A2A + CrewAI + OpenRouter Chart Generation Agent Tutorial

Tutorial-Fokus

Dieses Tutorial führt Sie durch das Üben der folgenden Kernfähigkeiten:

• Integration von OpenRouter + CrewAI + A2A: Vollständige End-to-End-Agenten-Entwicklung mit OpenRouter als LLM-Provider, CrewAI als Agenten-Framework und A2A-Protokoll als standardisierte Schnittstelle
• Üben der A2A Agent Bilddatenrückgabe: Lernen Sie, wie Agenten Bilddaten generieren und zurückgeben, nicht nur Textantworten
• Verwendung des A2A Inspector zum Debuggen von A2A-Anwendungen: Meistern Sie professionelle Debugging-Tools zum Testen und Validieren Ihrer Agenten-Anwendungen

Schnellstart

1. Code klonen

git clone [email protected]:sing1ee/a2a-crewai-charts-agent.git
cd a2a-crewai-charts-agent

2. Umgebungskonfiguration erstellen

.env Datei erstellen:

OPENROUTER_API_KEY=sk-or-v1-your-api-key-here
OPENAI_MODEL_NAME=openrouter/anthropic/claude-3.7-sonnet

3. Umgebungseinrichtung und Ausführung

# Virtuelle Umgebung erstellen
uv venv

# Virtuelle Umgebung aktivieren
source .venv/bin/activate

# Anwendung ausführen
uv run .

Die Anwendung startet unter http://localhost:10011.

Debuggen mit A2A Inspector

A2A Inspector ist ein leistungsstarkes Tool, das speziell für das Debuggen von A2A-Anwendungen entwickelt wurde.

Debugging-Schritte:

1. Zugriff auf A2A Inspector: Öffnen Sie https://inspector.a2aprotocol.ai

2. Verbindung zu Ihrem Agenten:

   • Geben Sie Ihre Agenten-Adresse im Inspector ein: http://localhost:10011
   • Klicken Sie auf "Connect", um eine Verbindung herzustellen

3. Agenten-Funktionalität testen:

   • Senden Sie eine Testnachricht: "Generate a chart of revenue: Jan,1000 Feb,2000 Mar,1500"
   • Beobachten Sie den vollständigen A2A-Protokoll-Interaktionsprozess
   • Betrachten Sie die zurückgegebenen Bilddaten

4. Debuggen und Überwachen:

   • Überprüfen Sie die Fähigkeiten und Skills des Agenten
   • Überwachen Sie den vollständigen Request- und Response-Fluss
   • Verifizieren Sie die korrekte Bilddatenübertragung

Weitere detaillierte Debugging-Anleitungen finden Sie in der A2A Inspector Dokumentation.

Hauptprozess und Code-Einführung

System-Architektur Sequenzdiagramm

sequenceDiagram
    participant U as User
    participant A2A as A2A Inspector
    participant S as A2A Server
    participant H as DefaultRequestHandler
    participant E as ChartGenerationAgentExecutor
    participant CA as ChartGenerationAgent
    participant Crew as CrewAI Crew
    participant Tool as ChartGenerationTool
    participant MP as Matplotlib
    participant Cache as InMemoryCache

    U->>A2A: Send prompt "Generate chart: A,100 B,200"
    A2A->>S: HTTP POST /tasks with A2A message
    S->>H: Handle request with RequestContext
    H->>E: execute(context, event_queue)
    E->>CA: invoke(query, session_id)
    CA->>Crew: kickoff with inputs
    Crew->>Tool: generate_chart_tool(prompt, session_id)
    Tool->>Tool: Parse CSV data
    Tool->>MP: Create bar chart with matplotlib
    MP-->>Tool: Return PNG bytes
    Tool->>Cache: Store image with ID
    Cache-->>Tool: Confirm storage
    Tool-->>Crew: Return image ID
    Crew-->>CA: Return image ID
    CA-->>E: Return image ID
    E->>CA: get_image_data(session_id, image_key)
    CA->>Cache: Retrieve image data
    Cache-->>CA: Return Imagedata
    CA-->>E: Return Imagedata
    E->>H: Create FilePart with image bytes
    H->>S: enqueue completed_task event
    S-->>A2A: Return A2A response with image
    A2A-->>U: Display generated chart

Details der Kernkomponenten

1. A2A Server Initialisierung (__main__.py)

