📚 AI 工程与智能体教程站

简体中文

English

简体中文

English

第 6 章 · 人工审批、中断与恢复 ​

本章实现一个不会发送任何真实邮件的审批流。先展示草稿，再显式批准或拒绝，最后记录结果；模型不能替人决定是否批准。

工作机制与设计选择 ​

interrupt() 暂停图并保存待恢复状态，需要 checkpointer 与 thread_id。恢复使用相同线程的 Command(resume=...)。被中断的节点从头重新执行，不是从 Python 源码下一行继续；因此中断前的副作用可能重复。把外部写操作放在审批之后，并用业务幂等键保护。

示例 ​

python
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

class State(TypedDict):
    draft: str
    status: str

def review(state: State):
    approved = interrupt({"draft": state["draft"], "question": "Approve this draft?"})
    return {"status": "approved" if approved is True else "rejected"}

builder = StateGraph(State)
builder.add_node("review", review)
builder.add_edge(START, "review")
builder.add_edge("review", END)
graph = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "draft-001"}}
paused = graph.invoke({"draft": "Welcome to the course!", "status": "pending"}, config)
print(paused["__interrupt__"][0].value)
answer = input("Type APPROVE to approve; anything else rejects: ")
result = graph.invoke(Command(resume=answer == "APPROVE"), config)
assert result["status"] in {"approved", "rejected"}
print(result["status"])

运行与验收 ​

分别运行两次：第一次输入 APPROVE，第二次输入其他文本。两条路径都应结束。本例默认返回格式从 interrupt 读取暂停；如果显式设置 version="v2"，则应使用 result.interrupts 和 result.value，不要混用。

容易踩的坑 ​

不要用宽泛 try/except 吞掉 interrupt 的控制信号，也不要把第一次 invoke 误当作完成。恢复时新建 thread_id 会找不到原暂停点。

🧪 随堂测验

点击你认为正确的选项。答错时会展示正确答案与原因解析。

1. 恢复应该使用哪个线程？

2. 恢复时节点从哪里开始？

3. 需要什么才能暂停恢复？

4. 审批超时可以默认批准吗？

动手实践 ​

  1. 把审批值改为包含备注的 JSON 字典，并验证字段。
  2. 中断前打印一条日志，观察恢复时是否重复。
  3. 设计审批超时后的拒绝策略，不自动批准。

官方参考 ​

核验日期：2026-09-17。