LangChain 人工审批：危险操作先让人确认

这篇适合谁、读完能做到什么

这篇适合想让智能体在写文件、执行 SQL 这类有副作用的操作之前先停下来、等人确认的开发者。读完你能用 HumanInTheLoopMiddleware 给工具调用设置审批策略，分清 approve、edit、reject、respond 四种决定各自的含义，并能在中断之后用同一个线程把执行接着跑下去。

是什么：工具调用前的人工审批

人工介入（Human-in-the-loop，简称 HITL）中间件让人可以监督智能体的工具调用。模型提出一个可能需要审查的操作时，比如写文件或执行 SQL，中间件可以暂停执行，等待人的决定。^[1]

它的工作方式是：用一份可配置的策略逐个检查工具调用，只有配置过的工具才会被检查，没有配置的工具照常运行。需要干预时，它发出一个 interrupt，让执行停下来。图的状态由 LangGraph 的持久化层保存，所以暂停是安全的，之后可以随时恢复。^[1]

人的决定决定接下来怎么走，一共四种：approve 原样执行；edit 修改后再执行；reject 拒绝并附上反馈；respond 用人的回复代替工具的输出，适合 ask_user 这类"问用户"的工具。^[1]

Python 版本的写法如下；JavaScript 版本的中间件叫 humanInTheLoopMiddleware，用 interruptOn 配置，参数采用驼峰写法，例如 allowedDecisions。^[1]

准备

• 语言和版本：示例同时提供 Python 和 JavaScript 两套写法。逐个审批模式（per_call）需要 langchain 1.4.6 或更高版本。^[1]
• 恢复：用 Command 恢复时要使用同一个 thread ID，它指向被暂停的那段对话。^[1]
• 持久化：中断必须配合 checkpointer 才能保存图的状态。本地测试用 InMemorySaver，生产环境用持久化的 checkpointer，例子有 AsyncPostgresSaver 和 MongoDBSaver。^[1]
• 线程：调用智能体时要传入带 thread ID 的 config，它把一次执行和一段对话关联起来。^[1]

上手：给工具设置审批策略

1. 从 langchain.agents.middleware 导入 HumanInTheLoopMiddleware，把它放进 create_agent 的 middleware 列表。^[1]
2. 用 interrupt_on 给每个工具配策略。示例中 write_file 设为 True，四种决定都允许；execute_sql 只允许 approve 和 reject，不能编辑；read_data 设为 False，属于安全操作，不需要审批。^[1]
3. 创建 InMemorySaver 并传给 checkpointer 参数。完整代码如下：

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware # [!code highlight]
from langgraph.checkpoint.memory import InMemorySaver # [!code highlight]

agent = create_agent(
    model="gpt-6.1-sol",
    tools=[write_file, execute_sql, read_data],
    middleware=[
        HumanInTheLoopMiddleware( # [!code highlight]
            interrupt_on={
                "write_file": True,  # All decisions (approve, edit, reject, respond) allowed
                "execute_sql": {"allowed_decisions": ["approve", "reject"]},  # No editing allowed
                "read_data": False, # Safe operation, no approval needed
            },
            # Prefix for interrupt messages - combined with tool name and args to form the full message
            # e.g., "Tool execution pending approval: execute_sql with query='DELETE FROM...'"
            # Individual tools can override this by specifying a "description" in their interrupt config
            description_prefix="Tool execution pending approval",
        ),
    ],
    # Human-in-the-loop requires checkpointing to handle interrupts.
    # In production, use a persistent checkpointer like AsyncPostgresSaver or MongoDBSaver.
    checkpointer=InMemorySaver(),  # [!code highlight]
)

1. 调用智能体时传入带 thread ID 的 config，例如 config = {"configurable": {"thread_id": "some_id"}}。执行会跑到中断为止，返回结果的 interrupts 属性列出了待审核的操作。使用 version="v2" 时结果是 GraphOutput。^[1]
2. 人审完之后，用 Command 恢复执行，决定放在 decisions 列表里，同一个 thread ID：

agent.invoke(
    Command( # [!code highlight]
        resume={"decisions": [{"type": "approve"}]}  # or "reject" [!code highlight]
    ), # [!code highlight]
    config=config, # Same thread ID to resume the paused conversation
    version="v2",
)

一次有多个操作待审：决定要按顺序给

