Human in the Loop - Docs by LangChain

Human in the Loop（HITL）中间件允许你为智能体的工具调用添加人工监督。当模型提议一个可能需要审查的操作时——例如写入文件或执行SQL——该中间件可以暂停执行并等待决策。它通过根据可配置的策略检查每个工具调用来实现这一点。如果需要干预，中间件会发出一个中断，暂停执行。图状态使用LangGraph的持久化层保存，因此执行可以安全地暂停并在稍后恢复。然后，人工决策决定下一步操作：操作可以按原样批准（approve）、在运行前修改（edit）、拒绝并附带反馈（reject），或直接响应（respond）用于“询问用户”类型的工具。

中断决策类型

中间件定义了四种内置的人工响应中断的方式：

决策类型	描述	示例用例
✅ `approve`	操作按原样批准并执行，不做更改。	按原样发送电子邮件草稿
✏️ `edit`	工具调用在修改后执行。	在发送电子邮件前更改收件人
❌ `reject`	工具调用被拒绝，并在对话中添加解释。	拒绝电子邮件草稿并解释如何重写它
💬 `respond`	跳过工具执行；人工的消息成为工具结果。	直接回复“ask_user”提示

每个工具可用的决策类型取决于你在 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-5.4",
    tools=[write_file, execute_sql, read_data],
    middleware=[
        HumanInTheLoopMiddleware(
            interrupt_on={
                "write_file": True,  # 允许所有决策（approve, edit, reject, respond）
                "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",
        ),
    ],
    # Human in the Loop需要检查点来处理中断。
    # 在生产环境中，使用持久化检查点，如 AsyncPostgresSaver。
    checkpointer=InMemorySaver(),
)

你必须配置一个检查点以在中断期间持久化图状态。在生产环境中，使用持久化检查点，如 AsyncPostgresSaver。对于测试或原型设计，使用 InMemorySaver。调用智能体时，传递一个包含 thread ID 的 config，以将执行与对话线程关联。详情请参阅 LangGraph 中断文档。

配置选项

interrupt_on

dict

required

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

description_prefix

string

default:"Tool execution requires approval"

操作请求描述的前缀

InterruptOnConfig 选项：

allowed_decisions

list[string]

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

description

string | callable

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

响应中断

当你调用智能体时，它会运行直到完成或触发中断。当中间件匹配到你在 interrupt_on 中配置的策略时，会触发中断。使用 version="v2" 时，结果是一个带有 interrupts 属性的 GraphOutput，其中包含需要审查的操作。然后你可以将这些操作呈现给审查者，并在提供决策后恢复执行。

from langgraph.types import Command

# Human in the Loop利用了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': '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以恢复暂停的对话
    version="v2",
)

决策类型

✅ approve
✏️ edit
❌ reject
💬 respond

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

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

使用 edit 在执行前修改工具调用。提供带有新工具名称和参数的已编辑操作。

agent.invoke(
    Command(
        # 决策以列表形式提供，每个待审查的操作一个。
        # 决策的顺序必须与中断请求中操作的顺序匹配。
        resume={
            "decisions": [
                {
                    "type": "edit",
                    # 带有工具名称和参数的已编辑操作
                    "edited_action": {
                        # 要调用的工具名称。
                        # 通常与原始操作相同。
                        "name": "new_tool_name",
                        # 传递给工具的参数。
                        "args": {"key1": "new_value", "key2": "original_value"},
                    }
                }
            ]
        }
    ),
    config=config,  # 相同的线程ID以恢复暂停的对话
    version="v2",
)

编辑工具参数时，请谨慎进行更改。对原始参数的重大修改可能导致模型重新评估其方法，并可能多次执行工具或采取意外操作。

使用 reject 来拒绝工具调用并提供反馈，而不是执行。

agent.invoke(
    Command(
        # 决策以列表形式提供，每个待审查的操作一个。
        # 决策的顺序必须与中断请求中操作的顺序匹配。
        resume={
            "decisions": [
                {
                    "type": "reject",
                    # 关于操作被拒绝原因的解释
                    "message": "No, this is wrong because ..., instead do this ...",
                }
            ]
        }
    ),
    config=config,  # 相同的线程ID以恢复暂停的对话
    version="v2",
)

message 作为反馈添加到对话中，以帮助智能体理解操作被拒绝的原因以及它应该做什么。

多个决策

当有多个操作待审查时，为每个操作提供一个决策，顺序与它们在中断中出现的顺序相同：

{
    "decisions": [
        {"type": "approve"},
        {
            "type": "edit",
            "edited_action": {
                "name": "tool_name",
                "args": {"param": "new_value"}
            }
        },
        {
            "type": "reject",
            "message": "This action is not allowed"
        }
    ]
}

对于“询问用户”类型的工具，使用 respond，其中工具的真实实现是人工的回复。message 内容直接作为工具结果返回；工具本身不执行。

agent.invoke(
    Command(
        # 决策以列表形式提供，每个待审查的操作一个。
        # 决策的顺序必须与中断请求中操作的顺序匹配。
        resume={
            "decisions": [
                {
                    "type": "respond",
                    # 人工的回复，直接作为工具结果返回
                    "message": "Blue.",
                }
            ]
        }
    ),
    config=config,  # 相同的线程ID以恢复暂停的对话
    version="v2",
)

message 作为成功的 ToolMessage 返回给智能体。当工具有意作为人工输入的占位符时——例如，一个提示澄清的 ask_user 工具——使用 respond。

使用Human in the Loop进行流式传输

你可以使用 stream() 代替 invoke()，在智能体运行和处理中断时获取实时更新。使用 stream_mode=['updates', 'messages'] 和 version="v2" 以统一的v2格式流式传输智能体进度和LLM令牌。

from langgraph.types import Command

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

# 流式传输智能体进度和LLM令牌直到中断
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, 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)

有关流模式的更多详情，请参阅流式传输指南。

执行生命周期

中间件定义了一个 after_model 钩子，该钩子在模型生成响应后但在任何工具调用执行前运行：

智能体调用模型生成响应。
中间件检查响应中的工具调用。
如果任何调用需要人工输入，中间件会构建一个包含 action_requests 和 review_configs 的 HITLRequest，并调用 interrupt。
智能体等待人工决策。
基于 HITLResponse 决策，中间件执行批准或编辑的调用，为拒绝的调用合成 ToolMessage，对于 respond 决策直接将人工回复作为 ToolMessage 返回，并恢复执行。

自定义HITL逻辑

对于更专业的工作流，你可以直接使用 interrupt 原语和 middleware 抽象来构建自定义的HITL逻辑。请查阅上面的执行生命周期以了解如何将中断集成到智能体的操作中。

将这些文档连接到Claude、VSCode等，通过MCP获取实时答案。

在GitHub上编辑此页面或提交问题。

​中断决策类型

​配置中断

​响应中断

​决策类型

​多个决策

​使用Human in the Loop进行流式传输

​执行生命周期

​自定义HITL逻辑

中断决策类型

配置中断

响应中断

决策类型

多个决策

使用Human in the Loop进行流式传输

执行生命周期

自定义HITL逻辑