状态图编排:持久执行与检查点

状态图编排:持久执行与检查点

本节摘要:Agent 与工作流共享一个问题:当一个 40 步的运行在第 38 步失败,你想从第 38 步续上,而不是从头重来。二流的状态模型让运维只能在一个「假设每次都是全新运行」的库周围 hack 重试。LangGraph 的设计回答是:状态是一等公民的类型化对象,变更显式,每个节点之后都持久化检查点——恢复就是一次 load_state(session_id) 调用。本节吃透 LangGraph 的核心模型:Agent 是状态机,节点是纯函数 (state) -> state_update,边是条件或直连转移,状态在每个节点后被检查点化,可在任意失败处从上一个成功检查点恢复。它提供四大能力——持久执行、流式、人在环中、综合记忆——以及三种编排拓扑:监督者、对等群体(Swarm)、层次化嵌套子图。Klarna、Uber、J.P. Morgan 等生产用户证明了「图形状 + 检查点」让恢复变廉价。本节用标准库从零实现一个带类型化状态、条件边、检查点/恢复循环的状态图。

对应原课程:Phase 14 · Lesson 13 · langgraph-stateful-graphs(原英文 phases/14-agent-engineering/13-langgraph-stateful-graphs/docs/en.md)。

学习目标

阅读完本节,你应当能够:

1. 描述 LangGraph 的核心模型:带类型化状态的状态机、函数节点、条件边、节点后检查点。
2. 说出文档强调的四大能力:持久执行、流式、人在环中、综合记忆。
3. 解释 LangGraph 支持的三种编排拓扑:监督者、对等(Swarm)、层次化(嵌套子图)。
4. 用标准库实现一个状态图:类型化状态、条件边、检查点/恢复循环。
5. 诊断检查点过小、非确定节点、条件边滥用三大反模式。

一、问题与直觉

状态为什么必须一等公民

Agent 与工作流共享一个问题:当一个 40 步的运行在第 38 步失败,你想从第 38 步续上,而不是从头重来。二流的状态模型让运维只能在一个「假设每次都是全新运行」的库周围 hack 重试。

LangGraph 的设计回答:状态是一等公民的类型化对象,变更显式,检查点在每个节点后持久化。恢复是一次 load_state(session_id) 调用。

图的定义

一个图由四样东西定义:

• 状态类型(State Type) —— 一个类型化字典(或 Pydantic 模型),每个节点读它、改它。
• 节点(Nodes) —— 纯函数 (state) -> state_update,返回后把 update 合并进状态。
• 边(Edges) —— 节点间的条件或直连转移。
• 入口与出口(Entry/Exit) —— START 与 END 哨兵节点标出边界。

例子:一个带 classify、refund、bug、sales、done 节点的 Agent——把第 12 节的路由工作流表达成图。

持久执行

每个节点返回后,运行时把状态序列化写进一个检查点器(SQLite、Postgres、Redis、自定义)。第 N 步失败时,运行时可 resume(session_id),带着精确状态从第 N+1 步续上。

LangGraph 文档明确点名了这件事关乎生产用户:Klarna、Uber、J.P. Morgan。论点不在图形状本身,而在图形状 + 检查点让恢复变廉价。

💡 设计要点:状态一等公民的真正价值不在「能存」,而在「能精确恢复」。当失败发生,你不是重跑整个昂贵流程,而是从最后一个成功节点带着完全一样的状态续上。这对长流程(审批、合规、多步研究)是质变——一次 LLM 调用的钱省下来,远超检查点器的运维成本。

四大能力

1. 持久执行(Durable Execution) —— 节点后序列化状态,失败可精确恢复。
2. 流式(Streaming) —— 每个节点可 yield 部分输出,图按「每节点增量」事件流给调用方,UI 随图运行更新。
3. 人在环中(Human-in-the-Loop) —— 节点间检查并修改状态。实现:关键节点前暂停、把状态摆给人、接受修改、恢复。检查点器让这件事变简单,因为状态本就已序列化。
4. 综合记忆(Memory) —— 短期(单次运行内,状态里的对话历史)与长期(跨运行,通过检查点器加独立长期库)。LangGraph 通过工具接入外部记忆系统(Mem0、自建)。

三种编排拓扑

1. 监督者(Supervisor) —— 中央路由 LLM 派发给专家子 Agent。langgraph-supervisor 里的 create_supervisor()(不过 LangChain 团队 2026 年建议直接通过工具调用做这件事,以获得更多上下文控制)。
2. 群体 / 对等(Swarm / Peer-to-Peer) —— Agent 通过共享工具面直接交接,无中央路由。
3. 层次化(Hierarchical) —— 监督者管理子监督者,实现为嵌套子图。

这个模式在哪里出错

• 检查点过小 —— 只检查点对话轮,会让工具状态与记忆写入不可恢复。完整状态必须序列化。
• 非确定节点 —— 恢复假设节点输入产生同样的状态更新。随机种子、墙钟、外部 API 必须被捕获。
• 条件边滥用 —— 每条边都条件的图,是一个无法推理的状态机。优先线性链加偶尔分支。

二、从零实现

原课程 code/main.py 用标准库实现一个状态图:

