第 10 篇结尾留了一句话：下一篇讲人机协同，让 Agent 在关键动作前停下来等你点头。这篇兑现承诺。create_agent 参数表里那两个"停下来"的入口——interrupt_before 参数与 HumanInTheLoopMiddleware——只是表层封装，底下是同一套机制：LangGraph 的动态中断。本篇把这套机制拆到最小零件：图在哪停、停下来状态存哪、人的决定怎么传回去。
Agent 接管的动作越真实，"会不会闯祸"就越不是概率问题而是设计问题——工具列表里只要有写操作，模型的每次决策都带着真实代价，而人工审批是给不可逆动作上的最后一道保险。全文按"介入场景 → 中断原理 → 最小示例 → 恢复值形态 → 断点选型 → 工具审批实战 → 总结"推进，代码基于 langgraph 1.2.x（2026-10 基线），环境沿用第 2 篇的 uv 项目。
1、为什么需要人工介入
第 1 篇给 Agent 下过定义：自主使用工具完成目标。自主性是收益，也是风险源。读错一个接口，重跑一次就完；给全公司客户群发一封错邮件、删掉一张生产表、付一笔不该付的款——这些动作不可撤销，错误没有重试机会。模型的可靠性可以逼近但永远到不了 100%，不可逆动作的可靠性却必须接近 100%，中间的缺口只能由人来补。
工程上的解法不是把自主性整个收回，而是分级放行。三类典型场景值得停下来：
- 高风险动作审批：发邮件、删数据、付款、发部署，执行前过一道人工确认
- 内容审核：AI 起草、人工把关——公告、客户回复、代码变更，发布前改两笔
- 方向纠偏：长任务中途汇报下一步计划，人工点头或转向后再继续
人工介入不是对 Agent 的不信任，而是对不可逆动作的工程保险——闸门只装在最贵的那一步。
这套模式叫 HITL（Human-in-the-Loop，人机协同），要回答三个工程问题：图在哪停、状态存哪、人的决定怎么传回去。LangGraph 的答案依次是 interrupt()、checkpointer、Command(resume=...)，下一节串起来看。
2、interrupt 机制原理
interrupt() 是节点内可调用的中断原语，与恢复用的 Command 同在 langgraph.types 模块。它在节点函数体内任意位置调用，参数是一份 JSON 可序列化的"申请单"——想让审批人看到什么，就放什么。完整的暂停-恢复时序如下：
调用方 LangGraph 运行时 checkpointer
│ │ │
│ invoke(input, thread_id) │ │
│───────────────────────────────►│ │
│ │ 执行 plan：产出待审动作 │
│ │ 执行 approve：调用 interrupt() │
│ │──暂停 + 写入断点状态────────────►│
│◄───提前返回，携带 __interrupt__──│ │
│ │ ……等待人工，时长不限…… │
│ （人工查看申请单，做出决定） │ │
│ invoke(Command(resume=值)) │ │
│───────────────────────────────►│──按 thread_id 读断点状态────────►│
│ │ approve 节点从头重跑 │
│ │ interrupt() 返回人工的值 │
│◄────────────最终 state─────────│ │
时序图里有五个关键点。其一，节点执行到 interrupt() 时向运行时抛出内部信号，图在这一步停住。其二，当前状态连同断点位置写入 checkpointer，按 thread_id 归档——所以 HITL 有两个硬前提：编译时挂 checkpointer、调用时带 thread_id，缺一个中断时直接报错。其三，invoke 提前返回，结果里多一个 __interrupt__ 键，装着申请单。其四，等待期不占任何计算资源，隔几分钟还是隔一天由审批人定；跨进程、跨机器恢复则要求持久化后端，InMemorySaver 只活在当前进程，第 12 篇展开。其五，恢复时用同一个 thread_id 再调一次 invoke，输入换成 Command(resume=值)，interrupt() 那一行返回人工给的值，节点继续往下执行。
第 5 点藏着最容易踩的坑：恢复不是从 interrupt() 那一行续跑，而是把整个节点从头重跑。机制细节见 Interrupts 官方文档与 interrupt API 参考。
注意：节点恢复时从头执行，interrupt() 之前的代码会跑第二遍——副作用要么做成幂等，要么挪到 interrupt() 之后；也别用裸 try/except 包住 interrupt()，它靠特殊异常实现暂停，会被吞掉。
3、最小审批示例
用一张两节点的图把机制跑通：plan 产出高危动作，approve 停下来等人。先定义 State 与两个节点：
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
class State(TypedDict):
action: str # 待审批的高危动作
result: str # 执行结果
def plan(state: State) -> dict:
# 真实场景里这一步由模型节点产出，示例直接写死
return {"action": "DELETE FROM orders WHERE id=1001"}
def approve(state: State) -> dict:
decision = interrupt({ # 申请单：暂停并递给前端
"action": state["action"],
"risk": "高危：删除订单，不可撤销",
})
if decision: # resume 值流回这里
return {"result": f"已执行：{state['action']}"}
return {"result": "已取消：人工驳回"}
解读：plan 在完整 Agent 里就是模型节点的产出位，示例写死一条删除语句，把注意力留给审批。approve 里的 interrupt() 携带申请单（动作 + 风险说明）暂停，返回值就是人工的决定——True 执行、False 驳回，语义完全由你自己定义。接着组装成图，第一次调用：
from langgraph.checkpoint.memory import InMemorySaver
builder = StateGraph(State)
builder.add_node("plan", plan)
builder.add_node("approve", approve)
builder.add_edge(START, "plan")
builder.add_edge("plan", "approve")
builder.add_edge("approve", END)
graph = builder.compile(checkpointer=InMemorySaver()) # HITL 的前提
config = {"configurable": {"thread_id": "t-001"}}
r1 = graph.invoke({}, config) # 第一次调用：断在 approve
print(r1["__interrupt__"][0].value) # 打印申请单
输出如下：
{'action': 'DELETE FROM orders WHERE id=1001', 'risk': '高危：删除订单，不可撤销'}
解读：invoke 既没报错也没跑完——它带着 __interrupt__ 正常返回，图停在 approve 中途，状态已存进 checkpointer。r1["__interrupt__"] 是列表（并行分支可能同时断多处），每项的 .value 是申请单，.id 是这条中断的编号。接着恢复，批准与驳回各走一遍：
r2 = graph.invoke(Command(resume=True), config) # 批准
print(r2["result"])
config2 = {"configurable": {"thread_id": "t-002"}} # 另开线程走驳回
graph.invoke({}, config2) # 再次断在 approve
r3 = graph.invoke(Command(resume=False), config2) # 驳回
print(r3["result"])
输出如下：
已执行：DELETE FROM orders WHERE id=1001
已取消：人工驳回
解读：恢复就是再调一次 invoke，输入从普通消息换成 Command(resume=True)——True 流回 approve 里 interrupt() 的返回值，节点重跑后走到 END。驳回那条线另开了 thread_id：t-001 已经跑到终态，没有挂起的中断可恢复；真实系统里一张审批单对应一个线程。两次调用、一次暂停，审批闭环完整落地——Web 服务里这两次 invoke 往往隔着几小时、来自两台机器，靠 thread_id 加持久化 checkpointer 串起来。
4、恢复值的两种形态
第 3 节的 resume 传的是布尔，够用但浪费了通道——resume 值可以是任意 JSON，审批接口的契约由你设计。第一种形态是简单决策：布尔或 "approve"/"reject" 字符串，节点据此分支，适合"放不放行"的二元问题。第二种形态是结构化改写：人工的决定、修改后的内容一起传回去，审批从"过或不过"升级为"改完再过"。内容审核场景最直观——Agent 起草邮件，人工改写定稿：
class DraftState(TypedDict):
draft: str # Agent 起草的邮件
final: str # 定稿
def review(state: DraftState) -> dict:
# 草稿全文作为申请单，人工的返回值直接定稿
decision = interrupt({"draft": state["draft"]})
if decision["approved"]:
return {"final": decision["text"]} # 人工改写后的版本
return {"final": "已取消发送"}
graph.invoke(Command(resume={ # 审批卡片的提交结果
"approved": True,
"text": "尊敬的客户：您的订单已于今日发出，预计 3 个工作日内送达。",
}), config)
解读：图结构换成 draft 节点接 review 节点即可，与第 3 节同构。申请单带草稿全文，人工在卡片上改两笔，resume 回一份 dict——review 拿到的是人工版本，直接定稿。改写的粒度自由发挥：改参数、换收件人、重写正文，都是往这份 dict 里加字段。langgraph 1.2.12 起 interrupt() 还接受 response_schema 参数（Pydantic 模型），人工提交的值先过校验再返回——格式错抛 ValidationError，原线程重答即可，审批接口从此有了类型契约。
这两种形态会贯穿后面所有模式：第 6 节官方中间件的四种决策，本质就是把结构化 resume 标准化了。
5、节点级断点对比
第 10 篇 create_agent 签名里的 interrupt_before 是另一族中断：编译期声明的节点级断点，还有个孪生兄弟 interrupt_after。把第 3 节的图改一下——approve 换成不含 interrupt() 的 execute 节点（builder2 同理组装成 plan → execute），compile 时点名断点：
def execute(state: State) -> dict: # 不含 interrupt() 的普通节点
return {"result": f"已执行：{state['action']}"}
graph2 = builder2.compile(
checkpointer=InMemorySaver(),
interrupt_before=["execute"], # 进入 execute 前断住
)
config = {"configurable": {"thread_id": "t-101"}}
r1 = graph2.invoke({}, config)
print(graph2.get_state(config).next) # 断点位置
r2 = graph2.invoke(None, config) # 恢复：不带人工载荷
print(r2["result"])
解读：进入 execute 前图暂停，get_state(config).next 打印 ('execute',)——断点停在哪个节点一目了然；恢复传 None，图继续执行到 END。它与 interrupt() 的差异一张表说清：
| 维度 | interrupt() | interrupt_before / after |
|---|---|---|
| 声明位置 | 节点函数体内 | compile 参数，也可 invoke 时传 |
| 申请单 | 任意 JSON payload | 无，前端不知道为何而停 |
| 人工输入 | resume 值流回节点 | 无通道，恢复只传 None |
| 节点执行 | 恢复时从头重跑 | 断点前或后，不重复执行 |
| 官方定位 | HITL 主路径 | 调试、单步、检查状态 |
官方文档的定位很直白：节点级断点是调试工具，人机协同场景一律推荐 interrupt()——差别就在"人的决定能不能传进图"。
补一句公道话：interrupt_before 并非无用。审批已由外部工单系统完成、图只负责执行的场景，或者想在暂停时用 get_state 检查甚至改一把状态再放行，节点级断点反而顺手。选型口诀：要传人工决定用 interrupt()，只要停一下用 interrupt_before。
6、工具审批实战
6.1、工具节点内嵌审批
真实 Agent 的危险动作几乎都发生在工具执行时——ReAct 循环里模型决定调用，工具节点负责执行，闸门就装在执行前。把第 9 篇手写的 call_tools 升级（AgentState 与 tools_by_name 沿用原定义，compile 记得挂 checkpointer）：
from langchain_core.messages import ToolMessage
DANGEROUS = {"send_email", "delete_rows"} # 危险工具名单
def call_tools(state: AgentState) -> dict:
results = []
for tc in state["messages"][-1].tool_calls:
if tc["name"] in DANGEROUS:
decision = interrupt({ # 申请单：工具名 + 参数
"tool": tc["name"],
"args": tc["args"],
})
if not decision["approved"]: # 驳回：理由回喂模型
results.append(ToolMessage(
content=decision.get("message", "人工驳回本次调用"),
tool_call_id=tc["id"]))
continue
tc = {**tc, "args": decision["args"]} # 批准且可能改过参数
output = tools_by_name[tc["name"]].invoke(tc["args"])
results.append(ToolMessage(content=str(output),
tool_call_id=tc["id"]))
return {"messages": results}
解读：名单外的工具原路直通；命中的先把工具名与参数打包成申请单 interrupt 出去。批准后可能拿到人工改过的参数——decision["args"] 覆盖原值再执行，第 4 节的改写形态落了地；驳回则把理由包成 ToolMessage 回喂，模型读到"人工驳回"通常会调整方案或放弃，循环不崩——正是第 9 篇"错误回传、模型自愈"思路的审批版。
两个工程提醒。其一，同一节点内多次 interrupt() 按出现顺序与恢复值配对：本例遍历的是 checkpoint 里固定的 tool_calls，首跑与重跑序列一致，配对才成立——"哪些调用会中断"绝不能依赖易变的外部条件，否则重跑时序列错位、决定张冠李戴。其二，并行分支同时中断时，resume 可以传 {中断 id: 决定} 的字典一次作答，中断对象的 .id 就是为这个场景准备的。
6.2、批量审批数据结构
上面是手搓版。官方 HumanInTheLoopMiddleware（第 10 篇提过）把这套模式标准化，create_agent 挂上即用：
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
agent = create_agent(
model, # 沿用 init_chat_model 实例
tools=[send_email, execute_sql, read_data],
checkpointer=InMemorySaver(), # 中断状态必须持久化
middleware=[HumanInTheLoopMiddleware(
interrupt_on={ # 审批策略表
"send_email": True, # 四种决策全放开
"execute_sql": {"allowed_decisions": ["approve", "reject"]},
"read_data": False, # 只读工具直接放行
},
)],
)
解读：interrupt_on 是策略表——True 放开全部决策，dict 限定可选决策，False 放行；langchain 1.3.3 起还支持 when 条件函数，比如 SQL 以 SELECT 开头就不打断。默认的批量模式把一轮所有待审调用攒成一次中断——payload 里的 action_requests 列表逐项带工具名、参数、描述，review_configs 告诉前端每个动作允许哪些决策。审批 UI 的数据结构照着这两份清单设计即可：一张卡一个动作，按 allowed_decisions 渲染按钮。人工作答时按顺序回一张决定列表：
agent.invoke(Command(resume={ # 与 action_requests 顺序对应
"decisions": [
{"type": "approve"},
{"type": "edit", "edited_action": {
"name": "send_email",
"args": {"to": "ops@example.com"}}},
{"type": "reject", "message": "先归档再删除，不要重试"},
],
}), config=config, version="v2")
四种决策的语义是固定的：
| 决策 | 行为 | 适用 |
|---|---|---|
approve |
原样执行 | 认可动作与参数 |
edit |
按改写后的工具与参数执行 | 人工修正 Agent 产出 |
reject |
不执行，message 回喂模型 | 否决并给理由 |
respond |
不执行，message 直接代答 | 人工替工具给结果 |
version="v2" 是 create_agent 的新版调用形态，挂起的申请单直接在结果的 .interrupts 属性里取；嫌批量粒度粗可设 interrupt_mode="per_call"（langchain 1.4.6 起），一个调用一次中断、按中断 id 作答。前后端打通的完整链路：后端轮询挂起线程——get_state(config).next 非空即有断点，snapshot.tasks 里的 interrupts 逐个取申请单渲染成卡片；人工提交后 Command(resume=...) 一次送回。细节见 HITL 中间件文档与 HumanInTheLoopMiddleware API 参考。
7、总结
本篇把 HITL 三件套配齐：interrupt() 在节点内任何位置挂起图，申请单随 __interrupt__ 交给前端；checkpointer 按 thread_id 存下断点状态，等待期时长不限；Command(resume=...) 把人工的决定送回 interrupt() 的返回值，节点从头重跑继续执行。恢复值从布尔到结构化 dict 都是自由设计空间，1.2.12 起还能用 response_schema 上类型契约；节点级断点留给调试，人工决定要进图就用 interrupt()；工具审批的手搓版与中间件标准版，你手里都有了。
留一个练习，回到第 9 篇的手写 ReAct：
- 加一件
send_email工具，并维护 DANGEROUS 危险名单 - 按 6.1 节升级工具节点：危险工具 interrupt 出申请单，驳回理由回喂模型
- 验收：提问"给 alice@example.com 发一封会议通知"，断点处看到申请单；批准、驳回各跑一次，驳回后模型应改为文字回复或放弃
- 进阶：给决定定义 Pydantic 模型，用
response_schema上校验
对于准备上生产的系统，建议从 6.2 节的中间件起步，interrupt_on 策略表足以覆盖多数审批需求，手搓 interrupt() 留给非标准拓扑。下一篇深入 checkpointer——本篇只用了它的内存版，SQLite 与 Postgres 后端、时间旅行、按线程回放，是持久化的完整版图。
