怎样手写一个可验证的 Agent 图工作流

下午五点，CI 报了一条失败：test_order_total_rounding 预期得到 19.99，实际得到 20.00。你把任务交给编码 Agent，它很快读了测试、改了一个金额计算函数、又跑了一遍测试。结果第一条测试过了，另外两条折扣测试开始失败。

这时最常见的处理方式，是把报错继续塞回同一个对话里，让模型再试一次。小任务通常没问题。可任务稍微复杂一点，那个对话会开始承担太多职责：它既要判断故障范围，又要找资料、提出方案、改文件、跑测试、决定是否继续，还得记住哪些命令已经执行过。失败时你只能看到一段越来越长的聊天记录，很难回答几个很实际的问题：

  • 它是在什么证据下改了这个文件？
  • 哪一步允许写入，谁能阻止写入？
  • 测试失败后为什么又改了一次，而不是停下来？
  • 如果进程中途退出，能从哪一步继续？

图工作流把这些混在一起的责任拆开。任务被分成有边界的节点，状态、分支、重试和停止条件也变成程序能执行的规则。

本文会从零写一个小型运行时。它不调用真实模型，也不改真实文件，这样刚接触 Agent 的读者可以先把控制流看清。等最后一节，你会知道怎样把其中的 plan_fix()、apply_fix() 换成 LLM 和受限工具，并能看懂 LangGraph 里的 State、Node、Edge 为什么正好对应这些代码。

一、先把几个词说清

这篇文章里的四个词会反复出现，先别急着记术语，先看它们在修复任务中分别是什么。

词 在本文中指什么 例子
Loop 同一角色根据新观察不断重复处理 “读报错，修改，再跑测试”
Node 只承担一种职责的步骤 verify 只运行测试，不改代码
Edge 从一个步骤走向下一个步骤的规则 测试通过走 finish，失败走 repair
State 节点之间传递、保存并恢复的信息 报错、计划、改动摘要、测试结果、重试次数

一个 Loop 本身也可以画成图。它有一个工作节点和一条回到自己的边：

1
2
3
4
5
6
7
                     测试失败
                         │
                         ▼
用户任务 → [诊断、修改、验证] ────→ 完成
                         ▲
                         │
                     再试一次

从图的角度看，Loop 只是最简单的形状。把它展开后，”诊断”、”修改”、”验证” 各自有输入和输出，程序也能在它们之间插入检查、并行或人工审批。

多 Agent 是另一件事。一个节点可以调用 LLM，也可以只是普通函数、测试命令、数据库查询，或者一个等待人工确认的暂停点。把每个节点都换成 Agent 往往没有好处。固定的路径校验、预算比较和测试退出码，本来就应该由代码决定。

LangGraph 的官方抽象也是如此：State 是共享状态，Node 是读取状态并返回更新的函数，Edge 决定下一个节点；节点不要求是模型。Graph API overview

二、不要因为任务长就上 Graph

网上常见一种说法：一个 Loop 多跑几轮就该升级成 Graph。这个判断不可靠。

让 Agent 查三个文件、汇总一段说明，可能走了六轮工具调用，却仍然只需要一个 Loop。它使用同一份上下文、同一种权限、同一个验收条件。相反，”给生产环境改支付配置”也许只走两步，却应该被拆开，因为读取配置、生成修改建议、提交变更和人工批准的风险完全不同。

我会用下面四个信号判断要不要拆图。

2.1 节点的权限不同

只读诊断与写文件不应共用一套工具。前者可以反复探索，后者应该少而慎重。如果一个节点只需要 read_file 和 search_logs，就不要顺手把 write_file、部署命令和生产凭据一起给它。

2.2 节点的输入不同

审阅一个补丁时，审阅者通常只需要任务要求、diff、测试结果和项目规范，不需要继承执行 Agent 的全部思考过程。把长对话原样交给它，审阅者更容易顺着上游的判断走，而不是检查补丁本身。

2.3 下游要等待多个独立结果

排查故障时，可以并行读取错误日志、依赖版本和最近的代码变更。修复节点却必须等这些结果回来，或者清楚知道哪一项缺失。这里的价值在于写出汇合条件，而不是为了同时调用几个模型。

2.4 失败后需要可恢复

如果任务在”测试失败”处停止，你希望下一次只重跑修复和验证，而不是让模型重新读完所有文件。此时需要检查点和可序列化状态。LangGraph 将 checkpoint 用于恢复、回放、分支和故障处理；用不用框架是后话，先把状态设计出来才有意义。Persistence

