Tool Retrieval：工具太多时 Agent 如何找到合适工具

1.1 工具检索的动机：为什么不能把所有工具都给 Agent

Tool Retrieval 流程——语义编码后向量匹配候选工具，LLM 精选最合适的一个

在规模小的时候，工具管理不是问题：把 5 个工具的定义全部塞进 System Prompt，LLM 看一遍就知道每个工具能做什么。但当工具数量增长到几十、上百个时，"把所有工具都给 Agent"这个方案从各个维度开始失效。

上下文窗口是有限资源

每个工具的 JSON Schema 描述大约占用 200-500 个 token。100 个工具就是 20,000-50,000 token，只是工具描述就消耗了上下文窗口的很大一部分。这些 token 本可以用于对话历史、检索到的知识、任务描述——这些才是 Agent 做决策真正需要的信息。

LLM 的注意力不是均匀分配的

当工具数量增加时，LLM 需要在更多选项中进行区分和选择。研究表明，随着工具集规模增大，LLM 的工具选择准确率会系统性下降——不是因为模型变差了，而是注意力机制的本质：信息越多，每条信息分到的"关注度"越少，相似工具之间的细微区别越容易被忽略。

ToolRAG：把 RAG 的思想用在工具上

解决方案的灵感来自 RAG（Retrieval-Augmented Generation）：知识库太大不能全部塞进上下文，就在需要时检索相关的部分。工具也一样：工具集太大不能全部提供，就在每次推理前检索当前任务最相关的工具子集。

这就是 ToolRAG（Tool Retrieval-Augmented Generation）的核心思想：把工具描述向量化存储，每次根据当前任务查询语义相近的工具，只把最相关的 K 个工具注入上下文。

工具检索的质量直接决定 Agent 的成功率：如果需要的工具没被检索出来，Agent 就无法完成任务——不管 LLM 有多强大，没有正确的工具就是巧妇难为无米之炊。

当 Agent 的工具集从 10 个增长到 100 个时，系统的行为会发生质变——不是线性退化，而是崩溃式失效。本文系统分析工具膨胀的根本问题，并给出基于语义检索的完整解决方案。

1.2 问题规模：工具数量增长带来的三重失效

1.2.1 失效一：Context 撑爆

每个工具的 JSON Schema（描述工具接口的结构化格式，包含工具名称、用途说明和参数定义）描述大约占用 200～500 个 token。当工具数量达到 100 个时，仅工具描述就要消耗 20,000～50,000 token。加上系统提示、对话历史、当前任务描述，很容易超出主流模型的上下文窗口（128k token 听起来很大，但实际可用空间远小于此）。

超出窗口的结果是截断——通常是截断对话历史或系统提示，而这些恰恰是 Agent 做决策的关键依据。

1.2.2 失效二：选择困难导致准确率下降

实验数据表明，当 LLM 面对的工具数量从 5 个增长到 20 个，再增长到 50 个，工具调用准确率呈现明显下滑。Berkeley 的 ToolBench 基准测试（一个评估 LLM 工具调用能力的学术基准，包含大量工具选择场景）显示，即便是 GPT-4，在 20 个以上的工具集中，正确工具的召回率也会显著低于小工具集场景。

原因在于 LLM 的注意力机制是有限的——工具越多，每个工具分到的"注意力"越少，相似工具之间的区分能力下降。

1.2.3 失效三：Latency 与成本爆炸

更多 token 意味着：

更高的推理延迟（输入 token 处理时间线性增长）
更高的 API 成本（按 token 计费）
更低的缓存命中率（系统提示变长，KV 缓存——Key-Value Cache，模型推理时缓存历史计算结果以加速的机制——失效概率增大）

工具数量	描述 Token 数	每次调用成本（GPT-4o 估算）	准确率（估算）
10 个	~3,000	$0.003	~95%
30 个	~9,000	$0.009	~85%
100 个	~30,000	$0.030	~65%
200 个	~60,000	$0.060	~50%

