From 6bbbd65aeb772508b3434dcc3085a064c1f11623 Mon Sep 17 00:00:00 2001 From: Jose Carlos Moro Date: Wed, 28 Oct 2026 00:01:48 +0100 Subject: [PATCH] feat(assistant): candidate selection with Claude assessor and read model columns --- README.md | 20 +- ...dd_catalog_columns_to_product_documents.py | 33 ++ docs/plan-assistant-genai.md | 527 +++++++++++++----- docs/step-17-assistant-recommend.md | 220 ++++++++ docs/step-18-assistant-agent.md | 58 ++ src/contexts/assistant/application/dtos.py | 33 ++ .../application/use_cases/index_catalog.py | 10 +- .../use_cases/recommend_candidate.py | 128 +++++ .../assistant/domain/assessment_usage.py | 9 + src/contexts/assistant/domain/candidate.py | 51 ++ src/contexts/assistant/domain/exceptions.py | 13 +- .../domain/ports/candidate_assessor.py | 21 + .../assistant/domain/ports/document_index.py | 5 +- .../assistant/domain/product_document.py | 8 + src/contexts/assistant/domain/rubric.py | 29 + src/contexts/assistant/domain/selection.py | 60 ++ .../infrastructure/llm/anthropic_errors.py | 20 + .../llm/anthropic_llm_client.py | 14 +- .../llm/claude_candidate_assessor.py | 137 +++++ .../infrastructure/persistence/models.py | 6 +- .../persistence/sqlalchemy_document_index.py | 22 +- src/shared/infrastructure/settings.py | 32 ++ tests/fakes/candidate_assessor.py | 59 ++ tests/fakes/documents.py | 24 +- .../test_recommend_candidate_deadline.py | 80 +++ .../test_claude_candidate_assessor.py | 228 ++++++++ .../test_sqlalchemy_documents.py | 56 +- .../application/test_index_catalog.py | 35 ++ .../application/test_recommend_candidate.py | 235 ++++++++ .../assistant/domain/test_product_document.py | 14 + .../assistant/domain/test_selection.py | 173 ++++++ .../shared/infrastructure/test_settings.py | 5 + 32 files changed, 2199 insertions(+), 166 deletions(-) create mode 100644 alembic/versions/ff5e81205e1e_add_catalog_columns_to_product_documents.py create mode 100644 docs/step-17-assistant-recommend.md create mode 100644 docs/step-18-assistant-agent.md create mode 100644 src/contexts/assistant/application/use_cases/recommend_candidate.py create mode 100644 src/contexts/assistant/domain/assessment_usage.py create mode 100644 src/contexts/assistant/domain/candidate.py create mode 100644 src/contexts/assistant/domain/ports/candidate_assessor.py create mode 100644 src/contexts/assistant/domain/rubric.py create mode 100644 src/contexts/assistant/domain/selection.py create mode 100644 src/contexts/assistant/infrastructure/llm/anthropic_errors.py create mode 100644 src/contexts/assistant/infrastructure/llm/claude_candidate_assessor.py create mode 100644 tests/fakes/candidate_assessor.py create mode 100644 tests/integration/contexts/assistant/application/test_recommend_candidate_deadline.py create mode 100644 tests/integration/contexts/assistant/infrastructure/test_claude_candidate_assessor.py create mode 100644 tests/unit/contexts/assistant/application/test_recommend_candidate.py create mode 100644 tests/unit/contexts/assistant/domain/test_selection.py diff --git a/README.md b/README.md index 8b6a94a..a09aad9 100644 --- a/README.md +++ b/README.md @@ -7,7 +7,7 @@ A reference **Generative AI backend** in Python: a **RAG** assistant that answers from a product catalog, built with **FastAPI**, **Domain-Driven Design (DDD)** and the **Hexagonal (Ports & Adapters)** pattern. The LLM, the embeddings and the vector store are adapters behind ports — swap providers, fake them in tests, and keep them out of your domain. [![CI](https://github.com/jcmoro/python-fastapi/actions/workflows/ci.yml/badge.svg?branch=main)](https://github.com/jcmoro/python-fastapi/actions/workflows/ci.yml) -![Tests](https://img.shields.io/badge/tests-137%20passing-2e7d32?style=flat) +![Tests](https://img.shields.io/badge/tests-192%20passing-2e7d32?style=flat) ![Architecture](https://img.shields.io/badge/architecture-DDD%20%2B%20Hexagonal-2e7d32?style=flat) ![mypy](https://img.shields.io/badge/mypy-strict-1565c0?style=flat) ![GenAI](https://img.shields.io/badge/GenAI-RAG%20assistant-6a1b9a?style=flat) @@ -172,27 +172,35 @@ src/ question.py # Value object (stripped, non-empty, ≤ 1,000 chars) completion.py # Completion: text, model, tokens, latency catalog_product.py # assistant's own view of a product - product_document.py # ProductDocument (content + content_hash), EmbeddedDocument + product_document.py # ProductDocument (catalog fields, content, content_hash) retrieved_document.py # Search hit with cosine similarity - exceptions.py # InvalidQuestionError, LLMUnavailableError + candidate.py # Candidate, RubricAnswers, Assessment (+ usage, failure reason) + assessment_usage.py # API calls and tokens per assessment + rubric.py # The two yes/no rubric questions, defined once + selection.py # select(): filter, rank, tie-break, threshold — pure + exceptions.py # InvalidQuestionError, LLMUnavailableError, InvalidAssessmentError ports/ llm_client.py # LLMClient ABC — documents which failures it raises embedding_client.py # One embedding per text, same order document_search.py # Top-k, most similar first document_index.py # Hashes, upsert, delete catalog_reader.py # Products from the catalog, no duplicates + candidate_assessor.py # One candidate per call; probabilities in [0, 1] application/ - dtos.py # AskCommand, AnswerDTO, SourceDTO, UsageDTO, IndexReportDTO + dtos.py # Ask/Answer/Source/Usage/IndexReport/Candidate/Recommendation DTOs use_cases/ ask_assistant.py # Embed → retrieve → generate, fallback without generation - index_catalog.py # Incremental reindex + orphan deletion + index_catalog.py # Incremental reindex + orphan deletion (before upsert) + recommend_candidate.py # Top-N, deadline, cancellation, select(), one log line infrastructure/ # The only place anthropic, fastembed and pgvector live. llm/ + anthropic_errors.py # Shared transient-vs-configuration classification anthropic_llm_client.py # Timeout, retries, error classification + claude_candidate_assessor.py # Rubric via structured output; 3 samples; per-call concurrency limit embeddings/ fastembed_embedding_client.py persistence/ - models.py # ProductDocumentModel (vector(384), HNSW cosine index) + models.py # ProductDocumentModel (vector(384), HNSW, unique sku) sqlalchemy_document_index.py sqlalchemy_document_search.py catalog/ diff --git a/alembic/versions/ff5e81205e1e_add_catalog_columns_to_product_documents.py b/alembic/versions/ff5e81205e1e_add_catalog_columns_to_product_documents.py new file mode 100644 index 0000000..d337fa9 --- /dev/null +++ b/alembic/versions/ff5e81205e1e_add_catalog_columns_to_product_documents.py @@ -0,0 +1,33 @@ +"""add catalog columns to product_documents + +Revision ID: ff5e81205e1e +Revises: 4aa1b2bf2668 +Create Date: 2026-10-27 22:38:12 + +""" +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa + + +revision: str = 'ff5e81205e1e' +down_revision: str | None = '4aa1b2bf2668' +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + op.add_column('product_documents', sa.Column('sku', sa.String(length=64), nullable=True)) + op.add_column('product_documents', sa.Column('name', sa.String(length=200), nullable=True)) + op.add_column('product_documents', sa.Column('price_cents', sa.Integer(), nullable=True)) + op.add_column('product_documents', sa.Column('currency', sa.String(length=3), nullable=True)) + op.create_index(op.f('ix_product_documents_sku'), 'product_documents', ['sku'], unique=True) + + +def downgrade() -> None: + op.drop_index(op.f('ix_product_documents_sku'), table_name='product_documents') + op.drop_column('product_documents', 'currency') + op.drop_column('product_documents', 'price_cents') + op.drop_column('product_documents', 'name') + op.drop_column('product_documents', 'sku') diff --git a/docs/plan-assistant-genai.md b/docs/plan-assistant-genai.md index dcc0379..fa572d0 100644 --- a/docs/plan-assistant-genai.md +++ b/docs/plan-assistant-genai.md @@ -1,182 +1,443 @@ -# Plan — Assistant: GenAI over the catalog (LLM port → RAG → agent → deploy) - -Bounded context: `assistant`. Architecture: DDD + Hexagonal (per `CLAUDE.md`). -Goal: answer natural-language questions about the catalog, with the LLM, the embeddings and the -agent framework treated as **adapters behind ports**. The domain never imports `anthropic`, -`langgraph` or `pgvector`. - -Delivered in four PRs. Each one is shippable on its own and keeps CI green. - -## Decisions taken - -- **LLM provider**: Anthropic (`anthropic` SDK, `AsyncAnthropic`). Model and timeout come from - settings; default to the cheapest current Claude model for dev. -- **Vector store**: `pgvector` inside the existing Postgres. No new infrastructure. -- **Embeddings**: behind an `EmbeddingClient` port. Dev/CI adapter: `fastembed` (local ONNX model, - no API key, no cost) with `sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2` - (384 dimensions, ~50 languages: Spanish and English questions both work). A hosted adapter - (e.g. Voyage AI) can be added later without touching the domain. -- **Agent framework**: LangGraph, confined to `infrastructure/agent/`. -- **Cross-context rule**: `assistant` never imports `catalog.domain`. It keeps its own read model - (`product_documents`) fed through DTOs. -- **Unit tests never call a real model**: `FakeLLMClient` / `FakeEmbeddingClient` at the port - boundary, same pattern as `FakeProductRepository`. -- **When the LLM fails**: PR 1 answers 503 (there is nothing to retrieve yet). From PR 2 on, a - deterministic fallback answer (top-k search results without generation) with `fallback_used=true`. -- **Indexing trigger: reindex on demand.** `assistant` owns an `IndexCatalog` use case that reads - products through its own `CatalogReader` port (DTOs only) and updates `product_documents`. - `catalog` does not change and does not know `assistant` exists. It runs only via `make reindex`; - there is no public endpoint while the API has no authentication. Trade-off accepted: the index - is stale until the next reindex. - - *Rejected — synchronous inside `CreateProduct` / `UpdateProduct`*: makes `catalog` depend on - `assistant` and puts embedding latency on every write. - - *Future evolution (not in PR 2)*: `catalog` publishes `ProductCreated` / `ProductUpdated` / - `ProductDeleted` domain events through a transactional **outbox**; `assistant` consumes them - and reindexes only what changed. Needs an event/outbox mechanism the repo does not have yet. +# Plan — Assistant: IA generativa sobre el catálogo (puerto LLM → RAG → selección de candidatos → agente → despliegue → evaluador jev) + +Bounded context: `assistant`. Arquitectura: DDD + Hexagonal (según `CLAUDE.md`). + +Objetivo: responder preguntas en lenguaje natural sobre el catálogo, tratando el LLM, los +embeddings y el framework de agentes como **adaptadores detrás de puertos**. El dominio nunca +importa `anthropic`, `langgraph` ni `pgvector`. + +Se entrega en seis PR (el PR 3 se divide en 3a y 3b). Cada uno se puede publicar por sí solo y mantiene el CI en verde. + +## Decisiones tomadas + +- **Proveedor LLM**: Anthropic (SDK `anthropic`, `AsyncAnthropic`). Modelo y timeout salen de los + settings; por defecto, el modelo de Claude más barato para desarrollo. +- **Almacén vectorial**: `pgvector` dentro del Postgres existente. Sin infraestructura nueva. +- **Embeddings**: detrás del puerto `EmbeddingClient`. Adaptador para desarrollo y CI: `fastembed` + (modelo ONNX local, sin API key y sin coste) con + `sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2` (384 dimensiones, ~50 idiomas: + funcionan preguntas en español y en inglés). Más adelante se puede añadir un adaptador alojado + (p. ej. Voyage AI) sin tocar el dominio. +- **Framework de agentes**: LangGraph, confinado en `infrastructure/agent/`. +- **Regla entre contextos**: `assistant` nunca importa `catalog.domain`. Mantiene su propio read + model (`product_documents`), alimentado mediante DTOs. +- **Los tests unitarios nunca llaman a un modelo real**: `FakeLLMClient` / `FakeEmbeddingClient` en + el límite del puerto, con el mismo patrón que `FakeProductRepository`. +- **Cuando el LLM falla**: el PR 1 responde 503 (todavía no hay nada que recuperar). Desde el PR 2, + una respuesta de fallback determinista (los resultados top-k sin generación) con + `fallback_used=true`. +- **Disparador del indexado: reindexado bajo demanda.** `assistant` tiene un caso de uso + `IndexCatalog` que lee los productos a través de su propio puerto `CatalogReader` (solo DTOs) y + actualiza `product_documents`. `catalog` no cambia y no sabe que `assistant` existe. Solo se + ejecuta con `make reindex`: no hay endpoint público mientras la API no tenga autenticación. + Compromiso aceptado: el índice queda desactualizado hasta el siguiente reindexado. + - *Descartado — síncrono dentro de `CreateProduct` / `UpdateProduct`*: haría que `catalog` + dependiera de `assistant` y cargaría la latencia de los embeddings en cada escritura. + - *Evolución futura (no entra en el PR 2)*: `catalog` publica los eventos de dominio + `ProductCreated` / `ProductUpdated` / `ProductDeleted` mediante un **outbox** transaccional; + `assistant` los consume y reindexa solo lo que cambió. Necesita un mecanismo de eventos/outbox + que el repo todavía no tiene. --- -## PR 1 — `LLMClient` port + `AskAssistant` use case (no RAG yet) +## PR 1 — Puerto `LLMClient` + caso de uso `AskAssistant` (todavía sin RAG) -### Domain — `src/contexts/assistant/domain/` +### Dominio — `src/contexts/assistant/domain/` -- `ports/llm_client.py` — `LLMClient` ABC: +- `ports/llm_client.py` — ABC `LLMClient`: `async complete(*, system: str, prompt: str, max_tokens: int) -> Completion`. -- `completion.py` — `Completion` frozen dataclass: `text`, `model`, `input_tokens`, +- `completion.py` — dataclass inmutable `Completion`: `text`, `model`, `input_tokens`, `output_tokens`, `latency_ms`. - `exceptions.py` — `InvalidQuestionError(InvalidInputError)`, `LLMUnavailableError(ServiceUnavailableError)`. -- `question.py` — `Question` value object: non-empty, max length (e.g. 1,000 chars), stripped. +- `question.py` — value object `Question`: no vacío, longitud máxima (p. ej. 1.000 caracteres), + sin espacios en los extremos. -### Application — `src/contexts/assistant/application/` +### Aplicación — `src/contexts/assistant/application/` - `dtos.py` — `AskCommand(question: str)`, `AnswerDTO(text, usage)`, `UsageDTO(model, input_tokens, output_tokens, latency_ms)`. -- `use_cases/ask_assistant.py` — builds the prompt, calls `LLMClient`, maps `Completion` → `AnswerDTO`. - `LLMUnavailableError` propagates. +- `use_cases/ask_assistant.py` — construye el prompt, llama a `LLMClient` y convierte `Completion` + en `AnswerDTO`. `LLMUnavailableError` se propaga. -### Infrastructure — `src/contexts/assistant/infrastructure/` +### Infraestructura — `src/contexts/assistant/infrastructure/` -- `llm/anthropic_llm_client.py` — `AnthropicLLMClient(LLMClient)`; wraps `AsyncAnthropic`, sets - timeout and retries once at construction, measures latency, reads token usage from the response. - Transient SDK errors (connection, timeout, 429, 5xx) become `LLMUnavailableError`; configuration - errors (400/401/403/404) propagate unchanged (500). +- `llm/anthropic_llm_client.py` — `AnthropicLLMClient(LLMClient)`; envuelve `AsyncAnthropic`, fija + timeout y reintentos una sola vez al construirse, mide la latencia y lee el uso de tokens de la + respuesta. Los errores transitorios del SDK (conexión, timeout, 429, 5xx) pasan a + `LLMUnavailableError`; los de configuración (400/401/403/404) se propagan sin cambios (500). ### UI — `src/contexts/assistant/ui/` - `schemas.py` — `AskRequest`, `AnswerResponse` (Pydantic, `extra="forbid"`). - `routers/assistant.py` — `POST /assistant/ask`, tag `assistant`. -- `dependencies.py` — per-context DI wiring. +- `dependencies.py` — cableado de dependencias del contexto. ### Shared - `domain/errors.py` — `ServiceUnavailableError(DomainError)`. -- `infrastructure/exception_handlers.py` — maps `ServiceUnavailableError` to 503. +- `infrastructure/exception_handlers.py` — convierte `ServiceUnavailableError` en 503. - `settings.py` — `anthropic_api_key: SecretStr | None`, `llm_model: str`, `llm_timeout_s: float` - (per attempt), `llm_max_retries: int` (default 0), `llm_max_tokens: int`. -- `.env.example` — the new variables, key left empty. -- `pyproject.toml` — add `anthropic`. + (por intento), `llm_max_retries: int` (0 por defecto), `llm_max_tokens: int`. +- `.env.example` — las variables nuevas, con la clave vacía. +- `pyproject.toml` — añadir `anthropic`. ### Tests -- `tests/fakes/llm_client.py` — `FakeLLMClient`, shared by unit and e2e. -- `tests/integration/contexts/assistant/` — adapter against `httpx2.MockTransport`: error - classification and a single attempt when `max_retries=0`. -- `tests/e2e/test_exception_handlers.py` — each shared domain error maps to its status code. -- `tests/unit/contexts/assistant/` — tests for `Question` - invariants, happy path, usage propagated to the DTO, `LLMUnavailableError` propagated when the - fake raises. -- `tests/e2e/test_assistant.py` — override the `LLMClient` dependency with the fake; 200, 422 on - empty question, 503 when the fake fails. +- `tests/fakes/llm_client.py` — `FakeLLMClient`, compartido por los tests unitarios y los e2e. +- `tests/integration/contexts/assistant/` — el adaptador contra `httpx2.MockTransport`: + clasificación de errores y un único intento con `max_retries=0`. +- `tests/e2e/test_exception_handlers.py` — cada error de dominio compartido devuelve su código de + estado. +- `tests/unit/contexts/assistant/` — tests de las invariantes de `Question`, del caso feliz, del uso + propagado al DTO y de `LLMUnavailableError` propagado cuando el fake falla. +- `tests/e2e/test_assistant.py` — sustituye la dependencia `LLMClient` por el fake: 200, 422 con la + pregunta vacía y 503 cuando el fake falla. --- -## PR 2 — RAG with pgvector +## PR 2 — RAG con pgvector -- `docker-compose.yml` — `db` and `db-test` move to `pgvector/pgvector:pg16`. -- Alembic migration — `CREATE EXTENSION IF NOT EXISTS vector`; table `product_documents` +- `docker-compose.yml` — `db` y `db-test` pasan a `pgvector/pgvector:pg16`. +- Migración de Alembic — `CREATE EXTENSION IF NOT EXISTS vector`; tabla `product_documents` (`product_id UUID PK`, `content TEXT`, `content_hash TEXT`, `embedding vector(384)`, - `updated_at`), HNSW index on `embedding` (cosine). -- Domain ports — + `updated_at`), índice HNSW sobre `embedding` (coseno). +- Puertos de dominio — - `EmbeddingClient.embed(texts) -> list[list[float]]`. - `DocumentSearch.search(embedding, k) -> list[RetrievedDocument]`. - - `DocumentIndex`: read the stored `product_id → content_hash`, upsert documents, delete - documents by `product_id`. - - `CatalogReader.read_all() -> list[CatalogProduct]` — `assistant`'s own view of a product, - not `catalog`'s DTO. `CatalogReader.missing(ids)` confirms which ids really no longer exist - (paging is not snapshot-consistent, so absence from `read_all` is not proof of deletion). - - `EmbeddingClient.model_name` — part of the `content_hash`, so changing the model re-embeds. -- Application — - - `IndexCatalog` use case, **incremental**: builds each product's document content and its - `content_hash`; embeds and upserts only new or changed products (Product has no - `updated_at`, so the hash — over model + content — is the change detector), in batches of - 256; **deletes orphaned documents** only after `CatalogReader.missing` confirms the product no - longer exists. Returns counts (indexed, unchanged, deleted). - - `AskAssistant` now embeds the question, retrieves top-k, puts them in the prompt and returns - them as `sources`. `AnswerDTO` gains `sources` and `fallback_used`. -- Fallback — `AskAssistant` catches `LLMUnavailableError` and returns the top-k results without - generation, `fallback_used=true`, 200. The 503 from PR 1 no longer reaches the client. -- Infrastructure — - - `FastEmbedEmbeddingClient` (model, cache dir and `local_files_only` from settings). The - `Dockerfile` downloads the model at build time; at runtime it is only loaded from disk and - warmed up in the app lifespan. - - `SqlAlchemyDocumentSearch` (own short-lived session, so no connection is held during the LLM - call) / `SqlAlchemyDocumentIndex` (`pgvector.sqlalchemy.Vector`). `RAG_TOP_K` ≤ 40 - (HNSW default `ef_search`). - - `CatalogReader` adapter: the **only** place `assistant` touches `catalog`. It calls - `catalog`'s public application API (`ListProducts`, paginated) and maps `ProductDTO` to - `CatalogProduct`. Never `catalog.domain` or `catalog.infrastructure`. -- UI (input adapters) — - - `ui/cli/reindex.py` — entry point `python -m contexts.assistant.ui.cli.reindex`. Builds - `IndexCatalog` with the **same dependency providers** as the API (`ui/dependencies.py`, - `shared/infrastructure/database.py`), runs it and prints the counts. An input adapter like - the router; reusable as a Kubernetes `Job` in PR 4. - - `Makefile` — `make reindex` runs that module in the `api` container. - - No HTTP endpoint for indexing. -- `pyproject.toml` — add `pgvector`, `fastembed`. + - `DocumentIndex`: leer el `product_id → content_hash` guardado, hacer upsert de documentos y + borrar documentos por `product_id`. + - `CatalogReader.read_all() -> list[CatalogProduct]` — la vista propia de `assistant` sobre un + producto, no el DTO de `catalog`. `CatalogReader.missing(ids)` confirma qué ids ya no existen + de verdad (la paginación no es consistente con una instantánea, así que no aparecer en + `read_all` no prueba que se haya borrado). + - `EmbeddingClient.model_name` — forma parte del `content_hash`, así que cambiar de modelo vuelve + a calcular los embeddings. +- Aplicación — + - Caso de uso `IndexCatalog`, **incremental**: construye el contenido del documento de cada + producto y su `content_hash`; calcula embeddings y hace upsert solo de los productos nuevos o + cambiados (`Product` no tiene `updated_at`, así que el hash —sobre modelo + contenido— es lo + que detecta los cambios), en lotes de 256. **Borra los documentos huérfanos** solo después de + que `CatalogReader.missing` confirme que el producto ya no existe. Devuelve los recuentos + (indexados, sin cambios, borrados). + - `AskAssistant` ahora calcula el embedding de la pregunta, recupera el top-k, lo pone en el + prompt y lo devuelve como `sources`. `AnswerDTO` gana `sources` y `fallback_used`. +- Fallback — `AskAssistant` captura `LLMUnavailableError` y devuelve los resultados top-k sin + generación, con `fallback_used=true` y 200. El 503 del PR 1 deja de llegar al cliente. +- Infraestructura — + - `FastEmbedEmbeddingClient` (modelo, directorio de caché y `local_files_only` desde los + settings). El `Dockerfile` descarga el modelo al construir la imagen; en ejecución solo se carga + de disco y se calienta en el lifespan de la app. + - `SqlAlchemyDocumentSearch` (con su propia sesión corta, para no retener ninguna conexión durante + la llamada al LLM) / `SqlAlchemyDocumentIndex` (`pgvector.sqlalchemy.Vector`). `RAG_TOP_K` ≤ 40 + (el `ef_search` por defecto de HNSW). + - Adaptador de `CatalogReader`: el **único** sitio donde `assistant` toca `catalog`. Llama a la + API pública de aplicación de `catalog` (`ListProducts`, paginado; `GetProduct` para confirmar + bajas) y convierte `ProductDTO` en `CatalogProduct`. Nunca `catalog.domain` ni + `catalog.infrastructure`. +- UI (adaptadores de entrada) — + - `ui/cli/reindex.py` — punto de entrada `python -m contexts.assistant.ui.cli.reindex`. Construye + `IndexCatalog` con los **mismos proveedores de dependencias** que la API (`ui/dependencies.py`, + `shared/infrastructure/database.py`), lo ejecuta e imprime los recuentos. Es un adaptador de + entrada igual que el router, reutilizable como `Job` de Kubernetes en el PR 4. + - `Makefile` — `make reindex` ejecuta ese módulo en el contenedor `api`. + - Sin endpoint HTTP para el indexado. +- `pyproject.toml` — añadir `pgvector` y `fastembed`. - Tests — - - Unit with fakes: incremental indexing (unchanged products are not re-embedded), **orphan - deletion**, retrieval in `AskAssistant`, fallback when the fake LLM raises. - - Integration against real `db-test`: search adapter (seed three documents, assert the nearest - one comes back first) and index adapter (upsert, hashes, delete). - - E2E asserts the fallback flag instead of 503. -- README — mark roadmap step 2 as done and replace "`assistant` never imports `catalog`" with the - real rule (only the `CatalogReader` adapter, through `catalog`'s application API). + - Unitarios con fakes: indexado incremental (los productos sin cambios no recalculan su + embedding), **borrado de huérfanos**, recuperación en `AskAssistant` y fallback cuando el LLM + falso falla. + - Integración contra el `db-test` real: adaptador de búsqueda (se siembran tres documentos y se + comprueba que vuelve primero el más cercano) y adaptador del índice (upsert, hashes, borrado). + - Los e2e comprueban el indicador de fallback en lugar del 503. +- README — marcar el paso 2 del roadmap como hecho y sustituir «`assistant` never imports + `catalog`» por la regla real (solo el adaptador `CatalogReader`, a través de la API de aplicación + de `catalog`). --- -## PR 3 — Agent with LangGraph +## PR 3a — Read model y selección de candidatos + +Sin punto de entrada todavía: `RecommendCandidate` solo lo ejercitan sus tests hasta que el PR 3b +añade la herramienta del agente. No cambia ningún comportamiento existente. + +- **Read model** — una migración añade `sku` (índice único), `name` y el precio (`price_cents`, + `currency`) como columnas de `product_documents`. Las rellena `IndexCatalog` desde el DTO del + catálogo; `catalog` no cambia. Las filas con alguna de estas columnas a nulo cuentan como + cambiadas, así que el reindexado incremental las rellena (con test). `get_product_by_sku` lee la + columna `sku`; la respuesta sin recomendación lista nombre, SKU y precio. + - **Orden en `IndexCatalog`**: los huérfanos confirmados se borran **antes** del upsert. Si no, un + producto borrado y otro nuevo con el mismo SKU entre dos reindexados harían chocar el upsert con + el índice único, y todo el reindexado haría rollback. Lo cubre un test que reproduce ese caso. + +### Selección de candidatos + +**Objetivo** — tras la recuperación, se puede recomendar un producto de entre los +candidatos top-k para preguntas como *«¿cuál de estos me recomiendas para X?»*, o declinar +explícitamente la recomendación. + +**Principio de diseño — el modelo lee, el código decide.** Es una decisión de diseño de este +proyecto, no una recomendación externa. + +- **Cada candidato se evalúa por separado, nunca todos en una sola pregunta.** Elegir entre varios + elementos en una sola pregunta puede depender del orden en que se presentan; evaluar cada uno por + separado, en paralelo, y ordenar en código elimina esa dependencia. La **eval de invariancia al + orden** (PR 3b) es la que comprueba que se cumple. +- Cada evaluación son unas pocas preguntas pequeñas de sí/no (`noul`), un juicio cada una, con un + significado explícito para sí y para no. +- El filtrado, la ordenación y los umbrales viven en el código, nunca en el modelo. +- El estado que se envía por candidato contiene solo lo que necesitan las preguntas. + +**Rúbrica** (definida una vez, en código; todos los evaluadores la usan sin cambios) + +| Id | Pregunta (`noul`) | Papel | +|---|---|---| +| `meets_requirements` | «¿Cumple los requisitos explícitos de la pregunta?». Criterio: sí si cumple todos los requisitos que la pregunta enuncia explícitamente; **si la pregunta no enuncia requisitos explícitos, se responde sí**. | **Filtro.** Se excluyen los candidatos cuya probabilidad queda por debajo de `selector_requirements_threshold`. | +| `suitable_for_use` | «¿Sirve para el uso X?», siendo X el uso por el que pregunta la pregunta. | **Ordenación.** Los candidatos que quedan se ordenan solo por esta probabilidad. Sin pesos. | + +Sin preguntas `score`: la expectativa de `score` no está calibrada entre niveles, así que nunca +entra en una ordenación. Los números (precios, tamaños) nunca se le preguntan al modelo; cuando +hace falta, se calculan en código. + +**Dominio** + +- `Candidate` — `product_id`, `sku`, `name`, `price_cents`, `currency`, `content` (el texto del + documento indexado) y `retrieval_score`. El catálogo no tiene descripción de producto, así que el + texto del candidato es el `content` del documento. +- `Assessment` — `product_id`, la probabilidad de sí de cada pregunta de la rúbrica + (`meets_requirements`, `suitable_for_use`), o `None` si ese candidato no se pudo evaluar. +- `Selection` — `product_id` elegido o `None`, `score` (la probabilidad de `suitable_for_use` del + elegido), `fallback_used`, `reason`: `chosen` | `below_threshold` | `no_assessments`. +- Puerto `CandidateAssessor.assess(question, candidate) -> Assessment`. Su docstring fija el + contrato: un candidato por llamada, nunca ve a los demás candidatos, las respuestas son + probabilidades en `[0, 1]` y solo lanza `LLMUnavailableError` ante fallos transitorios. +- `select(candidates, assessments, requirements_threshold, min_score) -> Selection` — función + pura. Recibe también los candidatos porque el desempate usa su `retrieval_score` y porque solo + puede devolver uno de ellos: + - descarta los candidatos que no se pudieron evaluar; + - se queda con los que tienen `meets_requirements >= requirements_threshold`; + - ordena el resto por `suitable_for_use`, de mayor a menor. Los **empates** los gana el de mayor + `retrieval_score`; si también empatan, el de menor `product_id`, para que la elección sea + siempre determinista; + - recomienda el primero si su `suitable_for_use >= min_score`; si no, `None`. + + Las dos comparaciones son **mayor o igual**. Con 3 muestras las probabilidades posibles son 0; + 0,33; 0,67 y 1, así que un umbral de 0,5 equivale a «mayoría sí» (2 de 3). + +**Aplicación** + +- Caso de uso `RecommendCandidate`: recibe los candidatos de la recuperación y solo pasa al + evaluador los **`selector_top_n`** primeros (10 por defecto). En el PR 3b, el agente pide + `selector_top_n` resultados al recomendar. Es independiente de `rag_top_k`, que sigue siendo la + cantidad de contexto que recibe el LLM para responder. El caso de uso lanza la evaluación de todos + los candidatos a la vez, con un plazo global `selector_timeout_s` = 10 s, y después llama a + `select`. El límite de concurrencia (`selector_max_concurrency` = 10) se aplica **por llamada a la + API** dentro del evaluador: las muestras de un candidato salen juntas, así que los candidatos van + terminando uno tras otro y un plazo corto deja sin evaluar solo a algunos, no a todos. +- **Un candidato solo participa con todas sus muestras válidas.** Una muestra inválida (incluida una + respuesta cortada por `max_tokens`), fallida o que no llega a tiempo lo deja sin evaluar, y el log + registra la razón (`invalid_sample`, `error`, `deadline`). Si se cancela la petición, se cancelan + todas las evaluaciones; un error de configuración se propaga en el acto. +- Un fallo transitorio sigue lanzando `LLMUnavailableError`, como subclase + `AssessmentUnavailableError` que lleva el uso acumulado, para que el log no pierda los tokens de + las muestras que sí respondieron. +- Si no se pudo evaluar ningún candidato: fallback determinista — los candidatos ordenados por + `retrieval_score`, sin recomendación (`fallback_used = true`). + +**Infraestructura** + +- `ClaudeCandidateAssessor` — el único evaluador de este PR. Es un adaptador, así que **habla con el + SDK de Anthropic directamente**, con salida estructurada, y **reutiliza la clasificación de + errores del PR 1**; el puerto `LLMClient` no cambia. Claude no devuelve probabilidades, así que la + probabilidad de cada respuesta sale de la **self-consistency**: se pregunta la rúbrica + `selector_samples` veces (**3** por defecto: impar, para que sí y no no puedan empatar), con las + dos preguntas de la rúbrica en la misma llamada y un sí/no para cada una, y se usa la + **frecuencia del sí** como probabilidad. No se usa la confianza que declara el propio modelo. Las + muestras usan el muestreo por defecto del modelo (la Messages API actual rechaza `temperature` / + `top_p` / `top_k`). +- Las preguntas de la rúbrica se definen en código, no en prompts repartidos por los adaptadores, + para que un segundo evaluador (PR 5) las reutilice sin cambios. + +**Settings** — `selector_provider` (por ahora solo `claude`; el PR 5 añade `jev`), `selector_top_n` +(10 por defecto), `selector_max_concurrency` (10), `selector_timeout_s` (10), +`selector_samples` (3), `selector_max_tokens` (1024), `selector_requirements_threshold` (0,5) y +`selector_min_score` (0,5). Las evals fijan los valores finales de los umbrales. El evaluador usa +`effort: "low"` y no desactiva el razonamiento, porque `thinking: disabled` devuelve 400 en +`claude-opus-5-5` y `claude-sonnet-5-5`. + +**Presupuesto de llamadas** — `selector_top_n × selector_samples` = 10 × 3 = **30 llamadas a +Claude** por recomendación; con 10 en paralelo son unas tres rondas (~3 s), muy por debajo del +plazo de 10 s. Una ráfaga de 30 llamadas puede chocar con un límite bajo de Anthropic: un `429` es +transitorio según la clasificación del PR 1, así que ese candidato queda sin evaluar y la selección +se hace con el resto. + +**Log** — una línea por selección: proveedor, id del modelo, **llamadas hechas y llamadas +contestadas** (para que las evals detecten `429` y timeouts), latencia, tokens, coste y la razón +(`chosen`, `below_threshold`, `no_assessments`). Nunca datos de producto más allá de los ids. + +**Tests unitarios** (`CandidateAssessor` falso; para el adaptador de self-consistency, el cliente +de Anthropic sobre `httpx2.MockTransport` con respuestas guionizadas, como en el PR 1) + +- Gana el mayor `suitable_for_use`; en un empate, el de mayor `retrieval_score`; en un doble + empate, el de menor `product_id`. +- Un candidato por debajo de `requirements_threshold` queda excluido aunque sea el mejor ordenado. +- Sin requisitos explícitos → `meets_requirements` es sí y no filtra nada. +- Todos por debajo de `min_score` → sin recomendación; justo en `min_score` (o en + `requirements_threshold`) pasa (`>=`). +- Falla un candidato → se elige entre el resto. +- Fallan todos → fallback determinista por `retrieval_score`. +- Entrada barajada → misma selección. +- Nunca se devuelve un producto que no esté entre los candidatos. +- Self-consistency (3 muestras): 3 de 3 sí → 1,0; 2 de 3 → 0,67; 1 de 3 → 0,33; una muestra + inválida o cortada por `max_tokens` deja el candidato sin evaluar. +- Solo el top `selector_top_n` de la recuperación llega al evaluador. + +**Decidido** (con el usuario): rúbrica de dos preguntas con un filtro de requisitos y ordenación +solo por idoneidad (sin pesos); la puntuación es la probabilidad de «sirve para X»; +self-consistency para Claude con 3 muestras; desempate por `retrieval_score` (y después por +`product_id`); los dos umbrales a 0,5 con `>=`; el agente pide `selector_top_n` resultados al +recomendar, separado de `rag_top_k`; 10 en paralelo, plazo de 10 s, top 10 al selector; llamadas +contestadas registradas por selección; sin pregunta de seguimiento; `score` nunca en una +ordenación; el principio de diseño registrado como decisión de este proyecto; `name` como columna +igual que `sku` y el precio; el `content` del documento en lugar de `description`; huérfanos +borrados antes del upsert; `ClaudeCandidateAssessor` directo sobre el SDK; errores del LLM del +agente traducidos con la misma clasificación. -- Domain port — `CatalogAgent.run(question) -> AgentResult` (answer, tool calls made, usage). -- Infrastructure — `infrastructure/agent/langgraph_catalog_agent.py`: a small graph with two tools, - `semantic_search` (uses `DocumentSearch`) and `get_product_by_sku` (reads the read model), - then a final answer node. -- Guardrails — max steps per run, tool allow-list, same timeout as PR 1 and fallback as PR 2, token usage - summed across steps. -- `AskAssistant` delegates to the agent behind a setting flag (`assistant_mode = "rag" | "agent"`). -- `pyproject.toml` — add `langgraph`, `langchain-anthropic`. -- Tests — unit with a fake agent; one integration test of the graph wiring with a scripted fake model. - -### Evals (not in default CI) +--- -- `tests/evals/` + marker `eval` — ~10 questions with the expected SKU; reports hit rate and cost. - Run manually with a real key: `make eval`. +## PR 3b — Agente con LangGraph + +Se apoya en el PR 3a: usa `RecommendCandidate`, el read model con `sku`, `name` y precio, y la +rúbrica. + +- Puerto de dominio — `CatalogAgent.run(question) -> AgentResult` (respuesta, llamadas a + herramientas hechas, uso). +- Infraestructura — `infrastructure/agent/langgraph_catalog_agent.py`: un grafo pequeño con las + herramientas `semantic_search` (usa `DocumentSearch`), `get_product_by_sku` (lee el read model) y + `recommend_candidate` (más abajo), y después un nodo de respuesta final. + - El LLM del agente va por `langchain-anthropic`, que es un segundo camino hacia Anthropic fuera de + `AnthropicLLMClient`. El adaptador del agente **traduce sus errores a `LLMUnavailableError` con + la misma clasificación** que el PR 1 (transitorios frente a configuración) y aplica **el mismo + timeout**. +- Límites — **como mucho 5 llamadas a herramientas por ejecución** (`recommend_candidate` cuenta + como una). Al llegar al límite, el agente responde en modo RAG con lo recuperado y registra la + razón `max_steps`. Lista blanca de herramientas, el mismo timeout que el PR 1 y el mismo fallback + que el PR 2, con el uso de tokens sumado entre pasos. +- `AskAssistant` delega en el agente según un setting (`assistant_mode = "rag" | "agent"`), **`rag` + por defecto**: el agente es opt-in hasta que las evals demuestren que es mejor. Los e2e del PR 2 + no cambian. +- Dependencias — los proveedores de `RecommendCandidate` y del evaluador se añaden aquí (no en 3a), + reutilizando **el único cliente de Anthropic del asistente**, que se cierra en el lifespan. +- `pyproject.toml` — añadir `langgraph` y `langchain-anthropic`. +- Tests — unitarios con un agente falso; un test de integración del cableado del grafo con un modelo + falso guionizado. + +### Herramienta `recommend_candidate` + +**Agente** — herramienta nueva de LangGraph `recommend_candidate`. Si no hay recomendación, el +agente dice explícitamente que no hay una recomendación clara y lista los candidatos top-k (nombre, +SKU y precio). Por ahora, sin pregunta de seguimiento. + +### Evals (marker `eval`, fuera del CI por defecto) + +Necesitan una `ANTHROPIC_API_KEY` real, que el usuario pone en `.env`. Su valor nunca aparece en logs, +salidas de comandos ni documentos. **Antes de la primera ejecución real se informa del número de +llamadas** que hará. + +- Dataset de preguntas con el SKU esperado. +- Precisión y cobertura (con qué frecuencia recomienda) para varios valores de `selector_min_score` + y `selector_requirements_threshold`; los umbrales se eligen a partir de esa tabla. +- **Invariancia al orden**: se barajan los candidatos y la recomendación no debe cambiar. Es la + comprobación del principio de diseño. +- Consistencia al repetir: se pregunta dos veces y la recomendación no debe cambiar. +- Latencia y coste por selección, y llamadas contestadas frente a llamadas hechas. +- `tests/evals/` + marker `eval` — ~10 preguntas con el SKU esperado; informa del porcentaje de + aciertos y del coste del agente completo. Se ejecuta a mano con la clave real: `make eval`. --- -## PR 4 — Deploy +## PR 4 — Despliegue + +- `k8s/` — `Deployment` (liveness en `/healthz`), `Service`, `ConfigMap`, `Secret` (solo plantilla), + `Job` que ejecuta `alembic upgrade head`, `Job` que ejecuta + `python -m contexts.assistant.ui.cli.reindex` y Postgres con pgvector para uso local. +- `make k8s-up` / `make k8s-down` — se ejecuta en `kind` en local. +- CI — job opcional que crea un clúster kind, aplica los manifiestos y llama a `/healthz`. +- Opcional — la misma imagen en AWS (App Runner o ECS Fargate + RDS con pgvector). + +--- -- `k8s/` — `Deployment` (liveness on `/healthz`), `Service`, `ConfigMap`, `Secret` (template only), - `Job` running `alembic upgrade head`, `Job` running `python -m contexts.assistant.ui.cli.reindex`, - Postgres with pgvector for local use. -- `make k8s-up` / `make k8s-down` — runs on `kind` locally. -- CI — optional job that creates a kind cluster, applies the manifests and hits `/healthz`. -- Optional — the same image on AWS (App Runner or ECS Fargate + RDS with pgvector). +## PR 5 — Evaluador jev + +Añade un segundo `CandidateAssessor` sobre un modelo de decisión. Nada fuera del adaptador cambia: +el puerto, `select`, `RecommendCandidate` y la rúbrica vienen del PR 3a; la herramienta del agente, +del PR 3b. + +**Proveedor — Decisions API de OpenRouter** (decidido), con `httpx` directo. + +- `JevCandidateAssessor` — `POST {jev_base_url}/alpha/decisions` + (`https://openrouter.ai/api/alpha/decisions`), `Authorization: Bearer `. + Una petición por candidato: `{model, state, questions}`, con `questions` indexado por el id de la + rúbrica. Las dos preguntas de la rúbrica van en la misma petición, como dos preguntas `noul`. jev + devuelve directamente la probabilidad de cada `noul`, así que este evaluador **no necesita + self-consistency**: `selector_samples` solo se aplica a `ClaudeCandidateAssessor`. +- **Reglas de la petición** (OpenAPI): toda pregunta necesita `type` e `instructions`. + `noul.criteria`, si se envía, debe llevar exactamente `true` y `false`. `choice.criteria` es un + mapa de opciones obligatorio. `score.criteria` es una lista ordenada obligatoria con al menos un + nivel. `state` es un texto, un objeto o un array. +- **Respuesta**: `{id?, model, provider?, answers, usage{input_tokens, output_tokens, cost?}}`, una + respuesta por cada clave pedida. `noul` solo trae su probabilidad (sin `confidence`); `choice` y + `score` pueden añadir `probabilities` y `confidence`, y `score` puede añadir `legend`. + - **Parseo estricto**: se comprueba el `type` de cada respuesta. Una clave que falta o un `type` + inesperado deja el `Assessment` de ese candidato en `None`. Los campos opcionales + (`probabilities`, `confidence`, `legend`, `usage.cost`) se leen comprobando su presencia. +- **Clasificación de errores** (la misma regla que `AnthropicLLMClient` en el PR 1): + - Transitorios → `LLMUnavailableError`: errores de conexión, timeouts, `429`, `500`, `502`, + `503`, `524` (timeout en el borde de la red) y `529` (proveedor sobrecargado). + - De configuración o de la cuenta → se propagan (500) con una línea de log clara: `400`, `401`, + `402` (créditos insuficientes: reintentar no lo arregla), `403`, `404` y `413`. +- **Modelo fijado al build con fecha** `typesafe/jev-1.14-20261014` (el `canonical_slug` del + catálogo en vivo; la referencia de OpenRouter dice que las peticiones lo aceptan). Nunca el alias + `~typesafe/jev-latest`. Se registran en el log el `model` que devuelve cada respuesta (el build + exacto que respondió) y `usage.cost` (USD). +- **Privacidad**: se envían las preferencias de proveedor `zdr: true` y `data_collection: "deny"`. + El catálogo no tiene datos personales, pero es buena práctica. El OpenAPI define las dos; la guía + de jev no dice si jev las admite, así que **hay que comprobarlo con la primera petición real** (una + preferencia no admitida devuelve un error). +- **Tamaño del estado**: cada petición debe quedar muy por debajo de ~32.000 bytes de texto. Por + encima, TypeSafe pasa a su modo de contexto largo, que describe como más lento y menos preciso. + Una petición por candidato con solo la pregunta y el texto del candidato queda muy lejos de ese + límite. +- **Endpoint alpha, totalmente aislado en este adaptador**: nada fuera de él conoce la ruta, el + payload ni la forma de la respuesta. Pasar a TypeSafe directo (`POST + https://api.typesafe.ai/v1/systemone`, con el id del modelo sin prefijo) es un cambio de settings + más este adaptador. Queda registrado como riesgo conocido en el documento del paso. + +**Settings** — `selector_provider` gana `jev`. Nuevos: `jev_model` (build con fecha), +`jev_base_url` (`https://openrouter.ai/api` por defecto) y `jev_api_key: SecretStr | None` (la API +key de OpenRouter). + +**Evals** — jev frente a Claude sobre el dataset del PR 3b: precisión y cobertura por umbral, +invariancia al orden, consistencia al repetir, latencia y coste por selección. **Los umbrales se +eligen de nuevo para cada modelo**: las probabilidades cambian entre modelos para la misma +pregunta. Otros modelos de decisión sobre la misma Decisions API (22 de 13 fabricantes a +16-10-2026) se pueden comparar cambiando solo `model`. + +**Antes de escribir el adaptador** — confirmar el contrato en las fuentes oficiales. Si algo difiere +de este plan, avisar en lugar de adaptarlo en silencio. Fuentes: + +- Guía de jev en OpenRouter: `https://openrouter.ai/docs/guides/community/jev.md` +- Referencia de la Decisions API (OpenAPI): + `https://openrouter.ai/docs/api/api-reference/alphadecisions/submit-a-decisions-questions-and-answers-request.md` +- Catálogo en vivo: `GET https://openrouter.ai/api/v1/models?output_modalities=decisions` +- Documentación de TypeSafe: `https://docs.typesafe.ai` («How to build with System One», + «Confidence») --- -## Done means +## Criterio de «hecho» -- `make lint` and `make test` green; mypy strict passes on the new context. -- README updated: architecture section mentions the `assistant` context; repo description and - topics set (`fastapi`, `ddd`, `hexagonal-architecture`, `rag`, `pgvector`, `langgraph`). +- `make lint` y `make test` en verde; mypy strict pasa sobre el contexto nuevo. +- README actualizado: la sección de arquitectura menciona el contexto `assistant`; descripción y + topics del repo puestos (`fastapi`, `ddd`, `hexagonal-architecture`, `rag`, `pgvector`, + `langgraph`). diff --git a/docs/step-17-assistant-recommend.md b/docs/step-17-assistant-recommend.md new file mode 100644 index 0000000..d42e527 --- /dev/null +++ b/docs/step-17-assistant-recommend.md @@ -0,0 +1,220 @@ +# Paso 17 — Contexto `assistant`: read model y selección de candidatos (PR 3a) + +Rama `feature/assistant-recommend`. Plan: `docs/plan-assistant-genai.md`, sección «PR 3a — Read +model y selección de candidatos». El agente LangGraph va en el PR 3b (`docs/step-18-assistant-agent.md`). + +## Objetivo + +Que, dada una pregunta como «¿cuál de estos me recomiendas para X?» y los candidatos recuperados, +`RecommendCandidate` devuelva una de dos cosas: + +- **un producto recomendado**, elegido por reglas en código; +- **ninguno**, con su razón (`below_threshold` o `no_assessments`). + +La elección es siempre determinista y la decisión nunca la toma el modelo. Además, el índice gana +las columnas `sku`, `name` y precio que necesitará el PR 3b. + +## Qué + +1. **Read model:** una migración añade a `product_documents` las columnas `sku` (índice único), + `name`, `price_cents` y `currency`. Las rellena `IndexCatalog` desde el DTO del catálogo. Las + filas con alguna de esas columnas a nulo cuentan como cambiadas, así que el siguiente reindexado + las rellena. +2. **Orden en `IndexCatalog`:** los huérfanos confirmados se borran **antes** del upsert. +3. **Dominio:** + - `Candidate`, `Assessment` y `Selection`; + - el puerto `CandidateAssessor`; + - la rúbrica, definida una vez en código; + - la función pura `select`. +4. **Aplicación:** el caso de uso `RecommendCandidate`: + - pasa al evaluador solo el top `selector_top_n`; + - evalúa en paralelo, como mucho 10 a la vez, con un plazo global de 10 s; + - aplica `select`, con fallback determinista si no se pudo evaluar ningún candidato; + - escribe una línea de log por selección. +5. **Infraestructura:** `ClaudeCandidateAssessor`, directo sobre el SDK de Anthropic con salida + estructurada y self-consistency (3 muestras). Reutiliza la clasificación de errores del PR 1. +6. **Settings:** `selector_provider` (solo `claude`), `selector_top_n`, `selector_max_concurrency`, + `selector_timeout_s`, `selector_samples`, `selector_requirements_threshold` y + `selector_min_score`. +7. **Fake** `FakeCandidateAssessor` para los tests. + +## Por qué + +- El PR 2 responde con contexto, pero no decide. Preguntas como «¿cuál me recomiendas para X?» + necesitan elegir uno, o reconocer que no se puede, sin que la elección dependa de cómo se + presentan los candidatos. +- **El modelo lee y el código decide.** Es una decisión de diseño de este proyecto: elegir entre + varios productos en una sola pregunta puede depender del orden de presentación. Evaluar cada uno + por separado y ordenar en código lo evita, y la eval de invariancia al orden (PR 3b) lo + comprueba. +- Separar 3a de 3b deja la parte determinista y comprobable sin red (selección, índice) en un PR + propio, antes de introducir LangGraph. + +## Fuera de alcance + +- **PR 3b:** el agente LangGraph, sus herramientas (incluida `recommend_candidate` y cómo presenta + el caso sin recomendación), el límite de pasos, `assistant_mode` y las evals. +- **PR 5:** el evaluador jev. +- **PR 4:** el despliegue. +- **Punto de entrada:** en 3a, `RecommendCandidate` solo lo ejercitan sus tests. No cambia ningún + comportamiento existente. +- **Cambios en `catalog`**: ni una línea. +- **Llamadas reales a Anthropic**: todo se demuestra con fakes y `httpx2.MockTransport`. + +## Decisiones tomadas (con el usuario) + +| Tema | Decisión | Porqué | +|---|---|---| +| Rúbrica | Dos `noul`. `meets_requirements` es un **filtro**: si la pregunta no tiene requisitos explícitos, se responde sí. `suitable_for_use` **ordena**, sin pesos. | Un requisito explícito no se compensa con idoneidad; la idoneidad es lo único que ordena. | +| Puntuación | La probabilidad de `suitable_for_use` | Una sola magnitud, fácil de interpretar y de calibrar. | +| `score` y números | `score` nunca entra en una ordenación, y los números nunca se le preguntan al modelo | La expectativa de `score` no está calibrada entre niveles, y los modelos de decisión no comparan magnitudes con fiabilidad. | +| Probabilidad con Claude | Self-consistency con `selector_samples` = **3**. Las dos preguntas van en la misma llamada. | Claude no devuelve probabilidades. La confianza que declara el propio modelo está mal calibrada. | +| Umbrales | `selector_requirements_threshold` = 0,5 y `selector_min_score` = 0,5, ambos con `>=`. Las evals (PR 3b) fijan los valores finales. | Con 3 muestras los valores posibles son 0; 0,33; 0,67 y 1, así que 0,5 equivale a «mayoría sí». | +| Desempate | Gana el mejor `retrieval_score`; si también empatan, el `product_id` menor | Elección siempre determinista. El segundo criterio lo añadí yo para cubrir el doble empate. | +| Candidatos | Solo el top `selector_top_n` = 10 llega al evaluador, independiente de `rag_top_k` | Son dos necesidades distintas: entre cuántos se elige y cuánto contexto recibe el LLM. | +| Concurrencia y plazo | Como mucho 10 evaluaciones a la vez, con un plazo global de 10 s | 10 × 3 = 30 llamadas, unas 3 rondas (~3 s). | +| 429 | Error transitorio (clasificación del PR 1): ese candidato queda sin evaluar. El log registra las llamadas hechas y las contestadas. | Una ráfaga de 30 llamadas puede chocar con límites bajos de la cuenta, y las evals deben detectarlo. | +| Columnas del read model | `sku` (índice único), `name`, `price_cents` y `currency`, rellenadas desde el DTO. Las filas con nulos se tratan como cambiadas. | Antes estaban solo dentro de `content`. Así `catalog` no cambia. | +| `Candidate` | Lleva `content` (el texto del documento) en lugar de `description` | El catálogo no tiene descripción de producto. | +| Orden en `IndexCatalog` | Huérfanos confirmados borrados **antes** del upsert | Con el índice único, un SKU reutilizado entre dos reindexados haría chocar el upsert y el reindexado entero haría rollback. | +| `ClaudeCandidateAssessor` | Directo sobre el SDK de Anthropic, con salida estructurada y la clasificación de errores del PR 1 | Es un adaptador. El puerto `LLMClient` solo devuelve texto y no cambia. | +| PR 3 dividido | 3a (este) y 3b (agente) | Petición del usuario. | + +## Criterios de aceptación + +| # | Criterio | Cómo se demuestra | +|---|---|---| +| A1 | `select` filtra por `meets_requirements >= umbral`, ordena por `suitable_for_use`, desempata por `retrieval_score` y después por `product_id`, y aplica `>= min_score` | Tests unitarios de la función pura, incluido el caso exacto en el umbral | +| A2 | Nunca se devuelve un producto que no esté entre los candidatos, y barajar la entrada no cambia la selección | Tests unitarios | +| A3 | Self-consistency: 3/3 → 1,0; 2/3 → 0,67; 1/3 → 0,33. Una muestra inválida, o cortada por `max_tokens`, deja el candidato sin evaluar (CR2) | Tests del adaptador con el cliente de Anthropic sobre `httpx2.MockTransport` y respuestas guionizadas | +| A4 | `ClaudeCandidateAssessor` clasifica los errores como el PR 1: los transitorios dan `LLMUnavailableError` y los de configuración se propagan | Tests del adaptador | +| A5 | `RecommendCandidate` solo evalúa el top `selector_top_n`. Nunca hay más de `selector_max_concurrency` **llamadas** en vuelo (CR1). Lo que no llega en plazo cuenta como no evaluado. | Tests con un evaluador falso que simula retrasos; test del evaluador que mide las llamadas en vuelo | +| A6 | Si falla un candidato, se elige entre el resto. Si fallan todos, fallback ordenado por `retrieval_score` con `fallback_used=true` y razón `no_assessments` | Tests unitarios | +| A7 | `product_documents` tiene `sku` único, `name`, `price_cents` y `currency`. El reindexado los rellena desde el DTO, y las filas con nulos se recalculan | Test unitario del relleno, test de integración de la migración y `make reindex` real | +| A8 | Borrar un producto y crear otro con el mismo SKU entre dos reindexados no rompe el reindexado | Test que **falla con el orden actual** y pasa con el nuevo | +| A9 | Una línea de log por selección con proveedor, modelo, llamadas hechas y contestadas, latencia, tokens, coste y razón, sin datos de producto más allá de los ids | Test con `caplog` | +| A10 | El dominio y la aplicación de `assistant` no importan `anthropic` | `grep` | +| A11 | Los quality gates y el CI pasan | `make lint`, `make test` y CI de la PR | +| A12 | El plan y el README coinciden con el código | Revisión y badge de tests | + +## Decisiones tomadas durante la implementación + +| Tema | Decisión | Porqué | +|---|---|---| +| Firma de `select` | `select(candidates, assessments, requirements_threshold, min_score)`, no `select(assessments, …)` | El desempate necesita el `retrieval_score` de cada candidato, y solo se puede devolver uno de ellos. El plan está corregido. | +| Incompletos en el índice | `DocumentIndex.content_hashes()` devuelve `None` para las filas sin columnas de catálogo | `IndexCatalog` las trata como cambiadas sin lógica nueva, porque `None != hash`. | +| Muestras de Claude | ~~Secuenciales, con la concurrencia por candidato~~ → **en paralelo, con un semáforo por llamada** (CR1) | Con muestras seguidas, todos los candidatos acababan en la 3.ª ronda y un plazo corto los dejaba a todos sin evaluar. | +| Muestras inválidas | ~~No cuentan como sí~~ → **cualquier muestra inválida, fallida o que no llega deja el candidato sin evaluar** (CR2) | Regla del usuario. Las puntuaciones quedan en k/3 y los fallos del modelo se separan de sus juicios. | +| `effort` y `max_tokens` del evaluador | `output_config.effort = "low"` y ~~`max_tokens = llm_max_tokens` (4096)~~ → `selector_max_tokens` = 1024 (CR6) | El modelo por defecto (`claude-haiku-5-5`) piensa por defecto y esos tokens cuentan contra `max_tokens`. Con un límite pequeño, una muestra podría acabarse sin JSON. `low` mantiene corto el razonamiento de una clasificación sí/no. No se desactiva el thinking porque en otros modelos (`claude-opus-5-5`) `disabled` devuelve 400. | +| Clasificación de errores | Extraída a `anthropic_errors.transient_errors_as_unavailable()` y compartida por `AnthropicLLMClient` y `ClaudeCandidateAssessor` | Una sola regla; los tests del PR 1 siguen en verde sin cambios. | +| Uso por evaluación | `Assessment.usage` (`api_calls`, `input_tokens`, `output_tokens`, `cost_usd`) | El log de la selección suma el uso real de las evaluaciones completadas. | +| «Llamadas hechas y contestadas» en el log | `candidates` (evaluaciones pedidas), `assessed` (con respuesta útil), `api_calls` y tokens (también los de las evaluaciones que fallan a medias, CR4), y `unassessed=:` | Un 429 o un timeout dejan `assessed < candidates` con su razón, que es lo que las evals tienen que detectar. | +| Coste con Claude | `cost_usd=None` | La Messages API no devuelve el coste. Calcularlo exigiría una tabla de precios en settings, que no está pedida. jev (PR 5) sí lo devuelve. | +| Errores de configuración en paralelo | `asyncio.wait(..., return_when=FIRST_EXCEPTION)`, cancelación de todo lo pendiente en un `finally`, y relanzar la primera excepción tras recuperarlas todas (CR3) | Se propaga en el acto, no deja llamadas huérfanas al cancelar la petición, y asyncio no avisa de excepciones sin recoger. | +| `selector_provider` | `Literal["claude"]` | Solo existe ese evaluador; el PR 5 añade `jev`. | +| Proveedores de dependencias | Ninguno en 3a (CR5). En 3b, reutilizando el único cliente de Anthropic del asistente, el que se cierra en el lifespan. | En 3a no hay punto de entrada, y eran código muerto con un segundo cliente HTTP. | +| Uso al fallar | `AssessmentUnavailableError(LLMUnavailableError)` lleva `usage`. `AssessmentUsage` vive en su propio módulo. | El puerto no cambia de contrato (CR4), y el módulo propio evita una importación circular entre `candidate` y `exceptions`. | + +## Correcciones tras la code review + +La revisión (`/code-review medium`) dio 10 hallazgos y se comprobó cada uno contra el código. 4 son +reales y cambian el comportamiento, 2 son mejoras y se aplican, y 3 son falsos positivos (detalle al +final de esta sección). El hallazgo #2 se resolvió con una regla nueva del usuario. Cada corrección +empieza por un test que falla con el código anterior. + +| # | Qué | Por qué | Criterio de aceptación | +|---|---|---|---| +| CR1 | **El límite de concurrencia se aplica por llamada, no por candidato.** `selector_max_concurrency` (10) es el número de llamadas a la API en vuelo. Las 3 muestras de un candidato salen juntas, así que los candidatos van terminando uno tras otro. | Antes, los 10 candidatos arrancaban a la vez y cada uno hacía 3 muestras seguidas: todos acababan en la 3.ª ronda. Con llamadas lentas (~4 s), a los 10 s no terminaba **ninguno** y la respuesta caía siempre al fallback (todo o nada). | Nunca hay más de `selector_max_concurrency` llamadas en vuelo. Con llamadas lentas y el plazo a mitad, **algunos** candidatos llegan evaluados en lugar de ninguno. | +| CR2 | **Un candidato solo participa con las 3 muestras válidas.** Cualquier muestra inválida, fallida o que no llegue lo deja sin evaluar, y su razón (`invalid_sample`, `error` o `deadline`) queda en el log. Una respuesta cortada por `max_tokens` es una muestra inválida. | Decisión del usuario. Las puntuaciones se quedan en 0 / 0,33 / 0,67 / 1, y en las evals los fallos del modelo se separan de sus juicios. | Una muestra inválida deja el candidato sin evaluar con `invalid_sample`. Un corte por `max_tokens` es inválido. Si todos quedan sin evaluar, fallback determinista. El log trae la razón de cada candidato sin evaluar. | +| CR3 | **Cancelar y cortar antes de tiempo.** Si se cancela la petición, se cancelan todas las evaluaciones en curso. Ante un error de configuración se propaga en el acto, sin esperar al plazo. | Antes las llamadas seguían ejecutándose y facturándose en segundo plano, y un 401 esperaba los 10 s. | Al cancelar `execute`, se cancelan todas las evaluaciones. Un error de configuración se propaga en mucho menos que el plazo. | +| CR4 | **No perder el uso.** El puerto sigue lanzando `LLMUnavailableError` ante fallos transitorios, pero la excepción lleva el uso acumulado (`AssessmentUnavailableError`, subclase). El caso de uso lo suma al de la selección. | Antes, si fallaba la 3.ª muestra, se perdían los tokens de las dos primeras y el log se quedaba corto. | Si falla una muestra, el log incluye los tokens de las demás. | +| CR5 | **Quitar los proveedores sin uso** `get_candidate_assessor` y `get_recommend_candidate_use_case`. En el 3b se añaden, reutilizando el único cliente de Anthropic del asistente, el que se cierra en el lifespan. | Eran código muerto en 3a y creaban un segundo cliente HTTP. | `grep`: no existen en 3a. | +| CR6 | **`selector_max_tokens` propio, 1024 por defecto.** El razonamiento no se desactiva: se mantiene `effort: "low"`. | Dos booleanos no necesitan 4096. No se desactiva el razonamiento porque `thinking: disabled` devuelve 400 en `claude-opus-5-5` y en `claude-sonnet-5-5`, y el evaluador usa `llm_model`, que es configurable. | La petición lleva `max_tokens = selector_max_tokens`. Un corte por `max_tokens` es inválido (CR2). | + +**Falsos positivos:** + +- **«Intercambiar SKUs rompe el índice único».** En `catalog` el SKU no se puede modificar + (`UpdateProductCommand` no tiene `sku`). El único caso real, borrar y crear con el mismo SKU, ya lo + cubre A8. +- **«Borrar huérfanos antes del upsert deja huecos si algo falla».** El reindexado va en una sola + transacción (`get_session`) y ante cualquier error hace rollback completo. +- **«Docstrings narrativos en los puertos».** El `CLAUDE.md` del repo los permite en los puertos, + para lo que los tipos no expresan. + +## Evidencia + +### Rojo antes del verde + +``` +Paso 1 (read model): +FAILED test_product_document.py::TestCatalogFields::test_a7_carries_the_catalog_fields +FAILED test_index_catalog.py::TestReadModelColumns::test_a7_reindexes_documents_missing_catalog_columns +FAILED test_index_catalog.py::TestReadModelColumns::test_a8_a_reused_sku_does_not_break_the_reindex +FAILED test_sqlalchemy_documents.py::TestReadModelColumns::test_a7_upsert_stores_the_catalog_columns +FAILED test_sqlalchemy_documents.py::TestReadModelColumns::test_a7_a_row_missing_catalog_columns_reports_no_hash +FAILED test_sqlalchemy_documents.py::TestReadModelColumns::test_a8_sku_is_unique_in_the_index +A8 con las columnas ya añadidas, rojo por el bug de orden: + tests.fakes.documents.DuplicateSkuInIndexError: SKU-REUSED +Paso 2 (select): ModuleNotFoundError: contexts.assistant.domain.candidate +Paso 3 (evaluador): ModuleNotFoundError: ...infrastructure.llm.claude_candidate_assessor +Paso 4 (caso de uso): ModuleNotFoundError: ...application.use_cases.recommend_candidate +``` + +El A8 se separó a propósito: primero se añadieron las columnas, se comprobó que el test seguía +fallando por el **orden** (upsert antes de borrar el huérfano con el mismo SKU) y solo entonces se +cambió el orden. + +Correcciones de la code review, sobre el código anterior: + +``` +CR1 (comportamiento, API anterior): + test_cr1_slow_calls_still_assess_some_candidates_before_the_deadline → assert 0 >= 1 + (con llamadas de 0,2 s y plazo de 0,5 s, ningún candidato llegaba evaluado) +Con el dominio nuevo añadido, rojo por comportamiento: + CR3 test_cr3_a_configuration_error_does_not_wait_for_the_deadline → esperó 3,0 s (> 1,0) + CR3 test_cr3_cancelling_the_request_cancels_every_assessment → assert 0 == 3 + CR4 test_cr4_usage_of_a_failed_assessment_is_logged → api_calls=3 (se perdía el uso) + CR2 test_cr2_unassessed_candidates_are_logged_with_their_reason → sin razones en el log +Evaluador (API nueva: max_concurrency): + CR2 one_invalid_sample / cut_by_max_tokens, CR4 carries_the_usage, CR1 never_more_calls → FAILED +CR6 test_cr6_the_assessor_has_its_own_output_limit → FAILED (no existía el setting) +``` + +Verde de las correcciones: `10 passed` (CR1×2, CR2×3, CR3×2, CR4×2, CR6). CR5 con `grep`: ni +`get_candidate_assessor` ni `get_recommend_candidate_use_case` existen en 3a. + +### Verde + +| # | Resultado | Evidencia | +|---|---|---| +| A1 | ✅ | `test_selection.py` (18 tests): ordenación, desempate por `retrieval_score` y por `product_id`, filtro de requisitos, umbrales exactos con `>=` y todos por debajo | +| A2 | ✅ | `test_a2_never_returns_a_product_outside_the_candidates`; `test_a2_shuffled_input_gives_the_same_selection` (todas las permutaciones de 4); el mismo resultado con entrada barajada en `RecommendCandidate` | +| A3 | ✅ | `test_claude_candidate_assessor.py`: 3/3 → 1,0; 2/3 → 0,67; 1/3 → 0,33; una muestra inválida, o cortada por `max_tokens`, deja el candidato sin evaluar con `invalid_sample` (CR2); esquema estricto; sin `temperature`; `effort=low` | +| A4 | ✅ | 429, 500, 529 y la conexión dan `LLMUnavailableError`; 400 y 401 se propagan | +| A5 | ✅ | `test_recommend_candidate.py`: solo el top N por `retrieval_score`; el candidato fuera de plazo cuenta como no evaluado (`deadline`). `test_cr1_never_more_calls_in_flight_than_the_limit`: pico de llamadas = límite. `test_cr1_slow_calls_still_assess_some_candidates_before_the_deadline`: con llamadas lentas, llegan candidatos evaluados | +| A6 | ✅ | Un fallo transitorio salta solo ese candidato; si fallan todos → fallback ordenado por `retrieval_score` con `no_assessments`; un error de configuración se propaga | +| A7 | ✅ | Tests unitarios y de integración del relleno. **`make reindex` real** tras la migración, sobre la BD de desarrollo: `indexed=3 unchanged=0 deleted=0` (relleno) y después `indexed=0 unchanged=3 deleted=0`; las filas tienen `sku`, `name`, `price_cents` y `currency` | +| A8 | ✅ | `test_a8_a_reused_sku_does_not_break_the_reindex` (rojo por el orden → verde); `test_a8_sku_is_unique_in_the_index` contra Postgres real (`IntegrityError`) | +| A9 | ✅ | `test_a9_one_line_per_selection_with_ids_only`: una sola línea con `provider`, `model`, `candidates`, `assessed`, `api_calls`, tokens, `cost_usd`, `latency_ms`, `reason` y `product_id`; no aparecen SKU, nombre ni texto de la pregunta | +| A10 | ✅ | `grep` sin resultados: el dominio y la aplicación de `assistant` no importan `anthropic`, `sqlalchemy`, `fastapi`, `pgvector`, `fastembed` ni `httpx`. `catalog` sin cambios. Los e2e del PR 2 sin cambios. Ningún `type: ignore`, `noqa` ni `pylint: disable` nuevo. | +| A11 | ✅ local | `make lint`: ruff OK, pylint 10.00/10, mypy sin errores (80 ficheros). `make test`: **192 passed** (antes 137), sin avisos de excepciones sin recoger. _CI: pendiente de la PR._ | +| A12 | ✅ | Plan con la firma real de `select` y las correcciones CR1–CR6; README con el badge de 192 tests y el layout de los ficheros nuevos | + +### Tests nuevos o ampliados + +| Fichero | Tests | +|---|---| +| `tests/unit/contexts/assistant/domain/test_selection.py` | 18 (nuevo) | +| `tests/unit/contexts/assistant/application/test_recommend_candidate.py` | 12 (nuevo) | +| `tests/integration/contexts/assistant/infrastructure/test_claude_candidate_assessor.py` | 17 (nuevo) | +| `tests/integration/contexts/assistant/application/test_recommend_candidate_deadline.py` | 1 (nuevo, CR1) | +| `tests/unit/shared/infrastructure/test_settings.py` | 3 (+1, CR6) | +| `tests/unit/contexts/assistant/application/test_index_catalog.py` | 11 (+2) | +| `tests/unit/contexts/assistant/domain/test_product_document.py` | 7 (+1) | +| `tests/integration/contexts/assistant/infrastructure/test_sqlalchemy_documents.py` | 10 (+3) | + +### Fuera de lo verificado + +- **Llamadas reales a Anthropic:** en 3a no se hace ninguna. El evaluador se prueba con + `httpx2.MockTransport`. Las llamadas reales llegan con las evals del PR 3b, y antes de la primera se + informa del número de llamadas y del modelo. diff --git a/docs/step-18-assistant-agent.md b/docs/step-18-assistant-agent.md new file mode 100644 index 0000000..872fc53 --- /dev/null +++ b/docs/step-18-assistant-agent.md @@ -0,0 +1,58 @@ +# Paso 18 — Contexto `assistant`: agente con LangGraph (PR 3b) + +Plan: `docs/plan-assistant-genai.md`, sección «PR 3b — Agente con LangGraph». Se apoya en el PR 3a +(`docs/step-17-assistant-recommend.md`). Este documento se completa al empezar el PR 3b; de momento +recoge el alcance y los criterios ya decididos, para que no se pierdan con la división del PR 3. + +## Objetivo + +Que, con `assistant_mode=agent`, una pregunta como «¿cuál de estos me recomiendas para X?» acabe +de una de dos formas: + +- con **un producto recomendado**, a través de `recommend_candidate`; +- con un **«no hay una recomendación clara»** explícito y el top-k (nombre, SKU y precio). + +En ambos casos el agente queda acotado en pasos, tiempo y coste. Con `assistant_mode=rag`, el valor +por defecto, todo sigue igual que en el PR 2. + +## Qué + +1. Agente LangGraph detrás del puerto `CatalogAgent`, con las herramientas `semantic_search`, + `get_product_by_sku` y `recommend_candidate`. +2. Como mucho 5 llamadas a herramientas por ejecución (`recommend_candidate` cuenta como una). Al + llegar al límite, el agente responde en modo RAG con lo recuperado y registra la razón + `max_steps`. +3. El LLM del agente va por `langchain-anthropic`. Sus errores se traducen a `LLMUnavailableError` + con la clasificación del PR 1, y se aplica el mismo timeout. +4. `assistant_mode` (`rag` | `agent`), con `rag` por defecto. +5. Evals con el marker `eval`, fuera del CI. + +## Decisiones tomadas (con el usuario) + +| Tema | Decisión | +|---|---| +| Límite de pasos | 5 llamadas a herramientas. Al llegar, respuesta en modo RAG y razón `max_steps` en el log. | +| `assistant_mode` | `rag` por defecto; los e2e del PR 2 no cambian. | +| Sin recomendación | «No hay una recomendación clara» y el top-k con nombre, SKU y precio, sin pregunta de seguimiento. | +| LLM del agente | Errores de `langchain-anthropic` traducidos con la misma clasificación que el PR 1, con el mismo timeout. | +| Clave para las evals | El usuario pone `ANTHROPIC_API_KEY` en `.env`. **Su valor nunca aparece en logs, salidas de comandos ni documentos.** | +| Primera ejecución real de evals | **Antes de lanzarla se informa al usuario del número de llamadas y del modelo** que usará. | +| Cliente de Anthropic | Los proveedores de `RecommendCandidate` y de `ClaudeCandidateAssessor` se crean aquí y reutilizan el único cliente de Anthropic del asistente, que se cierra en el lifespan (decidido en la code review del 3a). | + +## Criterios de aceptación + +| # | Criterio | Cómo se demuestra | +|---|---|---| +| B1 | Sin recomendación, la respuesta dice que no hay una clara y lista el top-k con nombre, SKU y precio | Test del agente | +| B2 | Al llegar a la 5.ª llamada a herramienta se corta, responde en modo RAG y el log registra `max_steps` | Test del grafo con un modelo falso guionizado | +| B3 | `assistant_mode=rag` es el valor por defecto, y los e2e del PR 2 pasan **sin modificarse** | `git diff` vacío en esos tests y suite en verde | +| B4 | Los errores del LLM del agente se clasifican como en el PR 1: los transitorios llevan al fallback y los de configuración se propagan | Test del adaptador del agente con errores simulados | +| B5 | El dominio y la aplicación de `assistant` no importan `langgraph` ni `langchain` | `grep` | +| B6 | Las evals existen con el marker `eval` y el CI por defecto no las ejecuta | Recolección de pytest y CI de la PR | +| B7 | Antes de la primera ejecución real se informa del número de llamadas, y el valor de la clave no aparece en ninguna salida | Mensaje al usuario antes de ejecutar, y revisión de las salidas | +| B8 | Los quality gates y el CI pasan | `make lint`, `make test` y CI de la PR | +| B9 | El plan y el README coinciden con el código | Revisión, anclas y badges | + +## Evidencia + +_Se completa al terminar._ diff --git a/src/contexts/assistant/application/dtos.py b/src/contexts/assistant/application/dtos.py index 15753d4..8608805 100644 --- a/src/contexts/assistant/application/dtos.py +++ b/src/contexts/assistant/application/dtos.py @@ -3,8 +3,10 @@ from dataclasses import dataclass from uuid import UUID +from contexts.assistant.domain.candidate import Candidate from contexts.assistant.domain.completion import Completion from contexts.assistant.domain.retrieved_document import RetrievedDocument +from contexts.assistant.domain.selection import SelectionReason @dataclass(frozen=True, slots=True) @@ -62,3 +64,34 @@ class IndexReportDTO: indexed: int unchanged: int deleted: int + + +@dataclass(frozen=True, slots=True) +class CandidateDTO: + product_id: UUID + sku: str + name: str + price_cents: int + currency: str + retrieval_score: float + + @classmethod + def from_candidate(cls, candidate: Candidate) -> CandidateDTO: + return cls( + product_id=candidate.product_id, + sku=candidate.sku, + name=candidate.name, + price_cents=candidate.price_cents, + currency=candidate.currency, + retrieval_score=candidate.retrieval_score, + ) + + +@dataclass(frozen=True, slots=True) +class RecommendationDTO: + product_id: UUID | None + score: float | None + reason: SelectionReason + fallback_used: bool + candidates: tuple[CandidateDTO, ...] + assessed: int diff --git a/src/contexts/assistant/application/use_cases/index_catalog.py b/src/contexts/assistant/application/use_cases/index_catalog.py index 582f4ef..e932cfa 100644 --- a/src/contexts/assistant/application/use_cases/index_catalog.py +++ b/src/contexts/assistant/application/use_cases/index_catalog.py @@ -29,17 +29,17 @@ async def execute(self) -> IndexReportDTO: ] stored_hashes = await self._document_index.content_hashes() + unlisted_ids = sorted(set(stored_hashes) - {doc.product_id for doc in documents}) + orphan_ids = await self._catalog_reader.missing(unlisted_ids) if unlisted_ids else [] + if orphan_ids: + await self._document_index.delete(orphan_ids) + changed = [ doc for doc in documents if stored_hashes.get(doc.product_id) != doc.content_hash ] for start in range(0, len(changed), self._batch_size): await self._index_batch(changed[start : start + self._batch_size]) - unlisted_ids = sorted(set(stored_hashes) - {doc.product_id for doc in documents}) - orphan_ids = await self._catalog_reader.missing(unlisted_ids) if unlisted_ids else [] - if orphan_ids: - await self._document_index.delete(orphan_ids) - return IndexReportDTO( indexed=len(changed), unchanged=len(documents) - len(changed), diff --git a/src/contexts/assistant/application/use_cases/recommend_candidate.py b/src/contexts/assistant/application/use_cases/recommend_candidate.py new file mode 100644 index 0000000..bdae713 --- /dev/null +++ b/src/contexts/assistant/application/use_cases/recommend_candidate.py @@ -0,0 +1,128 @@ +import asyncio +import logging +import time +from collections.abc import Sequence + +from contexts.assistant.application.dtos import CandidateDTO, RecommendationDTO +from contexts.assistant.domain.candidate import Assessment, AssessmentFailure, Candidate +from contexts.assistant.domain.exceptions import AssessmentUnavailableError, LLMUnavailableError +from contexts.assistant.domain.ports.candidate_assessor import CandidateAssessor +from contexts.assistant.domain.selection import Selection, select + +logger = logging.getLogger(__name__) + + +class RecommendCandidate: + def __init__( + self, + assessor: CandidateAssessor, + *, + top_n: int, + timeout_s: float, + requirements_threshold: float, + min_score: float, + ) -> None: + self._assessor = assessor + self._top_n = top_n + self._timeout_s = timeout_s + self._requirements_threshold = requirements_threshold + self._min_score = min_score + + async def execute(self, question: str, candidates: Sequence[Candidate]) -> RecommendationDTO: + started = time.perf_counter() + top = sorted(candidates, key=lambda c: (-c.retrieval_score, c.product_id))[: self._top_n] + assessments = await self._assess_all(question, top) + selection = select(top, assessments, self._requirements_threshold, self._min_score) + self._log(top, assessments, selection, started) + return RecommendationDTO( + product_id=selection.product_id, + score=selection.score, + reason=selection.reason, + fallback_used=selection.fallback_used, + candidates=tuple(CandidateDTO.from_candidate(candidate) for candidate in top), + assessed=_assessed(assessments), + ) + + async def _assess_all( + self, question: str, candidates: Sequence[Candidate] + ) -> list[Assessment]: + if not candidates: + return [] + tasks = { + asyncio.create_task(self._assess_one(question, candidate)): candidate + for candidate in candidates + } + try: + done, _ = await asyncio.wait( + tasks, timeout=self._timeout_s, return_when=asyncio.FIRST_EXCEPTION + ) + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + + finished = [task for task in tasks if task in done] + errors = [error for task in finished if (error := task.exception()) is not None] + if errors: + raise errors[0] + return [ + task.result() + if task in done + else Assessment( + product_id=candidate.product_id, + answers=None, + failure=AssessmentFailure.DEADLINE, + ) + for task, candidate in tasks.items() + ] + + async def _assess_one(self, question: str, candidate: Candidate) -> Assessment: + try: + return await self._assessor.assess(question, candidate) + except AssessmentUnavailableError as error: + return Assessment( + product_id=candidate.product_id, + answers=None, + usage=error.usage, + failure=AssessmentFailure.ERROR, + ) + except LLMUnavailableError: + return Assessment( + product_id=candidate.product_id, answers=None, failure=AssessmentFailure.ERROR + ) + + def _log( + self, + candidates: Sequence[Candidate], + assessments: Sequence[Assessment], + selection: Selection, + started: float, + ) -> None: + costs = [a.usage.cost_usd for a in assessments] + cost_usd = sum(c for c in costs if c is not None) if costs and None not in costs else None + unassessed = ",".join( + f"{a.product_id}:{a.failure.value}" + for a in sorted(assessments, key=lambda a: a.product_id) + if a.answers is None and a.failure is not None + ) + logger.info( + "candidate_selection provider=%s model=%s candidates=%d assessed=%d api_calls=%d " + "input_tokens=%d output_tokens=%d cost_usd=%s latency_ms=%d reason=%s product_id=%s " + "unassessed=%s", + self._assessor.provider, + self._assessor.model_name, + len(candidates), + _assessed(assessments), + sum(a.usage.api_calls for a in assessments), + sum(a.usage.input_tokens for a in assessments), + sum(a.usage.output_tokens for a in assessments), + cost_usd, + round((time.perf_counter() - started) * 1000), + selection.reason.value, + selection.product_id, + unassessed, + ) + + +def _assessed(assessments: Sequence[Assessment]) -> int: + return sum(1 for assessment in assessments if assessment.answers is not None) diff --git a/src/contexts/assistant/domain/assessment_usage.py b/src/contexts/assistant/domain/assessment_usage.py new file mode 100644 index 0000000..ac2ce82 --- /dev/null +++ b/src/contexts/assistant/domain/assessment_usage.py @@ -0,0 +1,9 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True, slots=True) +class AssessmentUsage: + api_calls: int = 0 + input_tokens: int = 0 + output_tokens: int = 0 + cost_usd: float | None = None diff --git a/src/contexts/assistant/domain/candidate.py b/src/contexts/assistant/domain/candidate.py new file mode 100644 index 0000000..9215cf7 --- /dev/null +++ b/src/contexts/assistant/domain/candidate.py @@ -0,0 +1,51 @@ +from dataclasses import dataclass, field +from enum import Enum +from uuid import UUID + +from contexts.assistant.domain.assessment_usage import AssessmentUsage +from contexts.assistant.domain.exceptions import InvalidAssessmentError + +__all__ = [ + "Assessment", + "AssessmentFailure", + "AssessmentUsage", + "Candidate", + "RubricAnswers", +] + + +@dataclass(frozen=True, slots=True) +class Candidate: + product_id: UUID + sku: str + name: str + price_cents: int + currency: str + content: str + retrieval_score: float + + +@dataclass(frozen=True, slots=True) +class RubricAnswers: + meets_requirements: float + suitable_for_use: float + + def __post_init__(self) -> None: + for name in ("meets_requirements", "suitable_for_use"): + value = getattr(self, name) + if not 0.0 <= value <= 1.0: + raise InvalidAssessmentError(f"{name} must be a probability in [0, 1]") + + +class AssessmentFailure(str, Enum): + INVALID_SAMPLE = "invalid_sample" + ERROR = "error" + DEADLINE = "deadline" + + +@dataclass(frozen=True, slots=True) +class Assessment: + product_id: UUID + answers: RubricAnswers | None + usage: AssessmentUsage = field(default_factory=AssessmentUsage) + failure: AssessmentFailure | None = None diff --git a/src/contexts/assistant/domain/exceptions.py b/src/contexts/assistant/domain/exceptions.py index f8c1125..bba0ee1 100644 --- a/src/contexts/assistant/domain/exceptions.py +++ b/src/contexts/assistant/domain/exceptions.py @@ -1,4 +1,5 @@ -from shared.domain.errors import InvalidInputError, ServiceUnavailableError +from contexts.assistant.domain.assessment_usage import AssessmentUsage +from shared.domain.errors import DomainError, InvalidInputError, ServiceUnavailableError class InvalidQuestionError(InvalidInputError): @@ -7,3 +8,13 @@ class InvalidQuestionError(InvalidInputError): class LLMUnavailableError(ServiceUnavailableError): pass + + +class InvalidAssessmentError(DomainError): + pass + + +class AssessmentUnavailableError(LLMUnavailableError): + def __init__(self, usage: AssessmentUsage) -> None: + super().__init__("the language model is unavailable") + self.usage = usage diff --git a/src/contexts/assistant/domain/ports/candidate_assessor.py b/src/contexts/assistant/domain/ports/candidate_assessor.py new file mode 100644 index 0000000..aa0a381 --- /dev/null +++ b/src/contexts/assistant/domain/ports/candidate_assessor.py @@ -0,0 +1,21 @@ +from abc import ABC, abstractmethod + +from contexts.assistant.domain.candidate import Assessment, Candidate + + +class CandidateAssessor(ABC): + @property + @abstractmethod + def provider(self) -> str: ... + + @property + @abstractmethod + def model_name(self) -> str: ... + + @abstractmethod + async def assess(self, question: str, candidate: Candidate) -> Assessment: + """Assesses one candidate against the rubric; it never sees the other candidates. + Answers are probabilities of yes in [0, 1]. Returns an ``Assessment`` with + ``answers=None`` when the model's output cannot be used. Raises + ``LLMUnavailableError`` only for transient failures; configuration errors propagate. + """ diff --git a/src/contexts/assistant/domain/ports/document_index.py b/src/contexts/assistant/domain/ports/document_index.py index b16a837..3a77f0f 100644 --- a/src/contexts/assistant/domain/ports/document_index.py +++ b/src/contexts/assistant/domain/ports/document_index.py @@ -6,7 +6,10 @@ class DocumentIndex(ABC): @abstractmethod - async def content_hashes(self) -> dict[UUID, str]: ... + async def content_hashes(self) -> dict[UUID, str | None]: + """Maps every indexed ``product_id`` to its ``content_hash``. ``None`` marks a document + that lacks catalog fields (indexed before they existed): it must be indexed again. + """ @abstractmethod async def upsert(self, documents: list[EmbeddedDocument]) -> None: diff --git a/src/contexts/assistant/domain/product_document.py b/src/contexts/assistant/domain/product_document.py index 074e62b..54dc63a 100644 --- a/src/contexts/assistant/domain/product_document.py +++ b/src/contexts/assistant/domain/product_document.py @@ -10,6 +10,10 @@ @dataclass(frozen=True, slots=True) class ProductDocument: product_id: UUID + sku: str + name: str + price_cents: int + currency: str content: str content_hash: str @@ -21,6 +25,10 @@ def from_catalog_product( fingerprint = f"{embedding_model}\n{content}" return cls( product_id=product.product_id, + sku=product.sku, + name=product.name, + price_cents=product.price_cents, + currency=product.currency, content=content, content_hash=hashlib.sha256(fingerprint.encode("utf-8")).hexdigest(), ) diff --git a/src/contexts/assistant/domain/rubric.py b/src/contexts/assistant/domain/rubric.py new file mode 100644 index 0000000..d48cc05 --- /dev/null +++ b/src/contexts/assistant/domain/rubric.py @@ -0,0 +1,29 @@ +from dataclasses import dataclass + + +@dataclass(frozen=True, slots=True) +class RubricQuestion: + id: str + instructions: str + yes_means: str + no_means: str + + +MEETS_REQUIREMENTS = RubricQuestion( + id="meets_requirements", + instructions="Does the product meet the explicit requirements stated in the question?", + yes_means=( + "The product satisfies every requirement the question states explicitly. " + "If the question states no explicit requirements, answer yes." + ), + no_means="The product fails at least one requirement the question states explicitly.", +) + +SUITABLE_FOR_USE = RubricQuestion( + id="suitable_for_use", + instructions="Is the product suitable for the use the question asks about?", + yes_means="The product is a good fit for that use.", + no_means="The product is not a good fit for that use.", +) + +RUBRIC = (MEETS_REQUIREMENTS, SUITABLE_FOR_USE) diff --git a/src/contexts/assistant/domain/selection.py b/src/contexts/assistant/domain/selection.py new file mode 100644 index 0000000..6282e78 --- /dev/null +++ b/src/contexts/assistant/domain/selection.py @@ -0,0 +1,60 @@ +from __future__ import annotations + +from collections.abc import Sequence +from dataclasses import dataclass +from enum import Enum +from uuid import UUID + +from contexts.assistant.domain.candidate import Assessment, Candidate, RubricAnswers + + +class SelectionReason(str, Enum): + CHOSEN = "chosen" + BELOW_THRESHOLD = "below_threshold" + NO_ASSESSMENTS = "no_assessments" + + +@dataclass(frozen=True, slots=True) +class Selection: + product_id: UUID | None + score: float | None + fallback_used: bool + reason: SelectionReason + + +def select( + candidates: Sequence[Candidate], + assessments: Sequence[Assessment], + requirements_threshold: float, + min_score: float, +) -> Selection: + by_id = {candidate.product_id: candidate for candidate in candidates} + answered = [ + (by_id[assessment.product_id], assessment.answers) + for assessment in assessments + if assessment.product_id in by_id and assessment.answers is not None + ] + if not answered: + return Selection(None, None, fallback_used=True, reason=SelectionReason.NO_ASSESSMENTS) + + eligible = [ + (candidate, answers) + for candidate, answers in answered + if answers.meets_requirements >= requirements_threshold + ] + ranked = sorted(eligible, key=_ranking_key) + if not ranked or ranked[0][1].suitable_for_use < min_score: + return Selection(None, None, fallback_used=False, reason=SelectionReason.BELOW_THRESHOLD) + + best, answers = ranked[0] + return Selection( + product_id=best.product_id, + score=answers.suitable_for_use, + fallback_used=False, + reason=SelectionReason.CHOSEN, + ) + + +def _ranking_key(pair: tuple[Candidate, RubricAnswers]) -> tuple[float, float, UUID]: + candidate, answers = pair + return (-answers.suitable_for_use, -candidate.retrieval_score, candidate.product_id) diff --git a/src/contexts/assistant/infrastructure/llm/anthropic_errors.py b/src/contexts/assistant/infrastructure/llm/anthropic_errors.py new file mode 100644 index 0000000..0397bc0 --- /dev/null +++ b/src/contexts/assistant/infrastructure/llm/anthropic_errors.py @@ -0,0 +1,20 @@ +from collections.abc import Iterator +from contextlib import contextmanager + +from anthropic import APIConnectionError, APIStatusError, RateLimitError + +from contexts.assistant.domain.exceptions import LLMUnavailableError + +UNAVAILABLE_MESSAGE = "the language model is unavailable" + + +@contextmanager +def transient_errors_as_unavailable() -> Iterator[None]: + try: + yield + except (APIConnectionError, RateLimitError) as exc: + raise LLMUnavailableError(UNAVAILABLE_MESSAGE) from exc + except APIStatusError as exc: + if exc.status_code < 500: + raise + raise LLMUnavailableError(UNAVAILABLE_MESSAGE) from exc diff --git a/src/contexts/assistant/infrastructure/llm/anthropic_llm_client.py b/src/contexts/assistant/infrastructure/llm/anthropic_llm_client.py index 2e4977d..148a444 100644 --- a/src/contexts/assistant/infrastructure/llm/anthropic_llm_client.py +++ b/src/contexts/assistant/infrastructure/llm/anthropic_llm_client.py @@ -1,13 +1,11 @@ import time -from anthropic import APIConnectionError, APIStatusError, AsyncAnthropic, RateLimitError +from anthropic import AsyncAnthropic from anthropic.types import TextBlock from contexts.assistant.domain.completion import Completion -from contexts.assistant.domain.exceptions import LLMUnavailableError from contexts.assistant.domain.ports.llm_client import LLMClient - -_UNAVAILABLE_MESSAGE = "the language model is unavailable" +from contexts.assistant.infrastructure.llm.anthropic_errors import transient_errors_as_unavailable class AnthropicLLMClient(LLMClient): @@ -19,19 +17,13 @@ def __init__( async def complete(self, *, system: str, prompt: str, max_tokens: int) -> Completion: started = time.perf_counter() - try: + with transient_errors_as_unavailable(): message = await self._client.messages.create( model=self._model, max_tokens=max_tokens, system=system, messages=[{"role": "user", "content": prompt}], ) - except (APIConnectionError, RateLimitError) as exc: - raise LLMUnavailableError(_UNAVAILABLE_MESSAGE) from exc - except APIStatusError as exc: - if exc.status_code < 500: - raise - raise LLMUnavailableError(_UNAVAILABLE_MESSAGE) from exc latency_ms = round((time.perf_counter() - started) * 1000) return Completion( diff --git a/src/contexts/assistant/infrastructure/llm/claude_candidate_assessor.py b/src/contexts/assistant/infrastructure/llm/claude_candidate_assessor.py new file mode 100644 index 0000000..14537fa --- /dev/null +++ b/src/contexts/assistant/infrastructure/llm/claude_candidate_assessor.py @@ -0,0 +1,137 @@ +import asyncio +import json +from dataclasses import dataclass + +from anthropic import AsyncAnthropic +from anthropic.types import Message, OutputConfigParam, TextBlock + +from contexts.assistant.domain.candidate import ( + Assessment, + AssessmentFailure, + AssessmentUsage, + Candidate, + RubricAnswers, +) +from contexts.assistant.domain.exceptions import AssessmentUnavailableError, LLMUnavailableError +from contexts.assistant.domain.ports.candidate_assessor import CandidateAssessor +from contexts.assistant.domain.rubric import RUBRIC +from contexts.assistant.infrastructure.llm.anthropic_errors import transient_errors_as_unavailable + +PROVIDER = "claude" + +SYSTEM_PROMPT = ( + "You assess one catalog product against a question. Answer each rubric question with " + "true (yes) or false (no), using only the product data given.\n\n" + + "\n".join( + f"- {question.id}: {question.instructions} " + f"Yes means: {question.yes_means} No means: {question.no_means}" + for question in RUBRIC + ) +) + +OUTPUT_SCHEMA: dict[str, object] = { + "type": "object", + "properties": {question.id: {"type": "boolean"} for question in RUBRIC}, + "required": [question.id for question in RUBRIC], + "additionalProperties": False, +} + +OUTPUT_CONFIG: OutputConfigParam = { + "effort": "low", + "format": {"type": "json_schema", "schema": OUTPUT_SCHEMA}, +} + + +@dataclass(frozen=True, slots=True) +class _Verdict: + meets_requirements: bool + suitable_for_use: bool + + +class ClaudeCandidateAssessor(CandidateAssessor): + def __init__( + self, + *, + client: AsyncAnthropic, + model: str, + timeout_s: float, + max_retries: int, + samples: int, + max_tokens: int, + max_concurrency: int, + ) -> None: + self._client = client.with_options(timeout=timeout_s, max_retries=max_retries) + self._model = model + self._samples = samples + self._max_tokens = max_tokens + self._calls = asyncio.Semaphore(max_concurrency) + + @property + def provider(self) -> str: + return PROVIDER + + @property + def model_name(self) -> str: + return self._model + + async def assess(self, question: str, candidate: Candidate) -> Assessment: + outcomes = await asyncio.gather( + *(self._ask(question, candidate) for _ in range(self._samples)), + return_exceptions=True, + ) + messages = [outcome for outcome in outcomes if isinstance(outcome, Message)] + usage = AssessmentUsage( + api_calls=self._samples, + input_tokens=sum(message.usage.input_tokens for message in messages), + output_tokens=sum(message.usage.output_tokens for message in messages), + ) + + errors = [outcome for outcome in outcomes if isinstance(outcome, BaseException)] + fatal = [error for error in errors if not isinstance(error, LLMUnavailableError)] + if fatal: + raise fatal[0] + if errors: + raise AssessmentUnavailableError(usage) from errors[0] + + verdicts = [_parse(message) for message in messages] + if any(verdict is None for verdict in verdicts): + return Assessment( + product_id=candidate.product_id, + answers=None, + usage=usage, + failure=AssessmentFailure.INVALID_SAMPLE, + ) + valid = [verdict for verdict in verdicts if verdict is not None] + answers = RubricAnswers( + meets_requirements=sum(v.meets_requirements for v in valid) / self._samples, + suitable_for_use=sum(v.suitable_for_use for v in valid) / self._samples, + ) + return Assessment(product_id=candidate.product_id, answers=answers, usage=usage) + + async def _ask(self, question: str, candidate: Candidate) -> Message: + state = json.dumps({"question": question, "product": candidate.content}) + async with self._calls: + with transient_errors_as_unavailable(): + return await self._client.messages.create( + model=self._model, + max_tokens=self._max_tokens, + system=SYSTEM_PROMPT, + messages=[{"role": "user", "content": state}], + output_config=OUTPUT_CONFIG, + ) + + +def _parse(message: Message) -> _Verdict | None: + if message.stop_reason != "end_turn": + return None + text = "".join(block.text for block in message.content if isinstance(block, TextBlock)) + try: + data = json.loads(text) + except json.JSONDecodeError: + return None + if not isinstance(data, dict) or set(data) != {question.id for question in RUBRIC}: + return None + meets, suitable = data["meets_requirements"], data["suitable_for_use"] + if not isinstance(meets, bool) or not isinstance(suitable, bool): + return None + return _Verdict(meets_requirements=meets, suitable_for_use=suitable) diff --git a/src/contexts/assistant/infrastructure/persistence/models.py b/src/contexts/assistant/infrastructure/persistence/models.py index a8a112e..2c3a734 100644 --- a/src/contexts/assistant/infrastructure/persistence/models.py +++ b/src/contexts/assistant/infrastructure/persistence/models.py @@ -2,7 +2,7 @@ from uuid import UUID from pgvector.sqlalchemy import Vector -from sqlalchemy import DateTime, Index, String, Text +from sqlalchemy import DateTime, Index, Integer, String, Text from sqlalchemy.orm import Mapped, mapped_column from shared.infrastructure.db_base import Base @@ -23,6 +23,10 @@ class ProductDocumentModel(Base): ) product_id: Mapped[UUID] = mapped_column(primary_key=True) + sku: Mapped[str | None] = mapped_column(String(64), unique=True, index=True, nullable=True) + name: Mapped[str | None] = mapped_column(String(200), nullable=True) + price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True) + currency: Mapped[str | None] = mapped_column(String(3), nullable=True) content: Mapped[str] = mapped_column(Text, nullable=False) content_hash: Mapped[str] = mapped_column(String(64), nullable=False) embedding: Mapped[list[float]] = mapped_column(Vector(EMBEDDING_DIMENSIONS), nullable=False) diff --git a/src/contexts/assistant/infrastructure/persistence/sqlalchemy_document_index.py b/src/contexts/assistant/infrastructure/persistence/sqlalchemy_document_index.py index 3176fcd..9ba3ee8 100644 --- a/src/contexts/assistant/infrastructure/persistence/sqlalchemy_document_index.py +++ b/src/contexts/assistant/infrastructure/persistence/sqlalchemy_document_index.py @@ -1,7 +1,7 @@ from datetime import datetime, timezone from uuid import UUID -from sqlalchemy import delete, select +from sqlalchemy import case, delete, or_, select from sqlalchemy.dialects.postgresql import insert from sqlalchemy.ext.asyncio import AsyncSession @@ -14,10 +14,16 @@ class SqlAlchemyDocumentIndex(DocumentIndex): def __init__(self, session: AsyncSession) -> None: self._session = session - async def content_hashes(self) -> dict[UUID, str]: - result = await self._session.execute( - select(ProductDocumentModel.product_id, ProductDocumentModel.content_hash) + async def content_hashes(self) -> dict[UUID, str | None]: + model = ProductDocumentModel + missing_catalog_column = or_( + model.sku.is_(None), + model.name.is_(None), + model.price_cents.is_(None), + model.currency.is_(None), ) + stored_hash = case((missing_catalog_column, None), else_=model.content_hash) + result = await self._session.execute(select(model.product_id, stored_hash)) return dict(result.all()) async def upsert(self, documents: list[EmbeddedDocument]) -> None: @@ -28,6 +34,10 @@ async def upsert(self, documents: list[EmbeddedDocument]) -> None: [ { "product_id": item.document.product_id, + "sku": item.document.sku, + "name": item.document.name, + "price_cents": item.document.price_cents, + "currency": item.document.currency, "content": item.document.content, "content_hash": item.document.content_hash, "embedding": item.embedding, @@ -39,6 +49,10 @@ async def upsert(self, documents: list[EmbeddedDocument]) -> None: stmt = stmt.on_conflict_do_update( index_elements=[ProductDocumentModel.product_id], set_={ + "sku": stmt.excluded.sku, + "name": stmt.excluded.name, + "price_cents": stmt.excluded.price_cents, + "currency": stmt.excluded.currency, "content": stmt.excluded.content, "content_hash": stmt.excluded.content_hash, "embedding": stmt.excluded.embedding, diff --git a/src/shared/infrastructure/settings.py b/src/shared/infrastructure/settings.py index f401a63..a193332 100644 --- a/src/shared/infrastructure/settings.py +++ b/src/shared/infrastructure/settings.py @@ -1,4 +1,5 @@ from functools import lru_cache +from typing import Literal from pydantic import Field, SecretStr from pydantic_settings import BaseSettings, SettingsConfigDict @@ -63,6 +64,37 @@ class Settings(BaseSettings): description="Only load the model from the cache, never download it. " "The Docker image bakes the model in and sets this to true.", ) + selector_provider: Literal["claude"] = Field( + default="claude", + description="Candidate assessor used by the recommendation tool.", + ) + selector_top_n: int = Field( + default=10, ge=1, description="Retrieval results passed to the candidate selector." + ) + selector_max_concurrency: int = Field( + default=10, ge=1, description="Assessor API calls in flight at the same time." + ) + selector_timeout_s: float = Field( + default=10.0, gt=0, description="Overall deadline for assessing all candidates." + ) + selector_samples: int = Field( + default=3, ge=1, description="Self-consistency samples per candidate (Claude assessor)." + ) + selector_max_tokens: int = Field( + default=1024, ge=1, description="Output token limit per assessor call." + ) + selector_requirements_threshold: float = Field( + default=0.5, + ge=0, + le=1, + description="Minimum probability of 'meets the explicit requirements' to stay eligible.", + ) + selector_min_score: float = Field( + default=0.5, + ge=0, + le=1, + description="Minimum probability of 'suitable for the use' to recommend a candidate.", + ) rag_top_k: int = Field( default=5, gt=0, diff --git a/tests/fakes/candidate_assessor.py b/tests/fakes/candidate_assessor.py new file mode 100644 index 0000000..696184c --- /dev/null +++ b/tests/fakes/candidate_assessor.py @@ -0,0 +1,59 @@ +import asyncio +from uuid import UUID + +from contexts.assistant.domain.candidate import ( + Assessment, + AssessmentFailure, + AssessmentUsage, + Candidate, + RubricAnswers, +) +from contexts.assistant.domain.ports.candidate_assessor import CandidateAssessor + +Outcome = RubricAnswers | Exception | None + + +class FakeCandidateAssessor(CandidateAssessor): + def __init__( + self, + outcomes: dict[UUID, Outcome] | None = None, + delays: dict[UUID, float] | None = None, + default: Outcome = None, + ) -> None: + self._outcomes = outcomes or {} + self._delays = delays or {} + self._default = default + self.assessed: list[UUID] = [] + self.in_flight = 0 + self.max_in_flight = 0 + self.cancelled = 0 + + @property + def provider(self) -> str: + return "fake" + + @property + def model_name(self) -> str: + return "fake-assessor" + + async def assess(self, question: str, candidate: Candidate) -> Assessment: + self.assessed.append(candidate.product_id) + self.in_flight += 1 + self.max_in_flight = max(self.max_in_flight, self.in_flight) + try: + await asyncio.sleep(self._delays.get(candidate.product_id, 0.01)) + except asyncio.CancelledError: + self.cancelled += 1 + self.in_flight -= 1 + raise + try: + outcome = self._outcomes.get(candidate.product_id, self._default) + if isinstance(outcome, Exception): + raise outcome + usage = AssessmentUsage(api_calls=3, input_tokens=100, output_tokens=10) + failure = AssessmentFailure.INVALID_SAMPLE if outcome is None else None + return Assessment( + product_id=candidate.product_id, answers=outcome, usage=usage, failure=failure + ) + finally: + self.in_flight -= 1 diff --git a/tests/fakes/documents.py b/tests/fakes/documents.py index 5e76443..615f7f9 100644 --- a/tests/fakes/documents.py +++ b/tests/fakes/documents.py @@ -25,18 +25,38 @@ async def missing(self, product_ids: list[UUID]) -> list[UUID]: return [pid for pid in product_ids if pid not in existing] +class DuplicateSkuInIndexError(Exception): + pass + + class FakeDocumentIndex(DocumentIndex): def __init__(self) -> None: self.documents: dict[UUID, EmbeddedDocument] = {} + self.incomplete: set[UUID] = set() self.upsert_sizes: list[int] = [] - async def content_hashes(self) -> dict[UUID, str]: - return {pid: item.document.content_hash for pid, item in self.documents.items()} + async def content_hashes(self) -> dict[UUID, str | None]: + return { + pid: None if pid in self.incomplete else item.document.content_hash + for pid, item in self.documents.items() + } async def upsert(self, documents: list[EmbeddedDocument]) -> None: self.upsert_sizes.append(len(documents)) for item in documents: + owner = next( + ( + pid + for pid, stored in self.documents.items() + if stored.document.sku == item.document.sku + and pid != item.document.product_id + ), + None, + ) + if owner is not None: + raise DuplicateSkuInIndexError(item.document.sku) self.documents[item.document.product_id] = item + self.incomplete.discard(item.document.product_id) async def delete(self, product_ids: list[UUID]) -> None: for product_id in product_ids: diff --git a/tests/integration/contexts/assistant/application/test_recommend_candidate_deadline.py b/tests/integration/contexts/assistant/application/test_recommend_candidate_deadline.py new file mode 100644 index 0000000..ccc336b --- /dev/null +++ b/tests/integration/contexts/assistant/application/test_recommend_candidate_deadline.py @@ -0,0 +1,80 @@ +import asyncio +import json +from uuid import uuid4 + +import anthropic +import httpx2 + +from contexts.assistant.application.use_cases.recommend_candidate import RecommendCandidate +from contexts.assistant.domain.candidate import Candidate +from contexts.assistant.infrastructure.llm.claude_candidate_assessor import ( + ClaudeCandidateAssessor, +) + +CALL_SECONDS = 0.2 + + +async def _slow_yes(_request: httpx2.Request) -> httpx2.Response: + await asyncio.sleep(CALL_SECONDS) + return httpx2.Response( + 200, + json={ + "id": "msg_test", + "type": "message", + "role": "assistant", + "model": "claude-test", + "content": [ + { + "type": "text", + "text": json.dumps({"meets_requirements": True, "suitable_for_use": True}), + } + ], + "stop_reason": "end_turn", + "stop_sequence": None, + "usage": {"input_tokens": 40, "output_tokens": 10}, + }, + ) + + +def _candidates(count: int) -> list[Candidate]: + return [ + Candidate( + product_id=uuid4(), + sku=f"SKU-{i}", + name=f"Name {i}", + price_cents=100, + currency="EUR", + content=f"SKU: SKU-{i}", + retrieval_score=1.0 - i / 10, + ) + for i in range(count) + ] + + +async def test_cr1_slow_calls_still_assess_some_candidates_before_the_deadline() -> None: + client = anthropic.AsyncAnthropic( + api_key="test-key", + max_retries=0, + http_client=anthropic.DefaultAsyncHttpxClient(transport=httpx2.MockTransport(_slow_yes)), + ) + assessor = ClaudeCandidateAssessor( + client=client, + model="claude-test", + timeout_s=5.0, + max_retries=0, + samples=3, + max_tokens=64, + max_concurrency=3, + ) + use_case = RecommendCandidate( + assessor, + top_n=10, + timeout_s=CALL_SECONDS * 2.5, + requirements_threshold=0.5, + min_score=0.5, + ) + + result = await use_case.execute("Which do you recommend for hiking?", _candidates(3)) + + assert result.assessed >= 1 + assert result.product_id is not None diff --git a/tests/integration/contexts/assistant/infrastructure/test_claude_candidate_assessor.py b/tests/integration/contexts/assistant/infrastructure/test_claude_candidate_assessor.py new file mode 100644 index 0000000..f2869a9 --- /dev/null +++ b/tests/integration/contexts/assistant/infrastructure/test_claude_candidate_assessor.py @@ -0,0 +1,228 @@ +import asyncio +import json +from collections.abc import Awaitable, Callable +from dataclasses import replace +from uuid import uuid4 + +import anthropic +import httpx2 +import pytest + +from contexts.assistant.domain.candidate import AssessmentFailure, Candidate +from contexts.assistant.domain.exceptions import AssessmentUnavailableError, LLMUnavailableError +from contexts.assistant.domain.rubric import RUBRIC +from contexts.assistant.infrastructure.llm.claude_candidate_assessor import ( + ClaudeCandidateAssessor, +) + +QUESTION = "Which of these do you recommend for hiking?" +CANDIDATE = Candidate( + product_id=uuid4(), + sku="BOOT-1", + name="Trail boot", + price_cents=8900, + currency="EUR", + content="SKU: BOOT-1\nName: Trail boot\nPrice: 89.00 EUR", + retrieval_score=0.8, +) + +Handler = Callable[[httpx2.Request], httpx2.Response] +AsyncHandler = Callable[[httpx2.Request], Awaitable[httpx2.Response]] + + +def _message(text: str, stop_reason: str = "end_turn") -> httpx2.Response: + return httpx2.Response( + 200, + json={ + "id": "msg_test", + "type": "message", + "role": "assistant", + "model": "claude-test", + "content": [{"type": "text", "text": text}], + "stop_reason": stop_reason, + "stop_sequence": None, + "usage": {"input_tokens": 40, "output_tokens": 10}, + }, + ) + + +def _verdict(meets: bool, suitable: bool) -> str: + return json.dumps({"meets_requirements": meets, "suitable_for_use": suitable}) + + +class Scripted: + def __init__(self, *responses: httpx2.Response) -> None: + self._responses = list(responses) + self.requests: list[httpx2.Request] = [] + + def __call__(self, request: httpx2.Request) -> httpx2.Response: + self.requests.append(request) + return self._responses.pop(0) + + +def _assessor( + handler: Handler | AsyncHandler, samples: int = 3, max_concurrency: int = 10 +) -> ClaudeCandidateAssessor: + client = anthropic.AsyncAnthropic( + api_key="test-key", + max_retries=0, + http_client=anthropic.DefaultAsyncHttpxClient(transport=httpx2.MockTransport(handler)), + ) + return ClaudeCandidateAssessor( + client=client, + model="claude-test", + timeout_s=1.0, + max_retries=0, + samples=samples, + max_tokens=64, + max_concurrency=max_concurrency, + ) + + +def _error(status_code: int) -> httpx2.Response: + return httpx2.Response( + status_code, json={"type": "error", "error": {"type": "api_error", "message": "boom"}} + ) + + +class TestSelfConsistency: + @pytest.mark.parametrize( + ("suitable_votes", "expected"), + [((True, True, True), 1.0), ((True, True, False), 2 / 3), ((True, False, False), 1 / 3)], + ) + async def test_a3_probability_is_the_frequency_of_yes( + self, suitable_votes: tuple[bool, bool, bool], expected: float + ) -> None: + handler = Scripted(*(_message(_verdict(True, vote)) for vote in suitable_votes)) + + assessment = await _assessor(handler).assess(QUESTION, CANDIDATE) + + assert assessment.answers is not None + assert assessment.answers.suitable_for_use == pytest.approx(expected) + assert assessment.answers.meets_requirements == 1.0 + + async def test_cr2_one_invalid_sample_leaves_the_candidate_unassessed(self) -> None: + handler = Scripted( + _message(_verdict(True, True)), + _message("not json"), + _message(_verdict(True, True)), + ) + + assessment = await _assessor(handler).assess(QUESTION, CANDIDATE) + + assert assessment.answers is None + assert assessment.failure is AssessmentFailure.INVALID_SAMPLE + + async def test_cr2_a_response_cut_by_max_tokens_is_an_invalid_sample(self) -> None: + handler = Scripted( + _message(_verdict(True, True)), + _message(_verdict(True, True), stop_reason="max_tokens"), + _message(_verdict(True, True)), + ) + + assessment = await _assessor(handler).assess(QUESTION, CANDIDATE) + + assert assessment.answers is None + assert assessment.failure is AssessmentFailure.INVALID_SAMPLE + + async def test_a3_all_samples_invalid_means_no_assessment(self) -> None: + handler = Scripted(*(_message('{"unexpected": 1}') for _ in range(3))) + + assessment = await _assessor(handler).assess(QUESTION, CANDIDATE) + + assert assessment.answers is None + assert assessment.product_id == CANDIDATE.product_id + + async def test_makes_exactly_one_call_per_sample_and_reports_usage(self) -> None: + handler = Scripted(*(_message(_verdict(True, True)) for _ in range(3))) + + assessment = await _assessor(handler).assess(QUESTION, CANDIDATE) + + assert len(handler.requests) == 3 + assert assessment.usage.api_calls == 3 + assert assessment.usage.input_tokens == 120 + assert assessment.usage.output_tokens == 30 + + async def test_sends_only_this_candidate_and_the_rubric_with_a_strict_schema(self) -> None: + handler = Scripted(*(_message(_verdict(True, True)) for _ in range(3))) + + await _assessor(handler).assess(QUESTION, CANDIDATE) + + body = json.loads(handler.requests[0].content) + sent = json.dumps(body) + assert QUESTION in sent + assert "BOOT-1" in sent + for question in RUBRIC: + assert question.instructions in sent + schema = body["output_config"]["format"]["schema"] + assert schema["required"] == ["meets_requirements", "suitable_for_use"] + assert schema["additionalProperties"] is False + assert "temperature" not in body + assert body["output_config"]["effort"] == "low" + + +class TestErrors: + @pytest.mark.parametrize("status_code", [429, 500, 529]) + async def test_a4_transient_status_raises_llm_unavailable(self, status_code: int) -> None: + with pytest.raises(LLMUnavailableError): + await _assessor(Scripted(_error(status_code))).assess(QUESTION, CANDIDATE) + + async def test_a4_connection_failure_raises_llm_unavailable(self) -> None: + def handler(_request: httpx2.Request) -> httpx2.Response: + raise httpx2.ConnectError("refused") + + with pytest.raises(LLMUnavailableError): + await _assessor(handler).assess(QUESTION, CANDIDATE) + + @pytest.mark.parametrize( + ("status_code", "sdk_error"), + [(400, anthropic.BadRequestError), (401, anthropic.AuthenticationError)], + ) + async def test_a4_configuration_error_propagates( + self, status_code: int, sdk_error: type[anthropic.APIStatusError] + ) -> None: + with pytest.raises(sdk_error): + await _assessor(Scripted(_error(status_code))).assess(QUESTION, CANDIDATE) + + +class TestUsageOnFailure: + async def test_cr4_a_failing_sample_carries_the_usage_of_the_others(self) -> None: + handler = Scripted( + _message(_verdict(True, True)), _message(_verdict(True, True)), _error(529) + ) + + with pytest.raises(AssessmentUnavailableError) as raised: + await _assessor(handler).assess(QUESTION, CANDIDATE) + + assert isinstance(raised.value, LLMUnavailableError) + assert raised.value.usage.api_calls == 3 + assert raised.value.usage.input_tokens == 80 + assert raised.value.usage.output_tokens == 20 + + +class TestCallConcurrency: + async def test_cr1_never_more_calls_in_flight_than_the_limit(self) -> None: + in_flight = 0 + peak = 0 + + async def handler(_request: httpx2.Request) -> httpx2.Response: + nonlocal in_flight, peak + in_flight += 1 + peak = max(peak, in_flight) + await asyncio.sleep(0.05) + in_flight -= 1 + return _message(_verdict(True, True)) + + assessor = _assessor(handler, max_concurrency=2) + others = [replace(CANDIDATE, product_id=uuid4()) for _ in range(3)] + + await asyncio.gather(*(assessor.assess(QUESTION, c) for c in others)) + + assert peak == 2 + + +class TestIdentity: + def test_reports_provider_and_model(self) -> None: + assessor = _assessor(Scripted()) + + assert (assessor.provider, assessor.model_name) == ("claude", "claude-test") diff --git a/tests/integration/contexts/assistant/infrastructure/test_sqlalchemy_documents.py b/tests/integration/contexts/assistant/infrastructure/test_sqlalchemy_documents.py index 1679222..2b76ad2 100644 --- a/tests/integration/contexts/assistant/infrastructure/test_sqlalchemy_documents.py +++ b/tests/integration/contexts/assistant/infrastructure/test_sqlalchemy_documents.py @@ -1,6 +1,8 @@ from uuid import UUID, uuid4 import pytest +from sqlalchemy import text +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from contexts.assistant.domain.product_document import EmbeddedDocument, ProductDocument @@ -21,11 +23,20 @@ def _vector(*weights: tuple[int, float]) -> list[float]: def _embedded( - content: str, vector: list[float], product_id: UUID | None = None + content: str, + vector: list[float], + product_id: UUID | None = None, + sku: str | None = None, ) -> EmbeddedDocument: return EmbeddedDocument( document=ProductDocument( - product_id=product_id or uuid4(), content=content, content_hash=f"hash-{content}" + product_id=product_id or uuid4(), + sku=sku or f"SKU-{content.upper()}", + name=f"Name {content}", + price_cents=1999, + currency="EUR", + content=content, + content_hash=f"hash-{content}", ), embedding=vector, ) @@ -119,3 +130,44 @@ async def test_limits_results_to_k( async def test_empty_index_returns_no_results(self, search: SqlAlchemyDocumentSearch) -> None: assert await search.search(_vector((0, 1.0)), k=5) == [] + + +class TestReadModelColumns: + async def test_a7_upsert_stores_the_catalog_columns( + self, index: SqlAlchemyDocumentIndex, db_session: AsyncSession + ) -> None: + item = _embedded("widget", _vector((0, 1.0)), sku="SKU-W") + + await index.upsert([item]) + + row = ( + await db_session.execute( + text( + "SELECT sku, name, price_cents, currency FROM product_documents " + "WHERE product_id = :pid" + ), + {"pid": item.document.product_id}, + ) + ).one() + assert tuple(row) == ("SKU-W", "Name widget", 1999, "EUR") + + async def test_a7_a_row_missing_catalog_columns_reports_no_hash( + self, index: SqlAlchemyDocumentIndex, db_session: AsyncSession + ) -> None: + product_id = uuid4() + await db_session.execute( + text( + "INSERT INTO product_documents " + "(product_id, content, content_hash, embedding, updated_at) " + "VALUES (:pid, 'legacy', 'hash-legacy', :embedding, now())" + ), + {"pid": product_id, "embedding": str(_vector((0, 1.0)))}, + ) + + assert (await index.content_hashes())[product_id] is None + + async def test_a8_sku_is_unique_in_the_index(self, index: SqlAlchemyDocumentIndex) -> None: + await index.upsert([_embedded("first", _vector((0, 1.0)), sku="SKU-DUP")]) + + with pytest.raises(IntegrityError): + await index.upsert([_embedded("second", _vector((1, 1.0)), sku="SKU-DUP")]) diff --git a/tests/unit/contexts/assistant/application/test_index_catalog.py b/tests/unit/contexts/assistant/application/test_index_catalog.py index d55e506..088e813 100644 --- a/tests/unit/contexts/assistant/application/test_index_catalog.py +++ b/tests/unit/contexts/assistant/application/test_index_catalog.py @@ -173,3 +173,38 @@ async def test_r3_embeds_and_upserts_in_batches( assert report.indexed == 5 assert [len(call) for call in embeddings.calls] == [2, 2, 1] assert index.upsert_sizes == [2, 2, 1] + + +class TestReadModelColumns: + async def test_a7_reindexes_documents_missing_catalog_columns( + self, + use_case: IndexCatalog, + catalog: FakeCatalogReader, + embeddings: FakeEmbeddingClient, + index: FakeDocumentIndex, + ) -> None: + await use_case.execute() + embeddings.calls.clear() + incomplete = catalog.products[0] + index.incomplete.add(incomplete.product_id) + + report = await use_case.execute() + + assert report == IndexReportDTO(indexed=1, unchanged=2, deleted=0) + expected = ProductDocument.from_catalog_product(incomplete, embedding_model=MODEL) + assert embeddings.calls == [[expected.content]] + assert index.incomplete == set() + + async def test_a8_a_reused_sku_does_not_break_the_reindex( + self, embeddings: FakeEmbeddingClient, index: FakeDocumentIndex + ) -> None: + original = _product("reused") + catalog = FakeCatalogReader([original]) + await IndexCatalog(catalog, embeddings, index).execute() + replacement = replace(original, product_id=uuid4(), name="reused v2") + catalog.products = [replacement] + + report = await IndexCatalog(catalog, embeddings, index).execute() + + assert report == IndexReportDTO(indexed=1, unchanged=0, deleted=1) + assert set(index.documents) == {replacement.product_id} diff --git a/tests/unit/contexts/assistant/application/test_recommend_candidate.py b/tests/unit/contexts/assistant/application/test_recommend_candidate.py new file mode 100644 index 0000000..7b8ec54 --- /dev/null +++ b/tests/unit/contexts/assistant/application/test_recommend_candidate.py @@ -0,0 +1,235 @@ +import asyncio +import logging +import random +import time +from uuid import uuid4 + +import pytest +from tests.fakes.candidate_assessor import FakeCandidateAssessor + +from contexts.assistant.application.use_cases.recommend_candidate import RecommendCandidate +from contexts.assistant.domain.candidate import AssessmentUsage, Candidate, RubricAnswers +from contexts.assistant.domain.exceptions import AssessmentUnavailableError, LLMUnavailableError +from contexts.assistant.domain.selection import SelectionReason + +QUESTION = "Which of these do you recommend for hiking?" +YES = RubricAnswers(meets_requirements=1.0, suitable_for_use=1.0) +WEAK = RubricAnswers(meets_requirements=1.0, suitable_for_use=0.67) + + +def _candidates(count: int) -> list[Candidate]: + return [ + Candidate( + product_id=uuid4(), + sku=f"SECRET-SKU-{i}", + name=f"Secret name {i}", + price_cents=1000 + i, + currency="EUR", + content=f"SKU: SECRET-SKU-{i}", + retrieval_score=round(1.0 - i / 100, 2), + ) + for i in range(count) + ] + + +def _use_case( + assessor: FakeCandidateAssessor, top_n: int = 10, timeout_s: float = 5.0 +) -> RecommendCandidate: + return RecommendCandidate( + assessor, + top_n=top_n, + timeout_s=timeout_s, + requirements_threshold=0.5, + min_score=0.5, + ) + + +class TestCandidatesAssessed: + async def test_a5_only_the_top_n_by_retrieval_score_are_assessed(self) -> None: + candidates = _candidates(12) + assessor = FakeCandidateAssessor(default=WEAK) + + await _use_case(assessor, top_n=10).execute(QUESTION, list(reversed(candidates))) + + assert sorted(assessor.assessed) == sorted(c.product_id for c in candidates[:10]) + + async def test_a5_a_candidate_past_the_deadline_counts_as_not_assessed(self) -> None: + slow, fast = _candidates(2) + assessor = FakeCandidateAssessor( + outcomes={slow.product_id: YES, fast.product_id: WEAK}, + delays={slow.product_id: 1.0}, + ) + + result = await _use_case(assessor, timeout_s=0.2).execute(QUESTION, [slow, fast]) + + assert result.product_id == fast.product_id + assert result.assessed == 1 + + +class TestFailures: + async def test_a6_a_transient_failure_skips_only_that_candidate(self) -> None: + failing, ok = _candidates(2) + assessor = FakeCandidateAssessor( + outcomes={failing.product_id: LLMUnavailableError("down"), ok.product_id: WEAK} + ) + + result = await _use_case(assessor).execute(QUESTION, [failing, ok]) + + assert result.product_id == ok.product_id + assert result.reason is SelectionReason.CHOSEN + + async def test_a6_all_failing_falls_back_to_retrieval_order(self) -> None: + candidates = _candidates(3) + assessor = FakeCandidateAssessor(default=LLMUnavailableError("down")) + + result = await _use_case(assessor).execute(QUESTION, list(reversed(candidates))) + + assert result.product_id is None + assert result.fallback_used is True + assert result.reason is SelectionReason.NO_ASSESSMENTS + assert [c.product_id for c in result.candidates] == [c.product_id for c in candidates] + + async def test_a6_a_configuration_error_propagates(self) -> None: + assessor = FakeCandidateAssessor(default=PermissionError("bad key")) + + with pytest.raises(PermissionError): + await _use_case(assessor).execute(QUESTION, _candidates(2)) + + +class TestResult: + async def test_candidates_are_returned_in_retrieval_order(self) -> None: + candidates = _candidates(4) + shuffled = random.Random(7).sample(candidates, k=4) + assessor = FakeCandidateAssessor(default=WEAK) + + result = await _use_case(assessor).execute(QUESTION, shuffled) + + assert [c.product_id for c in result.candidates] == [c.product_id for c in candidates] + first = result.candidates[0] + assert (first.sku, first.name, first.price_cents) == ("SECRET-SKU-0", "Secret name 0", 1000) + + async def test_a2_shuffled_input_gives_the_same_recommendation(self) -> None: + candidates = _candidates(6) + outcomes = {c.product_id: WEAK for c in candidates} + outcomes[candidates[3].product_id] = YES + + results = { + (await _use_case(FakeCandidateAssessor(outcomes=dict(outcomes))).execute( + QUESTION, random.Random(seed).sample(candidates, k=6) + )).product_id + for seed in range(5) + } + + assert results == {candidates[3].product_id} + + +class TestLogging: + async def test_a9_one_line_per_selection_with_ids_only( + self, caplog: pytest.LogCaptureFixture + ) -> None: + failing, chosen, other = _candidates(3) + assessor = FakeCandidateAssessor( + outcomes={ + failing.product_id: LLMUnavailableError("429"), + chosen.product_id: YES, + other.product_id: WEAK, + } + ) + caplog.set_level(logging.INFO, logger="contexts.assistant") + + await _use_case(assessor).execute(QUESTION, [failing, chosen, other]) + + lines = [r.getMessage() for r in caplog.records if r.name.startswith("contexts.assistant")] + assert len(lines) == 1 + line = lines[0] + for expected in ( + "provider=fake", + "model=fake-assessor", + "candidates=3", + "assessed=2", + "api_calls=6", + "input_tokens=200", + "output_tokens=20", + "cost_usd=None", + "reason=chosen", + f"product_id={chosen.product_id}", + "latency_ms=", + ): + assert expected in line + for leaked in ("SECRET", "Secret name", "hiking"): + assert leaked not in line + + +class TestReviewFixes: + async def test_cr3_a_configuration_error_does_not_wait_for_the_deadline(self) -> None: + broken, slow = _candidates(2) + assessor = FakeCandidateAssessor( + outcomes={broken.product_id: PermissionError("bad key"), slow.product_id: WEAK}, + delays={slow.product_id: 5.0}, + ) + started = time.perf_counter() + + with pytest.raises(PermissionError): + await _use_case(assessor, timeout_s=3.0).execute(QUESTION, [broken, slow]) + + assert time.perf_counter() - started < 1.0 + assert assessor.cancelled == 1 + + async def test_cr3_cancelling_the_request_cancels_every_assessment(self) -> None: + candidates = _candidates(3) + assessor = FakeCandidateAssessor( + default=WEAK, delays={c.product_id: 5.0 for c in candidates} + ) + task = asyncio.create_task(_use_case(assessor).execute(QUESTION, candidates)) + await asyncio.sleep(0.05) + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + assert assessor.cancelled == 3 + + async def test_cr4_usage_of_a_failed_assessment_is_logged( + self, caplog: pytest.LogCaptureFixture + ) -> None: + failing, ok = _candidates(2) + partial = AssessmentUsage(api_calls=3, input_tokens=80, output_tokens=20) + assessor = FakeCandidateAssessor( + outcomes={ + failing.product_id: AssessmentUnavailableError(partial), + ok.product_id: WEAK, + } + ) + caplog.set_level(logging.INFO, logger="contexts.assistant") + + await _use_case(assessor).execute(QUESTION, [failing, ok]) + + [line] = [r.getMessage() for r in caplog.records if r.name.startswith("contexts.assistant")] + assert "api_calls=6" in line + assert "input_tokens=180" in line + assert "output_tokens=30" in line + + async def test_cr2_unassessed_candidates_are_logged_with_their_reason( + self, caplog: pytest.LogCaptureFixture + ) -> None: + failing, slow, invalid, ok = _candidates(4) + assessor = FakeCandidateAssessor( + outcomes={ + failing.product_id: LLMUnavailableError("429"), + slow.product_id: WEAK, + invalid.product_id: None, + ok.product_id: WEAK, + }, + delays={slow.product_id: 5.0}, + ) + caplog.set_level(logging.INFO, logger="contexts.assistant") + + result = await _use_case(assessor, timeout_s=0.3).execute( + QUESTION, [failing, slow, invalid, ok] + ) + + assert result.product_id == ok.product_id + [line] = [r.getMessage() for r in caplog.records if r.name.startswith("contexts.assistant")] + assert f"{failing.product_id}:error" in line + assert f"{slow.product_id}:deadline" in line + assert f"{invalid.product_id}:invalid_sample" in line diff --git a/tests/unit/contexts/assistant/domain/test_product_document.py b/tests/unit/contexts/assistant/domain/test_product_document.py index 70d420c..94ba48d 100644 --- a/tests/unit/contexts/assistant/domain/test_product_document.py +++ b/tests/unit/contexts/assistant/domain/test_product_document.py @@ -69,3 +69,17 @@ def test_r2_same_content_with_another_model_has_another_hash(self) -> None: assert first.content == second.content assert first.content_hash != second.content_hash + + +class TestCatalogFields: + def test_a7_carries_the_catalog_fields(self) -> None: + product = _product() + + document = ProductDocument.from_catalog_product(product, embedding_model=MODEL) + + assert (document.sku, document.name, document.price_cents, document.currency) == ( + "WIDGET-001", + "Blue widget", + 1999, + "EUR", + ) diff --git a/tests/unit/contexts/assistant/domain/test_selection.py b/tests/unit/contexts/assistant/domain/test_selection.py new file mode 100644 index 0000000..e48e9d8 --- /dev/null +++ b/tests/unit/contexts/assistant/domain/test_selection.py @@ -0,0 +1,173 @@ +import itertools +from uuid import UUID, uuid4 + +import pytest + +from contexts.assistant.domain.candidate import Assessment, Candidate, RubricAnswers +from contexts.assistant.domain.exceptions import InvalidAssessmentError +from contexts.assistant.domain.rubric import ( + MEETS_REQUIREMENTS, + RUBRIC, + SUITABLE_FOR_USE, + RubricQuestion, +) +from contexts.assistant.domain.selection import Selection, SelectionReason, select + +REQUIREMENTS = 0.5 +MIN_SCORE = 0.5 + + +def _candidate(retrieval_score: float = 0.5, product_id: UUID | None = None) -> Candidate: + return Candidate( + product_id=product_id or uuid4(), + sku="SKU", + name="Name", + price_cents=100, + currency="EUR", + content="SKU: SKU | Name: Name | Price: 1.00 EUR", + retrieval_score=retrieval_score, + ) + + +def _assessed(candidate: Candidate, meets: float, suitable: float) -> Assessment: + return Assessment( + product_id=candidate.product_id, + answers=RubricAnswers(meets_requirements=meets, suitable_for_use=suitable), + ) + + +def _not_assessed(candidate: Candidate) -> Assessment: + return Assessment(product_id=candidate.product_id, answers=None) + + +def _select(candidates: list[Candidate], assessments: list[Assessment]) -> Selection: + return select(candidates, assessments, REQUIREMENTS, MIN_SCORE) + + +class TestRanking: + def test_a1_highest_suitability_wins(self) -> None: + low, high = _candidate(), _candidate() + + selection = _select([low, high], [_assessed(low, 1, 0.67), _assessed(high, 1, 1)]) + + assert selection.product_id == high.product_id + assert selection.score == 1 + assert selection.reason is SelectionReason.CHOSEN + assert selection.fallback_used is False + + def test_a1_a_tie_goes_to_the_higher_retrieval_score(self) -> None: + weaker, stronger = _candidate(retrieval_score=0.4), _candidate(retrieval_score=0.9) + + selection = _select( + [weaker, stronger], [_assessed(weaker, 1, 1), _assessed(stronger, 1, 1)] + ) + + assert selection.product_id == stronger.product_id + + def test_a1_a_double_tie_goes_to_the_lower_product_id(self) -> None: + first = _candidate(product_id=UUID("00000000-0000-0000-0000-000000000001")) + second = _candidate(product_id=UUID("00000000-0000-0000-0000-000000000002")) + + selection = _select( + [second, first], [_assessed(second, 1, 1), _assessed(first, 1, 1)] + ) + + assert selection.product_id == first.product_id + + +class TestThresholds: + def test_a1_requirements_filter_excludes_even_the_most_suitable(self) -> None: + unmet, met = _candidate(), _candidate() + + selection = _select([unmet, met], [_assessed(unmet, 0.33, 1), _assessed(met, 0.67, 0.67)]) + + assert selection.product_id == met.product_id + + def test_a1_exactly_at_both_thresholds_passes(self) -> None: + candidate = _candidate() + + selection = _select([candidate], [_assessed(candidate, 0.5, 0.5)]) + + assert selection.product_id == candidate.product_id + + def test_a1_all_below_min_score_recommends_nothing(self) -> None: + a, b = _candidate(), _candidate() + + selection = _select([a, b], [_assessed(a, 1, 0.33), _assessed(b, 1, 0)]) + + assert selection.product_id is None + assert selection.reason is SelectionReason.BELOW_THRESHOLD + assert selection.fallback_used is False + + def test_a1_all_failing_requirements_recommends_nothing(self) -> None: + a = _candidate() + + selection = _select([a], [_assessed(a, 0, 1)]) + + assert selection.product_id is None + assert selection.reason is SelectionReason.BELOW_THRESHOLD + + +class TestUnassessed: + def test_a6_a_failed_candidate_is_skipped(self) -> None: + failed, ok = _candidate(retrieval_score=0.99), _candidate() + + selection = _select([failed, ok], [_not_assessed(failed), _assessed(ok, 1, 1)]) + + assert selection.product_id == ok.product_id + + def test_a6_no_assessment_at_all_falls_back(self) -> None: + a, b = _candidate(), _candidate() + + selection = _select([a, b], [_not_assessed(a), _not_assessed(b)]) + + assert selection.product_id is None + assert selection.reason is SelectionReason.NO_ASSESSMENTS + assert selection.fallback_used is True + + +class TestSafety: + def test_a2_never_returns_a_product_outside_the_candidates(self) -> None: + candidate, stranger = _candidate(), _candidate() + + selection = _select([candidate], [_assessed(stranger, 1, 1), _assessed(candidate, 1, 0.67)]) + + assert selection.product_id == candidate.product_id + + def test_a2_shuffled_input_gives_the_same_selection(self) -> None: + candidates = [_candidate(retrieval_score=s) for s in (0.2, 0.9, 0.5, 0.9)] + assessments = [ + _assessed(candidates[0], 1, 1), + _assessed(candidates[1], 1, 1), + _assessed(candidates[2], 0.33, 1), + _assessed(candidates[3], 1, 1), + ] + expected = _select(candidates, assessments) + + for order in itertools.permutations(range(len(candidates))): + shuffled = [candidates[i] for i in order] + assert _select(shuffled, list(reversed(assessments))) == expected + + +class TestRubric: + def test_rubric_has_the_two_questions_in_order(self) -> None: + assert [question.id for question in RUBRIC] == ["meets_requirements", "suitable_for_use"] + + def test_requirements_question_says_yes_when_there_are_no_explicit_requirements(self) -> None: + assert "no explicit requirements" in MEETS_REQUIREMENTS.yes_means.lower() + + @pytest.mark.parametrize("question", RUBRIC) + def test_every_question_explains_yes_and_no(self, question: RubricQuestion) -> None: + assert question.instructions + assert question.yes_means + assert question.no_means + + def test_suitability_is_the_ranking_question(self) -> None: + assert SUITABLE_FOR_USE.id == "suitable_for_use" + + +class TestRubricAnswers: + @pytest.mark.parametrize("value", [-0.01, 1.01]) + def test_rejects_values_that_are_not_probabilities(self, value: float) -> None: + with pytest.raises(InvalidAssessmentError, match="probability"): + RubricAnswers(meets_requirements=value, suitable_for_use=0.5) diff --git a/tests/unit/shared/infrastructure/test_settings.py b/tests/unit/shared/infrastructure/test_settings.py index 03463a4..761bc04 100644 --- a/tests/unit/shared/infrastructure/test_settings.py +++ b/tests/unit/shared/infrastructure/test_settings.py @@ -11,3 +11,8 @@ def test_r4_accepts_up_to_the_hnsw_default_ef_search(self) -> None: def test_r4_rejects_more_than_the_hnsw_default_ef_search(self) -> None: with pytest.raises(ValidationError, match="rag_top_k"): Settings(rag_top_k=41) + + +class TestSelectorMaxTokens: + def test_cr6_the_assessor_has_its_own_output_limit(self) -> None: + assert Settings().selector_max_tokens == 1024