下面的任务满足这些条件：诊断是只读的，执行会产生副作用，验证由确定性命令完成，失败后最多修两次，仍然不通过就交给人。它适合做成一张很小的图。

这张图没有把”修复”拆成三个听起来很专业的角色。节点是否该独立，取决于它有没有不同的输入、权限、输出或停止条件。两个节点合并后什么都不损失，就把它们放在一起。

三、从 State 开始，不要先写 Agent

许多工作流项目一开始先写 prompt，之后才补状态。最后会得到一个大字典：研究结果、模型草稿、命令输出、错误字符串、用户消息全堆在一起。节点看见什么、能改什么，全靠开发者记忆。

先写状态会麻烦一点，但能让问题暴露得更早。下面用 dataclass 描述本文的任务状态。代码可以放进一个空文件 graph_agent.py，Python 3.11 可以直接运行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
from __future__ import annotations

from dataclasses import asdict, dataclass, field
from enum import StrEnum
from typing import Literal


class Stage(StrEnum):
    INTAKE = "intake"
    DIAGNOSE = "diagnose"
    PLAN = "plan"
    EXECUTE = "execute"
    VERIFY = "verify"
    HUMAN = "human"
    DONE = "done"
    FAILED = "failed"


@dataclass
class Observation:
    """只读工具得到的事实，不能把模型猜测写进这里。"""

    source: str
    content: str


@dataclass
class Verification:
    command: str
    exit_code: int
    summary: str


@dataclass
class TaskState:
    task_id: str
    request: str
    workspace: str
    stage: Stage = Stage.INTAKE
    observations: list[Observation] = field(default_factory=list)
    plan: str | None = None
    changed_files: list[str] = field(default_factory=list)
    verification: Verification | None = None
    retries: int = 0
    max_retries: int = 2
    spent_turns: int = 0
    max_turns: int = 12
    risk: Literal["low", "high"] = "low"
    route_reason: str | None = None
    events: list[str] = field(default_factory=list)

这份结构里有几个看上去不起眼的选择。

observations 只放工具看到的事实。比如测试输出、文件摘要、git diff 的结果。模型提出的猜测不应伪装成事实，所以计划单独放在 plan。verification 也不是一句”测试通过”，它需要留下执行了什么命令、退出码是多少、关键输出是什么。

retries、spent_turns 和 risk 由运行时治理。不要让模型自己写”我已经重试两次”或”这个改动风险很低”，然后把它当成系统状态。模型可以给出建议，最后写入这些字段的应该是受控代码或人工审批。

最后的 events 很朴素，它只是给初学者看的运行轨迹。生产系统会把它替换为结构化事件、trace ID、时间戳和持久化日志。先把”发生过什么”留下来，才能调试为什么走错路。

四、给每个节点写一份小合同

图工作流中的节点不是”一段 prompt”。它至少要说清楚四件事：

  1. 读取哪些状态字段。
  2. 可以调用哪些工具，是否能产生副作用。
  3. 修改哪些状态字段。
  4. 成功、失败或无法继续时交给哪条边。

为了避免一开始就接模型，下面先用固定函数模拟诊断、计划和执行。它们不会真的访问磁盘，输出却和真实系统应保存的信息相同。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def log(state: TaskState, message: str) -> None:
    state.events.append(f"{state.stage}: {message}")


def intake(state: TaskState) -> None:
    """入口节点只检查任务范围，不做分析，也不写文件。"""
    if not state.request.strip():
        state.stage = Stage.FAILED
        state.route_reason = "用户没有提供任务描述"
        log(state, "拒绝空任务")
        return

    if not state.workspace.startswith("workspace/"):
        state.stage = Stage.HUMAN
        state.risk = "high"
        state.route_reason = "工作目录不在允许范围内"
        log(state, "工作目录越界，等待人工决定")
        return

    state.stage = Stage.DIAGNOSE
    log(state, "输入通过，进入只读诊断")


def diagnose(state: TaskState) -> None:
    """真实版本应只暴露 read_file、search 和受限的测试日志读取工具。"""
    state.observations.extend([
        Observation("tests/test_orders.py", "test_order_total_rounding: 19.99 != 20.00"),
        Observation("src/pricing.py", "total = round(subtotal * (1 - discount), 2)"),
        Observation("git diff --stat", "工作区没有未提交改动"),
    ])
    state.spent_turns += 2
    state.stage = Stage.PLAN
    log(state, "收集到测试、实现和工作区状态")


