首页
/ DeerFlow Guardrails 深度解析:基于中间件的工具调用前置授权机制

DeerFlow Guardrails 深度解析:基于中间件的工具调用前置授权机制

2026-09-05 23:46:02作者:裴锟轩Denise

DeerFlow(deer-flow)是一个长时程 SuperAgent 运行框架:Agent 借助沙箱、记忆、工具、技能、子代理和消息网关,执行从几分钟到数小时的自主多步任务。但自主性带来一个核心安全问题——Agent 可以带着任意参数执行任何已加载的工具。本文基于仓库文档 backend/docs/GUARDRAILS.md 并结合 backend/packages/harness/deerflow/guardrails/ 源码,系统讲解 DeerFlow 的 Guardrails 前置授权体系:GuardrailMiddleware 如何在工具执行前拦截并评估每一次调用,三种 Provider(内置 Allowlist、OAP 护照、自定义)如何配置与实现,以及 fail_closed、运行归因(runtime attribution)、授权结果记录等生产级细节。读完后你将能够为任意 DeerFlow 部署设计一套确定性的、策略驱动的工具授权方案。

为什么需要 Guardrails:沙箱与人工审批之外的第三层

DeerFlow 已有的安全机制各有边界,Guardrails 文档用一张流程图点出了两者的盲区:

Without guardrails:                      With guardrails:

  Agent                                    Agent
    │                                        │
    ▼                                        ▼
  ┌──────────┐                             ┌──────────┐
  │ bash     │──▶ executes immediately     │ bash     │──▶ GuardrailMiddleware
  │ rm -rf / │                             │ rm -rf / │        │
  └──────────┘                             └──────────┘        ▼
                                                         ┌──────────────┐
                                                         │  Provider    │
                                                         │  evaluates   │
                                                         │  against     │
                                                         │  policy      │
                                                         └──────┬───────┘
                                                                │
                                                          ┌─────┴─────┐
                                                          │           │
                                                        ALLOW       DENY
                                                          │           │
                                                          ▼           ▼
                                                      Tool runs   Agent sees:
                                                      normally    "Guardrail denied:
                                                                   rm -rf blocked"

三层机制的定位差异如下:

  • 沙箱(Sandboxing):提供进程级隔离,但不提供语义级授权。沙箱里的 bash 仍然可以 curl 把数据传出去。
  • 人工审批(ask_clarification:每个动作都需要人在环,无法支撑自主工作流。
  • Guardrails:确定性的、策略驱动的授权层,无需人工干预即可在工具执行前作出允许/拒绝决策。

架构:GuardrailMiddleware 在中间件链中的位置

backend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.py 的源码可以确认,GuardrailMiddleware 通过 tail.append(...) 注册进 wrap_tool_call 中间件链,与文档架构图一致:

┌─────────────────────────────────────────────────────────────────────┐
│                        Middleware Chain                               │
│                                                                      │
│  1. ThreadDataMiddleware     ─── per-thread dirs                     │
│  2. UploadsMiddleware        ─── file upload tracking                │
│  3. SandboxMiddleware        ─── sandbox acquisition                 │
│  4. DanglingToolCallMiddleware ── fix incomplete tool calls           │
│  5. GuardrailMiddleware ◄──── EVALUATES EVERY TOOL CALL             │
│  6. ToolErrorHandlingMiddleware ── convert exceptions to messages     │
│  7-12. (Summarization, Title, Memory, Vision, Subagent, Clarify)    │
│                                                                      │
└─────────────────────────────────────────────────────────────────────┘
                         │
                         ▼
           ┌──────────────────────────┐
           │    GuardrailProvider     │  ◄── pluggable: any class
           │    (configured in YAML)  │      with evaluate/aevaluate
           └────────────┬─────────────┘
                        │
              ┌─────────┼──────────────┐
              │         │              │
              ▼         ▼              ▼
         Built-in   OAP Passport    Custom
         Allowlist  Provider        Provider
         (zero dep) (open standard) (your code)

源码中的装配逻辑(tool_error_handling_middleware.py)值得注意:

guardrails_config = app_config.guardrails
if guardrails_config.enabled and guardrails_config.provider:
    import inspect
    from deerflow.guardrails.middleware import GuardrailMiddleware
    from deerflow.reflection import resolve_variable

    provider_cls = resolve_variable(guardrails_config.provider.use)
    provider_kwargs = dict(guardrails_config.provider.config) if guardrails_config.provider.config else {}
    # Pass framework hint if the provider accepts it (e.g. for config discovery).
    # Built-in providers like AllowlistProvider don't need it, so only inject
    # when the constructor accepts 'framework' or '**kwargs'.
    if "framework" not in provider_kwargs:
        try:
            sig = inspect.signature(provider_cls.__init__)
            if "framework" in sig.parameters or any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values()):
                provider_kwargs["framework"] = "deerflow"
        except (ValueError, TypeError):
            pass
    provider = provider_cls(**provider_kwargs)
    tail.append(GuardrailMiddleware(provider, fail_closed=guardrails_config.fail_closed, passport=guardrails_config.passport))

