第 5 篇用 create_agent 把标准 Agent 循环收进一个函数：模型推理、工具调用、结果回填全部自动流转，接工具只是往参数里添几行。自动化的另一面随之浮现——转账、删库、群发邮件这类不可逆操作，模型一旦看走眼，损失瞬间落地，没有撤销键。工具接得越多、Agent 跑得越自主，越需要在关键动作前加一道人工闸门：机器装填，人来扣扳机。这就是人机协同（human-in-the-loop），第 1 篇预告的四大超能力之一"中断恢复"，本篇正面兑现。
版本口径与全系列一致：langgraph 1.2.x，PyPI 独立发布，机制以interrupts 官方文档为准。主角两个：interrupt() 在节点内按下暂停键，Command(resume=...) 带着人的决定续跑。中断依赖 checkpointer 存档，本篇只给 MemorySaver 的最小配置、够跑通即可，持久化的完整机制留到第 8 篇展开。全文按"场景 → 最小示例 → 恢复 → 响应形态 → 审批单 → 多中断点 → 邮件审批实战 → 总结"组织。
1、为什么需要人工审批
先给工具分个类。查询类工具——搜索、读库、看天气——错了顶多答非所问，重跑一次就翻篇，可逆。写操作类工具——转账、下单、发邮件、删记录——改变外部世界，执行即定局。大模型的幻觉率永远不是零，把不可逆动作完全交给模型，等于把风险敞口开到最大。
官方人机协同文档列的场景几乎全是这一类：审批关键 API 调用、审改模型产出、人工校验输入。思路一致：自动流程跑到关键决策点，控制权交还给人，人处理完再交回机器接着跑。注意这不是降低自动化程度，而是把"最后一步授权"从概率系统手里挪到确定性系统手里。
LangGraph 提供两档刹车。老的一档是静态断点：compile 时传 interrupt_before=["节点名"]，图跑到该节点之前必停，恢复时输入传 None 续跑，适合调试时逐节点单步。新的一档是 interrupt() 动态中断：写在节点函数内部任意位置，可以按业务条件触发，还能把待审信息带出去。对比如下：
| 维度 | 静态断点 | 动态中断 |
|---|---|---|
| 声明位置 | compile 的参数 |
节点函数内部 |
| 触发条件 | 到达节点即停 | 任意位置按条件停 |
| 中断时带数据 | 不带 | 载荷交给调用方 |
| 典型用途 | 调试单步 | 审批、审改、校验 |
一句话分工：静态断点管"在哪儿停"，动态中断管"为什么停、停下来交给人什么"。生产级审批流用后者，本篇全程只讲它。
2、interrupt 最小示例
最短可运行闭环：一笔转账，执行前必须有人放行。代码如下：
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt
from langgraph.checkpoint.memory import MemorySaver
class State(TypedDict):
amount: int # 转账金额
approved: bool # 审批结论
def transfer(state: State) -> dict:
# 图停在这一行：先拿到人的决定
ok = interrupt(f"转账 {state['amount']} 元，是否放行？")
return {"approved": ok}
builder = StateGraph(State)
builder.add_node("transfer", transfer)
builder.add_edge(START, "transfer")
builder.add_edge("transfer", END)
graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "t1"}} # 档案抽屉号
result = graph.invoke({"amount": 500, "approved": False}, config)
解读分两层。用法层只有一行：interrupt() 接收任意可 JSON 序列化的值（这里是一句审批问话），第一次执行到此处函数不会返回，而是抛出特殊异常通知运行时暂停，运行时用 checkpointer 把当前状态存档，invoke 随即正常返回。配置层三件套缺一不可：checkpointer、thread_id、interrupt 调用——不挂 checkpointer，机制无从存档；不传 thread_id，运行时不知道档案存进哪个抽屉；同一张图可同时挂起 N 笔审批，各抽屉互不串门。真正的转账 API 调用应写在 interrupt 之后，原因第 4 节展开。
顺带交代 MemorySaver：来自 langgraph.checkpoint.memory，官方新名是 InMemorySaver，MemorySaver 是沿用多年的别名，两条导入等价，细节见官方 Memory 文档。它把档案放在进程内存里，重启即失，只适合开发验证，生产要换落盘实现。暂停到恢复的完整生命周期画成时序：
解读：暂停与恢复是同一条线程上的两次 invoke，中间隔多久都行——五秒的点击确认，或五天的周五下午审批，对图来说没有差别，这就是"等待外部输入"的全部含义。
3、用 Command 恢复执行
第 2 节的图此刻停在问句上，恢复就是再 invoke 一次，输入换成 Command。放行与驳回两路演示如下：
from langgraph.types import Command
result = graph.invoke(Command(resume=True), config)
print(result["approved"]) # True：放行，转账执行
# 换一条新线程，演示驳回
config2 = {"configurable": {"thread_id": "t2"}}
graph.invoke({"amount": 9800, "approved": False}, config2)
r2 = graph.invoke(Command(resume=False), config2)
print(r2["approved"]) # False：驳回，路由可走取消分支
解读三件事。其一，Command 与 interrupt 同住 langgraph.types，resume 的值就是 interrupt() 的返回值——第一次执行停在问句，恢复后 transfer 节点从头重跑，再执行到那一行时，函数带着人的决定返回了。其二，恢复必须用同一个 thread_id：它是档案抽屉号，换新号等于开新档案，图从头跑起，审批自然对不上号。其三，approve 与 reject 就是两个布尔分支，节点里拿返回值 if 一下，再配上第 4 篇的条件边，即可分流到"执行"与"取消"两个下游节点。另有进阶形态 Command(resume=..., update={字段: 值})，恢复的同时给黑板补写增量，审改场景偶尔顺手。
4、中断时的响应形态
中断发生时 invoke 不报错，而是正常返回：黑板已有字段照旧——amount 在、approved 还是初始 False——旁边多出一个 __interrupt__ 键。官方口径：默认 invoke() API 把本轮触发的中断放在这个键下，值是 Interrupt 对象的列表。读取代码与结果如下：
intr = result["__interrupt__"][0] # 本轮触发的中断，可能不止一个
print(intr.value) # 载荷：转账 500 元，是否放行？
print(intr.id) # 中断编号，恢复时按它配对
解读：value 就是 interrupt() 传入的载荷，id 是这处中断的唯一编号——单中断用不上它，第 6 节的并行场景按它配对。列表形态说明一轮可能触发多个中断。事件流接口 stream_events 则把中断挂在 stream.interrupts 上、用 stream.interrupted 标记是否停机，同一机制的另一副面孔，本系列主线保持 invoke。
中断不是返回值，是异常；恢复不是续行，是重跑。
后半句就是官方明文的规则：恢复执行时，整个节点从头重跑，不接着 interrupt 那一行往下走。由此引出四条纪律，整理如下：
| 纪律 | 原因 |
|---|---|
别用 try 包住 interrupt() |
暂停靠抛特殊异常，except 会把它截走，中断失效 |
| interrupt 之前要幂等 | 节点重跑，之前的副作用会再执行一遍 |
| 别条件跳过节点内 interrupt | 多个 interrupt 按出现顺序以索引配对，数量变了就错乱 |
| 载荷只放可序列化的简单值 | checkpointer 要存档，函数与类实例存不进去 |
这四条的根子是同一条：重跑。工程上更稳的姿势是一个节点只放一个 interrupt，多道审批拆成多个节点，下一节正是这么做的。
5、审批单模式传递数据
True 与 False 只够二值审批，真实的审批要"看得见细节、改得了内容"。interrupt() 的载荷可以是任意 JSON 结构——把事由、金额、草稿正文打包成一张审批单带出去，人看完或改完，把结果作为 resume 值送回来。以报销审批为例，黑板上有 reason、total、receipts 三个字段：
def approve_expense(state: State) -> dict:
decision = interrupt({
"事由": state["reason"],
"金额": state["total"],
"单据": state["receipts"], # 待审材料整包带出
})
if decision is False: # 驳回
return {"status": "rejected"}
return {"status": "passed", "final": decision}
# 人工改了金额再放行：resume 值是修改后的内容
graph.invoke(Command(resume={"金额": 800, "事由": "客户招待"}), config)
解读：载荷在停的时候带出去，resume 值在续的时候送进来，一来一回构成完整的审批交互；final 存的是人改过的版本，而不是机器的原稿——审改语义由此落地。回传值的形态由你与前端约定，常见三态如下：
| 回传形态 | 含义 |
|---|---|
| 布尔 True 或 False | 二值审批，放行或驳回 |
| 修改后的字典或字符串 | 审改模式，人的版本写回黑板 |
| 函数或类实例 | 不可行，序列化这关过不去 |
进阶一句：interrupt 还接受第二个参数 response_schema（Pydantic 模型或 TypedDict），回传值自动校验，前端还能据此渲染输入表单，需要强约束的审批位值得用。
6、多中断点依次审批
一张图不止一道闸。采购单要先过金额审、再过折扣审，两道闸门两个节点：图停在第一道，放行后前进到第二道再停，一次一停、逐个人审。导入沿用第 2、3 节，完整可跑：
class POState(TypedDict):
total: int
off: int
amount_ok: bool
discount_ok: bool
def approve_amount(s: POState) -> dict:
return {"amount_ok": interrupt(f"金额 {s['total']} 元，是否同意？")}
def approve_discount(s: POState) -> dict:
return {"discount_ok": interrupt(f"折扣 {s['off']}%，是否同意？")}
builder = StateGraph(POState)
builder.add_node("approve_amount", approve_amount)
builder.add_node("approve_discount", approve_discount)
builder.add_edge(START, "approve_amount")
builder.add_edge("approve_amount", approve_discount")
builder.add_edge("approve_discount", END)
graph = builder.compile(checkpointer=MemorySaver())
运行与两次恢复如下：
config = {"configurable": {"thread_id": "po-1"}}
r = graph.invoke({"total": 9800, "off": 15,
"amount_ok": False, "discount_ok": False}, config)
print(r["__interrupt__"][0].value) # 第一道闸先停
r = graph.invoke(Command(resume=True), config)
print(r["__interrupt__"][0].value) # 放行，停在第二道
r = graph.invoke(Command(resume=False), config)
print(r["amount_ok"], r["discount_ok"])
输出如下：
金额 9800 元，是否同意？
折扣 15%，是否同意？
True False
解读：第一次 invoke 停在 approve_amount；放行后流程前进，立刻停在 approve_discount——每停一次就要一次恢复，几道闸就是几轮交互，每轮都能换人审。同一节点内放多个 interrupt 也被支持，恢复值按出现顺序以索引配对，但条件分支一旦让两次执行的 interrupt 数量不同，配对立刻错乱，这正是第 4 节纪律三的来历；并行分支同时中断时，一次恢复可按 Interrupt 的 id 配对多个值，官方文档另有专节，本篇不展开。
7、邮件草稿审批流实战
机制凑齐，合体做一个真东西：给客户写邮件——机器生成草稿，人工审改（可改可驳），通过则发送、驳回则带意见回炉再写。流程画出来：
解读：人工审改节点内部是 interrupt，审完之后用第 4 篇的条件边按"黑板上有无定稿"分流。State 与三个节点如下：
class MailState(TypedDict):
request: str # 写信要求
draft: str # 当前草稿
note: str # 驳回意见
email: str # 定稿正文
sent: bool
def write_draft(s: MailState) -> dict:
if s["note"]: # 驳回后按意见重写
return {"draft": f"按[{s['note']}]重写的草稿 v2"}
return {"draft": f"关于{s['request']}的草稿 v1"}
def review(s: MailState) -> dict:
decision = interrupt({"主题": s["request"], # 审批单
"草稿": s["draft"]})
if decision is False: # 驳回：留意见回炉
return {"note": "语气太生硬，重写"}
return {"email": decision} # 放行：定稿入黑板
def send(s: MailState) -> dict:
return {"sent": True} # 演示口径：真发送换 SMTP 客户端
建图连边、首轮运行与两轮恢复一气呵成：
def after_review(s: MailState) -> str:
return "send" if s["email"] else "write_draft"
builder = StateGraph(MailState)
builder.add_node("write_draft", write_draft)
builder.add_node("review", review)
builder.add_node("send", send)
builder.add_edge(START, "write_draft")
builder.add_edge("write_draft", "review")
builder.add_conditional_edges("review", after_review)
builder.add_edge("send", END)
graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "mail-1"}}
r = graph.invoke({"request": "续约事宜", "draft": "", "note": "",
"email": "", "sent": False}, config)
print(r["__interrupt__"][0].value["草稿"]) # 草稿 v1 待审
r = graph.invoke(Command(resume=False), config) # 驳回：回炉重写
print(r["__interrupt__"][0].value["草稿"]) # 草稿 v2 再次待审
final = "王总您好，续约方案已备好，详见附件。"
r = graph.invoke(Command(resume=final), config) # 改好的定稿放行
print(r["sent"], r["email"] == final)
输出如下：
关于续约事宜的草稿 v1
按[语气太生硬，重写]重写的草稿 v2
True True
走读一遍：write_draft 首轮产出 v1；review 打出中断，审批单带出主题与草稿；resume=False 驳回，note 写入意见，路由送回 write_draft 重写 v2，review 再次中断；resume 传入定稿文本，email 写入黑板，路由放行，send 收尾。三个工程点。其一，review 里 interrupt 之前的代码只做载荷拼装，天然幂等，节点重跑无害——第 4 节纪律的直接应用。其二，真实场景把 write_draft 换成一次模型调用（第 5 篇的口径），图结构一字不改。其三，驳回循环也是循环，别忘了第 2 篇的 recursion_limit 兜底（1.0.6 起默认 1000），机器不该无限重写，人也不该无限驳回。
8、总结
本篇把第 1 篇预告的"中断恢复"兑现成一组固定动作：节点内 interrupt(payload) 按下暂停，checkpointer 存档、thread_id 定抽屉；invoke 返回里的 __interrupt__ 读载荷与编号；Command(resume=...) 用同一 thread_id 续跑，True 放行、False 驳回、改后的内容就是审改结果；多道闸门拆成多个节点依次过。纪律四条——不 try 包、之前幂等、不条件跳过、载荷可序列化——根子都是一条：恢复时节点从头重跑。
练习两道。练习一：给第 7 节的邮件流加一道抄送闸门——send 之前新增 cc_check 节点，interrupt 带出收件人与抄送列表，驳回时回到 write_draft，跑通一图两处中断。练习二：给第 5 节的报销审批挂上 response_schema（提示：Pydantic 模型，金额字段约束大于零），传一个金额为负的字典恢复，观察 ValidationError 当场报错。
最后交代存档的去处：本篇的 MemorySaver 把档案放在进程内存里，进程一退、所有待审单子蒸发，这与审批"挂起等人"的天性相冲——生产环境必须换 SQLite、Postgres 这类落盘的 checkpointer，快照怎么打、线程怎么管、时间旅行怎么用，见持久化文档，第 8 篇把完整机制展开。
