人在回路中

原文：Human-in-the-loop

一句话

Human-in-the-loop (HITL) middleware 允许在 agent 执行工具调用时添加人工监督，通过可配置的策略暂停执行并等待人工决策。

什么时候翻这页

当你的 agent 需要执行可能需要审查的操作（如写入文件、执行 SQL）时，需要人工干预以确保安全性和准确性；当你希望对 agent 的关键决策进行控制时。

核心概念

Human-in-the-loop middleware

• 作用：在 agent 工具调用过程中添加人工监督
• 工作原理：通过检查每个工具调用与可配置策略的匹配度，决定是否暂停执行
• 执行中断：当需要干预时，middleware 会发出 interrupt 暂停执行
• 状态保存：使用 LangGraph 的 persistence layer 保存图状态，确保安全暂停和恢复

中断决策类型

条件中断

• 使用 when 谓词实现条件中断
• 谓词接收 ToolCallRequest 并返回 True（中断）或 False（自动批准）
• 可基于工具参数进行中断控制

怎么做

配置中断

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

agent = create_agent(
    model="gpt-5.4",
    tools=[write_file, execute_sql, read_data],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "write_file": True,  # 允许所有决策
                "execute_sql": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
                "read_data": False, # 安全操作，无需批准
            },
            description_prefix="Tool execution pending approval",
        ),
    ],
    checkpointer=InMemorySaver(),
)

响应中断

from langgraph.types import Command

# 提供线程 ID 以关联执行与对话线程
config = {"configurable": {"thread_id": "some_id"}}

# 运行 agent 直到触发中断
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Delete old records from the database",
            }
        ]
    },
    config=config,
    version="v2",
)

# 恢复执行
agent.invoke(
    Command(
        resume={"decisions": [{"type": "approve"}]}
    ),
    config=config,
    version="v2",
)

流式处理与人在回路中

from langgraph.types import Command

config = {"configurable": {"thread_id": "some_id"}}

# 流式处理直到中断
for chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Delete old records from the database"}]},
    config=config,
    stream_mode=["updates", "messages"],
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if token.content:
            print(token.content, end="", flush=True)
    elif chunk["type"] == "updates":
        if "__interrupt__" in chunk["data"]:
            print(f"\n\nInterrupt: {chunk['data']['__interrupt__']}")

# 人工决策后恢复流式处理
for chunk in agent.stream(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config=config,
    stream_mode=["updates", "messages"],
    version="v2",
):
    if chunk["type"] == "messages":
        token, metadata = chunk["data"]
        if token.content:
            print(token.content, end="", flush=True)

命令 / API 速查

核心类

• HumanInTheLoopMiddleware: 人在回路中间件
• ToolCallRequest: 工具调用请求对象
• Command: 恢复执行命令

关键参数

• interrupt_on: 工具名称到批准配置的映射
  • True: 中断并使用默认配置
  • False: 自动批准
  • InterruptOnConfig: 自定义配置对象
• description_prefix: 操作请求描述的前缀
• allowed_decisions: 允许的决策列表
• when: 中断谓词函数

决策类型

• {"type": "approve"}: 批准操作
• {"type": "edit", "edited_action": {...}}: 修改后执行
• {"type": "reject", "message": "..."}: 拒绝操作
• {"type": "respond", "message": "..."}: 直接响应

与 LangGraph / RAG 手册的联系

• LangGraph 提供了底层 interrupt 和 persistence 机制，HITL middleware 构建于此之上
• RAG 系统可以结合 HITL 实现对检索结果的人工审核
• 三者共同构成了完整的智能代理系统：RAG 提供知识基础，LangGraph 提供执行框架，HITL 提供人工监督

初学者易错点

1. 缺少 checkpointer: 忘记配置 checkpointer 会导致无法保存状态，中断后无法恢复
2. 错误使用 respond: 不要用 respond 来拒绝有副作用的工具，因为它会被视为成功执行
3. 决策顺序错误: 多个操作需要决策时，必须按照中断请求中的顺序提供决策
4. 过度编辑工具参数: 显著修改原始参数可能导致模型重新评估并意外执行工具多次
5. 忽略线程 ID: 每次调用必须使用相同的 thread ID 来关联对话线程

相关词条

• middleware: 中间件系统
• interrupt: 中断机制
• persistence layer: 持久化层
• ToolCallRequest: 工具调用请求
• ToolMessage: 工具消息