这印证了三个事实:中间件只有在 enabled: true 且配置了 provider 时才会被装配;Provider 通过 resolve_variable() 按类路径加载(与模型、工具、沙箱 Provider 同一机制);framework="deerflow" 提示参数只会在 Provider 构造函数接受 framework**kwargs 时注入,因此内置的 AllowlistProvider 这类零依赖 Provider 无需关心该参数。

另外,该文件中还有一处值得了解的细节:当启用了独立的细粒度授权(authorization 配置段)时,GuardrailAuthorizationAdapter 会包裹成另一个 GuardrailMiddleware 并作为外层守卫先执行——也就是说 DeerFlow 把执行期的拒绝、审计与 fail-closed 处理收敛在同一个久经考验的中间件实现中(见 tool_error_handling_middleware.py 的注释)。

中间件的六步执行逻辑

GuardrailMiddleware 实现的是 wrap_tool_call / awrap_tool_call 钩子(与 ToolErrorHandlingMiddleware 相同的 AgentMiddleware 模式),在 backend/packages/harness/deerflow/guardrails/middleware.py 中的完整行为为:

  1. ToolCallRequest 的运行时上下文构建 GuardrailRequest,包含工具名、参数、护照引用及运行归因字段;
  2. 调用已配置 Provider 的 provider.evaluate(request)(异步路径调用 aevaluate);
  3. 拒绝:返回 ToolMessage(status="error"),Agent 看到拒绝原因并自行调整策略;
  4. 允许:透传给真实的工具处理器;
  5. Provider 异常fail_closed=true(默认):阻止该调用;fail_closed=false 时放行并记录警告;
  6. GraphBubbleUp 异常(LangGraph 的 interrupt/pause/resume 控制信号)始终向上抛出,绝不被捕获

拒绝时 Agent 看到的消息格式由 middleware.py_build_denied_message 定义:

return ToolMessage(
    content=f"Guardrail denied: tool '{tool_name}' was blocked ({reason_code}). Reason: {reason_text}. Choose an alternative approach.",
    tool_call_id=tool_call_id,
    name=tool_name,
    status="error",
)

这正对应文档中“Try it”示例里 Agent 看到的 Guardrail denied: tool 'bash' was blocked (oap.tool_not_allowed)。除返回消息外,源码还实现了两个文档未展开但生产上很关键的行为:每次决策都会通过 put_authorization_outcome 写入授权结果(供后续收据/审计链路消费),并通过 _record_guardrail_eventmiddleware:guardrail 标签尽力写入 RunJournal 审计事件——审计是 best-effort 的,失败只打警告、不改变工具执行行为(见 middleware.py)。

三种 Provider 选项

选项一:内置 AllowlistProvider(零依赖)

最简单的起步方案,随 DeerFlow 一同发布,按工具名做黑名单/白名单,无外部包、无护照、无网络。

config.yaml(按名封禁):

guardrails:
  enabled: true
  provider:
    use: deerflow.guardrails.builtin:AllowlistProvider
    config:
      denied_tools: ["bash", "write_file"]

