Guide de développement d'outils d'agent : de la conception à l'optimisation-Tutoriel Python-php.cn

Agent Tool Development Guide: From Design to Optimization

1. Introduction

Imaginez que vous assemblez un robot majordome super intelligent (Agent). Ce robot a besoin de divers outils pour vous aider à accomplir des tâches, tout comme la poche 4D de Doraemon. Cet article vous apprendra comment créer ces outils puissants pour rendre votre majordome IA plus performant et efficace.

2. Deux modèles de conception d'outils de base

2.1 Outils synchrones : mode de réponse instantanée

Pensez à utiliser une machine à café en libre-service :

Insérez des pièces et appuyez sur le bouton "Americano"
Attendez quelques secondes
Le café coule à flot, prêt à boire

Il s'agit d'un modèle d'outil synchrone typique. L'agent appelle l'outil et attend des résultats immédiats - simple et rapide.

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

Copier après la connexion

Cas d'utilisation :

Requêtes rapides : météo, taux de change, calculs simples
Opérations simples : envoi de messages, commutation des commandes
Commentaires en temps réel : vérifications du code de vérification, demandes de solde

2.2 Outils asynchrones : mode de suivi des tâches

Imaginez commander de la nourriture via une application de livraison :

Après avoir passé une commande, l'APP vous donne un numéro de commande
Vous pouvez vérifier l'état de la commande à tout moment
L'APP vous avertit lorsque la livraison est terminée

C'est ainsi que fonctionnent les outils asynchrones, parfaits pour les tâches dont le traitement est plus long.

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

Copier après la connexion

Cas d'utilisation :

Opérations chronophages : traitement de fichiers volumineux, analyse de données
Tâches en plusieurs étapes : rendu vidéo, génération de rapports
Suivi des progrès nécessaire : formation du modèle, traitement par lots

3. Standardisation de l’interface des outils : établissement de spécifications universelles

Tout comme tous les appareils électriques suivent des normes de prise unifiées, nos interfaces d'outils doivent être standardisées. Cela garantit que tous les outils fonctionnent parfaitement avec l'agent.

3.1 Spécifications de description de l'outil

Imaginez écrire un manuel de produit, vous devez dire clairement aux utilisateurs :

Ce que fait l'outil
Quels paramètres sont nécessaires
Quels résultats seront renvoyés

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

Copier après la connexion

3.2 Classe de base unifiée

Tout comme tous les appareils électriques nécessitent des interrupteurs d'alimentation et des interfaces d'alimentation, tous les outils doivent suivre des spécifications de base :

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

Copier après la connexion

4. Gestion des erreurs : rendre les outils plus fiables

Tout comme les appareils électroménagers ont besoin d'une protection contre l'eau, les chocs et les surcharges, les outils ont besoin de mécanismes de protection complets.

4.1 Classification et traitement des erreurs

Imaginez gérer la livraison express :

Mauvaise adresse → Erreur de paramètre
Maintenance du système → Service temporairement indisponible
Courrier trop occupé → Besoin d'une limitation de débit et réessayez

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

Copier après la connexion

4.2 Mécanisme de nouvelle tentative

Comme organiser automatiquement une deuxième livraison lorsque la première tentative échoue :

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

Copier après la connexion

5. Optimisation des performances : rendre les outils plus efficaces

5.1 Mécanisme de mise en cache

Comme un dépanneur plaçant des articles populaires à des endroits bien en vue :

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

Copier après la connexion

5.2 Contrôle de concurrence

Comme le système de rendez-vous d'un hôpital, contrôlant le nombre de services simultanés :

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

Copier après la connexion

6. Tests et documentation : garantir la fiabilité des outils

6.1 Tests unitaires

Comme l'inspection qualité avant le lancement d'un nouveau produit :

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
        }

Copier après la connexion

6.2 Normes documentaires

C'est comme écrire un manuel produit détaillé et clair :

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)

Copier après la connexion

7. Résumé

Développer de bons outils d'agent, c'est comme créer une boîte à outils parfaite :

Classification appropriée des outils - Sync/Async chacun a son utilité
Interfaces standardisées - Facile pour une gestion unifiée
Mécanismes de protection - Gérer diverses exceptions
Poursuite de l'efficacité - Cache en cas de besoin, limite de débit si nécessaire
L'accent sur la qualité - Des tests approfondis, une documentation claire

Rappelez-vous : de bons outils peuvent rendre les agents deux fois plus efficaces, tandis que de mauvais outils limiteront les agents à chaque tour.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!