跳转至

Graph Engineering:从单 Agent Loop 到可治理的执行图

这篇回答一个具体问题:

一个 Agent 已经能自己调用工具、检查结果并继续行动,为什么还要把系统改成 Graph?

先说结论:

Loop 没有“死”。Graph Engineering 是把多个 Loop、模型调用、普通函数、工具和人工审批,组织成一张可执行、可恢复、可观测的图。

Graph 的价值也不在“Agent 更多”,而在于把关键控制权从自然语言里拿出来,交给程序可以检查的节点、边、状态和策略。

本文根据微信公众号文章《Loop Engineering 已死?一文带你了解 Graph Engineering》重新组织,并结合 Anthropic、LangGraph 和 Google ADK 的官方资料校正工程边界。它不是原文逐段改写,而是一篇面向实际设计和落地的独立教程。

先用一个生活例子理解

假设你要做一份行业研究报告。

最简单的做法是找一个能力很强的人:

查资料
  ↓
判断还缺什么
  ↓
继续搜索
  ↓
写初稿
  ↓
自己检查
  ↓
修改并交付

这就是一个 Loop。一个执行者围绕目标不断“观察、行动、检查、继续”,直到完成或停止。

如果任务变大,你可能会改成团队协作:

负责人拆分研究方向
  ↓
几位研究员并行查资料
  ↓
编辑统一写作
  ↓
审校人员独立检查
  ↓
不合格则退回修改

这就是 Graph 的直觉:

  • 每个岗位是一个节点。
  • 任务如何流转是边。
  • 资料、草稿和审校结果是状态。
  • 谁能发布、最多修改几次、何时必须人工确认是策略。

团队里的每个人仍然可以有自己的工作 Loop。Graph 并没有消灭 Loop,只是在更外层组织它们。

从 Prompt 到 Graph 是逐层加工程能力

可以把 Agent 系统理解成五层:

Graph:多个执行单元如何分工、流转和恢复
  ↑
Loop:一个执行单元如何持续行动直到停止
  ↑
Harness:模型能看到什么、能用什么、受什么约束
  ↑
Context:这一轮给模型哪些信息
  ↑
Prompt:怎样向模型描述任务

它们不是互相替代的流行词。

层级 主要问题 典型产物
Prompt 这一次怎样向模型说清任务 system prompt、任务模板、few-shot
Context 模型此刻应该看到什么 检索结果、任务状态、相关文件、记忆
Harness 模型怎样接入真实环境 工具、权限、沙箱、预算、trace
Loop 怎样持续行动并正确停止 状态机、重试、验证、停止条件
Graph 多个执行单元怎样协作 节点、边、共享状态、路由、检查点

一个 Graph 节点内部通常仍然需要 Prompt、Context、Harness 和 Loop。

Graph 到底是什么

可以用一个简单表达式描述:

Graph = Nodes + Edges + State + Policies

Node:谁来完成一步

节点不一定是 Agent,也不一定调用大模型。

一个节点可以是:

  • 一次模型调用。
  • 一个会自主使用工具的 Agent Loop。
  • 一个普通 Python 函数。
  • 一段数据库查询。
  • 一个测试或规则校验器。
  • 一个人工审批步骤。

例如“验证 JSON 是否符合 schema”用普通代码更可靠,不必让另一个 Agent 来判断。

Edge:下一步去哪里

边负责连接节点,并表达状态如何流转。

边可以是固定的:

检索 -> 写作

也可以带条件:

审核通过 -> 发布
审核失败且可修改 -> 重写
达到重试上限 -> 转人工

生产系统里,关键边应该能在代码中被明确判断和测试。

State:大家围绕什么协作

状态是任务当前的事实记录,例如:

{
  "task_id": "brief-2026-07-29",
  "question": "本周 Agent 工程有哪些重要进展?",
  "sources": [],
  "draft": "",
  "review": null,
  "revision_count": 0,
  "status": "researching"
}

不要把“所有聊天记录”直接当状态。更好的状态应该:

  • 结构化。
  • 有版本。
  • 可持久化。
  • 能区分事实、产物和临时消息。
  • 能按节点裁剪成不同上下文。

Policy:谁能做什么

