A2A Protocol

A2A + CrewAI + OpenRouter Chart Generation Agent Tutorial

deutschEnglishespañolfrançaisहिंदी日本語한국어portuguêsрусский中文
MILO•
Share
A2A + CrewAI + OpenRouter Chart Generation Agent Tutorial

チュートリアルの焦点

このチュートリアルでは、以下のコアスキルの練習を通してガイドします：

  • OpenRouter + CrewAI + A2Aの統合: LLMプロバイダーとしてOpenRouter、エージェントフレームワークとしてCrewAI、標準化されたインターフェースとしてA2Aプロトコルを使用した完全なエンドツーエンドエージェント開発
  • A2Aエージェント画像データ返却の練習: エージェントにテキスト応答だけでなく画像データを生成・返却させる方法を学習
  • A2A Inspectorを使用したA2Aアプリケーションのデバッグ: エージェントアプリケーションをテスト・検証するためのプロフェッショナルなデバッグツールをマスター

クイックスタート

1. コードのクローン

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

2. 環境設定の作成

.envファイルを作成：

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

3. 環境セットアップと実行

# 仮想環境の作成
uv venv

# 仮想環境の有効化
source .venv/bin/activate

# アプリケーションの実行
uv run .

アプリケーションはhttp://localhost:10011で開始されます。

A2A Inspectorでのデバッグ

A2A Inspectorは、A2Aアプリケーションのデバッグ専用に設計された強力なツールです。

デバッグ手順：

  1. A2A Inspectorへのアクセス: https://inspector.a2aprotocol.aiを開く

  2. エージェントへの接続:

    • Inspectorにエージェントアドレスを入力: http://localhost:10011
    • "Connect"をクリックして接続を確立

  3. エージェント機能のテスト:

    • テストメッセージを送信: "Generate a chart of revenue: Jan,1000 Feb,2000 Mar,1500"
    • 完全なA2Aプロトコルインタラクションプロセスを観察
    • 返却された画像データを確認

  4. デバッグとモニタリング:

    • エージェントの能力とスキルを確認
    • 完全なリクエストとレスポンスフローをモニタリング
    • 正しい画像データ送信を検証

より詳細なデバッグガイドについてはA2A Inspectorドキュメンテーションを参照してください。

メインプロセスとコード紹介

システムアーキテクチャシーケンス図

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

コアコンポーネントの詳細

1. A2Aサーバーの初期化 (__main__.py)

# エージェントの能力とスキルを定義
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'],
)

# エージェントカードの作成
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],
)

キーポイント:

  • AgentCapabilitiesでサポートされるエージェント機能を定義（ここではストリーミング無効）
  • AgentSkillで特定のエージェントスキルと使用例を記述
  • AgentCardはA2Aプロトコルにおけるエージェントのアイデンティティ

2. CrewAIエージェントの実装 (agent.py)

class ChartGenerationAgent:
    def __init__(self):
        # 専門チャート生成エージェントの作成
        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],
        )

        # タスクの定義
        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,
        )

キーポイント:

  • CrewAIのAgentクラスがAIアシスタントの役割と能力を定義
  • Taskクラスが特定のタスク実行ロジックを記述
  • カスタムツールはtoolsパラメータを通してエージェントに統合

3. チャート生成ツール

@tool('ChartGenerationTool')
def generate_chart_tool(prompt: str, session_id: str) -> str:
    """Generates a bar chart image from CSV-like input using matplotlib."""
    
    # CSVデータの解析
    df = pd.read_csv(StringIO(prompt))
    df.columns = ['Category', 'Value']
    df['Value'] = pd.to_numeric(df['Value'], errors='coerce')
    
    # 棒グラフの生成
    fig, ax = plt.subplots()
    ax.bar(df['Category'], df['Value'])
    ax.set_xlabel('Category')
    ax.set_ylabel('Value')
    ax.set_title('Bar Chart')
    
    # PNGバイトとして保存
    buf = BytesIO()
    plt.savefig(buf, format='png')
    plt.close(fig)
    buf.seek(0)
    image_bytes = buf.read()
    
    # 画像のエンコードとキャッシュ
    data = Imagedata(
        bytes=base64.b64encode(image_bytes).decode('utf-8'),
        mime_type='image/png',
        name='generated_chart.png',
        id=uuid4().hex,
    )
    
    # キャッシュに画像を保存
    session_data = cache.get(session_id) or {}
    session_data[data.id] = data
    cache.set(session_id, session_data)
    
    return data.id

キーポイント:

  • @toolデコレータを使用して関数をCrewAIツールに変換
  • pandasでCSVデータを解析、matplotlibでチャートを生成
  • ネットワーク送信のために画像をbase64エンコーディングで保存
  • 複数ユーザーのデータ分離管理のためにセッションIDを使用

4. A2Aエグゼキューター (agent_executor.py)

class ChartGenerationAgentExecutor(AgentExecutor):
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # ユーザー入力の取得
        query = context.get_user_input()
        
        # CrewAIエージェントの呼び出し
        result = self.agent.invoke(query, context.context_id)
        
        # 生成された画像データの取得
        data = self.agent.get_image_data(
            session_id=context.context_id, 
            image_key=result.raw
        )
        
        if data and not data.error:
            # 画像バイトを含むファイル部分の作成
            parts = [
                Part(
                    root=FilePart(
                        file=FileWithBytes(
                            bytes=data.bytes,
                            mimeType=data.mime_type,
                            name=data.name,
                        )
                    )
                )
            ]
        else:
            # エラー時のテキストメッセージ返却
            parts = [Part(root=TextPart(text=data.error or 'Failed to generate chart image.'))]
        
        # 完了したタスクをイベントキューに追加
        event_queue.enqueue_event(
            completed_task(
                context.task_id,
                context.context_id,
                [new_artifact(parts, f'chart_{context.task_id}')],
                [context.message],
            )
        )

キーポイント:

  • AgentExecutorはA2Aプロトコル実行レイヤー
  • RequestContextを通してユーザーリクエストを取得
  • CrewAI応答をA2Aプロトコル形式に変換
  • ファイルタイプデータ（画像）の返却をサポート

5. キャッシュシステム (utils.py)

class InMemoryCache:
    """有効期限なしのシンプルなスレッドセーフインメモリキャッシュ。"""
    
    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

キーポイント:

  • スレッドセーフなインメモリキャッシュ実装
  • 生成された画像データの保存に使用
  • ユーザーデータの混乱を防ぐためのセッション分離をサポート

技術スタック概要

  • A2Aプロトコル: 標準化されたエージェント通信プロトコル
  • CrewAI: マルチエージェント協調フレームワーク
  • OpenRouter: LLM API集約サービス
  • Matplotlib: Pythonチャート生成ライブラリ
  • Pandas: データ処理ライブラリ
  • UV: モダンなPythonパッケージマネージャー

拡張提案

  1. より多くのチャートタイプをサポート: 円グラフ、線グラフ、散布図など
  2. データ検証の追加: より強力な入力データ検証とエラーハンドリング
  3. 永続キャッシュ: Redisやファイルシステムを使用した画像保存
  4. ストリーミングサポート: リアルタイムチャート生成進捗のサポート
  5. マルチモーダル入力: テキストのみの入力ではなくCSVファイルアップロードのサポート

このチュートリアルを通して、現代のAI技術スタックを使用した実用的なエージェント構築のコアスキルをマスターしました。このチャート生成エージェントは、より複雑なデータ分析アプリケーションの基盤として活用できます。

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.