Outils d'agents IA et intégration externe
Ce chapitre présente comment les agents interagissent avec le monde extérieur.
La capacité d’invocation d’outils est la clé pour connecter les agents à l’environnement externe.
En intégrant divers outils, les agents peuvent exécuter du code, accéder à des API, manipuler des systèmes de fichiers, et plus encore.
Utilisation de l’ordinateur
L’utilisation de l’ordinateur permet aux agents d’utiliser les interfaces informatiques comme des humains.
Cela inclut les interfaces graphiques telles que les navigateurs et les applications de bureau.
C’est une étape importante vers la réalisation de l’intelligence artificielle générale (AGI).
Capacités principales
Analyse de captures d’écran: Comprendre le contenu à l’écran et identifier les éléments interactifs.
Reconnaissance des éléments d'interface: Reconnaître les éléments d'interface tels que les boutons, les zones de saisie et les menus.
Contrôle souris/clavier: Simuler des opérations humaines, en effectuant des actions telles que cliquer et taper.
Automatisation du navigateur: Navigation web, remplissage de formulaires, recherche, etc.
Principe de fonctionnement
Le flux de travail de Computer Use peut se résumer ainsi : observer, comprendre, décider et exécuter.
Première étape, capture d'écran. Capturez le contenu de l'écran ou de la fenêtre actuelle.
Deuxième étape, analyse visuelle. Utilisez un modèle de vision pour analyser la capture d'écran et identifier les éléments et états de l'interface.
Troisième étape, prise de décision. Déterminez l'action suivante en fonction de l'objectif de la tâche.
Quatrième étape, exécution. Effectuez des actions telles que des clics de souris et des saisies au clavier.
Répétez la boucle jusqu'à ce que la tâche soit terminée.
Implémentation du code
Implémentation de base de Computer Use Agent
"""
Implémentation de Computer Use Agent
Capable d'utiliser des interfaces informatiques comme un humain
"""
def __init__(self, vision_model, action_executor, planner):
# Modèle de vision : Analyser la capture d'écran
self.vision_model = vision_model
# Exécuteur d'actions : Effectuer des opérations de souris et de clavier
self.action_executor = action_executor
# Planificateur : Décider de l'action suivante
self.planner = planner
# Nombre maximal d'étapes d'exécution
self.max_steps = 100
def observe(self, screenshot):
"""
Analyser la capture d'écran, identifier les éléments interactifs
:param screenshot: Image de capture d'écran
:return: Liste des éléments d'interface
"""
# Utiliser un modèle de vision pour analyser la capture d'écran
analysis = self.vision_model.analyze(screenshot)
# Renvoyer les éléments d'interface reconnus
# Chaque élément inclut : type, emplacement, contenu, interactivité
return analysis.elements
def act(self, action):
"""
Exécuter l'action
:param action: Description de l'action, par ex., {"type": "click", "x": 100, "y": 200}
"""
return self.action_executor.execute(action)
def run(self, task):
"""
Exécuter la boucle de tâches
:param task: Description de la tâche
:return: Résultat de la tâche
"""
# Initialiser l'état de la tâche
self.planner.set_task(task)
for step in range(self.max_steps):
# Étape 1 : Capturer l'écran actuel
screenshot = self.get_screen()
# Étape 2 : Observer - Identifier les éléments de l'interface
elements = self.observe(screenshot)
# Étape 3 : Décider - Déterminer l'action suivante
action = self.planner.decide_action(elements)
# Vérifier si la tâche est terminée
if action.is_final:
return action.result
# Étape 4 : Exécuter l'action
self.act(action)
# Facultatif : attendre la mise à jour de l'interface
self.wait_for_update()
return "Limite maximale d'étapes atteinte"
class VisionModel:
"""Modèle de vision : Analyser la capture d'écran"""
def analyze(self, screenshot):
"""
Analyser la capture d'écran
Retourne la liste des éléments d'interface
"""
# Utiliser un modèle multimodal pour l'analyse
prompt = """
Analysez cette capture d'écran et identifiez tous les éléments interactifs de l'interface.
Y compris : boutons, champs de saisie, liens, menus, etc.
Pour chaque élément, veuillez fournir :
1. Type (bouton, champ de saisie, lien, etc.)
2. Emplacement (coordonnées de la boîte englobante)
3. Contenu (texte du bouton, texte indicatif du champ de saisie, etc.)
4. Interactivité (visibilité, état activé)
"""
result = self.vision_model.analyze_image(screenshot, prompt)
return ScreenAnalysisResult(elements=result.elements)
class ActionExecutor:
"""Exécuteur d'actions : effectuer des opérations de souris et de clavier"""
def execute(self, action):
"""
Exécuter l'action
:param action: Objet Action
"""
if action.type == "click":
self.mouse.click(action.x, action.y)
elif action.type == "type":
self.keyboard.type_text(action.text)
elif action.type == "scroll":
self.mouse.scroll(action.direction, action.amount)
elif action.type == "press":
self.keyboard.press_key(action.key)
return ActionResult(success=True)
class Planner:
"""Planificateur : décider de la prochaine action"""
def __init__(self, llm):
self.llm = llm
self.task = None
self.history = []
set_task self(task, task):
"""Définir la tâche actuelle"""
self.task = task
self.history = []
def decide_action(self, elements):
"""
Déterminer l'action suivante en fonction de l'état actuel de l'interface
"""
prompt = f"""
Tâche actuelle : {self.task}
Historique des actions exécutées :
{self.history}
Éléments interactifs sur l'interface actuelle :
{elements}
Veuillez décider de l'action suivante.
Si la tâche est terminée, retournez is_final=True.
Sinon, retournez l'action à exécuter (type, coordonnées, paramètres, etc.).
"""
response = self.llm.generate(prompt)
return Action.parse(response)
Scénarios d'application
Computer Use convient particulièrement aux scénarios suivants :
Automatisation web : comme remplir automatiquement des formulaires, extraire des données et effectuer des opérations web.
Opérations sur les applications de bureau : comme ouvrir des fichiers, modifier des documents et utiliser des logiciels.
Automatisation des tests : comme exécuter automatiquement des tests d'interface utilisateur et des tests de régression.
Remarque : Computer Use est encore en développement et présente des problèmes tels qu'une exécution lente et des erreurs occasionnelles. Pour les scénarios avec une API claire, appeler directement l'API est généralement plus efficace que Computer Use.
Explication détaillée du protocole MCP
MCP (Model Context Protocol) est un protocole standard ouvert.
Il permet aux modèles d'IA de se connecter en toute sécurité à des outils et des sources de données externes.
L'objectif de conception de MCP est de devenir l'« interface USB » du domaine de l'IA.
Principes de conception fondamentaux
Standardisation: format de protocole unifié, permettant à différents Agents et outils d'interopérer.
Sécurité: contrôle clair des permissions, les Agents ne peuvent accéder qu'aux ressources autorisées.
Extensibilité: Ajout facile de nouveaux outils et sources de données.
Composants de l'architecture
Hôte MCP: Environnement hôte pour exécuter des applications d'IA, telles que Claude Desktop, des assistants de codage IA, etc.
Client MCP: Un client qui maintient une connexion 1:1 avec le serveur MCP.
Serveur MCP: Programme serveur qui fournit des outils et des ressources.
Flux de communication
Première étape, établissement de la connexion. Le Client établit une connexion avec le Serveur et échange des informations sur les capacités.
Deuxième étape, découverte des outils. Le Client interroge le Serveur pour obtenir les outils et ressources disponibles.
Troisième étape, invocation d'outil. Le Client envoie une requête d'invocation d'outil, et le Serveur l'exécute et renvoie le résultat.
Quatrième étape, accès aux ressources. Le Client peut lire et écrire les ressources fournies par le Serveur.
Exemple d'implémentation de serveur MCP
Implémentation de serveur MCP
from mcp.types import Tool, Resource
import asyncio
# Créer une instance de serveur MCP
app = Server("filesystem")
# Définir les outils du système de fichiers
@app.list_tools()
async def list_tools():
"""Lister tous les outils disponibles"""
return [
Outil(
name="read_file",
description="Lire le contenu du fichier",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Chemin du fichier"
},
"encoding": {
"type": "string",
"description": "Encodage du fichier, par défaut utf-8",
"default": "utf-8"
}
},
"required": ["path"]
}
),
Outil(
name="write_file",
description="Écrire le contenu du fichier",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Chemin du fichier"
},
"content": {
"type": "string",
"description": "Contenu du fichier"
}
},
"required": ["path", "content"]
}
),
Outil(
name="list_directory",
description="Lister le contenu du répertoire",
inputSchema={
"type": "object",
"properties": {
"path": {
"type": "string",
"description": "Chemin du répertoire"
}
}
}
)
]
@app.call_tool()
async def call_tool(name, arguments):
"""Exécuter l'appel d'outil"""
if name == "read_file":
return await read_file(arguments["path"], arguments.get("encoding", "utf-8"))
elif name == "write_file":
return await write_file(arguments["path"], arguments["content"])
elif name == "list_directory":
return await list_directory(arguments["path"])
else:
raise ValueError(f"Outil inconnu : {name}")
async def read_file(path, encoding):
"""Lire le fichier"""
with open(path, "r", encoding=encoding) as f:
content = f.read()
return [{"type": "text", "text": content}]
async def write_file(path, content):
"""Écrire un fichier"""
with open(path, "w", encoding="utf-8") as f:
f.write(content)
return [{"type": "text", "text": "Fichier écrit avec succès"}]
async def list_directory(path):
"""Lister le répertoire"""
import os
entries = os.listdir(path)
return [{"type": "text", "text": "\n".join(entries)}]
# Lancer le serveur
if __name__ == "__main__":
import mcp.server.stdio
asyncio.run(mcp.server.stdio.serve(app))
Modèles d'intégration d'API
Il existe trois façons courantes pour un agent d'intégrer des API externes.
L'API REST est le style d'API Web le plus couramment utilisé.
GraphQL offre des capacités d'interrogation de données plus flexibles.
gRPC est adapté aux appels de procédure distante dans les scénarios à haute performance.
Intégration d'API REST
REST (Representational State Transfer) est un style de conception d'API Web.
Il utilise les méthodes HTTP (GET, POST, PUT, DELETE) pour les opérations.
Implémentation du client d'API REST
from typing import Dict, Any, Optional
class RESTAPIClient:
"""
Client d'API REST
Encapsule les requêtes HTTP et fournit une interface d'appel concise
"""
def __init__(self, base_url: str, headers: Optional[Dict] = None):
"""
Initialiser le client d'API
:param base_url: URL de base de l'API
:param headers: en-têtes de requête par défaut
"""
self.base_url = base_url.rstrip("/")
self.session = requests.Session()
# Définir les en-têtes de requête par défaut
default_headers = {
"Content-Type": "application/json",
"Accept": "application/json"
}
if headers:
default_headers.update(headers)
self.session.headers.update(default_headers)
def call(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
data: Optional[Dict] = None,
**kwargs
) -> Dict[Any, Any]:
"""
Envoyer une requête API
:param method: méthode HTTP (GET, POST, PUT, DELETE)
:param endpoint: point de terminaison de l'API
:param params: paramètres de requête d'URL
:param data: données du corps de la requête
:return: données de la réponse
"""
url = f"{self.base_url}/{endpoint.lstrip('/')}"
response = self.session.request(
method=method.upper(),
url=url,
params=params,
json=data,
**kwargs
)
# Vérifier le code d'état HTTP
response.raise_for_status()
# Analyser la réponse
if response.content:
return response.json()
return {}
def get(self, endpoint: str, params: Optional[Dict] = None):
"""Requête GET"""
return self.call("GET", endpoint, params=params)
def post(self, endpoint: str, data: Dict):
"""Requête POST"""
return self.call("POST", endpoint, data=data)
def put(self, endpoint: str, data: Dict):
"""Requête PUT"""
return self.call("PUT", endpoint, data=data)
def delete(self, endpoint: str, params: Optional[Dict] = None):
"""Requête DELETE"""
return self.call("DELETE", endpoint, params=params)
class APIAgent:
"""
Agent API
Appeler des API externes en langage naturel
"""
def __init__(self, api_client: RESTAPIClient):
self.api_client = api_client
self.tool_descriptions = self.define_tools()
def define_tools(self):
"""
Définir les outils disponibles pour l'Agent
Renvoyer les descriptions des outils afin que le LLM comprenne comment les appeler
"""
return [
{
"name": "get_weather",
"description": "Obtenir les informations météorologiques pour une ville spécifiée",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "Nom de la ville, par exemple, Beijing, Shanghai"
}
},
"required": ["city"]
}
},
{
"name": "get_forecast",
"description": "Obtenir les prévisions météo pour une ville spécifiée",
"parameters": {
"type": "object",
"properties": {
"city": {
"type": "string",
"description": "Nom de la ville"
},
"days": {
"type": "integer",
"description": "Jours de prévision, par défaut à 3",
"default": 3
}
},
"required": ["city"]
}
}
]
def execute_tool(self, tool_name: str, arguments: Dict):
"""
Exécuter l'appel d'outil
:param tool_name: nom de l'outil
:param arguments: arguments de l'outil
:return: résultat d'exécution
"""
if tool_name == "get_weather":
return self.api_client.get(f"/weather/{arguments['city']}")
elif tool_name == "get_forecast":
params = {"days": arguments.get("days", 3)}
return self.api_client.get(f"/forecast/{arguments['city']}", params=params)
else:
raise ValueError(f"Outil inconnu : {tool_name}")
Bonnes pratiques d'intégration d'API
Gestion des erreurs: Gérer avec élégance les erreurs réseau, les délais d'attente, l'indisponibilité du service, etc.
Mécanisme de réessai: Pour les erreurs transitoires, implémenter une nouvelle tentative avec backoff exponentiel.
Limitation de débit: Respecter les limites de débit de l'API pour éviter d'être banni.
Stratégie de mise en cache: Mettre en cache les données fréquemment demandées pour réduire les appels à l'API.
Considérations de sécurité: Ne pas coder en dur les informations sensibles telles que les clés d'API ; utiliser des variables d'environnement pour les gérer.
Résumé du chapitre
Ce chapitre présente les technologies fondamentales permettant aux Agents d'interagir avec le monde extérieur.
Utilisation de l'ordinateurPermet aux Agents de manipuler des interfaces GUI, adaptée aux scénarios nécessitant une interaction avec des applications graphiques.
Protocole MCPFournit un moyen standardisé d'intégrer des outils, servant d'« interface USB » de l'ère des agents.
Intégration APIEst la méthode la plus courante pour intégrer des systèmes externes, notamment REST, GraphQL, gRPC, etc.
Le choix de la méthode d'intégration appropriée dépend du scénario spécifique.
Pour les opérations d'interface graphique, Computer Use est une solution universelle.
Pour l'intégration standardisée d'outils, MCP est l'orientation future du développement.
Pour l'intégration d'API Web, REST API reste le choix le plus répandu.
Autres extensions