人机协同（Human-in-the-Loop，HITL）中间件允许您为智能体（agent）的工具调用添加人工监督。 当模型提出一个可能需要审查的操作时——例如写入文件或执行 SQL——该中间件可以暂停执行并等待决策。

它通过根据可配置的策略检查每个工具调用来实现此功能。如果需要干预，中间件会发出一个 interrupt，从而停止执行。图状态会使用 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-4o",
    tools=[write_file_tool, execute_sql_tool, read_data_tool],
    middleware=[
        HumanInTheLoopMiddleware( 
            interrupt_on={
                "write_file": True,  # 允许所有决策（approve, edit, reject）
                "execute_sql": {"allowed_decisions": ["approve", "reject"]},  # 不允许编辑
                # 安全操作，无需批准
                "read_data": False,
            },
            # 中断消息的前缀 - 与工具名称和参数组合形成完整消息
            # 例如："Tool execution pending approval: execute_sql with query='DELETE FROM...'"
            # 单个工具可以通过在其中断配置中指定 "description" 来覆盖此设置
            description_prefix="Tool execution pending approval",
        ),
    ],
    # 人机协同需要检查点（checkpointing）来处理中断。
    # 在生产环境中，请使用持久化的检查点保存器，如 AsyncPostgresSaver。
    checkpointer=InMemorySaver(),  
)

您必须配置一个检查点保存器（checkpointer），以便在中断之间持久化图状态。 在生产环境中，请使用持久化的检查点保存器，如 AsyncPostgresSaver。对于测试或原型设计，可以使用 InMemorySaver。

在调用智能体时，传递一个包含线程 ID 的 config，以便将执行与会话线程关联起来。 详情请参阅 LangGraph 中断文档。

配置选项

interrupt_ondictrequired

工具名称到批准配置的映射。值可以是 True（使用默认配置中断）、False（自动批准）或一个 InterruptOnConfig 对象。

description_prefixstringdefault:"Tool execution requires approval"

操作请求描述的前缀

InterruptOnConfig 选项：

allowed_decisionslist[string]

允许的决策列表：'approve'、'edit' 或 'reject'

descriptionstring | callable

用于自定义描述的静态字符串或可调用函数

响应中断

当您调用智能体时，它会一直运行直到完成或引发中断。当工具调用与您在 interrupt_on 中配置的策略匹配时，会触发中断。在这种情况下，调用结果将包含一个 __interrupt__ 字段，其中列出了需要审查的操作。然后，您可以将这些操作呈现给审查者，并在提供决策后恢复执行。

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 
)

# 中断包含完整的 HITL 请求，包括 action_requests 和 review_configs
print(result['__interrupt__'])
# > [
# >    Interrupt(
# >       value={
# >          'action_requests': [
# >             {
# >                'name': 'execute_sql',
# >                'arguments': {'query': 'DELETE FROM records WHERE created_at < NOW() - INTERVAL \'30 days\';'},
# >                'description': 'Tool execution pending approval\n\nTool: execute_sql\nArgs: {...}'
# >             }
# >          ],
# >          'review_configs': [
# >             {
# >                'action_name': 'execute_sql',
# >                'allowed_decisions': ['approve', 'reject']
# >             }
# >          ]
# >       }
# >    )
# > ]

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

决策类型

✅ approve

✏️ edit

❌ reject

使用 approve 按原样批准工具调用并执行，不做更改。

agent.invoke(
    Command(
        # 决策以列表形式提供，每个被审查的操作一个。
        # 决策的顺序必须与 `__interrupt__` 请求中列出的操作顺序匹配。
        resume={
            "decisions": [
                {
                    "type": "approve",
                }
            ]
        }
    ),
    config=config  # 相同的线程 ID 以恢复暂停的对话
)

人机协同流式处理

您可以使用 stream() 代替 invoke()，以便在智能体运行和处理中断时获得实时更新。使用 stream_mode=['updates', 'messages'] 来同时流式传输智能体进度和 LLM 令牌。

from langgraph.types import Command

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

# 流式传输智能体进度和 LLM 令牌直到中断
for mode, chunk in agent.stream(
    {"messages": [{"role": "user", "content": "Delete old records from the database"}]},
    config=config,
    stream_mode=["updates", "messages"],  
):
    if mode == "messages":
        # LLM 令牌
        token, metadata = chunk
        if token.content:
            print(token.content, end="", flush=True)
    elif mode == "updates":
        # 检查中断
        if "__interrupt__" in chunk:
            print(f"\n\nInterrupt: {chunk['__interrupt__']}")

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

:::js

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

// 流式传输智能体进度和 LLM 令牌直到中断
for await (const [mode, chunk] of await agent.stream(
{ messages: [{ role: "user", content: "Delete old records from the database" }] },
{ ...config, streamMode: ["updates", "messages"] }  
)) {
if (mode === "messages") {
// LLM 令牌
const [token, metadata] = chunk;
if (token.content) {
process.stdout.write(token.content);
}
} else if (mode === "updates") {
// 检查中断
if ("__interrupt__" in chunk) {
console.log(`\n\nInterrupt: ${JSON.stringify(chunk