《Deep Agents 实战笔记》第 10 篇 | 基于 deepagents (Python) 官方文档 原文:https://docs.langchain.com/oss/python/deepagents/permissions 原文:https://docs.langchain.com/oss/python/deepagents/human-in-the-loop
上一篇我们给了 Agent 沙箱和解释器,它会跑代码、会执行 shell 了。能力越大,越要装笼子。这一篇讲 Deep Agents 的两层安全围栏:声明式权限规则(代码自动判定,零延迟)和人机协同审批(HITL,把关键决策交还给人)。一个管"哪些事永远不让做",一个管"哪些事必须问过人再做"。
一、为什么权限必须是声明式的
先想清楚我们要防什么。Agent 的每次文件操作都是一次 LLM 决策,而 LLM 决策不可靠——提示注入、模型幻觉、任务理解偏差,都可能让它去读不该读的文件、写不该写的路径。靠在系统提示词里写"请不要访问 secrets 目录"来设防,约等于靠自觉。
Deep Agents 的解法是把权限做成声明式规则:把 FilesystemPermission 规则列表传给 create_deep_agent 的 permissions= 参数,所有内置文件系统工具(ls、read_file、glob、grep、write_file、edit_file、delete)自动遵守。权限功能要求 deepagents>=0.5.2。
每条规则三个字段:
| 字段 | 类型 | 含义 |
|---|---|---|
operations | "read" / "write" | "read" 覆盖 ls、read_file、glob、grep;"write" 覆盖 write_file、edit_file、delete |
paths | Glob 模式列表 | 匹配文件路径,支持 ** 递归和 {a,b} 多选,必须是绝对路径,不能含 .. 或 ~ |
mode | "allow" / "deny" / "interrupt" | 允许、拒绝,或暂停等待人类批准(interrupt 要求 deepagents>=0.6.8) |
判定逻辑只有一条:规则按声明顺序求值,首条匹配生效;没有规则匹配,默认允许。画成流程图:
flowchart TD
CALL["一次文件系统工具调用<br/>如 write_file 到某路径"] --> LOOP["按声明顺序遍历规则"]
LOOP --> MATCH{"当前规则匹配?<br/>operations 与 paths 同时命中"}
MATCH -->|否| MORE{"还有下一条规则吗?"}
MORE -->|有| MATCH
MORE -->|没有| ALLOW["默认:允许执行"]
MATCH -->|是| MODE{"规则的 mode"}
MODE -->|allow| ALLOW
MODE -->|deny| DENY["拒绝操作"]
MODE -->|interrupt| HITL["抛出 HITL 中断<br/>等待人类审批"]
style DENY fill:#fde8e8,color:#8b1a1a,stroke:#e05252
style HITL fill:#fbf3e6,color:#6b4a12,stroke:#c99a3d
style ALLOW fill:#f0f7f0,color:#1d4a1d,stroke:#4a9e4a
正在渲染图表…
注意那个默认允许——这和我直觉中的"防火墙默认拒绝"相反。所以写规则时,标准姿势是"先写窄的例外,最后用宽的 deny 兜底",显式把默认行为封死。
二、四个经典配方
配方一:只读代理。拒绝一切写入,适合分析型场景:
from deepagents import FilesystemPermission, create_deep_agent
agent = create_deep_agent(
model=model,
backend=backend,
permissions=[
FilesystemPermission(
operations=["write"],
paths=["/**"],
mode="deny",
),
],
)
配方二:路径隔离。只允许在 /workspace/ 下活动,其他全拒绝——生产环境多租户场景的标配:
permissions=[
FilesystemPermission(
operations=["read", "write"],
paths=["/workspace/**"],
mode="allow",
),
FilesystemPermission(
operations=["read", "write"],
paths=["/**"],
mode="deny",
),
]
配方三:保护特定文件。在 allow 之前插一条更窄的 deny,把 .env 和示例目录挡住:
permissions=[
FilesystemPermission(
operations=["read", "write"],
paths=["/workspace/.env", "/workspace/examples/**"],
mode="deny",
),
FilesystemPermission(
operations=["read", "write"],
paths=["/workspace/**"],
mode="allow",
),
FilesystemPermission(
operations=["read", "write"],
paths=["/**"],
mode="deny",
),
]
配方四:只读记忆。这是我认为最有想象力的一个——配合第 8 篇的组合 Backend,/memories/ 和 /policies/ 路由到 StoreBackend,再用权限挡住写入:
from deepagents.backends import CompositeBackend, StateBackend, StoreBackend
agent = create_deep_agent(
model=model,
backend=CompositeBackend(
default=StateBackend(),
routes={
"/memories/": StoreBackend(
namespace=lambda rt: (rt.server_info.user.identity,),
),
"/policies/": StoreBackend(
namespace=lambda rt: (rt.context.org_id,),
),
},
),
permissions=[
FilesystemPermission(
operations=["write"],
paths=["/memories/**", "/policies/**"],
mode="deny",
),
],
)
场景很具体:用户自己的记忆允许 Agent 读写(第 8 篇讲过),但组织级策略、共享知识库只能由应用代码更新——Agent 读得、改不得。权限规则和 Backend 路由组合,把"谁能改什么"表达得清清楚楚。
这四个配方还可以自由叠加,因为它们本质上是同一个求值引擎里的不同规则组合:只读代理是"一条兜底 deny",路径隔离是"一条 allow 加一条兜底 deny",保护文件是"窄 deny + 宽 allow + 兜底 deny"。先想清楚威胁模型,再从这几个模板里挑着拼,比从零设计一套权限体系省力得多。同时也要认清这套机制的边界:它只覆盖内置文件系统工具,不包括你自定义的工具和 MCP 工具,也不适用于沙箱 Backend——后者能通过 execute 执行任意命令,本来就是另一个层级的风险,需要上一篇讲的沙箱隔离和下一篇之外的其他手段配合。
三、规则顺序:首条匹配生效,顺序就是逻辑
first-match-wins 意味着顺序错误 = 权限失效,而且不报错:
# 正确:先 deny .env,再 allow workspace
correct_permissions = [
FilesystemPermission(
operations=["read", "write"],
paths=["/workspace/.env"],
mode="deny",
),
FilesystemPermission(
operations=["read", "write"],
paths=["/workspace/**"],
mode="allow",
),
FilesystemPermission(
operations=["read", "write"],
paths=["/**"],
mode="deny",
),
]
# 错误:/workspace/** 先匹配到 .env,deny 永远不会被求值
incorrect_permissions = [
FilesystemPermission(
operations=["read", "write"],
paths=["/workspace/**"],
mode="allow",
),
FilesystemPermission(
operations=["read", "write"],
paths=["/workspace/.env"],
mode="deny", # 永远不会到达
),
]
这类 bug 的阴险之处在于:.env 的保护规则明明写在配置里,测试时如果恰好没触发那条路径,你根本发现不了。把"更具体的规则在前,更宽泛的规则在后"当成铁律。
两个边角行为也值得知道。其一,删除目录是全有或全无:delete 会检查目标及其每个后代路径的 write 权限,任何一个被拒,整个操作就拒绝,不会删一半留一半。其二,删除普通文件从 deepagents>=0.7.3 起采用精确匹配,和 write_file 一样走首条匹配,所以更早、更窄的 allow 能胜过之后的兜底 deny。
还有一条实用提示:给 interrupt 模式的规则用字面前缀锚定的模式(如 /secrets/**)。批量工具(ls、glob、grep)在搜索子树可能与规则前缀重叠时就会触发中断,像 /**/secrets 这种没有锚定的模式会保守地过度触发——审批弹窗满天飞,用户很快就会开始无脑点"同意",这才是最大的安全隐患。
四、组合 Backend 下,权限不是想写就能写
第 4 篇讲过 CompositeBackend 可以按路径前缀路由到不同存储。当组合 Backend 的默认 Backend 是沙箱时,权限规则有一条硬约束:每条规则的路径必须限定在已知的路由前缀之内。
原因很本质:沙箱支持任意命令执行,仅靠基于路径的规则无法阻止代理通过 shell 命令访问文件系统。把权限限定到路由指向的非沙箱 Backend,才能避免这个口子。
from deepagents.backends import CompositeBackend
composite = CompositeBackend(
default=sandbox,
routes={"/memories/": memories_backend},
)
# 可行:权限限定在 /memories/ 路由内
agent = create_deep_agent(
model=model,
backend=composite,
permissions=[
FilesystemPermission(
operations=["write"],
paths=["/memories/**"],
mode="deny",
),
],
)
# 抛出 NotImplementedError:/workspace/** 命中沙箱默认 Backend
# 同理,/** 同时覆盖路由和默认 Backend,也会抛出
这个设计其实是在强迫你面对现实:你想给一段路径上权限,就先把那段路径路由到一个真正可控的 Backend 上。顺便提醒,权限只覆盖内置文件系统工具——自定义工具和 MCP 工具不在此列;需要限流、审计、内容检查这类自定义校验时,官方建议改用 Backend 策略钩子。
五、HITL:interrupt-审批-恢复
权限规则是二值的(allow/deny),但很多操作不是"禁止"或"放行"能概括的——发邮件、删文件、花钱调用外部 API,这些事需要人看一眼再定。这就是 HITL(human-in-the-loop,人机协同)的领地。
Deep Agents 通过 interrupt_on 参数配置哪些工具需要批准。设置后,HumanInTheLoopMiddleware 会自动加入中间件栈;如果一次运行在工具返回前被取消,同栈的 PatchToolCallsMiddleware 会自动修复消息历史。HITL 必须配 checkpointer——中断时要持久化代理状态,恢复时才能从断点继续:
from langchain.tools import tool
from deepagents import create_deep_agent
from langgraph.checkpoint.memory import MemorySaver
@tool
def remove_file(path: str) -> str:
"""从文件系统删除一个文件。"""
return f"Deleted {path}"
@tool
def fetch_file(path: str) -> str:
"""从文件系统读取一个文件。"""
return f"Contents of {path}"
@tool
def notify_email(to: str, subject: str, body: str) -> str:
"""发送一封邮件。"""
return f"Sent email to {to}"
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
tools=[remove_file, fetch_file, notify_email],
interrupt_on={
"remove_file": True, # 默认:approve、edit、reject、respond
"fetch_file": False, # 无需中断
"notify_email": {"allowed_decisions": ["approve", "reject"]}, # 不允许编辑
},
checkpointer=MemorySaver(),
)
审批者对被中断的工具调用有四种决策:
| 决策 | 行为 | 适用 |
|---|---|---|
approve | 按原始参数执行 | 按原样发送邮件草稿 |
edit | 执行前修改参数 | 发送前改收件人 |
reject | 跳过执行,拒绝反馈回传给代理 | 拒绝删除文件并说明原因 |
respond | 人的消息直接作为合成的工具结果返回 | 回答 ask_user 类提示 |
完整的执行-中断-审批-恢复时序如下:
sequenceDiagram
participant U as 应用与审查者
participant A as 代理
participant T as 敏感工具
U->>A: invoke(删除 temp.txt,带 thread_id)
A->>A: 模型决定调用 remove_file
Note over A: HITL 中间件命中 interrupt_on 规则
A-->>U: 暂停,返回 __interrupt__<br/>含 action_requests 与 review_configs
Note over U: 展示工具名、参数、允许的决策类型
U->>A: invoke Command 携带 decisions<br/>(必须使用相同 thread_id)
alt approve 或 edit
A->>T: 按决策执行或修改后执行
T-->>A: 工具结果
else reject 或 respond
Note over A: 跳过执行<br/>决策消息作为 ToolMessage 回传
end
A-->>U: 继续循环直到任务完成
正在渲染图表…
恢复的完整代码长这样(摘自官方文档,略有精简):
from langchain_core.utils.uuid import uuid7
from langgraph.types import Command
config = {"configurable": {"thread_id": str(uuid7())}}
result = agent.invoke(
{"messages": [{"role": "user", "content": "Delete the file temp.txt"}]},
config=config,
)
# 检查执行是否被中断
if result.get("__interrupt__"):
interrupt_value = result["__interrupt__"][0].value
action_requests = interrupt_value["action_requests"]
review_configs = interrupt_value["review_configs"]
# 按 action_requests 的相同顺序提供决策
decisions = [
{
"type": "reject",
"message": "User rejected deleting temp.txt. Do not retry deletion.",
}
]
# 携带决策恢复执行,config 必须相同
result = agent.invoke(
Command(resume={"decisions": decisions}),
config=config,
)
print(result["messages"][-1].content)
三个细节最容易被忽略:
- 恢复必须用同一个
thread_id。checkpointer 靠它找到暂停时的状态,config 变了等于另起炉灶。 - decisions 的顺序必须和
action_requests一一对应。当代理一次调用多个敏感工具时,所有中断会合并为一个中断,你必须按顺序给每个行动一个决策。 - reject 时带上领域相关的
message。省略的话默认反馈只说"工具未执行,除非用户要求否则别重试";传入明确的消息可以告诉代理"放弃这个行动"、"追问一个问题"还是"换更安全的方案"。另外注意:respond是给"人扮演工具"的场景用的,别用它拒绝有副作用的工具——它的消息可能被模型当成一次成功的工具结果。
对实时 UI,官方建议用 .stream() 而非 .invoke():"messages" 模式流式输出 token,"updates" 模式捕获 __interrupt__ 载荷,中断后收集决策、用 Command(resume=...) 再次流式恢复,形成"流式 → 暂停 → 审批 → 再流式"的循环。产品视角看,这个循环就是 HITL 前端的骨架:token 流负责"活着"的感觉,中断载荷负责"刹车"的时机,两者共用同一条 stream 通道,不需要额外的轮询或回调机制。另外,子代理触发的中断同样会出现在父级的 updates 流里,投递中断不需要开 subgraphs=True——只有想看子代理内部的 token 和进度事件时才需要。
六、条件中断:只拦该拦的
默认情况下,interrupt_on 列出的每一次工具调用都会暂停。如果写文件操作高频但只有"写到工作区之外"才危险,就该用条件中断:在配置里加一个 when 谓词(要求 langchain>=1.3.3),谓词接收 ToolCallRequest,返回 True 中断、False 自动放行:
from deepagents import create_deep_agent
from langchain.agents.middleware import ToolCallRequest
from langgraph.checkpoint.memory import MemorySaver
def writes_outside_workspace(request: ToolCallRequest) -> bool:
"""对工作区目录之外的写入暂停。"""
path = request.tool_call["args"].get("file_path", "")
return not path.startswith("/workspace/")
agent = create_deep_agent(
model="google_genai:gemini-3.6-flash",
interrupt_on={
"write_file": {
"allowed_decisions": ["approve", "edit", "reject"],
"when": writes_outside_workspace,
},
},
checkpointer=MemorySaver(),
)
返回 False 的调用直接运行,永远不会进入中断批次——审查者只会看到真正需要决策的行动。这是审批体验和安全性之间最好的平衡点。
条件中断之外,还有一招按风险分级配置。官方文档给出的分层思路值得直接抄作业:
interrupt_on = {
# 高风险:完全控制(approve、edit、reject)
"delete_file": {"allowed_decisions": ["approve", "edit", "reject"]},
"send_email": {"allowed_decisions": ["approve", "edit", "reject"]},
# 中风险:不允许编辑
"write_file": {"allowed_decisions": ["approve", "reject"]},
# 低风险:不中断
"read_file": False,
"ls": False,
}
高风险工具给审查者全部选项(可能还要顺手改改参数),中风险只留批准或拒绝,低风险完全不拦。这里的取舍逻辑是:审批次数要和数据的风险密度成正比。每多一次审批,用户的耐心就少一分——审得越少,每次审批才越被认真对待。宁可把中断点设计得少而重,也不要让审批弹窗沦为"同意"按钮的连点器训练器。edit 尤其要慎开:官方特别提醒编辑参数时保守行事,对原始参数的大幅改动可能导致模型重新评估做法,进而多次执行该工具或采取意外行动。
七、子代理与权限的继承
权限和 HITL 在子代理上的行为是本篇最后一块拼图:
- 权限默认继承:子代理自动获得父代理的权限规则。在子代理 spec 里设置
permissions字段则完全替换父规则(不是合并);设为空列表[]表示无限制,省略则继承。 interrupt_on可以覆盖:每个子代理可有自己的interrupt_on配置。比如主代理里read_file不需要审批,但在某个子代理里可以改成需要。
agent = create_deep_agent(
model=model,
backend=backend,
permissions=[
FilesystemPermission(
operations=["read", "write"],
paths=["/workspace/**"],
mode="allow",
),
FilesystemPermission(
operations=["read", "write"],
paths=["/**"],
mode="deny",
),
],
subagents=[
{
"name": "auditor",
"description": "Read-only code reviewer",
"system_prompt": "Review the code for issues.",
"permissions": [
FilesystemPermission(
operations=["write"],
paths=["/**"],
mode="deny",
),
FilesystemPermission(
operations=["read"],
paths=["/workspace/**"],
mode="allow",
),
FilesystemPermission(
operations=["read"],
paths=["/**"],
mode="deny",
),
],
}
],
)
这套机制组合起来,就是一个生产级的最小权限矩阵:主代理能读写工作区,审计子代理只读,记忆目录谁都不能写。把整个安全体系画成一张图:
flowchart TB
P["主代理权限规则<br/>permissions"] -->|"默认继承"| SA1["子代理 A"]
P -->|"spec 设置 permissions<br/>完全替换"| SA2["子代理 B 定制权限"]
SA2 -->|"permissions 为空列表"| SA3["无限制访问"]
IO["工具级中断 interrupt_on"] --> MERGE["合并为同一个审查步骤"]
FP["权限规则 mode=interrupt"] --> MERGE
MERGE --> HITL["HITL 中断<br/>approve / edit / reject / respond"]
style HITL fill:#fbf3e6,color:#6b4a12,stroke:#c99a3d
style P fill:#eef4ff,color:#1a3a8b,stroke:#4a72d9
正在渲染图表…
值得一提:权限规则的 mode="interrupt" 触发的中断,会和 interrupt_on 的工具中断合并——单个审查步骤可以同时覆盖自定义工具和受保护的文件系统路径,前端只需要实现一套审批 UI。
八、本篇踩坑点
- 默认行为是允许,不是拒绝:没有规则匹配时操作直接放行。写权限一定要有兜底的
deny规则,别指望"没写规则就是禁止"。 - 规则顺序写反:宽泛的 allow 排在窄的 deny 前面,deny 永远不会被求值,且无任何报错。更具体的规则永远在前。
- 忘了 checkpointer:
interrupt_on和mode="interrupt"都需要 checkpointer 才能暂停恢复,生产环境配持久化实现,别只用MemorySaver起演示最后忘了换。 - 恢复时换了 thread_id 或 config:状态找不回来,中断悬挂。恢复必须复用中断时的同一个 config。
- decisions 顺序错位:多个工具调用合并为一个中断时,决策列表必须与
action_requests严格按顺序对应,错位会导致"批准了 A 实际执行了 B"。 - 用
respond当拒绝用:它会把人的消息当作成功的工具结果回传,模型会以为操作成功了。拒绝一律用reject+message。 - interrupt 规则没锚定前缀:
/**/secrets这类模式会让批量工具保守地过度触发中断,审批疲劳比没有审批更危险。 - 以为权限能管住沙箱和自定义工具:权限只覆盖内置文件系统工具;组合 Backend 默认是沙箱时,规则还必须限定在路由前缀内,否则构造时直接抛
NotImplementedError。另记一笔:第 9 篇提过,解释器 PTC 调用也不走审批路径——敏感工具别放进 PTC 允许列表。
机制讲到这里,工具箱已经齐了:文件系统、Backend、子代理、上下文工程、记忆技能、沙箱解释器、权限与 HITL。从下一篇开始进入实战三部曲:第一步,完整复刻官方的 Deep Research 深度研究代理——多子代理分工、联网搜索、结构化报告,把前十章的机制在一个真实项目里全部串起来。
参考资料:
- Permissions(官方文档,本文主要素材):https://docs.langchain.com/oss/python/deepagents/permissions
- Human-in-the-loop(官方文档,本文主要素材):https://docs.langchain.com/oss/python/deepagents/human-in-the-loop
- LangChain 人机协同文档(条件中断更多配置):https://docs.langchain.com/oss/langchain/human-in-the-loop
- Backends 策略钩子(自定义校验逻辑):https://docs.langchain.com/oss/deepagents/backends