同一轮模型调用里可能有多个工具调用需要审批，每个操作都要有一个决定，决定的顺序必须和中断请求里操作的顺序一致。^[1] 下面的例子给出了一个 approve、一个 edit 和一个 reject：

{
    "decisions": [
        {"type": "approve"},
        {
            "type": "edit",
            "edited_action": {
                "name": "tool_name",
                "args": {"param": "new_value"}
            }
        },
        {
            "type": "reject",
            "message": "This action is not allowed"
        }
    ]
}

默认的 batched 模式每轮模型调用产生一个中断，列出所有需要审核的调用，用有序列表一次回答。per_call 模式则给每个需要审核的工具调用各自一个中断，用中断 ID 逐个回答。^[1] 新的智能体最好直接用逐个模式，LangChain 计划在下一个大版本里把它设为默认。^[1]

只拦截需要审的调用

默认情况下，interrupt_on 里列出的工具每次调用都会暂停。这意味着读取类的工具如果也列在里面，同样会被拦下，所以要么去掉它，要么用 when 只拦截需要审的那部分。如果只想拦截部分调用，可以给某个工具加上 when 谓词。谓词收到一个 ToolCallRequest，示例中 write_file 和 execute_sql 各有一个谓词，返回 True 表示中断，返回 False 表示自动通过。^[1] 条件中断需要 langchain 1.3.3 或更高版本。^[1] 下面的例子根据参数来判断：写到 workspace 之外的文件要审，不是只读 SELECT 的 SQL 也要审。

def writes_outside_workspace(request: ToolCallRequest) -> bool:
    """Pause writes to paths outside the workspace directory."""
    path = request.tool_call["args"].get("path", "")
    return not path.startswith("/workspace/")

def is_write_query(request: ToolCallRequest) -> bool:
    """Pause SQL that isn't a read-only SELECT."""
    query = request.tool_call["args"].get("query", "")
    return not query.lstrip().upper().startswith("SELECT")

审批流程是怎么走的

执行顺序是：模型生成回复；中间件检查回复里的工具调用；如果有需要人工处理的调用，中间件构造一个 HITLRequest 并调用 interrupt；然后智能体等待决定。拿到决定后，中间件执行被批准或修改过的调用，为被拒绝的调用生成 ToolMessage，并恢复执行。^[1] 逐个模式下，中间件在每个被拦截的调用开始运行时暂停，等这个调用的决定到了才执行工具。^[1]

流式输出时，可以用 stream_events() 并设置 version="v3"。stream.messages 逐个输出模型的 token，stream.interrupted 告诉你这次运行是否因为等待人工输入而暂停。^[1]

限制和常见问题

• 谓词在恢复执行时会被再次调用，所以它对同一个调用必须返回相同的结果。如果恢复时变成 False，工具会直接运行，人的回答会被忽略，即使人点的是 reject。^[1]
• 批量模式下，如果只有部分调用的谓词结果变了，中间件会抛出 ValueError。^[1]
• 修改参数要保守。大幅改动原始参数，可能让模型重新判断，结果是多次执行工具，或者做出意料之外的动作。^[1]
• 拒绝要用 reject。respond 的消息会被当作成功的工具结果返回给模型，用它来否决有副作用的工具是错的。^[1]
• 编辑后，模型会收到一条提示，说明审核人替换了这次调用；edit_notice 可以改这段话，设为 None 则去掉。^[1]
• 如果拒绝时没有写 message，中间件会用默认消息告诉模型：工具没有执行，除非用户要求，不要重试同一个调用。^[1]
• 答案不合法时，LangGraph 会按中断的 response_schema 校验并抛出 ValidationError，什么都不保存，可以对同一个中断 ID 重新提交修正后的答案。^[1]
• 如果有工具重试或错误处理中间件，要把 HumanInTheLoopMiddleware 放在它们前面，让它包住它们；否则重试的调用会再次打扰审核人。^[1]
• 生产环境要用持久化的 checkpointer，InMemorySaver 只适合测试和原型。^[1]

下一步

中断机制的细节可以接着读 LangGraph 的中断文档，流式输出的各种模式见 LangChain 的 Streaming 指南。^[1]

本文改编自 LangChain 官方文档《Human-in-the-loop》^[1]，原文以 MIT 许可证发布。