从字节跳动 DeerFlow 源码看 Agent 平台设计（二）：工具系统设计 — 从全量绑定到按需加载

qhvssonic

156人浏览 · 2026-06-14 08:30:00

qhvssonic · 2026-06-14 08:30:00 发布

系列导航

第一篇：什么是 Agent？一个成熟 Agent 平台的 8 个核心组件

本篇：工具系统设计 — 从全量绑定到按需加载

第三篇：五个核心中间件深度解析

第四篇：Agent 生命周期与状态管理

摘要

当 Agent 接入大量 MCP 工具时，将所有工具的完整 JSON Schema 一次性绑定给模型会导致严重的 token 浪费和选择困难问题。本文分析 DeerFlow 如何通过 tool_search 延迟加载机制实现"仅暴露名称、按需加载 schema"的工具发现策略，并讨论该方案在业界的独特性。

一、三层工具架构

DeerFlow 的工具系统采用三层来源架构，由 get_available_tools() 函数统一合并：

1.1 Built-in Tools（内置工具）

代码级内置，提供 Agent 运行时的基础能力：

BUILTIN_TOOLS = [
    present_file_tool,       # 向用户展示文件
    ask_clarification_tool,  # 向用户发起澄清提问
]

SUBAGENT_TOOLS = [
    task_tool,               # 委派任务给子 Agent
]

此外还有条件性内置工具：

view_image_tool：仅当模型配置 supports_vision: true 时加载
skill_manage_tool：仅当 skill_evolution.enabled 时加载
update_agent：仅自定义 Agent（非默认 Agent）时加载

1.2 Configured Tools（配置工具）

通过 config.yaml 声明，启动时通过反射加载：

tools:
  - name: web_search
    use: deerflow.community.search.duckduckgo:ddg_search_tool
    group: search
  - name: bash
    use: deerflow.sandbox.tools:bash_tool
    group: bash
  - name: read_file
    use: deerflow.sandbox.tools:read_file_tool
    group: file

resolve_variable(cfg.use, BaseTool) 将 "package.module:variable" 格式的字符串解析为实际的工具实例。DeerFlow 会校验配置中的 name 字段与工具实例的 .name 属性是否一致，不一致时输出警告日志。

1.3 MCP Tools（协议工具）

通过 MCP（Model Context Protocol）服务器动态加载。MCP 服务器配置在 extensions_config.json 中：

{
  "mcpServers": {
    "github": {
      "enabled": true,
      "type": "stdio",
      "command": "npx",
      "args": ["-y", "@modelcontextprotocol/server-github"],
      "env": {"GITHUB_TOKEN": "$GITHUB_TOKEN"}
    }
  }
}

MCP 工具通过 langchain-mcp-adapters 库适配为 LangChain 的 BaseTool 接口，支持 stdio、SSE、HTTP 三种传输方式。工具列表带有文件 mtime 缓存失效机制——当配置文件变更时自动重新初始化 MCP 客户端。

1.4 工具合并与去重

all_tools = loaded_tools + builtin_tools + mcp_tools + acp_tools
# 按名称去重，优先级：config > builtin > MCP > ACP
seen_names: set[str] = set()
unique_tools: list[BaseTool] = []
for t in all_tools:
    if t.name not in seen_names:
        unique_tools.append(t)
        seen_names.add(t.name)

二、全量绑定的问题

在标准的 LangChain/LangGraph 实现中，所有可用工具的完整 JSON Schema 通过 model.bind_tools(tools) 一次性发送给模型。当工具数量较少时（10 个以内），这种方式没有问题。

但当接入多个 MCP 服务器时（GitHub、Postgres、Slack、Puppeteer 等），每个服务器可能暴露 5-20 个工具，总计可达 50-100 个。此时全量绑定面临三个问题：

2.1 Token 浪费

每个工具的 JSON Schema 包含名称、描述、参数定义（类型、必填、枚举值等），平均占 200-500 token。50 个工具即占用 10000-25000 token 的上下文空间，严重挤压实际对话内容的空间。

2.2 模型选择困难

研究表明，当可选工具过多时，LLM 的工具选择准确率会下降。模型可能会：

选择功能相似但不正确的工具
产生"幻觉调用"——调用参数格式正确但逻辑错误的工具
在多个候选工具之间犹豫，增加推理 token 消耗

2.3 前缀缓存失效

LLM 推理服务通常支持 KV Cache 前缀缓存——如果系统提示词（包含工具 schema）不变，则可复用先前的 KV Cache 计算结果。工具列表的频繁变动（如 MCP 服务器重启后工具列表变化）会导致缓存失效，增加推理延迟。

三、DeerFlow 的 tool_search 延迟加载方案

DeerFlow 设计了一套完整的延迟加载机制，由四个协作组件构成。

3.1 构建时：识别与分类

在 Agent 构建阶段（_make_lead_agent()），经过技能策略过滤后，调用 build_deferred_tool_setup() 对工具进行分类：

