主题
原文:https://www.mstx.cn/harness/421.html · 创建:2026-09-05 · 更新:2026-09-05 · 阅读:1099
工具是Agent的"手"
回顾一下你已经构建的Harness:你有一个Runtime驱动核心循环,一个MemorySystem管理对话历史,一个ToolRegistry管理工具调用。这个架构能跑通,但工具的定义和注册方式还很粗糙——你手动写JSON Schema,手动注册每个工具,没有统一的规范。
在真实项目中,你会面对这样的问题:
- 团队中有5个开发者,每个人写的工具定义风格都不一样
- 工具数量从10个增长到50个,管理变得越来越困难
- 同一个工具被多个Agent复用,但每个Agent需要不同的权限配置
- 工具的Schema和实际函数签名不一致,导致LLM填的参数和函数期望的参数不匹配
这些问题指向一个核心需求:工具的定义和注册需要一套统一的规范。 这套规范要解决三个问题:如何描述一个工具、如何验证工具定义的完整性、如何管理工具的生命周期。
本节将构建一个完整的工具定义与注册系统。这是第4章的基础——后续各节讨论的工具安全、错误处理、沙箱执行,都建立在本节定义的工具模型之上。
工具定义的完整模型
一个工具不仅仅是"一个函数加一段描述"。从工程角度看,一个完整的工具定义需要包含以下信息:
python
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对象太繁琐。用装饰器来定义工具会更自然:
python
# 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")这个装饰器的设计极大简化了工具的定义。看一个使用示例:
python
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:
python
@registry.tool(
description="查询指定城市的天气信息",
parameters={
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称,如'北京'、'上海'。支持中文名称。"
}
},
"required": ["city"]
}
)
def get_weather(city: str) -> str:
...工具的验证
注册工具时,需要验证定义的完整性。你在3.3节中已经实现了基本的Schema验证,这里扩展为对完整ToolDefinition的验证:
python
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的迭代,工具会被修改、废弃、替换。一个完整的工具注册表需要支持生命周期管理:
python
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是订单编号,可以从用户的对话中提取。"
"返回订单的当前状态(待支付/已发货/已送达)和预计到达时间。"
)好的描述包含四个要素:
- 是什么:这个工具做什么
- 何时用:什么场景下应该调用这个工具
- 怎么用:参数从哪里来,如何正确填写
- 返回什么:调用后会得到什么结果
一个完整的示例
将以上内容整合,看一个完整的工具定义和使用流程:
python
# 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"用得准"的工具描述和参数定义。