Un pipeline de LangGraph que mantiene precisa la memoria del código base para los agentes de IA
Construir un pipeline de LangGraph que cree archivos de conocimiento por servicio, los valide con Ragas y DeepEval, haga un seguimiento de las ejecuciones en Langfuse y también las ejecute localmente en Ollama.
Fuentes: Codebase Memory for AI Agents: A LangGraph Pipeline That Actually Stays Accurate
Cualquiera que haya utilizado agentes de codificación en un repositorio grande se enfrenta a una pregunta: ¿cómo puede un agente comprender la base de código sin tener que leer cada archivo nuevamente para cada tarea? Lo que sigue es el tercer intento por resolverlo. Los intentos uno y dos emplearon RAG basado en embeddings sobre las docstrings. Ambos dejaron de funcionar en el plazo de un mes, ya que las documentaciones se quedaban atrás con cada cambio en el código. Esta tercera versión ha funcionado bien incluso con un equipo que realiza treinta o cuarenta commits al día.
No se trata de una revisión del formato de archivo de otra persona. Se refiere al propio pipeline: un grafo de estado LangGraph que monitorea el repositorio, elabora un archivo de conocimiento estructurado para cada servicio, evalúa la precisión de sus propios resultados antes de que cualquier agente los lea y registra cada ejecución para hacer visible el costo. También se incluyen los fallos del primer intento.
Qué abarca: por qué los problemas de presupuesto de tokens en entornos de programación multiagente son en realidad un problema de obsolescencia, el diseño completo de LangGraph (esquema de estados, nodos, bordes condicionales, puntos de control en DynamoDB), una prueba de citación en Amazon Bedrock con Claude Haiku 4.5, una puerta de evaluación basada en Ragas y DeepEval, capacidad de observabilidad con Langfuse autohospedado, una variante completamente local en Ollama para equipos que no pueden enviar código a un modelo en la nube, el gancho de git que lo integra todo y una revisión de los modos de fallo.
El problema
Cada vez que un agente de programación inicia una tarea en un repositorio, no sabe que ambas opciones son deficientes. Puede leer todo, lo cual consume tokens en archivos no relacionados con la tarea. O puede utilizar recuperación basada en embeddings, lo cual devuelve fragmentos semánticamente similares pero no ofrece una respuesta explícita sobre quién llama a qué o quién es el propietario de un módulo.
La alternativa viable consiste en un conjunto reducido de archivos estructurados y cuidadosamente seleccionados, uno por servicio, que indican cuál es su función y en qué dependen; estos archivos se generan una vez y se actualizan automáticamente. Un agente coordinador consulta un índice, selecciona los dos o tres archivos relevantes para la tarea y luego envía únicamente esos a un subagente. Se trata tanto de utilizar los tokens con inteligencia como de representar el conocimiento de manera eficaz.
El formato del archivo no es donde radica la dificultad. Markdown junto con frontmatter en YAML no es nada novedoso. La verdadera dificultad radica en el mecanismo que impide que esos archivos se vuelvan obsoletos a medida que cambia el código subyacente.
La arquitectura
git push / commit
|
v
+-------------------------+
| scan_diff (node) |
| git diff --name-only |
| scope to changed pkgs |
+-------------------------+
|
v
+-------------------------+
| draft_concept (node) |
| Bedrock Claude Haiku |
| one file per service |
+-------------------------+
|
v
+-------------------------+
| add_citations (node) |
| link to runbooks/PRs |
+-------------------------+
|
v
+-------------------------+
| eval_gate (node) |
| Ragas faithfulness |
| DeepEval assert_test |
+-------------------------+
pass | | fail
| +----> re-draft (max 2 retries) --> human review queue
v
+-------------------------+
| relink (node) |
| fix cross-references |
+-------------------------+
|
v
+-------------------------+
| publish (node) |
| commit bundle to repo |
+-------------------------+
every node emits a Langfuse trace
every super-step checkpoints to DynamoDB
El grafo es deliberadamente pequeño, con cinco nodos reales y un arista de intento repetido. LangGraph fomenta la creación de estructuras más complejas de las que realmente requiere el problema. Un primer intento con once nodos resultó imposible de explicar, ya que nadie podía indicar por qué existía un archivo determinado.
La pila y por qué cada componente está allí
+------------------+---------------------------+--------------------------------------+
| Layer | Tool | Why this one |
+------------------+---------------------------+--------------------------------------+
| Orchestration | LangGraph | Explicit state, conditional retries, |
| | | resumable from checkpoint |
| Draft model | AWS Bedrock, Claude | Cheap enough to run on every commit, |
| | Haiku 4.5 | fast enough not to block CI |
| Checkpoint store | DynamoDB | Single-digit ms reads, TTL for expiry, |
| | (langgraph-checkpoint-aws)| survives a crashed worker mid-run |
| Eval, faithfulness| Ragas | Reference-free scoring of a generated |
| | | file against the actual code diff |
| Eval, CI gate | DeepEval | pytest-native, hard-fails a bad draft |
| | | instead of silently publishing it |
| Observability | Langfuse (self-hosted) | Per-node token/cost/latency traces, |
| | | MIT licensed, own your own data |
| Local alternative| Ollama (llama3.1:8b, | Same pipeline, zero cloud calls, for |
| | nomic-embed-text) | air-gapped or cost-sensitive teams |
+------------------+---------------------------+--------------------------------------+
Paso 1: el esquema de estado
Todo en LangGraph comienza con la estructura del estado que se transmite entre nodos. Si se equivoca, cada nodo posterior tendrá que adivinar qué produjo el anterior.
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, END
class ConceptDraft(TypedDict):
path: str # e.g. "services/billing-service.md"
concept_type: str # "Service", "Runbook", "API Endpoint", ...
title: str
body: str # the drafted markdown body
citations: list[str]
faithfulness_score: float
retry_count: int
class PipelineState(TypedDict):
repo_root: str
commit_sha: str
changed_files: list[str]
drafts: list[ConceptDraft]
failed_drafts: list[ConceptDraft]
status: Literal["scanning", "drafting", "evaluating", "publishing", "done", "needs_review"]
No hay nada complicado en eso. El campo importante es retry_count, ya que evita que un borrador deficiente siga intentándose una y otra vez frente a la barrera de evaluación.
Paso 2: limitar el escaneo al diferencial
Escanear todo el repositorio en cada commit es lo que hace que pipelines como este sean demasiado costosos para ejecutarse de forma continua. Limitar el escaneo al diferencial los hace lo suficientemente económicos como para aplicarse en cada commit y no solo en ejecuciones nocturnas.
import subprocess
def scan_diff(state: PipelineState) -> PipelineState:
result = subprocess.run(
["git", "diff", "--name-only", f"{state['commit_sha']}~1", state["commit_sha"]],
cwd=state["repo_root"], capture_output=True, text=True, check=True
)
changed = [f for f in result.stdout.splitlines() if f.endswith((".py", ".go", ".ts"))]
# map file paths to the service/package that owns them
packages = sorted({f.split("/")[1] for f in changed if "/" in f})
state["changed_files"] = packages
state["status"] = "drafting"
return state
Asignar archivos a paquetes es el único aspecto que realmente depende de cada repositorio. Un monorepo con una estructura ordenada services// necesita aproximadamente tres líneas. Un repositorio heredado y complejo requiere un mapa de propietarios mantenido por separado, idealmente al analizar un archivo CODEOWNERS y no basándose en las rutas.
Paso 3: redactar con Bedrock
Este proceso cuesta dinero real, por lo que merece un manejo cuidadoso. Claude Haiku 4.5 en Bedrock es lo suficientemente rápido y económico como para ejecutarse con cada commit, mientras que un modelo más grande se reserva únicamente para el proceso de revisión humana, nunca para la fase rutinaria de redacción.
import boto3
import json
bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")
MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0" # cross-region inference profile
DRAFT_PROMPT = """You are documenting a software service for an AI coding agent's memory.
Given the file contents below, write a concept file with:
- A one-paragraph Responsibilities section
- A Dependencies section listing what this service calls and what calls it
- Nothing else. Do not invent dependencies you cannot see in the code.
Files:
{file_contents}
"""
def draft_concept(state: PipelineState) -> PipelineState:
drafts = []
for package in state["changed_files"]:
contents = read_package_source(state["repo_root"], package)
response = bedrock.invoke_model(
modelId=MODEL_ID,
body=json.dumps({
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 800,
"messages": [{"role": "user", "content": DRAFT_PROMPT.format(file_contents=contents)}]
})
)
body = json.loads(response["body"].read())
text = body["content"][0]["text"]
drafts.append(ConceptDraft(
path=f"services/{package}.md",
concept_type="Service",
title=package,
body=text,
citations=[],
faithfulness_score=0.0,
retry_count=0
))
state["drafts"] = drafts
state["status"] = "evaluating"
return state
Una instrucción en ese prompt realmente funciona: no invente dependencias que no se puedan ver en el código. Dejarlo así, Haiku describirá con confianza una dependencia de un servicio que no existe. Se dice que esa sola oración redujo la tasa de dependencias falsas de aproximadamente un archivo cada seis a casi cero en las pruebas. No lo considere una constante universal, ya que depende de cuánto código se incluya en el prompt.
Paso 4: la puerta de evaluación que la mayoría omite
Un archivo markdown con un campo de tipo frontmatter no puede saber si dice la verdad. Nada en ese formato impide que un LLM escriba una descripción confiada, bien formateada y errónea de un servicio. Si se omite esta etapa, el sistema puede corromper silenciosamente la comprensión que un agente tiene de su propio códigobase, al tiempo que parece tan fiable como uno correcto.
Ragas se encarga de calcular la puntuación y DeepEval actúa como filtro de aprobación/rechazo, ya que abordan problemas diferentes. Ragas genera una puntuación de fidelidad que indica hasta qué punto el borrador está basado en la fuente original, sin necesidad de una respuesta de referencia escrita a mano. DeepEval convierte ese número en algo con el que CI pueda actuar.
from ragas.metrics import faithfulness
from ragas import evaluate
from datasets import Dataset
def score_faithfulness(draft: ConceptDraft, source_contents: str) -> float:
eval_dataset = Dataset.from_dict({
"question": ["Describe this service's responsibilities and dependencies."],
"answer": [draft["body"]],
"contexts": [[source_contents]],
})
result = evaluate(eval_dataset, metrics=[faithfulness])
return result["faithfulness"][0]
from deepeval import assert_test
from deepeval.test_case import LLMTestCase
from deepeval.metrics import FaithfulnessMetric
def eval_gate(state: PipelineState) -> PipelineState:
passed, failed = [], []
for draft in state["drafts"]:
source = read_package_source(state["repo_root"], draft["title"])
score = score_faithfulness(draft, source)
draft["faithfulness_score"] = score
test_case = LLMTestCase(
input="Describe this service.",
actual_output=draft["body"],
retrieval_context=[source]
)
metric = FaithfulnessMetric(threshold=0.75)
try:
assert_test(test_case, [metric])
passed.append(draft)
except AssertionError:
draft["retry_count"] += 1
failed.append(draft)
state["drafts"] = passed
state["failed_drafts"] = failed
state["status"] = "publishing" if not failed else "needs_review"
return state
El umbral de 0.75 es un punto de partida y no una ley. Se elevó de 0.6 después de dos versiones iniciales en las que se aprobaron reclamaciones de dependencia ligeramente incorrectas. El equipo que adopte este valor debe estar preparado para pasar una semana analizando sus propias tasas de falsos positivos y falsos negativos antes de confiar en esa cifra.
Paso 5: conectar el gráfico con puntos de control de DynamoDB
from langgraph_checkpoint_aws import DynamoDBSaver
checkpointer = DynamoDBSaver(
table_name="codebase-memory-checkpoints",
ttl_seconds=60 * 60 * 24 * 7 # expire checkpoints after a week
)
graph = StateGraph(PipelineState)
graph.add_node("scan_diff", scan_diff)
graph.add_node("draft_concept", draft_concept)
graph.add_node("eval_gate", eval_gate)
graph.add_node("relink", relink)
graph.add_node("publish", publish)
graph.set_entry_point("scan_diff")
graph.add_edge("scan_diff", "draft_concept")
graph.add_edge("draft_concept", "eval_gate")
graph.add_conditional_edges(
"eval_gate",
lambda s: "relink" if s["status"] == "publishing" else END,
{"relink": "relink", END: END}
)
graph.add_edge("relink", "publish")
graph.add_edge("publish", END)
pipeline = graph.compile(checkpointer=checkpointer)
La elección de DynamoDB no es arbitraria. El pipeline se ejecuta dentro de CI, donde los procesos suelen terminar abruptamente durante su ejecución con más frecuencia de lo que la gente admite. Un grafo con puntos de control reanuda su ejecución desde el último paso supercompletado, en lugar de tener que redactar cinco archivos solo porque el sexto se agotó el tiempo asignado.
Paso 6: conocer el costo
El pipeline no se consideró fiable hasta que se pudieron ver el uso de tokens y la duración de cada nodo. El decorador @observe() de Langfuse convirtió eso en una adición de cinco minutos en lugar de requerir un rediseño completo.
from langfuse.decorators import observe
@observe()
def draft_concept(state: PipelineState) -> PipelineState:
# same body as above, now traced automatically
...
El autoalojamiento solo requiere un archivo docker-compose, lo cual es necesario si su código, incluso las diferencias dentro de los comandos, debe permanecer dentro de su red:
git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up -d
Después de que ClickHouse adquirió el proyecto, Langfuse autohospedado mantiene registros tanto en Postgres como en ClickHouse, y su uso autohospedado parece tan sólido como antes. Vale la pena confirmarlo antes de integrar la capacidad de observabilidad en el entorno de producción, ya que es lógico preguntarse si una herramienta será descartada o pondrá un muro de pago.
El panel muestra tres cosas: el costo del token por commit, la distribución de fidelidad a lo largo del tiempo y la tasa de intentos repetidos. Una lenta disminución en la fidelidad indica un problema en la lógica de lectura de fuentes y no que el modelo esté empeorando. Una tasa de intentos repetidos superior al 10% se ha atribuido siempre a la prompt, nunca al modelo.
Una versión local sin nube
Algunos equipos ni siquiera pueden enviar diferencias a un modelo alojado en la nube. Basta con reemplazar el nodo de draft por Ollama para que el mismo grafo funcione sin ningún otro cambio en la arquitectura.
import ollama
def draft_concept_local(state: PipelineState) -> PipelineState:
drafts = []
for package in state["changed_files"]:
contents = read_package_source(state["repo_root"], package)
response = ollama.chat(
model="llama3.1:8b",
messages=[{"role": "user", "content": DRAFT_PROMPT.format(file_contents=contents)}]
)
drafts.append(build_draft(package, response["message"]["content"]))
state["drafts"] = drafts
return state
Para realizar búsquedas basadas en incrustación sobre el paquete publicado, algo útil cuando hay más de cien archivos de concepto y index.md por sí solo no es suficiente, nomic-embed-text a través de Ollama es el modelo que ya utilizan la mayoría de las configuraciones locales de RAG, por lo que no es necesario justificar ninguna nueva dependencia.
+------------------+------------------------+---------------------------+
| Component | Cloud version | Local version |
+------------------+------------------------+---------------------------+
| Draft model | Bedrock Claude Haiku | Ollama llama3.1:8b |
| Embeddings | Bedrock Titan Embed | Ollama nomic-embed-text |
| Checkpoint store | DynamoDB | SQLite (LangGraph built-in)|
| Observability | Langfuse (self-hosted) | Langfuse (self-hosted) |
+------------------+------------------------+---------------------------+
En pruebas con los mismos archivos de origen, la fidelidad de llama3.1:8b resultó aproximadamente 10 a 15 puntos por debajo de Haiku 4.5, principalmente porque este último exige descripciones de dependencias más detalladas. Esa diferencia es la razón por la cual existe la verificación de evaluación independientemente del modelo utilizado. No se quiere descubrir una diferencia en calidad tres semanas después a través de un agente confundido.
El gancho git
#!/usr/bin/env bash
# .git/hooks/post-commit
set -euo pipefail
COMMIT_SHA=$(git rev-parse HEAD)
python -m codebase_memory.pipeline --commit "$COMMIT_SHA" --repo-root "$(git rev-parse --show-toplevel)"
Ejecútelo de ambas maneras: como un gancho post-commit en la máquina del desarrollador y como una tarea de CI obligatoria en la rama compartida. A nivel local, detecta desviaciones antes de que exista una PR. En CI funciona como punto de control, ya que algunos colaboradores nunca instalan el gancho, y un gancho que no se puede hacer cumplir es simplemente un consejo.
Una verificación de la realidad en producción
Aquí está lo que vale la pena saber antes de crear esto.
- El costo es real pero moderado. Para un repositorio con unos 40 commits al día que modifican dos o tres paquetes cada uno, la prueba de borrador de Haiku cuesta alrededor de unas decenas de dólares al mes, y la prueba de evaluación duplica más o menos esa cantidad, ya que cada borrador se evalúa en comparación con su código fuente. Es un precio bajo para evitar enviar documentación incorrecta a un agente.
needs_review en la máquina de estados.Qué construir a continuación
Actualmente, la herramienta de evaluación solo considera la fidelidad cuando se escribe un borrador y nunca vuelve a analizar un archivo ya publicado. El siguiente paso lógico es realizar una reevaluación programada de todo el conjunto de archivos en relación con el código fuente actual, con el fin de detectar archivos conceptuales que eran precisos en su momento pero que fallaron después de tres refactorizaciones sin que se activara una nueva versión. Eso aún no se ha implementado.
Los umbrales de evaluación son valores que cada uno ajusta de manera diferente y casi nadie los publica, así que si construyes algo similar, monitorea dónde se sitúa el tuyo y si sigue siendo válido después de unas pocas semanas.
Leer más
- Notas prácticas: LangGraph Memory: La guía práctica completa para gestionar — Guía paso a paso de las Notas prácticas: LangGraph Memory: La guía práctica completa para gestionar: contratos, verificaciones y espacios de código reutilizable para equipos que implementan este patrón.
- Harness Engineering: El runtime que convierte un LLM en un agente — Qué hace realmente funcionar a un agente de programación: el bucle, el envío de herramientas que funciona incluso ante llamadas defectuosas, y los permisos de permitir/pedir/denegar, mostrado con unas 600 líneas de Python puro.