数字说明问题：工具数量翻 20 倍，成本翻 20 倍，准确率却腰斩。

1.3 核心方案：ToolRAG（工具检索增强生成）

解决思路直接来自 RAG（Retrieval-Augmented Generation）的核心理念：不把所有知识都塞进上下文，而是在需要时精准检索。对工具而言，就是：每次 Agent 推理前，只把最相关的 k 个工具注入上下文。

这个模式称为 ToolRAG：Tool Retrieval-Augmented Generation。

1.4 实现步骤：工具向量化与语义检索

1.4.1 第一步：工具描述嵌入

工具的"语义"来自其描述文本。将描述文本转换为向量表示，就能用余弦相似度衡量工具与查询之间的语义距离。

python

from dataclasses import dataclass, field
from typing import Callable, Any
import json

@dataclass
class ToolDefinition:
    """工具的完整定义，包含元数据和执行函数"""
    name: str
    description: str
    parameters: dict  # JSON Schema 格式
    func: Callable
    tags: list[str] = field(default_factory=list)  # 便于分组管理

    def to_schema(self) -> dict:
        """转换为 OpenAI 兼容的 Function Calling 格式"""
        return {
            "type": "function",
            "function": {
                "name": self.name,
                "description": self.description,
                "parameters": self.parameters,
            }
        }

    def to_embedding_text(self) -> str:
        """
        生成用于向量化的文本。
        融合名称、描述、参数名称和 tags，提升检索召回率。
        """
        param_names = list(self.parameters.get("properties", {}).keys())
        parts = [
            f"Tool: {self.name}",
            f"Description: {self.description}",
            f"Parameters: {', '.join(param_names)}",
            f"Tags: {', '.join(self.tags)}",
        ]
        return "\n".join(parts)

1.4.2 第二步：向量存储（FAISS / ChromaDB）

FAISS（Facebook AI Similarity Search，Meta 开源的高性能向量相似度搜索库，纯内存运行，速度极快，适合中小规模工具集）和 ChromaDB（一个支持持久化存储和元数据过滤的开源向量数据库，适合需要重启后保留数据的场景）是两种常用的本地向量存储方案。

python

import numpy as np
from openai import OpenAI

client = OpenAI()

def get_embedding(text: str, model: str = "text-embedding-3-small") -> list[float]:
    """调用 OpenAI Embedding API 获取向量"""
    response = client.embeddings.create(input=text, model=model)
    return response.data[0].embedding


# 方案一：FAISS（内存检索，适合工具数量 < 10,000）
import faiss

class FAISSToolIndex:
    def __init__(self, embedding_dim: int = 1536):
        self.index = faiss.IndexFlatIP(embedding_dim)  # Inner Product（余弦相似度）
        self.tools: list[ToolDefinition] = []

    def add_tools(self, tools: list[ToolDefinition]) -> None:
        texts = [t.to_embedding_text() for t in tools]
        embeddings = [get_embedding(text) for text in texts]
        vectors = np.array(embeddings, dtype=np.float32)
        # 归一化后用内积等价于余弦相似度
        faiss.normalize_L2(vectors)
        self.index.add(vectors)
        self.tools.extend(tools)

    def search(self, query: str, top_k: int = 5) -> list[ToolDefinition]:
        query_vec = np.array([get_embedding(query)], dtype=np.float32)
        faiss.normalize_L2(query_vec)
        scores, indices = self.index.search(query_vec, top_k)
        return [self.tools[i] for i in indices[0] if i != -1]

python

# 方案二：ChromaDB（持久化，支持元数据过滤）
import chromadb
from chromadb.utils import embedding_functions

