HITL 人在回路：给 AI Agent 装一根"安全带"——LangGraph / DeepAgents 实现拆解

❝你敢让一个 AI Agent 全自动地帮你删数据库、付一笔钱、给客户群发邮件吗？大概率不敢。 但你又希望它能把这些活儿干完——只是在「真正按下回车那一刻」，让你瞄一眼、点个"确认"。这就是 「HITL（Human-in-the-Loop，人在回路）」：在 Agent 自主奔跑的链路上，挑几个高风险节点插一道「人工闸门」。 这篇文章不停留在"HITL 是个好东西"的层面，而是把它「怎么实现」讲透：一个无状态的大模型循环，到底是怎么做到"跑到一半停下来、等你几个小时、你回来再原地满血续上"的？我们重点拆 「LangGraph」（底层引擎）和 「DeepAgents」（上层 Harness）两层的真实源码，再横向扫一遍其他主流框架的实现思路。图文并茂，建议收藏。❞

一、先把概念说清楚：HITL 到底是什么

一句话：「HITL 就是在 Agent 的执行回路里，给人留一个"喊停 + 介入"的位置。」

把 Agent 想成一辆能自己开的车。全自动很爽，但下高速、并线、闯红灯这种关键动作，你希望方向盘还在自己手里。HITL 就是那套"关键时刻请人接管"的机制。

它和旁边几个安全机制的分工，很多人会混：

注意：HITL 不只是"批准/拒绝"两个按钮。成熟的实现把人的介入拆成「四种动作」——这是后面看 DeepAgents 源码的关键，先记住：

二、核心难题：一个无状态的循环，凭什么能"暂停几小时"

要理解 HITL 的实现，得先看清它难在哪。

大模型本身是「无状态」的：每次调用都是一锤子买卖，它不记得上一句。Agent 之所以显得连贯，是靠外面那个 「Agent loop」（调模型 → 执行工具 → 把结果塞回去 → 再调模型）不停地把历史喂回去。

现在问题来了：HITL 要求这个循环能在"模型说要调 delete_database"这一刻「冻结」，把请求摆给人看，然后——可能等 3 秒，也可能等 3 小时，甚至进程都重启了——人回来点了"批准"，循环要能「从冻结的那一点原地满血续上」，就像什么都没发生。

一个普通的 while 循环做不到这点。真正能扛住"暂停—持久化—恢复"的，是下面这套「三件套」:

1. 「interrupt（喊停）」：在节点内部抛出一个特殊的"可恢复异常"，把要给人看的数据带出去，让整张图停摆。
2. 「checkpointer（存档）」：把图停摆那一刻的「完整状态」（消息历史、变量、走到哪一步）落盘。这是能"等几小时"的物理基础——状态在硬盘上，不怕进程重启。
3. 「resume（读档续上）」：人给出决策后，用一个 Command(resume=...) 把决策值喂回去，框架从存档点重放，并把决策值"注入"到当初喊停的地方。

❝一句话记住：「HITL = 可恢复的暂停。而"可恢复"三个字，全靠 checkpointer 把状态存下来。」 这也是为什么所有正经的 HITL 实现，都强制要求你配一个 checkpointer——没有存档，就没有读档。 ❞

下面我们看这套三件套在 LangGraph 里到底长什么样。

三、底层引擎：LangGraph 的 interrupt + Command(resume)

LangGraph 是把上面三件套做成「一等公民」的框架。它的 HITL 就两个 API：节点里调 interrupt(value) 喊停，外面用 Command(resume=value) 续上。

3.1 interrupt()：一个会"重放"的暂停

看一眼它的真实实现（去掉细节后的核心逻辑）：

