Guardrails 安全守卫

本章目标:
理解 DeerFlow 为什么需要在工具执行之前插入一层可插拔的策略授权,以及它与沙箱隔离、人工确认的本质差异。
掌握 GuardrailProvider 协议、内置 AllowlistProvider、GuardrailMiddleware 三个组件的职责与协作方式,看清一次工具调用从拦截到 deny ToolMessage 的完整数据流。
能够按照协议约束实现并接入自己的自定义 provider(或 OAP 策略 provider),并安全地配置 guardrails.enabled 等开关。

TL;DR

Guardrails 是一个可选的、工具调用前的授权层:当 guardrails.enabled=true 时,GuardrailMiddleware 包裹每一次工具调用,把工具名、参数、passport 引用打包成 GuardrailRequest 交给配置的 GuardrailProvider.evaluate() 判定。判定为 deny 时返回一条 status="error" 的 ToolMessage,让 agent 看到拒绝原因并自行改换策略,而不是中断整个 run。Provider 是一个 Protocol(无需继承基类),通过 resolve_variable() 按类路径加载,提供三种选项:零依赖的内置 AllowlistProvider、OAP 标准策略 provider(如 aport-agent-guardrails)、或任何带 evaluate/aevaluate 方法的自定义类。Provider 抛异常时按 fail_closed(默认 True)决定阻断还是放行。

Overview

为什么 agent 需要工具调用前的可插拔授权

DeerFlow 是一个能自主执行多步任务的 super agent,它能调用 bash、write_file 等高权限工具。系统已有两道安全机制,但都有结构性缺口:沙箱提供进程隔离却不做语义授权(沙箱里的 bash 仍可 curl 把数据外传),ask_clarification 的人工确认对每个动作都要人在回路里、无法用于自主工作流 GUARDRAILS.md:34-36。

Guardrails 填补的正是这道缺口:一层确定性的、由策略驱动的、无需人工介入的授权。它的核心设计取舍有三点,都能从源码读出:

拦截点在工具执行前而非之后:GuardrailMiddleware 实现 wrap_tool_call,在调用真正的 tool handler 前先做判定 middleware.py:54-75。
deny 不是抛异常而是回写一条 error ToolMessage:agent 能看到拒绝文案并改换替代方案,run 继续推进而不是崩溃 middleware.py:42-52。
判定逻辑外置为可插拔 provider:GuardrailProvider 是一个 runtime_checkable 的 Protocol,任何带 evaluate/aevaluate 方法的类都满足契约,无需继承基类,按类路径用 resolve_variable() 加载——与 DeerFlow 加载 model/tool/sandbox 完全相同的机制 provider.py:39-46。

Architecture

Guardrails 由两条线组成:一条是数据契约 + 协议(provider.py),一条是拦截执行(middleware.py)。两者通过 GuardrailRequest(入参)和 GuardrailDecision(出参)解耦。Middleware 永远不关心 provider 内部如何判定,provider 永远不关心它被插在中间件链的哪个位置。

中间件链装配发生在 build_middlewares(tool_error_handling_middleware.py):仅当 guardrails_config.enabled 且 guardrails_config.provider 都满足时,才用 resolve_variable() 加载 provider 类、实例化、追加 GuardrailMiddleware 到链中,位置在 LLMErrorHandlingMiddleware 之后、SandboxAuditMiddleware 之前 tool_error_handling_middleware.py:99-124。

Source 列表:

文件	职责	关键符号
`guardrails/provider.py`	数据契约与协议定义	`GuardrailRequest` / `GuardrailDecision` / `GuardrailReason` / `GuardrailProvider` provider.py:9-57
`guardrails/builtin.py`	零依赖内置 provider	`AllowlistProvider` builtin.py:6-23
`guardrails/middleware.py`	工具调用拦截与判定执行	`GuardrailMiddleware` middleware.py:20-99
`guardrails/__init__.py`	公开 API 出口	`__all__` 导出 6 个符号 init.py:7-14
`config/guardrails_config.py`	配置模型与单例加载	`GuardrailsConfig` / `get_guardrails_config` guardrails_config.py:6-48
`agents/middlewares/tool_error_handling_middleware.py`	按配置装配中间件	`build_middlewares` 中的 guardrail 分支 tool_error_handling_middleware.py:99-124

Components / Subsystems

GuardrailProvider(协议与数据契约)

