Appearance
人机协作 HITL
前置阅读：内置中间件 · Agent 实战指南
本节解决什么问题
Human-in-the-Loop（HITL，人机协作 / 人工介入）是一种在 Agent 执行关键操作前暂停并等待人工审批的机制。并非所有 AI 决策都应该自动执行--发送邮件、修改数据库、执行支付等高风险操作需要人类确认后才能继续。
HITL 的核心思想是：让 AI 处理日常工作，让人类把关关键决策。
为什么需要 HITL
| 场景 | 没有 HITL | 有 HITL |
|---|---|---|
| 发送邮件 | Agent 直接发送，内容可能不当 | 人工审核邮件内容后确认发送 |
| 数据库操作 | Agent 直接执行 DELETE，可能误删 | 人工确认 SQL 语句后再执行 |
| 费用支出 | Agent 自动调用付费 API | 超过阈值时暂停，等待审批 |
| 代码部署 | Agent 直接推送到生产环境 | 人工 Review 代码变更后批准 |
必须深刻理解，不能跳过：interrupt 是服务端持久化暂停，可跨请求恢复
HITL 的中断不是内存中的"挂起线程"，而是通过 LangGraph 的
interrupt机制将完整状态持久化到 Checkpointer，然后完全停止执行。人工审批可能在几分钟后、几小时后甚至几天后才返回，此时用同一个thread_id恢复执行，Agent 从中断点精确恢复，不丢失任何上下文。这意味着：中断和恢复可以发生在不同的 HTTP 请求中，甚至在不同的进程 / 服务器中（只要共享同一个 Checkpointer）。这是 Durable Execution（持久化执行）的核心能力。
前端类比
HITL 类似于表单提交前的确认对话框（window.confirm()），但有本质区别：
javascript
// 前端确认 - 同步阻塞，客户端内存
if (window.confirm('确定发送邮件？')) {
sendEmail() // 用户点击后立即执行
}LangChain HITL 是服务端持久化暂停--Agent 将状态保存到数据库后完全停止，可能几分钟后甚至几天后才收到人工审批，然后从中断点恢复执行。这更像是审批工作流系统（如 OA 系统的请假审批），而不是简单的 window.confirm()。
原生语义
window.confirm() 是同步阻塞：浏览器主线程暂停，等待用户点击。一旦关闭浏览器或刷新页面，状态丢失。LangChain 的 interrupt 是持久化暂停：状态写入 Checkpointer（数据库），进程可以安全退出。恢复时通过 thread_id 从 Checkpointer 加载状态，从中断点继续。这不是"暂停线程"，而是"保存进度"。
HITL 决策流程
HumanInTheLoopMiddleware
LangChain 内置了 HumanInTheLoopMiddleware，提供开箱即用的 HITL 能力。
完整的 API 参数和配置选项请参考 内置中间件。
基本用法
python
import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""发送电子邮件"""
return f"邮件已发送至 {to}"
@tool
def search(query: str) -> str:
"""搜索信息"""
return f"搜索结果: {query}"
model = init_chat_model(os.environ["LLM_MODEL"])
agent = create_agent(
model=model,
tools=[send_email, search],
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={
"send_email": {
"allowed_decisions": ["approve", "edit", "reject"],
},
},
),
],
checkpointer=InMemorySaver(), # HITL 需要 Checkpointer 保存中断状态
)注意
HumanInTheLoopMiddleware 的参数是 interrupt_on=，不是 tools=。传 tools= 会抛出 TypeError。interrupt_on 是一个字典，key 为工具名，value 为 True（简写，启用全部决策）或 {"allowed_decisions": [...]}（指定允许的决策）。
python
# 错误 - 会 TypeError
HumanInTheLoopMiddleware(tools=["send_email"])
# 正确 - 简写
HumanInTheLoopMiddleware(interrupt_on={"send_email": True})
# 正确 - 完整配置
HumanInTheLoopMiddleware(
interrupt_on={"send_email": {"allowed_decisions": ["approve", "edit", "reject"]}}
)interrupt_on 配置
python
# 简写：启用审批，默认允许 approve / edit / reject
interrupt_on={"send_email": True}
# 完整配置：指定允许的决策
interrupt_on={
"send_email": {"allowed_decisions": ["approve", "edit", "reject"]},
"delete_account": {"allowed_decisions": ["approve", "reject"]}, # 不允许编辑
}
# 对所有工具启用审批（简写）
interrupt_on={"*": True}中断与恢复流程
第 1 步：发起请求，触发中断
python
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
# ... agent 创建同上 ...
# 使用 thread_id 标识会话（恢复时必须使用同一个 thread_id）
config = {"configurable": {"thread_id": "email-thread-1"}}
result = agent.invoke(
{"messages": [{"role": "user", "content": "给 alice@example.com 发一封会议通知邮件"}]},
config=config,
)
# Agent 在调用 send_email 前暂停
# 中断信息可从 result 中获取第 2 步：获取中断信息
中断发生后，Agent 的状态保存在 Checkpointer 中。中断信息（即将调用的工具名、参数等）可从两个位置获取：
python
# 方式 1：从 __interrupt__ 获取
interrupts = result.get("__interrupt__")
if interrupts:
for interrupt in interrupts:
# interrupt.value 包含中断详情（工具名、参数、描述等）
print(f"中断信息: {interrupt.value}")
# 方式 2：从 messages 获取
last_msg = result["messages"][-1]
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
for tc in last_msg.tool_calls:
print(f"待审批工具: {tc['name']}")
print(f"参数: {tc['args']}")__interrupt__ 是 LangGraph 的标准中断信息字段，包含 Interrupt 对象列表。每个 Interrupt 的 value 属性携带中断上下文（工具名、参数、审批描述等）。
第 3 步：人工审批后恢复执行
恢复使用 Command(resume=...)，不是 普通用户消息：
python
from langgraph.types import Command
# 使用同一个 thread_id 恢复
config = {"configurable": {"thread_id": "email-thread-1"}}
# 方式 A：批准执行（使用原始参数）
result = agent.invoke(
Command(resume={"action": "approve"}),
config=config,
)
# Agent 恢复执行，使用原始参数调用 send_email
# 方式 B：编辑参数后批准
result = agent.invoke(
Command(resume={
"action": "edit",
"args": {
"to": "alice@example.com",
"subject": "紧急：会议通知", # 修改了标题
"body": "您好，明天下午 3 点有紧急项目周会...",
},
}),
config=config,
)
# Agent 使用修改后的参数调用 send_email
# 方式 C：拒绝执行
result = agent.invoke(
Command(resume={"action": "reject"}),
config=config,
)
# Agent 收到拒绝信号，生成替代回复（如"邮件发送已取消"）必须深刻理解，不能跳过：恢复命令不是普通用户消息
恢复执行必须使用
Command(resume={"action": ...})，不能用{"messages": [{"role": "user", "content": "approve"}]}这样的普通用户消息。普通消息会被 Agent 当作新的用户输入处理，启动新一轮推理，而不是从中断点恢复。
Command是 LangGraph 的控制流原语，resume字段告诉运行时"这是对之前 interrupt 的回应"。中断和恢复通过thread_id关联--恢复时从 Checkpointer 加载中断时的完整状态，然后根据action决定如何继续。
approve / edit / reject 语义
| 决策 | resume 载荷 | 行为 |
|---|---|---|
approve | {"action": "approve"} | 使用原始参数执行工具 |
edit | {"action": "edit", "args": {...}} | 使用 args 中的修改后参数执行工具 |
reject | {"action": "reject"} | 不执行工具，向 Agent 返回拒绝信息，Agent 可调整策略 |
幂等性
中断和恢复是幂等的：如果恢复命令因网络问题重复发送，Checkpointer 会确保工具只执行一次。这是因为恢复操作基于 checkpoint 状态机--只有处于"中断"状态的 thread 才能接受 Command(resume=...)，已恢复的 thread 再次收到 resume 不会重复执行。
完整示例：邮件发送审批系统
python
import os
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
@tool
def send_email(to: str, subject: str, body: str) -> str:
"""发送电子邮件
Args:
to: 收件人邮箱地址
subject: 邮件主题
body: 邮件正文
"""
print(f"[邮件已发送] 收件人: {to}, 主题: {subject}")
return f"邮件已成功发送至 {to}"
@tool
def search_contacts(name: str) -> str:
"""搜索联系人信息
Args:
name: 联系人姓名
"""
contacts = {
"Alice": "alice@example.com",
"Bob": "bob@company.com",
}
email = contacts.get(name)
if email:
return f"{name} 的邮箱是 {email}"
return f"未找到联系人 {name}"
@tool
def get_schedule(date: str) -> str:
"""查询日程安排
Args:
date: 日期，格式 YYYY-MM-DD
"""
return f"{date} 的日程：14:00 技术评审，16:00 周会"
# 通过环境变量配置模型（示例名称，可能随时间变化）
model = init_chat_model(os.environ["LLM_MODEL"])
agent = create_agent(
model=model,
tools=[send_email, search_contacts, get_schedule],
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={
"send_email": {
"allowed_decisions": ["approve", "edit", "reject"],
},
},
),
],
checkpointer=InMemorySaver(),
system_prompt=(
"你是一个办公助手，可以帮助用户搜索联系人、查询日程和发送邮件。\n"
"发送邮件前请先确认收件人信息。"
),
)
# --- 运行流程 ---
config = {"configurable": {"thread_id": "email-demo"}}
# 第 1 步：用户请求发送邮件
result = agent.invoke(
{
"messages": [
{
"role": "user",
"content": "帮我给 Alice 发一封邮件，通知她明天的技术评审时间",
}
]
},
config=config,
)
# Agent 会先搜索联系人、查询日程（自动执行，不需要审批）
# 然后在发送邮件前暂停，等待审批
# 第 2 步：获取中断信息
last_msg = result["messages"][-1]
if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
for tc in last_msg.tool_calls:
if tc["name"] == "send_email":
print("=== 待审批的邮件 ===")
print(f"收件人: {tc['args'].get('to')}")
print(f"主题: {tc['args'].get('subject')}")
print(f"正文: {tc['args'].get('body')}")
print("==================")
# 第 3 步：人工批准
result = agent.invoke(
Command(resume={"action": "approve"}),
config=config,
)
print(f"最终回复: {result['messages'][-1].content}")与 Web 应用集成
在实际生产环境中，HITL 通常需要集成到 Web 应用中，通过 UI 界面展示待审批的操作。
审批 UI 模式
┌─────────────────────────────────────────┐
│ AI 助手请求执行以下操作： │
│ │
│ 📧 发送邮件 │
│ ├─ 收件人: alice@example.com │
│ ├─ 主题: 会议通知 │
│ └─ 正文: 您好，明天下午 3 点有项目周会... │
│ │
│ ┌──────┐ ┌──────┐ ┌──────┐ │
│ │ 批准 │ │ 编辑 │ │ 拒绝 │ │
│ └──────┘ └──────┘ └──────┘ │
└─────────────────────────────────────────┘FastAPI 后端示例
python
import os
from fastapi import FastAPI
from pydantic import BaseModel
from langchain.agents import create_agent
from langchain.chat_models import init_chat_model
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
app = FastAPI()
model = init_chat_model(os.environ["LLM_MODEL"])
agent = create_agent(
model=model,
tools=[send_email, search_contacts],
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={
"send_email": {"allowed_decisions": ["approve", "edit", "reject"]},
},
),
],
checkpointer=InMemorySaver(),
)
class ApprovalRequest(BaseModel):
thread_id: str
action: str # "approve" | "edit" | "reject"
modified_args: dict | None = None # action="edit" 时提供
@app.post("/chat")
async def chat(message: str, thread_id: str):
"""用户发送消息，可能触发中断"""
config = {"configurable": {"thread_id": thread_id}}
result = agent.invoke(
{"messages": [{"role": "user", "content": message}]},
config=config,
)
# 检查是否中断
interrupts = result.get("__interrupt__")
if interrupts:
# 提取待审批的工具调用信息
last_msg = result["messages"][-1]
tool_calls = getattr(last_msg, "tool_calls", [])
return {
"status": "pending_approval",
"tool_calls": [
{"name": tc["name"], "args": tc["args"]}
for tc in tool_calls
],
"message": "Agent 请求执行以下操作，请审批：",
}
return {
"status": "completed",
"response": result["messages"][-1].content,
}
@app.post("/approve")
async def approve(request: ApprovalRequest):
"""处理人工审批 - 使用 Command 恢复"""
config = {"configurable": {"thread_id": request.thread_id}}
if request.action == "approve":
resume_data = {"action": "approve"}
elif request.action == "edit" and request.modified_args:
resume_data = {"action": "edit", "args": request.modified_args}
elif request.action == "reject":
resume_data = {"action": "reject"}
else:
return {"error": "无效的审批决策"}
# 使用 Command 恢复，不是普通用户消息
result = agent.invoke(
Command(resume=resume_data),
config=config,
)
return {
"status": "completed",
"response": result["messages"][-1].content,
}关键区别
恢复执行时使用 Command(resume=...)，不是 {"messages": [{"role": "user", "content": "approve"}]}。普通用户消息会被当作新输入，不会从中断点恢复。Command 是 LangGraph 的控制流原语，resume 字段告诉运行时"这是对 interrupt 的回应"。
EnviroNexus 映射：发布前人工审核证据
在 EnviroNexus 环保知识库中，HITL 对应发布前人工审核证据环节：
用户问题 -> factor_alias 实体匹配 -> 标准号/因子结构化过滤
-> keyword/vector/hybrid retrieval -> MethodCard -> evidence_refs 校验
-> 结构化候选答案 -> interrupt 人工审核 -> 发布或驳回python
# EnviroNexus 场景：发布环保标准解读前，需人工审核证据链
agent = create_agent(
model=model,
tools=[search_regulations, fetch_factor_data, publish_report],
middleware=[
HumanInTheLoopMiddleware(
interrupt_on={
"publish_report": {
"allowed_decisions": ["approve", "reject"],
# 不允许 edit：报告内容由证据链决定，人工不应修改结论
},
},
),
],
checkpointer=InMemorySaver(),
system_prompt=(
"你是环保知识库助手。发布报告前必须确保 evidence_refs 完整。\n"
"无 evidence_refs 不得发布标准结论。"
),
)铁律：Retriever 找 Document 不回答；Tool 是模型能力入口、内部可调 Retriever；无 evidence_refs 不得出标准结论；Prompt 约束 ≠ 程序级证据校验。HITL 是最后一道人工防线。
与 LangGraph Interrupts 的关系
LangChain 的 HumanInTheLoopMiddleware 底层使用了 LangGraph 的 interrupt 机制。如果你需要更细粒度的控制（如在自定义节点中设置中断、多步审批流程等），可以直接使用 LangGraph 的 Interrupts API。
| 特性 | LangChain HITL Middleware | LangGraph Interrupts |
|---|---|---|
| 使用门槛 | 低，声明式配置 | 中，需要了解图结构 |
| 灵活性 | 工具级别的中断 | 任意节点级别的中断 |
| 多步审批 | 不支持 | 支持 |
| 自定义中断逻辑 | 有限 | 完全自定义 |
| 恢复方式 | Command(resume=...) | Command(resume=...)（相同） |
| 适用场景 | 简单的工具审批 | 复杂的审批工作流 |
详细了解 LangGraph 的中断机制，请参考 LangGraph Interrupts。
最佳实践
1. 合理选择需要审批的工具
不是所有工具都需要 HITL。遵循最小权限原则：
python
# 推荐：只对有副作用的工具启用
HumanInTheLoopMiddleware(
interrupt_on={
"send_email": True,
"delete_file": True,
"execute_sql": {"allowed_decisions": ["approve", "reject"]},
}
)
# 不推荐：对所有工具启用（用户体验差）
HumanInTheLoopMiddleware(interrupt_on={"*": True})2. 精简 allowed_decisions
不可逆操作不应允许编辑：
python
# 删除操作：只允许批准或拒绝（不应编辑删除参数）
"delete_account": {"allowed_decisions": ["approve", "reject"]}
# 发送邮件：允许编辑内容后发送
"send_email": {"allowed_decisions": ["approve", "edit", "reject"]}3. 提供清晰的审批上下文
在 system_prompt 中指导 Agent 解释为什么要执行该操作：
python
system_prompt = """
在调用需要审批的工具之前，请先向用户说明：
1. 你准备执行什么操作
2. 为什么要执行该操作
3. 操作的具体参数
"""4. 记录审批日志
所有审批决策都应记录，便于审计追溯：
python
import logging
logger = logging.getLogger("hitl_audit")
# 在恢复时记录
logger.info(
"HITL 审批: thread=%s, tool=%s, action=%s, reviewer=%s",
thread_id,
tool_name,
action,
reviewer_id,
)5. 设置审批超时
长时间未审批的请求应有超时处理：
python
from datetime import datetime, timedelta
APPROVAL_TIMEOUT = timedelta(hours=24)
async def check_pending_approvals():
"""定期检查并过期超时的审批请求"""
pending = await get_pending_approvals()
for approval in pending:
if datetime.now() - approval.created_at > APPROVAL_TIMEOUT:
await expire_approval(approval.thread_id)
# 可以选择自动 reject 或通知用户审批已超时下一步
- 内置中间件 - HumanInTheLoopMiddleware 的完整 API 参考
- Agent 实战指南 - Agent 的核心用法与工具绑定
- LangGraph Interrupts - 更灵活的图级别中断机制