def build_deferred_tool_setup(filtered_tools, *, enabled: bool) -> DeferredToolSetup:
    if not enabled:
        return DeferredToolSetup(None, frozenset(), None)
    
    # 仅 MCP 工具被标记为延迟加载
    deferred = [t for t in filtered_tools if _is_mcp_tool(t)]
    if not deferred:
        return DeferredToolSetup(None, frozenset(), None)
    
    catalog = DeferredToolCatalog(tuple(deferred))
    return DeferredToolSetup(
        tool_search_tool=build_tool_search_tool(catalog),
        deferred_names=catalog.names,
        catalog_hash=catalog.hash
    )

识别依据是工具的 metadata 标记——在 get_available_tools() 中，MCP 来源的工具会被打上 {"deerflow_mcp": True} 标签：

for t in mcp_tools:
    t.metadata = {**(t.metadata or {}), "deerflow_mcp": True}

3.2 系统提示词：仅列名称

get_deferred_tools_prompt_section() 生成一段包含在系统提示词中的内容：

<available-deferred-tools>
github_create_issue
github_list_repos
postgres_query
slack_send_message
</available-deferred-tools>

模型能看到这些工具的名称（知道它们存在），但没有参数 schema，因此无法直接调用。

3.3 运行时中间件：拦截未授权调用

DeferredToolFilterMiddleware 承担两项职责：

职责一：过滤模型绑定的工具 schema

def _filter_tools(self, request: ModelRequest) -> ModelRequest:
    hide = self._hidden(request.state)  # 延迟集 - 已提升集
    active = [t for t in request.tools if t.name not in hide]
    return request.override(tools=active)

即使 ToolNode 持有所有工具实例（用于执行），模型通过 bind_tools() 看到的 schema 只包含活跃工具和已提升工具。

职责二：拦截越权调用

def _blocked_tool_message(self, request: ToolCallRequest) -> ToolMessage | None:
    name = request.tool_call.get("name")
    if name in self._hidden(request.state):
        return ToolMessage(
            content=f"Error: Tool '{name}' is deferred and has not been promoted yet. "
                    f"Call tool_search first.",
            status="error"
        )

如果模型凭记忆尝试直接调用未提升的工具，系统返回错误而非执行，引导模型先通过 tool_search 获取 schema。

3.4 提升机制：tool_search 解锁

当模型调用 tool_search 工具时：

@tool
def tool_search(query: str, tool_call_id: Annotated[str, InjectedToolCallId]) -> Command:
    matched = catalog.search(query)[:MAX_RESULTS]
    
    content = json.dumps([convert_to_openai_function(t) for t in matched])
    names = [t.name for t in matched]
    
    return Command(
        update={
            "promoted": {"catalog_hash": catalog_hash, "names": names},
            "messages": [ToolMessage(content=content, tool_call_id=tool_call_id)]
        }
    )

Command(update={"promoted": ...}) 将提升记录写入图状态。ThreadState 的 merge_promoted reducer 处理合并逻辑：

def merge_promoted(existing, new):
    if existing.get("catalog_hash") != new["catalog_hash"]:
        return new  # catalog 变更 → 替换（防漂移）
    return {
        "catalog_hash": existing["catalog_hash"],
        "names": list(dict.fromkeys(existing["names"] + new["names"]))  # 去重合并
    }

下一轮模型调用时，中间件检测到该工具已提升，不再隐藏其 schema，模型即可正常调用。

四、搜索能力设计

DeferredToolCatalog 支持三种查询模式：

4.1 精确选取

select:github_create_issue,postgres_query

直接按名称选取指定工具，适合模型已明确知道需要哪个工具的场景。

4.2 前缀 + 关键词排序

+slack send

要求名称必须包含 slack，再按 send 关键词对候选排序。

4.3 正则搜索

notebook jupyter

对工具名称和描述做正则匹配，名称匹配权重（2）高于描述匹配（1），返回得分最高的前 5 个结果。

def search(self, query: str) -> list[BaseTool]:
    try:
        regex = re.compile(query, re.IGNORECASE)
    except re.error:
        regex = re.compile(re.escape(query), re.IGNORECASE)
    
    scored = []
    for t in self.tools:
        searchable = f"{t.name} {t.description or ''}"
        if regex.search(searchable):
            scored.append((2 if regex.search(t.name) else 1, t))
    scored.sort(key=lambda x: x[0], reverse=True)
    return [t for _, t in scored][:MAX_RESULTS]

五、安全设计

5.1 Catalog Hash 防漂移

每次构建目录时计算所有工具 schema 的 SHA256 哈希：

@cached_property
def hash(self) -> str:
    canon = [{"name": t.name, "schema": convert_to_openai_function(t)} 
             for t in sorted(self.tools, key=lambda t: t.name)]
    blob = json.dumps(canon, sort_keys=True, ensure_ascii=False, default=str)
    return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16]

