DeepAgents 之interrupt_on人工审批-编程阁

一、30 秒看懂 interrupt_on

interrupt_on = 「哪些工具调用必须先经过人工审批」的配置表 Agent 准备调用敏感工具 → 暂停（interrupt）→ 等人 approve/edit/reject/respond → 用 Command(resume=...) 恢复 → Agent 继续跑

类比：

角色	对应
Agent	想执行操作的员工
`interrupt_on`	敏感操作清单
人工审批	主管点头 / 改参数 / 拒绝
`Command(resume=...)`	把审批结果交回系统

二、核心概念

2.1 interrupt_on 是什么

interrupt_on是create_deep_agent的一个参数：

interrupt_on:dict[str,bool|InterruptOnConfig]|None=None

含义：工具名 → 是否/如何中断的映射。

配置了 interrupt_on	框架自动做什么
非空（或与 permissions interrupt 合并后非空）	挂载`HumanInTheLoopMiddleware`
工具被调用且命中规则	图执行暂停，返回`result.interrupts`
用户`Command(resume=...)`	按决策执行/跳过/改参，Agent 继续

2.2 执行流程

用户提问 │ ▼ Agent 推理 → 决定调用 write_file │ ▼ interrupt_on 命中？ ──否──▶ 直接执行工具 │ 是 ▼ 暂停，返回 interrupts（含 tool 名、参数、允许的决策） │ ▼ 人工 approve / edit / reject / respond │ ▼ agent.invoke(Command(resume={"decisions": [...]}), 同一 config) │ ▼ Agent 继续（可能再次 interrupt，直到结束）

2.3 必备前置条件

条件	是否必须	说明
`checkpointer`	✅ 必须	保存暂停时的图状态，否则无法 resume
`thread_id`	✅ 必须	同一对话线程，resume 时要与首次 invoke 相同
`version="v2"`	✅ 推荐	读取`result.interrupts`、resume 的标准写法
`HumanInTheLoopMiddleware`	自动	有 interrupt_on 时框架自动安装，无需手写

三、易混淆对比（重点）

3.1 interrupt_on vs permissions mode=“interrupt”

两者都能「暂停等人」，但入口和粒度不同：

对比项	`interrupt_on`	`permissions`+`mode="interrupt"`
配置方式	按工具名配置	按路径 + 操作类型配置
作用对象	任意工具（自定义 + 内置）	仅内置文件系统工具
粒度	整个工具（可用`when`细化）	路径级（`/secrets/**`等）
HITL 中间件	手动配或自动（有规则时）	有 interrupt 规则时自动合并进 interrupt_on
典型场景	删库、发邮件、调支付 API	敏感目录写入要审批
能否同时用	✅ 可以，用户 interrupt_on 同工具名优先	✅ 合并到同一 HITL 配置

# 方式 A：直接按工具名interrupt_on={"write_file":True}# 方式 B：按路径（框架自动转成 write_file/edit_file 的 when 谓词）permissions=[FilesystemPermission(operations=["write"],paths=["/secrets/**"],mode="interrupt")]# 方式 C：两者并存，同工具名时 interrupt_on 参数覆盖 permissions 生成的配置

3.2 approve / edit / reject / respond

决策	含义	工具是否执行	典型用途
`approve`	原样执行	✅ 执行	确认 Agent 提案
`edit`	改参数后执行	✅ 执行（改过的 args）	改收件人、改路径
`reject`	拒绝执行	❌ 不执行，返回拒绝消息给 Agent	禁止删文件、禁止发信
`respond`	人直接充当工具返回结果	❌ 不执行，把人说的话当 ToolMessage	`ask_user`类工具

易错点：

错误用法	正确理解
对`delete_file`用`respond`表示「拒绝」	拒绝用`reject`；`respond`是「人代替工具回答」
`reject`不写`message`	可以，但有默认文案；敏感操作建议写清楚「勿重试」

3.3 interrupt_on 的 True / False / InterruptOnConfig