职责:定义 middleware 与判定逻辑之间的契约。它是一个 @runtime_checkable 的 Protocol,带一个 name: str 属性和 evaluate / aevaluate 两个方法。文档字符串明确说明:任何拥有这些方法的类都能工作,无需继承基类;provider 通过 resolve_variable() 按类路径加载,与 model/tool/sandbox 的加载机制相同 provider.py:39-57。

关键类:

GuardrailRequest:每次工具调用传给 provider 的上下文,字段含 tool_name、tool_input、agent_id、thread_id、is_subagent、timestamp,后四个有默认值 provider.py:9-18。
GuardrailReason:结构化的允许/拒绝理由(对齐 OAP reason 对象),含 code 与可选 message provider.py:21-26。
GuardrailDecision:provider 的裁决(对齐 OAP Decision 对象),含 allow: bool、reasons 列表、可选 policy_id、metadata 字典 provider.py:29-36。

AllowlistProvider(内置零依赖 provider)

职责:最简单的开箱即用 provider,按工具名做白名单/黑名单过滤,无任何外部依赖、无 passport、无网络。name = "allowlist"。构造参数为两个仅关键字参数 allowed_tools 与 denied_tools builtin.py:6-13。

判定规则按顺序:若设置了白名单且工具不在其中 → deny(code oap.tool_not_allowed);若工具在黑名单中 → deny;否则 allow(code oap.allowed)。aevaluate 直接复用同步 evaluate builtin.py:15-23。测试验证它在运行时满足 GuardrailProvider 协议(isinstance(AllowlistProvider(), GuardrailProvider))test_guardrail_middleware.py:202-205。

GuardrailMiddleware(拦截执行)

职责:继承 AgentMiddleware[AgentState],实现 wrap_tool_call / awrap_tool_call,在工具执行前调用 provider 判定。构造参数:provider、仅关键字 fail_closed(默认 True)、passport middleware.py:20-33。

行为要点:_build_request 从 ToolCallRequest 提取工具名、参数,把构造时传入的 passport 写进 agent_id,并打上 UTC ISO 时间戳 middleware.py:34-40;deny 时 _build_denied_message 生成 status="error" 的 ToolMessage,文案含工具名、reason code 与 message,并提示 agent "Choose an alternative approach" middleware.py:42-52。GraphBubbleUp 异常(LangGraph 的 interrupt/pause/resume 控制信号)始终向上抛出、绝不吞掉 middleware.py:63-65。

Data Flow

下图展示一次工具调用从模型决策到 deny ToolMessage 的完整路径(以同步 wrap_tool_call 为例,异步 awrap_tool_call 逻辑对称 middleware.py:77-98)。

Implementation Details

内置 provider 的 deny 判定(AllowlistProvider.evaluate,9 行):

python

def evaluate(self, request: GuardrailRequest) -> GuardrailDecision:
    if self._allowed is not None and request.tool_name not in self._allowed:
        return GuardrailDecision(allow=False, reasons=[GuardrailReason(code="oap.tool_not_allowed", message=f"tool '{request.tool_name}' not in allowlist")])
    if request.tool_name in self._denied:
        return GuardrailDecision(allow=False, reasons=[GuardrailReason(code="oap.tool_not_allowed", message=f"tool '{request.tool_name}' is denied")])
    return GuardrailDecision(allow=True, reasons=[GuardrailReason(code="oap.allowed")])

解读:白名单优先于黑名单——只要配了 allowed_tools,不在其中的工具一律 deny;未配白名单时,只检查黑名单。每个分支都返回带 oap.* code 的结构化理由,middleware 会把它原样拼进 deny 文案 builtin.py:15-20。

middleware 的 deny 路径与 fail-closed 容错(wrap_tool_call 核心,< 20 行):

python

try:
    decision = self.provider.evaluate(gr)
except GraphBubbleUp:
    raise  # 保留 LangGraph 控制流信号
except Exception:
    logger.exception("Guardrail provider error (sync)")
    if self.fail_closed:
        decision = GuardrailDecision(allow=False, reasons=[GuardrailReason(code="oap.evaluator_error", message="guardrail provider error (fail-closed)")])
    else:
        return handler(request)
if not decision.allow:
    logger.warning("Guardrail denied: tool=%s policy=%s code=%s", gr.tool_name, decision.policy_id, ...)
    return self._build_denied_message(request, decision)
return handler(request)

解读:三段式异常处理——GraphBubbleUp 必须重新抛出以保留中断/恢复语义;其它异常时按 fail_closed 选择"安全失败"(构造 oap.evaluator_error 的 deny)或"宽松放行"(直接 handler(request));只有 decision.allow 为真才调用真正的 handler middleware.py:60-75。

