Pipeline LangGraph, który utrzymuje dokładność pamięci bazy kodu dla agentów AI
Stwórz pipeline LangGraph, który tworzy pliki wiedzy dla poszczególnych usług, filtruje je za pomocą Ragas i DeepEval, śledzi ich wykonywanie w Langfuse oraz pozwala na uruchomienie lokalnie na Ollama.
Źródła: Codebase Memory for AI Agents: A LangGraph Pipeline That Actually Stays Accurate
Każdy, kto używał agentów do kodowania w dużych repozytoriach, napotyka jeden problem: jak agent może zrozumieć bazę kodu, nie czytając ponownie każdego pliku przy każdym zadaniu? Poniżej przedstawiono trzecią próbę rozwiązania tego problemu. Pierwsze dwa podejścia wykorzystywały RAG oparte na embeddingach w docstringach. Obie metody przestały być skuteczne w ciągu miesiąca, ponieważ dokumentacja nie nadążała za każdą zmianą w kodzie. Ta trzecia wersja radzi sobie z zespołem dokonującym trzydziestu lub czterdziestu komitetów dziennie.
To nie jest recenzja formatu plików kogoś innego. Chodzi o sam proces: graf stanów LangGraph, który monitoruje repozytorium, tworzy strukturyzowany plik wiedzy dla każdej usługi, ocenia dokładność swoich wyników przed ich odczytaniem przez agenta oraz śledzi każdy proces, aby koszty były widoczne. Uwzględnione są również błędy z pierwszej próby.
To, co obejmuje: dlaczego problemy z budżetem tokenów w środowiskach kodowania wielu agentów to w rzeczywistości problem przestarzałości danych, pełny projekt LangGraph (schemat stanu, węzły, warunkowe krawędzie, mechanizm checkpointingu w DynamoDB), wykorzystanie Amazon Bedrock z Claude Haiku 4.5 do cytowania, system oceny oparty na Ragas i DeepEval, możliwości monitorowania dzięki samodzielnie hostowanemu Langfuse, wersja całkowicie lokalna Ollamy dla zespołów, które nie mogą wysyłać kodu do modeli chmurowych, hook gitowy łączący wszystko razem oraz analiza możliwych scenariuszy awarii.
Problem
Za każdym razem, gdy agent kodujący rozpoczyna zadanie w repozytorium, nie wie, że obie jego opcje są słabe. Może przeczytać wszystko, co zużywa tokeny na plikach nieskorelowanych z zadaniem. Albo może użyć metod wyszukiwania opartych na embeddingach, które zwracają fragmenty o podobnym znaczeniu, ale nie dają wyraźnej odpowiedzi na pytania o to, kto co wywołuje lub kto jest właścicielem modułu.
Opcją, która działa, jest niewielka liczba starannie przygotowanych, ustrukturyzowanych plików – po jednym na usługę – które określają, czym się zajmuje i od czego zależą, są generowane raz i automatycznie aktualizowane. Agent koordynujący sprawdza indeks, wybiera dwa lub trzy pliki istotne dla zadania, a następnie przekazuje je wyłącznie podagentowi. Chodzi tu zarówno o mądre wykorzystanie tokenów, jak i o reprezentację wiedzy.
Kłopot nie leży w formacie pliku. Markdown w połączeniu z YAML frontmatter nie jest niczym nowym. Trudność polega na mechanizmach, które zapobiegają starzeniu się tych plików w miarę zmian w kodzie, który je stanowi.
Architektura
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
Wykres jest celowo mały – zawiera pięć rzeczywistych węzłów oraz jedną krawędź powtórki. LangGraph zachęca do tworzenia czegoś bardziej skomplikowanego, niż wymagają tego zadania. Pierwsza próba z jedenaścioma węzłami okazała się niemożliwa do wyjaśnienia, ponieważ nikt nie potrafił powiedzieć, dlaczego dany plik istnieje.
Stack i powód obecności każdej jego części
+------------------+---------------------------+--------------------------------------+
| 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 |
+------------------+---------------------------+--------------------------------------+
Krok 1: schemat stanu
We wszystkim w LangGraph zaczyna się od kształtu stanu przekazywanego pomiędzy węzłami. Jeśli popełnisz błąd, każdy kolejny węzeł będzie musiał zgadywać, jaki stan wygenerował poprzedni.
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"]
Nie ma tam nic skomplikowanego. Kluczowym polem jest retry_count, ponieważ zapobiega ono nieskutecznym próbom niekończącego się przetwarzania przez mechanizm walidacji.
Krok 2: ograniczenie skanowania do różnic
Skanowanie całego repozytorium przy każdym komitowaniu sprawia, że takie procesy są zbyt kosztowne do ciągłego uruchamiania. Ograniczenie skanowania do różnic sprawia, że stają się wystarczająco tanie do wykonywania przy każdym komitowaniu, a nie tylko raz dziennie.
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
Przyporządkowywanie plików do pakietów to jedyna kwestia, która jest naprawdę specyficzna dla każdego repozytorium. Monorepo z uporządkowaną strukturą services// wymaga około trzech wierszy kodu. Rozległe repozytorium typu legacy wymaga osobnego mapowania właścicieli, najlepiej uzyskanego poprzez analizę pliku CODEOWNERS, a nie przez domyślanie się na podstawie ścieżek.
Krok 3: opracowanie projektu za pomocą Bedrock
Ten proces kosztuje prawdziwe pieniądze, więc wymaga starannego podejścia. Claude Haiku 4.5 na Bedrock działa szybko i jest wystarczająco tani, by być używanym przy każdym commitie, natomiast większe modele są przeznaczone wyłącznie do procesu weryfikacji przez człowieka, a nie do rutynowych kroków opracowywania projektu.
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
Jedno polecenie w tym promptzie naprawdę działa: nie wymyślaj zależności, których nie widać w kodzie. Jeśli nic nie zrobić, Haiku z pewnością opisze zależność od usługi, której nie ma. Podobno ta jedna zdanie zmniejszyło liczbę fałszywych zależności ze średnio jednego pliku na sześć do niemal zeru podczas testów. Nie traktuj tego jako uniwersalnej zasady, ponieważ zależy to od ilości kodu zawartego w promptzie.
Krok 4: brama ewaluacji, którą większość osób pomija
Plik markdown z polem typu frontmatter nie może wiedzieć, czy mówi prawdę. Nic w tym formacie nie zapobiega temu, by model językowy napisał pewny siebie, dobrze sformatowany, ale błędny opis usługi. Pominięcie tego etapu może sprawić, że system potajemnie zniekształci rozumienie agenta jego własnego kodu, wyglądając przy tym dokładnie tak samo wiarygodnie jak poprawny przypadek.
Ragas oblicza wynik wierności, czyli to, na jak dobrą podstawę opiera się projekt w odniesieniu do oryginalnego źródła, bez konieczności posiadania ręcznie napisanej odpowiedzi referencyjnej. DeepEval przekształca tę wartość w coś, na czym może oprzeć się system CI.
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
Progiem 0,75 jest punkt wyjścia, a nie prawo. Został on podniesiony z 0,6 po dwóch wersjach projektu, w których przeszły nieco błędne zgłoszenia dotyczące zależności. Zespół stosujący ten próg powinien spodziewać się poświęcenia tygodnia na obserwację własnych wskaźników fałszywie pozytywnych i fałszywie negatywnych, zanim zaufa tym wartościom.
Krok 5: łączenie wykresu z punktami kontrolnymi 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)
Wybór DynamoDB nie jest przypadkowy. Proces pipeline wykonywany jest w środowisku CI, gdzie procesory są częściej zamykane w trakcie działania, niż przyznają to ludzie. Graf z punktami kontrolnymi kontynuuje pracę od ostatniego ukończonego superkroku, zamiast przepisywać pięć plików tylko dlatego, że szósty przekroczył czas wykonywania.
Krok 6: sprawdzenie kosztów
Do procesu pipeline nie można było mieć pełnego zaufania, dopóki nie były widoczne ilość użytych tokenów oraz czas trwania każdego węzła. Dekorator @observe() z Langfuse umożliwił to w ciągu pięciu minut, bez konieczności całkowitej przebudowy rozwiązania.
from langfuse.decorators import observe
@observe()
def draft_concept(state: PipelineState) -> PipelineState:
# same body as above, now traced automatically
...
Aby uruchomić rozwiązanie lokalnie, potrzebny jest jedynie plik docker-compose, co ma znaczenie wtedy, gdy kod, włączając różnice pokazywane w komendach, musi pozostać w obrębie Twojej sieci:
git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up -d
Po przejęciu projektu przez ClickHouse, wersja samodzielnie hostowana Langfuse przechowuje dane zarówno w Postgres, jak i w ClickHouse, przy czym samodzielne hostowanie wydaje się równie niezawodne jak wcześniej. Warto to sprawdzić przed wdrożeniem narzędzi do monitorowania w środowisku produkcyjnym, ponieważ pytanie o to, czy narzędzie zostanie porzucone, czy będzie dostępne tylko za opłatą, jest uzasadnione.
Panel sterowania pokazuje trzy informacje: koszt tokena za każdą zmianę, rozkład dokładności w czasie oraz częstotliwość ponawiania prób. Powolny spadek dokładności wskazuje na błędy w logice odczytu źródeł, a nie na pogorszenie się jakości modelu. Częstotliwość ponawiania prób powyżej około 10% zawsze wynikała z problemów z promptem, a nie z samego modelu.
Wersja lokalna bez chmury
Część zespołów nie może nawet wysyłać różnic do modelu hostowanego w chmurze. Wystarczy zastąpić Ollamę węzłem do tworzenia projektów, a ten sam schemat będzie działał bez żadnych innych zmian w architekturze.
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
Dla wyszukiwania opartego na embeddingach w opublikowanym zestawie plików, przydatnego gdy jest ich ponad sto, a sam index.md nie wystarcza, nomic-embed-text dostępny przez Ollama to model, którego już używają większość lokalnych rozwiązań RAG, więc nie trzeba uzasadniać dodawania nowych zależności.
+------------------+------------------------+---------------------------+
| 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) |
+------------------+------------------------+---------------------------+
W testach na tych samych plikach źródłowych dokładność llama3.1:8b była o około 10 do 15 punktów niższa niż w przypadku Haiku 4.5, głównie dlatego, że ten model trudniej radzi sobie z opisywaniem zależności. To właśnie ta różnica jest powodem istnienia mechanizmu oceny niezależnie od tego, który model jest używany. Nie chcesz przecież dowiedzieć się o różnicy jakości dopiero po trzech tygodniach od zdezorientowanego agenta.
Hook 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)"
Zastosuj to na dwa sposoby: jako hook po zatwierdzeniu zmian na komputerze programisty oraz jako obowiązkowe zadanie w CI na wspólnym branchu. Lokalnie wykrywa odchylenia jeszcze przed utworzeniem PR. W środowisku CI stanowi punkt egzekwowania zasad, ponieważ niektórzy współpracownicy nigdy nie instalują tego hooka, a hook, którego nie da się egzekwować, jest jedynie zaleceniem.
Kontrola rzeczywistości w produkcji
Oto co warto wiedzieć przed rozpoczęciem pracy nad tym rozwiązaniem.
- Koszt jest rzeczywisty, ale niewielki. Dla repozytorium z około 40 komitami dziennie, dotyczących po dwa lub trzy pakiety każdy, koszt przeprowadzenia testu wersji roboczej Haiku wynosi około kilkunastu dolarów miesięcznie, a koszt przeprowadzenia oceny jest mniej więcej dwukrotnie wyższy, ponieważ każda wersja robocza jest oceniana w porównaniu z jej źródłem. To niska cena za uniknięcie dostarczania błędnej dokumentacji do agenta.
needs_review w maszynie stanów.Co budować dalej
Obecnie mechanizm eval ocenia wierność tylko wtedy, gdy napisany zostanie szkic, i nigdy nie sprawdza ponownie opublikowanego pliku. Naturalnym następnym krokiem jest zaplanowane ponowne ocenienie całego pakietu na podstawie aktualnego źródła, aby wykryć pliki koncepcyjne, które były poprawne w momencie stworzenia, ale uległy błędom po trzech modyfikacjach, przy czym nie został uruchomiony proces ponownego tworzenia szkicu. Coś takiego jeszcze nie zostało zaimplementowane.
Progi oceny to liczby, które każdy dostosowuje inaczej i których prawie nikt nie publikuje, więc jeśli stworzysz coś podobnego, śledź, na jakim poziomie znajdzie się twoja ocena i czy nadal będzie aktualna po pierwszych kilku tygodniach.
Pozycje pokrewne
- Praktyczne notatki: LangGraph Memory: Pełny praktyczny przewodnik po zarządzaniu — Szczegółowy opis Praktycznych notatek: LangGraph Memory: Pełny praktyczny przewodnik po zarządzaniu, w tym kontrakty, sprawdzenia oraz miejsca na kod do wstawienia dla zespołów wdrażających ten wzorzec.
- Harness Engineering: Sprzężenie ruchu, które przekształca LLM w agenta — Co naprawdę sprawia, że agent programistyczny funkcjonuje: pętla, mechanizm wysyłania narzędzi odporny na błędne wywołania oraz uprawnienia typu allow/ask/deny, przedstawione przy użyciu około 600 linii zwykłego Pythona.