第 15 章 Human-in-the-Loop：interrupt 原语

Human-in-the-Loop（HITL，人在环路）指在 Agent 自动执行流程中设置受控暂停点，将关键决策环节交给人类确认、补充或修正后再继续执行。该机制用于五类核心业务场景：高风险操作审批（银行转账、批量发邮件、生产数据删除、代码上线）必须经人工二次确认；模型置信度低时（分类概率低于阈值、检索结果歧义、工具返回不确定）由人类补充信息或纠正方向；人工纠错（修改 Agent 生成的文案、SQL、代码草稿再发布）；多轮表单填写（向导式收集姓名/手机号/地址等多字段，用户可以逐项输入、随时修改）；分级自主权（Tiered Autonomy）按风险等级自动通过（低风险）、通知人类（中风险）、或强制暂停等待显式审批（高风险）。LangGraph 1.2.x 提供三种层级由浅入深的 HITL 机制，其中 interrupt() 原语是最灵活、最核心的能力，配合 checkpointer 可实现从秒级弹窗审批到数天级异步工单审批的统一抽象。

15.1 三种 HITL 机制对比

机制	触发时机	触发条件	是否需要 checkpointer	可携带 payload	典型场景
interrupt_before/after（编译期/调用期断点）	指定节点执行前/后	无条件，到达该节点即暂停	是	否（只能通过 state 查看）	工具调用前强制审批、回答发出前内容审核
interrupt()（节点内动态断点）	节点函数内执行到该语句时	可按业务条件（if 分支）触发	是	是（任意可序列化对象）	低置信度审批、草稿确认编辑
节点内人工数据收集（多次 interrupt）	节点内顺序调用 interrupt 语句	按代码顺序依次触发	是	是（每次独立 payload）	多步表单、向导式信息收集

interrupt_before/after 属于静态断点：节点名在 compile() 参数或 stream()/invoke() 调用参数中固定，运行到该节点位置必然暂停，不依赖任何运行时数据。该机制适合位置固定、业务上每次都必经的审批点（例如"所有工具调用前都要审核""最终回复发出前必须审核"）。interrupt() 属于动态断点：直接写在节点函数内部，可以包裹在 if score < 0.7: 之类的业务条件里，只在满足条件时才触发；并且可以把任意可序列化对象（draft 文本、待审批金额、候选选项、风险评估明细）作为 payload 透出给调用方，人类基于 payload 内容做决策。节点内人工数据收集本质是同一节点内多次调用 interrupt()，每次 interrupt 暂停等待一份 resume 值，多份值按调用顺序匹配，适合同一个交互节点内的多轮对话式信息采集（向导表单）。

15.2 interrupt() 原语的行为模型与恢复机制

interrupt() 定义在 langgraph.types 模块，签名为 interrupt(value: Any) -> Any。其完整行为时序如下：

1. 节点函数执行到 human_input = interrupt(value) 语句时，LangGraph 抛出特殊内部异常 GraphInterrupt，该异常被 Pregel 运行时捕获（不会冒泡到用户业务代码）；
2. 运行时立即调用 checkpointer 的 put_writes() 方法，把当前 Channel 快照、interrupt 写入记录、以及传入的 value（即 payload）一并持久化到 checkpointer；
3. 当前 invoke/stream 调用此时产出中断事件后正常返回（不是抛异常），调用方通过 graph.get_state(config) 读取中断 payload——payload 存放在 state.tasks[0].interrupts[0].value（多 interrupt 时按列表索引对应）；
4. 人类在任意时间（毫秒到数天乃至数周）完成审批/输入，调用方再次执行 graph.invoke(Command(resume=human_value), config={"configurable":{"thread_id": tid}})；
5. 运行时从 checkpointer 加载中断时快照，从该节点的头部重新执行节点函数（不是从 interrupt 调用点继续），此时已经匹配 resume 值的 interrupt(value) 调用直接返回之前 resume 的 human_value，节点函数从头部顺次执行，直到下一 interrupt 或节点返回。

