1. Introduction
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.
2. Two Core Tool Design Patterns
2.1 Synchronous Tools: Instant Response Mode
Think of using a self-service coffee machine:
- Insert coins and press the "Americano" button
- Wait for a few seconds
- Coffee flows out, ready to drink
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:
- Quick queries: weather, exchange rates, simple calculations
- Simple operations: sending messages, switch controls
- Real-time feedback: verification code checks, balance inquiries
2.2 Asynchronous Tools: Task Tracking Mode
Imagine ordering food through a delivery APP:
- After placing an order, the APP gives you an order number
- You can check the order status anytime
- The APP notifies you when delivery is complete
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:
- Time-consuming operations: large file processing, data analysis
- Multi-step tasks: video rendering, report generation
- Progress tracking needed: model training, batch processing
3. Tool Interface Standardization: Establishing Universal Specifications
Just like all electrical appliances follow unified socket standards, our tool interfaces need standardization. This ensures all tools work perfectly with the Agent.
3.1 Tool Description Specifications
Imagine writing a product manual, you need to clearly tell users:
- What the tool does
- What parameters are needed
- What results will be returned
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 Unified Base Class
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
4. Error Handling: Making Tools More Reliable
Just like household appliances need protection against water, shock, and overload, tools need comprehensive protection mechanisms.
4.1 Error Classification and Handling
Imagine handling express delivery:
- Wrong address → Parameter error
- System maintenance → Service temporarily unavailable
- Courier too busy → Need rate limiting and retry
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
}
4.2 Retry Mechanism
Like automatically arranging a second delivery when the first attempt fails:
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)
5. Performance Optimization: Making Tools More Efficient
5.1 Caching Mechanism
Like a convenience store placing popular items in prominent positions:
class CachedSearchTool(BaseTool):
def __init__(self):
self.cache = {} # Simple memory cache
self.cache_ttl = 3600 # Cache for 1 hour
async def execute(self, query: str):
# First check if it's on the "shelf"
cache_key = f"search:{query}"
if cache_key in self.cache:
return self.cache[cache_key]
# If not, get it from the "warehouse"
result = await self._do_search(query)
self.cache[cache_key] = result
return result
5.2 Concurrency Control
Like a hospital's appointment system, controlling the number of simultaneous services:
class RateLimiter:
def __init__(self, max_concurrent: int = 5):
self._semaphore = Semaphore(max_concurrent) # Handle max 5 requests simultaneously
@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. Testing and Documentation: Ensuring Tool Reliability
6.1 Unit Testing
Like quality inspection before a new product launch:
class TestWeatherTool:
@pytest.mark.asyncio
async def test_normal_weather(self):
"""Test normal weather query"""
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):
"""Test invalid city name"""
tool = WeatherTool()
result = await tool.execute(city="NonexistentCity")
assert result["status"] == "error"
6.2 Documentation Standards
Like writing a detailed and clear product manual:
class WeatherTool(BaseTool):
"""
Weather Query Tool
Function: Query real-time weather information for specified cities
Usage Example:
```
python
tool = WeatherTool()
result = await tool.execute(city="Beijing")
print(f"Temperature: {result['data']['temperature']}°C")
```
Notes:
1. City name must be a valid Chinese city name
2. Maximum 10 queries per minute
"""
7. Summary
Developing good Agent tools is like crafting a perfect toolbox:
- Proper tool classification - Sync/Async each has its use
- Standardized interfaces - Easy for unified management
- Protection mechanisms - Handle various exceptions
- Pursuit of efficiency - Cache when needed, rate limit when necessary
- Quality focus - Thorough testing, clear documentation
Remember: Good tools can make Agents twice as effective, while poor tools will limit Agents at every turn.
Top comments (0)