Home >Backend Development >Python Tutorial >Agent Tool Development Guide: From Design to Optimization
Imagine you're assembling a super-intelligent robot butler (Agent). This robot needs various tools to help you complete tasks - just like Doraemon's 4D pocket. This article will teach you how to create these powerful tools to make your AI butler more capable and efficient.
Think of using a self-service coffee machine:
This is a typical synchronous tool pattern. The Agent calls the tool and waits for immediate results - quick and simple.
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 } }
Use cases:
Imagine ordering food through a delivery APP:
This is how asynchronous tools work, perfect for tasks that take longer to process.
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) }
Use cases:
Just like all electrical appliances follow unified socket standards, our tool interfaces need standardization. This ensures all tools work perfectly with the Agent.
Imagine writing a product manual, you need to clearly tell users:
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"] } }
Just like all electrical appliances need power switches and power interfaces, all tools need to follow basic specifications:
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
Just like household appliances need protection against water, shock, and overload, tools need comprehensive protection mechanisms.
Imagine handling express delivery:
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 } }
Like automatically arranging a second delivery when the first attempt fails:
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) }
Like a convenience store placing popular items in prominent positions:
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"] } }
Like a hospital's appointment system, controlling the number of simultaneous services:
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
Like quality inspection before a new product launch:
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 }
Like writing a detailed and clear product manual:
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)
Developing good Agent tools is like crafting a perfect toolbox:
Remember: Good tools can make Agents twice as effective, while poor tools will limit Agents at every turn.
The above is the detailed content of Agent Tool Development Guide: From Design to Optimization. For more information, please follow other related articles on the PHP Chinese website!