作为一名深耕 AI Agent 开发的工程师，我在过去三年中构建了超过20个生产级 Agent 系统。从最初的简单对话机器人到如今的复杂多工具协作系统，Tool Use（工具调用）一直是核心能力。今天我将分享在工具注册、发现与调用链设计上的实战经验。
先看一组直接影响你钱包的数字：GPT-4.1 output $8/MTok、Claude Sonnet 4.5 output $15/MTok、Gemini 2.5 Flash output $2.50/MTok、DeepSeek V3.2 output $0.42/MTok。以每月100万 token 输出量计算：
- GPT-4.1：$8/月（官方渠道 ¥58.4）
- Claude Sonnet 4.5：$15/月（官方渠道 ¥109.5）
- DeepSeek V3.2：$0.42/月（官方渠道 ¥3.07）
如果你使用 HolySheep API 中转，按 ¥1=$1 无损汇率结算，相比官方 ¥7.3=$1 汇率，每月可节省超过85%成本。DeepSeek V3.2 百万 token 仅需 ¥0.42，GPT-4.1 也只需 ¥8。这对于日均调用量大的 Agent 应用来说是巨大优势。
一、Tool Use 核心概念与工作原理
Tool Use 本质上是让 LLM 能够调用外部函数来扩展能力边界。当模型需要执行搜索、计算、数据库查询等操作时，通过 function calling 机制将请求路由到预定义的工具函数。这解决了纯语言模型的三大局限：无法访问实时数据、无法执行副作用操作、无法处理结构化计算。
1.1 Tool Call 的完整生命周期
一次完整的工具调用经历以下阶段：用户输入 → 模型推理决定调用哪个工具 → 构造调用请求 → 执行工具函数 → 将结果返回模型 → 模型整合结果生成最终回答。这个链路中任何环节出问题都会导致工具调用失败。
二、工具注册机制设计与实现
工具注册是 Tool Use 系统的根基。一个健壮的注册机制需要支持动态注册、版本管理、依赖注入等功能。我见过太多系统因为注册机制设计缺陷导致生产环境频繁出问题。
2.1 基于装饰器的声明式注册
最推荐的注册方式是基于装饰器的声明式注册。这种方式代码侵入性低，开发者只需在函数上加装饰器即可完成注册。
from typing import Callable, Any
from dataclasses import dataclass
import json
@dataclass
class ToolMetadata:
name: str
description: str
parameters: dict
version: str = "1.0.0"
class ToolRegistry:
"""统一工具注册中心，支持动态注册与发现"""
def __init__(self):
self._tools: dict[str, Callable] = {}
self._metadata: dict[str, ToolMetadata] = {}
def register(self, metadata: ToolMetadata):
"""装饰器注册工具"""
def decorator(func: Callable) -> Callable:
self._tools[metadata.name] = func
self._metadata[metadata.name] = metadata
return func
return decorator
def get_tool(self, name: str) -> Callable | None:
return self._tools.get(name)
def get_metadata(self, name: str) -> ToolMetadata | None:
return self._metadata.get(name)
def list_tools(self) -> list[dict]:
return [
{
"name": meta.name,
"description": meta.description,
"parameters": meta.parameters
}
for meta in self._metadata.values()
]
全局注册中心实例
registry = ToolRegistry()
@registry.register(ToolMetadata(
name="weather_query",
description="查询指定城市的实时天气信息，返回温度、湿度、风力等数据",
parameters={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名称，如北京、上海"},
"country": {"type": "string", "description": "国家代码，如 CN、US"}
},
"required": ["city"]
}
))
def weather_query(city: str, country: str = "CN") -> dict:
"""模拟天气查询工具"""
return {
"city": city,
"temperature": 22,
"humidity": 65,
"wind": "东南风 3级",
"condition": "多云"
}
@registry.register(ToolMetadata(
name="calculator",
description="执行数学计算，支持加减乘除、幂运算、平方根",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式，如 2+3*4"}
},
"required": ["expression"]
}
))
def calculator(expression: str) -> dict:
"""安全计算器"""
allowed_chars = set("0123456789+-*/().^ ")
if not all(c in allowed_chars for c in expression):
return {"error": "表达式包含非法字符"}
try:
# 安全评估：仅处理数学运算
result = eval(expression.replace("^", "**"))
return {"result": result, "expression": expression}
except Exception as e:
return {"error": str(e)}2.2 工具版本管理与灰度发布
生产环境中，工具升级不可避免但风险极高。我推荐使用版本化注册配合灰度策略：
class VersionedToolRegistry(ToolRegistry):
"""支持版本管理的工具注册中心"""
def __init__(self):
super().__init__()
self._versions: dict[str, list[str]] = {} # tool_name -> [versions]
self._default_version: dict[str, str] = {}
def register_versioned(
self,
metadata: ToolMetadata,
is_default: bool = False
):
"""注册指定版本的工具"""
def decorator(func: Callable) -> Callable:
tool_key = f"{metadata.name}:{metadata.version}"
self._tools[tool_key] = func
self._metadata[tool_key] = metadata
# 维护版本列表
if metadata.name not in self._versions:
self._versions[metadata.name] = []
if metadata.version not in self._versions[metadata.name]:
self._versions[metadata.name].append(metadata.version)
# 设置默认版本
if is_default or metadata.name not in self._default_version:
self._default_version[metadata.name] = metadata.version
return func
return decorator
def get_versions(self, tool_name: str) -> list[str]:
return self._versions.get(tool_name, [])
def get_default_version(self, tool_name: str) -> str | None:
return self._default_version.get(tool_name)
def get_tool_by_version(self, name: str, version: str = None) -> Callable | None:
if version is None:
version = self.get_default_version(name)
return self._tools.get(f"{name}:{version}")三、工具发现机制与智能路由
当 Agent 需要完成一个复杂任务时，如何让模型快速找到最合适的工具？这就是工具发现问题。我实现了一套基于语义相似度 + 关键词匹配的混合发现机制。
import json
from typing import Optional
class ToolDiscovery:
"""基于语义匹配的工具发现引擎"""
def __init__(self, registry: ToolRegistry):
self.registry = registry
self._cache: dict[str, list[dict]] = {}
def discover(
self,
query: str,
max_results: int = 5,
threshold: float = 0.3
) -> list[dict]:
"""
根据用户查询发现相关工具
使用关键词匹配 + 语义相似度的混合策略
"""
# 从注册中心获取所有工具
all_tools = self.registry.list_tools()
scored_tools = []
query_lower = query.lower()
query_keywords = set(query_lower.split())
for tool in all_tools:
score = 0.0
reasons = []
# 关键词精确匹配（权重0.6）
tool_name = tool["name"].lower()
if query_lower in tool_name or tool_name in query_lower:
score += 0.6
reasons.append("名称匹配")
elif any(kw in tool_name for kw in query_keywords if len(kw) > 2):
score += 0.3
reasons.append("关键词命中")
# 描述相关性（权重0.4）
desc = tool["description"].lower()
desc_keywords = desc.split()
overlap = len(query_keywords & set(desc_keywords))
if overlap > 0:
score += min(0.4, overlap * 0.1)
reasons.append(f"描述命中{overlap}个词")
if score >= threshold:
scored_tools.append({
"tool": tool,
"score": score,
"reasons": reasons
})
# 按分数排序并返回 top-k
scored_tools.sort(key=lambda x: x["score"], reverse=True)
return scored_tools[:max_results]
def build_tools_for_llm(
self,
query: str,
context_tools: list[str] = None
) -> list[dict]:
"""
为 LLM 构建合适的 tools 参数
如果指定了 context_tools，则优先使用这些工具
"""
if context_tools:
return [
self.registry.get_metadata(name)
for name in context_tools
if self.registry.get_metadata(name)
]
discovered = self.discover(query)
return [item["tool"] for item in discovered]
使用示例
discovery = ToolDiscovery(registry)
语义搜索：找到与"计算"相关的工具
calc_tools = discovery.discover("数学运算", max_results=3)
print(f"发现 {len(calc_tools)} 个相关工具")
for item in calc_tools:
print(f" - {item['tool']['name']} (score: {item['score']:.2f})")
print(f" 原因: {', '.join(item['reasons'])}")四、调用链设计：构建可靠的工具执行流程
工具调用链是整个 Tool Use 系统的核心。我设计了一套包含重试、降级、超时控制的可靠调用链。
4.1 调用链核心组件
import asyncio
import time
from typing import Any, Optional
from dataclasses import dataclass
from enum import Enum
class ExecutionStatus(Enum):
SUCCESS = "success"
FAILED = "failed"
TIMEOUT = "timeout"
RATE_LIMITED = "rate_limited"
FALLBACK_USED = "fallback_used"
@dataclass
class ExecutionResult:
status: ExecutionStatus
result: Any
execution_time_ms: float
tool_name: str
attempt: int
error: Optional[str] = None
class ToolExecutor:
"""工具执行器，支持重试、降级、超时控制"""
def __init__(
self,
max_retries: int = 3,
base_delay: float = 0.5,
timeout: float = 30.0,
use_fallback: bool = True
):
self.max_retries = max_retries
self.base_delay = base_delay
self.timeout = timeout
self.use_fallback = use_fallback
self._fallbacks: dict[str, Callable] = {}
def register_fallback(
self,
tool_name: str,
fallback_func: Callable
):
"""为指定工具注册降级函数"""
self._fallbacks[tool_name] = fallback_func
async def execute(
self,
tool_func: Callable,
tool_name: str,
arguments: dict,
use_cache: bool = True
) -> ExecutionResult:
"""
异步执行工具调用，包含完整错误处理
"""
cache_key = f"{tool_name}:{json.dumps(arguments, sort_keys=True)}"
# 简单的内存缓存
if use_cache and hasattr(self, '_cache'):
cached = self._cache.get(cache_key)
if cached and time.time() - cached['timestamp'] < 300:
return cached['result']
last_error = None
for attempt in range(1, self.max_retries + 1):
start_time = time.time()
try:
# 根据函数类型决定同步/异步执行
if asyncio.iscoroutinefunction(tool_func):
result = await asyncio.wait_for(
tool_func(**arguments),
timeout=self.timeout
)
else:
result = tool_func(**arguments)
execution_time = (time.time() - start_time) * 1000
return ExecutionResult(
status=ExecutionStatus.SUCCESS,
result=result,
execution_time_ms=execution_time,
tool_name=tool_name,
attempt=attempt
)
except asyncio.TimeoutError:
last_error = f"执行超时（{self.timeout}s）"
status = ExecutionStatus.TIMEOUT
except Exception as e:
last_error = str(e)
status = ExecutionStatus.FAILED
# 指数退避重试
if attempt < self.max_retries:
delay = self.base_delay * (2 ** (attempt - 1))
await asyncio.sleep(delay)
# 尝试降级
if self.use_fallback and tool_name in self._fallbacks:
try:
fallback_result = self._fallbacks[tool_name](**arguments)
return ExecutionResult(
status=ExecutionStatus.FALLBACK_USED,
result=fallback_result,
execution_time_ms=0,
tool_name=tool_name,
attempt=self.max_retries,
error=f"主工具失败，使用降级方案: {last_error}"
)
except Exception as e:
last_error = f"降级也失败: {e}"
return ExecutionResult(
status=ExecutionStatus.FAILED,
result=None,
execution_time_ms=0,
tool_name=tool_name,
attempt=self.max_retries,
error=last_error
)
注册降级函数
executor = ToolExecutor(max_retries=3, timeout=10.0)
executor.register_fallback("weather_query", lambda **kw: {"city": kw.get("city"), "temperature": "N/A", "source": "fallback"})4.2 完整调用链集成
import openai
class AgentToolCallChain:
"""完整的 Agent 工具调用链"""
def __init__(
self,
api_key: str,
registry: ToolRegistry,
executor: ToolExecutor,
discovery: ToolDiscovery,
base_url: str = "https://api.holysheep.ai/v1"
):
self.client = openai.OpenAI(
api_key=api_key,
base_url=base_url # 使用 HolySheep API 中转
)
self.registry = registry
self.executor = executor
self.discovery = discovery
self.max_iterations = 10
self.execution_log: list[ExecutionResult] = []
async def run(self, user_message: str) -> str:
"""
运行 Agent 处理用户消息
自动处理工具调用循环
"""
messages = [
{"role": "system", "content": "你是一个智能助手，可以通过工具来扩展能力。"}
]
messages.append({"role": "user", "content": user_message})
iteration = 0
while iteration < self.max_iterations:
iteration += 1
# 获取当前上下文相关的工具
available_tools = self.discovery.build_tools_for_llm(user_message)
# 调用 LLM
response = self.client.chat.completions.create(
model="gpt-4.1",
messages=messages,
tools=available_tools if available_tools else None,
temperature=0.7
)
assistant_message = response.choices[0].message
messages.append({
"role": "assistant",
"content": assistant_message.content,
"tool_calls": assistant_message.tool_calls
})
# 检查是否需要调用工具
if not assistant_message.tool_calls:
# 没有更多工具调用，返回最终回复
return assistant_message.content
# 执行工具调用
for tool_call in assistant_message.tool_calls:
tool_name = tool_call.function.name
arguments = json.loads(tool_call.function.arguments)
tool_func = self.registry.get_tool(tool_name)
if not tool_func:
tool_result = {
"error": f"工具 {tool_name} 不存在"
}
else:
# 执行工具
exec_result = await self.executor.execute(
tool_func=tool_func,
tool_name=tool_name,
arguments=arguments
)
self.execution_log.append(exec_result)
if exec_result.status == ExecutionStatus.SUCCESS:
tool_result = exec_result.result
else:
tool_result = {
"error": exec_result.error,
"status": exec_result.status.value
}
# 添加工具结果到对话
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(tool_result, ensure_ascii=False)
})
return "已达到最大迭代次数，请简化您的请求"
def get_execution_summary(self) -> dict:
"""获取执行摘要"""
if not self.execution_log:
return {"total_calls": 0}
success = sum(1 for r in self.execution_log if r.status == ExecutionStatus.SUCCESS)
avg_time = sum(r.execution_time_ms for r in self.execution_log) / len(self.execution_log)
return {
"total_calls": len(self.execution_log),
"success_count": success,
"success_rate": f"{success/len(self.execution_log)*100:.1f}%",
"avg_execution_time_ms": f"{avg_time:.2f}"
}
初始化 Agent（使用 HolySheep API）
agent = AgentToolCallChain(
api_key="YOUR_HOLYSHEEP_API_KEY", # 替换为你的 HolySheep API Key
registry=registry,
executor=executor,
discovery=discovery
)
运行异步任务
async def main():
result = await agent.run("北京今天天气怎么样？顺便帮我计算一下 256 * 128 的结果")
print(result)
print(agent.get_execution_summary())
asyncio.run(main())
五、实战案例：构建多工具协作系统
在 HolySheep API 环境下，我构建了一个实际的多工具协作系统，用于自动化数据分析和报告生成。这个系统整合了搜索、计算、数据处理等多个工具。
from typing import TypedDict
class ReportAgent:
"""数据分析报告生成 Agent"""
def __init__(self, api_key: str):
self.client = openai.OpenAI(
api_key=api_key,
base_url="https://api.holyshe