速查表

三种 provider 选项对比矩阵:

维度	内置 AllowlistProvider	OAP 策略 provider	自定义 provider	Source
类路径 `use`	`deerflow.guardrails.builtin:AllowlistProvider`	`aport_guardrails.providers.generic:OAPGuardrailProvider`(APort 示例)	`my_package:MyGuardrailProvider`	config.example.yaml:1051-1074
外部依赖	无(零依赖)	需 `pip install aport-agent-guardrails` 等	取决于实现	config.example.yaml:1051-1066
判定方式	按工具名白/黑名单	按 OAP passport + 策略码	任意逻辑	builtin.py:15-20
需要 passport	否	是(`guardrails.passport`)	可选	guardrails_config.py:23
`framework` 注入	否(不接受该参数)	是(构造器含 `framework` 或 `**kwargs` 时自动注入 `"deerflow"`)	视构造器签名而定	tool_error_handling_middleware.py:109-119
适用场景	快速锁掉危险工具	跨框架统一策略治理	业务自定义规则	GUARDRAILS.md:81-85

扩展指南

实现自定义 GuardrailProvider 的最小模板:

python

# my_package/guardrail.py
from deerflow.guardrails import GuardrailDecision, GuardrailReason, GuardrailRequest


class MyGuardrailProvider:
    name = "my-policy"  # 协议要求的 name 属性

    def __init__(self, *, blocked_prefixes: list[str] | None = None, **kwargs):
        # config.yaml 中 provider.config 的键作为 kwargs 传入
        self._blocked = blocked_prefixes or []
        # 接受 **kwargs 时,装配器会额外注入 framework="deerflow"

    def evaluate(self, request: GuardrailRequest) -> GuardrailDecision:
        for p in self._blocked:
            if request.tool_name.startswith(p):
                return GuardrailDecision(
                    allow=False,
                    reasons=[GuardrailReason(code="custom.blocked", message=f"{request.tool_name} blocked")],
                )
        return GuardrailDecision(allow=True, reasons=[GuardrailReason(code="custom.allowed")])

    async def aevaluate(self, request: GuardrailRequest) -> GuardrailDecision:
        return self.evaluate(request)  # 无异步 I/O 时直接复用同步实现

对应配置:

yaml

guardrails:
  enabled: true
  provider:
    use: my_package.guardrail:MyGuardrailProvider
    config:
      blocked_prefixes: ["bash", "write_"]

约束清单(直接从协议定义读出):

必须有 name: str 属性 provider.py:48。
必须实现 evaluate(self, request: GuardrailRequest) -> GuardrailDecision(同步)provider.py:50-52。
必须实现 async def aevaluate(self, request: GuardrailRequest) -> GuardrailDecision(异步变体)——GuardrailMiddleware.awrap_tool_call 会 await self.provider.aevaluate(gr) provider.py:54-56、middleware.py:85。
无需继承任何基类——Protocol 是结构化类型,鸭子类型即可;协议是 @runtime_checkable,isinstance 检查只校验方法存在 provider.py:39-46。
构造器参数由 provider.config 作为 kwargs 注入;若构造器含 framework 形参或 **kwargs,装配器会自动注入 framework="deerflow"(内置 provider 不接受该参数因而不注入)tool_error_handling_middleware.py:107-119。
切勿捕获 GraphBubbleUp——它是 LangGraph 的控制流信号,middleware 会原样向上传播,provider 内部也不应吞掉 middleware.py:63-65。

注意:agents/factory.py 的 create_* 工厂路径目前没有内置 GuardrailMiddleware——通过该工厂启用 guardrail 需传入自定义 AgentMiddleware 实例,否则会抛 ValueError factory.py:208-213。lead agent 的标准装配路径(build_middlewares)则会自动接入。

Configuration

配置位于 config.yaml 的 guardrails 段,模型为 GuardrailsConfig:

字段	默认值	含义	Source
`enabled`	`false`	是否启用 guardrail 中间件	guardrails_config.py:21
`fail_closed`	`true`	provider 报错时是否阻断工具调用	guardrails_config.py:22
`passport`	`null`	OAP passport 路径或托管 agent ID	guardrails_config.py:23
`provider.use`	(无默认)	provider 类路径,如 `deerflow.guardrails.builtin:AllowlistProvider`	guardrails_config.py:9
`provider.config`	`{}`	传给 provider 构造器的 kwargs	guardrails_config.py:10