策略负责约束图怎样运行,例如:

  • 哪个节点能访问网络。
  • 哪个节点能写文件。
  • 哪些动作必须人工确认。
  • 单个节点最多重试几次。
  • 整张图最多花多少 token、时间和费用。
  • 哪类失败可以自动恢复。
  • 哪类结果允许发布。

如果没有策略,Graph 只是连在一起的 Agent,谈不上可治理。

Graph 不是什么

不是知识图谱

知识图谱描述“实体之间有什么关系”。

Graph Engineering 里的图主要描述“任务下一步怎么执行”。两者都叫 Graph,但解决的问题不同。

不只是画一张流程图

流程图只负责展示。可执行图还必须回答:

每个节点的输入输出是什么?
状态保存在哪里?
条件边怎样判断?
失败后从哪里恢复?
重试是否会产生重复副作用?
谁能批准高风险动作?

一张图能画出来,不代表它已经能可靠运行。

不等于 Multi-Agent

Graph 可以只有一个 Agent,甚至完全没有 Agent:

普通函数 -> 单个 Agent -> 规则校验 -> 人工审批

Multi-Agent 也可以没有明确 Graph,例如多个 Agent 自由聊天。前者强调控制结构,后者强调执行角色的数量。

Loop 和 Graph 的关系

Loop 是一个执行单元的内部控制,Graph 是多个执行单元之间的外部控制。

flowchart LR
    G["Graph Runtime"] --> R["Research Node"]
    G --> W["Writer Node"]
    G --> V["Verifier Node"]
    R --> RL["Research Agent Loop"]
    W --> WL["Writer Agent Loop"]
    V --> C["Deterministic Checks"]

二者不是二选一:

维度 单 Agent Loop Graph
控制单位 一段连续循环 多个节点和状态转移
决策者 多数由当前 Agent 决定 Agent 与程序共同决定
上下文 容易把历史持续累积 可以给每个节点独立上下文
并行 需要 Loop 临时调度 可以显式 fan-out / fan-in
验证 容易由同一个 Agent 自检 可以拆出独立验证节点
恢复 常从长对话中寻找现场 可从检查点恢复指定节点
观测 主要看 step 和 tool call 还能看节点、边和状态转移
成本 起步成本低 设计、运行和维护成本更高

因此更准确的判断是:

任务简单、路径开放 -> 先用一个 Loop
流程稳定、风险较高 -> 先用 Workflow
需要隔离、并行、独立验证或恢复 -> 考虑 Graph

为什么一个大 Loop 会逐渐变难

1. 上下文越来越脏

同一个 Agent 同时查资料、写作、审校时,上下文会混入:

  • 已经失效的计划。
  • 搜索过程的噪声。
  • 工具返回的长日志。
  • 自己写过的初稿。
  • 自己对初稿的解释。

信息越多不一定越聪明。真正相关的信息反而可能被淹没。

Graph 可以让 Writer 只看到经过整理的资料,让 Reviewer 只看到任务标准、引用和最终草稿。

2. 错误会一路传播

大 Loop 早期误解目标后,后面的搜索、分析和写作都可能建立在错误前提上。

Graph 可以在关键边插入校验:

研究结果是否有来源?
  ↓
输入格式是否合格?
  ↓
风险动作是否获批?

3. 一个 Agent 同时当运动员和裁判

生成者通常知道自己“想表达什么”,容易忽略实际输出的问题。

把 Generator 和 Verifier 拆开,可以让验证节点使用干净上下文和独立标准。对于代码、计算、schema、测试结果等任务,验证节点还可以直接使用真实环境,而不是依赖模型自我评价。

4. 工具越多,选择越难

把几十个工具同时给一个 Agent,会增加选择错误、参数错误和越权风险。

Graph 可以按节点缩小能力面:

Researcher:只读、搜索、抓取
Writer:只读资料、写草稿
Verifier:只读草稿、运行校验
Publisher:只有审批后才能发布

5. 难以知道失败发生在哪里

大 Loop 只告诉你“最终失败了”。Graph 可以进一步定位:

  • 哪个节点失败。
  • 走了哪条边。
  • 当时的状态是什么。
  • 重试了几次。
  • 是模型问题、工具问题还是路由问题。

