回顾一下你已经构建的Harness：你有一个Runtime驱动核心循环，一个MemorySystem管理对话历史，一个ToolRegistry管理工具调用。这个架构能跑通，但工具的定义和注册方式还很粗糙——你手动写JSON Schema，手动注册每个工具，没有统一的规范。
在真实项目中，你会面对这样的问题：
团队中有5个开发者，每个人写的工具定义风格都不一样
工具数量从10个增长到50个，管理变得越来越困难
同一个工具被多个Agent复用，但每个Agent需要不同的权限配置
工具的Schema和实际函数签名不一致，导致LLM填的参数和函数期望的参数不匹配
这些问题指向一个核心需求：工具的定义和注册需要一套统一的规范。 这套规范要解决三个问题：如何描述一个工具、如何验证工具定义的完整性、如何管理工具的生命周期。
本节将构建一个完整的工具定义与注册系统。这是第4章的基础——后续各节讨论的工具安全、错误处理、沙箱执行，都建立在本节定义的工具模型之上。
一个工具不仅仅是"一个函数加一段描述"。从工程角度看，一个完整的工具定义需要包含以下信息：
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, List, Optional
from enum import Enum
class ToolCategory(Enum):
"""工具的分类"""
READ = "read" # 读操作：查询信息，无副作用
WRITE = "write" # 写操作：修改数据，有副作用
EXECUTE = "execute" # 执行操作：运行代码或命令
COMMUNICATE = "communicate" # 通信操作：发送消息、邮件等
DANGEROUS = "dangerous" # 危险操作：删除数据、执行交易等
class ToolPermission(Enum):
"""工具执行所需的权限级别"""
AUTO = "auto" # 自动执行，无需确认
USER_CONFIRM = "user_confirm" # 需要用户确认后执行
ADMIN_ONLY = "admin_only" # 仅管理员可触发
@dataclass
class ToolDefinition:
"""一个工具的完整定义"""
name: str # 工具名称（唯一标识）
description: str # 工具描述（给LLM看）
func: Callable # 实际的Python函数
parameters: Dict[str, Any] # 参数JSON Schema
category: ToolCategory = ToolCategory.READ
permission: ToolPermission = ToolPermission.AUTO
timeout_seconds: int = 10 # 执行超时
max_retries: int = 2 # 失败重试次数
max_result_length: int = 2000 # 结果最大长度
tags: List[str] = field(default_factory=list) # 标签，用于工具发现
version: str = "1.0.0" # 工具版本
examples: List[str] = field(default_factory=list) # 使用示例
deprecated: bool = False # 是否已废弃
metadata: Dict[str, Any] = field(default_factory=dict) # 额外元数据
这个模型比你在3.2节中使用的"名称+描述+函数+参数"要丰富得多。让我们逐一理解每个字段的用途：
category（分类）：标识工具的副作用级别。这决定了工具执行的默认安全策略——READ类工具可以自动执行，DANGEROUS类工具需要额外的确认。
permission（权限）：标识工具执行所需的权限级别。AUTO意味着Agent可以自主调用，USER_CONFIRM意味着每次调用前需要用户确认，ADMIN_ONLY意味着只有管理员能触发。
timeout_seconds：工具独立的超时设置。不同工具有不同的合理超时——查询数据库可能5秒就够，调用外部API可能需要30秒。
max_retries：工具独立的失败重试次数。一个偶尔超时的查询工具可以多试几次，一个发送邮件的工具不应该重试（可能重复发送）。
max_result_length：工具独立的结果截断阈值。一个列表查询工具的结果可能很大，需要截断；一个状态查询工具的结果很小，不需要截断。
tags：用于工具发现。当工具数量增多时，你不需要把所有工具都塞给LLM，而是根据对话内容选择相关的工具。标签是工具选择的基础。
version：工具版本。当工具的行为发生变化时，版本号帮助你追踪哪个Agent使用的是哪个版本的工具。
examples：工具的使用示例。这些示例可以包含在描述中，帮助LLM理解如何正确使用工具。
deprecated：标记工具是否已废弃。废弃的工具不应该被新Agent使用，但为了兼容性可能暂时保留。
直接构造ToolDefinition对象太繁琐。用装饰器来定义工具会更自然：
# harness/tools.py 的扩展
import functools
import inspect
from typing import Any, Callable, Dict, Optional, get_type_hints
class ToolRegistry:
def __init__(self, tool_timeout: int = 10, max_result_length: int = 2000):
self._tools: Dict[str, ToolDefinition] = {}
self.tool_timeout = tool_timeout
self.max_result_length = max_result_length
def tool(self,
name: Optional[str] = None,
description: Optional[str] = None,
category: ToolCategory = ToolCategory.READ,
permission: ToolPermission = ToolPermission.AUTO,
timeout_seconds: Optional[int] = None,
max_retries: Optional[int] = None,
max_result_length: Optional[int] = None,
tags: Optional[List[str]] = None,
examples: Optional[List[str]] = None):
"""装饰器：将一个Python函数注册为工具"""
def decorator(func: Callable) -> Callable:
# 自动推断工具名称
tool_name = name or func.__name__
# 自动推断描述（从docstring的第一行）
tool_description = description
if tool_description is None:
doc = inspect.getdoc(func)
tool_description = doc.split('\n')[0] if doc else tool_name
# 自动生成参数Schema（从函数签名和类型注解）
tool_parameters = self._generate_schema_from_function(func)
# 注册工具
self.register(
definition=ToolDefinition(
name=tool_name,
description=tool_description,
func=func,
parameters=tool_parameters,
category=category,
permission=permission,
timeout_seconds=timeout_seconds or self.tool_timeout,
max_retries=max_retries if max_retries is not None else 2,
max_result_length=max_result_length or self.max_result_length,
tags=tags or [],
examples=examples or [],
)
)
# 返回原函数，不影响其他使用
return func
return decorator
def _generate_schema_from_function(self, func: Callable) -> Dict[str, Any]:
"""从函数的签名和类型注解自动生成参数Schema"""
signature = inspect.signature(func)
type_hints = get_type_hints(func) if hasattr(func, '__annotations__') else {}
properties = {}
required = []
for param_name, param in signature.parameters.items():
# 跳过self参数（如果是方法）
if param_name == 'self':
continue
# 获取类型注解
param_type = type_hints.get(param_name, None)
json_type = self._python_type_to_json_type(param_type)
# 获取默认值
has_default = param.default != inspect.Parameter.empty
default_value = param.default if has_default else None
# 构建参数定义
param_def = {
"type": json_type,
"description": f"参数 {param_name}", # 默认描述，可后续手动优化
}
if has_default:
param_def["default"] = default_value
else:
required.append(param_name)
properties[param_name] = param_def
return {
"type": "object",
"properties": properties,
"required": required,
}
def _python_type_to_json_type(self, python_type) -> str:
"""将Python类型映射为JSON Schema类型"""
if python_type is None:
return "string"
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object",
}
# 处理Optional类型
origin = getattr(python_type, "__origin__", None)
if origin is not None:
# Optional[X] -> X
args = getattr(python_type, "__args__", ())
non_none_args = [a for a in args if a is not type(None)]
if non_none_args:
return self._python_type_to_json_type(non_none_args[0])
return type_map.get(python_type, "string")
这个装饰器的设计极大简化了工具的定义。看一个使用示例：
registry = ToolRegistry()
@registry.tool(
description="查询指定城市的天气信息",
category=ToolCategory.READ,
tags=["weather", "query"],
examples=["get_weather(city='北京') -> '晴，26°C'"],
)
def get_weather(city: str) -> str:
"""查询指定城市的天气信息"""
mock_data = {
"北京": "晴，26°C",
"上海": "多云，29°C",
}
return mock_data.get(city, f"暂无{city}的天气数据")
装饰器的自动推断逻辑：如果函数有类型注解（如city: str），装饰器会自动生成参数Schema；如果函数有docstring，装饰器会自动提取第一行作为描述。这样，一个简单函数加上类型注解，就能被注册为工具。
但自动推断有局限：自动生成的参数描述是"参数 city"这样的占位文本。对于需要精确参数描述的工具，你仍然需要手动提供详细的Schema。装饰器允许你在需要时覆盖自动生成的Schema：
@registry.tool(
description="查询指定城市的天气信息",
parameters={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称，如'北京'、'上海'。支持中文名称。"
}
},
"required": ["city"]
}
)
def get_weather(city: str) -> str:
...
注册工具时，需要验证定义的完整性。你在3.3节中已经实现了基本的Schema验证，这里扩展为对完整ToolDefinition的验证：
class ToolValidationError(Exception):
"""工具定义验证失败"""
pass
class ToolRegistry:
def _validate_definition(self, definition: ToolDefinition):
"""验证工具定义的完整性"""
errors = []
# 名称验证
if not definition.name or not definition.name.strip():
errors.append("工具名称不能为空")
elif not definition.name.replace('_', '').isalnum():
errors.append(f"工具名称 '{definition.name}' 只能包含字母、数字和下划线")
# 描述验证
if len(definition.description) < 10:
errors.append(
f"工具 '{definition.name}' 的描述过短（{len(definition.description)}字符）。"
f"描述应该清晰说明工具的用途，建议至少20字符。"
)
# 参数Schema验证
if definition.parameters.get("type") != "object":
errors.append(f"工具 '{definition.name}' 的parameters必须是object类型")
properties = definition.parameters.get("properties", {})
required = definition.parameters.get("required", [])
for prop_name, prop_def in properties.items():
if "type" not in prop_def:
errors.append(f"工具 '{definition.name}' 的参数 '{prop_name}' 缺少type定义")
if "description" not in prop_def:
errors.append(f"工具 '{definition.name}' 的参数 '{prop_name}' 缺少description")
if prop_def.get("description") == f"参数 {prop_name}":
errors.append(
f"工具 '{definition.name}' 的参数 '{prop_name}' 使用了默认描述。"
f"请提供有意义的参数描述。"
)
for req_param in required:
if req_param not in properties:
errors.append(
f"工具 '{definition.name}' 的required参数 '{req_param}' 不在properties中"
)
# 函数签名与Schema的一致性验证
import inspect
sig = inspect.signature(definition.func)
func_params = [p for p in sig.parameters if p != 'self']
# 检查Schema中的参数是否在函数签名中
for prop_name in properties:
if prop_name not in func_params:
errors.append(
f"工具 '{definition.name}' 的Schema参数 '{prop_name}' "
f"不在函数签名中。函数参数：{func_params}"
)
# 检查required参数是否在函数签名中
for req_param in required:
if req_param not in func_params:
errors.append(
f"工具 '{definition.name}' 的required参数 '{req_param}' "
f"不在函数签名中"
)
# 分类与权限的组合验证
if definition.category == ToolCategory.DANGEROUS:
if definition.permission == ToolPermission.AUTO:
errors.append(
f"危险工具 '{definition.name}' 不能设置为AUTO权限。"
f"危险操作至少需要USER_CONFIRM权限。"
)
if errors:
raise ToolValidationError(
f"工具 '{definition.name}' 定义验证失败：\n" +
"\n".join(f" - {e}" for e in errors)
)
验证的核心价值：在注册时就发现问题，而非运行时。一个工具定义的问题（比如Schema参数和函数参数不匹配）如果不及时发现，会导致LLM填了参数但函数调用失败。
工具不是静态的。随着Agent的迭代，工具会被修改、废弃、替换。一个完整的工具注册表需要支持生命周期管理：
class ToolRegistry:
def deprecate(self, name: str, reason: str = ""):
"""标记工具为已废弃"""
if name not in self._tools:
raise KeyError(f"工具 '{name}' 不存在")
self._tools[name].deprecated = True
self._tools[name].metadata["deprecation_reason"] = reason
print(f"警告：工具 '{name}' 已标记为废弃。原因：{reason}")
def remove(self, name: str):
"""完全移除工具"""
if name not in self._tools:
raise KeyError(f"工具 '{name}' 不存在")
del self._tools[name]
def get_active_tools(self) -> List[ToolDefinition]:
"""获取所有未废弃的工具"""
return [t for t in self._tools.values() if not t.deprecated]
def get_tools_by_category(self, category: ToolCategory) -> List[ToolDefinition]:
"""按分类获取工具"""
return [t for t in self._tools.values()
if t.category == category and not t.deprecated]
def get_tools_by_tag(self, tag: str) -> List[ToolDefinition]:
"""按标签获取工具"""
return [t for t in self._tools.values()
if tag in t.tags and not t.deprecated]
def get_schemas(self, only_active: bool = True) -> list:
"""获取所有工具的Schema（默认只返回活跃工具）"""
tools = self.get_active_tools() if only_active else list(self._tools.values())
schemas = []
for tool in tools:
schemas.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.parameters,
}
})
return schemas
生命周期管理的核心原则：
废弃不是删除：一个工具被废弃后，仍然保留在注册表中（标记为deprecated=True），但不会被新Agent使用。这保证了旧版本的轨迹记录仍然可查询。
删除是最终操作：只有确认工具不再被任何地方引用后，才彻底删除。
活跃工具是默认选择：get_schemas()默认只返回活跃工具，废弃工具自动被排除。
工具的描述是LLM理解工具用途的唯一途径。一个写得好的描述和写得差的描述，对Agent表现的影响是巨大的。
一个差描述的例子：
description="处理用户请求"
这个描述什么都没说。LLM不知道这个工具是做什么的、什么时候该用、什么时候不该用。
一个好描述的例子：
description=(
"查询用户在系统中的订单状态。"
"当用户询问'我的订单到哪了'、'订单状态'、'发货了吗'等问题时使用此工具。"
"参数order_id是订单编号，可以从用户的对话中提取。"
"返回订单的当前状态（待支付/已发货/已送达）和预计到达时间。"
)
好的描述包含四个要素：
是什么：这个工具做什么
何时用：什么场景下应该调用这个工具
怎么用：参数从哪里来，如何正确填写
返回什么：调用后会得到什么结果
将以上内容整合，看一个完整的工具定义和使用流程：
# examples/tool_definition_demo.py
from harness.tools import ToolRegistry, ToolCategory, ToolPermission
# 创建注册表
registry = ToolRegistry()
# 用装饰器定义工具
@registry.tool(
description=(
"查询指定城市的天气信息。"
"当用户询问天气相关问题时使用此工具。"
"参数city是城市名称，如'北京'、'上海'。"
"返回天气状况和温度。"
),
category=ToolCategory.READ,
permission=ToolPermission.AUTO,
tags=["weather", "query"],
examples=["get_weather(city='北京') -> '晴，26°C'"],
)
def get_weather(city: str) -> str:
"""查询指定城市的天气信息"""
mock_data = {
"北京": "晴，26°C，湿度40%",
"上海": "多云，29°C，湿度65%",
}
return mock_data.get(city, f"暂无{city}的天气数据")
@registry.tool(
description=(
"向指定邮箱发送邮件。"
"当用户要求发送邮件、通知某人时使用此工具。"
"参数to是收件人邮箱，subject是邮件主题，body是邮件正文。"
"使用此工具前应确认收件人和内容。"
),
category=ToolCategory.COMMUNICATE,
permission=ToolPermission.USER_CONFIRM, # 发邮件需要用户确认
tags=["email", "communication"],
timeout_seconds=30,
max_retries=1, # 邮件不重试，避免重复发送
)
def send_email(to: str, subject: str, body: str) -> str:
"""发送邮件"""
# 模拟发送
return f"邮件已发送至{to}，主题：{subject}"
# 查看注册的工具
print("注册的工具：")
for name, tool in registry._tools.items():
print(f" {name} | {tool.category.value} | {tool.permission.value} | v{tool.version}")
# 获取Schema
schemas = registry.get_schemas()
print(f"\n工具Schema数：{len(schemas)}")
# 演示废弃
registry.deprecate("get_weather", "使用新的天气服务工具替代")
active = registry.get_active_tools()
print(f"\n活跃工具数：{len(active)}")
print(f"活跃工具：{[t.name for t in active]}")
本节构建了一个完整的工具定义与注册系统。核心内容包括：
ToolDefinition：工具的完整数据模型，涵盖分类、权限、超时、重试、版本等维度
装饰器注册：用@registry.tool()简化工具定义，自动推断Schema和描述
验证机制：在注册时检查定义的完整性，早发现问题
生命周期管理：废弃、删除、过滤活跃工具
这套系统的核心设计原则是：工具的每一个属性都有明确的工程意义。 分类决定了安全策略，权限决定了执行门槛，超时和重试决定了容错能力，版本和废弃状态决定了生命周期管理。
在下一节中，我们将讨论工具Schema设计的深入话题——如何写出让LLM"用得准"的工具描述和参数定义。
评论专区
评论加载中...登录后即可发表评论