4.4 人机协作 (Human-in-the-Loop) 与状态干预

4.4 人机协作 (Human-in-the-Loop) 与状态干预

让人在关键处刹车

本节在全册位置：人机协作是 LangGraph 的招牌。interrupt 在函数节点里挂起整个图，把当前状态交给人；人审核后用 Command(resume=...) 带着修正结果继续。状态全程保留，中断点精确。我们认为这是让智能体「敢上线」的关键能力。

机制：interrupt 与 resume

在节点中调用 interrupt(值) 会暂停执行并把「待审核内容」返回给调用方。外部拿到后决定批准或改写，再用同一 thread_id、config 里加 thread_id 配合 invoke/Command(resume=...)。这比「重新跑一遍」安全：已完成的步骤不重做。类比游戏：interrupt 像存档点，玩家读完档再继续，不重打已过的关。中断点设计在哪，决定了人工能看到什么。

from langgraph.types import interrupt, Command
from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class S(TypedDict):
    draft: str

def review(s: S):
    # 挂起，等待人工确认草稿
    decision = interrupt({"need_review": s["draft"]})
    if decision == "reject":
        return {"draft": s["draft"] + "[已退回]"}
    return {"draft": s["draft"]}

b = StateGraph(S)
b.add_node("review", review)
b.add_edge(START, "review")
b.add_edge("review", END)
g = b.compile(checkpointer=MemorySaver())
cfg = {"configurable": {"thread_id": "h1"}}
print(g.invoke({"draft": "发布此文?"}, cfg))   # 返回 interrupt 值
print(g.invoke(Command(resume="ok"), cfg))        # 带审核结果继续
​

#  4.4 人机协作 (Human-in-the-Loop) 与状态干预
​

案例：合同发送前确认

背景：自动邮件在发送前必须人工过目，错了就是对外事故。

操作：发送节点前插 review 节点，interrupt 把邮件内容交人。

结果：人点确认才发送，点退回则改写重走，绝不静默发出。

解读：中断点精确，已生成内容不浪费，合规与体验兼得，这是图式编排的独占优势。

变式：多道审核可串多个 interrupt，形成审批链，每道都可独立 resume。

工程清单

• interrupt 在节点里挂起整图，把当前状态交人审核，resume 带结果继续，已完成步不重做。

• 中断点要精确落在写操作之前，人工能看到且仅决定去留，已生成内容不浪费。

• 多道审核可串多个 interrupt，形成审批链，每道都可独立 resume，合规更稳。

常见误区

把人工确认写在节点外部轮询，已生成内容被重复执行；interrupt 才是框架内原语，它能精确停在节点边界，比重新跑一遍安全得多。

interrupt 的运行时语义

interrupt 不是普通函数调用，它通过检查点机制挂起整张图，细节决定了你该怎么用它：

from langgraph.types import interrupt

def review(s):
    # 第一次执行到这里：整图挂起，值进入检查点的中断记录
    decision = interrupt({"draft": s["draft"], "need": "确认"})
    # 第二次 resume 进来：decision 就是外部传入的值
    return {"draft": s["draft"] + ("[已改]" if decision == "改" else "")}
​

执行规律是：interrupt 之前的状态已被检查点保存；resume 后节点从头重新执行，但 interrupt 调用会直接返回外部值，不重复执行节点内前面的代码。所以不要在 interrupt 之后依赖「只执行一次」的副作用——如果节点内 interrupt 前面有外部调用，resume 时会再执行一次。要精确控制，就把外部调用放在别的节点，或把「是否已执行」记录在状态里。

多道审核与审批链

复杂流程需要多道人工把关时，在多个节点里分别 interrupt 即可，形成审批链，每道可独立 resume：

def draft_check(s):
    if s["draft"] == "草稿":
        interrupt({"stage": "draft", "draft": s["draft"]})
    return {}

def send_check(s):
    interrupt({"stage": "send", "mail": s["mail"]})
    return {}
​

两道 interrupt 的语义：第一次 resume 只推进到第二个中断点，不会直接跑完；每道审核都可以看到前一道的产物。审批链的好处是每个节点只暴露该阶段该看的字段，避免把内部细节一次全抛给人。设计时遵循「最小暴露」：人只需要看到做决策所需的信息，其余字段留给自己调试。

人工未响应与超时策略

人工确认最怕「人等不到、流程卡死」。interrupt 挂起天然是持久化的，进程重启后仍在等待，这是优点；但业务上需要超时和兜底，常见做法有三种：

# 超时兜底：外部定时器到期后，以"拒绝"继续
g.invoke(Command(resume="reject_timeout"), cfg)
# 节点里把超时值当作普通决定处理
​

设计时给每个中断点定义「允许的 resume 值集合」，人传了非法值也能安全回落。超时和降级策略要在文档里写明，不然排查「为什么流程自己走了默认分支」时又是一场事故。