工具搜索与命名空间

📑 目录

概述

向量检索原理与工具 Embedding 的生成策略

当 Agent 拥有大量工具(超过 10-15 个)时,将所有工具的 Schema 一次性注入 system prompt 会导致两个严重问题:一是 token 消耗激增,二是模型在过多选择中容易"迷茫",选错工具的概率显著上升。

ToolSearchTool 的解决方案是延迟加载 + 语义检索:工具在注册时可以被标记为 defer_loading=True,这意味着它们的 Schema 不会默认进入 prompt。当模型需要工具时,它会先调用 ToolSearchTool,传入一个描述目标工具功能的查询字符串;ToolSearchTool 内部通过向量相似度检索,返回最相关的工具 Schema,然后模型再决定调用哪个具体工具。

从实现机制来看,工具检索依赖以下技术栈:

  1. Embedding 生成:每个工具的 name + description 会被送入文本 embedding 模型(如 text-embedding-3-small),生成一个高维向量。
  2. 向量索引:所有工具的向量被存入一个高效的近似最近邻(ANN)索引结构(如 HNSW)。
  3. 语义查询:当模型发出搜索请求时,查询文本同样被 embedding,然后在 ANN 索引中查找 top-k 个最相似的候选工具。

命名空间(tool_namespace)则提供了另一维度的隔离机制。通过将工具分组到不同命名空间,你可以实现更细粒度的检索范围控制。例如,将"财务相关工具"放入 finance 命名空间,将"运维相关工具"放入 ops 命名空间,检索时就可以按领域过滤,提升准确率。

需要注意的是,工具检索引入了额外的 embedding 调用延迟(通常为 50-200ms)。对于延迟敏感的场景,建议将工具向量预计算并缓存到内存中,避免每次检索都调用 embedding API。

ToolSearchTool 和 tool_namespace:如何在运行时按需加载大量工具。

正文

相关阅读

参考文档

完整实战示例:大规模工具库的分层检索架构

以下示例展示了如何为拥有 50+ 工具的系统构建分层检索架构,结合命名空间隔离和语义搜索:

import asyncio
from typing import Annotated
from agents import Agent, Runner, ToolSearchTool, function_tool


# ====== 财务命名空间工具 ======
@function_tool(defer_loading=True, tool_namespace="finance")
def query_invoice(invoice_id: Annotated[str, "发票编号"]) -> str:
    """查询发票详情和付款状态。"""
    return f"Invoice {invoice_id}: $500, status=paid"


@function_tool(defer_loading=True, tool_namespace="finance")
def generate_report(month: Annotated[str, "月份,格式 YYYY-MM"]) -> str:
    """生成月度财务报表。"""
    return f"Report for {month}: revenue=$10k, expenses=$7k"


# ====== 运维命名空间工具 ======
@function_tool(defer_loading=True, tool_namespace="ops")
def restart_service(service_name: Annotated[str, "服务名称"]) -> str:
    """重启指定微服务实例。需要管理员权限。"""
    return f"Service {service_name} restarted successfully"


@function_tool(defer_loading=True, tool_namespace="ops")
def check_logs(service_name: Annotated[str, "服务名称"], lines: int = 50) -> str:
    """查看服务最近日志。"""
    return f"Last {lines} lines of logs for {service_name}: [OK] Healthy"


# ====== 客服命名空间工具 ======
@function_tool(defer_loading=True, tool_namespace="support")
def create_ticket(description: Annotated[str, "问题描述"], priority: str = "medium") -> str:
    """创建客户支持工单。"""
    return f"Ticket created: {description[:30]}... (priority={priority})"


@function_tool(defer_loading=True, tool_namespace="support")
def escalate_ticket(ticket_id: Annotated[str, "工单编号"], reason: str) -> str:
    """将工单升级给高级支持团队。"""
    return f"Ticket {ticket_id} escalated: {reason}"


async def main():
    # 构建分层检索 Agent
    agent = Agent(
        name="EnterpriseAssistant",
        instructions="""
你是企业智能助手  #拥有财务、运维和客服三大领域的工具。
当用户提出请求时:
1. 首先判断请求属于哪个领域(finance/ops/support)
2. 使用 ToolSearchTool 在该领域内搜索最合适的工具
3. 如果跨领域  #可以多次搜索
4. 操作类工具(如 restart_service)执行前需向用户确认
""".strip(),
        tools=[
            query_invoice, generate_report,
            restart_service, check_logs,
            create_ticket, escalate_ticket,
            ToolSearchTool(),
        ],
        model="gpt-5-nano")

常见问题与调试

问题一:工具搜索返回不相关结果