def interrupt(value: Any) -> Any:
    scratchpad = conf[CONFIG_KEY_SCRATCHPAD]
    idx = scratchpad.interrupt_counter()        # 这是本节点第几个 interrupt
    # 已经有人给了 resume 值？把它返回，循环继续
    if scratchpad.resume and idx < len(scratchpad.resume):
        return scratchpad.resume[idx]
    v = scratchpad.get_null_resume(True)
    if v isnotNone:
        scratchpad.resume.append(v)
        return v
    # 还没有 resume 值 → 抛出可恢复异常，整张图停摆
    raise GraphInterrupt(
        (Interrupt.from_ns(value=value, ns=conf[CONFIG_KEY_CHECKPOINT_NS]),)
    )

这段代码藏着 LangGraph HITL 最反直觉、也最关键的一个设计——「重放（replay）语义」：

• 「第一次」执行到 interrupt()：没有 resume 值，于是抛 GraphInterrupt，把 value（要给人看的内容）带出去，图停在这里，状态被 checkpointer 存档。
• 人给了决策、用 Command(resume=...) 恢复后，LangGraph 「从这个节点的开头重新执行一遍」。再次走到 interrupt() 时，scratchpad 里已经有 resume 值了，于是这次它「不再抛异常，而是直接把决策值返回」，循环就这么续上了。

❝⚠️ 踩坑预警：因为恢复时「整个节点会从头重跑」，所以 interrupt()「之前」的代码（比如发一封邮件、写一次数据库）会被执行两次！实践中要么把副作用放到 interrupt() 之后，要么把 interrupt() 放在节点最前面。这是 LangGraph HITL 最常见的 bug 来源。 ❞

3.2 一个最小例子：暂停 → 看 → 续

from langgraph.types import interrupt, Command

def node(state):
    answer = interrupt("请问你多大年纪？")   # 停在这，把问题抛给客户端
    return {"human_value": answer}

# 第一次跑：图停摆，吐出 __interrupt__
for chunk in graph.stream({"foo": "abc"}, config):
    print(chunk)
# > {'__interrupt__': (Interrupt(value='请问你多大年纪？', id='...'),)}

# 人给了答案，用 Command(resume=...) 续上
for chunk in graph.stream(Command(resume="30 岁"), config):
    print(chunk)
# > 节点从头重跑，这次 interrupt() 直接返回 "30 岁"
# > {'node': {'human_value': '30 岁'}}

注意那个 config——里面带着 thread_id，配合 checkpointer，LangGraph 才知道"续的是哪一次对话的档"。

Command 这个原语也值得记一下，它能携带三种续法：resume（喂回决策值）、update（顺手改一下状态）、goto（指定下一步跳到哪个节点）。HITL 主要用 resume。

「小结 LangGraph 层」：它提供的是「最通用的暂停/恢复底座」——你能 interrupt 任何东西，问年龄、要审批、求澄清都行。但它不关心"工具调用要不要审批"这种业务语义，那是上层的事。

四、上层 Harness：DeepAgents 的 interrupt_on 与 HITL 中间件

DeepAgents 底层就是 LangChain + LangGraph，所以它「不重新发明」 HITL，而是把 LangGraph 的 interrupt() 包装成一个面向"工具调用审批"的、「声明式」的高层 API。

4.1 一行配置开启工具审批

在 DeepAgents 里开 HITL，你只需要给 create_deep_agent 传一个 interrupt_on：

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver

agent = create_deep_agent(
    model=model,
    tools=[sample_tool, get_weather, get_soccer_scores],
    interrupt_on={
        "sample_tool": True,                                    # 全部 4 种决策都允许
        "get_weather": False,                                   # 自动放行，不打扰
        "get_soccer_scores": {"allowed_decisions": ["approve", "reject"]},  # 只能批准/拒绝
    },
    checkpointer=MemorySaver(),   # 注意：HITL 必须配 checkpointer
)

语义很干净：

• True = 这个工具每次调用都要人审批（允许 approve/edit/reject/respond 全部四种）；
• False（或不写）= 自动放行；
• 给一个 dict = 精细控制这个工具「允许哪几种」决策。

4.2 它在底层做了什么：挂一个中间件

翻 create_deep_agent 的源码，interrupt_on 的落地只有关键一行——在中间件栈的尾部挂上 LangChain 的 HumanInTheLoopMiddleware：

