第 14 章 中断与人在回路

本章目标：掌握 interrupt——图运行到关键节点主动暂停，等人来决策，再带着决策恢复执行。这是生产级 Agent 的分水岭能力：全自动与全自动 supervised 之间的区别，就在这一章。

14.1 有些决定不该交给模型

评估系统运转顺利，直到 HR 总监提出一条硬性合规要求：

候选人综合分超过 85 的，发 offer 前必须有人工确认环节；年薪期望超过预算 30% 的，必须 HR 主管亲自放行。

这类要求有三个共同特征，决定了它们不能靠 prompt 解决：

1. 责任归属在人。模型可以建议，签字的必须是人。
2. 决策依据在系统外。预算、编制、政策——这些信息不在简历里，模型看不见。
3. 决策时机不确定。什么时候需要人介入，取决于运行时的状态。

朴素的做法是在图外面写轮询：跑一段、停下、问人、再跑一段。但你很快发现要自己造一堆轮子：怎么标记"停在等人"、暂停多久不超时、恢复时怎么把人的决定送回图里、进程重启后怎么续上。

LangGraph 把这套东西做成了原生能力：interrupt 函数。它在节点内部调用，效果是图在此处冻结、invoke 返回；等外部带着答案再次调用时，从冻结点精确复活。这种"人机协作"模式有个通用名——HITL（Human-In-The-Loop，人在回路）。

14.2 interrupt 的基本用法

interrupt 的 API 设计出乎意料地简单——节点函数里调用 interrupt(问题)，下一行直接拿到人的回答：

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

class ScreenState(TypedDict):
    resume_text: str
    score: Annotated[int, operator.add]
    final_decision: str

def evaluate(state: ScreenState) -> dict:
    return {"score": 88}

def confirm(state: ScreenState) -> dict:
    # 关键三行：
    answer = interrupt({                       # 1. 发起中断，附带要人看的信息
        "question": "综合分 88，超过 85，是否通过该候选人？",
        "score": state["score"],
    })
    # 3. 恢复后，answer 就是人类提交的决策
    decision = "通过" if answer == "approve" else "驳回"
    return {"final_decision": f"{decision}（人工确认）"}

def notify(state: ScreenState) -> dict:
    return {"final_decision": state["final_decision"] + "，已归档"}

builder = StateGraph(ScreenState)
builder.add_node("evaluate", evaluate)
builder.add_node("confirm", confirm)
builder.add_node("notify", notify)
builder.add_edge(START, "evaluate")
builder.add_edge("evaluate", "confirm")
builder.add_edge("confirm", "notify")
builder.add_edge("notify", END)

# interrupt 依赖持久化：中断状态存在检查点里
memory = InMemorySaver()
graph = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "hitl-001"}}

# ---- 第一轮：跑到 confirm 处中断 ----
result = graph.invoke({"resume_text": "李思远 ..."}, config)
print(result)    # 只包含 evaluate 的产出：{'resume_text': ..., 'score': 88}

# ---- 查看中断详情 ----
snap = graph.get_state(config)
print(snap.next)                 # ('confirm',) —— 停在 confirm 节点
print(snap.tasks[0].interrupts[0].value)
# {'question': '综合分 88，超过 85，是否通过该候选人？', 'score': 88}

# ---- 模拟人类决策，恢复执行 ----
result = graph.invoke(
    Command(resume="approve"),    # 2. 人类的答案通过 resume 送回
    config,                        # 同一个 thread_id！
)
print(result["final_decision"])
# 通过（人工确认），已归档

三个 API 的分工：

• interrupt(payload)：在节点内调用。payload 是"给人看的问题"，会出现在中断信息里。调用它意味着本节点的本次执行作废——图冻结在此节点开始前。
• Command(resume=answer)：恢复调用。answer 会成为当初那行 interrupt() 的返回值，节点从头重新执行，执行到 interrupt() 那行时直接拿到 answer，继续往下走。
• get_state 的中断信息：snap.next 告诉你停在哪个节点，snap.tasks 里的 interrupts 数组携带当初的 payload——前端靠它渲染"待审核"界面。

最微妙也最优雅的一点值得强调：恢复时节点从头重跑，但 interrupt 那行不再真正中断。这意味着 interrupt 之前的节点逻辑会执行两遍——节点里如果有副作用（发通知、写库），必须容忍重入。这是使用 interrupt 的头号纪律。

14.3 恢复的两种答案形态

Command(resume=...) 送回的 answer 可以是任意可序列化值，两种常见形态：

简单值："approve"、True、一个数字。适合二选一、简单参数。

结构化值 + 分支处理：

def confirm(state: ScreenState) -> Command:
    answer = interrupt({"question": "请审核", "score": state["score"]})

    if answer["action"] == "approve":
        return Command(goto="notify")
    elif answer["action"] == "revise":
        # 人要求改条件重评：把人的批注带回评估
        return Command(update={"score": answer["adjusted_score"]},
                       goto="evaluate")
    return Command(goto=END)

人工决策直接驱动流向——interrupt 与第 6 章的 Command 在此会师：人的回答就是节点的运行时输入，goto 可以按它动态计算。HITL 不是特殊机制，只是"节点输入来自人类"的普通执行。

14.4 实战形态：ReAct 循环中的工具审批

更高阶的 HITL 场景：不是"节点前等人"，而是"工具调用前等人"。Agent 决定调用 send_offer 工具（发 offer 邮件），这类不可逆动作必须有人点头。