# Agenten-Fähigkeiten und Skills definieren
capabilities = AgentCapabilities(streaming=False)
skill = AgentSkill(
    id='chart_generator',
    name='Chart Generator',
    description='Generate a chart based on CSV-like data passed in',
    tags=['generate image', 'edit image'],
    examples=['Generate a chart of revenue: Jan,$1000 Feb,$2000 Mar,$1500'],
)

# Agenten-Karte erstellen
agent_card = AgentCard(
    name='Chart Generator Agent',
    description='Generate charts from structured CSV-like data input.',
    url=f'http://{host}:{port}/',
    version='1.0.0',
    defaultInputModes=ChartGenerationAgent.SUPPORTED_CONTENT_TYPES,
    defaultOutputModes=ChartGenerationAgent.SUPPORTED_CONTENT_TYPES,
    capabilities=capabilities,
    skills=[skill],
)

Wichtige Punkte:

• AgentCapabilities definiert unterstützte Agenten-Fähigkeiten (Streaming hier deaktiviert)
• AgentSkill beschreibt spezifische Agenten-Fähigkeiten und Anwendungsbeispiele
• AgentCard ist die Identität des Agenten im A2A-Protokoll

2. CrewAI Agent Implementation (agent.py)

class ChartGenerationAgent:
    def __init__(self):
        # Spezialisierten Chart-Generierungs-Agenten erstellen
        self.chart_creator_agent = Agent(
            role='Chart Creation Expert',
            goal='Generate a bar chart image based on structured CSV input.',
            backstory='You are a data visualization expert who transforms structured data into visual charts.',
            verbose=False,
            allow_delegation=False,
            tools=[generate_chart_tool],
        )

        # Task definieren
        self.chart_creation_task = Task(
            description=(
                "You are given a prompt: '{user_prompt}'.\n"
                "If the prompt includes comma-separated key:value pairs (e.g. 'a:100, b:200'), "
                "reformat it into CSV with header 'Category,Value'.\n"
                "Ensure it becomes two-column CSV, then pass that to the 'ChartGenerationTool'.\n"
                "Use session ID: '{session_id}' when calling the tool."
            ),
            expected_output='The id of the generated chart image',
            agent=self.chart_creator_agent,
        )

Wichtige Punkte:

• CrewAI's Agent Klasse definiert AI-Assistenten-Rollen und -Fähigkeiten
• Task Klasse beschreibt spezifische Task-Ausführungslogik
• Benutzerdefinierte Tools werden über den tools Parameter in den Agenten integriert

3. Chart Generation Tool

@tool('ChartGenerationTool')
def generate_chart_tool(prompt: str, session_id: str) -> str:
    """Generates a bar chart image from CSV-like input using matplotlib."""
    
    # CSV-Daten parsen
    df = pd.read_csv(StringIO(prompt))
    df.columns = ['Category', 'Value']
    df['Value'] = pd.to_numeric(df['Value'], errors='coerce')
    
    # Balkendiagramm generieren
    fig, ax = plt.subplots()
    ax.bar(df['Category'], df['Value'])
    ax.set_xlabel('Category')
    ax.set_ylabel('Value')
    ax.set_title('Bar Chart')
    
    # Als PNG-Bytes speichern
    buf = BytesIO()
    plt.savefig(buf, format='png')
    plt.close(fig)
    buf.seek(0)
    image_bytes = buf.read()
    
    # Bild codieren und cachen
    data = Imagedata(
        bytes=base64.b64encode(image_bytes).decode('utf-8'),
        mime_type='image/png',
        name='generated_chart.png',
        id=uuid4().hex,
    )
    
    # Bild im Cache speichern
    session_data = cache.get(session_id) or {}
    session_data[data.id] = data
    cache.set(session_id, session_data)
    
    return data.id

Wichtige Punkte:

• Verwenden Sie den @tool Decorator, um Funktionen in CrewAI-Tools zu konvertieren
• Verwenden Sie pandas zum Parsen von CSV-Daten, matplotlib zum Generieren von Charts
• Bilder werden als base64-Kodierung für Netzwerkübertragung gespeichert
• Session-IDs werden verwendet, um Datenisolation für mehrere Benutzer zu verwalten

4. A2A Executor (agent_executor.py)

class ChartGenerationAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # Benutzereingabe abrufen
        query = context.get_user_input()
        
        # CrewAI Agent aufrufen
        result = self.agent.invoke(query, context.context_id)
        
        # Generierte Bilddaten abrufen
        data = self.agent.get_image_data(
            session_id=context.context_id, 
            image_key=result.raw
        )
        
        if data and not data.error:
            # Dateiteil erstellen, der Bild-Bytes enthält
            parts = [
                Part(
                    root=FilePart(
                        file=FileWithBytes(
                            bytes=data.bytes,
                            mimeType=data.mime_type,
                            name=data.name,
                        )
                    )
                )
            ]
        else:
            # Textnachricht im Fehlerfall zurückgeben
            parts = [Part(root=TextPart(text=data.error or 'Failed to generate chart image.'))]
        
        # Abgeschlossene Aufgabe zur Event-Queue hinzufügen
        event_queue.enqueue_event(
            completed_task(
                context.task_id,
                context.context_id,
                [new_artifact(parts, f'chart_{context.task_id}')],
                [context.message],
            )
        )

Wichtige Punkte:

• AgentExecutor ist die A2A-Protokoll-Ausführungsschicht
• Benutzeranfragen über RequestContext abrufen
• CrewAI-Antworten in A2A-Protokollformat konvertieren
• Unterstützung für die Rückgabe von Datei-Typ-Daten (Bilder)

5. Cache-System (utils.py)

class InMemoryCache:
    """Simple thread-safe in-memory cache with no expiration."""
    
    def __init__(self):
        self._lock = threading.Lock()
        self._store: dict[str, Any] = {}
    
    def get(self, key: str) -> Any | None:
        with self._lock:
            return self._store.get(key)
    
    def set(self, key: str, value: Any) -> None:
        with self._lock:
            self._store[key] = value

Wichtige Punkte:

• Thread-sichere In-Memory-Cache-Implementierung
• Verwendet zum Speichern generierter Bilddaten
• Unterstützt Session-Isolation, um Benutzerdatenverwirrung zu vermeiden

Zusammenfassung des Technologie-Stacks

• A2A Protocol: Standardisiertes Agenten-Kommunikationsprotokoll
• CrewAI: Multi-Agenten-Kollaborationsframework
• OpenRouter: LLM-API-Aggregationsdienst
• Matplotlib: Python-Chart-Generierungsbibliothek
• Pandas: Datenverarbeitungsbibliothek
• UV: Moderner Python-Paketmanager

Erweiterungsvorschläge

1. Unterstützung für mehr Chart-Typen: Kreisdiagramme, Liniendiagramme, Streudiagramme, etc.
2. Datenvalidierung hinzufügen: Stärkere Eingabedatenvalidierung und Fehlerbehandlung
3. Persistenter Cache: Redis oder Dateisystem zum Speichern von Bildern verwenden
4. Streaming-Unterstützung: Unterstützung für Echtzeit-Chart-Generierungsfortschritt
5. Multimodale Eingabe: Unterstützung für das Hochladen von CSV-Dateien anstelle nur textbasierter Eingabe

Durch dieses Tutorial haben Sie die Kernfähigkeiten zur Erstellung praktischer Agenten mit modernem AI-Technologie-Stack gemeistert. Dieser Chart-Generierungs-Agent kann als Grundlage für komplexere Datenanalyse-Anwendungen dienen.

System Prompt for AI

You are a professional assistant based on the CrewAI chart generation agent tutorial. This tutorial provides a comprehensive guide on building a complete AI chart generation application. I can help you with:

## What I Can Help You With:
- **Quick Start**: Guide you from scratch to build an agent application using OpenRouter + CrewAI + A2A protocol
- **Solve Technical Issues**: Answer specific questions about environment setup, code implementation, and debugging
- **Understand Core Concepts**: Explain how A2A protocol, CrewAI framework, and agent architecture work
- **Practical Guidance**: Provide best practices for image data handling, tool integration, and cache design
- **Debug Support**: Teach you to use A2A Inspector for professional debugging and troubleshooting

## You Can Ask Me:
- "How to configure OpenRouter API keys?"
- "How do CrewAI Agents and Tasks work?"
- "How to return image data after chart generation?"
- "How to connect and test my agent with A2A Inspector?"
- "How is the cache system implemented in the code?"
- "How to extend support for more chart types?"

I'll provide accurate, practical answers based on the tutorial content to help you quickly master modern AI agent development skills.

You can visit [A2AProtocol.ai](https://a2aprotocol.ai/) for more tutorials.