def plan_fix(state: TaskState) -> None:
    """这里替换成 LLM 后，它也只能输出计划，不能直接改文件。"""
    sources = {item.source for item in state.observations}
    if "tests/test_orders.py" not in sources or "src/pricing.py" not in sources:
        state.stage = Stage.HUMAN
        state.route_reason = "诊断证据不完整，不能生成修复方案"
        log(state, "缺少关键证据")
        return

    state.plan = (
        "检查金额计算的舍入时机；保留现有折扣语义；"
        "只修改 src/pricing.py；现有测试仅用于验证，不在本轮任务中改写。"
    )
    state.spent_turns += 1
    state.stage = Stage.EXECUTE
    log(state, "生成最小修复计划")


def execute(state: TaskState) -> None:
    """唯一允许写入的节点。真实版本应在这里调用受限 edit_file。"""
    if state.plan is None:
        state.stage = Stage.HUMAN
        state.route_reason = "没有计划，禁止写入"
        log(state, "拒绝无计划执行")
        return

    if state.risk == "high":
        state.stage = Stage.HUMAN
        state.route_reason = "高风险任务需要批准"
        log(state, "高风险任务没有写入")
        return

    state.changed_files.append("src/pricing.py")
    state.spent_turns += 2
    state.stage = Stage.VERIFY
    log(state, "已应用修复并记录变更文件")

diagnose() 不写 plan，plan_fix() 也不改 changed_files。节点少时，这种划分像是多写了几行代码；节点多了以后，它能帮你追出一条错误结论到底在哪一步进入状态。

同一个理由也解释了为什么审阅 Agent 常常应当只读。它可以提出”这个补丁会破坏税费计算”，把问题写成一条 review observation；真正修改文件仍由执行节点负责。这样你不会得到三个 Agent 同时写同一个文件，然后试图从聊天记录里找出冲突来源。

五、验证节点不负责安慰人，它只负责给出证据

Agent 自己说”已经修好”没有意义。验证节点要把结论建立在外部可重跑的证据上。这里模拟两种结果：第一次验证失败，第二次通过。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
def verify(state: TaskState) -> None:
    """真实版本可运行 pytest、lint、类型检查或业务 API 探针。"""
    if state.retries == 0:
        state.verification = Verification(
            command="pytest tests/test_orders.py -q",
            exit_code=1,
            summary="test_discount_rounding 仍失败：18.675 被处理为 18.67",
        )
        log(state, "测试失败，留下可定位的错误")
        return

    state.verification = Verification(
        command="pytest tests/test_orders.py -q",
        exit_code=0,
        summary="3 passed",
    )
    log(state, "测试通过")


def repair(state: TaskState) -> None:
    """修复节点只根据验证证据处理，不重新编造整个任务。"""
    assert state.verification is not None

    state.retries += 1
    state.plan = (
        "针对 discount_rounding 补充 Decimal 舍入规则，"
        "保留上一轮已确认的订单总额修复。"
    )
    state.changed_files.append("src/pricing.py")
    state.spent_turns += 2
    state.stage = Stage.VERIFY
    log(state, f"根据测试失败进行第 {state.retries} 次修复")

注意 repair() 没有把失败结果覆盖掉。verification 在下一次验证时会被更新，事件列表仍然留着上一次失败。实际项目最好把验证结果做成 append-only 列表，再单独保存 latest_verification_id。本文为了控制代码长度，只保存最新结果和文本事件。

验证范围也要提前写清。只跑一个失败测试，不代表没有回归；跑完整测试集也不代表业务正确。不同任务需要不同组合：单测、静态检查、构建、迁移演练、截图对比或人工验收。系统必须分开记录”模型说完成”和”外部证据支持完成”。

六、验证也会失真：给优化循环留一个外部观察者

到这里，工作流已经有了一个看似完整的闭环：诊断、计划、执行、验证、失败后修复。可它还存在一个很现实的漏洞。它把 pytest 的退出码当作唯一的成功信号，而执行节点正试图让这个信号变成 0。

如果执行节点能改测试，它可以把断言从 19.99 改成 20.00，然后非常诚实地报告测试通过。即便禁止它改测试，也还有别的投机方式：缩小测试范围、跳过慢用例、只验证最初报错的函数，或者把错误吞掉。运行时没有坏，它完全按我们写的指标在优化。坏的是这个指标已经不再代表“订单金额计算正确”。

