DEV Community

James Lee
James Lee

Posted on

Agent Tool Development Guide: From Design to Optimization

#ai #aiagent #python #llm

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:

  1. Insert coins and press the "Americano" button
  2. Wait for a few seconds
  3. 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 } }
Enter fullscreen mode Exit fullscreen mode

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:

  1. After placing an order, the APP gives you an order number
  2. You can check the order status anytime
  3. 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) }
Enter fullscreen mode Exit fullscreen mode

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"] } }
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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 }
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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"
Enter fullscreen mode Exit fullscreen mode

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 """
Enter fullscreen mode Exit fullscreen mode

7. Summary

Developing good Agent tools is like crafting a perfect toolbox:

  1. Proper tool classification - Sync/Async each has its use
  2. Standardized interfaces - Easy for unified management
  3. Protection mechanisms - Handle various exceptions
  4. Pursuit of efficiency - Cache when needed, rate limit when necessary
  5. 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)