人工审批 Human-in-the-loop

让智能体自动执行一切并不总是安全的：转账、删数据、发邮件这类"不可逆或高影响"动作，最好在真正执行前停下来请人确认。Human-in-the-loop（人在回路）就是让流程在某节点挂起、等人批准或拒绝后再继续。第 18 章的 ReAct 图已经具备自动循环能力，本章给它加一道"审批闸门"。

为什么需要：危险操作不能全自动

一个财务智能体的理想行为：先规划出"转账 1000 元给张三"，但在调用支付接口前停下来，把动作展示给用户确认。手动实现要自己管理"挂起状态、恢复入口、会话绑定"，容易出错；LangGraph 原生提供 interrupt() 与 Command，把挂起/恢复变成图的一部分。

两个关键 API

interrupt(内容)：在节点内调用即"挂起"——执行现场保存到 checkpointer，invoke 返回后不再往下走；程序下次用 Command 恢复时，interrupt() 的返回值就是人给出的决定。
Command(resume=值)：从外部恢复执行，值会作为 interrupt() 的返回继续跑剩余流程。注意：挂起需要能保存执行现场，因此必须给图配 checkpointer（MemorySaver 即可，详见第 20 章）。interrupt 与 Command 的 API 随版本演进，精确用法以官方文档为准。

完整示例：转账审批

import os
from typing import Annotated, TypedDict
from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.types import Command, interrupt

class State(TypedDict):
    messages: Annotated[list, add_messages]
    pending: dict          # 待审批的动作
    approved: bool

def propose(state):
    # 演示：从固定句式解析动作；实际项目可交给模型抽取（第 4 章结构化输出）
    text = state["messages"][-1].content
    _, action, amount, to = text.split()
    return {"pending": {"action": action, "amount": int(amount), "to": to}}

def human_gate(state):
    decision = interrupt({
        "type": "approval",
        "message": f"是否批准：{state['pending']['action']} "
                   f"{state['pending']['amount']} 元给 {state['pending']['to']}？",
    })
    return {"approved": decision == "approve"}

def do_pay(state):
    p = state["pending"]
    print(f"执行：已{p['action']}{p['amount']}元给{p['to']}（此处应调用真实接口）")
    return {}

def reject(state):
    print("已拒绝该操作，流程终止")
    return {}

builder = StateGraph(State)
builder.add_node("propose", propose)
builder.add_node("human_gate", human_gate)
builder.add_node("do_pay", do_pay)
builder.add_node("reject", reject)
builder.add_edge(START, "propose")
builder.add_edge("propose", "human_gate")
builder.add_conditional_edges(
    "human_gate",
    lambda s: "do_pay" if s["approved"] else "reject",
    {"do_pay": "do_pay", "reject": "reject"},
)
builder.add_edge("do_pay", END)
builder.add_edge("reject", END)

graph = builder.compile(checkpointer=MemorySaver())
config = {"configurable": {"thread_id": "txn-001"}}

第一次运行：挂起等人审批

out = graph.invoke(
    {"messages": [HumanMessage("执行 转账 1000 张三")]}, config)
print(out["__interrupt__"][0].value)
# 输出：{'type': 'approval', 'message': '是否批准：转账 1000 元给 张三？'}

invoke 没有一路跑到 END，而是把审批请求放进返回的 interrupt 字段——在桌面脚本里，你可以据此弹出对话框或直接 input()；在 Web 服务里则把该内容发给前端渲染审批按钮。

恢复：批准或拒绝

# 场景一：批准 -> 继续执行
graph.invoke(Command(resume="approve"), config)
# 输出：执行：已转账1000元给张三（do_pay 节点里的 print）

# 场景二：拒绝 -> 走 reject 分支（换一个新 thread 演示）
config2 = {"configurable": {"thread_id": "txn-002"}}
graph.invoke({"messages": [HumanMessage("执行 转账 1000 张三")]}, config2)
graph.invoke(Command(resume="reject"), config2)
# 输出：已拒绝该操作，流程终止

批准分支走向 do_pay（真实项目里调用支付 API），拒绝分支走向 reject，两条路径互不干扰。因为整个执行现场都在 checkpointer 里，人思考多久都行：下次用同一 thread_id 发 Command(resume=...) 即可无缝继续。

桌面与服务器差异

桌面/脚本：挂起后可以立即阻塞等用户输入，再同步 resume，代码最直白。
Web 服务：invoke 与 resume 往往跨 HTTP 请求，前端收到 interrupt 展示确认界面，用户点按钮后后端再发 Command(resume=...)，靠 thread_id 关联同一会话。不同部署形态与框架版本的推荐写法以 LangGraph 官方文档为准，核心思路一致：interrupt 处挂起，resume 处继续。

多道闸门：一次流程多次审批

interrupt 可以放在多个节点上，流程会按顺序多次挂起，适合"先审方案、再审金额"的分步审批：

def review_plan(state):
    decision = interrupt({"type": "review_plan", "plan": state["plan"]})
    return {"plan_ok": decision == "approve"}
# 后面的节点里可再放一次 interrupt({"type": "review_money", ...})

每道闸门挂起都保存现场，resume 一次往前走一步，直到下一道闸门或 END。

常见问题

resume 只能传字符串吗？interrupt() 的返回值由你传入 Command(resume=...) 的内容决定，字符串、字典、数字都可以；想带审批备注就传 {"decision": "approve", "note": "同意"}。
一直没人审批怎么办？检查点把现场保存着，进程重启、几天后再用同一 thread_id resume 都行；也可以另起定时任务，超时后用第 20 章的 update_state 把流程改走拒绝分支。
interrupt 的内容要可序列化吗？要——它会随状态写进 checkpointer，传 JSON 友好的值最稳妥。
忘记配 checkpointer 会怎样？挂起现场无处保存，调用会报错；interrupt 必须配合 compile(checkpointer=...) 使用。

小结：人工审批 = 在危险操作前放一个 interrupt() 节点，流程挂起并返回审批请求；外部用 Command(resume=...) 传入批准/拒绝，条件边据此分流执行或终止——把"让 AI 干活前先问人"变成图上一个可复用的闸门。

笔记加载中…

标签：#Agent实战

返回知识库