五种最常用的图结构

这些结构并不是 Graph Engineering 发明的新算法。它们和 Anthropic 总结的 workflow patterns、传统状态机、数据流水线有明显继承关系。

1. 固定流水线:Prompt Chaining

flowchart LR
    A["提取事实"] --> B["生成提纲"]
    B --> C["撰写正文"]
    C --> D["格式校验"]

适合步骤固定、每一步都能给出清楚输入输出的任务。

关键不是把流程切得越细越好,而是在真正需要验证、复用或恢复的地方切开。

2. 条件路由:Routing

flowchart TD
    A["用户请求"] --> B{"任务分类"}
    B -->|"知识问答"| C["RAG 流程"]
    B -->|"代码修改"| D["Coding Agent"]
    B -->|"高风险操作"| E["人工确认"]
    B -->|"无法判断"| F["澄清或兜底"]

路由器可以使用规则、小模型或大模型,但路由结果必须可评测。因为第一步分错,后面节点再强也没有用。

3. 扇出与汇合:Parallelization

flowchart LR
    A["研究问题"] --> B["搜索官方资料"]
    A --> C["搜索论文"]
    A --> D["分析代码仓库"]
    B --> E["去重与冲突检查"]
    C --> E
    D --> E
    E --> F["统一写作"]

它适合可以独立探索的子任务。汇合节点必须处理:

  • 重复信息。
  • 互相冲突的结论。
  • 缺失结果。
  • 来源和产物的对应关系。

4. 调度者与工作者:Orchestrator-Workers

flowchart TD
    O["Orchestrator"] --> P["生成有边界的子任务"]
    P --> W1["Worker A"]
    P --> W2["Worker B"]
    P --> W3["Worker C"]
    W1 --> S["Synthesis"]
    W2 --> S
    W3 --> S

它和固定并行的区别在于:子任务数量和内容由 Orchestrator 根据当前任务动态决定。

每个子任务至少要带上:

owner
goal
allowed tools
input references
expected output schema
deadline
budget
done condition

否则 Worker 很容易重复劳动或无限扩张范围。

5. 生成与验证:Evaluator-Optimizer

flowchart LR
    G["Generator"] --> V{"Verifier"}
    V -->|"通过"| D["Done"]
    V -->|"可修复且未超限"| G
    V -->|"高风险或已超限"| H["Human / Stop"]

它适合有明确质量标准、但第一次生成很难稳定通过的任务。

必须同时有:

  • 明确的通过标准。
  • 只针对阻塞问题的反馈。
  • 最大修改次数。
  • 超限后的停止或转人工路径。

否则它只是两个 Agent 互相提意见,依然可能无限循环。

真正的核心:节点可以自主,边必须可控

一个实用原则是:

让模型处理语义判断和开放探索,让程序控制权限、预算、状态转移和不可逆动作。

更适合交给模型的事

  • 判断一个问题需要查哪些方向。
  • 从非结构化资料中提取要点。
  • 比较多个解释。
  • 起草内容。
  • 根据反馈修改。

更适合交给程序的事

  • JSON Schema 校验。
  • 文件是否存在。
  • 测试是否通过。
  • 数值是否越界。
  • 权限是否允许。
  • 重试次数是否超限。
  • 是否已获得人工批准。
  • 是否执行真实发布、付款或删除。

例如不要让模型仅凭感觉说“测试应该通过了”,而要让验证节点真正执行测试并读取退出码。

模型提出候选结果
  ↓
代码或真实环境给出证据
  ↓
策略引擎决定是否允许继续

Graph 的确定性主要来自这些硬锚点,而不是来自节点数量。

完整例子:每天生成一份技术简报

假设系统每天要回答:

过去 24 小时,Agent 工程领域有哪些值得关注的更新?

交付要求是:

  • 至少覆盖官方公告、官方文档和代码仓库。
  • 每条结论都能回到来源。
  • 不把发布日期和事件发生日期混为一谈。
  • 不确定信息必须明确标记。
  • 发布前经过独立审校。

如果全部塞进一个 Loop

