人工介入（HITL）中间件 让您可以为代理工具调用添加人工监督。当模型提出可能需要审核的操作时——例如，写入文件或执行 SQL——中间件可以暂停执行并等待决策。 它通过根据可配置的策略检查每个工具调用来做到这一点。如果需要干预，中间件会发出中断来停止执行。图形状态使用 LangGraph 的持久层保存，因此执行可以安全暂停并稍后恢复。 然后人工决策决定接下来发生什么：操作可以按原样批准（approve）、修改后再运行（edit）或拒绝并提供反馈（reject）。

​
中断决策类型

中间件 定义了人工响应中断的三种内置方式：
决策类型描述示例用例
✅ approve操作按原样批准并执行，无更改。按原样发送电子邮件草稿
✏️ edit工具调用在修改后执行。在发送电子邮件之前更改收件人
❌ reject拒绝工具调用，并添加解释到对话中。拒绝电子邮件草稿并解释如何重写
每个工具可用的决策类型取决于您在 interrupt_on 中配置的政策。当多个工具调用同时暂停时，每个操作都需要单独的决策。决策必须按照中断请求中操作出现的顺序提供。
当编辑工具参数时，请保守地进行。更改原始参数的显著修改可能导致模型重新评估其方法，并可能多次执行工具或采取意外操作。

​
配置中断

要使用 HITL，请在创建代理时将中间件添加到代理的 middleware 列表中。 您使用工具操作到允许每个操作的决策类型的映射来配置它。当工具调用匹配映射中的操作时，中间件将中断执行。 
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware   
from langgraph.checkpoint.memory import InMemorySaver   


agent = create_agent(
    model="gpt-4.1",
    tools=[write_file_tool, execute_sql_tool, read_data_tool],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "write_file": True,  # All decisions (approve, edit, reject) allowed
                "execute_sql": {"allowed_decisions": ["approve", "reject"]},  # No editing allowed
                # 安全操作，不需要批准
                "read_data": False,
            },
            # 中断消息的前缀——与工具名称和参数组合形成完整消息
            # 例如 "Tool execution pending approval: execute_sql with query='DELETE FROM...'"
            # 个别工具可以通过在其 interrupt config 中指定 "description" 来覆盖此设置
            description_prefix="Tool execution pending approval",
        ),
    ],
    # 人工介入需要检查点来处理中断。
    # 在生产中，使用持久化检查点如 AsyncPostgresSaver。
    checkpointer=InMemorySaver(),
)
您必须配置检查点以在中断之间持久化图形状态。 在生产中，使用持久化检查点如 AsyncPostgresSaver。对于测试或原型设计，使用 InMemorySaver。调用代理时，传递包含线程 ID 的 config 以将执行与对话线程关联。 有关详细信息，请参阅 LangGraph 中断文档。
​
interrupt_on
dict
required
工具名称到批准配置的映射。值可以是 True（使用默认配置中断）、False（自动批准）或 InterruptOnConfig 对象。
​
description_prefix
string
default:"工具执行需要批准"
操作请求描述的前缀
InterruptOnConfig 选项：
​
allowed_decisions
list[string]
允许的决策列表：'approve'、'edit' 或 'reject'
​
description
string | callable
静态字符串或用于自定义描述的可调用函数

​
响应中断

当您调用代理时，它会运行直到完成或提出中断。当工具调用匹配您在 interrupt_on 中配置的政策时，会触发中断。使用 version="v2" 时，结果是具有包含需要审核的操作的 interrupts 属性的 GraphOutput。然后您可以向审核者展示这些操作，并在提供决策后恢复执行。 
from langgraph.types import Command

# 人工介入利用 LangGraph 的持久层。
# 您必须提供线程 ID 以将执行与对话线程关联，
# 以便可以暂停对话并在需要人工审核时恢复。
config = {"configurable": {"thread_id": "some_id"}}
# 运行图形直到达到中断。
result = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Delete old records from the database",
            }
        ]
    },
    config=config,
    version="v2",
)

# result 是一个具有 .value 和 .interrupts 的 GraphOutput
print(result.interrupts)
# > (
# >    Interrupt(
# >       value={
# >          'action_requests': [
# >             {
# >                'name': 'execute_sql',
# >                'arguments': {'query': 'DELETE FROM records WHERE created_at < NOW() - INTERVAL \'30 days\';'},
# >                'description': '工具执行需要批准\n\n工具：execute_sql\n参数：{...}'
# >             }
# >          ],
# >          'review_configs': [
# >             {
# >                'action_name': 'execute_sql',
# >                'allowed_decisions': ['approve', 'reject']
# >             }
# >          ]
# >       }
# >    ),
# > )


# 使用批准决策恢复
agent.invoke(
    Command(
        resume={"decisions": [{"type": "approve"}]}    # 或 "reject"
    ),
    config=config,   # 使用相同的线程 ID 来恢复暂停的对话
    version="v2",
)

​
决策类型

使用 approve 按原样批准工具调用并执行，无更改。
agent.invoke(
    Command(
        # 决策作为列表提供，每个待审核的操作一个。
        # 决策的顺序必须与
        # 中断请求中操作的顺序匹配。
        resume={
            "decisions": [
                {
                    "type": "approve",
                }
            ]
        }
    ),
    config=config,    # 使用相同的线程 ID 来恢复暂停的对话
    version="v2",
)

​
使用人工介入进行流式传输

您可以使用 stream() 而不是 invoke() 来在代理运行和处理中断时获取实时更新。使用 stream_mode=['updates', 'messages'] 和 version="v2" 以统一的 v2 格式流式传输代理进度和 LLM 令牌。 
from langgraph.types import Command

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

# 使用 v2 格式流式传输
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":
        # LLM token
        token, metadata = chunk["data"]    # 每个块包含完整的图形状态和/或消息
        if token.content:
            print(token.content, end="", flush=True)
    elif chunk["type"] == "updates":
        # Check for interrupt
        if "__interrupt__" in chunk["data"]:
            print(f"\n\nInterrupt: {chunk['data']['__interrupt__']}")

# Resume with streaming after human decision
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)

​
最佳实践

​
何时使用人工介入

人工介入最适合：
  • 高风险操作：可能产生重大影响的操作（如财务交易、数据删除）
  • 合规要求：需要人工审核的监管要求
  • 敏感数据：处理私人或机密信息的操作

​
优化用户体验

  • 清晰的描述：为每个工具提供清晰的描述以帮助审核者理解操作
  • 上下文信息：在可能的情况下，向审核者提供相关上下文
  • 快速响应：尽量减少人工审核的延迟

​
安全考虑

  • 限制编辑：仅在必要时允许编辑操作
  • 审计日志：记录所有人工决策以进行审计
  • 超时处理：为长时间等待人工响应设置超时

​
下一步

在 GitHub 上编辑此页面或提交问题。
通过 MCP 将这些文档连接到 Claude、VSCode 等，获取实时答案。