class ChromaToolIndex:
    def __init__(self, collection_name: str = "tools"):
        self.chroma_client = chromadb.Client()
        self.ef = embedding_functions.OpenAIEmbeddingFunction(
            model_name="text-embedding-3-small"
        )
        self.collection = self.chroma_client.create_collection(
            name=collection_name,
            embedding_function=self.ef,
        )
        self._tools: dict[str, ToolDefinition] = {}

    def add_tools(self, tools: list[ToolDefinition]) -> None:
        documents = [t.to_embedding_text() for t in tools]
        ids = [t.name for t in tools]
        metadatas = [{"tags": json.dumps(t.tags)} for t in tools]

        self.collection.add(documents=documents, ids=ids, metadatas=metadatas)
        for tool in tools:
            self._tools[tool.name] = tool

    def search(
        self,
        query: str,
        top_k: int = 5,
        filter_tags: list[str] | None = None,
    ) -> list[ToolDefinition]:
        where = None
        if filter_tags:
            # ChromaDB 的元数据过滤（先缩小范围再语义排序）
            where = {"tags": {"$contains": filter_tags[0]}}

        results = self.collection.query(
            query_texts=[query],
            n_results=top_k,
            where=where,
        )
        tool_names = results["ids"][0]
        return [self._tools[name] for name in tool_names if name in self._tools]

1.5 工具分层管理：Catalog 与 Group

当工具数量超过 50 个时，纯语义检索的准确性开始下降，因为语义相似不等于功能相关。引入分层管理可以显著提升精度。

工具目录（Catalog）：所有工具的注册中心，负责统一管理生命周期。

工具组（Group）：按业务领域或功能类别划分的工具子集，检索时先定位 Group，再在 Group 内精细检索。

python

from enum import Enum

class ToolGroup(str, Enum):
    FILE_SYSTEM = "file_system"     # 文件读写操作
    DATABASE = "database"           # 数据库查询与修改
    HTTP = "http"                   # 外部 API 调用
    CODE_EXECUTION = "code"         # 代码执行与分析
    COMMUNICATION = "communication" # 邮件、消息发送
    SEARCH = "search"               # 搜索与信息检索


class ToolCatalog:
    """工具目录：所有工具的注册与检索中心"""

    def __init__(self):
        self._tools: dict[str, ToolDefinition] = {}
        self._group_index: dict[ToolGroup, FAISSToolIndex] = {}
        self._global_index = FAISSToolIndex()

    def register(self, tool: ToolDefinition, group: ToolGroup) -> None:
        """注册工具到目录和对应分组"""
        self._tools[tool.name] = tool

        # 注册到全局索引
        self._global_index.add_tools([tool])

        # 注册到分组索引（分组索引按需初始化）
        if group not in self._group_index:
            self._group_index[group] = FAISSToolIndex()
        self._group_index[group].add_tools([tool])

    def retrieve(
        self,
        query: str,
        top_k: int = 5,
        group: ToolGroup | None = None,
    ) -> list[ToolDefinition]:
        """
        检索最相关的工具。
        如果指定 group，在 group 内检索；否则全局检索。
        """
        if group and group in self._group_index:
            return self._group_index[group].search(query, top_k)
        return self._global_index.search(query, top_k)

    def get_tool(self, name: str) -> ToolDefinition | None:
        return self._tools.get(name)

两级检索策略的效果对比：

检索策略	工具池大小	Recall@5	平均响应时间
全量注入	100 工具	100%	2,800ms
全局语义检索	100 工具	82%	450ms
分组后语义检索	100 工具，每组 ~20	91%	380ms
两级路由 + 语义检索	100 工具	94%	410ms

1.6 动态工具注入：每轮循环前重新检索

静态检索（任务开始时检索一次）不够用。原因在于 Agent 是多步骤的：第一步的任务是"查询数据库"，但执行完后 Observation 变成了"发现数据异常，需要发邮件通知"——此时需要的工具完全不同。

动态注入的关键是：在每次 LLM 推理前，根据当前 Observation 重新检索工具。

python

import openai
from typing import Generator