单个 Agent 要在同一段上下文中完成搜索、去重、判断、写作、查错和发布。任务一长,就容易发生:

  • 搜索日志挤占写作上下文。
  • 相同来源被重复计算。
  • 初稿里的错误影响自我审校。
  • 中断后很难恢复到正确位置。
  • 修改一条引用却重做整份报告。

改成三段执行图

flowchart TD
    S["Start"] --> O["Research Orchestrator"]
    O --> A["官方公告 Worker"]
    O --> B["官方文档 Worker"]
    O --> C["代码仓库 Worker"]
    A --> M["Normalize & Merge"]
    B --> M
    C --> M
    M --> W["Writer"]
    W --> R{"Independent Reviewer"}
    R -->|"pass"| P["Publish"]
    R -->|"revise, retry < 2"| W
    R -->|"blocked / retry >= 2"| H["Human Review"]

这张图里不必所有节点都用大模型:

节点 实现方式 只接收什么 产出什么
Research Orchestrator 模型或规则 研究主题、来源范围、预算 有边界的子任务
Research Worker Agent Loop 单个子任务、限定工具 带证据的候选事实
Normalize & Merge 普通代码 + 小量模型 所有候选事实 去重、冲突标记后的资料包
Writer 一次模型调用或短 Loop 资料包、写作规范 草稿
Reviewer 规则 + 模型 + 链接检查 草稿、验收标准、来源 结构化审校结果
Publish 普通代码 已通过的草稿、审批状态 发布结果

状态要传事实,不要传整段聊天

可以把共享状态设计为:

from dataclasses import dataclass, field
from typing import Literal


@dataclass
class BriefState:
    topic: str
    evidence: list[dict] = field(default_factory=list)
    draft: str = ""
    review_issues: list[str] = field(default_factory=list)
    revision_count: int = 0
    status: Literal[
        "researching", "writing", "reviewing",
        "approved", "needs_human", "failed"
    ] = "researching"

每个节点只拿到自己需要的状态投影:

Researcher 看:topic、source policy、budget
Writer 看:topic、normalized evidence、style guide
Reviewer 看:acceptance criteria、draft、source map
Publisher 看:approved artifact、approval token

这叫“共享状态,隔离上下文”。共享状态不等于所有节点都读取所有信息。

一个最小的框架无关执行器

下面的伪代码省略了模型和工具实现,但保留了 Graph Runtime 的关键职责:

NODES = {
    "research": research_node,
    "write": write_node,
    "review": review_node,
    "publish": publish_node,
    "human": request_human_review,
}


def next_after_review(state: BriefState) -> str:
    if not state.review_issues:
        return "publish"
    if state.revision_count < 2:
        return "write"
    return "human"


EDGES = {
    "research": lambda state: "write",
    "write": lambda state: "review",
    "review": next_after_review,
    "publish": lambda state: "done",
    "human": lambda state: "done",
}


def run_graph(state: BriefState, start: str = "research") -> BriefState:
    current = start

    while current != "done":
        enforce_policy(current, state)
        state = NODES[current](state)
        save_checkpoint(node=current, state=state)
        current = EDGES[current](state)

    return state

真正上线时还要补:

  • 节点超时和取消。
  • 幂等键。
  • 重试与退避。
  • 并发调度。
  • 状态版本和迁移。
  • trace、指标和成本。
  • 敏感信息处理。
  • 人工审批和恢复入口。

重点不在是否使用某个框架,而在这些运行时语义是否明确。

检查点和恢复为什么重要

长任务最怕“最后一步失败,前面全部重来”。

检查点可以保存某个节点完成后的状态:

checkpoint-01:研究完成
checkpoint-02:资料去重完成
checkpoint-03:初稿完成
checkpoint-04:审校失败,等待第二次修改

恢复时,Runtime 可以从最近的安全检查点继续,而不是把整段历史重新交给模型猜现场。

检查点不是长期记忆

二者用途不同:

概念 保存什么 生命周期
Checkpoint 某次任务在某个时刻的完整状态 主要服务当前任务恢复
Long-term Store 跨任务复用的偏好、规则、经验 可以跨线程或用户长期存在

