阶段：G3-持久化与人机协作 ｜ 预计耗时：90 分钟 ｜ 前置：G3-01 Checkpointer 与 thread_id（本篇离不开 checkpointer） 代码：要你自己敲成
langgraph-learn/examples/g3_approval.py；概念示例可先看 概念指南示例 B 这部分是本框架市场价值最高的能力——企业要的就是"AI 打草稿、人拍板"
学习目标
- 掌握
interrupt(payload)+Command(resume=...)的完整协作回路 - 做出批准 / 拒绝 / 修改三种 resume 的审批流
- 能解释"为什么 interrupt 必须配 checkpointer"（面试级问题）
核心知识
1. 完整回路（四步）
from langgraph.types import interrupt, Command
def risky_step(state: State) -> dict:
decision = interrupt({ # ① 抛出：图在此暂停
"amount": state["amount"],
"question": "金额超过 500，请审批",
})
# ② 图停住，进程可以把 decision 问题交给任何人（命令行/微信/网页）
# ③ 人工处理后，用 Command(resume=...) 恢复，decision 就是 resume 传入的值
if decision["action"] == "approve":
return {"log": ["已批准，金额" + str(state["amount"])]}
if decision["action"] == "reject":
raise ValueError("审批人拒绝")
return {"amount": decision["new_amount"], "log": ["已修改金额"]} # ④ 处理结果写回 State
恢复侧（任何进程/任何时间）：
graph.invoke(Command(resume={"action": "approve"}), cfg) # cfg 里 thread_id 必须与暂停时一致
2. 为什么必须配 checkpointer（背下来）
interrupt 暂停的不是"异常"，是等待：图停在半路，进程甚至可以退出。能恢复靠的是 checkpointer 把暂停时的完整状态（包括"停在哪个节点"）存了下来。没有 checkpointer 的 interrupt = 挂掉就没了。
3. 三种 resume 的语义由你定义
resume 的值就是 interrupt() 的返回值——你传什么，节点里就拿到什么。所以"批准/拒绝/修改金额"不是框架功能，是你的业务约定：传 dict 最灵活（{"action": ..., "new_amount": ...}）。
实验任务
任务 1：命令行审批流（40 分钟）
写 g3_approval.py：
- 图：validate → risky_step(interrupt) → record（模拟报销流水线：读发票金额 → 超过 500 暂停审批 → 记账）
- 用 input() 在命令行问审批人，支持三个命令：
- y → Command(resume={"action": "approve"})
- n → Command(resume={"action": "reject"})
- 500 <数字> → Command(resume={"action": "modify", "new_amount": 数字})
- 三种各跑一遍，记录每种的最终 State
任务 2：跨进程审批（30 分钟，本篇精华）
把脚本拆成两个命令行窗口演示：
- 窗口 A：跑图到 interrupt（Agent 停住）→ 直接关掉窗口 A
- 窗口 B：重新打开脚本，用同一 thread_id 只执行 Command(resume={"action": "approve"}) → 流程从中断处继续走完
- 这段演示就是以后给客户演示的核心卖点：审批可以发生在完全不同的进程、不同的时间、不同的人
任务 3：思考题（15 分钟）
写进笔记：① 审批期间进程挂了/断电了，恢复时会丢什么？（什么都不丢——checkpointer 存的）② 什么操作值得 interrupt？全都要审批行不行？（摩擦成本：低风险高频操作不要打断人）
验收标准
衔接下一阶段
会暂停会恢复 → G3-03 断点续跑与时间旅行：把恢复能力再往上推一层。