代码仓库ChainReaction

模型能调工具之后，第一个让人睡不踏实的问题就是它什么时候会自己动手。

查天气、算数、读文档，模型自己跑没问题。发邮件、删数据库记录、调支付接口，这类动作发出去就收不回来。Human-in-the-Loop 中间件处理的就是这一层：模型提议一个工具调用，真正执行之前先停下来，等人点个头。

这篇讲四件事。中间件怎么配，interrupt 怎么把执行挂起、状态落在哪、又怎么恢复，审批粒度能细到什么程度，以及和前端对接时待审批的卡片需要哪些字段。代码都在 DeepSeek 上真跑过，输出是终端里直接抄下来的，脚本在仓库的 HumanInTheLoop/ 目录下。

闸门装在哪一步

create_agent 跑起来是个循环：模型产出 AIMessage，里面可能带 tool_calls，工具节点执行，结果作为 ToolMessage 回到模型，再产出下一轮。HumanInTheLoopMiddleware 挂的是 after_model 钩子，位置在”模型已经想好要调什么”和”工具真的被执行”之间。

钩子里做的事情很直接。把这条 AIMessage 里的 tool_calls 逐个拿去过 interrupt_on 这张表，表里没有的工具直接放行，命中的攒成一个 HITLRequest，然后调 interrupt()。执行停在这里。

有人会问，在 system prompt 里写一句”执行敏感操作前先问我”不行吗。行，但不稳。模型可能忘，也可能理解成”在回复里提一句就算问过”。中间件是代码，命中就是命中，没得商量。

官方文档把这段生命周期拆成五步：模型生成回复，中间件检查回复里的 tool_calls，命中策略就构造 HITLRequest 并调 interrupt，然后等决策；拿到 HITLResponse 之后，批准和编辑过的调用照常执行，被拒绝的合成一条 ToolMessage，人代答的内容也直接当成 ToolMessage 塞回去；之后继续跑。这套顺序值得记住，出问题时排查基本就沿着它走。

一次中断和一次批准

先看能跑的最小版本。两个工具，发邮件要审批，读邮件放行。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
import os
import sys
import json

sys.stdout.reconfigure(encoding="utf-8")

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

model = ChatOpenAI(
    api_key=os.getenv("DEEPSEEK_API_KEY"),
    base_url="https://api.deepseek.com/v1",
    model="deepseek-chat",
    temperature=0.1,
    max_tokens=1000,
)


def read_email(email_id: str) -> str:
    """Read an email by its ID."""
    return f"Email content for ID: {email_id}"


def send_email(recipient: str, subject: str, body: str) -> str:
    """Send an email to a recipient."""
    print(f"[TOOL EXECUTED] send_email -> {recipient} / {subject}")
    return f"Email sent to {recipient} with subject '{subject}'"


agent = create_agent(
    model=model,
    tools=[read_email, send_email],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "send_email": {"allowed_decisions": ["approve", "edit", "reject"]},
                "read_email": False,
            },
        ),
    ],
    checkpointer=InMemorySaver(),
)

config = {"configurable": {"thread_id": "demo-approve"}}

# 第一次 invoke 跑到中断为止
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Send an email to alice@example.com with subject 'Weekly sync moved' "
                "and body 'The weekly sync moves to Friday 3pm.'",
            }
        ]
    },
    config=config,
    version="v2",
)

print("result type:", type(result).__name__)
print(json.dumps(result.interrupts[0].value, ensure_ascii=False, indent=2))

# 用同一个 thread_id 恢复
final = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
    version="v2",
)

for msg in final.value["messages"]:
    print(f"--- {msg.type} ---")
    print(msg.content)

真实输出，第一段是中断载荷：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
result type: GraphOutput
{
  "action_requests": [
    {
      "name": "send_email",
      "args": {
        "recipient": "alice@example.com",
        "subject": "Weekly sync moved",
        "body": "The weekly sync moves to Friday 3pm."
      },
      "description": "Tool execution requires approval\n\nTool: send_email\nArgs: {'recipient': 'alice@example.com', 'subject': 'Weekly sync moved', 'body': 'The weekly sync moves to Friday 3pm.'}"
    }
  ],
  "review_configs": [
    {
      "action_name": "send_email",
      "allowed_decisions": ["approve", "edit", "reject"]
    }
  ]
}

第二段是 approve 之后跑完的结果：

