Agent 与 Workflow 如何协作？状态应该由谁保存
内容摘要：不是所有事都该交给模型决定。本章讲确定性工作流与模型驱动决策的边界，介绍状态机、有向无环图（DAG）、事件驱动三种编排模型，以及 Agent 状态的持久化与恢复——这是长任务与故障恢复的基础。 难度：进阶 | 前置知识：08 Agent Loop
常见误区
对话历史就是 Agent 的完整状态；进程重启后把 messages 重新发给模型，就能可靠地从断点继续。
先给结论
生产系统通常用确定性 Workflow 管理关键状态转换，只把无法穷举的局部决策交给模型。任务阶段、工具结果、审批和副作用必须进入显式状态与 Checkpoint，不能只存在自然语言历史中。
带着这三个问题阅读：
- 状态机、DAG 与事件驱动编排分别解决什么问题？
- 对话上下文、业务状态和执行 Checkpoint 有何区别？
- 恢复任务时如何避免重复执行已经成功的副作用？
一、确定性 vs 模型驱动：先划边界
原则：能确定的就确定。把流程骨架写成确定性工作流，只在真正需要判断的节点交给模型。这样更可测、更可控、更便宜。很多"Agent 项目"其实是"工作流 + 几个模型节点"。
二、三种编排模型
| 模型 | 结构 | 适合 | 说明 |
|---|---|---|---|
| 状态机（State Machine） | 状态 + 转移 | 有明确阶段、需审批暂停 | 易表达"等待人工"等停顿 |
| 有向无环图（DAG） | 无环任务依赖 | 批处理、并行、无循环 | 天然并行，不能表达循环 |
| 事件驱动（Event-driven） | 事件 + 处理器 | 异步、长任务、解耦 | 扩展性好，调试更难 |
Agent 常把三者组合：外层状态机管阶段、某阶段内用 DAG 并行检索、跨阶段用事件解耦。
三、状态：Agent 的事实来源
把状态设计成显式、可序列化的结构，节点之间只通过状态通信，不用隐式全局变量：
from dataclasses import dataclass, field, asdict
from typing import Literal
import json
@dataclass
class AgentState:
task_id: str
goal: str
phase: Literal["planning", "gathering", "await_approval", "executing", "done", "failed"] = "planning"
findings: list[str] = field(default_factory=list)
pending_action: dict | None = None # 待审批的高危动作
approved: bool = False
version: int = 1 # Schema 版本，便于迁移
def to_json(self) -> str:
return json.dumps(asdict(self), ensure_ascii=False)
@classmethod
def from_json(cls, s: str) -> "AgentState":
return cls(**json.loads(s))状态分层（务必分开存，不要混成一坨）：
四、持久化与恢复（Checkpoint）
长任务、需要暂停/审批的任务，必须把状态持久化到可靠存储，进程重启后能恢复：
class CheckpointStore:
"""最小 Checkpoint 接口。生产用 Postgres/Redis 等，示例用内存。"""
def __init__(self):
self._data: dict[str, str] = {}
def save(self, state: AgentState) -> None:
self._data[state.task_id] = state.to_json()
def load(self, task_id: str) -> AgentState | None:
raw = self._data.get(task_id)
return AgentState.from_json(raw) if raw else None
# 恢复即"读最近状态 + 从对应 phase 继续"
store = CheckpointStore()
store.save(AgentState(task_id="t1", goal="排查慢查询", phase="gathering"))
resumed = store.load("t1")
assert resumed and resumed.phase == "gathering"幂等要点：恢复后可能重放某个节点，写操作节点必须用幂等键防重复（15）。
五、失败路径
- 状态藏在内存/全局变量：进程一挂就全丢，无法恢复。
- 状态 Schema 变更不兼容：升级后旧 checkpoint 反序列化失败——用
version字段做迁移或失效策略。 - 恢复后重复写：无幂等键导致重复创建工单/重复扣款。
六、测试与验证
- 测"崩溃-恢复"：保存到某 phase，重建对象从 checkpoint 恢复，断言不重复执行已完成节点。
- 测状态迁移：旧版本 JSON 能否被新代码安全加载或明确拒绝。
七、安全与权限
- 持久化状态可能含敏感信息，需加密/脱敏与访问控制。
await_approval阶段的高危动作在获批前不得执行（13）。
八、性能与成本
- Checkpoint 频率是权衡：太频繁增加 IO，太稀疏丢进度。按节点边界保存通常合适。
九、框架差异
- LangGraph（1.0）以"状态图 + 持久化 checkpoint + interrupt 人工审批"为核心卖点；
langgraph.prebuilt的能力在 1.0 后迁入langchain.agents。LlamaIndex 用事件驱动 Workflows。选型见 20。
十、关联章节
面试复述
Workflow 决定可预测的主路径，Agent 处理局部动态决策；显式状态才是事实来源。恢复时应读取 Checkpoint 和幂等记录，而不是让模型从对话文本猜测进度。
复述时先给结论，再补机制、适用边界和一个可验证指标；不要只罗列框架名称或产品功能。
迁移练习
把一个三步 Agent 任务改写为显式状态机，列出每个状态的进入条件、可执行动作、持久化字段和恢复位置。
检索关键词
确定性工作流 vs Agent 状态机 DAG 事件驱动 Agent 状态持久化 Checkpoint 恢复 状态 Schema 版本 幂等恢复
官方来源
- LangGraph 持久化/中断文档；工作流引擎（如 Temporal）文档作为工程对照（
verified_at: 2026-09-07）。