让智能体状态可查、进度可恢复
文章主张用显式状态机替代隐含执行位置的循环,为智能体定义命名状态、类型化数据和持久化检查点。作者以 LangGraph 工单分流项目说明这种设计的工程价值。
AI 深度解读
Agent 能在演示里连续调用工具,不代表它能在生产环境里可靠运行。进程重启后从哪里继续,退款前怎样等人审批,出错时如何查清走到了哪一步,这些问题很难靠一份消息历史回答。
作者用一个基于 LangGraph 的工单分流项目 relayg,主张把 Agent 建成有明确节点、类型化状态和持久化检查点的状态机。这套做法值得看,但收益来自执行状态的管理,不能把所有可靠性问题都算作框架已经解决。
为什么消息列表不能代表执行进度
常见 Agent 示例只有一个 while True:把 state["messages"] 交给模型,发现 reply.tool_calls 就调用 run_tool,把结果追加回消息列表,否则返回回答。
写起来短,问题也藏得深。
首先,循环执行到哪里,并没有一个可以直接读取的业务状态。查一次失败,往往需要逐行翻日志,推断模型上轮做过什么、下一轮准备做什么。
其次,消息列表把上下文和业务数据混在了一起。退款金额是否存在、审批结论有哪些取值、当前步骤能读哪些字段,都可能等到运行时才暴露问题。
最后,循环通常依赖当前进程里的内存。要暂停到第二天,再接着处理,就得额外保存中间结果、记录执行位置,并编写恢复逻辑。
作者真正反对的是把执行进度藏在循环和消息历史里。
这也解释了为什么加几条日志还不够:日志可以帮助还原经过,却不直接提供下一步执行所需的完整状态。
把模型判断、业务规则和执行动作拆开
relayg 只有三个节点,执行顺序是 START → classify → policy_check → act → END。作者用 StateGraph(TicketState) 建图,通过 builder.add_node 注册节点,再用 builder.add_edge 连接流程。
三个节点承担不同责任:
classify读取工单,用 LLM 或确定性的 mock 完成分类。policy_check根据规则表,用纯 Python 判断退款政策。act执行模拟动作,并写入审计记录。
这里最有用的设计,是只有第一个节点接触 LLM。识别工单意图有模糊空间,适合交给模型;退款是否符合政策,则需要按明确规则执行。
状态对象把这些步骤之间的数据约定写了出来:
class TicketState(TypedDict, total=False):
ticket_id: str
subject: str
body: str
classification: Classification # pydantic 模型:intent、urgency、refund_amount
policy_decision: PolicyDecision # Literal["auto_approve", "needs_approval", ...]
approved: bool
actions: Annotated[list[dict], operator.add]
要关注最后一行:operator.add 是状态合并规则,节点返回新的 actions 时,会追加到列表,而不是覆盖旧记录。
作者还规定,policy_decision 只能取四种字面量值,不过原文没有列全;Classification 则是包含 intent、urgency、refund_amount 的 Pydantic 模型。这样,政策节点读取退款金额时,有明确的数据结构可依赖。
但类型约定也有边界。TypedDict 主要服务静态检查,total=False 还允许字段缺省,因此不能据此保证所有缺失字段都会变成编辑器报错。模型输出是否经过正确的 Pydantic 校验、节点是否满足前置条件,仍需要实现来保证。
检查点让等待审批成为执行流程的一部分
作者在编译图时传入 graph = build_graph(checkpointer=SqliteSaver(...)),让每个节点执行后的状态持久化。进程重启后,运行可以从最近的检查点恢复,不必只依赖内存。
更直观的例子是退款审批:relayg 规定,金额在 50 到 200 美元之间的退款需要人工批准。原文没有说明区间端点是否包含。
act 节点遇到这类退款,会提交暂停原因和审批所需信息:
verdict = interrupt({
"reason": "refund_needs_approval",
"amount": amount,
"ticket_id": state["ticket_id"],
})
关键是 interrupt():它把等待人工决定纳入图的执行过程,审批人员不用一直在线,运行进程也不用持续占着资源等待。
审核完成后,调用 graph.invoke(Command(resume={"approved": True, "note": "ok"}), config),就能把批准结果交回暂停的运行。作者表示,relayg 包含这条中断与恢复路径的测试,也能用离线 demo 观察暂停过程。
不过,原文没有展开 config 如何关联具体运行,也没有说明审批页面、身份校验和回调入口怎样实现。「不用手写恢复位置」可以成立,「整个审批系统都不需要接线」则超出了示例展示的范围。
我的判断
我认同把业务规则从模型里拿出来,也认同把执行位置变成可检查、可持久化的数据。对需要审批、排障和恢复的 Agent,这两点比继续堆提示词更直接。
但作者把 while-loop 和状态机的对立说得偏满了。循环同样可以配合数据库保存状态,也可以实现恢复;只是当分支、审批和失败处理逐渐增加时,团队很容易自行拼出一套不完整的工作流系统。状态机的优势,是提前给这些问题一个明确结构。
还有一个生产边界:保存状态,不等于外部动作只会执行一次。
示例中的 act 执行的是模拟动作。真实退款如果已经成功,而检查点尚未写入,恢复后如何防止重复退款?动作记录与磁盘审计日志怎样保持一致?原文没有交代。检查点也不应被理解为保存了整个 Python 调用栈,恢复时仍需考虑节点重入和副作用。
作者没有提供两种方案的故障率、开发成本或性能对照,所以这是有项目支撑的工程经验,不能当作普遍优越性的实验证明。对于简单脚本、无需人工介入、失败后可以直接重跑的任务,作者自己也建议继续用循环。
能拿来做什么
- 检查现有 Agent 能否直接回答「当前处于哪一步、下一步依赖什么数据」。如果只能翻消息历史推断,可以先定义业务状态,再决定是否引入 LangGraph。
- 把工单分类与退款政策分开:模型输出结构化分类,程序检查金额和规则,执行节点处理动作与审计,分别验证每一步的输入输出。
- 为审批流程验证一次完整恢复:触发
interrupt()后结束进程,再通过Command(resume=...)恢复同一运行,确认审批结果和状态都能接上。 - 接入真实退款等外部动作之前,补上重复执行的防护,并检查动作成功但状态保存失败时的处理方式,避免把检查点当成完整的事务保证。