这会封禁所有请求中的 bashwrite_file,其余工具放行。也可以改用白名单(仅允许列出的工具):

guardrails:
  enabled: true
  provider:
    use: deerflow.guardrails.builtin:AllowlistProvider
    config:
      allowed_tools: ["web_search", "read_file", "ls"]

实测步骤:

  1. 把上述配置加入 config.yaml
  2. 启动 DeerFlow:make dev
  3. 向 Agent 提问:“Use bash to run echo hello”;
  4. Agent 会看到:Guardrail denied: tool 'bash' was blocked (oap.tool_not_allowed)

builtin.py 的源码可以看到两个实现细节:

  • 判定顺序是先白名单后黑名单,两个名单返回的都是 oap.tool_not_allowed 码,但消息文本不同(not in allowlist vs is denied);
  • 构造函数刻意区分了“未配置白名单”(None → 全放行)与“显式空白名单”([] → 全拒绝):
def __init__(self, *, allowed_tools: list[str] | None = None, denied_tools: list[str] | None = None):
    # Distinguish "no allowlist configured" (None -> allow all) from an
    # explicitly empty allowlist ([] -> allow nothing).
    self._allowed = set(allowed_tools) if allowed_tools is not None else None
    self._denied = set(denied_tools) if denied_tools else set()

注释明确指出:如果用真值判断,空列表 [] 会被坍缩为 None 从而“fail open”——当运维意图是“一个工具都不允许”时,所有工具反而会全部通过。这是一个容易被忽略的安全语义细节。

选项二:OAP Passport Provider(策略驱动)

面向基于 Open Agent Passport(OAP) 开放标准的策略执行。OAP 护照是一个 JSON 文档,声明 Agent 的身份、能力与操作限额:

┌─────────────────────────────────────────────────────────────┐
│                    OAP Passport (JSON)                        │
│                   (open standard, any provider)              │
│  {                                                           │
│    "spec_version": "oap/1.0",                                │
│    "status": "active",                                       │
│    "capabilities": [                                         │
│      {"id": "system.command.execute"},                       │
│      {"id": "data.file.read"},                               │
│      {"id": "data.file.write"},                              │
│      {"id": "web.fetch"},                                    │
│      {"id": "mcp.tool.execute"}                              │
│    ],                                                        │
│    "limits": {                                               │
│      "system.command.execute": {                             │
│        "allowed_commands": ["git", "npm", "node", "ls"],     │
│        "blocked_patterns": ["rm -rf", "sudo", "chmod 777"]   │
│      }                                                       │
│    }                                                         │
│  }                                                           │
└──────────────────────────┬──────────────────────────────────┘

手动创建护照:OAP 护照就是一个 JSON 文件,可按 OAP 规范手写,并用 OAP 的 JSON Schema 校验;aport-spec 仓库的 examples 目录提供了模板。

以 APort 作为参考实现:APort Agent Guardrails 是 OAP 的开源(Apache 2.0)实现之一,覆盖护照创建、本地评估与可选的托管 API 评估:

pip install aport-agent-guardrails
aport setup --framework deerflow

该命令生成:

  • ~/.aport/deerflow/config.yaml — 评估器配置(本地或 API 模式)
  • ~/.aport/deerflow/aport/passport.json — 带能力与限额的 OAP 护照

config.yaml(使用 APort 作为 Provider):

guardrails:
  enabled: true
  provider:
    use: aport_guardrails.providers.generic:OAPGuardrailProvider

config.yaml(使用你自己的 OAP Provider):

guardrails:
  enabled: true
  provider:
    use: my_oap_provider:MyOAPProvider
    config:
      passport_path: ./my-passport.json

任何接受 framework 关键字参数并实现 evaluate/aevaluate 的 Provider 都能工作。OAP 标准定义护照格式与决策码;DeerFlow 不关心是哪个 Provider 在读取它们。

护照控制什么:

护照字段 作用 示例
capabilities[].id Agent 可用的工具类别 system.command.executedata.file.write
limits.*.allowed_commands 允许的命令 ["git", "npm", "node"],或 ["*"] 表示全部
limits.*.blocked_patterns 永远拒绝的模式 ["rm -rf", "sudo", "chmod 777"]
status 总开关 activesuspendedrevoked

