Agent工具开发指南：从设计到优化

一、简介

想象一下你正在组装一个超级智能的机器人管家（特工）。这个机器人需要各种工具来帮助你完成任务——就像哆啦A梦的4D口袋一样。本文将教您如何创建这些强大的工具，让您的AI管家更加得力和高效。

2. 两个核心工具设计模式

2.1 同步工具：即时响应模式

考虑使用自助咖啡机：

1. 插入硬币并按“美式咖啡”按钮
2. 稍等几秒
3. 咖啡流出，可以喝了

这是典型的同步工具模式。代理调用该工具并等待立即结果 - 快速而简单。

class WeatherTool(BaseTool):
    """Weather Query Tool - Synchronous Mode"""
    async def execute(self, city: str) -> dict:
        # Simple and direct like pressing a coffee machine button
        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 异步工具：任务跟踪模式

想象一下通过外卖应用程序订购食物：

1. 下单后，APP会给你一个订单号
2. 您可以随时查看订单状态
3. 配送完成后APP会通知您

这就是异步工具的工作原理，非常适合需要较长时间处理的任务。

class DocumentAnalysisTool(BaseTool):
    """Document Analysis Tool - Asynchronous Mode"""

    async def start_task(self, file_path: str) -> str:
        # Like placing a food delivery order, returns a task 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:
        # Like checking food delivery status
        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. 工具接口标准化：建立通用规范

就像所有电器都遵循统一的插座标准一样，我们的工具接口也需要标准化。这可确保所有工具与代理完美配合。

3.1 工具说明规格

想象一下写一份产品手册，你需要清楚地告诉用户：

• 该工具的作用
• 需要什么参数
• 会返回什么结果

from pydantic import BaseModel, Field

class ToolSchema(BaseModel):
    """Tool Manual Template"""
    name: str = Field(..., description="Tool name")
    description: str = Field(..., description="Tool purpose description")
    parameters: dict = Field(..., description="Required parameters")
    required: List[str] = Field(default_factory=list, description="Required parameters")

    class Config:
        schema_extra = {
            "example": {
                "name": "Weather Query",
                "description": "Query weather information for specified city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "City name"
                        }
                    }
                },
                "required": ["city"]
            }
        }

3.2 统一基类

就像所有电器都需要电源开关和电源接口一样，所有工具都需要遵循基本规格：

class BaseTool(ABC):
    """Base template for all tools"""

    @abstractmethod
    def get_schema(self) -> ToolSchema:
        """Tool manual"""
        pass

    def validate_input(self, params: Dict) -> Dict:
        """Parameter check, like a fuse in electrical appliances"""
        return ToolSchema(**params).dict()

    @abstractmethod
    async def execute(self, **kwargs) -> Dict:
        """Actual functionality execution"""
        pass

4. 错误处理：让工具更可靠

就像家用电器需要防水、防震、过载保护一样，工具也需要完善的保护机制。

4.1 错误分类及处理

想象一下处理快递：

• 地址错误→参数错误
• 系统维护→服务暂时无法使用
• 快递太忙→需要限制速率并重试

class WeatherTool(BaseTool):
    """Weather Query Tool - Synchronous Mode"""
    async def execute(self, city: str) -> dict:
        # Simple and direct like pressing a coffee machine button
        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
            }
        }

4.2 重试机制

就像第一次尝试失败时自动安排第二次送货：

class DocumentAnalysisTool(BaseTool):
    """Document Analysis Tool - Asynchronous Mode"""

    async def start_task(self, file_path: str) -> str:
        # Like placing a food delivery order, returns a task 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:
        # Like checking food delivery status
        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)
        }

5.性能优化：让工具更高效

5.1 缓存机制

就像便利店一样，将热门商品放在显眼的位置：

from pydantic import BaseModel, Field

class ToolSchema(BaseModel):
    """Tool Manual Template"""
    name: str = Field(..., description="Tool name")
    description: str = Field(..., description="Tool purpose description")
    parameters: dict = Field(..., description="Required parameters")
    required: List[str] = Field(default_factory=list, description="Required parameters")

    class Config:
        schema_extra = {
            "example": {
                "name": "Weather Query",
                "description": "Query weather information for specified city",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "City name"
                        }
                    }
                },
                "required": ["city"]
            }
        }

5.2 并发控制

像医院的预约系统，控制同时服务的数量：

class BaseTool(ABC):
    """Base template for all tools"""

    @abstractmethod
    def get_schema(self) -> ToolSchema:
        """Tool manual"""
        pass

    def validate_input(self, params: Dict) -> Dict:
        """Parameter check, like a fuse in electrical appliances"""
        return ToolSchema(**params).dict()

    @abstractmethod
    async def execute(self, **kwargs) -> Dict:
        """Actual functionality execution"""
        pass

6. 测试和文档：确保工具可靠性

6.1 单元测试

比如新产品上市前的质量检查：

class ToolError(Exception):
    """Tool error base class"""
    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:
        # Execute specific operation
        result = await self._do_work(**kwargs)
        return {"status": "success", "data": result}
    except ValidationError:
        # Parameter error, like wrong address
        return {"status": "error", "code": "INVALID_PARAMS"}
    except RateLimitError as e:
        # Need rate limiting, like courier too busy
        return {
            "status": "error", 
            "code": "RATE_LIMIT",
            "retry_after": e.retry_after
        }

6.2 文档标准

就像写一份详细清晰的产品手册：

class RetryableTool(BaseTool):
    @retry(
        stop=stop_after_attempt(3),  # Maximum 3 retries
        wait=wait_exponential(multiplier=1, min=4, max=10)  # Increasing wait time
    )
    async def execute_with_retry(self, **kwargs):
        return await self.execute(**kwargs)

七、总结

开发好的代理工具就像制作一个完美的工具箱：

1. 正确的工具分类 - 同步/异步各有其用途
2. 标准化接口-方便统一管理
3. 保护机制-处理各种异常
4. 追求效率——需要时缓存，需要时限速
5. 质量焦点 - 彻底的测试，清晰的文档

记住：好的工具可以让 Agent 的效率加倍，而差的工具则会处处限制 Agent。

以上是Agent工具开发指南：从设计到优化的详细内容。更多信息请关注PHP中文网其他相关文章！