# deepagents/graph.py 中间件装配
if interrupt_on is not None:
    deepagent_middleware.append(HumanInTheLoopMiddleware(interrupt_on=interrupt_on))

这就是 DeepAgents 的整体哲学：「所有能力都是中间件」（计划、文件、子代理、压缩、记忆……HITL 也只是其中一层）。HITL 这层挂在最靠近模型输出的位置，专门拦截"模型刚吐出来的工具调用"。

4.3 中间件的核心：after_model 拦截工具调用

HumanInTheLoopMiddleware 的精髓在 after_model 钩子里——「模型每次输出之后、工具真正执行之前」，它插进来检查一遍：

核心逻辑（精简后）：

def after_model(self, state, runtime):
    last_ai_msg = ...  # 模型最近一条带 tool_calls 的输出
    # 1. 挑出"需要审批"的工具调用
    action_requests, review_configs, interrupt_indices = [], [], []
    for idx, tool_call in enumerate(last_ai_msg.tool_calls):
        if (config := self.interrupt_on.get(tool_call["name"])) isnotNone:
            ar, rc = self._create_action_and_config(tool_call, config, state, runtime)
            action_requests.append(ar); review_configs.append(rc)
            interrupt_indices.append(idx)
    ifnot action_requests:
        returnNone   # 没有需要审批的，直接放行

    # 2. 一次性把所有待审批动作打包，调 LangGraph 的 interrupt() 喊停
    decisions = interrupt(HITLRequest(
        action_requests=action_requests,
        review_configs=review_configs,
    ))["decisions"]

    # 3. 人回来后，按每个决策改写工具调用
    revised_tool_calls, extra_tool_messages = [], []
    for idx, tool_call in enumerate(last_ai_msg.tool_calls):
        if idx in interrupt_indices:
            decision = decisions[...]
            revised, msg = self._process_decision(decision, tool_call, config)
            ...
    last_ai_msg.tool_calls = revised_tool_calls
    return {"messages": [last_ai_msg, *extra_tool_messages]}

看到没——「它最终还是落到了 interrupt()」。DeepAgents/LangChain 的活儿，是把"工具调用"翻译成"给人看的审批请求（HITLRequest）"，再把"人的决策"翻译回"改写后的工具调用"。三件套里的暂停和恢复，依旧是 LangGraph 在干。

那四种决策，就在 _process_decision 里逐一兑现：

if decision["type"] == "approve":       # 原样放行
    return tool_call, None
if decision["type"] == "edit":          # 用人改过的 name/args 重建调用
    return ToolCall(name=edited["name"], args=edited["args"], id=tool_call["id"]), None
if decision["type"] == "reject":        # 不执行，塞一条 error ToolMessage 给模型
    return tool_call, ToolMessage(content=..., status="error", ...)
if decision["type"] == "respond":       # 不执行，人直接代答一条 success ToolMessage
    return tool_call, ToolMessage(content=decision["message"], status="success", ...)

4.4 并行工具调用 & 子代理继承

两个值得一提的工程细节，从 DeepAgents 的集成测试里能直接读出来：

• 「并行审批」：模型一次吐出 3 个工具调用（比如同时查天气、查球赛、调 sample_tool），中间件会把「需要审批的那几个打包进一次 interrupt」，前端一次性展示、人一次性给出一个 decisions 列表。不需要审批的（如上面配了 False 的 get_weather）直接放行。恢复时这样续：

agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}, {"type": "approve"}]}),
    config=config,
)

• 「子代理继承」：主代理的 interrupt_on 会「自动传递给子代理」（subagent），所以子代理调危险工具时同样会触发审批；子代理也可以声明自己的 interrupt_on「覆盖」继承来的配置。这保证了"审批闸门"不会因为任务被外包给 subagent 就被绕过。

