阶段:G3-持久化与人机协作 | 预计耗时:60 分钟 | 前置:G3-02 interrupt 人工审批 代码:要你自己敲成
langgraph-learn/examples/g3_timetravel.py一句话定位:G3-02 教的是"等人",本篇教的是"回到过去"——两者都吃同一套检查点
学习目标
- 掌握
get_state / get_state_history:读当前状态、翻全部历史快照 - 实现两种"回到过去":从任意旧状态重跑(时间旅行)、中途改状态再续跑(分叉修正)
- 能把"断点续跑"讲成一个 5 分钟客户演示脚本
核心知识
1. 检查点历史是"版本树"
cfg = {"configurable": {"thread_id": "a"}}
graph.get_state(cfg) # 当前状态(停在哪个节点、State 全量)
for snap in graph.get_state_history(cfg): # 全部历史快照,从新到旧
print(snap.config, snap.values.get("messages", "")[:30])
print(" 下一步:", snap.next) # 这个快照之后将要执行的节点
每个快照都带自己的 config(含 checkpoint_id)——旧快照的 config 就是"时光机票"。
2. 三种玩法
| 玩法 | 做法 | 用途 |
|---|---|---|
| 时间旅行重跑 | graph.invoke(None, 旧快照的config) |
从任意历史点重新执行 |
| 修改后分叉 | graph.update_state(快照config, {"key": 新值}) 再 invoke(None) |
"要是当时改对了就好了" |
| 查看卡在哪 | graph.get_state(cfg).next |
调试:图停在哪个节点、State 里是什么 |
3. 断点续跑 ≠ interrupt(区别要分清)
- interrupt:业务流程主动暂停等人(审批)
- 断点续跑:故障中断后从最后快照恢复(机器重启、进程被杀)
- 实现上是同一机制:有 checkpointer 的图,崩溃后用同一 thread_id invoke(None) 就从"最后一个完成步骤"接着走
实验任务
任务 1:制造 5 步历史(15 分钟)
用 g0_studio_demo.py 那类纯逻辑图(不花 API 钱)扩展成 5 个节点(step1~step5 依次执行,每步往 log 追加一行),跑一次让它留下 6 个以上快照,get_state_history 全部打印出来。
任务 2:时间旅行(20 分钟)
- 挑第 3 步之后的旧快照,用它的 config
invoke(None, cfg)——观察 log 从那里重新生长 update_state把第 4 步的某个字段改成不同值再 invoke(None)——对比"分叉后"与"原时间线"的最终结果差异- 两个结果都打印存档
任务 3:断点续跑演示(20 分钟)
在 5 节点图的 step3 节点里故意 raise KeyboardInterrupt(模拟崩溃):
- 崩溃后 get_state(cfg).next 应显示 ("step3",)
- 修掉崩溃代码后 invoke(None, cfg)——从 step3 接着走完
- 这就是给客户演示的完整脚本:录成视频放进作品集(G8 会用)
验收标准
衔接下一阶段
会回到过去了 → G3-04 Store 长期记忆:checkpointer 只管一个会话,跨会话的记忆要用 Store。