DeepAgents - Human in the loop

前言

Human in the loop（人机协作）在企业级 Agent 应用中非常告急——AI 在执关键工具时必须颠末人类审批，克制误操纵影响业务。我之前用 LangGraph 0.3 裸写了一套（旧文），当时须要在 tool 函数里手动调 interrupt()，很啰嗦。如今有了 DeepAgents 和内置的 HumanInTheLoopMiddleware，只需设置一个 interrupt_on 字典，制止逻辑全主动——在实行前停息图实行，生存状态到 checkpointer，等待人类决定后规复。
不外官方文档的示例代码比力简朴，只演示了根本用法，没有阐明怎样在真实应用中整合。本文以"答应 AI 实行 shell 下令，但每次实行前需用户确认"为需求，一步步实现完备的人机协作流程。
焦点概念

在开始之前，先理清几个关键概念：
概念阐明interrupt（制止）当 Agent 预备调用某个被监控的 tool 时，HumanInTheLoopMiddleware 调用 LangGraph 的 interrupt() 停息图实行，并抛出包罗 action_requests 和 review_configs 的哀求checkpoint（查抄点）制止时图状态会被长期化。必须设置 checkpointer，否则制止后无法规复。生产环境发起用 AsyncPostgresSaver，测试用 InMemorySaverversion="v2"LangGraph 1.0 的 v2 模式，ainvoke() 返回 GraphOutput 对象（含 .interrupts 属性），astream() 的 updates 流中会出现 __interrupt__ 变乱Command(resume=)用户做出决定后，用 Command(resume={"decisions": [...]}) 从断点规复实行Decision（决定）四种范例：approve（答应）、reject（拒绝并反馈）、edit（修改参数后实行）、respond（人类直接回复，跳过 tool 实行）实行生命周期

1. 用户提问 → Agent 调用 LLM 生成回复
2.   → LLM 决定调用 tool（如 execute_shell_command）
3.     → after_model 钩子：检查 tool 是否在 interrupt_on 中
4.       → 是：构建 HITLRequest → interrupt() → 暂停 ⌛
5.       → 否：继续执行
6.   → 人类做出决策（approve / reject / edit / respond）
7.     → 恢复执行 → 执行/拒绝 tool → LLM 生成最终回复 → 返回

复制代码

流程逻辑

以 Chainlit 谈天应用为交互载体，消息处理惩罚流程如下：

• 用户在谈天页面发送消息（如"查抄下体系负载"）
  
• Agent 调用 LLM 天生复兴，LLM 决定调用 execute_shell_command
  
• HumanInTheLoopMiddleware 检测到该 tool 在 interrupt_on 列表中，触发制止
  
• Chainlit 应用检测到制止，向用户展示审批提示
  
• 用户复兴 答应 / 拒绝，应用用 Command(resume=) 规复实行
  
• Agent 根据决定实行或拒绝 tool，终极返回结果给用户
  

设置制止

起首须要在创建 Agent 时设置 HumanInTheLoopMiddleware：

1. from deepagents import create_deep_agent
2. from langchain.agents.middleware import HumanInTheLoopMiddleware
4. agent = create_deep_agent(
5.     model=llm,
6.     tools=[execute_shell_command],
7.     checkpointer=checkpointer,  # 必须配置！
8.     system_prompt="你是一位智能助手...",
9.     middleware=[
10.         HumanInTheLoopMiddleware(
11.             interrupt_on={
12.                 # 对 execute_shell_command 进行审批
13.                 "execute_shell_command": {
14.                     "allowed_decisions": ["approve", "reject"]
15.                 }
16.             }
17.         ),
18.     ],
19. )

复制代码

interrupt_on 是一个字典，key 为 tool 名称，value 的可设置项：

• True — 答应全部四种决定（approve / edit / reject / respond）
  
• False — 不拦截该 tool（等同于不写）
  
• {"allowed_decisions": [...]} — 只答应指定决定范例
  
• 还可以设置 when 谓词按参数条件判定是否拦截、description 自界说制止提示文本
  

invoke 模式中的实现

v2 模式下的 ainvoke() 返回 GraphOutput 对象，可通过 .interrupts 属性直接获取制止数据，不须要去查 state。
检测制止

1. resp = await agent.ainvoke(
2.     input={"messages": [HumanMessage(content=query)]},
3.     config=config,
4.     version="v2",
5. )
7. if resp.interrupts:
8.     # 存在中断，resp.interrupts 是 Interrupt 对象的元组
9.     interrupt = resp.interrupts[0]
10.     # interrupt.value 是 HITLRequest，包含 action_requests 和 review_configs
11.     print(interrupt.value["action_requests"])

复制代码

规复制止

用户做出决定后，用 Command(resume=) 规复：

1. from langgraph.types import Command
3. await agent.ainvoke(
4.     Command(resume={
5.         "decisions": [{"type": "approve"}]  # 或 {"type": "reject", "message": "..."}
6.     }),
7.     config=config,  # 必须用同一个 thread_id
8.     version="v2",
9. )

复制代码