LangGraph 的 persistence 文档也区分了 thread 内的 checkpoint 和跨 thread 的 store。生产环境还要考虑持久化数据库、加密、保留时间和删除机制,不能只用进程内内存。

有副作用的节点必须幂等

如果节点会发邮件、付款、发布内容或修改数据库,恢复时不能无脑重跑。

常见做法是:

task_id + node_id + operation_id -> idempotency key

执行前先查询这个操作是否已经成功。已经完成就复用结果,没有完成才继续。

如何判断要不要上 Graph

先问一个更重要的问题:

当前最简单的方案,是否已经在真实评测中暴露出结构性问题?

可以按下面顺序判断:

flowchart TD
    A["任务是否一次模型调用就能稳定完成?"] -->|"是"| B["单次调用"]
    A -->|"否"| C{"路径是否基本固定?"}
    C -->|"是"| D["Workflow"]
    C -->|"否"| E{"一个 Agent Loop 能否稳定完成?"}
    E -->|"是"| F["保留单 Loop"]
    E -->|"否"| G{"失败是否来自隔离、并行、验证或恢复?"}
    G -->|"是"| H["引入 Graph"]
    G -->|"否"| I["先修模型、工具、上下文或评测"]

Graph 特别适合这些情况

  1. 需要保护上下文

搜索、写作、验证需要不同信息,不希望互相污染。

  1. 任务可以真正并行

子任务彼此独立,汇合时可以去重和解决冲突。

  1. 需要专业分工

不同节点需要不同模型、工具、权限或提示。

  1. 需要独立验证

结果可以用测试、规则、真实环境或独立标准检查。

  1. 任务必须中断恢复

流程长、工具不稳定,或者需要等待人工批准。

  1. 高风险动作需要治理

发布、付款、删除、写生产数据等动作不能交给自由 Loop 直接决定。

这些情况通常不要急着上 Graph

  • 一次模型调用已经足够。
  • 任务短,而且路径非常开放。
  • 子任务高度依赖同一段上下文。
  • 所谓“并行”其实每一步都依赖前一步。
  • 没有评测,无法证明拆分后变好。
  • 只是为了使用新框架或增加 Agent 数量。

多 Agent 的收益和成本要一起看

Anthropic 在其 Research 系统工程复盘的内部评测中报告:

  • 多 Agent 系统相对单 Agent 提升了 90.2%。
  • token 使用量可以解释 BrowseComp 评测中 80% 的效果差异。
  • Agent 系统使用的 token 约为普通聊天的 4 倍,多 Agent 系统约为 15 倍。

这些数字说明多 Agent 在可并行、信息量大、价值足够高的研究任务里可能很有效,也说明收益很大一部分来自投入了更多计算。

但不要把这些数字当成所有业务的通用结论:

  • 它来自特定 Research 系统和内部评测。
  • 多花 token 不保证你的任务也提高。
  • 高度共享上下文、步骤强依赖的任务未必适合并行。
  • 延迟、费用、失败点和调试成本都会增加。

工程上应比较的是:

质量提升带来的业务价值
          vs
token + 延迟 + 工具 + 运维 + 调试成本

Graph 应该评测什么

只看最终答案,会掩盖图里的结构性问题。

节点级

指标 回答的问题
node_success_rate 这个节点能否按 contract 完成
schema_valid_rate 输出结构是否合格
node_latency 哪个节点最慢
node_cost 哪个节点最贵
retry_rate 哪个节点经常重做
evidence_coverage 研究节点是否给出足够证据

边和路由级

指标 回答的问题
route_accuracy 是否把任务送到正确分支
invalid_transition_rate 是否出现不允许的状态转移
loop_exit_rate 修改循环是否能正常退出
human_escalation_rate 哪类任务最常转人工

整图级

指标 回答的问题
end_to_end_success 最终任务是否完成
time_to_done 总耗时是否可接受
cost_per_success 每次成功平均花多少
recovery_success_rate 中断后能否恢复
duplicate_side_effect_rate 是否发生重复发布、写入或调用
quality_gain_vs_baseline 相比单次调用或单 Loop 是否真的提升

评测时至少保留一个简单基线。否则系统变复杂以后,只能看到“它能运行”,无法知道复杂度是否值得。

