diff --git a/CLAUDE.md b/CLAUDE.md index a294f7a..508a355 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -10,9 +10,12 @@ Target domains: ML/AI, health/biomedicine, tech news. ## Current Phase -- **Version:** V1 — RAG Robusto + Telegram -- **Weeks:** 1–4 -- **Focus:** Iterative RAG (fixed → semantic → hybrid → reranking), Telegram bot +- **Version:** V2 — Agente LangGraph + Briefing matutino +- **Window:** 17/08 – 02/10/2026 (milestone due 02/10/2026) +- **Focus:** Rebuild the RAG pipeline as a LangGraph agent (T18 in progress), + typed tools, conversational memory, morning briefing scheduler. See + `ROADMAP.md` for the full T18–T24 breakdown and `docs/architecture.md` + (ADR-005) for the LangGraph layer placement. > **UPDATE THIS** as you progress through versions. @@ -87,7 +90,8 @@ src/researchos/ ## What NOT to do -- Do NOT use LangChain/LangGraph in V1. Direct Claude API calls only. +- Do NOT import `langgraph` outside `infrastructure/orchestration/research_graph.py`. + Nodes and state stay framework-free — see ADR-005 in `docs/architecture.md`. - Do NOT put business logic in infrastructure/ - Do NOT import infrastructure in domain/ - Do NOT use class inheritance for agents — use composition via agent_utils.py diff --git a/docs/architecture.md b/docs/architecture.md index 999dabf..96db157 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -50,6 +50,25 @@ **Consequences:** Lighter than a `Protocol` for the common case of one behavior, still statically checkable via the `Callable` signature. Swapping retrieval strategy (vector-only → hybrid+rerank) required zero changes to `TelegramBot` — only the closure built in the composition root changed. Rule going forward: one behavior → function-type alias; several related behaviors sharing state → `Protocol` or a class. +--- + +## ADR-005: LangGraph layer placement — state in domain, nodes in application, assembly in infrastructure + +**Date:** 2026-08 +**Status:** Accepted + +**Context:** V2 introduces LangGraph, which is a borderline case for the dependency rule: unlike an SDK or a DB driver, it expresses control flow, and control flow is `application/`'s job. Three options were evaluated: +1. state, nodes and graph all in `application/agents/research_agent/`, accepting the `langgraph` import as a documented exception +2. state in `domain/`, pure nodes in `application/`, assembly in `infrastructure/`; +3. everything in `infrastructure/orchestration/`, with `application/` exposing only the functions nodes call. + +Option 1 is what every official example does but creates an erosion precedent — the project already carries `ingestion_service.py` importing `httpx`/`fitz` as *debt*, not as an accepted exception. Option 3 would put business rules like "if local retrieval is thin, search arXiv" (T19) inside `infrastructure/`. The deciding finding: a reducer is just a `Callable[[T, T], T]` (the official quickstart uses stdlib `operator.add`), so **only `StateGraph`/`START`/`END`/`compile()` truly require the framework** — the state and the nodes do not. A premise in favour of importing `add_messages` into the state was investigated and discarded: it dedupes messages by `id` *within a thread*, it is not what isolates concurrent users — that is the checkpointer's `thread_id` (T22). + +**Decision:** Option 2. `ResearchContext` (`query`, `documents`, `answer`) in `domain/models.py` with no external imports; nodes as pure `(ResearchContext) -> dict` functions in `application/agents/research_agent/nodes.py`; `StateGraph`, edges and `compile()` in `infrastructure/orchestration/research_graph.py` — the only file in the project importing `langgraph`. T19 routing uses **conditional edges**, not `Command(goto=...)`, so the routing function stays a pure `state -> str` in `application/` and can be tested without executing the node (and its LLM call). `messages` and `rewritten_query` are deliberately not declared yet: they arrive in T22, once the required merge semantics is known. + +**Consequences:** `domain/` and `application/` tests run without `langgraph` installed; nodes are tested by passing a fabricated state and asserting the returned dict. `answer_query` keeps working without the graph, which T24 needs to run the V1 pipeline and the V2 agent side by side over the same queries. Fourth instance of the same project pattern after `AnswerFn`, `RetrieveFn` and `with_logging`: the core never knows the mechanism invoking it. Cost: the agent lives across three files in two layers, no official example looks like this, so tutorial code must be relocated rather than copied. Framework portability was *not* a reason — node return conventions are LangGraph-specific and would be rewritten anyway; ADK appears in V6 as a comparison exercise, not a migration. Re-evaluate if: multi-agent in V6 forces `Command` (mandatory for subgraph→parent routing, which would make those nodes impure), the assembly file starts accumulating business logic, or a second piece outside the assembly requires the framework. For T22, `messages` will first get a hand-written reducer in `domain/` (a dedupe-by-`id` dict comprehension, ~15 lines) and later the split-state variant as a comparison exercise — `GraphState(TypedDict)` in `infrastructure/` composing `ResearchContext` from `domain/` plus `Annotated[list, add_messages]`, on the grounds that a LangChain-formatted message list with a LangGraph reducer is a framework structure, not a domain model. + + --- diff --git a/docs/learnings.md b/docs/learnings.md index 2ac6b7f..2f85d48 100644 --- a/docs/learnings.md +++ b/docs/learnings.md @@ -502,3 +502,78 @@ Regla simple para ResearchOS: - El mecanismo de flujo de la información ya que el patrón me muestra que la función que envuelve recibe los mismo argumentos de la función que quiero envolver, pero aún así, no asimilo muy bien cómo fluye la información ya que estoy acostrumbrado a un patrón más lineal (spaguetti) --- + +**Fecha:** 19/08/2026 + +### ¿Qué aprendí? + +- **Un reducer de LangGraph es solo un `Callable[[T, T], T]`.** El quickstart oficial usa `operator.add`, de la biblioteca estándar. `add_messages` es una conveniencia, no un requisito. Consecuencia arquitectónica: el estado y los nodos pueden definirse sin importar LangGraph; lo único inevitable es `StateGraph`/`START`/`END`/`compile()`, que pertenecen al ensamblado. + +- **`add_messages` no hace lo que creía.** Deduplica y actualiza mensajes por `id` *dentro de un mismo hilo*, para editar o corregir mensajes. El aislamiento entre usuarios concurrentes lo da el `thread_id` del checkpointer, no el reducer. Confundí las dos cosas y sobre esa premisa equivocada casi justifiqué una excepción a la regla de capas. + +- **Los nodos reciben el estado completo, siempre.** No existe vista parcial: la firma es `(ResearchContext) -> dict`. Lo que varía es qué campos lee cada nodo y qué devuelve. Y devuelven un **dict parcial** — LangGraph fusiona con el estado existente, no hay que reconstruir el objeto. + +- **Fábricas de nodos para inyectar dependencias.** Un nodo no admite parámetros extra, pero necesita `RetrieveFn` y `LLMProvider`. Tres salidas: meterlas al estado (revienta en T22, el checkpointer tiene que serializar y un cliente HTTP no es serializable), globales de módulo (mete infraestructura concreta en `application/`), o una fábrica que las capture en closure y devuelva el nodo. La tercera es la correcta — quinta aplicación del mismo patrón después de `AnswerFn`, `RetrieveFn` y `with_logging`. + +- **Beneficio no buscado de la fábrica:** el prompt de sistema se lee del disco **una vez** al construir el grafo, en el cuerpo de la fábrica. Hoy `answer_query` lo lee en cada pregunta. + +- **Campo obligatorio = fallo temprano.** `query` sin default hace que `ResearchContext()` lance `TypeError` al construir. Con `query: str = ""` se construiría bien y el error aparecería mucho después: embedding de string vacío, retrieval basura, respuesta rara, y hay que rastrear hacia atrás. Criterio: obligatorio lo que el sistema no puede inventar; con default lo que empieza vacío por naturaleza. + +- **Elegir dataclass sobre TypedDict cambia la sintaxis de acceso.** Los ejemplos oficiales usan `state["query"]` porque declaran el estado como `TypedDict`. Con dataclass es `state.query` — y mypy caza los typos, cosa que el acceso por string no permite. + +### ¿Qué no entendí bien / queda abierto? + +- `documents` no tiene reducer, así que se reemplaza. En T22, si el ciclo de reescritura corre el nodo de recuperación dos veces, la segunda tanda pisa la primera. Probablemente sea lo deseado (quiero los documentos de la mejor query, no la unión), pero es una decisión sin confirmar. +- No verifiqué si `StateGraph(ResearchContext)` intenta construir el estado sin argumentos en algún punto interno. Si lo hiciera, `query` obligatorio sería un problema. + +### Decisiones de diseño + +- **ADR-005**: estado en `domain/models.py`, nodos puros en `application/agents/research_agent/nodes.py`, ensamblado en `infrastructure/orchestration/research_graph.py`. Único archivo del proyecto que importa `langgraph` es el del ensamblado. Se evaluaron tres opciones y se descartó la portabilidad de framework como razón — las razones reales son correr V1 y V2 en paralelo (T24), testear ruteo sin montar el grafo (T19), y mantener `application/` libre de frameworks. +- Estado inicial de tres campos: `query`, `documents`, `answer`. `messages` y `rewritten_query` se posponen a T22, cuando se conozca la semántica de fusión que cada uno necesita. +- `AnswerFn` y `RetrieveFn` movidos a `domain/interfaces.py`. Son vocabulario de contratos, igual que los Protocols; tenerlos en `rag_service.py` obligaba a `nodes.py` a importar de un servicio con el que no tiene relación. +- Para T19 se usarán conditional edges, no `Command(goto=...)`: el criterio de aceptación pide testear el ruteo aislado, y con `Command` habría que ejecutar el nodo completo con su llamada al LLM. + +### Errores interesantes + +- Puse `await make_retrieve_node(...)` dentro de `build_research_graph`. Doble error: la fábrica es `def` normal y no devuelve corrutina, y la función contenedora tampoco es `async`, así que era `SyntaxError` al importar. Tercera vez que confundo el tiempo de la fábrica con el tiempo de la función que devuelve — la regla de async se aplica a cada función por separado, no al archivo. +- Escribí `state["query"]` copiando el patrón de los ejemplos oficiales, que usan `TypedDict`. Con dataclass es acceso por atributo. +- Cargué el prompt de sistema en `make_retrieve_node`, donde no se usa. Va en `make_generate_node`. +- Corrí `nodes.py` esperando ver salida. No tiene bloque `__main__` — solo define funciones. El bloque de prueba estaba en `research_graph.py`. +- El docstring de `research_graph.py` abría con cuatro comillas simples en vez de tres. + +--- + +**Fecha:** 20/08/2026 + +### ¿Qué aprendí? + +- **El grafo se compila una vez, no por consulta.** Mi primera versión llamaba `build_research_graph()` dentro del closure `answer_with_graph`, así que en cada pregunta se instanciaba `StateGraph`, se ejecutaban las dos fábricas, se cableaban las edges y se compilaba. Peor: `make_generate_node` lee el prompt de sistema del disco en el cuerpo de la fábrica, así que la ventaja de leerlo una sola vez se anulaba. Tercera vez que cometo el mismo error de poner construcción cara en el lugar de ejecución (antes: `LocalEmbedder()` dentro del closure de logging). + +- **Un composition root mezcla tres cosas distintas:** construcción de dependencias (embedder, Chroma, BM25, LLM), composición del motor (grafo, envolturas) y arranque del canal (`bot.run()`). Solo la primera es común a todos los puntos de entrada; por eso es la única que se extrajo a `_wiring.py`. + +- **`scripts/` no está en el paquete instalable, y eso cambia cómo se importa.** Al correr `uv run python scripts/x.py`, Python pone el directorio del *script* en `sys.path`, no la raíz del repo. Por eso el import es `from _wiring import ...` y no `from scripts._wiring import ...`. + +- **El parche de `pysqlite3` tiene que estar a nivel de módulo, no dentro de una función.** `chromadb` importa `sqlite3` al importarse, así que la sustitución en `sys.modules` debe ocurrir antes. Importar `_wiring` es lo que aplica el parche. + +- **Duplicación deliberada vs. accidental.** Extraje `build_dependencies` porque cómo se conecta a Chroma y cómo se reconstruye BM25 *deben* ser idénticos entre scripts. Consideré extraer también `retrieve_hybrid_rerank`, que está duplicada palabra por palabra, y lo descarté: es una elección de composición, no infraestructura. El smoke test puede legítimamente querer una estrategia distinta a la del bot, y centralizarla mataría ese aislamiento. Además T24 va a necesitar varias estrategias conviviendo. **La regla: se extrae cuando las copias deben cambiar juntas, no cuando se ven iguales.** + +- **Un `NetworkError` de `httpx.ReadError` en el long polling no es un fallo del aplicativo.** El traceback vive entero en `telegram/`, `httpx/` y `httpcore/`, y ocurre dentro de una función llamada `network_retry_loop`: la biblioteca ya lo contempla y reintenta. En un entorno corporativo con proxy es esperable. + +### ¿Qué no entendí bien / queda abierto? + +- El log dice `No error handlers are registered, logging exception`. Hoy eso aplica a errores de red que la biblioteca resuelve sola, pero si mañana el grafo lanza una excepción, el usuario en Telegram no recibe nada y yo veo un muro de traceback sin poder distinguir "la red parpadeó" de "el grafo reventó". Falta un `add_error_handler`. +- `build_dependencies` carga **todos** los documentos de Chroma en memoria en cada arranque para reconstruir BM25, porque BM25 no persiste. Con el corpus actual es instantáneo, pero T21 (briefing matutino) va a ingerir papers cada mañana y ese arranque se va a alargar. + +### Decisiones de diseño + +- `build_dependencies()` en `scripts/_wiring.py` devolviendo un `NamedTuple` (`chroma`, `bm25`, `llm`). NamedTuple sobre tupla suelta: `deps.chroma` se lee mejor que `deps[0]` y mypy lo verifica. +- **No** se extrae `retrieve_hybrid_rerank` a `_wiring.py` (ver arriba). La duplicación se mantiene a propósito. +- `answer_v1_pipeline` y `answer_v2_graph` conviven en `run_telegram_bot.py`, con el bot cableado al segundo. El primero se conserva para T24, donde hay que correr ambos sobre las mismas queries. +- `run_research_graph.py` acepta la query por `argparse` en vez de hardcodearla. + +### Errores interesantes + +- `build_research_graph()` dentro del closure en vez de a nivel de módulo (ver arriba). +- El criterio de aceptación de T18 decía que `git diff --stat` no debía tocar `telegram_bot.py`, y lo toca. Pero el diff es **solo** el import de `AnswerFn` desde `domain/interfaces` en vez de definirlo localmente — consecuencia del movimiento de alias de ayer, no adaptación al grafo. El criterio se cumple en lo que buscaba verificar: conectar el grafo no requirió modificar la lógica del adaptador. + +--- diff --git a/docs/work_log.md b/docs/work_log.md index 9da4a99..8977aeb 100644 --- a/docs/work_log.md +++ b/docs/work_log.md @@ -328,3 +328,97 @@ `rewritten_query`) --- + +## 2026-08-19 + +### Trabajo desarrollado +- Arrancado T18: primer grafo LangGraph mínimo (`retrieve → generate`), + siguiendo la ubicación de capas decidida en ADR-005 +- `domain/models.py`: nuevo `ResearchContext` (dataclass) — estado del agente + (`query`, `documents`, `answer`); `messages`/`rewritten_query` quedan para + T22, cuando se conozca la semántica de merge que necesitan +- `domain/interfaces.py`: `RetrieveFn` y `AnswerFn` se relocalizan aquí desde + `rag_service.py` y `telegram_bot.py` respectivamente — quedan como los + primeros alias de tipo compartidos entre más de un consumidor (ADR-004) +- `application/agents/research_agent/nodes.py` (nuevo): `make_retrieve_node` + y `make_generate_node`, fábricas que devuelven nodos puros + `(ResearchContext) -> dict`, inyectando `RetrieveFn`/`LLMProvider` por + clausura — cero imports de LangGraph, tal como fija ADR-005 +- `infrastructure/orchestration/research_graph.py` (nuevo): + `build_research_graph()` ensambla el `StateGraph` (`retrieve → generate`) + — único archivo del proyecto que importa `langgraph` +- `scripts/run_research_graph.py` (nuevo): script de humo con CLI (`-q`) que + corre el grafo contra Chroma/Anthropic reales para verificación manual +- `docs/architecture.md`: agregado ADR-005 (ubicación de capas para + LangGraph — estado en domain, nodos en application, ensamblaje en + infrastructure), con la comparación de las 3 opciones evaluadas +- Revisión de calidad sobre todo lo anterior: `ruff check --fix` + + `ruff format` (imports desordenados, whitespace, EOF); agregados + docstrings y type hints faltantes (`build_research_graph`, nodos); un + error real de `mypy` en `add_node` resulta ser una limitación de los stubs + de LangGraph (reproducida en un caso mínimo fuera del proyecto, incluso + pasando `input_schema` explícito) — silenciado con `type: ignore` puntual + y documentado, no es deuda de código propio +- `tests/unit/application/test_research_agent_nodes.py` (nuevo): cubre + ambos nodos (`retrieve_node`, `generate_node`) con `ResearchContext` + fabricado, verificando el dict parcial devuelto — sin ejecutar el grafo + +### Próximos pasos +- Conectar el grafo al bot de Telegram (T18, cierre) +- Extraer el wiring duplicado entre `run_telegram_bot.py` y + `run_research_graph.py` a una función compartida — ya hay dos + consumidores, la abstracción se justifica +- Silenciar los ~34 errores de mypy provenientes de `chromadb` con overrides + en `pyproject.toml`, y arreglar los de código propio (el `datetime | None` + en `ingestion_service.py:70`) +- Confirmar la decisión de que `documents` se reemplace y no se acumule + entre reintentos (relevante para T22) + +--- + +## 2026-08-20 + +### Trabajo desarrollado +- Correcciones a las 2 observaciones del tutor sobre el commit de nodos de + ayer: `research_graph.py` documenta la limitación de mypy con versión + (`langgraph==1.2.11`) y el issue upstream (`langchain-ai/langgraph#5000`); + `test_research_agent_nodes.py` ahora verifica el contenido real enviado al + LLM (query + texto de documentos), no solo la respuesta — confirmado que + detecta la regresión si el nodo ignora `state.documents` +- `CLAUDE.md`: "Current Phase" actualizada de V1 a V2 (ventana + 17/08–02/10/2026); la regla sobre LangGraph en "What NOT to do" corregida + para reflejar ADR-005 (prohibido solo fuera de `research_graph.py`, no en + general) +- Bot de Telegram conectado al grafo LangGraph (T18/#6): `answer_v2_graph` + invoca `build_research_graph(...).ainvoke(...)`, reemplazando + `answer_query` como función cableada a `TelegramBot`. `answer_v1_pipeline` + se conserva sin cablear, para T24. Nuevo `test_research_graph.py` que + invoca el grafo completo con retriever y LLM mockeados — cierra los 3 + criterios de aceptación pendientes de T18 +- Wiring duplicado entre `run_telegram_bot.py` y `run_research_graph.py` + extraído a `scripts/_wiring.py` (`build_dependencies()`), incluyendo el + parche de `pysqlite3` que estaba copiado en 4 scripts. + `retrieve_hybrid_rerank` se mantiene duplicada a propósito — decisión de + composición, no de infraestructura (ver `docs/learnings.md`) +- `run_telegram_bot.py` ya importa `AnswerFn` directo desde + `domain/interfaces` (no vía re-export de `telegram_bot.py`) +- `docs/learnings.md`: entrada de hoy documenta el error de compilar el + grafo por consulta en vez de una sola vez, la distinción composition-root + vs. wiring, y el criterio para decidir cuándo extraer duplicación + +### Próximos pasos +- `add_error_handler` en `telegram_bot.py`: loguear la excepción y + responder al usuario en vez de dejarlo esperando +- BM25 se reconstruye en memoria desde Chroma en cada arranque; revisar + cuando el corpus supere unos cientos de documentos (T21 lo va a hacer + crecer) +- `run_research_graph.py` imprime los documentos completos; dejar solo + `doc_id` y score para que la salida sea legible +- Deuda de mypy: 34 errores, la mayoría de `chromadb`; silenciar con + overrides en `pyproject.toml` y arreglar los de código propio +- Decidir si `documents` debe acumularse o reemplazarse cuando T22 + introduzca el ciclo de reescritura +- Al llegar a T23 (Dockerfile): decidir si el bootstrap se mueve de + `scripts/` a `src/researchos/` para que sea importable desde la imagen + +--- diff --git a/scripts/_wiring.py b/scripts/_wiring.py new file mode 100644 index 0000000..bf2facb --- /dev/null +++ b/scripts/_wiring.py @@ -0,0 +1,69 @@ +"""Shared wiring for the operational scripts in this directory. + +Not part of the installable package (scripts/ isn't). Each script runs via +``uv run python scripts/.py``, so this module is imported by its bare +name (``_wiring``, not ``scripts._wiring``) — Python puts the running +script's own directory on ``sys.path``, not the repo root. + +The pysqlite3 patch below must run before anything imports ``chromadb`` +(which imports ``sqlite3`` internally), so it stays at module level here +instead of inside a function — importing this module is what applies it. +""" + +import sys +from typing import NamedTuple + +if sys.platform == "linux": + __import__("pysqlite3") + sys.modules["sqlite3"] = sys.modules.pop("pysqlite3") + +import chromadb + +from researchos.domain.models import Document +from researchos.infrastructure.llm.anthropic_llm import AnthropicLLM +from researchos.infrastructure.retrieval.bm25 import BM25Retriever +from researchos.infrastructure.retrieval.chroma import ChromaVectorStore +from researchos.infrastructure.retrieval.embedder import LocalEmbedder +from researchos.paths import CHROMA_DIR + + +class Dependencies(NamedTuple): + """The embedder/Chroma/BM25/LLM stack shared by the bot and graph scripts.""" + + chroma: ChromaVectorStore + bm25: BM25Retriever + llm: AnthropicLLM + + +def build_dependencies(collection_name: str = "papers") -> Dependencies: + """Build retrievers and LLM against an existing Chroma collection. + + Loads every document already indexed in ``collection_name`` to build the + BM25 side of hybrid search — there is no BM25 persistence, so it's + rebuilt in memory from Chroma's stored documents on every run. + + Args: + collection_name: Chroma collection to read from. Defaults to "papers". + + Returns: + A Dependencies tuple with a ready-to-use chroma store, bm25 + retriever, and LLM provider. + """ + embedder = LocalEmbedder() + chroma = ChromaVectorStore(embedder=embedder, collection_name=collection_name) + llm = AnthropicLLM() + + raw = ( + chromadb.PersistentClient(path=str(CHROMA_DIR)) + .get_collection(collection_name) + .get(include=["documents", "metadatas"]) + ) + all_docs = [ + Document(doc_id=doc_id, text=text, metadata=metadata) + for doc_id, text, metadata in zip( + raw["ids"], raw["documents"], raw["metadatas"], strict=False + ) + ] + bm25 = BM25Retriever(documents=all_docs) + + return Dependencies(chroma=chroma, bm25=bm25, llm=llm) diff --git a/scripts/eval_retrieval.py b/scripts/eval_retrieval.py index f857433..dddcd3c 100644 --- a/scripts/eval_retrieval.py +++ b/scripts/eval_retrieval.py @@ -21,12 +21,8 @@ import asyncio import json -import sys - -if sys.platform == "linux": - __import__("pysqlite3") - sys.modules["sqlite3"] = sys.modules.pop("pysqlite3") +import _wiring # noqa: F401 # applies the pysqlite3 patch before chromadb is imported below import chromadb from researchos.application.services.retrieval_service import hybrid_rerank_search, hybrid_search diff --git a/scripts/ingest_documents.py b/scripts/ingest_documents.py index c239879..96e11a7 100644 --- a/scripts/ingest_documents.py +++ b/scripts/ingest_documents.py @@ -14,11 +14,8 @@ import argparse import asyncio -import sys -if sys.platform == "linux": - __import__("pysqlite3") - sys.modules["sqlite3"] = sys.modules.pop("pysqlite3") +import _wiring # noqa: F401 # applies the pysqlite3 patch before ingest_papers touches chromadb from researchos.application.services.ingestion_service import ingest_papers from researchos.paths import ensure_dirs diff --git a/scripts/run_research_graph.py b/scripts/run_research_graph.py new file mode 100644 index 0000000..6c3f9dd --- /dev/null +++ b/scripts/run_research_graph.py @@ -0,0 +1,29 @@ +import argparse +import asyncio + +from _wiring import build_dependencies + +from researchos.application.services.retrieval_service import hybrid_rerank_search, hybrid_search +from researchos.domain.models import Document, ResearchContext +from researchos.infrastructure.orchestration.research_graph import build_research_graph + +parser = argparse.ArgumentParser(description="Procesador de consultas.") +parser.add_argument( + "-q", "--query", default="Qué es RLHF?", help="La query(consulta) que deseas procesar" +) +args = parser.parse_args() + +K = 5 +deps = build_dependencies() + + +async def retrieve_hybrid_rerank(query: str) -> list[Document]: + candidates = await hybrid_search(query, retrievers=[deps.chroma, deps.bm25], k=K * 2) + return await hybrid_rerank_search(query=query, llm=deps.llm, documents=candidates, k=K) + + +graph = build_research_graph(retrieve=retrieve_hybrid_rerank, llm=deps.llm) +result = asyncio.run(graph.ainvoke(ResearchContext(query=args.query))) +print(result["query"], "\n") +print(result["documents"], "\n") +print(result["answer"], "\n") diff --git a/scripts/run_telegram_bot.py b/scripts/run_telegram_bot.py index e72d7b9..c017af4 100644 --- a/scripts/run_telegram_bot.py +++ b/scripts/run_telegram_bot.py @@ -1,55 +1,42 @@ -import sys - -if sys.platform == "linux": - __import__("pysqlite3") - sys.modules["sqlite3"] = sys.modules.pop("pysqlite3") - import json import logging from datetime import UTC, datetime -import chromadb +from _wiring import build_dependencies from researchos.application.services.rag_service import answer_query from researchos.application.services.retrieval_service import hybrid_rerank_search, hybrid_search from researchos.config import settings -from researchos.domain.models import Document -from researchos.infrastructure.bot.telegram_bot import AnswerFn, TelegramBot -from researchos.infrastructure.llm.anthropic_llm import AnthropicLLM -from researchos.infrastructure.retrieval.bm25 import BM25Retriever -from researchos.infrastructure.retrieval.chroma import ChromaVectorStore -from researchos.infrastructure.retrieval.embedder import LocalEmbedder -from researchos.paths import CHROMA_DIR, DATA_DIR +from researchos.domain.interfaces import AnswerFn +from researchos.domain.models import Document, ResearchContext +from researchos.infrastructure.bot.telegram_bot import TelegramBot +from researchos.infrastructure.orchestration.research_graph import build_research_graph +from researchos.paths import DATA_DIR query_logger = logging.getLogger("researchos.queries") -# ── Build retrievers ── -COLLECTION_NAME = "papers" K = 5 +deps = build_dependencies() + -embedder = LocalEmbedder() -chroma = ChromaVectorStore(embedder=embedder, collection_name=COLLECTION_NAME) -llm = AnthropicLLM() +async def retrieve_hybrid_rerank(query: str) -> list[Document]: + candidates = await hybrid_search(query, retrievers=[deps.chroma, deps.bm25], k=K * 2) + return await hybrid_rerank_search(query=query, llm=deps.llm, documents=candidates, k=K) -raw = ( - chromadb.PersistentClient(path=str(CHROMA_DIR)) - .get_collection(COLLECTION_NAME) - .get(include=["documents", "metadatas"]) -) -all_docs = [ - Document(doc_id=doc_id, text=text, metadata=metadata) - for doc_id, text, metadata in zip(raw["ids"], raw["documents"], raw["metadatas"], strict=False) -] -bm25 = BM25Retriever(documents=all_docs) +# V1 pipeline (function calls, no LangGraph). Not wired to the bot below — +# kept so T24 can run it and answer_v2_graph side by side over the same queries. +async def answer_v1_pipeline(query: str) -> str: + return await answer_query(query, deps.llm, retrieve=retrieve_hybrid_rerank) -async def retrieve_hybrid_rerank(query: str) -> list[Document]: - candidates = await hybrid_search(query, retrievers=[chroma, bm25], k=K * 2) - return await hybrid_rerank_search(query=query, llm=llm, documents=candidates, k=K) + +# V2 agent (LangGraph), wired to the bot below. +research_graph = build_research_graph(retrieve=retrieve_hybrid_rerank, llm=deps.llm) -async def answer(query: str) -> str: - return await answer_query(query, llm, retrieve=retrieve_hybrid_rerank) +async def answer_v2_graph(query: str) -> str: + result = await research_graph.ainvoke(ResearchContext(query=query)) + return result["answer"] def with_logging(answer_fn: AnswerFn) -> AnswerFn: @@ -81,5 +68,5 @@ async def logged_answer(query: str) -> str: return logged_answer -bot = TelegramBot(token=settings.telegram_bot_token, answer_fn=with_logging(answer)) +bot = TelegramBot(token=settings.telegram_bot_token, answer_fn=with_logging(answer_v2_graph)) bot.run() diff --git a/src/researchos/application/agents/research_agent/nodes.py b/src/researchos/application/agents/research_agent/nodes.py new file mode 100644 index 0000000..e9a1f4b --- /dev/null +++ b/src/researchos/application/agents/research_agent/nodes.py @@ -0,0 +1,57 @@ +"""Research agent nodes — pure functions consumed by the LangGraph assembly. + +Per ADR-005, nodes live in application/ as plain ``(ResearchContext) -> dict`` +functions with zero LangGraph imports. Dependencies (RetrieveFn, LLMProvider) +cannot be passed as extra node parameters — LangGraph only calls a node with +the state — so each node is built by a factory that closes over its +dependency once, in the composition root (infrastructure/orchestration/). +""" + +from collections.abc import Awaitable, Callable +from typing import Any + +from researchos.application.agents.agent_utils import build_rag_messages +from researchos.domain.interfaces import LLMProvider, RetrieveFn +from researchos.domain.models import ResearchContext +from researchos.domain.prompts import PromptTemplate + +NodeFn = Callable[[ResearchContext], Awaitable[dict[str, Any]]] + + +def make_retrieve_node(retrieve: RetrieveFn) -> NodeFn: + """Build a retrieve node bound to a specific retrieval strategy. + + Args: + retrieve: A RetrieveFn implementation (vector-only, hybrid, etc.), + injected as a closure so the node itself stays framework-free. + + Returns: + An async node function that reads ``state.query`` and returns the + partial state update ``{"documents": [...]}`` for LangGraph to merge. + """ + + async def retrieve_node(state: ResearchContext) -> dict[str, Any]: + return {"documents": await retrieve(state.query)} + + return retrieve_node + + +def make_generate_node(llm: LLMProvider) -> NodeFn: + """Build a generate node bound to a specific LLM provider. + + The system prompt is rendered once at construction time, not per call. + + Args: + llm: An LLMProvider implementation, injected as a closure. + + Returns: + An async node function that reads ``state.query``/``state.documents`` + and returns the partial state update ``{"answer": "..."}``. + """ + system_prompt = PromptTemplate("system", "agent").render() + + async def generate_node(state: ResearchContext) -> dict[str, Any]: + messages = build_rag_messages(state.query, state.documents, system_prompt) + return {"answer": await llm.generate(messages)} + + return generate_node diff --git a/src/researchos/application/services/rag_service.py b/src/researchos/application/services/rag_service.py index 9d8b8dc..ae73c36 100644 --- a/src/researchos/application/services/rag_service.py +++ b/src/researchos/application/services/rag_service.py @@ -1,12 +1,7 @@ -from collections.abc import Awaitable, Callable - from researchos.application.agents.agent_utils import build_rag_messages -from researchos.domain.interfaces import LLMProvider -from researchos.domain.models import Document +from researchos.domain.interfaces import LLMProvider, RetrieveFn from researchos.domain.prompts import PromptTemplate -RetrieveFn = Callable[[str], Awaitable[list[Document]]] - async def answer_query(query: str, llm: LLMProvider, retrieve: RetrieveFn) -> str: system_prompt = PromptTemplate("system", "agent").render() diff --git a/src/researchos/domain/interfaces.py b/src/researchos/domain/interfaces.py index f8815c4..a502ab4 100644 --- a/src/researchos/domain/interfaces.py +++ b/src/researchos/domain/interfaces.py @@ -16,11 +16,15 @@ async def upsert(self, documents: list[Document]) -> None: pass """ -from collections.abc import AsyncIterator +from collections.abc import AsyncIterator, Awaitable, Callable from typing import Protocol from .models import Document, Message +RetrieveFn = Callable[[str], Awaitable[list[Document]]] + +AnswerFn = Callable[[str], Awaitable[str]] + class LLMProvider(Protocol): """Contract for any LLM provider (Claude, Gemini, etc.).""" diff --git a/src/researchos/domain/models.py b/src/researchos/domain/models.py index 4d492c8..724df19 100644 --- a/src/researchos/domain/models.py +++ b/src/researchos/domain/models.py @@ -4,6 +4,7 @@ All other layers reference these models. They have zero external dependencies. """ +from dataclasses import dataclass, field from datetime import datetime from pydantic import BaseModel, Field @@ -91,3 +92,18 @@ class AgentOutput(BaseModel): answer: str sources: list[Document] = Field(default_factory=list) metadata: dict = Field(default_factory=dict) + + +@dataclass +class ResearchContext: + """State passed between LangGraph nodes in the research agent (ADR-005). + + A dataclass rather than a Pydantic BaseModel: LangGraph accepts + TypedDict/dataclass/BaseModel as state schemas equally, and a dataclass + avoids re-validating the state on every node's partial update. Revisit + as a BaseModel if cross-field validation becomes necessary. + """ + + query: str + documents: list[Document] = field(default_factory=list) + answer: str = "" diff --git a/src/researchos/infrastructure/bot/telegram_bot.py b/src/researchos/infrastructure/bot/telegram_bot.py index 34b7ac9..a37eb3b 100644 --- a/src/researchos/infrastructure/bot/telegram_bot.py +++ b/src/researchos/infrastructure/bot/telegram_bot.py @@ -1,15 +1,14 @@ """Telegram bot adapter — delivery channel for the RAG engine.""" import logging -from collections.abc import Awaitable, Callable from telegram import Update from telegram.constants import MessageLimit from telegram.ext import ApplicationBuilder, ContextTypes, MessageHandler, filters -logger = logging.getLogger(__name__) +from researchos.domain.interfaces import AnswerFn -AnswerFn = Callable[[str], Awaitable[str]] +logger = logging.getLogger(__name__) def _split_message(text: str, limit: int = MessageLimit.MAX_TEXT_LENGTH) -> list[str]: diff --git a/src/researchos/infrastructure/orchestration/__init__.py b/src/researchos/infrastructure/orchestration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/researchos/infrastructure/orchestration/research_graph.py b/src/researchos/infrastructure/orchestration/research_graph.py new file mode 100644 index 0000000..0fb2644 --- /dev/null +++ b/src/researchos/infrastructure/orchestration/research_graph.py @@ -0,0 +1,51 @@ +"""LangGraph assembly for the research agent. + +Per ADR-005, this is the only module in the project allowed to import +``langgraph`` — ``StateGraph``, edges and ``compile()`` are the one part of +the graph that truly requires the framework. Nodes and state stay +framework-free (see ``application/agents/research_agent/nodes.py`` and +``domain/models.ResearchContext``). +""" + +from langgraph.graph import END, START, StateGraph +from langgraph.graph.state import CompiledStateGraph + +from researchos.application.agents.research_agent.nodes import ( + make_generate_node, + make_retrieve_node, +) +from researchos.domain.interfaces import LLMProvider, RetrieveFn +from researchos.domain.models import ResearchContext + + +def build_research_graph( + retrieve: RetrieveFn, llm: LLMProvider +) -> CompiledStateGraph[ResearchContext, None, ResearchContext, ResearchContext]: + """Assemble and compile the minimal retrieve-then-generate research graph. + + Args: + retrieve: A RetrieveFn implementation, injected into the retrieve node. + llm: An LLMProvider implementation, injected into the generate node. + + Returns: + A compiled LangGraph graph ready to run via ``.ainvoke(ResearchContext(...))``. + """ + builder = StateGraph(ResearchContext) + + # Add nodes. The ignores below are a stub limitation (langgraph==1.2.11, + # mypy==1.20.1), not a real type error: mypy cannot bind add_node's generic + # NodeInputT against a plain async Callable (reproduced with a minimal + # StateGraph outside this project too, including passing input_schema + # explicitly). Tracked upstream as langchain-ai/langgraph#5000 (making + # StateGraph/CompiledStateGraph generic-safe is still open, no target + # version yet) — re-check this ignore next time langgraph is upgraded. + builder.add_node("retrieve", make_retrieve_node(retrieve)) # type: ignore[call-overload] + builder.add_node("generate", make_generate_node(llm)) # type: ignore[call-overload] + + # Add edges to connect nodes + builder.add_edge(START, "retrieve") + builder.add_edge("retrieve", "generate") + builder.add_edge("generate", END) + + # Compile the agent + return builder.compile() diff --git a/tests/unit/application/test_research_agent_nodes.py b/tests/unit/application/test_research_agent_nodes.py new file mode 100644 index 0000000..847bd94 --- /dev/null +++ b/tests/unit/application/test_research_agent_nodes.py @@ -0,0 +1,61 @@ +"""Unit tests for research agent nodes — pure functions, no LangGraph involved. + +Per ADR-005, each node is tested by passing a fabricated ResearchContext and +asserting the returned partial-state dict, with no graph execution needed. +""" + +import pytest + +from researchos.application.agents.research_agent.nodes import ( + make_generate_node, + make_retrieve_node, +) +from researchos.domain.models import Document, ResearchContext +from tests.conftest import MockLLMProvider + + +@pytest.mark.unit +class TestRetrieveNode: + @pytest.mark.asyncio + async def test_returns_documents_from_retrieve_fn(self, sample_documents: list[Document]): + async def fake_retrieve(query: str) -> list[Document]: + assert query == "What is RAG?" + return sample_documents + + retrieve_node = make_retrieve_node(fake_retrieve) + result = await retrieve_node(ResearchContext(query="What is RAG?")) + + assert result == {"documents": sample_documents} + + +@pytest.mark.unit +class TestGenerateNode: + @pytest.mark.asyncio + async def test_returns_llm_answer( + self, mock_llm: MockLLMProvider, sample_documents: list[Document] + ): + generate_node = make_generate_node(mock_llm) + state = ResearchContext(query="What is RAG?", documents=sample_documents) + + result = await generate_node(state) + + assert result == {"answer": "This is a mock response."} + assert len(mock_llm.calls) == 1 + + @pytest.mark.asyncio + async def test_passes_query_and_documents_to_llm( + self, mock_llm: MockLLMProvider, sample_documents: list[Document] + ): + """A node that ignored state.documents would still return an answer — + this asserts on what was actually sent to the LLM, not just the output.""" + generate_node = make_generate_node(mock_llm) + state = ResearchContext(query="What is RAG?", documents=sample_documents) + + await generate_node(state) + + messages = mock_llm.calls[0] + assert messages[0].role == "system" + assert messages[1].role == "user" + assert "What is RAG?" in messages[1].content + for doc in sample_documents: + assert doc.text in messages[1].content diff --git a/tests/unit/infrastructure/test_research_graph.py b/tests/unit/infrastructure/test_research_graph.py new file mode 100644 index 0000000..3447472 --- /dev/null +++ b/tests/unit/infrastructure/test_research_graph.py @@ -0,0 +1,29 @@ +"""Unit tests for the compiled research graph — retrieve and LLM mocked. + +Unlike tests/unit/application/test_research_agent_nodes.py (nodes tested in +isolation), this exercises the assembled graph end to end via .ainvoke(), +confirming the wiring between nodes/edges actually works. +""" + +import pytest +from tests.conftest import MockLLMProvider + +from researchos.domain.models import Document, ResearchContext +from researchos.infrastructure.orchestration.research_graph import build_research_graph + + +@pytest.mark.unit +class TestBuildResearchGraph: + @pytest.mark.asyncio + async def test_retrieve_then_generate_end_to_end( + self, mock_llm: MockLLMProvider, sample_documents: list[Document] + ): + async def fake_retrieve(query: str) -> list[Document]: + return sample_documents + + graph = build_research_graph(retrieve=fake_retrieve, llm=mock_llm) + result = await graph.ainvoke(ResearchContext(query="What is RAG?")) + + assert result["query"] == "What is RAG?" + assert result["documents"] == sample_documents + assert result["answer"] == "This is a mock response."