配置在 AppConfig 加载时通过 load_guardrails_config_from_dict(config.guardrails.model_dump()) 写入进程级单例 app_config.py:197;中间件装配仅在 enabled 且 provider 均存在时才生效 tool_error_handling_middleware.py:100-101。

安全标注:

⚠️ enabled=false 是默认值:不显式开启则没有任何工具调用授权层,沙箱里的 bash 可任意执行。生产自主工作流强烈建议开启 guardrails_config.py:21。
⚠️ fail_closed=true 是安全默认:切勿在生产把它改为 false——那会在 provider 故障(如网络抖动、策略服务宕机)时静默放行所有工具调用,等于绕过整个授权层 middleware.py:66-71。
⚠️ 白名单优先于黑名单:AllowlistProvider 一旦设置 allowed_tools,所有未列入的工具(包括将来新增的内置工具)都会被 deny;只想拦个别工具时应只配 denied_tools,不要配空/不全的 allowed_tools builtin.py:16-19。
⚠️ provider.config 的值会原样作为 kwargs 传入构造器:不要把密钥明文写在 config.yaml,优先用 $ENV_VAR 形式由配置系统解析环境变量。

Common Pitfalls / Tips

deny 不会中断 run:被拒绝的工具调用会得到一条 status="error" 的 ToolMessage,agent 会继续推理并可能改用其它工具。Guardrails 是"引导"而非"熔断",真正的硬停止由 LoopDetectionMiddleware 等其它机制负责 middleware.py:72-74。
reasons 可能为空:middleware 取 decision.reasons[0] 前已做空判断,空列表时回退为 message "blocked by guardrail policy"、code "oap.denied";自定义 provider 仍建议显式填 reasons 以便排障 middleware.py:45-46。
passport 来自构造时的固定值:GuardrailRequest.agent_id 填的是 GuardrailMiddleware 构造时传入的 passport,不是逐线程动态值;thread_id 字段在标准装配路径下保持默认空 middleware.py:34-40。
同步与异步分别走 evaluate / aevaluate:LangGraph 以异步路径运行时调用 aevaluate,自定义 provider 必须实现它(即使只是转调 evaluate),否则异步 run 会失败 middleware.py:83-85。
测试隔离用 reset_guardrails_config():配置是进程级单例,测试间务必重置以防止泄漏 guardrails_config.py:45-48。

References

backend/packages/harness/deerflow/guardrails/provider.py — 协议与数据契约
backend/packages/harness/deerflow/guardrails/builtin.py — AllowlistProvider
backend/packages/harness/deerflow/guardrails/middleware.py — GuardrailMiddleware
backend/packages/harness/deerflow/config/guardrails_config.py — 配置模型与单例
backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py — 中间件装配
backend/docs/GUARDRAILS.md — 设计动机与三种 provider 说明
config.example.yaml — 配置示例
backend/tests/test_guardrail_middleware.py — 协议一致性与行为回归测试

页面	关系
./12-中间件链机制.md	GuardrailMiddleware 是中间件链第 6 环,本章是该链中"工具调用前授权"一环的深入展开
./20-工具系统与内置工具.md	Guardrails 拦截的对象正是这里装配的工具(`bash`/`write_file` 等);理解工具集有助于设计白/黑名单
./14-鉴权-CSRF与授权.md	鉴权是"谁能进系统"的边界授权,Guardrails 是"agent 能调哪个工具"的运行时授权,二者互补
./04-配置系统与AppConfig.md	`GuardrailsConfig` 由 AppConfig 在启动时加载为进程单例,本章配置项的加载链路源于此

Guardrails 安全守卫 ​

TL;DR ​

Overview ​

为什么 agent 需要工具调用前的可插拔授权 ​

Architecture ​

Components / Subsystems ​

GuardrailProvider(协议与数据契约) ​

AllowlistProvider(内置零依赖 provider) ​

GuardrailMiddleware(拦截执行) ​

Data Flow ​

Implementation Details ​

速查表 ​

扩展指南 ​

Configuration ​

Common Pitfalls / Tips ​

References ​

Related Pages ​

Guardrails 安全守卫

TL;DR

Overview

为什么 agent 需要工具调用前的可插拔授权

Architecture

Components / Subsystems

GuardrailProvider(协议与数据契约)

AllowlistProvider(内置零依赖 provider)

GuardrailMiddleware(拦截执行)

Data Flow

Implementation Details

速查表

扩展指南

Configuration

Common Pitfalls / Tips

References

Related Pages