# 使用LangChain+LangGraph开发AI应用（9） - LangGraph 进阶


![](https://static.xiongneng.me/langchain-langgraph-cover-20261017190854.png)

## 学习进度

这一篇给图装上两个进阶器官。持久化执行，图跑一半进程死了，换个进程接着跑。人工协同，图跑到关键步骤停下来，把问题递给人，等一个答复再继续。这两件事共享同一个地基，checkpointer 在每个 super-step 落一个状态快照。

围绕这个地基往深里走，中断、恢复、回放、落盘。示例在 `source/hello_persist.py`，九个 demo，全部纯函数不依赖模型。完整案例是一条带人工审批的发布流水线 `source/release.py`，流程跑到审批节点把摘要递给终端，你按 y 或 n，它按你的选择走部署或驳回。

## 前言用途

agent 跑长任务有两类时刻最要命。一类是意外，进程崩了、网断了、超时了，从头重跑意味着把已经花钱的模型调用全部重来。另一类是主动的，流程走到删库、转账、群发这类不可逆动作前，得停下来等人点头。两类问题的解法是同一套，把状态周期性落盘，停了就从最近的快照接着来。

LangGraph 的答案是 checkpointer。图每推进一个 super-step，状态快照就写进 checkpointer，中断、崩溃、重启之后，带着同一个 thread_id 回来，图从快照处继续。配合 interrupt 函数，任意节点内部都能声明一个暂停点，把任意数据递给外部，恢复时拿回一个答复。审批、确认、填表、改参数，人机协同的所有场景都是这一个动作的变体。

什么时候需要这些。多轮对话要续接状态，长流程要容错，或者流程里有需要人拍板的步骤，就是这一篇的主场。多智能体编排里子图之间的状态恢复，同样压在这套机制上。

## 原理概念

checkpointer 的存储单位是 checkpoint，一个 thread 里按 super-step 排成一条链。每个 super-step 跑完，图状态的快照落一个 checkpoint，链上每个快照带自己的 config（thread_id 加 checkpoint_id）、元数据（step 步数、writes 来源）和下一步要执行的节点。第 7 篇见过的 get_state 就是取这条链的最新一环，get_state_history 把整条链翻出来。

落盘的节奏也能调。stream 或 invoke 时传 `durability` 参数，三档可选。`"exit"` 只在图退出时落盘，性能最好，进程崩了中间状态全丢。`"async"` 异步落盘，性能和持久性折中，崩溃时有小概率丢最近的快照。`"sync"` 在下一步开始前同步落盘，最稳，代价是每步多一次写盘。默认行为在 exit 与 sync 之间，生产上按业务能接受丢失多少来选。

![](https://static.xiongneng.me/interrupt-flow-20261017190835.png)

中断的实现比直觉朴素。interrupt 函数被调用时，往运行时抛一个特殊的异常，引擎捕获它，把当前状态落进 checkpointer，然后停住。这个实现决定了三件事。中断可以发生在节点内任意位置、任意条件下，是动态的。恢复时节点从第一行重新执行，interrupt 之前的代码会再跑一遍。interrupt 的恢复值按调用顺序严格对号，一个节点里有几个 interrupt，恢复值就按顺序填进去。

恢复只有一条路，`Command(resume=...)`。它的值原样成为 interrupt 函数的返回值，图从断点继续。checkpointer 用内存版跑单进程演示，换 SqliteSaver 或 PostgresSaver，快照落进数据库，进程重启、换台机器都能接着审批。官方还有一套静态断点 `interrupt_before`、`interrupt_after`，编译图时把节点名钉进去，主要用于调试，人机协同推荐用 interrupt 函数。

checkpointer 是个可替换的接口，官方维护了一排实现。`langgraph-checkpoint` 是基础包，InMemorySaver 在里面。SqliteSaver 和 PostgresSaver 在单独的包里，要自己装。MongoDB、Azure Cosmos DB 也各有官方包。想自己写一个，实现四个方法就够，put 存快照、put_writes 存中间写入、get_tuple 取快照、list 列快照，快照序列化默认用 JsonPlusSerializer，Pydantic 模型、dataclass、datetime 都能直接存，不支持的对象可以开 pickle 回退。存储里还有个容错设计值得知道，某个 super-step 中途失败时，同批已成功节点的写入会单独存档，恢复执行时已成功的节点不会重跑。

## 核心用法

九个用法，示例在 `source/hello_persist.py`，全部纯函数图。共享骨架是一个三节点的审批图，work 干活，approval 里放 interrupt，finish 收尾。

### 1. 中断闭环

最核心的一条链，暂停、恢复、按答复路由。

```python
from langgraph.types import Command, interrupt

def approval(state: State):
    answer = interrupt({"question": "批准吗？", "task": state["task"]})
    return {"approved": answer}

graph.add_conditional_edges(
    "approval", lambda s: "finish" if s["approved"] else END)
```

```python
result = app.invoke({"task": "删库跑路", ...}, config)
# result["__interrupt__"][0].value -> {'question': '批准吗？', 'task': '删库跑路'}
final = app.invoke(Command(resume=True), config)
```

```text
== 中断闭环 ==
  中断 payload: {'question': '批准吗？', 'task': '删库跑路'}
  恢复 True（批准）: 已处理 发布文档 -> 人工通过
```

第一次 invoke 返回值里多出 `__interrupt__` 键，payload 原样躺在里面。恢复值 True 让条件边选了 finish 分支，结论带着「人工通过」。恢复 False 则直接到 END，结果停在 work 的产出上，流程等于被人按了驳回。

用 stream 驱动时呈现方式不同，中断信息在事件流的 `stream.interrupts` 里，`stream.interrupted` 为 True。交互循环的写法是固定的三段，发起输入跑一遍，看 `interrupted` 判断要不要问人，问完把 `Command(resume=答复)` 作为下一轮输入，直到不再中断。官方把这段循环叫 human-in-the-loop loop， Web 服务里就是「请求进来跑图，发现中断返回审批页面，用户提交后 resume 继续」这个模式。

### 2. 结构化 payload

interrupt 的参数可以是任意 JSON 值，问题怎么问、选项给什么，自己定义。

```python
picked = interrupt({"options": ["加急", "普通"], "default": "普通"})
```

```text
== 结构化 payload ==
  发出去的问题: {'options': ['加急', '普通'], 'default': '普通'}
  收回来的答案: 加急
```

payload 递出去，答案递回来，中间隔了多久、隔了多少进程都无所谓，checkpointer 记着呢。

一个节点里放多个 interrupt 也行，恢复值按调用顺序严格对号，第一个 interrupt 拿第一个 resume。配套的纪律是别用条件分支跳过 interrupt 调用，这次执行问了两个问题、下次只问一个，恢复值就对不上号了。并行分支同时中断的场景，恢复值换成 `{interrupt.id: 答复}` 的映射，每个中断用自己的 id 领答案。还有一条，payload 只放 JSON 能序列化的简单值，函数、类实例传不进去。

### 3. 节点从头重跑

恢复时节点从第一行重新执行，这是中断机制最容易被忽略的性质。

```python
def flaky(state: State):
    counter["n"] += 1          # interrupt 之前的代码，恢复时会再跑
    answer = interrupt("确认？")
    return {"answer": answer}
```

```text
== 节点重跑 ==
  节点函数执行了 2 次（中断一次 + 恢复重跑一次）
```

计数器实锤了重跑。把副作用放在 interrupt 之前，就会执行两次。写节点的时候要么把副作用挪到 interrupt 之后，要么设计成幂等的，upsert 代替 insert。

### 4. 吞异常坑

用 try/except 包住 interrupt，中断就没了。

```python
def bad(state: State):
    try:
        answer = interrupt("确认？")
    except Exception:
        answer = "fallback"    # 中断靠异常实现，这里把它吞了
    return {"answer": answer}
```

```text
== 吞异常坑 ==
  图没有暂停，直接得到: fallback
```

图一口气跑完，没人被问到任何问题。interrupt 的暂停机制就是抛异常，except Exception 把它拦下来，引擎收不到信号。要处理业务异常，用具体的异常类型，别用裸的 Exception。

### 5. 静态断点

编译图时把断点钉死在节点边界上，恢复用 invoke 传 None。

```python
app = g.compile(checkpointer=checkpointer, interrupt_before=["b"])
app.invoke({"n": 2}, config)

snapshot = app.get_state(config)
# snapshot.next == ("b",)
resumed = app.invoke(None, config)
```

```text
== 静态断点 ==
  停在 ('b',) 前，状态 {'n': 3}
  invoke(None) 续跑: {'n': 30}
```

断点停在 b 节点之前，a 的产出已经在状态里。invoke 传 None 意思是不给新输入，从断点继续。断点除了编译时钉死，还能在每次 invoke 时临时指定，`app.invoke(inputs, interrupt_before=["b"], config=config)`，调试时想断哪步断哪步。这种断点在节点边界上，不能带条件，人机协同还是用 interrupt 函数。

### 6. 状态历史

get_state_history 把线程的整条 checkpoint 链翻出来。

```python
history = sorted(
    app.get_state_history(config), key=lambda s: s.metadata.get("step", 0))
```

```text
== 状态历史 ==
  step=-1 next=('__start__',) values={}
  step=0 next=('a',) values={'n': 2}
  step=1 next=('b',) values={'n': 3}
```

每个 super-step 一条快照，step 是步数，next 是下一步要跑的节点。注意这里只有三条，b 还没执行，step=2 的快照不存在，快照在 super-step 边界落地，没跑完就没有。链长了还可以按条件筛，`next(s for s in history if s.next == ("b",))` 找某节点执行前的快照，`s.metadata["step"] == 2` 按步数找，`s.metadata["source"] == "update"` 找 update_state 产生的分叉。排查「图停在哪了」，get_state_history 一眼看清。

### 7. 时间旅行

从历史快照的 config 重新 invoke，之前的步骤直接跳过，之后的重新执行。

```python
history = list(app.get_state_history(config))
step1 = next(s for s in history if s.metadata.get("step") == 1)
replayed = app.invoke(None, step1.config)
```

```text
== 时间旅行 ==
  从 step=1 的 checkpoint 回放: {'n': 30}
```

快照之前的计算成果都还在，不用重花钱。回放不等于修改，想改历史状态再继续，用 update_state 在快照上分叉出新线程。回放时 interrupt 会重新触发，适合重跑一次带人工确认的流程。

### 8. SqliteSaver 落盘

内存版进程一退就没了，换 SqliteSaver，快照落进一个 SQLite 文件。

```python
import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("release_state.db", check_same_thread=False)
app = build_approval_graph(SqliteSaver(conn))

# 进程 A 发起，中断退出
# 进程 B 打开同一个文件接着审批
final = app.invoke(Command(resume=True), config)
```

```text
== SqliteSaver 落盘 ==
  第一段连接中断: True
  新连接恢复: 已处理 发布上线 -> 人工通过
```

demo 里关掉第一个连接、开第二个新连接模拟换进程，恢复照常。连接要传 `check_same_thread=False`，引擎在内部线程池里写快照，默认的 sqlite 连接会报跨线程错误。生产换 PostgresSaver，接口不变。

### 9. response_schema 校验

给 interrupt 配一个 Pydantic 模型，恢复值不符合 schema 直接被拦。

```python
class Decision(BaseModel):
    approved: bool
    reason: str

answer = interrupt("审批", response_schema=Decision)
```

```text
== response_schema ==
  非法恢复值被拦下: ValidationError
  合法恢复值: True（没问题）
```

恢复 `{"approved": "yes"}` 这种类型不对的值，抛 ValidationError，图还停在断点上，换个合法的再按一次就行。校验通过的恢复值直接是 Decision 实例，字段访问不用再解析字典。要做的校验比 schema 复杂怎么办，官方的推荐姿势是把错误信息存进状态，用条件边跳回同一个节点再问一次，每个 interrupt 保持每次执行只调用一次。别在节点里写 while True 循环包 interrupt，恢复是从节点开头重跑的，循环几轮就把之前几轮全部重放，执行次数指数级涨。

## 完整案例

九招都练完了。收个尾，搭一条真的带人工闸门的流水线，`release.py`，发布审批。plan 节点把变更描述整理成摘要，approve 节点把摘要递给人工，等一个 Decision，批准走 deploy，驳回走 reject。

```bash
uv run python release.py "把搜索阈值从 0.8 调到 0.75" --auto-approve
```

```text
变更摘要: 变更「把搜索阈值从 0.8 调到 0.75」已通过静态检查，涉及 1 个配置文件，风险等级低。

结论: 已发布。审批备注：无人值守演示，自动批准
```

--auto-approve 是无人值守模式，跳过终端问答直接批准。去掉这个开关，流程会在审批处停下来，把摘要打在终端上，等你敲 y 或 n，再让你留一句备注。你敲的每个字经 Decision 校验后成为恢复值。交互长这样。

```text
变更摘要: 变更「把搜索阈值从 0.8 调到 0.75」已通过静态检查，涉及 1 个配置文件，风险等级低。
批准发布？(y/n) y
审批备注（可空）: 低风险配置变更

结论: 已发布。审批备注：低风险配置变更
```

按 n 走的是另一条分支，同一个图，结论变成「已驳回」，备注原样带上。路由写在 approve 节点内部，用上一节的 Command(goto=) 完成，批准与驳回共享同一个中断点。

```bash
uv run python release.py "把搜索阈值从 0.8 调到 0.75" --db release_state.db
```

--db 让快照落进 SQLite 文件。第一次运行发起发布，流程停在审批，你关掉终端下班。第二天再打开，用同样的 thread_id 把 Command(resume) 发过去，图从断点接着跑。审批可以隔夜，状态不会丢。

这套机制还能钻进工具内部。interrupt 在 @tool 装饰的函数里同样可用，模型调用 send_email 这类工具时，函数体里的 interrupt 先暂停，把收件人、主题、正文一起递给人，人可以批准、改参数再放行、或者取消，恢复值决定工具最终怎么执行。给 agent 加「敏感操作人工闸门」就是这一招，模型自己决定调工具，人决定放不放行。

测试在 `tests/test_hello_persist.py` 十二条加 `tests/test_release.py` 七条，中断 payload、恢复路由、节点重跑计数、吞异常、普通输入撞断点、静态断点、历史步数、时间旅行、schema 拦截与重试都钉住了。加上前八篇的，全仓一百五十二条，全绿。

## 源码

[https://github.com/yidao620c/xneng-langchain-langgraph](https://github.com/yidao620c/xneng-langchain-langgraph)

## 参考

- [LangGraph 官方文档 - Persistence](https://docs.langchain.com/oss/python/langgraph/persistence)
- [LangGraph 官方文档 - Checkpointers](https://docs.langchain.com/oss/python/langgraph/checkpointers)
- [LangGraph 官方文档 - Human-in-the-loop](https://docs.langchain.com/oss/python/langgraph/human-in-the-loop)


---

> 作者: XiongNeng  
> URL: https://xiongneng.me/posts/ai/langchain/langchain-langgraph-09-persistence/