Пайплайн LangGraph, сохраняющий точность данных кодовой базы для ИИ-агентов
Создайте пайплайн LangGraph, который формирует файлы знаний для каждого сервиса, проверяет их с помощью Ragas и DeepEval, отслеживает выполнение в Langfuse, а также запускается локально на Ollama.
Источники: Codebase Memory for AI Agents: A LangGraph Pipeline That Actually Stays Accurate
Любой, кто использовал инструменты для обработки кода в больших репозиториях, сталкивается с одним вопросом: как агент может ознакомиться с кодовой базой, не перечитывая каждый файл для каждого задания? Ниже приведена третья попытка решения этой проблемы. Первые две попытки основывались на технологии RAG с использованием эмбеддингов документации; обе они потеряли эффективность в течение месяца, поскольку документация отставала от каждой изменения в коде. Эта третья версия справляется даже с командой, делающей по тридцать–сорок коммитов в день.
Речь идет не о ревизии формата файлов кого-то другого, а об самом процессе: граф состояний LangGraph отслеживает репозиторий, создает структурированный файл с знаниями для каждого сервиса, оценивает точность своего вывода перед тем, как его прочитает агент, и отслеживает каждую операцию, чтобы можно было увидеть затраты. Также учитываются ошибки, возникшие при первой попытке обработки.
Что охватывается: почему проблема бюджета токенов в средах многопроцессного программирования на самом деле является проблемой устаревших данных, полная структура LangGraph (схема состояний, узлы, условные рёбра, сохранение состояния в DynamoDB), пример использования Amazon Bedrock с Claude Haiku 4.5 для цитирования, механизм оценки, основанный на Ragas и DeepEval, возможности отслеживания с помощью самостоятельно размещённого Langfuse, полностью локальная версия Ollama для команд, которые не могут отправлять код в облачные модели, гит-хук, объединяющий все компоненты, и анализ возможных способов сбоев.
Проблема
Каждый раз, когда агент для программирования начинает выполнение задачи в репозитории, он не знает, что оба его варианта являются неидеальными. Он может прочитать всё, что находится в репозитории, что приводит к трате токенов на файлы, не связанные с задачей. Или же он может использовать поиск на основе эмбеддингов, который возвращает семантически похожие фрагменты текста, но не даёт чёткого ответа о том, кто вызывает что или кому принадлежит тот или иной модуль.
Альтернативой, которая работает, является небольшое количество тщательно подобранных и структурированных файлов — по одному на каждый сервис, в которых указано, что он делает и от чего зависит; эти файлы генерируются один раз и автоматически обновляются. Агент-оркестратор запрашивает индекс, выбирает два-три файла, соответствующих задаче, и затем передает только их подагенту. Здесь важно не только правильное использование токенов, но и точное представление знаний.
Проблема не в формате файлов. Markdown с YAML-метаданными — это вовсе не что-то новое. Сложность заключается в механизмах, которые предотвращают устаревание этих файлов при изменении кода, находящегося внутри них.
Архитектура
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
Граф преднамеренно мал: в нем пять реальных узлов и одна ребро-попытка. LangGraph побуждает создавать что-то более сложное, чем требуется для решения задачи. Первая попытка с одиннадцатью узлами оказалась невозможной для объяснения, поскольку никто не мог сказать, почему существует тот или иной файл.
Стек и причина наличия каждого его элемента
+------------------+---------------------------+--------------------------------------+
| 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 |
+------------------+---------------------------+--------------------------------------+
Шаг 1: схема состояния
В LangGraph всё начинается с формы состояния, передаваемого между узлами. Если она неверна, каждый последующий узел вынужден гадать, что произвёл предыдущий узел.
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"]
Там нет ничего сложного. Важным является поле retry_count, поскольку оно предотвращает бесконечное повторение плохих вариантов проекта при проверке.
Шаг 2: ограничить сканирование только различиями
Полное сканирование всего репозитория при каждом коммите делает подобные пайплайны слишком дорогими для постоянной работы. Ограничение сканирования только различиями делает их достаточно дешевыми для каждого коммита, а не только для еженочных запусков.
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
Сопоставление файлов с пакетами — это единственный аспект, который действительно является уникальным для каждого репозитория. Для организованного монорепозитория с структурой services// достаточно примерно трех строк. В крупном репозитории с устаревшей структурой требуется отдельная карта владельцев, желательно полученная путем обработки файла CODEOWNERS, а не на основе предположений, основанных на путях.
Шаг 3: разработка с помощью Bedrock
Этот этап стоит реальных денег, поэтому к нему следует относиться тщательно. Claude Haiku 4.5 на Bedrock работает достаточно быстро и дешево для выполнения при каждом коммите, а более мощные модели используются только в случае необходимости проведения проверки человеком, никогда в обычном процессе разработки.
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
В том запросе есть одно указание, которое действительно помогает: не выдумывайте зависимости, которые невозможно увидеть в коде. Если оставить всё как есть, Haiku с уверенностью будет описывать зависимость от сервиса, который на самом деле не существует. Говорят, что эта одна фраза сократила частоту появления ложных зависимостей с примерно одного файла из шести до практически нуля во время тестирования. Не воспринимайте это как абсолютную константу, поскольку это зависит от объема исходного кода, указанного в запросе.
Шаг 4: проверка, которую большинство пропускает
Файл в формате Markdown с полем типа frontmatter не может знать, говорит ли он правду. Ничто в этом формате не мешает большой языковой модели написать уверенное, хорошо оформленное, но неверное описание сервиса. Если пропустить этот этап, система может незаметно исказить понимание агентом собственной базы кода, при этом выглядя так же надежно, как и корректное решение.
Ragas вычисляет оценку верности, то есть степень соответствия черновика исходному тексту, при этом не требуется ручного ответа-ориентира. DeepEval преобразует этот показатель в формат, пригодный для использования системой 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
Порог в 0,75 является лишь отправной точкой, а не законом. Он был повышен с 0,6 после того, как два проекта с косвенно неверными утверждениями о зависимостях прошли проверку. Команда, принимающая этот показатель, должна ожидать, что ей потребуется неделя на анализ своих показателей ложноположительных и ложноотрицательных результатов, прежде чем она сможет доверять этой цифре.
Шаг 5: подключение графика с использованием контрольных точек 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)
Выбор DynamoDB не является произвольным. Пайплайн выполняется в среде CI, где рабочие процессы чаще прерываются в процессе работы, чем признают это многие. Граф с точками контроля возобновляется с последнего завершённого супершага, вместо того чтобы пересоздавать пять файлов только из-за того, что шестой не завершился в установленное время.
Шаг 6: определение стоимости
В пайплайн начали верить только тогда, когда стало видно, сколько токенов используется и сколько времени занимает каждый узел. Декоратор @observe() от Langfuse позволил реализовать это за пять минут, без необходимости полной переработки кода.
from langfuse.decorators import observe
@observe()
def draft_concept(state: PipelineState) -> PipelineState:
# same body as above, now traced automatically
...
Для самостоятельной развертки требуется лишь файл docker-compose, что важно, если ваш код, включая различия во вводимых данных, должен оставаться внутри вашей сети:
git clone https://github.com/langfuse/langfuse.git
cd langfuse
docker compose up -d
После того как ClickHouse приобрел проект, самостоятельно развернутый Langfuse сохраняет записи как в Postgres, так и в ClickHouse, причем его работа кажется не менее надежной, чем раньше. Стоит это проверить перед внедрением инструментов мониторинга в производство, поскольку вопрос о том, будет ли инструмент устаревшим или станет платным, вполне обоснован.
На панели управления отображаются три показателя: стоимость токена за каждый коммит, динамика точности с течением времени и частота повторных попыток. Медленное снижение точности указывает на ошибки в логике обработки исходного текста, а не на ухудшение качества модели. Частота повторных попыток выше примерно 10% всегда связана с запросом, а не с самой моделью.
Локальная версия без облака
Некоторые команды не могут даже отправлять различия в изменениях на хостинговую модель. Достаточно заменить узел обработки на Ollama, и та же структура будет работать без каких-либо изменений в архитектуре.
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
Для поиска с использованием встроенных механизмов в опубликованном пакете, что полезно, когда количество файлов с концептами превышает сто, и одного лишь index.md недостаточно, модель nomic-embed-text через Ollama уже используется в большинстве локальных решений типа RAG, поэтому нет необходимости обосновывать добавление новой зависимости.
+------------------+------------------------+---------------------------+
| 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) |
+------------------+------------------------+---------------------------+
В тестах на тех же исходных файлах точность llama3.1:8b оказалась примерно на 10–15 пунктов ниже, чем у Haiku 4.5, в основном из-за того, что этот модель более строго обрабатывает описания зависимостей. Именно поэтому существует механизм оценки независимо от того, какой модель выполняет работу. Вы ведь не хотите узнавать о разнице в качестве лишь через три недели от неопытного агента.
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)"
Используйте его в обоих случаях: в качестве хука после коммита на машине разработчика и как обязательной задачи CI на общей ветке. Локально он выявляет отклонения ещё до создания PR. В CI он служит инструментом обеспечения соблюдения правил, поскольку некоторые участники так и не устанавливают хук, а хук, который невозможно применить на практике, является лишь рекомендацией.
Проверка на практике в производстве
Вот что стоит знать перед тем, как это реализовывать.
- Затраты реальны, но незначительны. Для репозитория с примерно 40 коммитами в день, касающимися по два-три пакета каждый, стоимость проверки через Haiku draft составляет несколько десятков долларов в месяц, а стоимость проверки eval примерно вдвое выше, поскольку каждый проект оценивается по сравнению со своим исходным кодом. Это недорогая цена за то, чтобы не передавать агенту неверную документацию.
needs_review в машине состояний.Что делать дальше
В настоящее время механизм оценки учитывает степень точности только при написании черновика и никогда не анализирует уже опубликованный файл. Следующим логичным шагом будет плановая переоценка всего набора на основе современного исходного кода, чтобы выявить концептуальные файлы, которые были точны в момент создания, но стали ошибочными после трёх рефакторингов без запуска процесса переписывания. Этого ещё не реализовано.
Пороги оценки — это показатели, которые каждый настраивает по-своему, и почти никто их не публикует; поэтому, если вы создадите что-то подобное, отслеживайте значения для своей системы и проверяйте, сохраняются ли они через несколько недель.