Human in the Loop:给 Agent 加上人工审核

Published by rcdfrd on 2026-02-15

Human in the Loop:给 Agent 加上人工审核

Agent 调用工具之前,有时候你想先看一眼,确认没问题再执行。比如 SQL 删除操作、发邮件这类不可逆动作。Human-in-the-Loop 就干这个。

基本用法

LangChain 1.0 用 HumanInTheLoopMiddleware 实现:

from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="openai:gpt-4o-mini",
    tools=[get_user_location, get_weather_for_location],
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={
            "get_user_location": True,
            "get_weather_for_location": {
                "allowed_decisions": ["approve", "edit"]
            }
        }
    )],
    checkpointer=InMemorySaver(),
)

interrupt_on 指定哪些工具需要人工确认。设成 True 表示支持批准、拒绝、编辑三种操作。

处理中断

config = {"configurable": {"thread_id": "1"}}
response = agent.invoke(
    {"messages": [{"role": "user", "content": "外面天气怎么样?"}]},
    config=config
)

if "__interrupt__" in response:
    for interrupt in response["__interrupt__"]:
        for req in interrupt.value["action_requests"]:
            print(req["description"])

    from langgraph.types import Command
    decisions = [{"type": "approve"}]
    response = agent.invoke(
        Command(resume={"decisions": decisions}),
        config=config
    )

执行到需要审核的工具时,Agent 暂停,返回中断信息。你检查之后,用 Command(resume=...) 恢复执行。Checkpointer 在这里不可少,暂停和恢复之间可能隔很久。

小结

HITL 的本质是在工具调用前插入一个同步检查点。配合 checkpointer,暂停可以跨进程、跨时间。生产环境里很实用,比如让运营人员审批 Agent 的操作。