评估模式(取决于 Provider 实现),以 APort 参考实现为例:

模式 工作方式 网络 延迟
Local 本地评估护照 约 300ms
API 发送护照 + 上下文到托管评估器,决策带签名 约 65ms

自定义 OAP Provider 可以实现任意评估策略——DeerFlow 中间件不关心 Provider 如何得出结论。

实测步骤: 安装配置后,先让 Agent “Create a file called test.txt with content hello”,再要求 “Now delete it using bash rm -rf”。Guardrail 会拦截:oap.blocked_pattern: Command contains blocked pattern: rm -rf

选项三:自定义 Provider(Bring Your Own)

任何具有 evaluate(request)aevaluate(request) 方法的 Python 类都可以直接用作 Provider——无需继承基类,这是结构性协议(见后文 Protocol 一节):

# my_guardrail.py

class MyGuardrailProvider:
    name = "my-company"

    def evaluate(self, request):
        from deerflow.guardrails.provider import GuardrailDecision, GuardrailReason

        # Example: block any bash command containing "delete"
        if request.tool_name == "bash" and "delete" in str(request.tool_input):
            return GuardrailDecision(
                allow=False,
                reasons=[GuardrailReason(code="custom.blocked", message="delete not allowed")],
                policy_id="custom.v1",
            )
        return GuardrailDecision(allow=True, reasons=[GuardrailReason(code="oap.allowed")])

    async def aevaluate(self, request):
        # This skeleton reuses the sync path. If policy evaluation performs
        # async I/O, call and await the async evaluator here instead.
        return self.evaluate(request)

config.yaml:

guardrails:
  enabled: true
  provider:
    use: my_guardrail:MyGuardrailProvider

注意 my_guardrail.py 必须在 Python 路径上(例如放在 backend 目录内,或安装为包)。

实测步骤: 创建 my_guardrail.py → 添加配置 → 启动后要求 “Use bash to delete test.txt” → 你的 Provider 将其拦截。

可选:运行归因字段(Runtime Attribution)

归因字段是可选的。需要更丰富策略上下文或审计记录的 Provider 可以读取它们,简单的工具 allow/deny Provider 可以直接忽略:

字段 示例用途
user_id 将已认证的 DeerFlow 用户关联到 Provider 侧策略或审计记录
user_role 应用简单基于角色的策略,如仅允许 admin 使用某工具。取自认证用户的 system_role(面向 guardrail 的字段改名,而非独立字段)
oauth_provider 决策关联到外部身份提供者(如存在)
oauth_id 决策关联到外部提供者的 subject/user id(如存在)
thread_id 决策关联回会话线程
run_id 决策关联回某次执行
tool_call_id 标识被允许/拒绝的那次具体工具调用

这些字段由 Gateway 从服务端认证状态填充(run worker 始终写入 thread_id/run_id):Web 认证请求中,inject_authenticated_user_contextrequest.state.user 写入 user_id/user_role/oauth_provider/oauth_id;对可信 IM / 内部认证请求(Slack、Discord、Telegram、Feishu、DingTalk 等提供可信 owner 头的内部调用方),Gateway 在服务端解析 owner 用户并写入同样的归因字段。客户端传入的值不能覆盖它们——服务端赋值优先。 若可信内部调用方解析不到 owner 用户,Gateway 会剥离客户端传入的 user_role/oauth_provider/oauth_id,而不是把它们当作权威值;已存在的 user_id 出于兼容旧渠道存储的行为保留,但角色/OAuth 策略只在服务端解析到 owner 后才生效。

如果部署有按用户维度的策略需求,可以启用“上下文感知 Provider”,把运行时字段传入外部策略文件——业务策略留在 Python 代码和 config.yaml 之外;Provider 只负责归一化上下文、评估策略并映射回 GuardrailDecision。文档给出了一个完整的示例骨架(此处保留其核心结构):

import asyncio
import json
from pathlib import Path

from deerflow.guardrails.provider import GuardrailDecision, GuardrailReason