节点从头部重新执行是最容易被误解、也最容易引发生产 Bug 的行为。根因在于 LangGraph 的执行模型以节点为原子单元，checkpoint 粒度是超步而非源代码行，resume 时无法在语句级恢复，必须重放节点函数。副作用代码（发邮件、写数据库、调用外部支付 API）如果放在 interrupt() 调用之前，会在首次触发 interrupt 时执行一次，resume 后重跑节点时又执行一次，导致重复发件、重复扣款、重复写库。因此必须保证所有副作用幂等，或把副作用放置在 interrupt() 调用之后（即"先收集全部人类输入，最后一次性执行副作用"）。

多次 interrupt的行为。同一节点可顺序调用多次 interrupt()：

def form_node(s):
    name = interrupt({"q": "请输入姓名"})
    phone = interrupt({"q": "请输入手机号", "name": name})
    addr = interrupt({"q": "请输入地址", "name": name, "phone": phone})
    # 所有信息收集完成后执行副作用
    save_to_db(name, phone, addr)
    return {"done": True}

每次调用 interrupt 都会单独落盘一个中断事件。resume 有两种方式：(1) 逐次 resume——每次调用 graph.invoke(Command(resume=value), config=cfg) 匹配当前未匹配的第一个 interrupt，节点重跑时前 N-1 个 interrupt 直接返回历史值、第 N 个返回本次 resume 值；(2) 批量 resume——传入列表 Command(resume=[v_name, v_phone, v_addr])，一次匹配多个 interrupt（适合前端一次性提交整个表单后批量恢复）。

checkpointer 是强制前提。未配置 checkpointer 时调用 interrupt() 直接抛 NotImplementedError: Checkpointer is required for interrupt，因为无处保存中断快照与 resume 值的对应关系，恢复语义无法成立。调试期使用 MemorySaver（内存实现），生产使用 PostgresSaver / SqliteSaver / RedisSaver（跨进程持久化）。

暂停时长基于 checkpointer 持久化实现：运行时线程在 interrupt 发生时立即释放（不占用 CPU、不占内存、不保持数据库连接），所有状态都在 checkpointer 中。resume 可以是秒级交互（前端弹窗确认），也可以是小时级异步审批（工单系统、邮件审批链接），甚至是天后级的离线复核。跨进程/跨服务恢复只需使用同一个 checkpointer 实例（连接同一数据库）与同一个 thread_id，在任意服务进程中 invoke Command(resume=...) 即可从断点继续，原执行进程早已退出也不影响恢复。

恢复方式有两种场景：

• 同进程内恢复：执行图的进程未退出（如 Web 服务常驻进程），直接在同一进程中调用 graph.invoke(Command(resume=value), config={"configurable":{"thread_id":tid}})；
• 跨进程/跨服务恢复：审批服务与执行服务分离（如执行在 Worker 集群、审批在 Web 后台），两边共享同一个 PostgresSaver（同一数据库连接串），审批服务处理完人类操作后，在自己进程内调用 graph.invoke(Command(resume=value), config=...)，Worker 集群下次调度时就能从 checkpointer 中读取到 resume 值（或 Worker 不主动调度，由审批服务直接 resume 并继续后续执行）。

interrupt 与 update_state 的区别：interrupt() 是"暂停-续跑"语义，恢复时节点从头部重新执行，可能触发 LLM 重新调用、工具重新执行；update_state() 是"直接改状态"语义，不执行任何节点代码，只把传入 values 作为指定节点（默认 __start__ 节点）的写入合并到 State，适合人类直接覆盖某个字段（如手动把 status 改成 approved）而不需要节点继续跑的场景。人工纠错场景（修改 Agent 草稿再继续）应使用 interrupt+resume；纯数据修正场景（管理员手动标记状态）用 update_state。

时间旅行+审批。配置 checkpointer 后，graph.get_state_history(config) 返回该 thread_id 全部历史快照迭代器（按 checkpoint 时间倒序）。前端可展示完整执行轨迹（每个快照的 State、next、created_at、parent_config），调用方可在任意历史 snapshot 的 config 上调用 graph.invoke(..., config=historical_config) 实现"回到过去重做"（fork），fork 出的新分支在 checkpointer 中形成独立子树，原轨迹不被覆盖。Agent Inbox 模式基于此能力：将所有待审批中断统一抽象为"收件箱"条目，UI 展示 payload、提供 approve/edit/reject 三种操作，审批通过后统一 resume。langgraph-prebuilt 库提供 AgentInbox 协议参考实现，前端可以直接对接其 REST/WebSocket 接口。