这就是反馈系统里常见的 Goodhart 问题：当某个数字成了优化目标，它很容易失去原本的测量意义。放到 Agent 工作流里，”测试通过”、”工单关闭”、”回答速度”、”评测分数”都可能发生同样的事。一个 Loop 只盯着它自己的成功条件时，往往看不见这种偏离。

解决办法不是再加一个会说”看起来不错”的审阅 Agent，而是为优化环增加独立的观察和约束。

6.1 每个主指标都该有一条护栏

主指标回答”这个任务完成了吗”，护栏回答”它是不是用不该用的方式完成了”。两者必须来自不同角度。

场景 主指标 护栏或反指标
修复 bug 指定失败测试通过 保留回归测试通过、diff 未改测试断言、没有新增跳过标记
客服 Agent 工单被解决 用户是否重新打开工单、升级率、抽样人工判定
代码生成 构建通过 静态检查、依赖漏洞扫描、变更规模和评审结果
检索问答 回答耗时低 引用是否可访问、关键事实是否来自允许的数据源、人工抽检

护栏不是越多越好。每多一个指标都会增加成本，也可能与主目标冲突。开始时挑最容易被投机的那一条即可。订单示例里最直接的护栏是：执行节点没有测试目录的写权限；验证节点运行一组执行节点从未见过的保留测试；审计节点检查 diff 中有没有删除断言、增加 skip 或修改测试收敛条件。

这里的”保留测试”和机器学习里的 held-out 集合很像。优化节点可以看见任务描述、失败输出和允许的源代码，但不该看见所有验收样例。否则它学到的可能是怎样让这组样例变绿，而不是怎样修正底层行为。保留测试不必神秘，可以是代码审查者维护的一组回归用例，也可以是线上灰度中的独立样本。

6.2 冻结规则不要放进可写 State

max_retries、允许路径、必须运行的测试命令、哪些目录禁止改写，这些规则不属于任务产物。它们是运行时配置，应该由调用方或人工维护，而不是让计划节点更新。

1
2
3
4
5
6
7
8
9
10
11
12
from dataclasses import dataclass


@dataclass(frozen=True)
class EvaluationPolicy:
    allowed_write_paths: tuple[str, ...] = ("src/pricing.py",)
    required_commands: tuple[str, ...] = (
        "pytest tests/test_orders.py -q",
        "pytest tests/regression_orders.py -q",
    )
    forbidden_diff_markers: tuple[str, ...] = ("pytest.mark.skip", "assert True")
    max_changed_files: int = 1

frozen=True 不会替代操作系统权限，也拦不住一段恶意 Python 直接绕过对象限制。但它把一个设计选择写得很明确：执行 Agent 读取 policy，不能更新 policy。真实系统还要在工具层检查路径，在 CI 或隔离环境中执行命令，并把 policy 版本写入 trace。

一条很实用的规则是：优化节点能修改的东西，不应同时定义它如何被判定成功。修复节点可以改业务代码；验证策略、保留测试和批准规则由它外面的层拥有。否则图里的每一层都能彼此确认，却没有哪一层还在检查现实。

6.3 审计环跑得比执行环慢

每一次工具调用都跑全量审计会很贵，也会把简单任务拖得很慢。可以让不同反馈环以不同节奏运行：

  • 快速执行环在每次任务中运行，负责受限修改和针对性测试。
  • 发布前审计环运行保留测试、diff 检查和安全扫描。
  • 更慢的质量环按周或按版本抽取真实任务，检查”通过率”是否仍然代表用户感受到的成功。
  • 人负责重新确认目标本身。例如，客服团队不能只因为关闭率上升就宣布系统变好，还要检查用户是否解决问题。

快环解决眼前任务，慢环检查快环是不是在朝错误方向加速。两者不该共享同一份可随时修改的评分标准。

七、边才是控制流。把每一种出口写出来

前面的节点只修改状态，并没有自己决定工作流何时结束。把判断集中到路由函数，才能清楚看见规则来自哪里。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def route_after_verify(state: TaskState) -> Stage:
    result = state.verification
    if result is None:
        state.route_reason = "验证节点没有返回结果"
        return Stage.HUMAN

    if result.exit_code == 0:
        state.route_reason = "验证命令退出码为 0"
        return Stage.DONE

    if state.spent_turns >= state.max_turns:
        state.route_reason = "已达到模型与工具调用预算"
        return Stage.HUMAN

    if state.retries >= state.max_retries:
        state.route_reason = "已达到最大修复次数"
        return Stage.HUMAN

    if "permission" in result.summary.lower():
        state.risk = "high"
        state.route_reason = "验证报告涉及权限问题"
        return Stage.HUMAN

    state.route_reason = "测试失败，但仍在允许的修复预算内"
    return Stage.EXECUTE