关键：怎样分辨"新消息"照旧"制止规复"

在谈天应用中，用户发来的每条消息都走同一个 @cl.on_message 处理惩罚函数。用户说"查抄负载"和复兴"答应"都只是文本。办理方法是——调用前先查抄是否有待处理惩罚的制止：

1. # 检查当前会话是否有待处理的中断
2. state = await agent.aget_state(config)
3. if state.next:
4.     # 有待处理中断 → 本次消息是审批回复，构建 resume 命令
5.     cmd = Command(resume={"decisions": [{"type": "approve"}]})
6.     await agent.ainvoke(cmd, config=config, version="v2")
7. else:
8.     # 无中断 → 正常对话
9.     resp = await agent.ainvoke(
10.         {"messages": [HumanMessage(content=query)]}, config=config, version="v2"
11.     )

复制代码

state.next 不为空表现图实行被停息了（有制止等待处理惩罚）。
stream 模式中的实现

流式模式须要用 stream_mode=["messages", "updates"]（官方保举同时开启两种流）：

• messages 流：获取 LLM 的 token 级输出
  
• updates 流：检测制止变乱 __interrupt__
  

1. async for chunk in agent.astream(
2.     input=input_data,
3.     stream_mode=["messages", "updates"],
4.     version="v2",
5.     config=config,
6. ):
7.     if chunk["type"] == "messages":
8.         msg, _meta = chunk["data"]
9.         # msg 是 AIMessageChunk，包含 content 和 tool_calls
10.         if isinstance(msg, AIMessageChunk) and msg.content:
11.             yield extract_text(msg)     # 流式输出文本
12.     elif chunk["type"] == "updates":
13.         if "__interrupt__" in chunk["data"]:
14.             interrupt = chunk["data"]["__interrupt__"][0]
15.             yield format_question(interrupt)  # 输出审批问题

复制代码

stream 模式的规复与 invoke 类似——在调用 astream() 之前同样要先查抄 state.next 来判定是正常对话照旧制止规复。
完备示例

下面是焦点代码。checkpointer 和 llm 的设置函数、日志模块等非焦点代码省略。

Agent 封装（internal/agent/agent.py 焦点部分）

1. # --- 审批关键词匹配 ---
2. _APPROVE_KEYWORDS = frozenset(
3.     {"yes", "accept", "approve", "ok", "是", "允许", "同意", "批准"}
4. )
6. def _parse_decision(query: str) -> str:
7.     return "approve" if query.strip().lower() in _APPROVE_KEYWORDS else "reject"
9. def _build_resume_command(decision_type: str, actions_count: int) -> Command:
10.     item = {"type": decision_type}
11.     if decision_type == "reject":
12.         item["message"] = "user rejected this action"
13.     return Command(resume={"decisions": [item for _ in range(actions_count)]})
15. def _extract_text(message) -> str:
16.     """从消息中提取纯文本（兼容 str 和 list[dict] 两种 content 格式）。"""
17.     if not message or not hasattr(message, "content"):
18.         return ""
19.     content = message.content
20.     if isinstance(content, str):
21.         return content
22.     if isinstance(content, list):
23.         return "".join(
24.             b.get("text", "")
25.             for b in content
26.             if isinstance(b, dict) and b.get("type") == "text"
27.         )
28.     return ""
30. def _format_interrupt_question(interrupt) -> str:
31.     """将中断数据格式化为用户的审批问题。"""
32.     action_requests = interrupt.value.get("action_requests", [])
33.     review_configs = interrupt.value.get("review_configs", [])
34.     allowed = (
35.         review_configs[0].get("allowed_decisions", ["approve", "reject"])
36.         if review_configs
37.         else ["approve", "reject"]
38.     )
40.     lines = []
41.     for req in action_requests:
42.         lines.append(
43.             "Do you approve me to execute this action?\n\n"
44.             f"- name: {req['name']}\n"
45.             f"- args: `{req['args']}`\n"
46.         )
47.     lines.append(f"Input your decision: {', '.join(allowed)}\n")
48.     return "\n".join(lines)
50. class AIAgent:
51.     # ... __init__, _init_deep_agent, _init_tools 省略 ...
53.     async def _has_pending_interrupt(self, config: RunnableConfig) -> bool:
54.         state = await self._agent.aget_state(config)
55.         return bool(state.next)
57.     # --- invoke 模式 ---
58.     async def ainvoke(self, query: str, config: RunnableConfig) -> str:
59.         if not self._agent:
60.             await self._init_deep_agent()
62.         # 优先处理中断恢复
63.         if await self._has_pending_interrupt(config):
64.             state = await self._agent.aget_state(config)
65.             actions_count = len(
66.                 state.interrupts[0].value["action_requests"]
67.             )
68.             decision = _parse_decision(query)
69.             cmd = _build_resume_command(decision, actions_count)
70.             await self._agent.ainvoke(cmd, config=config, version="v2")
72.             # 恢复后取最新消息
73.             state = await self._agent.aget_state(config)
74.             if state.values and "messages" in state.values:
75.                 return _extract_text(state.values["messages"][-1])
76.             return "Oops, something went wrong."
78.         # 正常对话
79.         resp = await self._agent.ainvoke(
80.             input={"messages": [HumanMessage(content=query)]},
81.             config=config,
82.             version="v2",
83.         )
84.         if resp.interrupts:
85.             return _format_interrupt_question(resp.interrupts[0])
86.         return _extract_text(resp.value["messages"][-1])
88.     # --- stream 模式 ---
89.     async def astream(self, query: str, config: RunnableConfig):
90.         if not self._agent:
91.             await self._init_deep_agent()
93.         # 判断是中断恢复还是正常对话
94.         state = await self._agent.aget_state(config)
95.         if state.next:
96.             actions_count = len(
97.                 state.interrupts[0].value["action_requests"]
98.             )
99.             decision = _parse_decision(query)
100.             input_data = _build_resume_command(decision, actions_count)
101.         else:
102.             input_data = {"messages": [HumanMessage(content=query)]}
104.         async for chunk in self._agent.astream(
105.             input=input_data,
106.             stream_mode=["messages", "updates"],
107.             version="v2",
108.             config=config,
109.         ):
110.             if chunk["type"] == "messages":
111.                 msg, _meta = chunk["data"]
112.                 if isinstance(msg, AIMessageChunk) and msg.content:
113.                     yield _extract_text(msg)
114.             elif chunk["type"] == "updates" and "__interrupt__" in chunk["data"]:
115.                 yield _format_interrupt_question(
116.                     chunk["data"]["__interrupt__"][0]
117.                 )

