Au cœur d'AgentGPT réside son système d'outils d'IA flexible, permettant aux agents de choisir et d'utiliser diverses fonctionnalités en fonction des exigences de leur mission. Cet article explore les principes de conception, les mécanismes d'implémentation et les méthodes d'extension de ce système. Les développeurs y découvriront comment intégrer des outils d'IA personnalisés à AgentGPT, augmentant ainsi ses capacités pour répondre à des besoins métier spécifiques.
1. Introduction
Dans les architectures d'agents autonomes, un système d'outils robuste est essentiel pour l'exécution de tâches complexes. AgentGPT, grâce à une conception soignée de ses outils, permet à ses agents d'identifier et d'employer les ressources appropriées pour atteindre leurs objectifs. Cette approche accroît non seulement la souplesse du système, mais aussi son efficacité dans la résolution de problèmes.
Nous examinerons la conception sous-jacente du système, ses mécanismes internes et fournirons un guide détaillé pour son extension.
2. Architecture du Système d'Outils IA
2.1 Vue d'ensemble architecturale
Le système d'uotils d'AgentGPT est conçu de manière modulaire et se compose des éléments clés suivants :
2.2 Définition de l'interface d'outil
Chaque outil d'IA doit implémenter l'interface de base unique AbstractTool :
# app/core/tools/base_tool.py
from abc import ABC, abstractmethod
from typing import Any, Dict
from fastapi.responses import StreamingResponse
class AbstractTool(ABC):
designation: str = "" # Nom technique de l'outil
exposition_publique: str = "" # Description visible par l'utilisateur
schema_arguments: str = "" # Schéma des arguments requis
icone_url: str = "" # URL de l'icône de l'outil
def __init__(self, agent_model: Any, langue_cible: str):
self.agent_model = agent_model
self.langue_cible = langue_cible
@staticmethod
@abstractmethod
def est_disponible() -> bool:
"""Vérifie si l'outil est opérationnel et configuré."""
pass
@abstractmethod
async def executer(
self, objectif_global: str, tache_courante: str, entree_utilisateur: str,
*args: Any, **kwargs: Any
) -> StreamingResponse:
"""Déclenche la fonction principale de l'outil."""
pass
3. Examen des Outils Intégrés
3.1 Outil de Recherche Web (RechercheWeb)
L'outil RechercheWeb permet à l'agent de glaner des informations actuelles via un moteur de recherche :
# app/core/tools/web_search_tool.py
import asyncio
from typing import List, Dict, Any
from urllib.parse import quote
from fastapi.responses import StreamingResponse
from loguru import logger
from app.config import app_settings # Supposons que les paramètres sont ici
from app.core.tools.base_tool import AbstractTool
from app.core.utils.stream_helpers import text_to_stream, summarize_with_citations # Fonctions d'aide
class RechercheWeb(AbstractTool):
designation = (
"Permet de rechercher des informations récentes et publiques sur le web pour des requêtes simples."
)
exposition_publique = "Recherche des informations sur les événements actuels via le web."
schema_arguments = "La requête de recherche. Cette valeur est toujours renseignée."
icone_url = "/assets/icons/search_globe.png"
@staticmethod
def est_disponible() -> bool:
# Vérifie la présence d'une clé API pour le service de recherche
return app_settings.search_api_key is not None and app_settings.search_api_key != ""
async def executer(
self, objectif_global: str, tache_courante: str, entree_utilisateur: str, *args: Any, **kwargs: Any
) -> StreamingResponse:
try:
return await self._effectuer_recherche(objectif_global, tache_courante, entree_utilisateur)
except Exception as e:
logger.error(f"Erreur lors de l'appel au service de recherche : {e}, retour au raisonnement.")
# En cas d'échec, se rabattre sur l'outil de Raisonnement
from app.core.tools.reasoning_tool import Raisonnement # Importation locale pour éviter la dépendance circulaire
return await Raisonnement(self.agent_model, self.langue_cible).executer(
objectif_global, tache_courante, entree_utilisateur, *args, **kwargs
)
async def _effectuer_recherche(
self, objectif_global: str, tache_courante: str, entree_utilisateur: str
) -> StreamingResponse:
# Simulation d'appel à un service de recherche externe
reponse_api = await self._simuler_appel_api_recherche(entree_utilisateur)
fragments_cites: List[Dict[str, str]] = [] # Liste des résultats avec citations
# Traitement des résultats de recherche
if reponse_api.get("boite_reponse"):
contenu_boite = reponse_api["boite_reponse"]
if contenu_boite.get("reponse"):
fragments_cites.append({"text": contenu_boite["reponse"], "url": f"https://example.com/search?q={quote(entree_utilisateur)}"})
elif contenu_boite.get("extrait"):
fragments_cites.append({"text": contenu_boite["extrait"].replace("\n", " "), "url": f"https://example.com/search?q={quote(entree_utilisateur)}"})
for idx, item in enumerate(reponse_api.get("resultats_organiques", [])[:5]): # Limite à 5 résultats
if "snippet" in item and "link" in item:
fragments_cites.append({"text": item["snippet"], "url": item["link"]})
elif "description" in item and "url" in item:
fragments_cites.append({"text": item["description"], "url": item["url"]})
if not fragments_cites:
return text_to_stream("Aucun résultat pertinent trouvé via la recherche web.", final=True)
# Résume les résultats et les renvoie avec les sources
return summarize_with_citations(
self.agent_model, self.langue_cible, objectif_global, tache_courante, fragments_cites
)
async def _simuler_appel_api_recherche(self, query: str) -> Dict[str, Any]:
"""Fonction simulée pour un appel d'API de recherche externe."""
await asyncio.sleep(0.5) # Simule le délai réseau
# Données de réponse simulées
if "AgentGPT" in query:
return {
"boite_reponse": {"reponse": "AgentGPT est un cadre d'agents autonomes qui utilise des grands modèles de langage pour accomplir des tâches."},
"resultats_organiques": [
{"snippet": "Découvrez comment AgentGPT utilise des outils pour accomplir des objectifs.", "link": "https://agentgpt.com/features"},
{"snippet": "Tutoriel sur l'intégration d'outils personnalisés dans AgentGPT.", "link": "https://agentgpt.com/docs/custom_tools"}
]
}
else:
return {
"resultats_organiques": [
{"snippet": f"Information générale sur {query}.", "link": "https://example.com/info"},
{"snippet": f"Actualités récentes concernant {query}.", "link": "https://example.com/news"}
]
}
3.2 Outil de Raisonnement (Raisonnement)
L'outil Raisonnement est dédié aux tâches de déduction logique et de synthèse :
# app/core/tools/reasoning_tool.py
from typing import Any
from fastapi.responses import StreamingResponse
from app.core.tools.base_tool import AbstractTool
from app.core.utils.stream_helpers import generate_llm_summary # Fonction d'aide pour le résumé LLM
class Raisonnement(AbstractTool):
designation = "Utilise le modèle de langage pour analyser et raisonner sur la tâche de l'utilisateur."
exposition_publique = "Effectue un raisonnement sur la tâche donnée."
schema_arguments = "La demande de raisonnement ou le problème à analyser."
@staticmethod
def est_disponible() -> bool:
return True # L'outil de raisonnement est toujours disponible
async def executer(
self, objectif_global: str, tache_courante: str, entree_utilisateur: str, *args: Any, **kwargs: Any
) -> StreamingResponse:
# Utilise le modèle de langage pour synthétiser des informations
return generate_llm_summary(
modele_ia=self.agent_model,
langue_de_travail=self.langue_cible,
objectif=objectif_global,
tache=tache_courante,
entree_textuelle=entree_utilisateur,
)
3.3 Outil Wikipédia (WikipediaInfo)
L'outil WikipediaInfo permet d'interroger l'encyclopédie Wikipédia pour obtenir des informations factuelles :
# app/core/tools/wikipedia_tool.py
import wikipedia # Nécessite l'installation de 'wikipedia-api'
from typing import Any
from fastapi.responses import StreamingResponse
from app.core.tools.base_tool import AbstractTool
from app.core.utils.stream_helpers import text_to_stream
class WikipediaInfo(AbstractTool):
designation = "Recherche des informations sur des concepts, personnes, lieux, etc., sur Wikipédia."
exposition_publique = "Recherche d'informations encyclopédiques sur Wikipédia."
schema_arguments = "Le terme de recherche à utiliser sur Wikipédia."
icone_url = "/assets/icons/wikipedia.png"
@staticmethod
def est_disponible() -> bool:
return True # L'outil Wikipédia est toujours disponible
async def executer(
self, objectif_global: str, tache_courante: str, entree_utilisateur: str, *args: Any, **kwargs: Any
) -> StreamingResponse:
try:
# Recherche une page sur Wikipédia
page_wiki = wikipedia.page(entree_utilisateur, auto_suggest=True, redirect=True, summary=True)
# Retourne le résumé de la page
return text_to_stream(page_wiki.summary)
except wikipedia.exceptions.DisambiguationError as e:
# Gère les erreurs d'ambiguïté (plusieurs résultats possibles)
options_disponibles = ", ".join(e.options[:5])
return text_to_stream(f"Plusieurs options trouvées : {options_disponibles}")
except wikipedia.exceptions.PageError:
# Gère les cas où la page n'existe pas
return text_to_stream(f"Aucune page Wikipédia trouvée pour : {entree_utilisateur}")
4. Gestion et Enregistrement des Outils
4.1 Mécanisme d'Enregistrement
Tous les outils sont centralisés et enregistrés via le fichier tools_registry.py :
# app/core/tools/tools_registry.py
from typing import List, Type
from app.core.tools.base_tool import AbstractTool
from app.core.tools.web_search_tool import RechercheWeb
from app.core.tools.reasoning_tool import Raisonnement
from app.core.tools.wikipedia_tool import WikipediaInfo
# Supposons d'autres outils comme ImageGenerator, CodeExecutor, TaskConcluder, etc.
# from app.core.tools.image_gen_tool import ImageGenerator
# from app.core.tools.code_executor_tool import CodeExecutor
# from app.core.tools.task_concluder_tool import TaskConcluder
# Liste de toutes les classes d'outils disponibles dans le système
SYSTEM_TOOLS: List[Type[AbstractTool]] = [
RechercheWeb, # Outil de recherche sur le web
Raisonnement, # Outil de raisonnement et de synthèse
WikipediaInfo, # Outil de recherche Wikipédia
# ImageGenerator, # Génération d'images (si implémenté)
# TaskConcluder, # Outil pour finaliser une tâche
# CodeExecutor, # Exécuteur de code
]
def recuperer_outils_actifs_utilisateur(noms_outils_requis: List[str]) -> List[Type[AbstractTool]]:
"""
Filtre les outils disponibles en fonction des noms requis et de leur disponibilité.
Args:
noms_outils_requis: Liste des noms de classes d'outils demandés par l'utilisateur.
Returns:
Une liste d'objets AbstractTool prêts à l'emploi.
"""
return [
outil_cls for outil_cls in SYSTEM_TOOLS
if outil_cls.__name__ in noms_outils_requis and outil_cls.est_disponible()
]
def lister_tous_les_outils_disponibles() -> List[Type[AbstractTool]]:
"""Retourne tous les outils du système qui sont actuellement disponibles."""
return [outil_cls for outil_cls in SYSTEM_TOOLS if outil_cls.est_disponible()]
def obtenir_identifiant_outil(outil_instance: AbstractTool) -> str:
"""Récupère l'identifiant (nom de classe) d'une instance d'outil."""
return outil_instance.__class__.__name__
4.2 Mécanisme de Sélection d'Outils
Durant la phase d'analyse des tâches, l'agent identifie l'outil le plus approprié :
# app/services/agent_logic/llm_task_analyzer.py
from typing import List, Dict, Any
from app.core.tools.base_tool import AbstractTool
from app.core.tools.tools_registry import recuperer_outils_actifs_utilisateur, obtenir_identifiant_outil
from app.models.analysis import ActionAnalysis, AnalysisArguments # Modèles pour l'analyse
from app.prompts.analysis_prompts import analyze_task_template # Modèle de prompt
# Autres importations pour la gestion des tokens, erreurs LLM, etc.
async def evaluer_action_agent(
self, objectif_global: str, tache_courante: str, noms_outils_actifs: List[str]
) -> ActionAnalysis:
# Récupère les outils que l'utilisateur a activés
capacites_agent = recuperer_outils_actifs_utilisateur(noms_outils_actifs)
definitions_fonctions = [generer_schema_fonction_outil(outil_cls) for outil_cls in capacites_agent]
# Construit le prompt d'analyse pour le LLM
requete_analyse = analyze_task_template.format_prompt(
goal=objectif_global,
task=tache_courante,
language=self.config.language,
)
# (Implémentation de la gestion des tokens et de la gestion des erreurs LLM ici)
# self.token_manager.calculer_tokens_max(self.llm_instance, requete_analyse.to_string(), str(definitions_fonctions))
# Appelle le modèle LLM pour analyser la tâche et suggérer un outil
response_llm = await self.llm_instance.predict_messages(
messages=requete_analyse.to_messages(),
functions=definitions_fonctions,
# callbacks=self.callbacks_handler,
)
# Extrait l'appel de fonction suggéré par le LLM
invocation_data = response_llm.additional_kwargs.get("function_call", {})
arguments_json_str = invocation_data.get("arguments", "")
try:
# Parse les arguments de la fonction appelée
parser_pydantic = PydanticOutputParser(pydantic_object=AnalysisArguments)
args_analyse = parser_pydantic.parse(arguments_json_str)
return ActionAnalysis(
action_name=invocation_data.get("name", obtenir_identifiant_outil(self._obtenir_outil_par_defaut())),
**args_analyse.dict(),
)
except Exception as e: # Gérer les erreurs de parsing ou d'appel LLM
logger.warning(f"Échec de l'analyse LLM, retour à l'analyse par défaut: {e}")
return ActionAnalysis.get_default_analysis(tache_courante)
def generer_schema_fonction_outil(outil_cls: Type[AbstractTool]) -> Dict[str, Any]:
"""Crée le schéma de fonction au format attendu par les modèles LLM (ex: OpenAI)."""
return {
"name": outil_cls.__name__,
"description": outil_cls.designation,
"parameters": {
"type": "object",
"properties": {
"justification": {
"type": "string",
"description": "Raisonnement expliquant pourquoi cette fonction est appelée",
},
"argument": {
"type": "string",
"description": outil_cls.schema_arguments,
},
},
"required": ["justification", "argument"],
},
}
5. Guide de Développement d'Outils Personnalisés
5.1 Création d'une Classe d'Outil de Base
Pour développer un outil personnalisé, il faut d'abord hériter de la classe AbstractTool :
# app/core/tools/mon_outil_special.py
from typing import Any
from fastapi.responses import StreamingResponse
from loguru import logger
from app.core.tools.base_tool import AbstractTool
from app.core.utils.stream_helpers import text_to_stream
class MonOutilSpecial(AbstractTool):
"""Exemple d'outil personnalisé pour des tâches spécifiques."""
designation = "Un outil unique pour traiter des informations spécifiques ou interagir avec un système interne."
exposition_publique = "Exécute une fonction spéciale définie par l'utilisateur."
schema_arguments = "L'entrée textuelle nécessaire à l'outil pour son exécution."
icone_url = "/assets/icons/custom_gear.png" # Chemin vers l'icône de l'outil
@staticmethod
def est_disponible() -> bool:
"""
Vérifie la disponibilité de l'outil.
Peut dépendre de variables d'environnement, clés API, ou l'état du système.
"""
# Exemple: `app_settings.ma_cle_api_personnalisee`
return True # Par défaut, toujours disponible pour cet exemple
async def executer(
self, objectif_global: str, tache_courante: str, entree_utilisateur: str, *args: Any, **kwargs: Any
) -> StreamingResponse:
"""
Contient la logique d'exécution principale de l'outil.
Args:
- objectif_global: L'objectif général de l'utilisateur.
- tache_courante: La tâche spécifique à réaliser maintenant.
- entree_utilisateur: La chaîne d'entrée que l'outil doit traiter.
"""
try:
# Appel à la fonction métier de l'outil
resultat_traitement = await self._proceder_traitement_personnalise(entree_utilisateur)
# Retourne une réponse en streaming
return text_to_stream(resultat_traitement)
except Exception as e:
logger.error(f"Erreur lors de l'exécution de MonOutilSpecial : {e}")
# Stratégie de gestion des erreurs: renvoyer un message ou se rabattre sur un autre outil
return text_to_stream("Échec de l'outil spécial, veuillez réessayer.")
async def _proceder_traitement_personnalise(self, texte_entree: str) -> str:
"""
Implémentation de la logique spécifique de l'outil.
Peut impliquer des appels API externes, des requêtes BDD, manipulations de fichiers, etc.
"""
# Exemple: simple transformation de texte
texte_modifie = f"Résultat du traitement personnalisé : '{texte_entree.strip().upper()}'"
return texte_modifie
5.2 Enregistrement de l'Outil Personnalisé
Ajoutez le nouvel outil à la liste des outils système :
# app/core/tools/tools_registry.py
from .mon_outil_special import MonOutilSpecial # Import de l'outil personnalisé
# Mise à jour de la liste des outils du système
SYSTEM_TOOLS: List[Type[AbstractTool]] = [
RechercheWeb,
Raisonnement,
WikipediaInfo,
# Autres outils...
MonOutilSpecial, # Ajout de l'outil personnalisé ici
]
5.3 Affichage des Outils dans l'Interface Utilisateur
Le front end interroge une API pour obtenir la liste des outils disponibles et les présente à l'utilisateur :
// frontend/src/hooks/useAgentCapabilities.ts
import { z } from "zod"; // Bibliothèque de validation de schéma
import { useQuery, useQueryClient } from "@tanstack/react-query";
import { useAgentInterfaceStore } from "@/stores/agentInterfaceStore"; // Un magasin d'état global
// Schéma de validation pour un outil
const AgentCapabilitySchema = z.object({
nomUnique: z.string(),
descriptionDetaillee: z.string(),
couleurThematique: z.string(),
urlImage: z.string().optional(),
});
// Schéma de la réponse de l'API
const AgentCapabilitiesResponseSchema = z.object({
capacites: z.array(AgentCapabilitySchema),
});
export function useAgentCapabilities() {
const definirCapacites = useAgentInterfaceStore.use.setAvailableCapabilities();
const clientDeRequete = useQueryClient();
const queryResult = useQuery(["capacitesAgent"], () => chargerCapacitesAPI("/api/agent/capabilities"), {
onSuccess: (data) => {
// Met à jour les capacités actives après un chargement réussi
mettreAJourCapacitesActives(data);
},
});
function mettreAJourCapacitesActives(listeCapacites: Capabilite[]) {
// Sauvegarder dans le stockage local ou le magasin d'état
localStorage.setItem("agentCapabilities", JSON.stringify(listeCapacites));
definirCapacites(listeCapacites.filter((cap) => cap.estActive)); // Filtrer celles qui sont actives
}
const basculerActivationCapacite = (nomCapacite: string, active: boolean) => {
clientDeRequete.setQueriesData(["capacitesAgent"], (ancienneListe) => {
const data = (ancienneListe as Capabilite[]).map((cap) =>
cap.nomUnique === nomCapacite ? { ...cap, estActive: active } : cap
);
mettreAJourCapacitesActives(data);
return data;
});
};
return {
capacitesActives: queryResult.data ?? [],
basculerActivationCapacite,
estChargementReussi: queryResult.isSuccess,
};
}
// Type pour une capacité (à définir ailleurs, ex: types.ts)
interface Capabilite {
nomUnique: string;
descriptionDetaillee: string;
couleurThematique: string;
urlImage?: string;
estActive: boolean;
}
async function chargerCapacitesAPI(endpoint: string): Promise<capabilite> {
// Fonction simulée pour charger les capacités depuis l'API
const response = await fetch(endpoint);
const data = await response.json();
return AgentCapabilitiesResponseSchema.parse(data).capacites.map(c => ({...c, estActive: true}));
}
</capabilite>
6. Exemples d'Extension du Système d'Outils
6.1 Exemple : Outil de Prévisions Météo
Voici l'implémentation d'un outil permettant d'obtenir les prévisions météorologiques :
# app/core/tools/outil_meteo.py
import aiohttp # Pour les requêtes HTTP asynchrones
from typing import Any
from fastapi.responses import StreamingResponse
from loguru import logger
from app.config import app_settings
from app.core.tools.base_tool import AbstractTool
from app.core.utils.stream_helpers import text_to_stream
class OutilMeteo(AbstractTool):
"""Outil pour interroger les conditions météorologiques actuelles d'une ville."""
designation = "Permet de récupérer les conditions météorologiques actuelles pour une ville spécifiée."
exposition_publique = "Obtient les prévisions météo."
schema_arguments = "Le nom de la ville pour laquelle obtenir la météo."
icone_url = "/assets/icons/weather_cloud.png"
@staticmethod
def est_disponible() -> bool:
# S'assure qu'une clé API météo est configurée
return app_settings.weather_api_key is not None and app_settings.weather_api_key != ""
async def executer(
self, objectif_global: str, tache_courante: str, entree_utilisateur: str, *args: Any, **kwargs: Any
) -> StreamingResponse:
try:
informations_meteo = await self._demander_meteo_ville(entree_utilisateur)
return text_to_stream(informations_meteo)
except Exception as e:
logger.error(f"Échec de l'outil Météo pour '{entree_utilisateur}': {e}")
return text_to_stream("Impossible de récupérer les informations météorologiques pour le moment.")
async def _demander_meteo_ville(self, nom_ville: str) -> str:
"""Récupère les données météo depuis une API externe."""
cle_api = app_settings.weather_api_key
url_api = f"http://api.meteo-service.com/v1/current?city={nom_ville}&apiKey={cle_api}&units=metric&lang=fr"
async with aiohttp.ClientSession() as session:
async with session.get(url_api) as reponse_http:
if reponse_http.status == 200:
donnees_meteo = await reponse_http.json()
description_ciel = donnees_meteo['current']['condition']
temperature_deg = donnees_meteo['current']['temperature']
humidite_pct = donnees_meteo['current']['humidity']
return f"Météo actuelle à {nom_ville}: {description_ciel}, Température: {temperature_deg}°C, Humidité: {humidite_pct}%"
else:
raise Exception(f"Erreur d'API météo, code: {reponse_http.status}, message: {await reponse_http.text()}")
6.2 Exemple : Outil d'Interrogation de Base de Données
# app/core/tools/outil_base_donnees.py
from typing import Any
from fastapi.responses import StreamingResponse
from sqlalchemy import text as sql_text_query
from loguru import logger
from app.database.session import get_async_session # Fonction pour obtenir une session SQLAlchemy asynchrone
from app.core.tools.base_tool import AbstractTool
from app.core.utils.stream_helpers import text_to_stream
class OutilBaseDonnees(AbstractTool):
"""Outil pour exécuter des requêtes SQL et récupérer des données de la base de données."""
designation = "Permet d'interroger la base de données interne de l'application."
exposition_publique = "Exécute des requêtes SQL."
schema_arguments = "La requête SQL à exécuter."
icone_url = "/assets/icons/database_query.png"
@staticmethod
def est_disponible() -> bool:
# La disponibilité peut dépendre de droits d'accès spécifiques pour l'agent
return True # Supposons que l'accès est toujours autorisé pour cet exemple
async def executer(
self, objectif_global: str, tache_courante: str, entree_utilisateur: str, *args: Any, **kwargs: Any
) -> StreamingResponse:
try:
resultat_requete = await self._lancer_commande_sql(entree_utilisateur)
return text_to_stream(resultat_requete)
except Exception as e:
logger.error(f"Échec de l'exécution de OutilBaseDonnees: {e}")
return text_to_stream("La requête de base de données a échoué.")
async def _lancer_commande_sql(self, commande_sql: str) -> str:
"""Exécute une commande SQL et formatte les résultats."""
# AVERTISSEMENT: En production, une validation STRICTE des entrées est cruciale pour prévenir les injections SQL.
async with get_async_session() as session_bd:
query_result = await session_bd.execute(sql_text_query(commande_sql))
fetched_rows = query_result.fetchall()
# Formatage simple des résultats
if fetched_rows:
entetes = query_result.keys() # Récupère les noms de colonnes
lignes_formatees = [", ".join(map(str, row)) for row in fetched_rows]
return f"Résultats de la requête:\nEntêtes: {', '.join(entetes)}\n" + "\n".join(lignes_formatees)
else:
return "Aucun enregistrement trouvé pour la requête."
7. Bonnes Pratiques pour le Système d'Outils
7.1 Gestion des Erreurs et Mécanismes de Repli
Une bonne gestion des erreurs permet à l'agent de rester résilient :
# Extrait d'une méthode executer() d'un outil
from app.core.tools.reasoning_tool import Raisonnement # Importation du repli
async def executer(
self, objectif_global: str, tache_courante: str, entree_utilisateur: str, *args: Any, **kwargs: Any
) -> StreamingResponse:
try:
# Tente d'exécuter la fonction principale de l'outil
outcome = await self._realiser_action_principale(entree_utilisateur)
return text_to_stream(outcome)
except CustomToolError as e: # Gère une erreur spécifique à l'outil
logger.warning(f"Problème spécifique détecté par l'outil: {e}")
return text_to_stream("Une erreur spécifique est survenue, tentant une solution alternative.")
except Exception as e: # Capture toutes les autres exceptions
logger.error(f"Erreur inattendue lors de l'exécution de l'outil: {e}")
# Se rabat sur l'outil de Raisonnement en cas d'échec grave
return await Raisonnement(self.agent_model, self.langue_cible).executer(
objectif_global, tache_courante, "Impossible d'utiliser l'outil. Veuillez raisonner sur la situation.", *args, **kwargs
)
7.2 Optimisation des Performances
L'utilisation de la mise en cache peut grandement améliorer l'efficacité :
# Utilisation du cache pour éviter les appels API redondants
from functools import lru_cache
import datetime
class EfficientTool(AbstractTool):
_cache = {} # Un dictionnaire simple comme cache
async def _appeler_api_externe_mise_en_cache(self, parametre_cle: str) -> str:
"""Appelle une API externe avec mise en cache des résultats."""
if parametre_cle in self._cache and (datetime.datetime.now() - self._cache[parametre_cle]['timestamp']).seconds < 3600:
logger.info(f"Résultat pour '{parametre_cle}' récupéré du cache.")
return self._cache[parametre_cle]['data']
# Simule un appel API coûteux
await asyncio.sleep(1)
resultat_api = f"Données fraîchement obtenues pour '{parametre_cle}'."
self._cache[parametre_cle] = {
'data': resultat_api,
'timestamp': datetime.datetime.now()
}
logger.info(f"Résultat pour '{parametre_cle}' mis en cache.")
return resultat_api
# La méthode executer() de cet outil appellerait _appeler_api_externe_mise_en_cache
7.3 Considérations de Sécurité
La validation et le nettoyage des entrées sont primordiaux :
# Validation et nettoyage des entrées utilisateur
import re
def filtrer_entree_securisee(entree_raw: str) -> str:
"""
Nettoie une chaîne d'entrée pour prévenir les injections ou les caractères dangereux.
Ex: supprime les balises HTML potentielles, les caractères de contrôle, etc.
"""
# Retire les balises HTML et les caractères potentiellement dangereux
cleaned_input = re.sub(r'[<>"\'`;]', '', entree_raw)
# Limite la longueur pour éviter les attaques par déni de service (DoS)
return cleaned_input[:255].strip()
# Utilisation:
# cleaned_user_query = filtrer_entree_securisee(input_from_llm)
8. Conclusion
Le système d'outils d'IA d'AgentGPT offre une capacité d'extension remarquable grâce à :
- **Une interface standardisée** : Tous les outils adhèrent à une interface commune.
- **Un mécanisme d'enregistrement souple** : Permet l'ajout et la gestion dynamiques des outils.
- **Une sélection intelligente** : L'IA choisit l'outil approprié en fonction du contexte de la tâche.
- **Une robustesse face aux erreurs** : Intègre des mécanismes de traitement des exceptions et de repli.
- **Une facilité d'extension** : Un cadre clair pour le développement d'outils personnalisés.
9. Recommandations Pratiques
- **Découplage des responsabilités** : Chaque outil doit se concentrer sur une fonction unique.
- **Documentation exhaustive** : Des descriptions précises aident l'IA à comprendre et utiliser l'outil.
- **Gestion proactive des erreurs** : Prévoir la gestion de toutes les situations exceptionnelles.
- **Attention aux performances** : Optimiser l'exécution des outils pour ne pas ralentir l'agent.
- **Sécurité renforcée** : Valider rigoureusement toutes les entrées et interactions.
Références
- Dépôt GitHub d'AgentGPT
- Architecture des systèmes d'outils LangChain
- Fonctionnalités d'appel de fonctions OpenAI
- Documentation officielle FastAPI
- Principes de la programmation asynchrone en Python