class ContextAwareGuardrailProvider:
    """Illustrative provider skeleton; policy loading/evaluation is provider-defined."""

    name = "context-aware-example"

    def __init__(self, *, policy_path, audit_path="./logs/guardrail-audit.jsonl", **kwargs):
        self.policy_path = Path(policy_path)
        self.audit_path = Path(audit_path)
        # Load policy rules here. In a real deployment this could call an
        # internal policy service, OPA/Cedar, AGT, or another rule engine.
        self.policy = self._load_policy(self.policy_path)

    def evaluate(self, request):
        decision = self._decide(request)
        self._write_audit(request, decision)
        return decision

    async def aevaluate(self, request):
        # ``_decide`` is in-memory policy work; the audit write is blocking
        # file I/O, so offload it off the event loop with ``asyncio.to_thread``
        # (DeerFlow enforces a blocking-IO gate in CI).
        decision = self._decide(request)
        await asyncio.to_thread(self._write_audit, request, decision)
        return decision

    def _decide(self, request):
        # 1. Normalize DeerFlow request data into policy context.
        context = {
            "tool_name": request.tool_name,
            "tool_input": request.tool_input,
            "user_id": request.user_id,
            "user_role": request.user_role,
            "oauth_provider": request.oauth_provider,
            "oauth_id": request.oauth_id,
            "thread_id": request.thread_id,
            "run_id": request.run_id,
            "tool_call_id": request.tool_call_id,
            "agent_id": request.agent_id,
            "timestamp": request.timestamp,
            # Derived fields make simple rule engines handle multi-field checks.
            "role_tool_key": f"{request.user_role or ''}:{request.tool_name}",
            "command": request.tool_input.get("command", ""),
            "message": json.dumps(request.tool_input, ensure_ascii=False, default=str),
        }
        # 2. Evaluate the provider-defined policy schema.
        result = self._evaluate_policy(self.policy, context)
        # 3. Convert the policy result back to DeerFlow's decision object.
        return GuardrailDecision(
            allow=result["allow"],
            reasons=[GuardrailReason(code=result["code"], message=result["message"])],
            policy_id=result.get("policy_id"),
            metadata={
                "user_id": request.user_id,
                "user_role": request.user_role,
                "oauth_provider": request.oauth_provider,
                "oauth_id": request.oauth_id,
                "thread_id": request.thread_id,
                "run_id": request.run_id,
                "tool_call_id": request.tool_call_id,
            },
        )
    # _write_audit / _load_policy / _evaluate_policy 见原文档完整实现

这个骨架里有两条值得强调的工程约束:异步路径中的审计文件写入是阻塞 I/O,必须用 asyncio.to_thread 移出事件循环(DeerFlow 在 CI 中强制执行 blocking-IO 门禁);如果策略评估本身涉及阻塞 I/O(外部策略服务、每次调用读文件),同样要放进 asyncio.to_thread,或实现原生异步评估器。

对应的 config.yaml 与一份示例策略文件(示意性 YAML,具体 Schema 由 Provider 自定):

guardrails:
  enabled: true
  provider:
    use: my_guardrail:ContextAwareGuardrailProvider
    config:
      policy_path: ./policies/guardrail-policy.yml
      audit_path: ./logs/guardrail-audit.jsonl
# policies/guardrail-policy.yml
version: "1.0"
rules:
  - name: allow-admin-bash
    condition:
      field: role_tool_key
      operator: eq
      value: admin:bash
    action: allow
    priority: 300
    message: Admin users may execute bash

  - name: deny-bash-for-other-roles
    condition:
      field: tool_name
      operator: eq
      value: bash
    action: deny
    priority: 200
    message: bash is restricted to admin users

  - name: deny-dangerous-command
    condition:
      field: message
      operator: matches
      value: "\\brm\\s+-rf\\b"
    action: deny
    priority: 100
    message: Dangerous shell command detected

defaults:
  action: allow

多数策略引擎都是“归一化请求上下文 → 按序评估规则 → 返回 allow/deny”的形态,这个例子展示了如何把 DeerFlow 的运行时归因字段(如 user_role + tool_name 拼出的 role_tool_key)喂给简单规则引擎,实现“admin 才能用 bash”这类策略。