当工具描述过于模糊或相似时,embedding 检索可能无法区分它们。优化策略:

  1. 为每个工具编写独特的、详细的描述,避免使用通用的"查询数据"、"获取信息"等模糊词汇。
  2. 在描述中明确包含工具的输入输出格式和适用场景。
  3. 如果两个工具功能高度重叠,考虑合并它们,通过参数区分行为。

问题二:命名空间配置错误导致工具不可见

如果工具注册了命名空间,但 Agent 的 instructions 中没有引导模型按命名空间搜索,模型可能无法发现这些工具。检查清单:

  1. 确认 tool_namespace 参数在 @function_tool 中正确设置。
  2. 在 instructions 中明确告知模型命名空间的存在和划分逻辑。
  3. 使用 ToolSearchTool 时,确认搜索查询中包含命名空间关键词。

问题三:延迟加载工具的首次调用延迟高

由于 ToolSearchTool 需要额外的 embedding 和检索步骤,首次调用延迟工具的响应时间可能比直接加载的工具长 200-500ms。优化建议:

  1. 预加载高频工具(将最常用的 3-5 个工具设为 defer_loading=False)。
  2. 在应用启动时预热 embedding 缓存。
  3. 对延迟敏感的场景,考虑使用更轻量的关键词匹配作为第一层过滤,embedding 检索作为第二层精排。

与其他方案对比

维度Agents SDK ToolSearch + NamespaceLangChain Tool Retrieval手动工具分组
检索方式语义向量检索向量检索或关键词匹配硬编码规则
命名空间原生支持需通过 metadata 模拟需自行实现
维护成本中(需写好描述)中(依赖向量存储)高(规则易腐化)
扩展性高(新工具自动索引)高(新工具自动索引)低(需改代码)

LangChain 的工具检索系统更为成熟,支持多种向量存储后端(FAISS、Pinecone、Chroma)。Agents SDK 的 ToolSearchTool 更适合与 OpenAI 生态紧密集成的场景,且命名空间机制在大型企业中具有天然的组织结构映射优势。手动工具分组虽然实现简单,但随着工具数量增长,规则维护会成为沉重负担,不推荐在超过 10 个工具的场景中使用。

语义检索引擎的定制与性能调优

mermaid
flowchart LR
Q[用户查询文本] --> E[Embedding 模型]
E --> V1[查询向量]
T[工具描述文本] --> E2[预计算 Embedding]
E2 --> V2[工具向量索引 HNSW]
V1 --> S[相似度计算 top-k]
V2 --> S
S --> F[命名空间过滤]
F --> R[返回候选工具 Schema]


`ToolSearchTool` 的默认实现已经足够应对多数场景,但在超大规模工具库(超过 100 个工具)或延迟敏感的业务中,开发者需要对检索引擎进行深度定制。优化的方向主要有三个:减少 Embedding API 调用、提升检索精度、加速索引查询。这三个方向相辅相成,共同决定了工具检索系统的生产可用性。

首先来看 Embedding 缓存。默认情况下,每次调用 `ToolSearchTool` 都会将查询文本发送到 embedding API,产生 50 到 200 毫秒的网络延迟。如果查询文本存在大量重复模式(如查询订单状态、查看用户信息、生成报表),本地缓存 embedding 向量可以显著降低延迟。以下实现展示了一个基于内存的查询向量缓存,键为查询文本的哈希值:

