1. Introduction
Imagine you’re assembling a super-intelligent robot Agent. This robot needs a variety of tools to help you complete the task – just like Doraemon’s four-dimensional pocket. This article will teach you how to create these powerful tools to make your AI housekeeper more capable and efficient.
2. Two core tool design patterns
2.1 Synchronization Tool: Instant Response Mode
Imagine using a self-service coffee machine:
- Insert a coin and press the “Americano” button
- Wait a few seconds
- The coffee flows out and is ready to be drunk
This is a typical synchronization tool pattern. The agent invokes the tool and waits for immediate results – quickly and easily.
class WeatherTool(BaseTool):
""Weather Finder - Sync Mode"""
async def execute(self, city: str) -> dict:
# Simple and straightforward, like pressing a coffee maker 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
}
}
Usage Scenarios:
- Quick query: weather, exchange rates, simple calculations
- Simple operation: send messages, switch control
- Real-time feedback: CAPTCHA check, balance inquiry
2.2 Asynchronous Tools: Task Tracking Mode
Imagine ordering food through a food delivery app:
- After placing an order, the APP gives you an order number
- You can check the status of your order at any time
- The APP will notify you when the order is completed
This is how asynchronous tools work, and are suitable for tasks that take a long time.
class DocumentAnalysisTool(BaseTool):
Document Analysis Tool - Asynchronous Mode
async def start_task(self, file_path: str) -> str:
# It's like placing a takeout order and returning the 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:
# It's like checking the status of a takeaway order
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)
}
Usage Scenarios:
- Time-consuming operations: large file processing and data analysis
- Multi-step tasks: video rendering, report generation
- Progress tracking is required: model training, batch processing
3. Standardization of tool interfaces: Establish common specifications
Just as all appliances follow a uniform socket standard, our tool interfaces need to be standardized. This ensures that all tools work perfectly with the Agent.
3.1 Tool Description Specification
Imagine writing a product manual where you need to clearly tell the user:
- Functions of the tool
- What parameters are required
- 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 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 the weather information of a specified city",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "City Name"
}
}
},
"required": ["city"]
}
}
3.2 Unified Base Classes
Just as all appliances require power switches and power interfaces, all tools need to follow basic specifications:
class BaseTool(ABC):
Basic Templates for All Tools
@abstractmethod
def get_schema(self) -> ToolSchema:
""Tool Manual"""
pass
def validate_input(self, params: Dict) -> Dict:
""""Parameter checking, like a fuse in an appliance""""
return ToolSchema(**params).dict()
@abstractmethod
async def execute(self, **kwargs) -> Dict:
""Actual Function Execution""
pass
4. Error handling: Make the tool more reliable
Just as household appliances need protection against water, shock, and overload, tools need comprehensive protection mechanisms.
4.1 Misclassification and Handling
Imagine handling a courier:
- Wrong address → The parameter is incorrect
- The System Maintenance → service is temporarily unavailable
- courier is too busy → needs to throttle and retries
The
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:
# Perform specific actions
result = await self._do_work(**kwargs)
return {"status": "success", "data": result}
except ValidationError:
# The parameter is wrong, just like the wrong address
return {"status": "error", "code": "INVALID_PARAMS"}
except RateLimitError as e:
# Need to limit the flow, just like the courier is too busy
return {
"status": "error",
"code": "RATE_LIMIT",
"retry_after": e.retry_after
}
4.2 Retry Mechanism
It’s like automatically scheduling a second delivery after the first delivery fails:
class RetryableTool(BaseTool):
@retry(
stop=stop_after_attempt(3), # Retry up to 3 times
wait=wait_exponential(multiplier=1, min=4, max=10) # Increase the wait time
)
async def execute_with_retry(self, **kwargs):
return await self.execute(**kwargs)
5. Performance optimization: Make the tool more efficient
5.1 Caching Mechanisms
It’s like a convenience store that puts hot items in a prominent position:
class CachedSearchTool(BaseTool):
def __init__(self):
self.cache = {} # Simple memory cache
self.cache_ttl = 3600 # Cached 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 repository
result = await self._do_search(query)
self.cache[cache_key] = result
return result
5.2 Concurrency Control
It’s 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 up to 5 requests at the same time
@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: Ensure tool reliability
6.1 Unit Testing
It’s like a quality check before a new product is launched:
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
It’s like writing a detailed and clear product manual:
class WeatherTool(BaseTool):
"""
Weather lookup tool
Function: Query the real-time weather information of a specified city
Examples:
```
python
tool = WeatherTool()
result = await tool.execute(city="Beijing")
print(f"Temperature: {result['data']['temperature']}°C")
```
Notes:
1. The city name must be a valid Chinese city name
2. A maximum of 10 queries per minute
"""
7. Summary
Developing a good agent tool is like building a perfect toolbox:
- Reasonable categorization of tools – synchronous/asynchronous has its own purpose
- Standardized interface – easy to manage in a unified manner
- Protection Mechanism – Handle all kinds of exceptions
- Pursuit of efficiency – caching when needed and throttling when necessary
- Quality focus – thorough testing, clear documentation
Remember: good tools can make the agent do more with less, while bad tools will limit the agent’s performance in every link.