关键操作前，让人点个头：LangChain Human-in-the-Loop 实战

Agent 能自己调用工具固然强大，但有些操作你绝不想让它"自作主张"：删库、转账、对外发邮件、改生产配置……一旦做错，代价是真金白银。

Human-in-the-Loop（HITL）中间件就是给这类操作加一道"人工闸门"：当模型打算执行敏感工具时，暂停下来，把这个动作交给人审批——批准、修改、拒绝，或直接替工具作答。这篇文章基于 LangChain 官方 HITL 文档，把它讲透。

一、它解决什么问题，怎么工作

HITL 中间件会逐个检查模型发起的工具调用，对照你配置的策略。若某个调用需要人工介入，它就发起一个interrupt（中断）：

模型生成回复 → 中间件检查工具调用 → 命中策略？ ├─ 是 → interrupt 暂停，存档状态，等人决策 └─ 否 → 直接执行

关键点：中断时整个图的状态会被 LangGraph 的持久化层保存下来，所以可以安全地暂停、过几分钟甚至几天后再恢复。这也是为什么HITL 必须配 checkpointer + thread_id。

四种人工决策

人收到中断后，有四种回应方式：

⚠️ 区分reject和respond：拒绝一个有副作用的动作用reject；只有当"人就是这个工具的实现"（如 ask_user 让用户澄清）时才用respond——因为respond的消息会被当成成功的工具结果，拿它去否决副作用工具会让模型误以为操作成功了。

公共：从 .env 初始化模型

下文示例共用这段（从.env读MODEL_NAME/OPENAI_API_BASE/OPENAI_API_KEY）：

importosfromdotenvimportload_dotenvfromlangchain.chat_modelsimportinit_chat_model load_dotenv()model=init_chat_model(os.getenv("MODEL_NAME","glm-5.1"),model_provider="openai",base_url=os.getenv("OPENAI_API_BASE"),api_key=os.getenv("OPENAI_API_KEY"),streaming=True,)

公共：示例工具（最小实现）

下文的write_file/execute_sql/read_data都是占位演示工具，最小实现如下（实际项目里换成你真正的实现即可）。注意参数名要和后面when谓词读取的一致（write_file用path、execute_sql用query）：

fromlangchain.toolsimporttool@tooldefwrite_file(path:str,content:str)->str:"""把内容写入指定文件路径。"""# 实际项目：with open(path, "w") as f: f.write(content)returnf"已写入{path}（{len(content)}字符）"@tooldefexecute_sql(query:str)->str:"""执行一条 SQL 语句。"""# 实际项目：连接数据库执行returnf"已执行 SQL：{query}"@tooldefread_data(table:str)->str:"""读取指定表的数据（只读、安全操作）。"""returnf"{table}的数据：[模拟若干行]"

二、配置中断：interrupt_on

把HumanInTheLoopMiddleware加进middleware，用interrupt_on配一张"工具 → 是否/如何审批"的映射表。值有三种：

• True—— 中断，允许全部四种决策
• False—— 自动放行（安全操作）
• InterruptOnConfig（dict）—— 精细控制，比如只允许某几种决策