实现 Provider 所需的接口

必备接口

┌──────────────────────────────────────────────────┐
│              GuardrailProvider Protocol            │
│                                                   │
│  name: str                                        │
│                                                   │
│  evaluate(request: GuardrailRequest)              │
│      -> GuardrailDecision                         │
│                                                   │
│  aevaluate(request: GuardrailRequest)   (async)   │
│      -> GuardrailDecision                         │
└──────────────────────────────────────────────────┘

GuardrailRequestGuardrailDecision 的完整结构定义在 backend/packages/harness/deerflow/guardrails/provider.py@runtime_checkable Protocol,isinstance 检查可用):

┌──────────────────────────┐    ┌──────────────────────────┐
│     GuardrailRequest      │    │    GuardrailDecision      │
│                           │    │                           │
│  tool_name: str           │    │  allow: bool              │
│  tool_input: dict         │    │  reasons: [GuardrailReason]│
│  agent_id: str | None     │    │  policy_id: str | None    │
│  thread_id: str | None    │    │  metadata: dict           │
│  is_subagent: bool        │    │                           │
│  timestamp: str           │    │  GuardrailReason:         │
│  user_id: str | None      │    │    code: str              │
│  user_role: str | None    │    │    message: str           │
│  oauth_provider: str | None│   │                           │
│  oauth_id: str | None     │    │                           │
│  run_id: str | None       │    │                           │
│  tool_call_id: str | None │    │                           │
│                           │    │                           │
└──────────────────────────┘    └──────────────────────────┘

对照当前源码,GuardrailRequest 比文档图示还多出三个向后兼容字段(默认值保证旧 Provider 不受影响):channel_user_idis_internalauthz_attributes(经 normalize_authz_attributes 归一化后填入)。中间件在 middleware.py_build_request 中完成字段填充,其中 agent_id 取自配置的 passporttool_call_id 取自 LLM 生成的工具调用 id,timestamp 为 UTC ISO 时间戳。

DeerFlow 工具名清单

Provider 在 request.tool_name 中看到的工具名:

工具 作用
bash 执行 Shell 命令
write_file 创建/覆盖文件
str_replace 编辑文件(查找替换)
read_file 读取文件内容
ls 列目录
web_search Web 搜索
web_fetch 抓取 URL 内容
image_search 图片搜索
present_files 向用户展示文件
view_image 显示图片
ask_clarification 向用户提问
task 委派给子代理
mcp__* MCP 工具(动态名称)

OAP Reason 码

含义
oap.allowed 工具调用已授权
oap.tool_not_allowed 工具不在白名单
oap.command_not_allowed 命令不在 allowed_commands
oap.blocked_pattern 命令命中封禁模式
oap.limit_exceeded 操作超出限额
oap.passport_suspended 护照状态为 suspended/revoked
oap.evaluator_error Provider 崩溃(fail-closed 时拒绝)

Provider 加载机制

DeerFlow 通过 resolve_variable() 加载 Provider——与模型、工具、沙箱 Provider 相同的机制。use: 字段是 Python 类路径:package.module:ClassName

设置了 config: 时,Provider 用 **config 关键字参数实例化,且始终注入 framework="deerflow"(前提是构造函数可接收)。保持前向兼容的建议写法:

class YourProvider:
    def __init__(self, framework: str = "generic", **kwargs):
        # framework="deerflow" tells you which config dir to use
        ...

配置参考(Configuration Reference)

guardrails:
  # Enable/disable guardrail middleware (default: false)
  enabled: true

  # Block tool calls if provider raises an exception (default: true)
  fail_closed: true

  # Passport reference -- passed as request.agent_id to the provider.
  # File path, hosted agent ID, or null (provider resolves from its config).
  passport: null

  # Provider: loaded by class path via resolve_variable
  provider:
    use: deerflow.guardrails.builtin:AllowlistProvider
    config:  # optional kwargs passed to provider.__init__
      denied_tools: ["bash"]

