把 Agent 的不确定性关进确定性的笼子里：外层流程可预测，内层判断交给模型。这就是生产系统最常见的样子。
为什么需要编排
一个 while 循环能跑通 Demo，但很难处理：
- 任务需要运行几十分钟，进程重启怎么办？
- 某一步失败，只重跑这一步，还是整个流程？
- 需要人工审批，审批可能隔天才回来；
- 多个分支、并行、汇合；
- 出问题时要看到完整路径。
这些是工作流引擎要解决的问题。图编排把流程显式建模为状态、节点和边，让系统可以在任意一步暂停、恢复、重试和观测。
图的基本元素
- State：整个流程共享的数据结构；
- Node：一个执行单元，可以是模型调用、工具、纯函数、人工等待；
- Edge：固定的下一步；
- Conditional Edge：根据状态选择分支；
- Checkpoint：每一步后的状态快照；
- Interrupt：暂停等待外部输入（人工审批、异步回调）。
State 设计：只放必要的、可序列化的数据
from typing import TypedDict, Literal, Annotated
import operator
class State(TypedDict):
task_id: str
goal: str
category: Literal["billing", "technical", "refund", "other"]
messages: Annotated[list, operator.add] # 只追加
evidence: list[dict] # 带来源的事实
draft: str | None
approved: bool | None
attempts: int
errors: Annotated[list[dict], operator.add]
trace_id: str原则：
- 可序列化：能存 JSON，才能做检查点；
- 显式：不要依赖闭包或全局变量；
- 增量更新：用 reducer 定义「追加」还是「覆盖」，避免并发写冲突；
- 有版本：State Schema 变更时，旧检查点要能迁移或标记不可恢复。
检查点与持久化执行
每一步执行后保存状态：
checkpoint = {
"task_id": "...",
"node": "draft",
"state": {...},
"created_at": "...",
"attempt": 2
}恢复时：
- 读取最新 checkpoint；
- 确认该节点的副作用是否已发生（幂等键）；
- 从下一个节点继续，而不是从头重跑。
这让长任务可以跨进程、跨重启、跨天运行。也是人工审批能挂起的前提。
如果节点内调用了外部 API 并成功，但在写 checkpoint 前崩溃，恢复时可能重复调用。写操作必须用幂等键，或先记录「意图」再执行。
人工介入：Interrupt 的正确姿势
错误做法：在 Agent 循环里 input() 等待，进程一直挂着。
正确做法：把人工审批建模为一个节点：
def approval_node(state):
if state.get("approved") is not None:
return state # 已恢复
raise Interrupt({
"type": "approval",
"summary": state["draft"][:500],
"action": "send_email",
"risk": "对外发送，不可撤回",
"options": ["approve", "reject", "edit"],
})流程挂起，状态落库。审批者在前端看到摘要、动作、影响范围，选择后通过 API 恢复执行。
审批界面要展示：
- 将要发生什么（动作、参数、目标）；
- 为什么（触发规则、模型理由）；
- 影响范围（金额、收件人、不可逆性）；
- 可选项（批准、拒绝、修改后批准）。
只给一个「同意」按钮，等于让人盲目签字。
重试、超时与幂等
每个节点独立配置：
| 配置 | 说明 |
|---|---|
timeout | 单节点最长执行时间 |
max_retries | 重试次数，指数退避 |
retry_on | 只对暂时性错误重试（超时、429、5xx） |
idempotency_key | 写操作去重 |
fallback | 重试耗尽后的降级路径 |
不要对整个流程统一重试。一个 10 步的流程从头重跑，可能重复发送 3 封邮件。应该只重跑失败节点，并保证副作用幂等。
一个最小图实现
不依赖框架也可以写得很清楚：
NODES = {
"classify": classify,
"retrieve": retrieve,
"draft": draft,
"validate": validate,
"approve": approve,
"send": send,
}
def run_graph(state, start="classify", max_steps=20):
node = start
for _ in range(max_steps):
save_checkpoint(state, node)
handler = NODES[node]
try:
state = handler(state)
except Interrupt as it:
state["interrupt"] = it.payload
save_checkpoint(state, node)
return state # 等待外部恢复
except RetryableError as e:
state = retry_or_fallback(state, node, e)
node = route(node, state)
if node == "END":
break
return state
def route(node, state):
if node == "classify":
return "retrieve" if state["category"] != "other" else "draft"
if node == "validate":
return "draft" if not state["valid"] and state["attempts"] < 2 else "approve"
if node == "approve":
return "send" if state["approved"] else "draft"
if node == "send":
return "END"
return NEXT[node]真实框架（LangGraph、Temporal、Cloudflare Workflows 等）会替你处理持久化、并发、可视化和恢复。但理解这些原语，才能判断框架是否适合。
图编排 vs 直接写代码
| 场景 | 直接写代码 | 图编排 |
|---|---|---|
| 3–5 步、无持久化 | ✓ 更简单 | 过度设计 |
| 需要人工审批 | 难 | ✓ |
| 长任务、可恢复 | 难 | ✓ |
| 多分支、并行 | 一般 | ✓ |
| 需要可视化 | 无 | ✓ |
| 团队非工程师维护 | 难 | ✓（低代码平台） |
| 极致性能 | ✓ | 有调度开销 |
一个实用判断：如果你需要「暂停后继续」，就需要编排。 否则普通函数 + 队列可能更简单。
测试图
图的最大好处是可测试。至少覆盖：
- 节点单测：给定 state，输出是否符合预期；
- 路径测试：分类为 refund 时是否走审批；validate 失败是否回到 draft；
- 恢复测试：在 approve 前杀掉进程，能否从检查点恢复；
- 幂等测试：重复执行 send，是否只发一封；
- 超时测试：某节点卡住，是否按策略降级。
这些测试不需要真实模型——用固定输出的 fake model 即可。模型质量用评估集测，流程正确性用单元测试测，两者分开。
反模式
「agent_node」里跑 20 步循环，外层图形同虚设。应该把关键步骤拆成节点，让流程可见、可恢复。
把 50MB 的 PDF 放进 state，检查点写入爆炸。存引用（URL/ID），用时再取。
检索需要 5 秒，人工审批需要 2 天。超时必须按节点配置。
只显示「是否批准退款」，不显示金额、订单、政策依据。审批者无法做判断，只能盲签。
State Schema 改了，旧检查点恢复失败。上线前考虑兼容或清理策略。
检查清单
- 流程有显式状态，且可序列化
- 每个节点有超时、重试策略、错误分类
- 写操作幂等，检查点与副作用一致
- 人工审批是节点，不是阻塞等待
- 审批界面展示动作、参数、影响、不可逆性
- 能从任意检查点恢复
- 关键路径有单元测试与恢复测试
- 全链路 trace 与节点级指标
小结
- 图编排把流程显式化，换来可恢复、可审批、可观测。
- State 要可序列化、显式、带 reducer 和版本。
- 检查点让长任务跨进程、跨天运行，但副作用仍需幂等。
- 人工审批应建模为可挂起的节点，并展示足够上下文。
- 短流程直接写代码，需要暂停/恢复时再上图。
下一篇进入记忆：Agent 如何跨步骤、跨会话记住该记的，忘掉该忘的。