• State —— 类型化字典,带 messages、step、route、output、human_approval。
• Node —— 可调用对象,取状态返回 update 字典。
• StateGraph —— 节点 + 边 + 条件边 + run + resume。
• SQLiteCheckpointer(内存版替身)—— 每个节点后序列化状态;load(session_id) 恢复。
• 演示图:classify → 分支(refund / bug / sales)→ 人工闸门 → send。

核心骨架如下,用伪代码展示。

Step 1:类型化状态 + 节点

from typing import TypedDict

class State(TypedDict):
    messages: list
    step: int
    route: str           # refund / bug / sales
    output: str
    human_approval: bool

def classify(state: State) -> dict:
    route = llm_classify(state["messages"][-1])
    return {"route": route, "step": state["step"] + 1}

def refund(state: State) -> dict:
    return {"output": "退款已处理", "step": state["step"] + 1}
​

Step 2:图 + 条件边

class StateGraph:
    def __init__(self, state_type):
        self.nodes = {}
        self.edges = {}             # node -> 下一步(直连名或条件函数)

    def add_node(self, name, fn):
        self.nodes[name] = fn

    def add_conditional_edges(self, name, router_fn, mapping):
        # router_fn(state) -> key; mapping[key] -> 下一个节点名
        self.edges[name] = ("cond", router_fn, mapping)

    def add_edge(self, src, dst):
        self.edges[src] = ("direct", dst)
​

Step 3:检查点 + 恢复

class SQLiteCheckpointer:
    def __init__(self):
        self.store = {}             # session_id -> [state 快照...]

    def save(self, session_id, state):
        self.store.setdefault(session_id, []).append(dict(state))

    def load(self, session_id):
        snaps = self.store.get(session_id, [])
        return snaps[-1] if snaps else None

def run(graph, state, session_id, ckpt):
    while state.get("step_node") != "END":
        node_fn = graph.nodes[state["step_node"]]
        update = node_fn(state)
        state.update(update)
        ckpt.save(session_id, state)       # 每个节点后检查点
        state["step_node"] = next_node(graph, state)
    return state

def resume(graph, session_id, ckpt):
    state = ckpt.load(session_id)          # 精确恢复
    return run(graph, state, session_id, ckpt)
​

运行 python3 code/main.py 会展示第一次运行在人工闸门失败、持久化、然后恢复产出最终输出。

💡 设计要点:resume 之所以能精确恢复,全靠「节点是纯函数 + 状态完全序列化」两个前提。如果某个节点读了墙钟、随机数、或外部 API 而没把这些捕获进状态,恢复后重放就会得到不同结果——这是「非确定节点」反模式的根。

三、框架对比

• LangGraph —— 参考,生产就绪。用 create_react_agent、create_supervisor,或自建图。
• AutoGen v0.4(第 14 节)—— Actor 模型替代,适合高并发场景。
• Claude Agent SDK(第 17 节)—— 托管 harness,内置会话存储。
• 自建 —— 需要精确控制状态形状或检查点器后端时。

四、可复用产物

本节产出一份可复用技能(原课程 outputs/skill-state-graph.md):

• skill-state-graph.md:给定任意目标运行时,生成一个 LangGraph 形状的状态图,内置检查点与恢复接线。包含一份状态设计清单(状态字段是否覆盖工具结果与记忆写入?哪些字段该进检查点?reducer 怎么合并并发 update?),以及一份恢复可靠性清单(节点是否纯函数?随机种子/墙钟/外部 API 是否捕获?条件边是否克制?)。

Python 代码(code/main.py)是独立可运行的状态图骨架,类型化状态、条件边、检查点/恢复都是后端无关的;把内存检查点器换成真实 SQLite/Postgres 即可投入生产。

五、练习

1. (Easy) 加一条从 classify 到 end 的条件边:分类置信度低于阈值时走它。人工手动设置 route 后恢复运行。
2. (Medium) 把类 SQLite 替身换成真实 SQLite 检查点器。测量每步序列化开销。
3. (Medium) 实现并行边:两个节点并发跑,用自定义 reducer 合并。不可变状态在这里买到了什么?
4. (Hard) 读 langgraph-supervisor 参考。把玩具移植到 create_supervisor。对比轨迹形状。
5. (Hard) 加流式:每个节点运行时 yield 部分状态。把增量边到达边打印。

本节要点回顾

1. 状态一等公民:类型化对象、显式变更、每节点后检查点,恢复是 load_state(session_id)。
2. 图四要素:状态类型、节点(纯函数)、边(条件/直连)、START/END 哨兵。
3. 持久执行:节点后序列化,失败从最后成功节点带精确状态续上。
4. 价值在精确恢复:省下重跑整个昂贵流程的钱,远超检查点器运维成本。
5. 四大能力:持久执行、流式、人在环中、综合记忆。
6. 三种拓扑:监督者(中央路由)、群体(对等交接)、层次化(嵌套子图)。
7. 检查点过小是反模式:只存对话轮会让工具状态与记忆写入不可恢复,完整状态必须序列化。
8. 非确定节点是恢复的天敌:随机种子、墙钟、外部 API 必须捕获进状态。
9. 条件边要克制:每条边都条件的图无法推理,优先线性链加偶尔分支。
10. 2026 推荐:直接用工具调用做监督者,比 create_supervisor 给更多上下文控制。

下一节,我们看 Actor 模型——AutoGen v0.4 用消息传递的并发原语替代状态机,当你的场景是高并发多 Agent 协作时,它比状态图更合身。