A2A Beispiel: Travel Planner OpenRouter

Dies ist eine Python-Implementierung, die dem A2A (Agent2Agent) Protokoll entspricht. Eine Reiseassistent-Demo, die auf Googles offiziellem a2a-python SDK und OpenAI Python SDK basiert. Es ist ein Reiseassistent, der den OpenAI-Modellspezifikationen entspricht und Ihnen Reiseplanungsdienste anbieten kann.

Quellcode

A2A Travel Planner OpenRouter

Projektarchitektur

Dieses Projekt demonstriert, wie man einen interoperablen Reiseplanungsagenten mit dem A2A-Protokoll erstellt, einschließlich der folgenden Kernkomponenten:

Travel Planner Agent: Kern-Reiseassistent-Logik basierend auf OpenAI-kompatiblem Interface
Agent Executor: A2A-Protokoll-Adapter, der Agent-Logik mit A2A-Server verbindet
A2A Server: A2A-protokollkonformer Server, der standardisierte Inter-Agent-Kommunikationsschnittstelle bereitstellt
Loop Client: Test-Client für die Interaktion mit dem A2A-Server

Workflow-Sequenzdiagramm

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

    Client->>A2AServer: Agent-Karte anfordern
    A2AServer->>Client: Agent-Karte zurückgeben (Fähigkeiten, Kompetenzen)

    Note over Client,A2AServer: Benutzer fragt Reiseplanung ab

    Client->>A2AServer: message/sendStream (Streaming-Anfrage)
    A2AServer->>RequestHandler: Streaming-Anfrage weiterleiten
    RequestHandler->>Executor: execute(context, event_queue)
    
    Executor->>Agent: stream(query)
    Agent->>LLM: chat.completions.create(stream=True)
    
    loop Streaming-Antwort-Verarbeitung
        LLM-->>Agent: Streaming-Inhaltsfragment zurückgeben
        Agent-->>Executor: yield {'content': chunk, 'done': False}
        Executor-->>RequestHandler: TaskArtifactUpdateEvent
        RequestHandler-->>A2AServer: SSE-Event senden
        A2AServer-->>Client: Stream-Inhalt-Update
    end
    
    LLM-->>Agent: Finale Antwort abgeschlossen
    Agent-->>Executor: yield {'content': '', 'done': True}
    Executor-->>RequestHandler: Finales TaskArtifactUpdateEvent
    RequestHandler-->>A2AServer: Finales SSE-Event
    A2AServer-->>Client: Streaming-Antwort abgeschlossen

Haupt-Workflow

Agent-Karten-Abruf: Client ruft zuerst Agent-Karte vom A2A-Server ab, um Agent-Fähigkeiten und -Kompetenzen zu verstehen
Streaming-Anfrage-Verarbeitung: Client sendet Streaming-Nachrichtenanfrage mit Benutzerabfrage
Agent-Ausführung: Agent-Executor verarbeitet die Anfrage und ruft Reiseplanungsagent-Kernlogik auf
LLM-Interaktion: Agent führt Streaming-Gespräch mit OpenAI-kompatiblem LLM
Echtzeit-Antwort: Streamt Antworten in Echtzeit über Server-Sent Events (SSE) an Client

Erste Schritte

Umgebungsvariablen konfigurieren:

Kopieren Sie die Beispieldatei und konfigurieren Sie Ihre API-Anmeldedaten.

cp env.example .env

Bearbeiten Sie die .env-Datei mit Ihren tatsächlichen Werten:

# Erforderlich: Ihr API-Schlüssel für den KI-Modelldienst
API_KEY=your_actual_api_key_here

# Optional: Modellname (Standard: google/gemini-2.0-flash-001)
MODEL_NAME=google/gemini-2.0-flash-001

# Optional: Basis-URL für den API-Dienst
BASE_URL=https://openrouter.ai/api/v1

Abhängigkeiten installieren und Server starten:

uv venv
source .venv/bin/activate

uv sync
uv run .

Loop-Client in neuem Terminal ausführen:

source .venv/bin/activate
uv run loop_client.py

Konfiguration

Die Anwendung verwendet Umgebungsvariablen für die Konfiguration:

API_KEY (erforderlich): Ihr API-Schlüssel für den KI-Modelldienst
MODEL_NAME (optional): Der zu verwendende Modellname (Standard: "google/gemini-2.0-flash-001")
BASE_URL (optional): Die Basis-URL für den API-Dienst (Standard: "https://openrouter.ai/api/v1")

Technische Merkmale

Aktuelle Implementierung

✅ A2A-Protokoll-Konformität: Vollständig konform mit Agent2Agent-Protokollspezifikation
✅ Streaming-Antwort: Unterstützt Echtzeit-Streaming-Inhaltsgenerierung
✅ OpenAI-kompatibel: Unterstützt jede OpenAI-kompatible API-Schnittstelle
✅ Modulares Design: Klare Trennung zwischen Agent-Logik und Protokoll-Anpassung
✅ Umgebungskonfiguration: Flexible Umgebungsvariablen-Konfiguration

Zukünftige Verbesserungspläne

Task-Status-Management-Verbesserung

Basierend auf Google A2A LangGraph Beispiel, geplant folgende Features hinzuzufügen:

🔄 Task-Lebenszyklus-Management: Vollständige Task-Status-Verfolgung implementieren (eingereicht → arbeitend → abgeschlossen/fehlgeschlagen)
🔄 Multi-Turn-Gesprächsunterstützung: input_required-Status hinzufügen, um komplexe Reiseplanungsszenarien zu unterstützen, die Benutzerklärung erfordern
🔄 Task-Persistenz: Task-Status-Persistenz für langfristige Planungsaufgaben implementieren
🔄 Verbesserte Fehlerbehandlung: Detailliertere Fehlerzustände und Wiederherstellungsmechanismen
🔄 Task-Stornierung: Stornierung laufender Tasks unterstützen

Status-Management-Beispiel

# Zukünftiges Status-Management-Implementierungsbeispiel
class TravelPlannerTaskManager:
    async def handle_complex_query(self, query: str, context: RequestContext):
        # Erkennen, ob mehr Informationen benötigt werden
        if self.needs_clarification(query):
            return TaskStatus(
                state=TaskState.input_required,
                message="Mehr Informationen benötigt: Bitte geben Sie spezifisches Ziel, Daten und Budgetbereich an"
            )
        
        # Komplexe mehrstufige Planung ausführen
        task_id = await self.create_long_running_task(query)
        return TaskStatus(
            state=TaskState.working,
            taskId=task_id,
            message="Detaillierten Reiseplan erstellen..."
        )

Geplante Feature-Ergänzungen

📋 Strukturierte Datenunterstützung: DataPart-Unterstützung für formularbasierte Reisepräferenz-Sammlung hinzufügen
🖼️ Multimedia-Unterstützung: FilePart für Generierung und Verarbeitung von Reisebildern, Karten usw. unterstützen
🔍 Tool-Integration: Externe APIs (Wetter, Flüge, Hotels usw.) als Tool-Aufrufe integrieren
🌐 Mehrsprachige Unterstützung: Mehrsprachige Reiseplanungsfähigkeiten erweitern
📊 Analyse-Metriken: Sammlung von Task-Ausführungszeit, Erfolgsraten-Metriken hinzufügen

Besuchen Sie A2A