一种能为人工智能代理保持代码库内存准确性的LangGraph流程

任何将编码智能体应用于大型代码库的人都会遇到一个问题：智能体如何在不为每个任务重新读取所有文件的情况下理解整个代码库？以下是解决该问题的第三次尝试。前两次尝试均采用基于文档字符串的嵌入式RAG技术，但由于文档更新速度跟不上代码变更，这两种方法都在一个月内失效了。而这个第三版本则能够应对每天有三十到四十次代码提交的团队需求。

这并非对他人文件格式的评估，而是针对整个处理流程：它利用LangGraph状态图监控代码库，为每个服务生成结构化的知识文件，在任何智能体读取之前先对输出结果的准确性进行评分，同时记录每次运行过程以便了解成本情况。第一次尝试中的失败案例也会被一并纳入分析。

涵盖内容：为何多智能体编码环境中的令牌预算问题实质上是过时性问题，LangGraph的完整设计（状态架构、节点、条件边以及DynamoDB检查点机制），使用Claude Haiku 4.5在Amazon Bedrock上的引用功能，基于Ragas和DeepEval构建的评估系统，通过自托管Langfuse实现的可观测性，为无法将代码上传到云端模型的团队设计的完全本地化Ollama版本，将各组件串联起来的git钩子，以及针对故障模式的分析。

问题所在

每当编码智能体在某个代码库中开始执行任务时，它都会面临两个都不理想的选项。其一是可以读取所有内容，但这会浪费令牌在与任务无关的文件上；其二则是使用基于嵌入的检索方式，虽然能获取语义相似的代码片段，却无法明确指出谁调用了什么功能或哪个模块属于谁所有。

可行的方案是准备少量经过精心整理的结构性文件，每个服务对应一个文件，说明其功能及依赖项，这些文件只需生成一次即可自动更新。协调代理会查询索引，选出与任务相关的两三个文件，然后仅将它们转发给子代理。这既关乎如何明智地使用令牌，也关乎知识的表达方式。

难点并不在于文件格式——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     |
+------------------+---------------------------+--------------------------------------+

第一步：状态模式

在 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 起草

此环节会消耗实际成本，因此需要谨慎处理。运行在 Bedrock 上的 Claude Haiku 4.5 每次提交时都能快速且低成本地完成处理，而更大型的模型则仅用于需要人工审核的环节，绝不会用在常规的起草流程中。

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步：大多数人会跳过的评估环节

带有 frontmatter 类型字段的 Markdown 文件无法判断自己所说是否属实。这种格式根本无法阻止大型语言模型写出看似可信、格式良好但实际上错误的服务描述。如果跳过这一环节，系统就可能在悄无声息地扭曲智能体对其自身代码库的理解，而外观上却仍会像正确理解的一样可靠。

Ragas负责评分，而DeepEval则承担通过/失败的判定，因为二者解决的是不同的问题。Ragas会生成一个忠实度分数，用来衡量草稿与真实来源的契合程度，无需手工提供的参考答案。DeepEval则将这个数值转化为机器可以依据的标准。

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 环境中运行，而那里的 worker 经常会在运行过程中意外终止，这一现象比人们承认的更为普遍。通过检查点机制，流程会从上一个完成的超级步骤继续，而不会因为第六个步骤超时就重新处理五个文件。

步骤 6：查看成本

在能够查看每个节点的令牌使用情况与运行时长之前，人们并不信任该流水线。Langfuse 的 @observe() 装饰器仅用了五分钟就解决了这个问题，无需进行整体重新设计。

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 已无法满足需求时，通过 Ollama 运行的 nomic-embed-text 是大多数本地 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的忠实度比Haiku 4.5低约10到15分，主要是因为它更难以准确描述依赖关系。正是由于这种差距，才需要无论哪种模型负责生成内容时都进行评估。你可不想在三周后从表现混乱的代理那里得知质量上的差异。

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 草稿检测的费用大约在每月几十美元左右，而评估检测的费用则约为其两倍，因为每份草稿都需要与其源代码进行对比评分。这样的成本对于避免向处理工具提供错误文档来说已经相当便宜了。

延迟是没人提及的权衡问题。在某套配置下，对单个包的提交进行完整扫描、草稿生成、评估和发布的全流程需要15到25秒。作为异步CI任务尚可，但若在提交时作为同步操作则不可行，因为会在进入后台任务处理之前就出错。

该检查机制有时会拦截合格的草稿。忠实度评分并不完美，一份准确但简略的草稿也可能低于阈值。任何两次测试都失败的草稿都会进入人工审核队列，而不会被悄悄丢弃，这在状态机中表现为needs_review状态。

如果在自己的代码库中不使用多个代理，就不值得构建此系统。其所有价值都来自于消耗该包的代理。最初构建的仅支持嵌入的版本因为无人使用而闲置了两个月。

包与文件的映射关系无法完整传输。其他部分几乎都可以直接复制粘贴。只有那个函数需要根据自身布局重新编写，而且应该花时间认真实现，而非用一行代码应付了事。

下一步该做什么

目前，评估系统仅在草稿编写时判断一致性，而不会重新检查已发布的文件。接下来的自然步骤是定期用当前的源代码重新评估整个项目包，以便发现那些在最初创建时准确无误，但在经过三次重构后出错却未被触发重新起草的方案文件。目前这一功能尚未实现。

评估阈值是每个人调整方式都不同的参数，而且几乎没人会公开自己的设定，因此如果你开发类似功能，一定要记录下你的阈值数值，并观察几周后它是否依然适用。