多数路由都不需要模型决定。退出码、重试次数、预算和权限命中都是确定条件，用 if 更稳定，也更容易测试。只有”这个错误是依赖服务暂时不可用，还是代码逻辑错误”这类需要理解文本语义的分支，才适合交给受限的模型分类器。

即使用模型分类，也不要让它任意输出下一步。让它只能在 repair、clarify、human_review 里选一个，并要求返回它引用的 observation ID。路由层再检查这个枚举和证据是否有效。模型负责解释，运行时负责授权。

路由函数返回 Stage.EXECUTE 看起来有点奇怪。因为验证失败后，流程不是回到完整的诊断和计划，而是进入 repair()。下面的运行器会根据当前 stage 和验证结果将它分发到正确函数。你也可以把 REPAIR 加进枚举，代码会更直白。本文保留较少状态值，方便先理解图的本质。

八、把节点和边接起来：一个很小的图运行时

接下来才轮到运行器。它的职责很窄：读取当前状态，执行一个节点，根据节点结果选择下一个节点，并在终态、预算或人工暂停处停止。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
def run_graph(state: TaskState) -> TaskState:
    while state.stage not in {Stage.DONE, Stage.HUMAN, Stage.FAILED}:
        if state.spent_turns >= state.max_turns:
            state.stage = Stage.HUMAN
            state.route_reason = "运行器在节点开始前发现预算耗尽"
            log(state, "预算耗尽")
            break

        if state.stage == Stage.INTAKE:
            intake(state)
            continue

        if state.stage == Stage.DIAGNOSE:
            diagnose(state)
            continue

        if state.stage == Stage.PLAN:
            plan_fix(state)
            continue

        if state.stage == Stage.EXECUTE:
            if state.verification and state.verification.exit_code != 0:
                repair(state)
            else:
                execute(state)
            continue

        if state.stage == Stage.VERIFY:
            verify(state)
            next_stage = route_after_verify(state)
            state.stage = next_stage
            log(state, f"路由到 {next_stage}: {state.route_reason}")
            continue

        raise RuntimeError(f"未知状态: {state.stage}")

    return state

入口只需要构造状态并运行：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
if __name__ == "__main__":
    state = TaskState(
        task_id="fix-order-rounding-001",
        request="修复订单金额舍入测试，并确认相关测试通过",
        workspace="workspace/demo-shop",
    )
    result = run_graph(state)

    print("最终状态:", result.stage)
    print("路由原因:", result.route_reason)
    print("改动文件:", result.changed_files)
    print("验证结果:", result.verification)
    print("运行轨迹:")
    print("\n".join(result.events))

这段程序运行后会经历下面这条路径：

1
2
3
4
5
6
7
8
INTAKE
  → DIAGNOSE
  → PLAN
  → EXECUTE
  → VERIFY          第一次测试失败
  → EXECUTE          实际调用 repair()
  → VERIFY          第二次测试通过
  → DONE

它看起来仍然是一个 while 循环。没错，很多图运行时的底层也会有循环或调度器。Graph 的区别不在于有没有 while，而在于循环不再把所有行为混成一个大函数。节点、状态和边都已经成为独立对象，可以单独检查、替换和测试。

九、一步一步看状态怎样变化

初学者读 Agent 代码容易只盯着 prompt 或工具调用，忽略状态已经怎么变了。下面把刚才这次运行压缩成一张表。

步骤 节点 新增或修改的状态 为什么能进入下一步
1 intake stage=DIAGNOSE 工作目录在允许范围内，任务非空
2 diagnose 三条 observations，spent_turns=2 只读诊断完成
3 plan plan 有最小修复方案 关键文件证据齐全
4 execute changed_files=[pricing.py] 有计划，且不是高风险任务
5 verify 第一份失败 verification 测试退出码为 1
6 route route_reason 写入重试原因 未超预算，未达到重试上限
7 repair retries=1，更新计划和改动记录 验证失败属于可修复问题
8 verify 第二份成功 verification 测试退出码为 0
9 route stage=DONE 外部验证通过

