A2A Exemple : Travel Planner OpenRouter

Il s'agit d'une implémentation Python qui adhère au protocole A2A (Agent2Agent). Une démonstration d'assistant de voyage implémentée basée sur le SDK officiel a2a-python de Google et le SDK Python OpenAI. C'est un assistant de voyage conforme aux spécifications du modèle OpenAI, capable de vous fournir des services de planification de voyage.

Code Source

A2A Travel Planner OpenRouter

Architecture du Projet

Ce projet démontre comment construire un agent de planification de voyage interopérable en utilisant le protocole A2A, incluant les composants principaux suivants :

Travel Planner Agent : Logique principale de l'assistant de voyage basée sur une interface compatible OpenAI
Agent Executor : Adaptateur de protocole A2A qui fait le pont entre la logique de l'agent et le serveur A2A
A2A Server : Serveur conforme au protocole A2A fournissant une interface de communication inter-agents standardisée
Loop Client : Client de test pour interagir avec le serveur A2A

Diagramme de Séquence du Flux de Travail

sequenceDiagram
    participant Client
    participant A2AServer
    participant RequestHandler
    participant Executor as TravelPlannerAgentExecutor
    participant Agent as TravelPlannerAgent
    participant LLM as OpenAI-Compatible LLM

    Client->>A2AServer: Demander la Carte d'Agent
    A2AServer->>Client: Retourner la Carte d'Agent (compétences, capacités)

    Note over Client,A2AServer: L'utilisateur interroge la planification de voyage

    Client->>A2AServer: message/sendStream (requête de streaming)
    A2AServer->>RequestHandler: Router la requête de streaming
    RequestHandler->>Executor: execute(context, event_queue)
    
    Executor->>Agent: stream(query)
    Agent->>LLM: chat.completions.create(stream=True)
    
    loop Traitement de la réponse en streaming
        LLM-->>Agent: Retourner un fragment de contenu en streaming
        Agent-->>Executor: yield {'content': chunk, 'done': False}
        Executor-->>RequestHandler: TaskArtifactUpdateEvent
        RequestHandler-->>A2AServer: Pousser l'événement SSE
        A2AServer-->>Client: Mise à jour du contenu en streaming
    end
    
    LLM-->>Agent: Réponse finale terminée
    Agent-->>Executor: yield {'content': '', 'done': True}
    Executor-->>RequestHandler: TaskArtifactUpdateEvent final
    RequestHandler-->>A2AServer: Événement SSE final
    A2AServer-->>Client: Réponse en streaming terminée

Flux de Travail Principal

Récupération de la Carte d'Agent : Le client récupère d'abord la carte d'agent du serveur A2A pour comprendre les capacités et compétences de l'agent
Traitement de Requête en Streaming : Le client envoie une requête de message en streaming avec la requête utilisateur
Exécution de l'Agent : L'exécuteur d'agent traite la requête et appelle la logique principale de l'agent planificateur de voyage
Interaction LLM : L'agent mène une conversation en streaming avec un LLM compatible OpenAI
Réponse en Temps Réel : Diffuse les réponses au client en temps réel via Server-Sent Events (SSE)

Commencer

Configurez les variables d'environnement :

Copiez le fichier d'exemple et configurez vos identifiants API.

cp env.example .env

Éditez le fichier .env avec vos valeurs réelles :

# Requis : Votre clé API pour le service de modèle IA
API_KEY=your_actual_api_key_here

# Optionnel : Nom du modèle (par défaut : google/gemini-2.0-flash-001)
MODEL_NAME=google/gemini-2.0-flash-001

# Optionnel : URL de base pour le service API
BASE_URL=https://openrouter.ai/api/v1

Installez les dépendances et démarrez le serveur :

uv venv
source .venv/bin/activate

uv sync
uv run .

Exécutez le client de boucle dans un nouveau terminal :
```
source .venv/bin/activate
uv run loop_client.py
```

Configuration

L'application utilise des variables d'environnement pour la configuration :

API_KEY (requis) : Votre clé API pour le service de modèle IA
MODEL_NAME (optionnel) : Le nom du modèle à utiliser (par défaut : "google/gemini-2.0-flash-001")
BASE_URL (optionnel) : L'URL de base pour le service API (par défaut : "https://openrouter.ai/api/v1")

Caractéristiques Techniques

Implémentation Actuelle

✅ Conformité au Protocole A2A : Entièrement conforme à la spécification du protocole Agent2Agent
✅ Réponse en Streaming : Prend en charge la génération de contenu en streaming en temps réel
✅ Compatible OpenAI : Prend en charge toute interface API compatible OpenAI
✅ Conception Modulaire : Séparation claire entre la logique de l'agent et l'adaptation du protocole
✅ Configuration d'Environnement : Configuration flexible des variables d'environnement

Plans d'Amélioration Future

Amélioration de la Gestion d'État des Tâches

Basé sur l'exemple Google A2A LangGraph, prévoyant d'ajouter les fonctionnalités suivantes :

🔄 Gestion du Cycle de Vie des Tâches : Implémenter un suivi complet de l'état des tâches (soumis → en cours → terminé/échoué)
🔄 Support de Conversation Multi-tours : Ajouter l'état input_required pour prendre en charge des scénarios complexes de planification de voyage nécessitant une clarification de l'utilisateur
🔄 Persistance des Tâches : Implémenter la persistance de l'état des tâches pour les tâches de planification de longue durée
🔄 Gestion d'Erreur Améliorée : États d'erreur plus détaillés et mécanismes de récupération
🔄 Annulation de Tâches : Prendre en charge l'annulation des tâches en cours

Exemple de Gestion d'État

# Exemple d'implémentation future de gestion d'état
class TravelPlannerTaskManager:
    async def handle_complex_query(self, query: str, context: RequestContext):
        # Détecter si plus d'informations sont nécessaires
        if self.needs_clarification(query):
            return TaskStatus(
                state=TaskState.input_required,
                message="Plus d'informations nécessaires : Veuillez fournir une destination spécifique, des dates et une fourchette de budget"
            )
        
        # Exécuter une planification complexe en plusieurs étapes
        task_id = await self.create_long_running_task(query)
        return TaskStatus(
            state=TaskState.working,
            taskId=task_id,
            message="Création d'un plan de voyage détaillé..."
        )

Ajouts de Fonctionnalités Prévus

📋 Support de Données Structurées : Ajouter le support DataPart pour la collecte de préférences de voyage basée sur des formulaires
🖼️ Support Multimédia : Prendre en charge FilePart pour générer et traiter des images de voyage, cartes, etc.
🔍 Intégration d'Outils : Intégrer des API externes (météo, vols, hôtels, etc.) comme appels d'outils
🌐 Support Multi-langues : Étendre les capacités de planification de voyage multi-langues
📊 Métriques d'Analyse : Ajouter la collecte de temps d'exécution des tâches, métriques de taux de succès

Visitez A2A