LangChain / Advanced usage / Human-in-the-loop-平芜编程栈

原文链接：https://docs.langchain.com/oss/python/langchain/human-in-the-loop

高级用法

人机协同

复制页面

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

它的工作方式是针对配置的策略检查每个工具调用。如果需要人工介入，中间件会发出一个中断来暂停执行。图状态会使用LangGraph的持久化层保存，因此执行可以安全地暂停并在稍后恢复。

随后由人工决策决定下一步操作：可以按原样批准操作（approve）、在运行前修改（edit）、附带反馈拒绝（reject），或直接回复（respond）——适用于“询问用户”类型的工具。

中断决策类型

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

决策类型	描述	示例用例
✅ 批准	按智能体提议的原始参数执行工具。	按原样发送邮件草稿
✏️ 编辑	在执行前修改工具参数。	发送邮件前更改收件人
❌ 拒绝	完全跳过该工具调用，并向智能体返回拒绝反馈。	拒绝删除文件并说明原因
💬 回复	将人工消息直接作为合成工具结果返回，跳过执行，适用于“询问用户”类型的工具。	用直接回复回答"ask_user"提示

每个工具可用的决策类型取决于您在interrupt_on中配置的策略。当多个工具调用同时被暂停时，每个操作都需要单独的决策。决策必须按照中断请求中操作出现的顺序提供。

当人工拒绝请求的操作时使用reject。仅当人工充当工具角色时（例如回答ask_user提示）才使用respond。不要使用respond来拒绝有副作用的工具，因为其消息会被视为成功的工具结果。

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

配置中断

要使用HITL，请在创建智能体时将中间件添加到智能体的中间件列表中。

配置时使用工具操作到允许的决策类型的映射。当工具调用匹配映射中的操作时，中间件将中断执行。

fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportHumanInTheLoopMiddlewarefromlanggraph.checkpoint.memoryimportInMemorySaver agent=create_agent(model="gpt-5.5",tools=[write_file,execute_sql,read_data],middleware=[HumanInTheLoopMiddleware(interrupt_on={"write_file":True,# 允许所有决策（批准、编辑、拒绝、回复）"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",),],# 人机协同需要检查点机制来处理中断。# 生产环境中，请使用持久化检查点器，如AsyncPostgresSaver或MongoDBSaver。checkpointer=InMemorySaver(),)

您必须配置检查点器来跨中断持久化图状态。在生产环境中，请使用持久化检查点器，如AsyncPostgresSaver或MongoDBSaver。对于测试或原型开发，可使用InMemorySaver。

调用智能体时，传入包含线程ID的配置，以将执行与对话线程关联。详情请参阅LangGraph中断文档。

配置选项

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

description_prefix
string，默认值："Tool execution requires approval"
操作请求描述的前缀

InterruptOnConfig 选项：

allowed_decisions
list[string]
允许的决策列表：'approve'（批准）、'edit'（编辑）、'reject'（拒绝）或'respond'（回复）

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

when
callable
可选谓词，接收一个ToolCallRequest对象，返回True则中断，返回False则自动批准。用于根据调用的参数进行门控中断。需要langchain>=1.3.3。

条件中断

默认情况下，interrupt_on中列出的每个工具调用都会暂停以供审核。要为部分调用添加条件暂停，请为工具的InterruptOnConfig添加when谓词。该谓词接收ToolCallRequest，返回True则中断，返回False则自动批准，因此您可以基于工具参数进行门控。

条件中断需要langchain>=1.3.3。

fromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportHumanInTheLoopMiddleware,ToolCallRequestfromlanggraph.checkpoint.memoryimportInMemorySaverdefwrites_outside_workspace(request:ToolCallRequest)->bool:"""暂停写入工作区目录之外路径的操作。"""path=request.tool_call["args"].get("path","")returnnotpath.startswith("/workspace/")defis_write_query(request:ToolCallRequest)->bool:"""暂停非只读SELECT的SQL操作。"""query=request.tool_call["args"].get("query","")returnnotquery.lstrip().upper().startswith("SELECT")agent=create_agent(model="gpt-5.5",tools=[write_file,execute_sql,read_data],middleware=[HumanInTheLoopMiddleware(interrupt_on={"write_file":{"allowed_decisions":["approve","edit","reject"],"when":writes_outside_workspace,},"execute_sql":{"allowed_decisions":["approve","reject"],"when":is_write_query,},},),],checkpointer=InMemorySaver(),)

当when谓词返回False时，调用会直接运行而不会中断。当返回True时，或省略when时，调用会照常暂停。返回False的调用永远不会被添加到中断批次中，因此审核者只会看到需要决策的操作。

响应中断

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

fromlanggraph.typesimportCommand# 人机协同利用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的GraphOutputprint(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可按原样批准工具调用，并在不做任何更改的情况下执行它。

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":"User rejected this action. Do not retry this tool call.",}]}),config=config,# 使用相同的线程ID恢复暂停的对话version="v2",)

该消息会作为反馈添加到对话中，帮助智能体理解操作被拒绝的原因以及它应该采取什么替代方案。当您省略message时，中间件会使用默认的拒绝消息，告诉模型该工具未被执行，并且除非用户要求，否则不要重试相同的工具调用。对于有副作用的工具，请提供领域特定的消息，明确说明智能体应该放弃该操作、提出后续问题，还是尝试更安全的替代方案。

💬 回复

使用respond适用于“询问用户”类型的工具，这类工具的实际实现就是人工的回复。消息内容会直接作为工具结果返回；工具本身不会被执行。

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

该消息会作为成功的ToolMessage返回给智能体。当工具故意作为人工输入的占位符时，请使用respond，例如用于提示澄清的ask_user工具。不要使用respond来拒绝提议的操作，因为它会告诉模型该工具已成功完成。

多个决策

当多个操作正在审核中时，为每个操作按它们在中断中出现的顺序提供决策：

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

人机协同流式传输

您可以在智能体运行和处理中断时使用stream_events()流式传输实时更新。使用stream.messages流式传输LLM令牌，使用stream.values检查智能体状态快照以获取中断信息。

fromlanggraph.typesimportCommand config={"configurable":{"thread_id":"some_id"}}# 流式传输智能体进度和LLM令牌直到中断stream=agent.stream_events({"messages":[{"role":"user","content":"Delete old records from the database"}]},config=config,version="v3",)formessageinstream.messages:fortokeninmessage.text:print(token,end="",flush=True)# 检查运行是否暂停等待人工输入ifstream.interrupted:print(f"\n\nInterrupt:{stream.interrupts}")# 人工决策后恢复并流式传输stream=agent.stream_events(Command(resume={"decisions":[{"type":"approve"}]}),config=config,version="v3",)formessageinstream.messages:fortokeninmessage.text:print(token,end="",flush=True)

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

执行生命周期

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

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