1
2
3
4
5
6
7
8
9
[TOOL EXECUTED] send_email -> alice@example.com / Weekly sync moved
--- human ---
Send an email to alice@example.com with subject 'Weekly sync moved' and body 'The weekly sync moves to Friday 3pm.'
--- ai ---
I'll send that email now.
--- tool ---
Email sent to alice@example.com with subject 'Weekly sync moved'
--- ai ---
The email has been sent to alice@example.com with the subject "Weekly sync moved" and the body "The weekly sync moves to Friday 3pm."

两个细节值得留意。

第一，version="v2" 让 invoke 返回 GraphOutput，中断载荷挂在 .interrupts 上，最终状态在 .value 里。不传 version 也能跑，返回的是普通 dict，中断载荷在 result["__interrupt__"] 里。两种格式都对，v2 读起来顺一点。

第二，[TOOL EXECUTED] 这行出现在 approve 之后。第一次 invoke 时它没有打印，这就是闸门生效的证据。

interrupt_on 怎么配

interrupt_on 是一个字典，key 是工具名，value 有三种写法。

value 含义
False 这个工具不审批，直接执行。不写在字典里也是一样的效果
True 允许全部决策：approve、edit、reject、respond
{"allowed_decisions": [...]} 只允许列出的决策，还能带 description、when、args_schema

允许的决策一共四种，来自官方文档。

决策 行为 典型场景
approve 按模型给的参数原样执行 邮件草稿没问题，直接发
edit 改掉参数再执行 收件人写错了，改一个再发
reject 不执行，把拒绝理由作为反馈交回模型 拒绝删除文件，并说明原因
respond 不执行，把人的回复直接当成工具结果 给 ask_user 这类占位工具做人肉回答

reject 和 respond 的区别容易搞混。reject 合成的 ToolMessage 状态是 error，语义是”这个动作被否了”；respond 合成的是 success，语义是”工具成功返回了人给的答案”。拿 respond 去否一个写操作，模型会以为写成功了。文档里专门提醒过这一点。

InterruptOnConfig 里还能配几个东西：

  • description：字符串或者 callable(tool_call, state, runtime)，决定中断载荷里 description 字段长什么样。做前端卡片就靠它。
  • description_prefix：构造在中间件上，默认 "Tool execution requires approval"，只在工具没有自定义 description 时生效。
  • when：callable(ToolCallRequest) -> bool，返回 True 才中断。需要 langchain>=1.3.3。
  • args_schema：可选的 JSON Schema，会出现在 review_configs 里，给前端判断哪些字段可以编辑。中间件目前不会自动从工具签名生成它，要自己填。

配置写成这样：

1
2
3
4
5
6
7
8
HumanInTheLoopMiddleware(
    interrupt_on={
        "write_file": True,
        "execute_sql": {"allowed_decisions": ["approve", "reject"]},
        "read_data": False,
    },
    description_prefix="Tool execution pending approval",
)

有个坑要提前说。如果某个工具的配置字典里 allowed_decisions 是空的或者拼错了 key，中间件在构造时直接抛 ValueError，不会静默跳过。这个设计是对的，否则你以为装了闸门，实际上门一直开着。

interrupt 挂起和恢复的完整时序

interrupt() 内部是抛一个特殊异常。运行时接住它，把当前图状态写进 checkpointer，然后把控制权交还给调用方。恢复时 Command(resume=...) 的值会成为 interrupt() 的返回值。

这里有个必须记住的语义：恢复时节点是从头重跑的，不是从断点那一行继续。文档写得很清楚，interrupt 之前已经执行过的代码会再跑一遍。中间件这个场景里重跑的是 after_model 钩子，它重新比对一遍 tool_calls，这次 interrupt() 直接拿到决策返回，然后继续往下处理。代价是钩子里的副作用会重复，所以别在钩子之前做不幂等的事。

checkpointer 是硬前提

文档里给 HumanInTheLoopMiddleware 的警告只有一句：需要 checkpointer 来跨中断保存状态。这句话的分量比它看起来重，缺了它整个机制根本转不起来。

没有 checkpointer 会怎样，我实际试了。中断本身还是会触发，载荷也在，但恢复直接炸：

1
2
3
4
5
6
step1 interrupted keys: ['__interrupt__']
step1 interrupt value: [{'name': 'send_email', 'args': {'recipient': 'bob@example.com'}, 'description': "Tool execution requires approval\n\nTool: send_email\nArgs: {'recipient': 'bob@example.com'}"}]