importosfromdotenvimportload_dotenvfromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportHumanInTheLoopMiddlewarefromlangchain.chat_modelsimportinit_chat_modelfromlanggraph.checkpoint.memoryimportInMemorySaver load_dotenv()model=init_chat_model(os.getenv("MODEL_NAME","glm-5.1"),model_provider="openai",base_url=os.getenv("OPENAI_API_BASE"),api_key=os.getenv("OPENAI_API_KEY"),streaming=True,)agent=create_agent(model=model,tools=[write_file,execute_sql,read_data],middleware=[HumanInTheLoopMiddleware(interrupt_on={"write_file":True,# 允许全部决策"execute_sql":{"allowed_decisions":["approve","reject"]},# 不允许 edit"read_data":False,# 安全，自动放行},description_prefix="工具执行待审批",# 中断描述前缀),],# HITL 依赖检查点持久化状态；生产用 PostgresSaver，测试用 InMemorySavercheckpointer=InMemorySaver(),)

必须配checkpointer，并在 invoke 时传带thread_id的config，把这次执行关联到一个会话线程，才能暂停后恢复。

三、条件中断：只拦该拦的（when）

默认interrupt_on里列的工具每次调用都暂停。但很多时候你只想拦"危险的那部分"——比如只拦写库 SQL、放行只读 SELECT。给InterruptOnConfig加一个when谓词即可，它收到ToolCallRequest、返回True才中断。

importosfromdotenvimportload_dotenvfromlangchain.agentsimportcreate_agentfromlangchain.agents.middlewareimportHumanInTheLoopMiddleware,ToolCallRequestfromlangchain.chat_modelsimportinit_chat_modelfromlanggraph.checkpoint.memoryimportInMemorySaver load_dotenv()model=init_chat_model(os.getenv("MODEL_NAME","glm-5.1"),model_provider="openai",base_url=os.getenv("OPENAI_API_BASE"),api_key=os.getenv("OPENAI_API_KEY"),streaming=True,)defwrites_outside_workspace(request:ToolCallRequest)->bool:"""只拦写到 /workspace/ 之外的路径。"""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=model,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的调用根本不会进审批队列——审批者只看到真正需要决策的动作，不被只读查询刷屏。

条件中断需要langchain>=1.3.3。

四、响应中断：暂停 → 审批 → 恢复

invoke 时加version="v2"，Agent 跑到中断点会返回一个带.interrupts的结果。你把待审批动作拿给人看，拿到决策后再用Command(resume=...)恢复。

fromlanggraph.typesimportCommand config={"configurable":{"thread_id":"some_id"}}# HITL 必须有 thread_id# 1. 跑到中断点result=agent.invoke({"messages":[{"role":"user","content":"删除数据库里的旧记录"}]},config=config,version="v2",)# result 是 GraphOutput，.interrupts 里是待审批的动作print(result.interrupts)# (Interrupt(value={# 'action_requests': [{'name': 'execute_sql',# 'arguments': {'query': "DELETE FROM records WHERE ..."},# 'description': '工具执行待审批\n\nTool: execute_sql\nArgs: {...}'}],# 'review_configs': [{'action_name': 'execute_sql',# 'allowed_decisions': ['approve', 'reject']}]# }),)# 2. 人决策后，用同一个 thread_id 恢复agent.invoke(Command(resume={"decisions":[{"type":"approve"}]}),# 或 "reject"config=config,version="v2",)

四种决策怎么写

✅ approve —— 原样执行

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

✏️ edit —— 改参数再执行（给出新的工具名和参数）

agent.invoke(Command(resume={"decisions":[{"type":"edit","edited_action":{"name":"execute_sql",# 通常和原来一样"args":{"query":"DELETE FROM records WHERE created_at < NOW() - INTERVAL '90 days'"},},}]}),config=config,version="v2")

改参数要保守。大改动可能让模型重新评估、甚至重复执行工具或做出意外动作。

❌ reject —— 拒绝并反馈（工具不执行，message 进对话）

agent.invoke(Command(resume={"decisions":[{"type":"reject","message":"用户拒绝了该操作，不要重试这个工具调用。",# 可选，省略则用默认拒绝语}]}),config=config,version="v2")

message会作为反馈加进对话，告诉模型为什么被拒、接下来该怎么做。对有副作用的工具，最好写明确：是放弃、还是换个更安全的方案。

💬 respond —— 人替工具作答（仅用于 ask_user 这类"问用户"工具）

agent.invoke(Command(resume={"decisions":[{"type":"respond","message":"蓝色。",# 人的回复，直接作为工具结果返回}]}),config=config,version="v2")

message会被当成一条成功的 ToolMessage返回给模型。再强调一次：别拿respond去否决副作用工具。

多个动作：决策按顺序一一对应

如果同时有多个工具调用被暂停，每个动作给一个决策，顺序要和中断里动作出现的顺序一致：

{"decisions":[{"type":"approve"},{"type":"edit","edited_action":{"name":"tool_name","args":{"param":"new_value"}}},{"type":"reject","message":"这个操作不允许"},]}

五、边流式边审批

想在 Agent 运行时实时看 token、并捕获中断，用stream_events()（version="v3"）：

fromlanggraph.typesimportCommand config={"configurable":{"thread_id":"some_id"}}# 流式跑到中断stream=agent.stream_events({"messages":[{"role":"user","content":"删除数据库里的旧记录"}]},config=config,version="v3",)formessageinstream.messages:# 流式 LLM tokenfortokeninmessage.text:print(token,end="",flush=True)ifstream.interrupted:# 是否暂停等待人工print(f"\n\n中断：{stream.interrupts}")# 人决策后，继续流式恢复stream=agent.stream_events(Command(resume={"decisions":[{"type":"approve"}]}),config=config,version="v3",)formessageinstream.messages:fortokeninmessage.text:print(token,end="",flush=True)

stream.messages流 token，stream.interrupted/stream.interrupts检查是否需要人工。

六、执行生命周期

HITL 中间件本质是一个after_model钩子——在模型生成回复后、工具执行前介入：

1. Agent 调模型生成回复；
2. 中间件检查回复里的工具调用；
3. 若有调用需要人工，构造HITLRequest（含action_requests和review_configs）并发起interrupt；
4. Agent 暂停，等人决策；
5. 按决策处理：执行 approve/edit 的调用、给 reject 的调用合成一条 ToolMessage 反馈、把 respond 的回复直接作为 ToolMessage 返回，然后恢复执行。

七、小结

落地三件套：

1. 配interrupt_on：哪些工具拦（True/InterruptOnConfig）、哪些放行（False）；只想拦危险参数就加when。
2. 配checkpointer+thread_id：中断要靠持久化暂存状态，缺一不可。
3. 用version="v2"+Command(resume=...)：跑到中断 → 取result.interrupts给人看 → 决策回传恢复。

HITL 的价值在于：把"高风险动作"从全自动改成"人确认"，用一点点延迟换巨大的安全边界。删库、转账、对外通信这类操作，值得这道闸门。