如何从现有 Loop 逐步迁移

不要一开始设计一张几十个节点的巨图。更稳妥的顺序是:

第一步:先让单 Loop 可观测

记录:

  • 每轮输入和输出。
  • 工具调用。
  • 预算。
  • 停止原因。
  • 最终结果。

没有 trace,就不知道应该在哪里切节点。

第二步:把状态从聊天历史中拿出来

把目标、产物、错误、预算和审批状态做成结构化字段。

第三步:先拆一个独立验证节点

生成与验证通常边界最清楚,也最容易证明价值。

Agent Loop -> Verifier -> Done / Retry / Human

第四步:只并行真正独立的部分

从一个明确的 fan-out / fan-in 场景开始,不要让所有节点都互相调用。

第五步:增加检查点和恢复

先覆盖最贵、最慢或最容易失败的节点。

第六步:把权限和预算放进 Runtime

不要只在 prompt 里写“最多重试两次”。Runtime 必须真的拒绝第三次重试。

第七步:用评测决定是否继续拆

如果拆分没有改善质量、恢复或成本,就合并节点。图不是越大越先进。

框架怎么选

Graph Engineering 不依赖特定框架。

你可以使用:

  • 普通函数加一个状态表。
  • 消息队列和任务数据库。
  • 工作流引擎。
  • LangGraph 这类有持久化、检查点和 human-in-the-loop 能力的图运行时。
  • Google ADK 这类提供顺序、循环、并行和图式工作流的 Agent 开发工具。

选型时不要先比较“支持多少 Agent”,先问:

状态模型是否明确?
能否持久化和恢复?
条件边能否测试?
有副作用的节点如何幂等?
能否暂停等待人工?
节点权限是否隔离?
trace 是否能串起整张图?
状态 schema 升级怎么办?

小系统用几十行代码就能验证设计。只有当持久化、并发、人工介入和运维需求变复杂时,框架的价值才会明显。

常见误区

“Graph 会取代 Loop”

不会。Graph 的节点内部仍然可以运行 Loop。两者负责不同层级。

“Agent 越多,系统越强”

不一定。Agent 增多意味着更多 token、通信、延迟、失败点和状态同步。

“每个节点都应该是 Agent”

不应该。能用确定性代码完成的检查,就优先用代码。

“所有节点共享完整上下文最方便”

这会重新制造大 Loop 的上下文污染。应该共享结构化状态,再为节点构建最小必要上下文。

“画出 Graph 就拥有确定性”

如果条件边仍由模糊自然语言决定,预算只写在 prompt 里,失败后无法恢复,这张图依然不可靠。

“用了框架就自动具备生产能力”

框架只是运行基础。权限、幂等、评测、观测、数据治理和业务验收仍然要自己设计。

最小设计清单

准备上线一张执行图前,至少检查:

  • 每个节点是否只有一个清楚职责?
  • 节点输入输出是否有 schema?
  • 哪些节点调用模型,哪些使用确定性代码?
  • 条件边是否可以单元测试?
  • 每个节点只拿到必要上下文吗?
  • 共享状态是否有版本?
  • 哪些节点可以并行?
  • fan-in 如何处理缺失、重复和冲突?
  • 每个循环是否有最大次数?
  • 图是否有总预算和取消机制?
  • 失败可以从哪个检查点恢复?
  • 有副作用的节点是否幂等?
  • 高风险动作是否需要人工批准?
  • 是否记录节点、边、状态和成本 trace?
  • 是否和单次调用或单 Loop 基线做过评测?

下一步

建议按这个顺序继续:

  1. Loop Engineering:先理解单个执行单元怎样循环、停止和恢复。
  2. Agent 模式与实现:对照 ReAct、Routing、Orchestrator-Workers 等常见模式。
  3. Multi-Agent 协作、自进化与记忆系统:继续学习责任边界、A2A、共享状态和记忆。
  4. Agent 安全与 Guardrails:把权限、审批和运行时边界补完整。
  5. Agent 效果评测框架:建立节点、路由和端到端评测。
  6. 大型 Agent 系统架构设计:把 Graph 放进任务、工具、记忆和观测平台。

参考资料