15.3 完整示例

示例一：工具调用前人工审批（静态断点 interrupt_before）。

# langgraph==1.2.x
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.checkpoint.memory import MemorySaver
from langchain_core.messages import BaseMessage
from langchain_openai import ChatOpenAI

class S(TypedDict):
    messages: Annotated[list[BaseMessage], add_messages]

def agent(s: S) -> dict:
    resp = llm.bind_tools(tools).invoke(s["messages"])
    return {"messages": [resp]}

# 真实业务中填入转账/发件/删数据等高风险工具
tools = []
llm = ChatOpenAI(model="gpt-4o-mini")

g = StateGraph(S)
g.add_node("agent", agent).add_node("tools", ToolNode(tools, handle_tool_errors=True))
g.add_edge(START, "agent")
g.add_conditional_edges("agent", tools_condition, {"tools": "tools", END: END})
g.add_edge("tools", "agent")
graph = g.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["tools"],  # 工具调用前无条件暂停给人工审批
)

cfg = {"configurable": {"thread_id": "tx-001"}}
graph.invoke({"messages": [("user", "给张三的账户 6222xxxx 转账 10000 元")]}, config=cfg)
state = graph.get_state(cfg)
last_ai_msg = state.values["messages"][-1]
print("待审批工具调用：", last_ai_msg.tool_calls)
# 人类审核后调用 resume（None 表示不注入返回值，继续执行原节点）
graph.invoke(None, config=cfg)

示例二：动态审批（低置信度时 interrupt 展示草稿）。

from langgraph.types import interrupt, Command

def writer(s: S) -> dict:
    draft = llm.invoke(s["messages"]).content
    # 简化版置信度判断：draft 含"待定"或过短视为低置信度
    if "待定" in draft or len(draft) < 20:
        # interrupt payload 是结构化对象，前端按 type 渲染审批卡片
        decision = interrupt({
            "type": "draft_review",
            "draft": draft,
            "reason": "low_confidence",
        })
        if not decision["approved"]:
            draft = decision["edited_draft"]
    return {"messages": [{"role": "assistant", "content": draft}]}

# 审批侧：渲染草稿，提供 approve/edit/reject 三个按钮
graph.invoke({"messages": [("user", "写一段产品广告词")]}, config=cfg)
# 用户修改后提交
graph.invoke(Command(resume={"approved": False,
                             "edited_draft": "人工修改后的广告词文案…"}), config=cfg)

示例三：多步表单（多次 interrupt 收集字段）。

class FormState(TypedDict):
    done: bool
    name: str
    phone: str
    address: str

def form_node(s: FormState) -> dict:
    name = interrupt({"question": "请输入您的姓名", "fields_so_far": {}})
    phone = interrupt({"question": "请输入手机号", "fields_so_far": {"name": name}})
    address = interrupt({"question": "请输入收货地址",
                         "fields_so_far": {"name": name, "phone": phone}})
    # 所有信息收集完才执行副作用（写库、下单）
    # create_order(name, phone, address)
    return {"name": name, "phone": phone, "address": address, "done": True}

本章小结

HITL 是高可信、可审计 Agent 系统的必备能力。LangGraph 提供从静态断点（interrupt_before/after，固定位置无条件暂停）、动态断点（interrupt()，可条件触发、可携带 payload）到节点内多步数据收集（多次 interrupt）三种机制，统一依赖 checkpointer 持久化中断快照。interrupt() 首次调用时抛 GraphInterrupt 异常并落盘，调用方通过 get_state 读取 payload，通过 Command(resume=...) 恢复；节点从头部重跑时 interrupt() 直接返回 resume 值，副作用必须放在 interrupt 之后或保证幂等。暂停时长不占用运行时资源（从毫秒到数天），跨进程恢复通过共享 checkpointer+thread_id 实现。interrupt 是暂停-续跑语义（节点重跑），update_state 是直接改状态语义（不执行节点）。时间旅行通过 get_state_history 支持 fork 重做；Agent Inbox 模式将待审批项抽象为收件箱统一管理。