以后某个任务说”Agent 改错文件”，你不必靠模型回忆。可以查第 4 步执行节点拿到了什么 plan，查第 2 步 observations 是否包含了错误文件，查路由有没有绕过高风险分支。

也要注意一个边界：这个例子里状态对象是可变的，节点会直接修改它。教学代码这样写最容易读。生产中更常见的做法是让节点返回一个 patch，例如 {"plan": "..."}，由运行时统一合并到状态。这能更方便地做审计和并发 reducer，也能防止节点悄悄改了不属于它的字段。LangGraph 的节点就是返回状态更新，再由状态 schema 和 reducer 决定怎样合并。Graph API overview

十、接入 LLM 后，节点应该怎样写

把上面的 plan_fix() 换成模型调用，并不等于把整个 state 原封不动塞进 prompt。模型看到越多信息，越容易抓住无关细节，也越难判断哪一条是事实、哪一条是旧计划。

计划节点只需要一个很小的上下文投影：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def plan_context(state: TaskState) -> dict:
    return {
        "request": state.request,
        "observations": [asdict(item) for item in state.observations],
        "previous_verification": (
            asdict(state.verification) if state.verification else None
        ),
        "allowed_files": ["src/pricing.py"],
        "required_output": {
            "hypothesis": "string",
            "steps": ["string"],
            "files_to_change": ["string"],
            "risks": ["string"],
        },
    }

真实 plan_fix() 可以把这份对象序列化后发送给模型，并要求模型返回 JSON。收到回复后不能直接相信它：要检查 files_to_change 是否都在允许列表中，steps 是否为空，JSON 是否符合 schema。任何一个检查失败，都把错误作为新的 observation 或转人工处理，而不是让模型直接调用编辑工具。

执行节点也不应该得到计划节点的全部权限。它需要的能力可能只有：读取计划、编辑白名单文件、运行一条指定测试、读取 diff。部署、删除目录、读取环境变量等工具默认不在这个节点的能力集中。

一个最小的权限表可以写成这样：

节点 可以读取 可以写入 可以执行
diagnose 代码、测试、日志 无 只读查询、受限测试日志
plan 已投影的 observations 无 无
execute 计划和目标文件 业务代码白名单文件 受限编辑
verify 变更摘要和测试文件 无 测试、lint、构建
human 全部必要证据 按审批决定 按审批决定

这张表不会因为提示词写了规则就自动生效。提示词只能说明约束，工具注册表、工作目录限制、容器或审批系统才负责执行。模型就算幻觉出 delete_database，执行层没有这个工具，它也做不了。

十一、并行不是同时跑几个 Agent 那么简单

现在回到诊断阶段。我们之前顺序收集测试、实现和 git diff。如果三个读取任务互不依赖，可以并行，但必须先回答两个问题：结果怎样写回状态，某个分支失败时是否还继续。

下面用 asyncio 模拟三个只读任务。为了演示顺序问题，第三个任务最快完成，第二个任务最慢完成。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import asyncio


async def read_test() -> Observation:
    await asyncio.sleep(0.02)
    return Observation("tests/test_orders.py", "rounding assertion failed")


async def read_implementation() -> Observation:
    await asyncio.sleep(0.06)
    return Observation("src/pricing.py", "uses binary float before rounding")


async def inspect_diff() -> Observation:
    await asyncio.sleep(0.01)
    return Observation("git diff --stat", "working tree clean")


async def diagnose_in_parallel(state: TaskState) -> None:
    jobs = [read_test(), read_implementation(), inspect_diff()]
    results = await asyncio.gather(*jobs, return_exceptions=True)

    for name, result in zip(["test", "implementation", "diff"], results):
        if isinstance(result, Exception):
            state.events.append(f"diagnose: {name} 读取失败: {result}")
        else:
            state.observations.append(result)

asyncio.gather() 的返回顺序与传入顺序一致，不是完成顺序。测试读取最先结束，结果列表仍按 test、implementation、diff 的固定顺序写回。调度略有不同，下一次运行的状态排列也不会跟着漂移。

如果你用的是多个子 Agent，道理相同。它们可以并发完成，主工作流保存结果时应使用稳定的任务 ID 或原始调用顺序，而不是谁先返回就先写谁。否则复现一次故障时，你连”后面的 Agent 到底看见了哪份先前状态”都说不清。

还要设计失败策略。对于我们的修复任务，读不到 git diff 也许只需要标记警告；读不到失败测试文件则不应继续计划。可以把这种规则明确写成：