merge_promoted 在合并时校验 hash——如果 MCP 服务器重启后工具列表发生了变化（名称相同但参数变了），旧的 promoted 记录自动失效，防止暴露已不存在或已变更的工具。

5.2 Fail-closed 策略

def _assemble_deferred(filtered_tools, *, enabled: bool):
    setup = build_deferred_tool_setup(filtered_tools, enabled=enabled)
    if enabled and not setup.deferred_names and any(_is_mcp_tool(t) for t in filtered_tools):
        raise RuntimeError(
            "tool_search enabled and MCP tools survived policy filtering, "
            "but no deferred set was recovered — refusing to bind MCP schemas (fail-closed)."
        )

如果配置启用了 tool_search 但目录构建失败（逻辑异常），系统拒绝启动而非退化为全量暴露。

5.3 每次最多 5 个

MAX_RESULTS = 5 限制每次 tool_search 最多返回 5 个工具 schema，避免模型通过一次搜索获取全量 schema（绕过延迟加载的初衷）。

六、完整数据流

┌────────────────────────────────────────────────────────────────┐
│                      Agent 构建阶段                             │
│                                                                │
│  所有工具（config + builtin + MCP）                             │
│       ↓ 技能策略过滤                                           │
│  filtered_tools                                                │
│       ↓ build_deferred_tool_setup()                            │
│  ┌─────────────────────────────────────────────────────────┐   │
│  │ DeferredToolSetup:                                      │   │
│  │   deferred_names = {tool_a, tool_b, tool_c, ...}        │   │
│  │   catalog_hash = "abc123..."                            │   │
│  │   tool_search_tool = <closure over DeferredToolCatalog> │   │
│  └─────────────────────────────────────────────────────────┘   │
│       ↓                                                        │
│  final_tools = filtered_tools + [tool_search_tool]             │
│  system_prompt 包含 <available-deferred-tools> 名称列表         │
└────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌────────────────────────────────────────────────────────────────┐
│                    运行时 — 第 1 轮                              │
│                                                                │
│  DeferredToolFilterMiddleware:                                  │
│    hidden = deferred_names - promoted = {tool_a, tool_b, ...}  │
│    bind_tools → 只暴露非延迟工具 + tool_search                  │
│                                                                │
│  模型看到系统提示词中的工具名称列表                               │
│  模型决定需要 tool_a → 调用 tool_search("select:tool_a")        │
└────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌────────────────────────────────────────────────────────────────┐
│                    tool_search 执行                              │
│                                                                │
│  1. catalog.search("select:tool_a") → 找到 tool_a 实例         │
│  2. 序列化为 JSON Schema                                       │
│  3. Command(update={"promoted": {hash, ["tool_a"]}})           │
│     → 写入 ThreadState.promoted                                │
│  4. 返回 ToolMessage(content=完整 schema JSON)                  │
└────────────────────────────────────────────────────────────────┘
                              │
                              ▼
┌────────────────────────────────────────────────────────────────┐
│                    运行时 — 第 2 轮                              │
│                                                                │
│  DeferredToolFilterMiddleware:                                  │
│    hidden = deferred_names - {tool_a} = {tool_b, tool_c, ...}  │
│    bind_tools → 暴露非延迟工具 + tool_search + tool_a           │
│                                                                │
│  模型正常调用: tool_a(param1="xxx", param2="yyy")               │
│  ToolNode 执行 → 返回结果                                      │
└────────────────────────────────────────────────────────────────┘

七、与业界方案的对比

方案	工具发现方式	适用场景
LangChain 标准	全量 bind_tools	工具数 < 15
CrewAI	全量绑定 + 角色分工减少每个 Agent 可见工具	多 Agent 协作
AutoGen	全量注册	工具数可控
MCP 协议	`tools/list` 一次性返回所有工具	协议层面无按需发现
OpenAI function calling	全量声明	讨论过 pagination 但未公开实现
DeerFlow tool_search	名称暴露 + 按需搜索获取 schema	MCP 工具数量大

DeerFlow 的 tool_search 方案在当前开源 Agent 框架中尚属独创。其本质类似操作系统的虚拟内存/页面置换——工具"存在但不加载"，直到实际需要时才"换入"上下文窗口。

八、配置与启用

该机制通过 config.yaml 控制：

tool_search:
  enabled: true   # 默认 false，渐进推广中

当前默认关闭（enabled: false），说明该特性仍在渐进推广阶段。对于 MCP 工具数量有限（< 15 个）的部署场景，全量绑定仍是更简单的选择。

下一篇预告

第三篇将深入分析 DeerFlow 的五个核心中间件：SandboxMiddleware（执行隔离）、SummarizationMiddleware（上下文管理）、MemoryMiddleware（长期记忆）、ClarificationMiddleware（澄清中断）、SafetyFinishReasonMiddleware（安全防护），探讨每个中间件的设计动机、实现机制和关键决策。