常见错误

1. 未配置 checkpointer 即调用 interrupt()。现象：NotImplementedError: Checkpointer is required for interrupt。原因：compile 时未传 checkpointer。修复：至少传入 MemorySaver 用于调试；生产用 PostgresSaver/SqliteSaver/RedisSaver；所有 interrupt 相关单元测试必须显式传 checkpointer。
2. resume 时使用新 thread_id 导致找不到断点。现象：GraphRecursionError 或 resume 后图从头开始执行。原因：thread_id 不一致，checkpointer 把它当作一个新的执行线程，找不到对应中断快照。修复：resume 必须严格复用触发 interrupt 时的 thread_id；前端保存 thread_id 到 URL/session；审批回调链接必须携带 thread_id。
3. 节点重跑导致副作用重复执行。现象：interrupt 前的工具/API 被调用两次——重复发邮件、重复扣款、重复写库。原因：resume 后节点从头部重跑，interrupt 前的副作用又执行一遍。修复：副作用代码严格放在最后一个 interrupt() 调用之后（即先收集完所有输入、最后一次执行副作用）；必须前置副作用时，工具函数必须幂等（用 request_id 去重、写库前先查唯一键）。
4. interrupt 后节点重跑导致 LLM 重复计费。现象：同一次审批触发两次 LLM 调用，token 成本翻倍。原因：LLM 调用放在 interrupt() 之前。修复：先 LLM 出草稿→再 interrupt 让人类审批→resume 后直接使用 resume 值（approved/edited_draft）不再调 LLM；或对 interrupt 前的 LLM 结果写入 State 缓存（resume 后判断草稿已存在则跳过再次调用）；compile 时开启 cache 参数缓存同参数 LLM 调用。
5. payload 中放入不可序列化对象。现象：checkpointer 写入时抛 TypeError: Object of type X is not JSON serializable 或 PicklingError。原因：payload 包含 LLM 实例、数据库连接、threading.Lock、生成器、lambda、打开的文件句柄等不可 pickle 对象。修复：payload 只放基础类型（str/int/float/bool/None）、dict、list、Pydantic BaseModel 实例；复杂对象序列化为 JSON 再放入 payload。

生产建议

• 分级自主权落地：工具元数据中声明 risk_level 字段（low/medium/high/critical），ToolNode 前统一路由节点判断：low 自动通过、medium interrupt 推送摘要通知人类（可异步 approve）、high/critical interrupt_before 阻塞等待显式 approve；
• interrupt payload 设计为结构化 Pydantic 模型（含 type、summary、risk_level、raw_data、options 候选操作），前端按 type 渲染不同审批卡片（转账审批、草稿编辑、信息补充），避免通用文本弹窗；
• 审批超时兜底：在 State 中维护 expires_at 时间戳，或在 checkpointer 存储层加 TTL 任务，超期未审批自动 reject 并进入降级分支（通知用户、转人工工单），避免线程长期悬挂占用资源；
• 跨服务审批：审批回调服务与执行服务共享同一个 PostgresSaver（同一数据库连接、同一表结构），回调中 invoke Command(resume=...)，执行服务不需要长连接或消息队列；
• Agent Inbox UI 组件化：统一展示待办列表（按 type/风险等级过滤）、payload 详情、approve/edit/reject 操作、历史轨迹回放（get_state_history），支持批量审批；
• 所有 resume 值写入审计日志（thread_id、审批人 user_id、审批时间、原始 payload、resume payload、耗时），金融/医疗等合规场景保留完整 checkpoint 快照用于事后审计；
• interrupt 前的 LLM/工具调用结果必须写入 State（不要只放局部变量），resume 后节点重跑时先判断对应字段是否已存在，已存在则跳过重复调用，既省 token 又避免重复副作用；
• 前端轮询+推送混合：interrupt 发生后前端切到审批 UI，通过 WebSocket/SSE 等待后端推送 resume 结果（或主动轮询 get_state 检测 next 字段变化），避免长轮询拖垮后端。