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
            }
        }

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)
        }

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 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 Retry Mechanism

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)
        }

5. Performance Optimization: Making Tools More Efficient

5.1 Caching Mechanism

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"]
            }
        }

5.2 Concurrency Control

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

6. Testing and Documentation: Ensuring Tool Reliability

6.1 Unit Testing

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
        }

6.2 Documentation Standards

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)

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.

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!