1
2
3
4
测试文件缺失                 → HUMAN
实现文件缺失                 → HUMAN
git diff 读取失败             → 继续，但 risk=high
任何并行任务超时             → 取消剩余任务并写入 timeout 事件

并行读通常比较容易。并行写复杂得多。两个执行节点同时修改同一个文件，最后保存的内容取决于调度而不是业务规则。除非你已经有 worktree、文件锁、patch 合并和冲突处理，否则让写入节点串行运行，通常比事后处理乱掉的工作区便宜。

十二、检查点不是“存一下聊天记录”

运行到一半进程崩了，下一次重跑时最糟糕的选择是把任务从头再做一遍。你会重复消耗模型调用，也可能在不知道上一轮已经写过文件的情况下再次写入。

检查点应该保存的是可以恢复的任务事实，而不是一整段未经筛选的聊天文本。对于本文的状态，最小实现可以每执行完一个节点就写一次 JSON：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import json
from pathlib import Path


def save_checkpoint(state: TaskState, path: Path) -> None:
    payload = asdict(state)
    payload["stage"] = state.stage.value
    path.write_text(
        json.dumps(payload, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )


def load_checkpoint(path: Path) -> TaskState:
    raw = json.loads(path.read_text(encoding="utf-8"))
    raw["stage"] = Stage(raw["stage"])
    raw["observations"] = [Observation(**item) for item in raw["observations"]]
    if raw["verification"] is not None:
        raw["verification"] = Verification(**raw["verification"])
    return TaskState(**raw)

在 run_graph() 每个节点之后加上 save_checkpoint(state, checkpoint_path)，就能在进入 VERIFY 前保存当前进度。恢复时也不能盲目继续：应该重新确认工作目录、检查上次记录的变更是否仍然存在、必要时重新运行验证。外部世界会变化，保存下来的状态只能说明”当时看见了什么”，不能保证今天仍然成立。

另一个常被忽略的问题是版本。半年后你给 TaskState 新增字段，或者把 Stage.EXECUTE 改名成 Stage.APPLY_PATCH，旧检查点可能无法读取。至少给持久化内容加一个 schema_version 和 workflow_version，再为旧版本提供迁移函数。没有这些，”支持恢复”只是开发环境里的一次演示。

十三、怎样测试一张图

测试图工作流时，不能只问最后有没有输出正确文本。路径本身也是产品行为。

先为路由写最小单元测试。它们不需要模型，也不需要真实文件：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
def make_failed_state(*, retries: int, spent_turns: int) -> TaskState:
    return TaskState(
        task_id="test",
        request="fix",
        workspace="workspace/demo",
        retries=retries,
        spent_turns=spent_turns,
        verification=Verification("pytest", 1, "assertion failed"),
    )


def test_failed_verification_enters_repair() -> None:
    state = make_failed_state(retries=0, spent_turns=3)
    assert route_after_verify(state) == Stage.EXECUTE


def test_retry_limit_enters_human_review() -> None:
    state = make_failed_state(retries=2, spent_turns=3)
    assert route_after_verify(state) == Stage.HUMAN


def test_budget_limit_enters_human_review() -> None:
    state = make_failed_state(retries=0, spent_turns=12)
    assert route_after_verify(state) == Stage.HUMAN

再测节点权限。这里不必真的运行沙箱，可以给工具函数一个会记录调用的 fake executor，断言诊断节点从不触发写入工具，执行节点拿不到白名单之外的路径。接入真实工具后，这些测试仍要留在执行层，不能只检查 prompt 里有没有写”请谨慎操作”。

最后测端到端轨迹。本文的模拟场景应该固定走到 DONE，事件中必须有一次失败验证、一次修复、一次成功验证。另一个场景把 max_retries 设为 0，预期停在 HUMAN，并且不再进行第二次写入。这样的测试会抓住很多肉眼看不出的错误，例如路由函数把 >= 写成 >，导致系统多改了一次文件。

评测模型节点时还要再加一层。为诊断、计划和审阅准备固定 fixture，分别检查：

  • 是否引用了提供的 observation，而不是编造文件内容；
  • 是否遵守了输出 JSON schema；
  • 是否拒绝越权文件或危险操作；
  • 在同一种失败输入下，是否选择了允许的路由枚举；
  • 当证据不足时，是否转向澄清或人工处理，而不是编造修复。

Agent 的正确性通常不是一个分数。模型输出可以评估，工具结果可以断言，路径是否越权也可以断言。把它们混成一句”通过率 85%”，很难知道下一次该修 prompt、修工具还是修路由。

十四、把这个小运行时映射到 LangGraph

看完手写版本后，再看框架会轻松很多。下面不是另一套新概念，只是名称对应：

本文代码 LangGraph 中常见的对应物
TaskState State schema，常用 TypedDict、dataclass 或 Pydantic 模型
intake、diagnose、verify add_node() 注册的节点函数
route_after_verify() add_conditional_edges() 的条件路由
Stage.DONE END
save_checkpoint() checkpointer 持久化的 state snapshot
run_graph() 编译后的图运行时与调度器

框架会帮你处理图的编译、状态合并、并发 super-step、流式事件和 checkpointer 接口，但它不会替你回答业务问题：verification 是覆盖还是追加？测试失败后能改几次？哪个节点有写权限？何时必须人工审批？这些决定仍然需要你在状态、节点和边里写出来。

如果你还没用过 LangGraph，不要急着把本文所有函数翻译成 API。先拿第三节的状态、第四节的节点合同、第六节的路由规则画在纸上。能在不运行程序时说明每个字段的写入者和每条边的条件，再上框架会快得多。

十五、几个很常见的错误

为了“多 Agent”而拆节点

把总结 PDF 拆成抓取、切块、摘要、二次摘要、润色、审阅六个 LLM 节点，看起来像一张复杂图，实际只是增加了调用次数和状态传递。若同一个模型、同一份上下文、同一套工具就能完成，保持一个 Loop 更好。

让所有节点读写全部 State

这是最容易把图做成共享泥潭的方式。诊断节点不需要改预算，审阅节点不需要写变更文件，执行节点也不需要覆盖原始请求。把字段所有权写进接口和测试，后期会少很多莫名其妙的回归。

把路由都交给模型

模型可以解释复杂错误，但不该决定已知的硬边界。预算耗尽、测试退出码、白名单命中、人工审批都是程序规则。让模型决定这些，结果既不稳定也不容易复现。

用另一个模型给第一个模型盖章

审阅必须看到独立证据。没有测试、没有规则清单、没有来源，第二个模型只是用不同措辞重复第一个模型的判断。它可以发现一些表达问题，却不能把猜测变成验证。

只存最终答案，不存失败路径

图系统最有价值的信息常常在失败处：哪条工具调用超时，哪次验证失败，状态在哪个版本被污染。只保存最终文本，等于在故障发生后扔掉证据。

一开始就并行写文件

并行读取、并行调研通常安全。并行写同一个工作区要面对锁、worktree、文件冲突和合并。先把唯一写入者做对，再考虑把独立写操作放到隔离环境中。

十六、给初学者的练习顺序

想动手跑一遍，可以按下面的顺序增加难度。

  1. 把第四节到第七节的代码复制到一个文件中，确认模拟任务会经历一次失败和一次修复后结束。
  2. 修改 verify()，让它连续失败三次，观察任务为什么会停在 HUMAN，而不是无限循环。
  3. 给 intake() 增加一个规则：请求里出现 deploy 或 delete 时直接进入 HUMAN。这一步能体会到权限路由为什么不该交给模型。
  4. 把 diagnose() 的固定 observation 换成受限工作目录里的真实文件读取，但仍然不要加入写文件工具。
  5. 只替换 plan_fix()，让模型从投影后的 observations 中返回 JSON 计划。先打印并校验 JSON，不让它执行。
  6. 最后才把 execute() 接到一个允许修改临时目录的编辑工具，并让 verify() 跑测试。

每一步都保留前一步的测试。很多人第一次接模型后发现路径乱了，就开始不断改 prompt。更有效的做法是先问：状态有没有缺字段，路由有没有让模型承担确定判断，执行器有没有放开不该有的工具。模型只是图里的一个节点，别让它成为所有问题的解释。

图只是责任清单的一种写法

一张好的 Agent 图不会让人觉得”节点真多”。读者应该能顺着它回答：发生了什么，谁做了什么，凭什么继续，失败后停在哪里。

Loop 仍然是最常用的起点。把它做得有预算、有验证、有工具边界，已经能解决很多任务。职责分开、状态需要恢复、分支需要解释、并行需要汇合时，再把它展开成 Graph。图里写下的，是运行时、测试和后续维护者都能检查的责任。

参考资料