「小结 DeepAgents 层」：它把 LangGraph 通用的 interrupt()，收敛成了 Coding/Research Agent 最高频的那个需求——「按工具名做调用级审批」，并附赠 approve/edit/reject/respond 四种语义、并行打包、子代理继承。一行 interrupt_on 背后，是"中间件拦截 + interrupt 喊停 + checkpointer 存档 + Command 续上"的完整链路。

五、横向对比：其他框架怎么做 HITL

跳出 LangChain 生态，主流框架的 HITL 思路可以归成「三大流派」：

流派 A：图/状态机 + 持久化暂停（最"重"，也最强）

代表就是 「LangGraph / DeepAgents」。优势是「真正可持久化」——能跨进程、跨小时甚至跨天恢复，天然支持"审批工单"这种异步形态；代价是你得接受"状态图 + checkpointer"这套心智模型，以及前面说的"节点重放"陷阱。

流派 B：工具调用审批策略（最贴合 Coding Agent）

代表是 「OpenAI Agents SDK / Codex」、「Claude Code」。它们不一定有"持久化暂停"，而是在「工具执行前插一个审批回调/策略」：

• OpenAI Agents SDK：工具上声明 needs_approval，运行时产出 interruption，宿主决定 approve/reject 后再 run 续上；
• Claude Code：内置 「permission（权限）模式」 + 「hooks」——bash、写文件等动作前弹确认，或用 hook 脚本自动判定放行/拦截。

特点：和"工具调用"这件事咬得最紧，开发体验直接；持久化能力通常弱于流派 A（更多是"同一进程内等用户点一下"）。

流派 C：把"人"建模成一个 Agent / 一次输入（最轻）

代表是 「AutoGen」、「CrewAI」：

• AutoGen 用一个 UserProxyAgent，human_input_mode 设成 ALWAYS / TERMINATE / NEVER，到点就在对话里向真人要输入；
• CrewAI 在 Task 上设 human_input=True，任务产出后让人审一道再继续。

特点：概念最直观（人就是对话里的一个参与者），上手快；但粒度粗（偏"回合级"而非"工具调用级"），异步/持久化能力有限。

一张表收口

❝一句话归纳趋势：「越往"图/状态机"走，HITL 越能持久化、越能做异步审批工单；越往"对话参与者"走，越轻、越直观但越难跨进程恢复。」 LangGraph/DeepAgents 是前者的标杆。 ❞

六、落地建议（少踩三个坑）

1. 「一定要配 checkpointer」。没有它 interrupt 抛出来就再也回不去了。开发用 MemorySaver，生产换 SqliteSaver / PostgresSaver，这样审批才能扛住进程重启。
2. 「警惕"节点重放"导致副作用执行两次」。把不可逆操作（发邮件、扣款、写库）放在 interrupt()「之后」，或让 interrupt 紧贴节点开头。这是 LangGraph HITL 的头号 bug。
3. 「审批粒度别一刀切」。用 DeepAgents 的话，把高风险工具设 True、只读工具设 False、需要"批准或拒绝但不让改"的设 {"allowed_decisions": ["approve", "reject"]}。审批太多会把人累垮，等于没有 HITL。

七、结语

HITL 不是给 Agent 加个"确认框"那么简单。它的难点从来不在 UI，而在「怎么让一个无状态的循环可靠地暂停、存档、再原地复活」。

把这篇的链路在脑子里过一遍就够了：

❝「人按下"批准" ← Command(resume) 续上 ← checkpointer 存档 ← interrupt() 喊停 ← 中间件 after_model 拦下工具调用 ← 模型吐出 delete_database。」 ❞

LangGraph 把"可恢复的暂停"做成底座，DeepAgents 在上面收敛成一行 interrupt_on 的工具审批。其他框架要么用审批回调贴近工具调用，要么把人建模成对话里的一个参与者——各有取舍，但底层都在回答同一个问题：「怎么在自动化和可控之间，留一道人能伸手进去的缝。」

这道缝，就是 Agent 能被放进生产环境的那根安全带。

延伸阅读

• LangGraph 官方 HITL 概念页：https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/