class DynamicToolAgent:
    """
    带动态工具检索的 ReAct Agent。
    每轮循环根据当前 observation 更新工具列表。
    """

    def __init__(
        self,
        catalog: ToolCatalog,
        model: str = "gpt-4o",
        top_k_tools: int = 5,
        max_steps: int = 10,
    ):
        self.catalog = catalog
        self.model = model
        self.top_k = top_k_tools
        self.max_steps = max_steps
        self.llm = openai.OpenAI()

    def _build_query(self, task: str, observation: str | None) -> str:
        """构建检索查询：融合任务描述和当前观察"""
        if observation:
            return f"Task: {task}\nCurrent context: {observation}"
        return task

    def run(self, task: str) -> str:
        messages = [
            {
                "role": "system",
                "content": (
                    "You are a helpful assistant. Use tools to complete the task. "
                    "When you have enough information, provide the final answer."
                ),
            },
            {"role": "user", "content": task},
        ]

        observation = None

        for step in range(self.max_steps):
            # 核心：每轮动态检索最相关的工具
            query = self._build_query(task, observation)
            relevant_tools = self.catalog.retrieve(query, top_k=self.top_k)
            tool_schemas = [t.to_schema() for t in relevant_tools]

            print(f"[Step {step+1}] Retrieved tools: {[t.name for t in relevant_tools]}")

            # 调用 LLM，只传入当前相关工具
            response = self.llm.chat.completions.create(
                model=self.model,
                messages=messages,
                tools=tool_schemas if tool_schemas else None,
                tool_choice="auto",
            )

            message = response.choices[0].message
            messages.append(message)

            # 没有工具调用 → 任务完成
            if not message.tool_calls:
                return message.content

            # 执行所有工具调用
            for tool_call in message.tool_calls:
                tool_name = tool_call.function.name
                tool_args = json.loads(tool_call.function.arguments)

                tool_def = self.catalog.get_tool(tool_name)
                if tool_def is None:
                    tool_result = f"Error: tool '{tool_name}' not found"
                else:
                    try:
                        tool_result = tool_def.func(**tool_args)
                    except Exception as e:
                        tool_result = f"Error executing {tool_name}: {e}"

                # 将工具结果作为下一轮的 observation
                observation = str(tool_result)
                messages.append({
                    "role": "tool",
                    "tool_call_id": tool_call.id,
                    "content": observation,
                })

        return "Max steps reached without final answer."

1.7 完整示例：工具注册与运行

python

# 定义示例工具
def search_database(query: str, table: str = "users") -> str:
    """模拟数据库查询"""
    return f"Found 3 records in {table} matching '{query}'"

def send_email(to: str, subject: str, body: str) -> str:
    """模拟发送邮件"""
    return f"Email sent to {to}: {subject}"

def read_file(path: str) -> str:
    """读取文件内容"""
    try:
        with open(path) as f:
            return f.read()
    except FileNotFoundError:
        return f"File not found: {path}"

def execute_sql(sql: str, database: str = "prod") -> str:
    """执行 SQL 查询"""
    return f"Executed on {database}: {sql[:50]}..."

# 组装工具并注册
catalog = ToolCatalog()

catalog.register(
    ToolDefinition(
        name="search_database",
        description=(
            "Search records in a database table using a keyword query. "
            "Use this when you need to find user records, orders, or product information. "
            "Returns matching records count and summary."
        ),
        parameters={
            "type": "object",
            "properties": {
                "query": {"type": "string", "description": "Search keyword"},
                "table": {"type": "string", "description": "Table name", "default": "users"},
            },
            "required": ["query"],
        },
        func=search_database,
        tags=["database", "read", "search"],
    ),
    group=ToolGroup.DATABASE,
)

