7.2 工具定义与注册
在上一节中，我们系统学习了 Function Calling 的基本原理：大语言模型通过特殊的提示工程与结构化输出，能够"决定"调用哪个函数、传入哪些参数，然后由外部运行时执行函数并将结果回传给模型。这一机制打通了模型与外部世界的隔阂，是 Agent 具备行动能力的基础。
然而，Function Calling 只是解决了"模型如何表达调用意图"的问题。在实际的 Agent 系统中，我们面对的挑战远不止于此：一个 Agent 可能需要同时管理天气查询、数据库操作、文件搜索、邮件发送等数十种工具；这些工具有的同步执行、有的异步调用，有的需要认证，有的存在版本迭代和依赖关系。如何系统化地定义、注册、发现和管理这些工具，正是本节要解决的核心问题。
本节将从工具的抽象设计出发，逐步构建一个完整的工具注册与管理体系，涵盖工具接口标准、注册中心设计、自动发现机制、依赖管理与版本控制等内容。
7.2.1 工具抽象设计
在 Agent 系统中，工具是模型与外部世界交互的最小执行单元。可以把工具理解为 Agent 的"手"——模型是"大脑"，负责决策；工具是"手"，负责执行。一个好的工具抽象应该具备以下特征：
- 自描述性：工具自身携带完整的元信息（名称、描述、参数），模型无需额外文档即可理解如何使用它
- 可组合性：工具之间可以组合使用，一个工具的输出可以作为另一个工具的输入
- 可管理性：支持注册、注销、启用、禁用等生命周期操作
- 可发现性：Agent 运行时能够按名称、分类、关键词等方式查找所需工具
下面是工具抽象层的核心组成结构图：
┌─────────────────────────────────────────────────┐
│ Tool 抽象层 │
├─────────────────────────────────────────────────┤
│ ┌──────────┐ ┌──────────┐ ┌──────────────┐ │
│ │ 元数据 │ │ 参数Schema │ │ 执行逻辑 │ │
│ │ - name │ │ - type │ │ - callable │ │
│ │ - desc │ │ - required│ │ - async/sync │ │
│ │ - version│ │ - enum │ │ - error handle│ │
│ └──────────┘ └──────────┘ └──────────────┘ │
├─────────────────────────────────────────────────┤
│ ┌──────────────────────────────────────────┐ │
│ │ 生命周期管理 │ │
│ │ - 注册(register) - 发现(discover) │ │
│ │ - 启用(enable) - 禁用(disable) │ │
│ │ - 升级(upgrade) - 回滚(rollback) │ │
│ └──────────────────────────────────────────┘ │
└─────────────────────────────────────────────────┘工具抽象层由三个核心模块和一组生命周期管理能力组成。元数据描述"工具是什么"，参数 Schema 描述"工具需要什么输入"，执行逻辑描述"工具做什么"。三者缺一不可——缺少元数据，模型无法发现工具；缺少参数 Schema，模型无法正确调用；缺少执行逻辑，工具无法产生实际效果。
工具接口标准设计
一个标准的工具接口应包含以下要素。我们先定义工具分类枚举和元数据结构：
python
from abc import ABC, abstractmethod # ABC：抽象基类，用于定义工具接口规范
from typing import Any, Dict, Optional, Callable # 类型标注，提升代码可读性和IDE支持
from dataclasses import dataclass, field # dataclass：用装饰器自动生成__init__等方法
from enum import Enum # Enum：定义枚举类型，用于工具分类
import json # 用于JSON序列化（如导出工具Schema）
class ToolCategory(Enum):
"""工具分类枚举——每个工具必须归属一个分类，便于管理和筛选"""
DATA_QUERY = "data_query" # 数据查询类：如数据库查询、缓存读取
ACTION = "action" # 执行操作类：如发邮件、创建任务
COMPUTATION = "computation" # 计算类：如数学运算、统计聚合
EXTERNAL_API = "external_api" # 外部API调用类：如天气、汇率查询
FILE_IO = "file_io" # 文件操作类：如读写文件、目录搜索ToolCategory 枚举是工具分类的基石。使用枚举而非自由字符串，可以避免拼写错误，也让 IDE 提供自动补全。接下来定义工具元数据：
python
@dataclass
class ToolMetadata:
"""工具元数据——描述工具"是什么"的所有静态信息"""
name: str # 工具唯一标识，模型通过此名称调用工具
description: str # 功能描述，这是模型理解工具用途的唯一线索，务必清晰准确
category: ToolCategory # 工具分类，用于分组管理和批量操作
version: str = "1.0.0" # 语义化版本号，格式为 主版本.次版本.修订号
author: str = "" # 作者信息，便于追溯和维护责任
tags: list = field(default_factory=list) # 标签列表，支持多维度搜索（如["天气","实时"]）
deprecated: bool = False # 是否已废弃，废弃工具不再推荐使用但仍保留兼容
requires_auth: bool = False # 是否需要认证，标记后Agent运行时会注入凭证description 字段是工具定义中最重要的部分之一——它是模型判断"是否应该调用此工具"的主要依据。一个模糊的描述（如"查询数据"）会导致模型频繁误调用，而一个精确的描述（如"执行SELECT语句查询MySQL数据库，返回JSON格式结果"）能大幅提升调用准确率。
下面定义工具参数结构：
python
@dataclass
class ToolParameter:
"""工具参数定义——描述工具"需要什么输入""""
name: str # 参数名，与execute方法的关键字参数名一致
type: str # 参数类型，对应JSON Schema类型：string/integer/number/boolean/array/object
description: str # 参数说明，告诉模型这个参数的含义和取值范围
required: bool = False # 是否必填，True表示模型必须提供此参数
enum: Optional[list] = None # 枚举值，限制参数只能取这些值中的一个
default: Any = None # 默认值，当模型未提供此参数时使用有了元数据和参数定义后，我们来构建工具基类。BaseTool 是所有工具的抽象接口，它强制子类实现 execute 方法，同时提供了向 OpenAI 和 Anthropic 两种 API 格式转换的能力：
python
class BaseTool(ABC):
"""工具基类 - 所有工具的抽象接口，定义了工具必须实现的契约"""
def __init__(self, metadata: ToolMetadata, parameters: list[ToolParameter]):
self.metadata = metadata # 存储工具元数据
self.parameters = parameters # 存储参数定义列表
self._enabled = True # 工具启用状态，默认启用
@property
def name(self) -> str:
"""快捷属性，方便通过 tool.name 访问工具名"""
return self.metadata.name
@abstractmethod
async def execute(self, **kwargs) -> Dict[str, Any]:
"""执行工具逻辑（子类必须实现）。
使用async以支持异步操作（如网络请求、数据库查询）。
所有参数通过**kwargs传入，返回值为字典格式。
"""
pass
def to_openai_schema(self) -> Dict[str, Any]:
"""转换为 OpenAI 兼容的工具定义格式。
OpenAI的function calling使用嵌套的function结构，
strict=True表示强制模型严格遵循参数Schema。
"""
properties = {} # 参数属性字典
required = [] # 必填参数名列表
for param in self.parameters: # 遍历所有参数定义
prop = {"type": param.type, "description": param.description}
if param.enum: # 如果参数有枚举约束
prop["enum"] = param.enum # 添加到属性中
properties[param.name] = prop # 存入属性字典
if param.required: # 如果是必填参数
required.append(param.name) # 加入必填列表
return {
"type": "function", # OpenAI格式固定为function类型
"function": {
"name": self.metadata.name,
"description": self.metadata.description,
"strict": True, # 启用严格模式，防止模型自由发挥
"parameters": {
"type": "object",
"properties": properties,
"required": required,
"additionalProperties": False # 禁止额外属性，防止模型传入未定义的参数
}
}
}
def to_anthropic_schema(self) -> Dict[str, Any]:
"""转换为 Anthropic 兼容的工具定义格式。
Anthropic的格式更扁平，使用input_schema而非嵌套function结构。
"""
properties = {}
required = []
for param in self.parameters: # 与OpenAI格式相同的参数处理逻辑
prop = {"type": param.type, "description": param.description}
if param.enum:
prop["enum"] = param.enum
properties[param.name] = prop
if param.required:
required.append(param.name)
return {
"name": self.metadata.name, # Anthropic格式：工具名直接在顶层
"description": self.metadata.description,
"input_schema": { # 使用input_schema而非parameters
"type": "object",
"properties": properties,
"required": required # Anthropic默认允许额外属性，不强制strict
}
}可以看到，to_openai_schema 和 to_anthropic_schema 的参数处理逻辑几乎完全相同，差异仅在于外层包装结构。在实际项目中，可以抽取一个 _build_properties 私有方法来消除重复代码。这里为了展示清晰，保留了完整逻辑。
设计要点：
additionalProperties: False是一个容易被忽略但非常重要的设置。它阻止模型"发明"参数 Schema 中不存在的字段。如果不设置，模型有时会自作主张地添加如format、timezone等未定义参数，导致执行报错。
具体工具实现示例
有了 BaseTool 之后，定义具体工具就变得很直观了。以下是一个天气查询工具的实现：
python
class WeatherTool(BaseTool):
"""天气查询工具——调用外部天气API获取实时天气信息"""
def __init__(self):
# 调用父类构造器，传入元数据和参数定义
super().__init__(
metadata=ToolMetadata(
name="get_weather", # 工具名，模型通过此名称调用
description="获取指定城市的实时天气信息，包括温度、湿度、天气状况", # 清晰描述功能
category=ToolCategory.EXTERNAL_API, # 外部API类
version="1.0.0",
tags=["天气", "实时数据"] # 标签，便于搜索
),
parameters=[
ToolParameter("city", "string", "城市名称", required=True), # 必填参数
ToolParameter("unit", "string", "温度单位", enum=["celsius", "fahrenheit"]), # 枚举参数
]
)
async def execute(self, city: str, unit: str = "celsius") -> Dict[str, Any]:
"""执行天气查询。实际项目中这里会调用天气API。"""
# 实际应调用天气 API，这里模拟返回数据
return {
"city": city,
"temperature": 26,
"unit": unit,
"condition": "晴",
"humidity": 45,
"timestamp": "2024-07-21T10:00:00Z"
}再看一个带安全检查的数据库查询工具。这个工具展示了如何在内置执行逻辑中加入输入验证，防止注入攻击：
python
class DatabaseQueryTool(BaseTool):
"""数据库查询工具——执行SQL查询并返回结果"""
def __init__(self, db_connection):
# 注意：db_connection通过构造器注入，实现依赖反转
super().__init__(
metadata=ToolMetadata(
name="query_database",
description="执行 SQL 查询并返回结果，支持 SELECT 语句",
category=ToolCategory.DATA_QUERY,
version="1.0.0",
requires_auth=True # 标记需要认证，提示Agent运行时注入凭证
),
parameters=[
ToolParameter("query", "string", "SQL 查询语句", required=True),
ToolParameter("limit", "integer", "返回行数上限"),
]
)
self.db = db_connection # 保存数据库连接，供execute使用
async def execute(self, query: str, limit: int = 100) -> Dict[str, Any]:
"""执行 SQL 查询，内置安全防护"""
# 安全检查：只允许SELECT语句，防止DELETE/UPDATE等危险操作
if not query.strip().upper().startswith("SELECT"):
return {"error": "仅支持 SELECT 查询"}
# 自动添加LIMIT子句，防止返回过多数据导致Token溢出
query = f"{query.rstrip(';')} LIMIT {limit}"
try:
results = await self.db.fetch(query) # 异步执行查询
return {"rows": results, "count": len(results)}
except Exception as e:
# 捕获异常并返回结构化错误信息，而非抛出异常
# 这样Agent可以根据错误信息决定重试或换用其他工具
return {"error": str(e)}DatabaseQueryTool 展示了两个重要的安全实践：一是输入校验，只允许 SELECT 语句；二是结果限制，自动添加 LIMIT 子句。在实际生产环境中，还应考虑参数化查询来防止 SQL 注入。
设计要点：
execute方法的错误处理应该返回结构化的错误字典，而不是抛出异常。原因是 Agent 运行时需要将工具结果回传给模型，如果直接抛异常会中断整个 Agent 流程。返回{"error": "..."}让模型有机会理解失败原因并采取补救措施。
7.2.2 注册与发现机制
工具定义好了之后，下一步就是要把它们管理起来。想象你有一个工具箱，里面放着扳手、螺丝刀、电钻等各种工具。如果你只是把工具随意丢进箱子里，用的时候翻箱倒柜找不到，那效率极低。合理的做法是给工具箱登记一份清单——每个工具有编号、有名称、有分类标签，清单上还记录了工具的状态（可用/损坏/已借出）。当你需要某个工具时，查清单即可快速定位。
工具注册中心（Tool Registry）就是 Agent 系统中的这份"工具箱清单"。它负责记录所有可用工具的元信息，提供查找、分类、状态管理等功能。随着 Agent 系统的复杂度增长，管理数十甚至上百个工具，一个中心化的注册机制不可或缺。
工具注册中心设计
下面是 ToolRegistry 的完整实现。这个类承担了工具生命周期的全部管理职责：
python
from typing import Dict, List, Optional, Type # 导入类型标注工具
import logging # 标准日志库，用于记录工具操作
logger = logging.getLogger(__name__) # 创建本模块的日志记录器
class ToolRegistry:
"""工具注册中心 - 管理所有工具的生命周期
类比工具箱清单：
- _tools 是清单正文，记录每个工具的实例
- _categories 是分类索引，方便按类别查找
- _deprecated 是废弃记录，标注替代方案
"""
def __init__(self):
self._tools: Dict[str, BaseTool] = {} # 清单正文：工具名 -> 工具实例
self._categories: Dict[ToolCategory, List[str]] = {} # 分类索引：分类 -> 工具名列表
self._deprecated: Dict[str, str] = {} # 废弃记录：工具名 -> 替代工具名
def register(self, tool: BaseTool) -> None:
"""注册一个工具——相当于往工具箱清单上添加一条新记录"""
name = tool.metadata.name # 取出工具名作为索引键
# 检查是否已存在同名工具（除非已废弃，否则警告覆盖）
if name in self._tools and not self._tools[name].metadata.deprecated:
logger.warning(f"工具 '{name}' 已存在，将被覆盖")
self._tools[name] = tool # 写入清单
# 更新分类索引——将工具名加入对应分类的列表
category = tool.metadata.category
if category not in self._categories:
self._categories[category] = [] # 该分类首次出现，创建空列表
if name not in self._categories[category]:
self._categories[category].append(name) # 追加到分类列表
logger.info(f"工具 '{name}' (v{tool.metadata.version}) 注册成功")
def register_many(self, tools: List[BaseTool]) -> None:
"""批量注册工具——适合初始化时一次性注册多个工具"""
for tool in tools:
self.register(tool)
def unregister(self, name: str) -> None:
"""注销工具——从清单中移除某个工具"""
if name in self._tools:
tool = self._tools.pop(name) # 从清单中删除
# 同时从分类索引中移除，保持索引一致性
category = tool.metadata.category
if category in self._categories:
self._categories[category].remove(name)
logger.info(f"工具 '{name}' 已注销")
def get(self, name: str) -> Optional[BaseTool]:
"""按名称获取工具实例——查清单"""
return self._tools.get(name)
def list_all(self, include_deprecated: bool = False) -> List[BaseTool]:
"""列出所有工具——浏览清单。
include_deprecated=False时，隐藏已废弃的工具。
"""
if include_deprecated:
return list(self._tools.values())
return [t for t in self._tools.values() if not t.metadata.deprecated]
def list_by_category(self, category: ToolCategory) -> List[BaseTool]:
"""按分类列出工具——按工具箱隔层查找"""
names = self._categories.get(category, []) # 从分类索引中获取工具名列表
return [self._tools[n] for n in names if n in self._tools] # 转换为工具实例列表
def search(self, query: str) -> List[BaseTool]:
"""搜索工具——按名称、描述或标签模糊匹配"""
results = []
query_lower = query.lower() # 统一转为小写进行不区分大小写匹配
for tool in self._tools.values():
# 三路匹配：工具名、描述文本、标签列表
if (query_lower in tool.metadata.name.lower() or
query_lower in tool.metadata.description.lower() or
any(query_lower in tag.lower() for tag in tool.metadata.tags)):
results.append(tool)
return results
def to_openai_tools(self, tool_names: Optional[List[str]] = None) -> List[Dict]:
"""导出为OpenAI工具列表格式——生成给模型的工具菜单。
tool_names参数可以指定只导出部分工具，实现工具的动态裁剪。
"""
tools = self._tools.values()
if tool_names: # 如果指定了工具名列表，只导出这些工具
tools = [self._tools[n] for n in tool_names if n in self._tools]
# 过滤掉已废弃的工具，只导出可用工具的Schema
return [t.to_openai_schema() for t in tools if not t.metadata.deprecated]
def deprecate(self, name: str, replacement: str = "") -> None:
"""标记工具为废弃——但不立即删除，保持向后兼容"""
if name in self._tools:
self._tools[name].metadata.deprecated = True # 标记废弃标志
if replacement:
self._deprecated[name] = replacement # 记录替代工具
logger.info(f"工具 '{name}' 已标记为废弃，替代: {replacement or '无'}")ToolRegistry 的设计有几点值得注意：
- 双索引结构：
_tools字典提供按名称的 O(1) 查找，_categories字典提供按分类的筛选能力。两者在注册和注销时同步更新，保持一致性。 - 软删除策略：
deprecate方法只标记而非删除工具。这样已有配置仍可引用旧工具名，同时list_all和to_openai_tools会自动过滤废弃项，不影响新用户。 - 动态裁剪：
to_openai_tools支持传入tool_names参数，允许 Agent 根据当前任务只暴露相关工具子集。这很重要——如果一次性把 100 个工具的 Schema 全发给模型，不仅消耗大量 Token，还会降低模型选择正确工具的准确率。
自动发现机制
对于大型项目，手动注册每个工具既繁琐又容易遗漏。就像仓库管理员不需要手动登记每一件货物——他只需要规定"所有贴了条码的货物进库时自动记录"，工具注册也可以通过装饰器模式实现自动注册。
下面实现一个 @tool 装饰器，它能在函数定义时自动提取参数信息并完成注册：
python
# 全局注册中心实例——整个进程共享一个工具箱
_global_registry = ToolRegistry()
def tool(name: str, description: str, category: ToolCategory = ToolCategory.ACTION,
version: str = "1.0.0", **kwargs):
"""装饰器：自动注册工具到全局注册中心。
用法：
@tool(name="send_email", description="发送邮件")
async def send_email(to: str, subject: str) -> dict:
...
装饰器会自动从函数签名推断参数类型和必填属性。
"""
def decorator(func):
# 从函数签名推断参数信息
import inspect # inspect模块用于内省函数签名
sig = inspect.signature(func) # 获取函数签名对象
parameters = [] # 收集参数定义
for param_name, param in sig.parameters.items(): # 遍历函数的每个参数
if param_name == 'self': # 跳过实例方法中的self参数
continue
# 推断参数类型：通过类型注解映射到JSON Schema类型
param_type = "string" # 默认类型为string
if param.annotation != inspect.Parameter.empty: # 如果有类型注解
type_map = {str: "string", int: "integer", float: "number", bool: "boolean"}
param_type = type_map.get(param.annotation, "string") # 查找映射，找不到则默认string
# 推断是否必填：没有默认值的参数为必填
is_required = param.default == inspect.Parameter.empty
parameters.append(
ToolParameter(param_name, param_type, f"{param_name}参数", required=is_required)
)
# 动态创建工具类——将普通函数包装为BaseTool子类
class AutoTool(BaseTool):
def __init__(self):
super().__init__(
metadata=ToolMetadata(
name=name,
description=description,
category=category,
version=version,
**kwargs # 透传额外的元数据字段（如tags）
),
parameters=parameters
)
async def execute(self, **kwargs_exec):
# 调用被装饰的原始函数
result = func(**kwargs_exec)
if inspect.iscoroutine(result): # 如果是协程函数，await等待结果
return await result
return result # 同步函数直接返回结果
tool_instance = AutoTool() # 实例化自动工具
_global_registry.register(tool_instance) # 自动注册到全局注册中心
return func # 返回原函数，保持向后兼容
return decorator
# ========== 使用装饰器自动注册 ==========
@tool(
name="send_email",
description="发送电子邮件给指定收件人",
category=ToolCategory.ACTION,
version="1.0.0"
)
async def send_email(to: str, subject: str, body: str) -> dict:
"""发送邮件"""
# 实际发送逻辑
return {"status": "sent", "to": to, "subject": subject}
@tool(
name="search_files",
description="在文件系统中搜索匹配的文件",
category=ToolCategory.FILE_IO
)
async def search_files(pattern: str, directory: str = ".") -> dict:
"""搜索文件"""
import glob
import os
results = glob.glob(os.path.join(directory, pattern), recursive=True)
return {"matches": results, "count": len(results)}
# 查看已注册工具
print("已注册工具:")
for t in _global_registry.list_all():
print(f" - {t.metadata.name} v{t.metadata.version}: {t.metadata.description}")装饰器模式的优势在于声明即注册——开发者只需在函数上加一行 @tool(...)，工具就自动进入注册中心，无需额外调用注册语句。这大幅减少了遗漏注册的可能性，也让代码更整洁。
7.2.3 依赖管理与版本控制
随着工具数量增长，工具之间的关系会变得复杂。"生成报告"工具可能需要先调用"数据查询"工具获取数据；"发送通知"工具可能依赖"模板渲染"工具。如果不显式管理这些依赖关系，运行时容易出现"工具找不到所需前置数据"或"依赖版本不兼容"等问题。
工具依赖声明
下面实现一个依赖声明和检查机制：
python
@dataclass
class ToolDependency:
"""工具依赖声明——描述一个工具依赖另一个工具的什么版本"""
tool_name: str # 依赖的工具名
version_constraint: str # 版本约束，如 ">=1.0.0,<2.0.0"
optional: bool = False # 是否可选依赖（可选依赖缺失不报错）
class ToolWithDependencies(BaseTool):
"""支持依赖管理的工具基类"""
def __init__(self, metadata, parameters, dependencies=None):
super().__init__(metadata, parameters)
self.dependencies: List[ToolDependency] = dependencies or [] # 依赖列表，默认为空
def check_dependencies(self, registry: ToolRegistry) -> List[str]:
"""检查依赖是否满足，返回缺失的依赖描述列表。
在工具执行前调用此方法，确保所有前置条件已满足。
"""
missing = [] # 收集所有缺失的依赖
for dep in self.dependencies: # 遍历声明的每个依赖
tool = registry.get(dep.tool_name) # 从注册中心查找依赖工具
if not tool: # 依赖工具不存在
if not dep.optional: # 如果是必选依赖
missing.append(f"缺失依赖: {dep.tool_name}")
else:
# 版本检查（简化版，实际应解析版本约束表达式）
from packaging import version # 使用packaging库进行版本比较
if not version.parse(tool.metadata.version) in dep.version_constraint:
missing.append(
f"版本不匹配: {dep.tool_name} 需要 {dep.version_constraint}, "
f"实际 {tool.metadata.version}"
)
return missing # 返回所有未满足的依赖check_dependencies 方法在工具执行前提供了一个"预检"步骤。如果返回的列表非空，说明存在未满足的依赖，Agent 可以选择先注册缺失的工具或告知用户无法完成此任务。
版本控制策略
工具会迭代升级——修复 bug、增加功能、调整接口。在生产环境中，同时维护多个版本的工具是常见需求：新版工具可能引入了不兼容的改动，但旧版调用方尚未迁移。版本管理器让我们能够多版本共存、按需切换、必要时回滚。
python
class ToolVersionManager:
"""工具版本管理器——支持多版本共存与版本切换"""
def __init__(self):
# 二维字典：工具名 -> {版本号 -> 工具实例}
self._versions: Dict[str, Dict[str, BaseTool]] = {}
# 当前激活版本：工具名 -> 版本号
self._active_versions: Dict[str, str] = {}
def register_version(self, tool: BaseTool) -> None:
"""注册一个工具版本——往版本库中添加一个版本"""
name = tool.metadata.name # 工具名
ver = tool.metadata.version # 版本号
if name not in self._versions: # 首次注册该工具
self._versions[name] = {} # 创建版本字典
self._versions[name][ver] = tool # 存入该版本
# 默认激活最新版本（简单字符串比较，生产中应使用语义化版本比较）
if name not in self._active_versions or ver > self._active_versions[name]:
self._active_versions[name] = ver
def get_active(self, name: str) -> Optional[BaseTool]:
"""获取当前激活版本的工具实例"""
if name in self._active_versions:
ver = self._active_versions[name] # 获取激活版本号
return self._versions.get(name, {}).get(ver) # 查找并返回对应实例
return None
def rollback(self, name: str, version: str) -> bool:
"""回滚到指定版本——在发现新版本有问题时快速恢复"""
if name in self._versions and version in self._versions[name]:
self._active_versions[name] = version # 切换激活版本
return True
return False # 目标版本不存在，回滚失败
def list_versions(self, name: str) -> List[str]:
"""列出工具的所有版本——用于查看可用版本和选择回滚目标"""
return sorted(self._versions.get(name, {}).keys()) # 排序后返回版本列表
# ========== 使用示例 ==========
version_manager = ToolVersionManager()
# 注册不同版本
weather_v1 = WeatherTool() # version 1.0.0
weather_v2 = WeatherTool() # 假设升级到 2.0.0
weather_v2.metadata.version = "2.0.0"
weather_v2.metadata.description = "支持多城市同时查询" # 新版本功能增强
version_manager.register_version(weather_v1) # 注册v1
version_manager.register_version(weather_v2) # 注册v2，自动设为激活版本
# 当前使用的是 v2.0.0
active = version_manager.get_active("get_weather")
print(f"当前版本: {active.metadata.version}") # 输出: 2.0.0
# 回滚到 v1.0.0
version_manager.rollback("get_weather", "1.0.0")
active = version_manager.get_active("get_weather")
print(f"回滚后版本: {active.metadata.version}") # 输出: 1.0.0版本管理器的核心是 _versions 二维字典和 _active_versions 指针。每个工具名下挂多个版本，_active_versions 记录当前生效的是哪个版本。rollback 方法只需修改指针即可实现版本切换，无需重新创建工具实例，非常高效。
设计要点：示例中的版本比较使用了简单的字符串比较（
ver > self._active_versions[name]）。在生产环境中，应使用packaging.version库进行语义化版本比较，确保"2.0.0" > "1.9.9"的判断是正确的（字符串比较在此处恰好也对，但"10.0.0" > "2.0.0"用字符串比较会出错）。
7.2.4 完整实战示例
下面将前面所有组件整合为一个可运行的完整示例。这个示例展示了从工具定义、注册到使用的完整流程：
python
import asyncio
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from enum import Enum
import json
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# ========== 工具分类与元数据 ==========
class ToolCategory(Enum):
SEARCH = "search" # 搜索类
DATABASE = "database" # 数据库类
API = "api" # 外部API类
FILE = "file" # 文件类
UTILITY = "utility" # 工具类
@dataclass
class ToolMeta:
"""简化版工具元数据"""
name: str # 工具名
description: str # 功能描述
category: ToolCategory = ToolCategory.UTILITY # 分类，默认为通用工具
version: str = "1.0.0" # 版本号
tags: List[str] = field(default_factory=list) # 标签列表
# ========== 工具类 ==========
class Tool:
"""简化的工具类——封装元数据、参数Schema和执行函数"""
def __init__(self, meta: ToolMeta, params_schema: Dict, executor):
self.meta = meta # 工具元数据
self.params_schema = params_schema # 参数的JSON Schema
self._executor = executor # 执行函数（可以是同步或异步）
self.enabled = True # 启用状态
async def run(self, **kwargs) -> Dict[str, Any]:
"""执行工具，自动处理同步/异步函数和异常"""
try:
result = self._executor(**kwargs) # 调用执行函数
if asyncio.iscoroutine(result): # 如果是协程，await它
result = await result
return {"success": True, "data": result} # 成功时包装结果
except Exception as e:
return {"success": False, "error": str(e)} # 失败时返回错误信息
def to_openai_tool(self) -> Dict:
"""转换为OpenAI工具格式"""
return {
"type": "function",
"function": {
"name": self.meta.name,
"description": self.meta.description,
"parameters": self.params_schema
}
}
# ========== 注册中心 ==========
class ToolRegistry:
"""工具注册中心——整合注册、查找、分类、搜索和导出功能"""
def __init__(self):
self._tools: Dict[str, Tool] = {} # 工具字典
self._by_category: Dict[ToolCategory, List[str]] = {} # 分类索引
def register(self, tool: Tool):
"""注册工具并更新分类索引"""
self._tools[tool.meta.name] = tool # 存入工具字典
cat = tool.meta.category # 获取分类
self._by_category.setdefault(cat, []).append(tool.meta.name) # 追加到分类索引
logger.info(f"注册工具: {tool.meta.name} v{tool.meta.version}")
def get(self, name: str) -> Optional[Tool]:
"""按名获取工具"""
return self._tools.get(name)
def list_enabled(self) -> List[Tool]:
"""列出所有已启用的工具"""
return [t for t in self._tools.values() if t.enabled]
def by_category(self, cat: ToolCategory) -> List[Tool]:
"""按分类列出工具"""
return [self._tools[n] for n in self._by_category.get(cat, []) if n in self._tools]
def search(self, keyword: str) -> List[Tool]:
"""关键词搜索工具（匹配名称、描述和标签）"""
kw = keyword.lower()
return [
t for t in self._tools.values()
if kw in t.meta.name.lower() or kw in t.meta.description.lower()
or any(kw in tag.lower() for tag in t.meta.tags)
]
def to_openai_tools(self) -> List[Dict]:
"""导出为OpenAI工具列表格式"""
return [t.to_openai_tool() for t in self.list_enabled()]
# ========== 创建工具并注册 ==========
registry = ToolRegistry()
# 工具1: 天气查询
registry.register(Tool(
meta=ToolMeta("get_weather", "获取城市天气", ToolCategory.API, tags=["天气"]),
params_schema={
"type": "object",
"properties": {
"city": {"type": "string", "description": "城市名"}
},
"required": ["city"]
},
executor=lambda city: {"city": city, "temp": 28, "condition": "晴"}
))
# 工具2: 计算器
registry.register(Tool(
meta=ToolMeta("calculator", "执行数学计算", ToolCategory.UTILITY, tags=["计算"]),
params_schema={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
},
executor=lambda expression: {"result": eval(expression)}
))
# 工具3: 数据库查询
registry.register(Tool(
meta=ToolMeta("db_query", "查询数据库", ToolCategory.DATABASE, tags=["数据", "SQL"]),
params_schema={
"type": "object",
"properties": {
"sql": {"type": "string", "description": "SELECT查询语句"},
"limit": {"type": "integer", "description": "返回行数", "default": 10}
},
"required": ["sql"]
},
executor=lambda sql, limit=10: {"rows": [{"id": 1, "name": "示例"}], "count": 1}
))
# ========== 测试功能 ==========
async def main():
# 列出所有工具
print("=" * 50)
print("所有已注册工具:")
for t in registry.list_enabled():
print(f" {t.meta.name} v{t.meta.version} [{t.meta.category.value}]")
print(f" {t.meta.description}")
# 搜索工具
print("\n搜索 '查询':")
for t in registry.search("查询"):
print(f" {t.meta.name}")
# 按分类查看
print("\nAPI 类工具:")
for t in registry.by_category(ToolCategory.API):
print(f" {t.meta.name}")
# 执行工具
weather_tool = registry.get("get_weather")
result = await weather_tool.run(city="Beijing")
print(f"\n执行 get_weather: {result}")
# 导出 OpenAI 格式
print("\nOpenAI 工具定义:")
for t in registry.to_openai_tools():
print(f" {json.dumps(t, ensure_ascii=False, indent=2)[:200]}...")
asyncio.run(main())运行上述代码，你将看到注册中心的完整工作流程：注册三个工具后，可以通过 list_enabled 列出全部工具，通过 search 按关键词搜索，通过 by_category 按分类筛选，通过 get 获取单个工具并执行，最后通过 to_openai_tools 导出给模型使用的工具 Schema。
使用 LangChain 的工具体系
在实际项目中，不必从零构建工具系统。LangChain 框架提供了成熟的工具定义方案，以下演示三种常见用法：
python
from langchain_core.tools import tool, StructuredTool # 导入LangChain工具组件
from pydantic import BaseModel, Field # Pydantic用于参数校验
from typing import Optional, List
import asyncio
# ========== 方式一：@tool 装饰器（自动推断参数） ==========
@tool
def search_web(query: str, num_results: int = 5) -> str:
"""搜索互联网，返回相关结果摘要。
Args:
query: 搜索关键词
num_results: 返回结果数量，默认5条
"""
# 模拟搜索
return f"搜索 '{query}' 返回 {num_results} 条结果"
# ========== 方式二：Pydantic 模型定义复杂参数 ==========
class WeatherInput(BaseModel):
"""天气查询参数——使用Pydantic模型获得类型校验和自动补全"""
city: str = Field(description="城市名称，如 Beijing") # 必填，有描述
date: Optional[str] = Field(default=None, description="日期，格式 YYYY-MM-DD") # 可选
include_forecast: bool = Field(default=False, description="是否包含未来预报") # 带默认值
@tool(args_schema=WeatherInput)
def get_weather_detail(city: str, date: Optional[str] = None, include_forecast: bool = False) -> str:
"""获取详细天气信息，支持指定日期和预报查询"""
forecast = f"，未来三天: 晴->多云->小雨" if include_forecast else ""
return f"{city} 天气: 28°C 晴{forecast}"
# ========== 方式三：StructuredTool 手动构建 ==========
def calculate(expression: str) -> str:
"""安全计算数学表达式"""
import re
if not re.match(r'^[\d\s\+\-\*\/\(\)\.]+$', expression): # 只允许数字和运算符
return "错误：表达式包含非法字符"
try:
return f"计算结果: {eval(expression)}"
except Exception as e:
return f"计算错误: {str(e)}"
calculator_tool = StructuredTool.from_function(
func=calculate, # 传入执行函数
name="calculator", # 指定工具名
description="安全计算数学表达式，支持加减乘除和括号"
)
# ========== 使用工具 ==========
tools = [search_web, get_weather_detail, calculator_tool]
print("已注册工具:")
for t in tools:
print(f" {t.name}: {t.description[:50]}...")
print(f" Schema: {t.args_schema.schema() if t.args_schema else 'N/A'}")
# 测试工具调用
async def test_tools():
result = await search_web.ainvoke({"query": "LangChain"})
print(f"\nsearch_web: {result}")
result = await get_weather_detail.ainvoke({
"city": "Shanghai",
"include_forecast": True
})
print(f"get_weather_detail: {result}")
result = await calculator_tool.ainvoke({"expression": "15 * 3 + 7"})
print(f"calculator: {result}")
asyncio.run(test_tools())LangChain 的 @tool 装饰器会自动从函数的 docstring 和类型注解生成工具 Schema，大幅减少了样板代码。Pydantic 模型则提供了更强的参数校验能力——当模型传入的参数不符合类型要求时，Pydantic 会自动报错并给出清晰的错误信息。
7.2.5 常见误区
在实践中，工具定义与注册环节有不少容易踩的坑。以下列出几个最常见的误区：
误区一：工具描述过于简短或模糊
python
# 错误：描述太模糊，模型无法判断何时该用这个工具
ToolMeta(name="query", description="查询数据")
# 正确：描述具体到数据源、操作类型和返回格式
ToolMeta(name="query_mysql", description="执行SELECT语句查询MySQL数据库，返回JSON格式结果，单次最多100行")工具的 description 是模型唯一的选型依据。如果多个工具的描述都写着"查询数据"，模型就无法区分该调用哪个。好的描述应包含三个要素：做什么（查询数据）、在哪做（MySQL数据库）、返回什么（JSON格式结果）。
误区二：一次暴露过多工具给模型
python
# 错误：一次性把所有工具Schema发给模型
tools = registry.to_openai_tools() # 可能有50+个工具
# 正确：根据当前任务只暴露相关工具
relevant_tools = registry.search("天气") # 只取天气相关工具
tools = [t.to_openai_schema() for t in relevant_tools]每个工具的 Schema 都会占用上下文 Token。当工具数量超过 15-20 个时，模型选择正确工具的准确率会显著下降。最佳实践是按任务上下文动态裁剪工具集——先通过语义搜索或分类筛选出相关工具，再导出给模型。
误区三：在 execute 方法中直接抛出异常
python
# 错误：异常会中断整个Agent流程
async def execute(self, query: str) -> Dict:
results = await self.db.fetch(query) # 如果数据库挂了，直接抛异常
return {"rows": results}
# 正确：捕获异常并返回结构化错误，让模型有机会重试
async def execute(self, query: str) -> Dict:
try:
results = await self.db.fetch(query)
return {"rows": results, "count": len(results)}
except Exception as e:
return {"error": str(e), "hint": "数据库连接失败，建议稍后重试"}Agent 运行时需要将工具的返回结果序列化为文本回传给模型。如果 execute 抛出未捕获异常，整个 Agent 循环会崩溃。返回 {"error": ...} 让模型理解失败原因，可能选择重试或换用其他工具。
误区四：忽略参数校验导致安全风险
python
# 错误：直接信任模型传入的SQL，存在注入风险
async def execute(self, query: str) -> Dict:
return await self.db.fetch(query) # 模型可能生成 DELETE FROM users
# 正确：白名单校验+参数化查询
async def execute(self, query: str) -> Dict:
if not query.strip().upper().startswith("SELECT"): # 只允许SELECT
return {"error": "仅支持SELECT查询"}
return await self.db.fetch(query) # 生产环境还应使用参数化查询模型生成的参数不可完全信任。任何涉及文件系统、数据库、Shell 命令的工具，都必须在 execute 方法中加入输入校验和权限控制。
误区五：工具命名不规范导致冲突
python
# 错误：命名不统一，容易冲突
@tool(name="GetWeather", description="...")
@tool(name="get-weather", description="...")
@tool(name="get_weather", description="...") # 三个工具本质相同但名字不同
# 正确：统一使用snake_case命名，语义清晰
@tool(name="get_weather", description="获取指定城市的天气信息")工具名应统一使用 snake_case 风格（小写+下划线），且全局唯一。同一个系统中不应出现 getWeather、get-weather、get_weather 三种命名风格共存的情况。
误区六：版本比较使用字符串而非语义化版本
python
# 错误：字符串比较在版本号超过10时会出错
"10.0.0" > "2.0.0" # False！因为字符 '1' < '2'
# 正确：使用packaging库进行语义化版本比较
from packaging import version
version.parse("10.0.0") > version.parse("2.0.0") # True当工具版本号的主版本号达到两位数（如 10.0.0）时，字符串比较会产生错误结果。始终使用 packaging.version 库进行版本比较，确保语义正确。
7.2.6 本节小结
本节从工具抽象设计出发，完整构建了一个工具定义与注册体系。核心要点如下：
| 序号 | 核心要点 |
|---|---|
| 1 | 工具抽象包含三部分：元数据（名称、描述、版本）、参数 Schema（类型、必填、枚举）、执行逻辑（异步调用、错误处理） |
| 2 | 工具描述是模型选型的唯一依据，应具体说明工具做什么、操作什么、返回什么格式 |
| 3 | 注册中心是工具管理的核心，采用双索引结构（按名称查找 + 按分类筛选），支持注册、注销、搜索和废弃管理 |
| 4 | 自动发现通过 @tool 装饰器实现声明即注册，利用 inspect 模块自动推断参数类型和必填属性 |
| 5 | 依赖管理通过 ToolDependency 声明工具间依赖，执行前预检确保前置条件满足 |
| 6 | 版本控制支持多版本共存和版本切换，回滚操作仅需修改版本指针，高效且无侵入 |
| 7 | 安全实践包括输入校验（白名单过滤）、结果限制（自动 LIMIT）、错误封装（返回而非抛出）三道防线 |
| 8 | 动态裁剪根据任务上下文只暴露相关工具子集，避免过多工具 Schema 消耗 Token 和干扰模型选择 |
| 9 | LangChain 提供 @tool 装饰器、Pydantic 参数模型和 StructuredTool 三种工具定义方式，各有适用场景 |
到目前为止，我们学习的工具定义与注册机制都是在单一进程内完成的——所有工具和注册中心都运行在同一个 Python 进程中。但在真实的 Agent 应用中，工具往往分布在不同的服务中：数据库工具运行在数据库服务器上，天气工具运行在第三方 API 服务上，文件工具运行在远程文件系统中。如何让 Agent 跨进程、跨服务地发现和调用工具？这就需要一种标准化的工具协议。
下一节（7.3）将介绍 MCP（Model Context Protocol）协议——一种由 Anthropic 提出的开放标准，专门用于规范模型与外部工具之间的通信。MCP 定义了工具的发现、描述和调用协议，使得不同来源的工具可以统一接入 Agent 系统，就像 USB 协议让各种外设都能接入电脑一样。理解了本节的工具抽象和注册机制后，你将能更好地理解 MCP 协议在分布式场景下的价值。