复制代码

Chainlit 应用层（chainlit_app.py 焦点部分）

1. @cl.on_message
2. async def main(msg: cl.Message):
3.     config = RunnableConfig(
4.         configurable={"thread_id": cl.context.session.id},
5.     )
7.     # stream 模式（推荐）
8.     final_answer = cl.Message(content="")
9.     async for chunk in ai_agent.astream(msg.content, config=config):
10.         await final_answer.stream_token(chunk)
11.     await final_answer.send()
13.     # 或者 invoke 模式
14.     # resp = await ai_agent.ainvoke(msg.content, config)
15.     # await cl.Message(resp).send()

复制代码

Chainlit 应用层的代码非常简便——由于制止检测和规复逻辑全部封装在 AIAgent 内部了。Chainlit 只须要流式输出 astream() / ainvoke() 的返回结果即可。
交互结果

1. [用户]: 检查下系统负载
3. [AI]: 🔧 正在调用工具: execute_shell_command...
5. [AI]: Do you approve me to execute this action?
7.        - name: execute_shell_command
8.        - args: `{"command": "cat /proc/loadavg && free -h", "timeout": 10}`
10.        Input your decision: approve, reject
12. [用户]: 批准
14. [AI]: 当前系统负载: 0.52 0.38 0.25，
15.        内存总容量 62Gi，已用 10Gi，剩余 46Gi，系统运行正常。

复制代码

进阶：利用 interrupt_on 的 when 谓词

假如不想拦截全部 shell 下令，只想拦截伤害操纵（如 rm、dd、写入体系目次等），可以用 when 谓词按参数条件判定：

1. from langgraph.prebuilt.tool_node import ToolCallRequest
3. def is_dangerous_command(request: ToolCallRequest) -> bool:
4.     """只拦截包含危险操作的命令。"""
5.     command = request.tool_call["args"].get("command", "")
6.     dangerous = {"rm ", "dd ", "mkfs", "shutdown", "reboot"}
7.     return any(d in command for d in dangerous)
9. HumanInTheLoopMiddleware(
10.     interrupt_on={
11.         "execute_shell_command": {
12.             "allowed_decisions": ["approve", "reject"],
13.             "when": is_dangerous_command,   # 只在危险命令时拦截
14.         }
15.     }
16. )

复制代码

when 谓词返回 True 才触发制止，返回 False 则主动答应。留意 when 须要 langchain >= 1.3.3。
改进点

• 如今 reject 时用的是固定消息，现实产物中可以让用户输入拒绝缘故起因，方便 LLM 调解后续举动
  
• 审批提示如今是纯文本，可以用 Chainlit 的 AskActionMessage 做成按钮交互（不外受制于 Chainlit Action 的 payload 范例限定，须要额外处理惩罚）
  
• 假如有多个 tool 同时被拦截，action_requests 列表中会有多项，本文为简化只取了第一个，生产环境应遍历处理惩罚
  

增补

DeepAgents 的 HumanInTheLoopMiddleware 把之前须要手写的制止逻辑全部封装好了。集成到真实应用的关键只有三步：

• 创建 Agent 时：设置 interrupt_on 字典 + 确保有 checkpointer
  
• 每次调用前：通过 state.next 判定是正常对话照旧制止规复
  
• 规复时：用 Command(resume={"decisions": [...]}) 传入用户决定
  

ainvoke 和 astream 两种模式的焦点逻辑同等，只是检测制止的方式差别（.interrupts 属性 vs updates 流中的 __interrupt__）。