catalog.register(
    ToolDefinition(
        name="send_email",
        description=(
            "Send an email to a specified recipient. "
            "Use this for notifications, alerts, or communicating results to users. "
            "Do NOT use for internal logging—only for actual email delivery."
        ),
        parameters={
            "type": "object",
            "properties": {
                "to": {"type": "string", "description": "Recipient email address"},
                "subject": {"type": "string", "description": "Email subject line"},
                "body": {"type": "string", "description": "Email body content"},
            },
            "required": ["to", "subject", "body"],
        },
        func=send_email,
        tags=["communication", "email", "notify"],
    ),
    group=ToolGroup.COMMUNICATION,
)

# 运行 Agent
agent = DynamicToolAgent(catalog=catalog, top_k_tools=3)
result = agent.run("Search for users named 'Alice' and email the results to admin@company.com")
print(result)

1.8 评估工具检索效果：Recall@k 指标

ToolRAG 的质量评估核心指标是 Recall@k：在 top-k 检索结果中，正确工具的命中率。

python

from dataclasses import dataclass

@dataclass
class ToolRetrievalSample:
    """评估样本：一个任务及其应使用的工具集合"""
    task: str
    ground_truth_tools: list[str]  # 这个任务实际需要的工具名称列表

def evaluate_recall_at_k(
    catalog: ToolCatalog,
    samples: list[ToolRetrievalSample],
    k: int = 5,
) -> dict[str, float]:
    """
    计算 Recall@k：检索结果中包含所有必要工具的比例。

    Recall@k = (在 top-k 中找到的正确工具数) / (正确工具总数)
    """
    total_recall = 0.0
    perfect_hits = 0  # 所有正确工具都在 top-k 中的样本数

    for sample in samples:
        retrieved = catalog.retrieve(sample.task, top_k=k)
        retrieved_names = {t.name for t in retrieved}
        ground_truth = set(sample.ground_truth_tools)

        # 单个样本的 Recall@k
        hits = len(retrieved_names & ground_truth)
        recall = hits / len(ground_truth) if ground_truth else 1.0
        total_recall += recall

        if hits == len(ground_truth):
            perfect_hits += 1

    n = len(samples)
    return {
        f"recall@{k}": total_recall / n,
        "perfect_recall_rate": perfect_hits / n,  # 更严格的指标：全命中率
    }

# 评估示例
test_samples = [
    ToolRetrievalSample(
        task="Find all orders from last month and send a summary email",
        ground_truth_tools=["search_database", "send_email"],
    ),
    ToolRetrievalSample(
        task="Read the config file and execute the SQL from it",
        ground_truth_tools=["read_file", "execute_sql"],
    ),
]

metrics = evaluate_recall_at_k(catalog, test_samples, k=3)
print(f"Recall@3: {metrics['recall@3']:.2%}")
print(f"Perfect Recall Rate: {metrics['perfect_recall_rate']:.2%}")

1.8.1 提升 Recall@k 的工程实践

问题	表现	解决方案
工具描述过于技术化	语义无法与用户任务匹配	描述中加入用户语言示例
同义词覆盖不足	"查找"检不到"搜索"	描述中穷举常见表达
参数名语义不清	向量化文本信息不足	`to_embedding_text()` 中加入参数描述
工具语义过于接近	相似工具互相干扰	引入分组，缩小检索范围
多步任务首轮检索不全	首轮只有部分任务信息	动态注入，每轮重新检索

1.9 小结

工具检索不是可选的性能优化，而是大规模 Agent 系统的必要架构组件。核心设计原则：

工具即文档：工具描述的质量决定检索质量，写工具描述时想象自己在写搜索引擎的语料。
分层管理：Catalog 做统一注册，Group 做范围缩小，语义检索做精细排序。
动态注入：每轮推理前根据当前 Observation 重新检索，而非任务开始时检索一次。
量化评估：用 Recall@k 和 Perfect Recall Rate 持续监控检索质量，驱动迭代优化。

下一篇将进入生产 Agent 的可观测性实践：当 Agent 在线上出问题时，如何用追踪工具定位根因。