Human in the Loop:给 Agent 加上人工审核
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 的操作。