=== step2: try to resume without a checkpointer ===
EXCEPTION: RuntimeError
Cannot use Command(resume=...) without checkpointer

原因不复杂。中断意味着进程要停在一个”待续”的位置上，消息历史、还没执行的 tool call、任务级的 resume 列表，这些都得有个地方存。checkpointer 就是那个地方，config 里的 thread_id 是取回这份状态的指针。没有它，第二次 invoke 只能从头开始，而从头开始就意味着模型重新想一遍，之前那个待审批的调用消失得无影无踪。

InMemorySaver 存在进程内存里，适合本地调试。进程一退，状态就没了，线上别用。生产上按文档换成 AsyncPostgresSaver 或者 MongoDBSaver，这样服务重启之后未审批的单子还在。

恢复执行的写法

四种决策的载荷长这样：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
from langgraph.types import Command

config = {"configurable": {"thread_id": "demo-approve"}}

# approve：原样执行
agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config, version="v2")

# edit：换掉参数再执行
agent.invoke(
    Command(resume={"decisions": [{
        "type": "edit",
        "edited_action": {
            "name": "send_email",
            "args": {"recipient": "right@example.com", "subject": "Report", "body": "See attached."},
        },
    }]}),
    config=config,
    version="v2",
)

# reject：不执行，把理由交回模型
agent.invoke(
    Command(resume={"decisions": [{
        "type": "reject",
        "message": "The meeting time is not confirmed yet. Do not send this email.",
    }]}),
    config=config,
    version="v2",
)

# respond：人替工具回答
agent.invoke(
    Command(resume={"decisions": [{"type": "respond", "message": "Blue."}]}),
    config=config,
    version="v2",
)

edit 的 edited_action 里要写完整的 name 和 args，不是只写改动的那一项。name 通常和原来一样，也允许换成别的工具。

我跑了一次 edit，把收件人从 wrong@example.com 改成 right@example.com，工具确实用新参数执行了。有意思的是回到模型的那条 ToolMessage：

1
2
3
4
5
--- tool ---
Note: a human reviewer replaced this tool call before it ran. The call recorded in your message is the one you produced, not the one that executed. This was intentional and authorized. Do not re-issue your original call. Executed instead: send_email with arguments {"recipient": "right@example.com", "subject": "Report", "body": "See attached."}.

Tool response:
Email sent to right@example.com

这段提示是中间件自动加的，可以通过 edit_notice=None 关掉。它的作用是防止模型看到参数被改之后，又自作主张把原始调用重发一遍。

reject 那条路我也跑了。拒绝理由会拼进 ToolMessage，状态标成 error：

1
2
3
4
5
6
--- tool (status=error) ---
User rejected the tool call for `send_email` with reason: The meeting time is not confirmed yet. Do not send this email. Ask the user to confirm the new time first.
--- ai ---
The email wasn't sent — the tool call was rejected because the meeting time isn't confirmed yet.

Before I send anything, could you confirm the new time? You mentioned Friday 3pm, but I'd like your explicit confirmation that this is final before notifying Alice.

模型没有硬来，转成向用户追问。这就是把 message 写具体的好处。不写 message 的话，中间件会用默认文案告诉模型”工具没执行，除非用户明确要求否则不要重试”。

不用中间件，自己写闸门

中间件是现成品。审批逻辑特别绕的时候，比如要根据外部工单系统的状态决定放不放行，可以直接拿 interrupt 原语自己写。核心只有两行：

1
2
3
4
5
6
from langgraph.types import interrupt


def approval_node(state):
    decision = interrupt({"question": "Approve this action?", "args": state["pending_args"]})
    return {"decision": decision}

节点跑到 interrupt 就停住，恢复时 Command(resume=...) 的值从 interrupt 的返回值进来。自己写要多留神三件事。别把 interrupt 包在裸的 try/except 里，它靠抛异常暂停，异常被捕获就静默失效，图看起来”跑过去了”，实际上闸门没了。同一个节点里有多个 interrupt 时，resume 值是按下标严格匹配的，不能有条件跳过某一次调用，循环里调也要保证每次执行的顺序完全一致。传给 interrupt 的载荷必须能 JSON 序列化，函数、类实例都不行，生产环境的 checkpointer 会直接序列化失败。这三条出自 LangGraph 的 interrupts 文档，中间件内部遵守的也是同一套规则。