实现思路：自定义工具内部先 interrupt，把工具调用本身变成审批点：

from langchain_core.tools import tool
from langgraph.types import interrupt

@tool
def send_offer(candidate: str, salary: int) -> str:
    """向候选人发送 offer 邮件。

    Args:
        candidate: 候选人姓名
        salary: 年薪（万元）
    """
    # 不可逆动作：先问人
    decision = interrupt({
        "action": "send_offer",
        "candidate": candidate,
        "salary": salary,
        "note": "此操作将发送正式邮件，请确认",
    })
    if decision == "approve":
        return f"已向 {candidate} 发出 offer（年薪 {salary} 万）"
    return "操作被拒绝，未发送邮件"

这个工具放进第 8 章的 ReAct 图后，行为变成：模型决定调 send_offer → ToolNode 执行该工具 → 工具内部 interrupt → 整个图冻结 → 人审批 → Command(resume=...) 恢复 → 工具返回结果 → 模型继续推理。

审批粒度从"节点级"细化到了"动作级"，而机制零新增——interrupt 在工具函数体内，与在节点函数体内是同一回事。生产上常见的进一步封装是"白名单工具免审、敏感工具必审"：工具注册表里打上 requires_approval 标记，审批逻辑写在统一的基类/装饰器里，业务工具保持干净。

14.5 中断与超时：挂起的治理

图挂起等人，人可能永远不来。工程上需要治理挂起：

• 超时策略：外层调度（你的服务）给挂起线程设 TTL，超时后用 Command(resume=默认答案) 自动兜底——比如"48 小时未审批视为驳回"。引擎本身不管超时，这是你服务层的职责。
• 通知触达：中断发生时（invoke 返回后）把审批任务推给审批人（IM、邮件工单），别指望人主动轮询。
• 审批界面：用 get_state 读出 interrupts payload 渲染表单，提交时调 invoke(Command(resume=...))。LangGraph 生态的 Agent Server / Studio 提供了现成的托管界面，但协议就是本章这两端——自己包一个 Web 审批页同样简单。

14.6 中断的本质（收束到第 12 章）

把机制拆到最底层，interrupt 做的事朴素得很：

1. 中断时：抛出一个特殊信号，引擎捕获后把"当前节点待执行、携带 interrupt payload"的状态写入检查点，然后让 invoke 正常返回。
2. 恢复时：Command(resume=answer) 触发从检查点恢复，引擎重放该节点的执行，将 answer 注入 interrupt 调用点。

所以：没有 checkpointer 就没有 interrupt（这也是 14.2 demo 里 compile 必须挂 saver 的原因）；中断状态和第 12 章的会话状态是同一套存储；恢复和第 13 章的"从检查点续跑"是同一族操作。四大能力——会话恢复、时间旅行、中断、（下一章的）长期记忆——全部长在持久化这一棵树上。这就是为什么第 12 章被称为"分水岭"。

常见坑与排错

1. interrupt 前的副作用执行两遍。14.2 反复强调的恢复语义：节点从头重跑。节点里 interrupt 之前如果发了通知邮件，恢复时会再发一封。纪律：interrupt 尽量放在节点第一行；有副作用的动作放在 interrupt 之后。

2. 恢复时用了新 thread_id。中断状态存在原线程的检查点里，换 thread_id 等于开了全新会话——图从头跑，再次中断在同一处。恢复必须用与中断时相同的 config。

3. 忘记挂 checkpointer。没持久化时 interrupt 会直接报错（无处存中断状态）。看到 interrupt requires checkpointer 类错误，回去给 compile 加 saver。

4. 把 resume 的 answer 类型搞错。中断时 payload 里约定的格式（比如期望 "approve" 字符串），恢复时送了 {"action": "approve"}——节点里 answer == "approve" 判断失败，走进意外分支。前后端约定 answer 的 schema 并做校验，这是系统边界，值得校验。

5. 并发恢复同一中断。两个管理员同时点了"通过"和"驳回"，两次 invoke(Command(resume=...)) 打到同一 thread——同线程并发问题（12.3），第二个调用可能报状态冲突或产生怪异结果。审批入口要加分布式锁或幂等令牌。

6. 在 interrupt payload 里塞大对象。payload 会进检查点、进审批界面。塞整份简历原文不如塞 ID + 摘要，审批页按需拉取详情。

7. interrupt 里写复杂分支逻辑。interrupt 调用本身应当"薄"——发起、拿答案、返回。审批流程的编排（多级审批、会签）放在图结构里（多个 confirm 节点 + 条件边），不要塞进一个巨大的 confirm 节点。

动手练习

1. 扩展 14.2 的 demo：score ≤ 85 时跳过 confirm 直接 notify（提示：evaluate 后用条件边，或 confirm 节点内部先判断再决定是否 interrupt）。验证两条路径：需要审批的与免审直通的。

2. 给第 8 章的 ReAct 评估图加上 send_offer 审批工具（14.4 的实现）。跑一次完整流程：模型调工具 → 中断 → 人工批准 → 邮件"发送" → 模型收尾。再跑一次拒绝路径，观察模型对"操作被拒绝"的反应。

3. 实现两级审批：综合分 ≥ 90 需要 HR 总监 + 业务负责人先后确认（两个 confirm 节点串联）。思考：用两个节点串联，还是在一个节点里 interrupt 两次？各有什么坑？（提示：坑 1 的副作用重入。）

本章速查