LangChain系列之LangGraph 进阶：interrupt、checkpointer 与时间旅行实战
LangGraph 进阶：interrupt、checkpointer 与时间旅行实战
一、Agent 差点把事办砸了
先讲一个产线切片。
我们有个运维助手 Agent，能力是改服务配置。那天它判断流量要涨，自作主张把一个测试参数推上了生产环境。模型没疯，判断也不算离谱，问题出在没人拦它。
更难受的是事后追责。没人说得清它当时看见了什么数据、怎么下的判断，因为跑完什么都没留下。我翻了一下午日志，才勉强拼出一条时间线。
两件事都要治：事前有人拦，事后留得清。
这类事故复盘下来，答案永远指向同一处：这张图太顺了。模型调工具，工具直接执行，一路冲到 END，中间没有任何一道闸。
你缺三样东西。
第一，危险动作前能不能先问我？第二，进程崩了，状态还在不在？第三，出了错，能不能回到几分钟前重新跑？
三问，LangGraph 都有官方机制对应。这篇借一个贯穿案例，把三道闸逐个装上，末尾再送一个复用利器。
先给案例起个名字：高危操作审核 Agent。它有改配置、跑脚本这类危险工具，每次动手前要过人工审批，执行要留档，出错能回溯。后面每一章，都在给这个案例添一块能力。
二、会等：interrupt，给图装一道人工闸口
LangGraph 的答案叫 interrupt。它能在图的执行中途按下暂停键，把控制权交还给人类。
最实用的位置是工具内部。执行危险操作前，先弹一张审批单：
from langchain_core.tools import tool
from langgraph.types import interrupt
@tool
def push_config(change: str) -> str:
"""把配置变更推到线上环境。危险操作，需人工放行。"""
approval = interrupt({
"action": "push_config",
"change": change,
"question": "确认推到生产？回复 yes 放行",
})
if approval == "yes":
return f"已推送：{change}"
return "人工驳回，未执行"
interrupt 接收一个 dict，内容随便写，它就是给人类看的审批单。图在这里挂起，进程可以退出，你的周末不受影响。
人审完怎么续？用 Command：
from langgraph.types import Command
graph.invoke(
Command(resume="yes"),
config={"configurable": {"thread_id": "ops-001"}},
)
resume 的值会成为 interrupt 的返回值，图从断点原地满血复活。
把一次审批的完整时序摆出来看：
| 时刻 | 图的行为 | 你的动作 |
|---|---|---|
| invoke 撞上 interrupt | 图挂起，状态写入 checkpointer，进程可安全退出 | 看审批单，人工判断 |
| invoke Command(resume) | 按 thread_id 找回断点，resume 值回放给 interrupt，节点重跑至原地继续 | 收到执行结果 |
顺带说明位置的选择。interrupt 不只写在工具里，节点函数里同样能调，区别只在粒度：工具内拦截精确到单次调用，节点内拦截覆盖整段逻辑。危险操作都在工具里，所以工具内是主力位置。
还有一门审批单的信息设计。上面那张 dict 就是要给人看的，内容质量决定审批速度。
一次合格的审批单至少带三样：做什么（action 与参数）、影响面（改哪台机器哪个环境）、怎么回滚（出问题用什么命令撤）。只扔一个工具名过去，审批人两眼一抹黑，最后只能全部驳回。
两个前提和两个坑，表格里一次说清：
| 事项 | 说明 |
|---|---|
| 前提 1 | 编译时必须挂 checkpointer，暂停靠它存档 |
| 前提 2 | invoke 时必须带 thread_id，恢复时按它找到断点 |
| 坑 1 | 别用 try/except 包住 interrupt，暂停语义会被异常处理吃掉 |
| 坑 2 | interrupt 之前的副作用要幂等，恢复后这段代码会重跑 |
坑 2 多说一句。恢复的原理是把该节点从头再执行一遍，跑到 interrupt 那行才"回放"出 resume 值。
interrupt 前面如果已经发过邮件，恢复时会再发一封。副作用要幂等，这句建议抄在工位上。
三、会存：checkpointer，进程死了也能续
上一节反复出现的 checkpointer，值得单独一节。上篇我们加过一行 InMemorySaver，那时它只是测试件：状态存在内存里，进程一重启就全丢。
产线要换后端，换法是零成本的，编译时换个对象：
上篇那行 InMemorySaver 没有白写，它让你在零依赖下先理解存档机制。后端只是把"存哪里"换掉，interrupt、恢复、时间旅行的用法一行不变。
from langgraph.checkpoint.sqlite import SqliteSaver
with SqliteSaver.from_conn_string("checkpoints.db") as checkpointer:
graph = builder.compile(checkpointer=checkpointer)
升级到 Postgres 也只换两行，先装包 pip install langgraph-checkpoint-postgres：
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string(DB_URI) as checkpointer:
checkpointer.setup() # 首次运行建表
graph = builder.compile(checkpointer=checkpointer)
如果你的应用是 asyncio 写的，对应换 AsyncSqliteSaver 与 AsyncPostgresSaver，API 同形，方法前加 await 即可。同步异步两套后端官方都维护着，不用自己造轮子。
三个后端怎么选，一张表定夺：
| 后端 | 适用场景 | 代价 |
|---|---|---|
| InMemorySaver | 本地调试、单元测试 | 进程重启全丢 |
| SqliteSaver | 单机工具、个人 Agent、demo 到产线的过渡 | 不扛高并发写 |
| PostgresSaver | 生产多实例部署 | 要管数据库，首次先 setup() 建表 |
thread_id 是另一根轴。每个 thread_id 是一条独立的状态线，隔离粒度自己定：
| thread_id 粒度 | 含义 | 典型场景 |
|---|---|---|
| 每会话一个 | 同一会话内续聊 | 客服对话、多轮任务 |
| 每用户一个 | 跨会话保留该用户上下文 | 个人助理 |
| 每任务一个 | 一次审批流一条线 | 运维工单、审核流 |
你会发现本章的机制全挂在同一个底座上。interrupt 的挂起靠存档，恢复靠取档，时间旅行靠历史档。checkpointer 就是这一切的地基，所以它是四件能力里最该先配的那个。
想随时看图停在哪一步，get_state 一眼便知：
snap = graph.get_state(config)
print(snap.next) # 下一个要执行的节点
snap 里不止有 next。values 是当前完整状态，config 带着当前 checkpoint_id，tasks 描述排队中的任务。排查"图怎么卡住了"时，先把它打出来看。
还有个容易忽略的点：checkpoint 是累积快照，每跑一步存一档，体积随会话变长。长会话场景记得给老 thread 定期归档清理，别让存档库悄悄膨胀。
说句实话，checkpointer 存的只是"会话状态"，离真正的"记忆"还差一层。那个话题留给下一篇，这里先按下。
四、会回：时间旅行，错了能回退重跑
有了 checkpointer，图的每一步都留了档。这让 LangGraph 有个很科幻的能力：时间旅行。
第一步，把一个 thread 的全部 checkpoint 列出来：
history = list(graph.get_state_history(config))
for snap in history:
print(snap.config["configurable"]["checkpoint_id"], snap.next)
第二步，挑一个断点，改掉状态，生成新分支：
target = history[3] # 回到出错前那一步
new_config = graph.update_state(
target.config,
{"messages": [("user", "改用归档脚本清理，别直接删")]},
)
第三步，从新分支续跑，参数已经修正：
graph.invoke(None, new_config)
update_state 不会覆盖原历史，它从断点岔出一条新分支。旧局完整留着，新局重新开跑，这正好是复盘需要的形态。
history 每一项都带元信息：created_at 时间戳、next、当时的 writes，按时间倒序排，最新在前。排查线上问题时，它就是本地的飞行记录仪，不用接外部追踪平台也能回放现场。
它和 interrupt 是一对组合拳。审批单挂起后，批准人三天后才来处理，期间服务重启、发版、扩容，都不影响：状态在数据库里，resume 时按 thread_id 找回断点接着跑。人审天然的异步节奏，被持久化稳稳接住。
给你一个好用的心智模型：图的状态管理就是一套 git。每个 checkpoint 是一次 commit，thread 是仓库，update_state 开分支，历史永远可查。带这套模型去读官方文档，处处都对得上。
这里必须插一个产线警示。时间旅行会重放节点，节点里的工具会真的再执行一次。工具是查库、读文件这类只读操作，无所谓。
工具是发邮件、下单、扣款这类写操作，重放就是真金白银的事故。把非幂等工具包一层幂等保护（用请求 ID 去重），是时间旅行上产线的门票。
顺带一个真实用法。我们排查线上问题时，会把出问题的 thread 整条 history 拉下来，逐个看每步的 next 和状态值，Agent 当时的决策路径一目了然。这个习惯救过我们两次，成本为零。
五、会拆：子图，把流程打包复用
审核流程写好一套，第二个 Agent 也要用同样的审核。复制粘贴一遍？以后改规则得改两处。
LangGraph 的答案是子图：把一张编译好的图，当另一个图里的节点来挂。
最省事的接法，父子图状态结构相同，直接挂：
review_graph = review_builder.compile()
builder.add_node("review", review_graph)
父图跑到 review 节点，整个子图原样跑一遍，两边共享同一份 messages。
状态结构不同，就包一层转换函数：
def call_review(state: MainState) -> dict:
result = review_graph.invoke({"task": state.task})
return {"verdict": result["verdict"]}
builder.add_node("review", call_review)
子图可以单独 invoke 测试，测好了再挂进父图。这层嵌套长这样：
两个实践提醒。子图自己不配 checkpointer 时，会共享父图的存档，interrupt 挂起与恢复在嵌套里照样能用。子图内部如果也有 interrupt，挂起时父图会一并停住，恢复后从子图断点继续，层次再多也只挂起一次。
何时该拆子图，看三个信号：流程要被第二个 Agent 复用、这块逻辑想单独写单元测试、团队想按图划分职责边界。
没踩到信号就先写单图。子图是抽函数，先有能跑的整体，再抽可复用的局部，顺序反了就是过度设计。这条和你写后端服务的直觉完全一致。
多说一句方向。今天你把审核流程打成了子图，明天把检索打成一个，把执行打成一个，父图就只剩调度。多 Agent 系统的地基就是子图，这一步我们后面会正式踩上去。
六、闭环 demo：三关串起来
三道闸都装好了，看它们如何协同。一个完整的高危操作审核流，核心 20 行出头：
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
@tool
def exec_change(cmd: str) -> str:
"""执行一条运维变更指令"""
ok = interrupt({"cmd": cmd, "question": "放行这条指令？"})
if ok != "yes":
return "已驳回"
if "rm " in cmd: # 模拟执行出事
raise RuntimeError("误伤生产目录")
return f"已执行：{cmd}"
def call_model(state: MessagesState):
return {"messages": [model_with_tools.invoke(state["messages"])]}
builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode([exec_change]))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", tools_condition)
builder.add_edge("tools", "model")
graph = builder.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "ops-demo"}}
graph.invoke({"messages": [("user", "清理 /data/tmp 下的旧日志")]}, cfg)
# 图在 interrupt 处挂起，等人审
graph.invoke(Command(resume="yes"), cfg)
# 放行，执行出错，但每一步状态已存档
history = list(graph.get_state_history(cfg))
fixed = graph.update_state(
history[2].config,
{"messages": [("user", "改用归档脚本清理，别直接删")]},
)
graph.invoke(None, fixed) # 回到出错前，换安全方案重跑
这段 demo 用 InMemorySaver 方便你零依赖跑通。想验证"进程死了也能续"，把 checkpointer 换成 SqliteSaver 再跑一遍，中途 Ctrl+C 退出程序，重新启动后用同一个 thread_id 继续 invoke，图会从上次挂起的地方接着走。
三道闸在这段代码里各就各位：interrupt 踩了刹车，checkpointer 留了底片，时间旅行给了倒车键。
拆开看三个关键动作。第一次 invoke 带 thread_id，图在 interrupt 处挂起，此时进程退出也不丢。第二次 invoke 用 Command(resume) 放行，执行炸了，但每一步都有存档。
第三次是时间旅行，回到出错前，换条路重跑。三次 invoke 之间的对照表：
| 关卡 | 机制 | 一句话 |
|---|---|---|
| 会等 | interrupt + Command(resume) | 危险动作停下来问人 |
| 会存 | checkpointer + thread_id | 每步留档，重启能续 |
| 会回 | get_state_history + update_state | 回到过去，开新分支 |
整个流程画出来是这样：
模型只会往前冲，图替你把住了方向。
七、收尾：上产线的三条 checklist
回顾这一篇。上篇你画出了图，这篇图学会了四件事：会等（interrupt 人工闸口）、会存（checkpointer 跨进程续跑）、会回（时间旅行回溯分叉）、会拆（子图打包复用）。
上产线前，三条 checklist 送你：
- 危险操作必须挂 interrupt，人审放行
- checkpointer 换数据库后端，SQLite 起步，Postgres 收尾
- 所有写操作工具做成幂等，时间旅行才有安全感
三条都过，你的 Agent 才算从"能跑"走到"敢上产线"。这三条不是终点，是我们从搜索产线一路踩坑攒下的底线配置，你可以照着加码。
下篇我们正式拆"记忆"。今天 checkpointer 存的会话状态只是短期记忆这一半，长期记忆是另一套系统：向量库、跨会话召回、记忆的写入与遗忘。
关注我的个人公众号后台回复「LangGraph」，可以领取LangGraph实战教程