审批粒度

粒度可以拧三档。

按工具是最粗的一档，interrupt_on 的 key 决定哪个工具进闸门。读类工具写 False，写类工具给完整决策集，破坏性工具只留 approve 和 reject，不允许改参数。

按条件细一档，用 when 谓词。它收到一个 ToolCallRequest，里面有 tool_call["args"]，可以拿参数做判断：

1
2
3
4
5
6
7
8
9
10
11
12
13
from langchain.agents.middleware import HumanInTheLoopMiddleware, ToolCallRequest


def is_external(request: ToolCallRequest) -> bool:
    recipient = request.tool_call["args"].get("recipient", "")
    return not recipient.endswith("@internal.com")


HumanInTheLoopMiddleware(
    interrupt_on={
        "send_email": {"allowed_decisions": ["approve", "reject"], "when": is_external},
    },
)

发内部同事直接放行，发外部地址才拦。实测效果：

1
2
3
4
5
6
7
[TOOL EXECUTED] send_email -> bob@internal.com / Hi
[cond-internal] recipient=bob@internal.com interrupted=False
  final: The email has been sent to bob@internal.com with the subject "Hi".
[cond-external] recipient=carol@partner.io interrupted=True
  pending: ['send_email']
[TOOL EXECUTED] send_email -> carol@partner.io / Hi
  final: Done — the email with subject "Hi" was sent to carol@partner.io.

内部那封一次没停，外部那封先挂起再执行。注意文档里的一句话：when 返回 False 的调用根本不会进中断批次，审批人看到的只有真正需要他决定的动作。

按参数是第三档，也就是 edit。when 决定要不要拦，edit 决定拦下来之后改哪个字段。

流式场景下的中断

界面要一边出字一边等审批，用 stream_events。它把 token 和中断状态分开展示，适合聊天窗口：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
config = {"configurable": {"thread_id": "some_id"}}

stream = agent.stream_events(
    {"messages": [{"role": "user", "content": "Delete old records from the database"}]},
    config=config,
    version="v3",
)
for message in stream.messages:
    for token in message.text:
        print(token, end="", flush=True)

if stream.interrupted:
    print(stream.interrupts)

# 决策拿到之后，同样用 Command 恢复
stream = agent.stream_events(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
    version="v3",
)

stream.interrupted 是判断要不要弹审批窗的信号，stream.interrupts 就是那张卡片的数据。注意 version="v3" 和前面 invoke 用的 "v2" 不是一回事，前者是事件流接口，后者是单次调用的返回格式，别混着传。

待审批卡片要渲染哪些字段

中断载荷就是前端卡片的数据源，结构是 HITLRequest：

字段 位置 用途
action_requests[].name 中断载荷 要调哪个工具，卡片标题
action_requests[].args 中断载荷 完整参数，卡片正文
action_requests[].description 中断载荷 可读描述，卡片摘要
review_configs[].action_name 中断载荷 和 action 对应
review_configs[].allowed_decisions 中断载荷 这张卡片上该显示哪几个按钮
review_configs[].args_schema 中断载荷 可编辑字段的 JSON Schema，只有配置里给了才有

description 默认是 description_prefix 加上工具名和参数字典，读起来像调试信息。要给用户看，就传一个 callable 自己拼：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
def card(tool_call, state, runtime):
    args = tool_call["args"]
    if tool_call["name"] == "send_email":
        return (
            f"待审批：发送邮件\n"
            f"收件人：{args.get('recipient')}\n"
            f"主题：{args.get('subject')}\n"
            f"正文：{args.get('body')}"
        )
    return f"待审批：删除记录 {args.get('record_id')}"


HumanInTheLoopMiddleware(
    interrupt_on={
        "send_email": {"allowed_decisions": ["approve", "edit", "reject"], "description": card},
        "delete_record": {"allowed_decisions": ["approve", "reject"], "description": card},
    },
)

一次让模型同时发邮件和删记录，会攒成一个中断、两个动作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
{
  "action_requests": [
    {
      "name": "send_email",
      "args": {"recipient": "alice@example.com", "subject": "Sync", "body": "Moved to Friday."},
      "description": "待审批：发送邮件\n收件人：alice@example.com\n主题：Sync\n正文：Moved to Friday."
    },
    {
      "name": "delete_record",
      "args": {"record_id": 42},
      "description": "待审批：删除记录 42"
    }
  ],
  "review_configs": [
    {"action_name": "send_email", "allowed_decisions": ["approve", "edit", "reject"]},
    {"action_name": "delete_record", "allowed_decisions": ["approve", "reject"]}
  ]
}

