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 查看订单状态
- 送达时,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="北京")
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="不存在的城市")
assert result["status"] == "error"
6.2 文档规范
就像产品说明书要详细清晰：
class WeatherTool(BaseTool):
"""
天气查询工具
功能：查询指定城市的实时天气信息
使用示例:
```python
tool = WeatherTool()
result = await tool.execute(city="北京")
print(f"温度: {result['data']['temperature']}°C")
```
注意事项:
1. 城市名必须是有效的中国城市名称
2. 每分钟最多查询 10 次
"""
7. 总结
开发好的 Agent 工具就像打造称手的工具箱：
- 工具分类要合理 - 同步/异步各有用途
- 接口要标准 - 便于统一管理
- 要有保护机制 - 处理各种异常情况
- 追求高效 - 该缓存缓存,该限流限流
- 重视质量 - 测试充分,文档清晰
记住：好的工具能让 Agent 事半功倍,糟糕的工具会让 Agent 处处受限。
标签：指南,task,self,Agent,result,async,工具,优化,def From： https://www.cnblogs.com/muzinan110/p/18554494