配置模型定义在 backend/packages/harness/deerflow/config/guardrails_config.py,为 Pydantic BaseModel

字段 类型 默认值 说明
enabled bool False 是否启用 guardrail 中间件
fail_closed bool True Provider 抛异常时是否阻断工具调用
passport str | None None OAP 护照路径或托管 Agent ID,作为 request.agent_id 传给 Provider
provider.use str 必填(启用时) 类路径,如 deerflow.guardrails.builtin:AllowlistProvider
provider.config dict {} 传给 provider.__init__ 的可选关键字参数

模块还维护一个单例:get_guardrails_config() 在未加载时返回默认实例,load_guardrails_config_from_dict() 在 AppConfig 加载时调用,reset_guardrails_config() 供测试清理单例防止泄漏(见 guardrails_config.py)。config.example.yaml 中同样收录了三段 Provider 示例配置,与上文一一对应。

测试

cd backend
uv run python -m pytest tests/test_guardrail_middleware.py -v

文档写作时该测试文件 backend/tests/test_guardrail_middleware.py 收录 25 个用例,覆盖以下领域(当前仓库中该文件的用例数量已进一步增长,覆盖范围只增不减):

  • AllowlistProvider:允许、拒绝、白名单+黑名单并存、异步路径;
  • GuardrailMiddleware:允许透传、带 OAP 码的拒绝、fail-closed、fail-open、护照转发、空 reasons 回退、空工具名、协议 isinstance 检查;
  • 异步路径awrap_tool_call 的允许、拒绝、fail-closed、fail-open;
  • GraphBubbleUp:LangGraph 控制信号穿透中间件(不被捕获);
  • Config:默认值、from_dict、单例加载/重置。

文件地图

packages/harness/deerflow/guardrails/
    __init__.py              # 公共导出
    provider.py              # GuardrailProvider 协议、GuardrailRequest、GuardrailDecision
    middleware.py             # GuardrailMiddleware(AgentMiddleware 子类)
    builtin.py               # AllowlistProvider(零依赖)

packages/harness/deerflow/config/
    guardrails_config.py     # GuardrailsConfig Pydantic 模型 + 单例

packages/harness/deerflow/agents/middlewares/
    tool_error_handling_middleware.py  # 在中间件链中注册 GuardrailMiddleware

config.example.yaml          # 三种 Provider 选项示例
tests/test_guardrail_middleware.py  # 单元测试
docs/GUARDRAILS.md           # 本文对应的原始文档

以上路径的仓库根相对形式分别为 backend/packages/harness/deerflow/guardrails/provider.pybackend/packages/harness/deerflow/guardrails/middleware.pybackend/packages/harness/deerflow/guardrails/builtin.pybackend/packages/harness/deerflow/config/guardrails_config.pybackend/packages/harness/deerflow/agents/middlewares/tool_error_handling_middleware.pyconfig.example.yamlbackend/tests/test_guardrail_middleware.py

小结:为 DeerFlow 选择授权方案

  • 快速止血AllowlistProvider + denied_tools 两行配置即可封禁高危工具,零依赖;注意 allowed_tools: [] 是“全拒绝”而非“全放行”的语义陷阱;
  • 策略化治理:采用 OAP 护照(capabilities + limits + status),配合本地或 API 评估模式,护照即总开关(suspended/revoked 一键停摆);
  • 深度集成:自定义 Provider 可读取 user_role/thread_id/run_id 等运行归因字段,把组织级业务策略(角色、身份、审计)落到外部规则引擎,DeerFlow 只负责请求归一化与决策执行;
  • 默认安全姿态fail_closed: true 保证 Provider 故障时宁可拒绝也不放行,GraphBubbleUp 控制信号则永远穿透,不干扰 LangGraph 的 interrupt/resume 流程;
  • 可观测性:每次决策都会写入授权结果并尽力落入 RunJournal(middleware:guardrail 标签),便于事后审计。

Guardrails 让 DeerFlow 的自主多步任务从“有能力但无边界”变为“能力与策略边界同步交付”——这正是长时程 Agent 走向生产部署的关键一环。

登录后查看全文
热门项目推荐
相关项目推荐