前端回传的形状是对称的，一个 decisions 数组，长度和顺序都必须和 action_requests 对齐：

1
2
3
4
Command(resume={"decisions": [
    {"type": "approve"},
    {"type": "reject", "message": "Deletion is not approved. Record 42 is still needed."},
]})

这里有个容易踩的地方：action_requests 里没有 tool call 的 id，只有 name、args、description。文档的表述是按顺序匹配，中间件源码里也是按下标去对。所以前端列表的排序必须稳定，用户提交时按同一个下标顺序拼 decisions。如果卡片列表支持拖拽或者重排，务必把原始下标一起带上，别用 UI 上的位置。

几个真实的坑

下面这些都在本机踩过，报错是从终端抄的。

中断能触发，恢复却抛 RuntimeError: Cannot use Command(resume=...) without checkpointer，多半是 agent 创建时没挂 checkpointer。这个前面演示过了。

decisions 的数量和挂起的 tool call 对不上，少给或多给都会炸：

1
ValueError : Number of human decisions (2) does not match number of hanging tool calls (1).

决策类型不在配置允许的范围内，同样是 ValueError。配置里只开了 approve 和 reject，却回传 edit：

1
ValueError : Unexpected human decision: {'type': 'edit', ...}. Decision type 'edit' is not allowed for tool 'send_email'. Expected one of ['approve', 'reject'] based on the tool's configuration.

恢复之后工具被重复执行，来源有两个。一是节点从头重跑，interrupt 之前的副作用会再发生一次，文档要求这些操作幂等，能挪到 interrupt 后面就挪过去。二是模型拿到 edit 后的结果，可能觉得不对劲，把原始调用重新发起；中间件用 edit_notice 那段提示压这个行为，reject 的默认文案也明确写了不要重试。

thread_id 换了，状态就丢了。恢复必须用中断时那个，用一个新的，checkpointer 会当成全新会话，空状态起跑。这个错误的表现是模型一脸茫然地重新问一遍需求，而不是报错，比较难查。

一次中断只处理当前这批 tool_calls。模型拿到结果之后如果又发起新的敏感调用，会再触发一次中断，界面上就是第二次审批。前端得有个循环，每次 invoke 完都检查 .interrupts 是否非空，非空就渲染卡片、等决策、再 resume。

剩下几条属于设计取舍，没有标准答案：

  • respond 合成的是 success 状态的工具结果。用它来否一个写操作，模型会认为写成功了，拒绝就该用 reject。
  • InMemorySaver 重启即失忆，未审批的单子全丢。这个错误在开发环境永远复现不了，上线前记得换成持久化的 saver。
  • 审批人关掉页面之后，图会无限期挂起，没有超时机制，这是文档写明的行为。挂起期间那个 thread_id 一直是”待续”状态，没人处理这条会话就卡死。要不要加超时、超时后自动 reject 还是转人工，属于业务决定，中间件不替你选；落地时通常在后端存一份待审批队列，配个定时任务清理。
  • args 是模型生成的原始参数，键名是工具签名里的英文，值可能是嵌套结构。直接 json.dumps 出来给用户看，信息是够的，但没人愿意读。至少给每个工具配一个 description callable 把参数翻译成人话，字段多的时候还可以用 args_schema 驱动一个表单。

小结

  • 闸门装在 after_model，模型提议工具调用之后、工具执行之前。配置入口是 HumanInTheLoopMiddleware(interrupt_on={...})。
  • 决策有四种：approve、edit、reject、respond。拒绝用 reject，respond 只留给需要人肉回答的占位工具。
  • interrupt() 靠抛异常暂停，状态写进 checkpointer，thread_id 是取回状态的指针。没有 checkpointer，Command(resume=...) 直接抛 RuntimeError。
  • 恢复时节点从头重跑，不是断点续传。interrupt 之前的副作用必须幂等。
  • 审批粒度三档：按工具（interrupt_on 的 key）、按条件（when 谓词读 tool_call["args"]）、按参数（edit 改 edited_action.args）。前端卡片按 action_requests 和 review_configs 渲染，回传的 decisions 顺序必须和 action_requests 一一对应。