配置值	含义
`True`	开启中断；默认允许 approve、edit、reject、respond
`False`	显式关闭该工具的中断（即使父级/permissions 有规则）
`{"allowed_decisions": [...]}`	自定义允许哪些决策
`{"allowed_decisions": [...], "when": fn}`	自定义决策 + 条件中断

interrupt_on={"remove_file":True,# 全开"fetch_file":False,# 不关心中断"notify_email":{"allowed_decisions":["approve","reject"]},# 不许改参数}

3.4 单工具中断 vs 批量中断

场景	行为
Agent 一次只调 1 个敏感工具	`action_requests`长度 = 1
Agent 一次调多个敏感工具	合并为一次 interrupt，`action_requests`有多项
resume 时	`decisions`列表顺序必须与 action_requests 一致

3.5 interrupt_on vs 工具内 interrupt()

对比项	`interrupt_on`	工具内`interrupt()`
配置位置	`create_deep_agent(interrupt_on=...)`	自定义工具函数内部
触发点	工具即将被调用前	工具执行过程中任意位置
数据结构	`action_requests`+`review_configs`	自定义 payload
resume 格式	`Command(resume={"decisions": [...]})`	`Command(resume={...})`自定义
适用	标准 HITL 审批流	工具内部多步确认

本文聚焦interrupt_on；工具内interrupt()属于进阶用法。

3.6 主智能体 vs 子智能体 interrupt_on

子智能体类型	是否继承主 agent 的 interrupt_on
声明式`SubAgent`字典	✅ 默认继承；子 agent 自己的`interrupt_on`完全覆盖
`CompiledSubAgent`	❌ 不继承，在子图内部自己配
`AsyncSubAgent`	❌ 不继承，在远程子 agent 配

四、interrupt_on 配置详解

4.1 参数签名

agent=create_deep_agent(model="qwen-plus",tools=[my_tool,...],interrupt_on={"tool_name":True|False|InterruptOnConfig,},checkpointer=InMemorySaver(),# 必须)

4.2 InterruptOnConfig 字段

字段	类型	说明
`allowed_decisions`	`list[str]`	可选：`approve`/`edit`/`reject`/`respond`
`when`	`Callable[[ToolCallRequest], bool]`	返回`True`才中断；`False`自动放行

4.3 与 Middleware 的关系

interrupt_on 非空（或与 permissions interrupt 合并后非空） ↓ 自动追加 HumanInTheLoopMiddleware（栈尾部） ↓ 工具调用前检查是否 interrupt ↓ 若中断后未 resume，PatchToolCallsMiddleware 会在下次启动前修复悬空 tool_call

组件	角色
`HumanInTheLoopMiddleware`	执行 interrupt / resume 逻辑
`PatchToolCallsMiddleware`	中断取消后修补消息历史

五、四种人工决策类型

5.1 approve —— 原样批准

decision={"type":"approve"}

工具按 Agent 原始参数执行。

5.2 edit —— 改参数后执行

decision={"type":"edit","edited_action":{"name":"notify_email",# 必须带工具名"args":{"to":"team@corp.com","subject":"...","body":"..."},},}

allowed_decisions里必须包含"edit"。

5.3 reject —— 拒绝执行

decision={"type":"reject","message":"用户拒绝删除该文件，请勿重试，改问用户要归档哪个目录。",}

Agent 收到的是拒绝反馈，不会执行工具。

5.4 respond —— 人代替工具回答

decision={"type":"respond","message":"用户说：先用测试环境，不要上生产。",}

适合「问用户」类工具；不要用来拒绝有副作用的操作。

5.5 决策配置示例对照

工具风险	推荐 allowed_decisions
删除、支付、发外部邮件	`["approve", "edit", "reject"]`
中等风险写操作	`["approve", "reject"]`
只读确认类	`["approve"]`或`["approve", "reject"]`
向用户提问	可含`"respond"`

六、条件中断 when 谓词

默认：interrupt_on里列出的工具每次调用都暂停。

加when后：只有谓词返回 True 才暂停。

fromlangchain.tools.tool_nodeimportToolCallRequestdefwrites_outside_workspace(request:ToolCallRequest)->bool:"""只有写到 /workspace/ 外才 interrupt。"""path=request.tool_call["args"].get("file_path","")returnnotpath.startswith("/workspace/")agent=create_deep_agent(model="qwen-plus",interrupt_on={"write_file":{"allowed_decisions":["approve","edit","reject"],"when":writes_outside_workspace,},},checkpointer=InMemorySaver(),)

when 返回值	行为
`True`	暂停，等人审批
`False`	不中断，直接执行
省略`when`	每次调用都中断

注意：request.tool_call["name"]/request.tool_call["args"]是标准取法（与wrap_tool_call相同）。

七、中断后的处理流程

7.1 标准两步法

fromuuidimportuuid4fromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.typesimportCommand config={"configurable":{"thread_id":str(uuid4())}}# ① 首次调用result=agent.invoke({"messages":[{"role":"user","content":"删除 temp.txt"}]},config=config,version="v2",)# ② 检查中断ifresult.interrupts:iv=result.interrupts[0].value action=iv["action_requests"][0]review={c["action_name"]:cforciniv["review_configs"]}print("待审批工具:",action["name"])print("参数:",action["args"])print("允许决策:",review[action["name"]]["allowed_decisions"])# ③ 人工决策后恢复（config 必须相同！）result=agent.invoke(Command(resume={"decisions":[{"type":"approve"}]}),config=config,version="v2",)print(result.value["messages"][-1].content)

7.2 result.interrupts 结构

字段	含义
`result.interrupts`	中断列表；非空表示暂停
`interrupts[0].value["action_requests"]`	待审批的工具调用列表
`interrupts[0].value["review_configs"]`	每个工具允许的决策类型
`action_requests[i]["name"]`	工具名
`action_requests[i]["args"]`	工具参数

7.3 多工具批量审批

# Agent 同时想 delete + send_email，两个都要审批decisions=[{"type":"approve"},# 对应 action_requests[0]{"type":"reject","message":"禁止发邮件，请勿重试"},# 对应 action_requests[1]]result=agent.invoke(Command(resume={"decisions":decisions}),config=config,version="v2")

规则：len(decisions) == len(action_requests)，且顺序一一对应。

7.4 流程图

invoke(用户消息) │ ├─ result.interrupts 为空 → 直接拿 result.value["messages"][-1] │ └─ result.interrupts 非空 │ ├─ 解析 action_requests ├─ 收集用户 decisions └─ invoke(Command(resume={"decisions": [...]}), 同一 config) │ └─ 可能再次 interrupt，循环直到结束

八、子智能体与 interrupt_on

8.1 SubAgent 覆盖主 agent 配置

agent=create_deep_agent(model="qwen-plus",interrupt_on={"delete_file":True,"read_file":False,# 主 agent：读文件不需审批},subagents=[{"name":"file-manager","description":"文件管理","system_prompt":"你是文件管理员","interrupt_on":{"delete_file":True,"read_file":True,# 子 agent：读也要审批（覆盖继承）},}],checkpointer=InMemorySaver(),)

8.2 继承规则表

场景	interrupt_on 来源
SubAgent 未配`interrupt_on`	继承主 agent + permissions 合并结果
SubAgent 配了`interrupt_on`	完全使用子 agent 自己的，不合并主 agent
general-purpose 子智能体	继承主 agent 的 interrupt_on

子 agent 触发 interrupt 时，处理方式与主 agent 相同：检查result.interrupts，Command(resume=...)恢复。

九、实战示例

示例 1：最小可运行（自定义工具 + interrupt_on）

importosfromuuidimportuuid4fromdeepagentsimportcreate_deep_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.toolsimporttoolfromlanggraph.checkpoint.memoryimportInMemorySaverfromlanggraph.typesimportCommand model=init_chat_model(model="qwen-plus",model_provider="openai",api_key=os.getenv("ali_api_key"),base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",temperature=0,)@tooldefdelete_file(path:str)->str:"""删除文件。"""returnf"已删除{path}"agent=create_deep_agent(model=model,tools=[delete_file],interrupt_on={"delete_file":True},checkpointer=InMemorySaver(),)config={"configurable":{"thread_id":str(uuid4())}}result=agent.invoke({"messages":[{"role":"user","content":"删除 /tmp/a.txt"}]},config=config,version="v2",)ifresult.interrupts:action=result.interrupts[0].value["action_requests"][0]print(f"待审批:{action['name']}{action['args']}")ok=input("批准？y/n：").strip().lower()=="y"decision={"type":"approve"}ifokelse{"type":"reject","message":"用户拒绝删除"}result=agent.invoke(Command(resume={"decisions":[decision]}),config=config,version="v2")print(result.value["messages"][-1].content)

示例 2：限制 allowed_decisions

agent=create_deep_agent(model=model,tools=[delete_file,notify_email],interrupt_on={"delete_file":{"allowed_decisions":["approve","reject"]},# 不许改参数"notify_email":True,},checkpointer=InMemorySaver(),)

示例 3：条件中断（仅危险路径）

fromlangchain.tools.tool_nodeimportToolCallRequestdefis_production_path(req:ToolCallRequest)->bool:path=req.tool_call["args"].get("file_path","")returnpath.startswith("/prod/")agent=create_deep_agent(model=model,interrupt_on={"write_file":{"allowed_decisions":["approve","edit","reject"],"when":is_production_path,},},checkpointer=InMemorySaver(),)

示例 4：与 permissions interrupt 配合（敏感目录写入）

fromdeepagentsimportFilesystemPermission,create_deep_agentfromdeepagents.backendsimportFilesystemBackend agent=create_deep_agent(model=model,backend=FilesystemBackend(root_dir=".",virtual_mode=True),permissions=[FilesystemPermission(operations=["write"],paths=["/secrets/**"],mode="interrupt"),],# 无需再写 interrupt_on={"write_file": True}，框架会自动合并checkpointer=InMemorySaver(),)

等价于：对/secrets/**的write_file/edit_file自动生成带when谓词的 interrupt 配置。

十、避坑清单

序号	坑	正确做法
1	不配`checkpointer`	必须`InMemorySaver()`或持久化 checkpointer
2	resume 时换了`thread_id`	必须用同一个`config`
3	多个待审批只传 1 个 decision	`decisions`数量与`action_requests`一致
4	用`respond`表示「拒绝」	拒绝用`reject`
5	`reject`不写 message	敏感操作建议写清「勿重试」
6	以为没写 interrupt_on 就不能 HITL	`permissions mode="interrupt"`也会自动装 HITL
7	CompiledSubAgent 期望继承主 agent interrupt_on	在子图内部自己配
8	不用`version="v2"`	读`interrupts`/ resume 建议用 v2 API
9	interrupt 后程序直接结束	必须循环：`interrupts`→`Command(resume=...)`→ 再 invoke
10	`edit`时漏`edited_action.name`	必须包含工具名 + 完整 args

十一、一张表总结

11.1 我想实现 X → 怎么配？

需求	配置方式
某个自定义工具调用前要审批	`interrupt_on={"tool_name": True}`
只允许批准/拒绝，不许改参数	`{"allowed_decisions": ["approve", "reject"]}`
仅部分参数/路径才审批	加`when`谓词
敏感目录写入要审批	`FilesystemPermission(..., mode="interrupt")`
子 agent 读文件也要审批	SubAgent 字典里单独配`interrupt_on`
中断后继续执行	`Command(resume={"decisions": [...]})`+ 同一`config`

11.2 核心 API 速查

API	作用
`interrupt_on={...}`	声明哪些工具要人工审批
`checkpointer=...`	保存暂停状态（必须）
`config={"configurable": {"thread_id": ...}}`	会话线程 ID
`agent.invoke(..., version="v2")`	标准调用
`result.interrupts`	判断是否暂停
`Command(resume={"decisions": [...]})`	提交审批结果并恢复

📚官方文档
Human-in-the-loop：https://docs.langchain.com/oss/python/deepagents/human-in-the-loop
Permissions interrupt（与 interrupt_on 合并）：https://docs.langchain.com/oss/python/deepagents/permissions

目录