```python
import hashlib
from typing import List

class EmbeddingCache:
    """基于 LRU 的查询向量缓存。"""

    def __init__(self, maxsize: int = 1000):
        self._maxsize = maxsize
        self._cache = {}
        self._order = []

    def _key(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()[:16]

    def get(self, text: str):
        k = self._key(text)
        if k in self._cache:
            self._order.remove(k)
            self._order.append(k)
            return self._cache[k]
        return None

    def put(self, text: str, vector: List[float]):
        k = self._key(text)
        if k in self._cache:
            self._order.remove(k)
        self._order.append(k)
        self._cache[k] = vector
        while len(self._order) > self._maxsize:
            old = self._order.pop(0)
            del self._cache[old]

其次是混合检索策略。纯语义检索在面对特定关键词查询时可能不够精准,例如用户输入 restart_service 时,embedding 可能返回 check_service_status 而不是精确匹配。将关键词匹配作为前置过滤器,再以语义相似度精排,可以在不牺牲召回率的前提下显著提升准确率:

import re
from typing import List, Callable

def hybrid_tool_search(
    query: str,
    tools: List[Callable],
    semantic_search_fn,
    top_k: int = 5,
) -> List[Callable]:
    keywords = re.findall(r"\w+", query.lower())
    keyword_hits = []
    for t in tools:
        desc = (t.__doc__ or "").lower()
        score = sum(1 for kw in keywords if kw in desc)
        if score > 0:
            keyword_hits.append((t, score))

    if len(keyword_hits) >= top_k:
        keyword_hits.sort(key=lambda x: x[1], reverse=True)
        return [t for t, _ in keyword_hits[:top_k]]

    semantic_results = semantic_search_fn(query, top_k=top_k)
    seen = {t for t, _ in keyword_hits}
    for t in semantic_results:
        if t not in seen:
            keyword_hits.append((t, 0))
    return [t for t, _ in keyword_hits[:top_k]]

最后是索引预热与效果评估。在应用启动阶段预先将全部工具的 embedding 加载到内存,并使用 faiss 或 hnswlib 构建 ANN 索引,可以将单次检索延迟从 200 毫秒降至 5 毫秒以内。对于命名空间较多的系统,建议为每个命名空间维护独立的索引分区,检索时先路由到对应分区,再执行向量搜索。这种分层索引策略在大规模场景下能有效降低搜索空间,提升整体吞吐量。同时,应建立检索效果的持续评估机制,定期采样真实查询并人工标注最相关的工具,计算命中率和平均倒数排名,根据评估结果迭代优化工具描述文本和 embedding 模型选型。对于命名空间的热更新需求,可以通过监听配置中心的变化事件,动态重建对应分区的索引而无需重启整个服务,从而实现工具库的无缝扩展和更新。在实际落地时,还应充分考虑 embedding 模型的多语言支持能力,确保中英文混合查询场景下的检索质量不会出现显著下降,必要时可为不同语言维护独立的向量空间。

此外,当工具库发生高频更新时,索引重建的代价不可忽视。增量更新策略可以有效缓解这一问题:仅对新增、修改和删除的工具进行局部向量更新,而非全量重建索引。结合写时复制技术,可以在不阻塞查询流量的前提下完成索引更新,确保检索服务的连续性。

生产环境部署与性能优化

向量索引运维的实践要点

将本章节的技术应用到生产环境时,首要考虑的是稳定性与可观测性。建议采用渐进式 rollout 策略:先在开发环境验证核心逻辑,再迁移到预发布环境进行压力测试,最后才全量上线。部署过程中应配置完善的日志收集和指标监控,确保任何问题都能被快速发现和定位。

具体来说,需要在基础设施层面做好以下准备:容器资源限制(CPU/内存)、网络策略配置(防火墙规则、服务网格)、持久化存储选型(SSD vs 标准盘)以及备份恢复方案。对于高可用要求严格的场景,建议部署多实例并配置负载均衡,避免单点故障导致服务中断。

检索准确率的关键指标

监控是生产系统的生命线。针对本章节涉及的功能,建议重点跟踪以下指标:请求延迟(P50/P95/P99)、错误率(4xx/5xx/超时)、吞吐量(QPS/TPS)以及资源利用率(CPU/内存/磁盘/网络)。这些指标应接入统一的监控大盘,并设置合理的告警阈值。

除了基础指标,还应关注业务层面的指标。例如功能成功率、用户满意度、成本消耗趋势等。通过将技术指标与业务指标关联分析,可以更准确地评估系统改进的实际价值,避免陷入"为了优化而优化"的陷阱。

大规模工具库的架构考量

随着业务规模增长,单实例部署很快会成为瓶颈。扩展性设计应在项目初期就纳入考量,而非事后补救。水平扩展通常比垂直扩展更具成本效益,但也引入了分布式系统的复杂性(数据一致性、服务发现、负载均衡等)。

在扩展过程中,建议遵循"无状态优先"原则:将状态外置到独立的存储层(如 Redis、PostgreSQL),使计算层可以随时水平扩容。对于无法避免的状态(如会话、缓存),采用分布式一致性协议或最终一致性模型来管理。定期进行容量规划和压力测试,确保系统在流量峰值时仍能稳定运行。

运维团队的协作建议

技术方案的落地离不开高效的团队协作。建议建立清晰的运维手册(Runbook),涵盖常见故障的诊断步骤、应急处理流程和升级路径。同时,通过定期的复盘会议,将线上事故转化为团队的学习素材,持续完善系统的健壮性。

在工具链方面,推荐将本章节的配置和脚本纳入版本控制(Git),并使用 Infrastructure as Code(IaC)工具(如 Terraform、Ansible)管理基础设施变更。这不仅能提高部署效率,还能确保环境一致性,减少"在我机器上能跑"的问题。