第 23 章
Citations API：文档问答系统的引用溯源机制与结构化输出的不兼容处理
第二十三章：Tool Search Tool：动态工具发现与自适应能力扩展
23.1 工具发现的必要性
在传统的 Tool Use 模式中，工具列表在每次 API 请求时静态传入。这在工具数量少（5-10个）时是合理的，但当系统需要支持数百个工具时，会遇到严重问题：
问题一：上下文窗口限制 大量工具定义会消耗大量输入 token。50个详细工具定义可能消耗 10,000+ tokens，占据宝贵的上下文空间。
问题二：模型注意力分散 研究表明，当工具列表过长时，Claude 的工具选择准确性会下降。模型需要在大量工具中寻找正确的一个，容易产生混淆。
问题三：工具生态动态变化 企业环境中，工具会不断添加、更新和淘汰。静态工具列表无法适应这种动态变化。
解决方案：Tool Search Tool（工具搜索工具）
Tool Search Tool 是一种元工具（meta-tool）模式，它允许 Claude 在需要时动态搜索和发现可用工具，而不是在请求开始时接收所有工具。
23.2 Tool Search Tool 的设计原则
Tool Search Tool 的核心思想是：将工具发现本身作为一个工具。
tool_search_tool = {
"name": "search_tools",
"description": """搜索可用工具列表。当你需要某种能力但不确定是否有对应工具时，
或者需要浏览所有可用工具时，使用此工具。
返回匹配的工具定义列表，每个工具包含名称、描述和参数 Schema。
找到合适的工具后，可以直接使用该工具的名称和参数调用它。
【注意】系统支持动态工具调用：搜索到工具后，即使它不在初始工具列表中，
你也可以通过 execute_dynamic_tool 工具来调用它。""",
"input_schema": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词，描述你需要的能力，如'发送邮件'、'查询数据库'、'处理图片'"
},
"category": {
"type": "string",
"description": "工具类别过滤（可选）",
"enum": [
"communication", # 通信工具：邮件、短信、通知
"data", # 数据工具：数据库、文件、表格
"integration", # 集成工具：第三方 API
"computation", # 计算工具：数学、代码执行
"media", # 媒体工具：图片、视频、音频
"utility" # 通用工具
]
},
"max_results": {
"type": "integer",
"default": 5,
"minimum": 1,
"maximum": 20,
"description": "返回工具数量上限"
}
},
"required": ["query"]
}
}
execute_dynamic_tool = {
"name": "execute_dynamic_tool",
"description": """执行通过 search_tools 发现的工具。
在调用此工具前，必须先用 search_tools 确认工具存在并了解其参数格式。""",
"input_schema": {
"type": "object",
"properties": {
"tool_name": {
"type": "string",
"description": "要执行的工具名称（必须是 search_tools 返回的工具）"
},
"tool_input": {
"type": "object",
"description": "传给工具的参数（必须符合该工具的 input_schema）"
}
},
"required": ["tool_name", "tool_input"]
}
}
23.3 工具注册表的实现
import json
import re
from typing import Dict, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
@dataclass
class ToolRegistration:
"""工具注册信息"""
definition: dict
func: callable
category: str
tags: List[str] = field(default_factory=list)
usage_count: int = 0
last_used: Optional[datetime] = None
version: str = "1.0"
class DynamicToolRegistry:
"""动态工具注册表"""
def __init__(self):
self.tools: Dict[str, ToolRegistration] = {}
self._search_index: Dict[str, List[str]] = {} # 关键词 -> 工具名列表
def register(
self,
tool_definition: dict,
func: callable,
category: str = "utility",
tags: List[str] = None
) -> "DynamicToolRegistry":
"""注册工具"""
name = tool_definition["name"]
registration = ToolRegistration(
definition=tool_definition,
func=func,
category=category,
tags=tags or []
)
self.tools[name] = registration
# 更新搜索索引
self._index_tool(name, tool_definition, tags or [])
return self
def _index_tool(self, name: str, definition: dict, tags: List[str]):
"""为工具建立搜索索引"""
# 从名称和描述中提取关键词
text = f"{name} {definition.get('description', '')} {' '.join(tags)}"
# 简单的分词（实际生产中可用 jieba 等）
keywords = set(re.findall(r'[\w]+', text.lower()))
for keyword in keywords:
if keyword not in self._search_index:
self._search_index[keyword] = []
if name not in self._search_index[keyword]:
self._search_index[keyword].append(name)
def search(
self,
query: str,
category: Optional[str] = None,
max_results: int = 5
) -> List[dict]:
"""搜索工具"""
# 分词查询
query_keywords = set(re.findall(r'[\w]+', query.lower()))
# 计算每个工具的相关性分数
tool_scores: Dict[str, float] = {}
for keyword in query_keywords:
# 完全匹配
if keyword in self._search_index:
for tool_name in self._search_index[keyword]:
tool_scores[tool_name] = tool_scores.get(tool_name, 0) + 1.0
# 前缀匹配
for indexed_keyword, tool_names in self._search_index.items():
if indexed_keyword.startswith(keyword) and indexed_keyword != keyword:
for tool_name in tool_names:
tool_scores[tool_name] = tool_scores.get(tool_name, 0) + 0.5
# 按类别过滤
if category:
tool_scores = {
name: score
for name, score in tool_scores.items()
if self.tools.get(name) and self.tools[name].category == category
}
# 排序并返回前 N 个
sorted_tools = sorted(tool_scores.items(), key=lambda x: x[1], reverse=True)
results = []
for tool_name, score in sorted_tools[:max_results]:
if tool_name in self.tools:
reg = self.tools[tool_name]
results.append({
"name": tool_name,
"description": reg.definition.get("description", ""),
"category": reg.category,
"tags": reg.tags,
"input_schema": reg.definition.get("input_schema", {}),
"relevance_score": round(score, 2)
})
return results
def execute(self, tool_name: str, tool_input: dict):
"""执行工具"""
if tool_name not in self.tools:
raise ValueError(f"工具不存在: {tool_name}")
registration = self.tools[tool_name]
registration.usage_count += 1
registration.last_used = datetime.now()
return registration.func(**tool_input)
def list_categories(self) -> Dict[str, int]:
"""列出所有类别及工具数量"""
categories = {}
for reg in self.tools.values():
categories[reg.category] = categories.get(reg.category, 0) + 1
return categories
def get_popular_tools(self, top_n: int = 10) -> List[dict]:
"""获取最常用的工具"""
sorted_tools = sorted(
self.tools.items(),
key=lambda x: x[1].usage_count,
reverse=True
)
return [
{
"name": name,
"usage_count": reg.usage_count,
"category": reg.category
}
for name, reg in sorted_tools[:top_n]
]
23.4 完整的动态工具发现 Agent
import anthropic
class DynamicToolAgent:
"""支持动态工具发现的 Agent"""
def __init__(self, registry: DynamicToolRegistry, model: str = "claude-opus-4-5"):
self.client = anthropic.Anthropic()
self.registry = registry
self.model = model
# 初始工具集：只包含元工具
self.meta_tools = [tool_search_tool, execute_dynamic_tool]
def _handle_search_tools(self, inputs: dict) -> str:
"""处理 search_tools 调用"""
results = self.registry.search(
query=inputs["query"],
category=inputs.get("category"),
max_results=inputs.get("max_results", 5)
)
if not results:
return json.dumps({
"found": 0,
"message": f"未找到与 '{inputs['query']}' 相关的工具",
"suggestion": "尝试更换关键词，或查询 list_all_categories 了解可用工具类别"
}, ensure_ascii=False)
return json.dumps({
"found": len(results),
"tools": results
}, ensure_ascii=False)
def _handle_execute_dynamic_tool(self, inputs: dict) -> str:
"""处理 execute_dynamic_tool 调用"""
tool_name = inputs["tool_name"]
tool_input = inputs["tool_input"]
try:
result = self.registry.execute(tool_name, tool_input)
return json.dumps(result, ensure_ascii=False, default=str)
except ValueError as e:
return json.dumps({
"error": str(e),
"suggestion": "请先用 search_tools 确认工具名称是否正确"
}, ensure_ascii=False)
except Exception as e:
return json.dumps({
"error": f"执行失败: {str(e)}",
"tool_name": tool_name,
"tool_input": tool_input
}, ensure_ascii=False)
def _process_tool_call(self, tool_name: str, tool_input: dict) -> str:
"""处理工具调用"""
if tool_name == "search_tools":
return self._handle_search_tools(tool_input)
elif tool_name == "execute_dynamic_tool":
return self._handle_execute_dynamic_tool(tool_input)
else:
return json.dumps({"error": f"未知元工具: {tool_name}"})
def run(self, user_message: str, system: str = "") -> str:
"""运行动态工具发现 Agent"""
default_system = """你是一个能够动态发现和使用工具的助手。
【工具使用流程】
1. 接到任务后，先思考需要哪些类型的能力
2. 使用 search_tools 搜索相关工具
3. 查看搜索结果，了解工具的参数格式
4. 使用 execute_dynamic_tool 执行找到的工具
5. 分析工具结果，如需更多工具则继续搜索
【注意】
- 不要假设工具存在，必须先通过 search_tools 确认
- 如果第一次搜索没找到，尝试不同的关键词
- 工具执行失败时，检查参数是否符合工具的 input_schema"""
messages = [{"role": "user", "content": user_message}]
while True:
response = self.client.messages.create(
model=self.model,
max_tokens=4096,
system=system or default_system,
tools=self.meta_tools,
messages=messages
)
if response.stop_reason == "end_turn":
return ' '.join(b.text for b in response.content if b.type == "text")
if response.stop_reason == "tool_use":
tool_results = []
for block in response.content:
if block.type != "tool_use":
continue
print(f"[元工具] {block.name}: {json.dumps(block.input, ensure_ascii=False)[:150]}")
result = self._process_tool_call(block.name, block.input)
tool_results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": result
})
messages.append({"role": "assistant", "content": response.content})
messages.append({"role": "user", "content": tool_results})
else:
break
return "任务完成"
23.5 构建工具生态系统
注册大量工具的实践
def build_enterprise_registry() -> DynamicToolRegistry:
"""构建企业级工具注册表（示例）"""
registry = DynamicToolRegistry()
# === 通信工具 ===
registry.register(
tool_definition={
"name": "send_email",
"description": "发送电子邮件给指定收件人",
"input_schema": {
"type": "object",
"properties": {
"to": {"type": "string"},
"subject": {"type": "string"},
"body": {"type": "string"}
},
"required": ["to", "subject", "body"]
}
},
func=lambda to, subject, body: {"status": "sent", "message_id": "msg_001"},
category="communication",
tags=["email", "通知", "邮件"]
)
registry.register(
tool_definition={
"name": "send_sms",
"description": "发送短信到指定手机号",
"input_schema": {
"type": "object",
"properties": {
"phone": {"type": "string"},
"message": {"type": "string", "maxLength": 160}
},
"required": ["phone", "message"]
}
},
func=lambda phone, message: {"status": "sent", "cost": 0.05},
category="communication",
tags=["sms", "短信", "手机"]
)
# === 数据工具 ===
registry.register(
tool_definition={
"name": "query_mysql",
"description": "执行 MySQL 数据库查询",
"input_schema": {
"type": "object",
"properties": {
"sql": {"type": "string"},
"database": {"type": "string"}
},
"required": ["sql", "database"]
}
},
func=lambda sql, database: {"rows": [], "affected": 0},
category="data",
tags=["database", "mysql", "数据库", "查询", "sql"]
)
registry.register(
tool_definition={
"name": "read_excel",
"description": "读取 Excel 文件并返回数据",
"input_schema": {
"type": "object",
"properties": {
"file_path": {"type": "string"},
"sheet_name": {"type": "string"}
},
"required": ["file_path"]
}
},
func=lambda file_path, sheet_name=None: {"rows": [], "columns": []},
category="data",
tags=["excel", "表格", "文件", "读取"]
)
# === 集成工具 ===
registry.register(
tool_definition={
"name": "create_jira_ticket",
"description": "在 Jira 中创建工单",
"input_schema": {
"type": "object",
"properties": {
"project": {"type": "string"},
"summary": {"type": "string"},
"description": {"type": "string"},
"issue_type": {"type": "string", "enum": ["Bug", "Story", "Task"]}
},
"required": ["project", "summary", "issue_type"]
}
},
func=lambda **kwargs: {"ticket_id": "PROJ-001", "url": "https://jira.example.com/PROJ-001"},
category="integration",
tags=["jira", "工单", "bug", "任务", "项目管理"]
)
registry.register(
tool_definition={
"name": "get_salesforce_contact",
"description": "从 Salesforce CRM 获取联系人信息",
"input_schema": {
"type": "object",
"properties": {
"email": {"type": "string"},
"name": {"type": "string"}
}
}
},
func=lambda **kwargs: {"contact_id": "003xx001", "name": "张三", "company": "示例公司"},
category="integration",
tags=["salesforce", "crm", "客户", "联系人"]
)
# 添加更多工具...（实际系统可能有数百个）
print(f"工具注册表：{len(registry.tools)} 个工具")
print(f"类别分布：{registry.list_categories()}")
return registry
23.6 自适应工具组合：处理未知任务
Tool Search Tool 的真正威力在于处理系统设计时未预见的任务：
def demonstrate_adaptive_capability():
"""演示自适应工具发现能力"""
registry = build_enterprise_registry()
agent = DynamicToolAgent(registry)
# 任务1：系统设计时预见的任务
result1 = agent.run("给 [email protected] 发一封邮件，通知他明天的会议取消了")
# 任务2：需要组合多个工具的复杂任务
result2 = agent.run("""
从 Excel 文件 /data/customers.xlsx 读取客户列表，
在 Salesforce 中查找每个客户的联系人信息，
然后为每个客户创建一个 Jira 跟进工单，
并发送邮件通知销售团队。
""")
# 任务3：完全未预见的组合需求
result3 = agent.run("""
我需要给最近购买了超过 $10000 的客户发短信优惠券。
先查询数据库获取客户列表，然后逐一发送短信。
""")
return result1, result2, result3
23.7 工具版本化与能力演进
随着业务发展，工具能力需要演进。Tool Search 模式天然支持工具的无缝更新：
class VersionedToolRegistry(DynamicToolRegistry):
"""支持工具版本管理的注册表"""
def register_versioned(
self,
tool_definition: dict,
func: callable,
version: str = "1.0",
deprecated_versions: List[str] = None,
**kwargs
):
"""注册带版本的工具"""
name = tool_definition["name"]
# 检查是否有旧版本
if name in self.tools:
old_version = self.tools[name].version
print(f"更新工具 {name}: v{old_version} -> v{version}")
# 添加版本信息到描述
description = tool_definition.get("description", "")
if deprecated_versions:
description += f"\n（当前版本：{version}，废弃版本：{', '.join(deprecated_versions)}）"
else:
description += f"\n（版本：{version}）"
enhanced_definition = {**tool_definition, "description": description}
reg = self.register(enhanced_definition, func, **kwargs)
self.tools[name].version = version
return reg
小结
Tool Search Tool 模式将工具使用从"静态配置"提升到"动态发现"，适合以下场景：
- 工具数量大（50个以上）：避免每次请求传递全量工具定义
- 工具生态动态变化：新工具上线后无需更改 Agent 代码
- 处理未预见任务：让 Claude 自主发现完成任务所需的能力
- 多团队工具共享：不同团队的工具统一注册，按需发现和使用
核心实现要素：
- 工具注册表（支持语义搜索）
- 元工具（
search_tools+execute_dynamic_tool） - 清晰的使用流程说明（在系统提示中指导 Claude 如何使用这种模式）
下一章将探讨 Advisor Tool——让 Claude 在执行任务前进行元认知规划的高级模式。