1. 引言
想象一下你正在组装一个超级智能的机器人管家（Agent）。这个机器人需要各种工具来帮助你完成任务——就像哆啦A梦的四维口袋一样。本文将教你如何创建这些强大的工具，使你的AI管家更有能力和效率。
2. 两种核心工具设计模式
2.1 同步工具：即时响应模式
想象使用自助咖啡机：
- 投币并按下“美式咖啡”按钮
- 等待几秒钟
- 咖啡流出，准备饮用
这是一种典型的同步工具模式。Agent调用工具并等待即时结果——快速而简单。
class WeatherTool(BaseTool):
"""天气查询工具 - 同步模式"""
async def execute(self, city: str) -> dict:
# 简单直接，就像按下咖啡机按钮
weather_data = await self.weather_api.get_current(city)
return {
"status": "success",
"data": {
"temperature": weather_data.temp,
"humidity": weather_data.humidity,
"description": weather_data.desc
}
}
使用场景：
- 快速查询：天气、汇率、简单计算
- 简单操作：发送消息、开关控制
- 实时反馈：验证码检查、余额查询
2.2 异步工具：任务跟踪模式
想象通过外卖APP点餐：
- 下单后，APP给你一个订单号
- 你可以随时查看订单状态
- 订单完成时，APP会通知你
这就是异步工具的工作方式，适合处理需要较长时间的任务。
class DocumentAnalysisTool(BaseTool):
"""文档分析工具 - 异步模式"""
async def start_task(self, file_path: str) -> str:
# 就像下外卖订单，返回任务ID
task_id = str(uuid.uuid4())
await self.task_queue.put({
"task_id": task_id,
"file_path": file_path,
"status": "processing"
})
return task_id
async def get_status(self, task_id: str) -> dict:
# 就像查看外卖订单状态
task = await self.task_store.get(task_id)
return {
"task_id": task_id,
"status": task["status"],
"progress": task.get("progress", 0),
"result": task.get("result", None)
}
使用场景：
- 耗时操作：大文件处理、数据分析
- 多步骤任务：视频渲染、报告生成
- 需要进度跟踪：模型训练、批量处理
3. 工具接口标准化：建立通用规范
就像所有电器遵循统一的插座标准，我们的工具接口也需要标准化。这确保所有工具都能完美地与Agent协作。
3.1 工具描述规范
想象编写产品手册，你需要清楚地告诉用户：
- 工具的功能
- 需要哪些参数
- 将返回什么结果
from pydantic import BaseModel, Field
class ToolSchema(BaseModel):
"""工具手册模板"""
name: str = Field(..., description="工具名称")
description: str = Field(..., description="工具用途描述")
parameters: dict = Field(..., description="所需参数")
required: List[str] = Field(default_factory=list, description="必需参数")
class Config:
schema_extra = {
"example": {
"name": "天气查询",
"description": "查询指定城市的天气信息",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "城市名称"
}
}
},
"required": ["city"]
}
}
3.2 统一基类
就像所有电器需要电源开关和电源接口，所有工具需要遵循基本规范：
class BaseTool(ABC):
"""所有工具的基本模板"""
@abstractmethod
def get_schema(self) -> ToolSchema:
"""工具手册"""
pass
def validate_input(self, params: Dict) -> Dict:
"""参数检查，就像电器中的保险丝"""
return ToolSchema(**params).dict()
@abstractmethod
async def execute(self, **kwargs) -> Dict:
"""实际功能执行"""
pass
4. 错误处理：让工具更可靠
就像家用电器需要防水、防震和过载保护，工具也需要全面的保护机制。
4.1 错误分类和处理
想象处理快递：
- 地址错误 → 参数错误
- 系统维护 → 服务暂时不可用
- 快递员太忙 → 需要限流和重试
class ToolError(Exception):
"""工具错误基类"""
def __init__(self, message: str, error_code: str, retry_after: Optional[int] = None):
self.message = message
self.error_code = error_code
self.retry_after = retry_after
@error_handler
async def execute(self, **kwargs):
try:
# 执行具体操作
result = await self._do_work(**kwargs)
return {"status": "success", "data": result}
except ValidationError:
# 参数错误，就像地址错误
return {"status": "error", "code": "INVALID_PARAMS"}
except RateLimitError as e:
# 需要限流，就像快递员太忙
return {
"status": "error",
"code": "RATE_LIMIT",
"retry_after": e.retry_after
}
4.2 重试机制
就像第一次投递失败后自动安排第二次投递：
class RetryableTool(BaseTool):
@retry(
stop=stop_after_attempt(3), # 最多重试3次
wait=wait_exponential(multiplier=1, min=4, max=10) # 增加等待时间
)
async def execute_with_retry(self, **kwargs):
return await self.execute(**kwargs)
5. 性能优化：让工具更高效
5.1 缓存机制
就像便利店将热门商品放在显眼位置：
class CachedSearchTool(BaseTool):
def __init__(self):
self.cache = {} # 简单的内存缓存
self.cache_ttl = 3600 # 缓存1小时
async def execute(self, query: str):
# 首先检查是否在“货架”上
cache_key = f"search:{query}"
if cache_key in self.cache:
return self.cache[cache_key]
# 如果不在，从“仓库”获取
result = await self._do_search(query)
self.cache[cache_key] = result
return result
5.2 并发控制
就像医院的预约系统，控制同时服务的数量：
class RateLimiter:
def __init__(self, max_concurrent: int = 5):
self._semaphore = Semaphore(max_concurrent) # 同时处理最多5个请求
@asynccontextmanager
async def acquire(self):
async with self._semaphore:
yield
class ApiTool(BaseTool):
def __init__(self):
self.rate_limiter = RateLimiter(max_concurrent=5)
async def execute(self, **kwargs):
async with self.rate_limiter.acquire():
return await self._call_api(**kwargs)
6. 测试和文档：确保工具可靠性
6.1 单元测试
就像新产品发布前的质量检验：
class TestWeatherTool:
@pytest.mark.asyncio
async def test_normal_weather(self):
"""测试正常天气查询"""
tool = WeatherTool()
result = await tool.execute(city="Beijing")
assert result["status"] == "success"
assert "temperature" in result["data"]
@pytest.mark.asyncio
async def test_invalid_city(self):
"""测试无效城市名称"""
tool = WeatherTool()
result = await tool.execute(city="NonexistentCity")
assert result["status"] == "error"
6.2 文档标准
就像编写详细清晰的产品手册：
class WeatherTool(BaseTool):
"""
天气查询工具
功能：查询指定城市的实时天气信息
使用示例：
```
python
tool = WeatherTool()
result = await tool.execute(city="Beijing")
print(f"Temperature: {result['data']['temperature']}°C")
```
注意事项：
1. 城市名称必须是有效的中国城市名称
2. 每分钟最多查询10次
"""
7. 总结
开发好的Agent工具就像打造一个完美的工具箱：
- 合理的工具分类——同步/异步各有用途
- 标准化的接口——便于统一管理
- 保护机制——处理各种异常
- 追求效率——需要时缓存，必要时限流
- 质量关注——全面测试，清晰文档
记住：好的工具可以让Agent事半功倍，而差的工具会在每个环节限制Agent的发挥。
