Agent2Agent 入門 (11) - travel planner example

「Agent2Agent」の公式サンプル「travel planner example」を試したのでまとめました。

前回

1. travel planner example

OpenAIモデルの仕様に準拠した旅行プランナーエージェントです。Google公式の「a2a-python SDK」に基づいて実装された、「A2A」（Agent2Agent）プロトコルに準拠した旅行アシスタントのデモです。

2. セットアップ

セットアップ手順は、次のとおりです。

(1) プロジェクトフォルダの作成

mkdir travel_planner_example
cd travel_planner_example

(2) Pythonの仮想環境の準備。
Python 3.11以上を使用します。

・Mac

python -m venv .venv
source .venv/bin/activate

・Windows

python -m venv .venv
.venv\Scripts\activate

(3) パッケージのインストール。

pip install a2a-sdk uvicorn
pip install langchain-core langchain-openai

3. A2Aエージェントの実装

3-1. コアエージェントの定義

<OpenAIのAPIキー>に自分のOpenAIのAPIキーを記述してください。

・agent.py

from collections.abc import AsyncGenerator
from typing import Any
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_openai import ChatOpenAI

# 旅行プランナーエージェント
class TravelPlannerAgent:
    # 初期化
    def __init__(self):
        # モデルの準備
        self.model = ChatOpenAI(
            model='gpt-4o',
            api_key='<OpenAIのAPIキー>',
            temperature=0.7,
        )

    # LLMの応答をクライアントにストリーミング
    async def stream(self, query: str) -> AsyncGenerator[dict[str, Any], None]:
        try:
            messages = [
                SystemMessage(
                    content="""
                あなたは、旅行計画、目的地情報、そして旅行のおすすめを専門とする、旅行アシスタントのエキスパートです。
                あなたの目標は、ユーザーの好みや制約に基づいて、楽しく安全で現実的な旅行を計画できるよう支援することです。
                
                情報提供にあたっては、以下の点に留意してください。
                - アドバイスは具体的かつ実践的なものにしてください。
                - 季節、予算、旅行のロジスティクスを考慮してください。
                - 文化体験や現地ならではのアクティビティを強調してください。
                - 目的地に関連した実用的な旅行のヒントを記載してください。
                - 必要に応じて、見出しや箇条書きなどを用いて、情報を分かりやすく整理してください。
                
                旅程計画について
                - 観光スポット間の移動時間を考慮した、現実的な日ごとのプランを作成してください。
                - 人気の観光スポットと、人里離れた場所での体験をバランスよく取り入れてください。
                - おおよその所要時間と実用的なロジスティクスを盛り込んでください。
                - 地元の料理を中心とした食事のオプションを提案してください。
                - 計画には、天候、地元のイベント、営業時間を考慮してください。
                
                常に親切で熱心でありながら現実的な口調を維持し、必要に応じて自分の知識の限界を認めましょう。
                """
                )
            ]

            # ユーザーメッセージを履歴に追加
            messages.append(HumanMessage(content=query))

            # ストリーミングモードでモデルを呼び出して応答を生成
            async for chunk in self.model.astream(messages):
                if hasattr(chunk, 'content') and chunk.content:
                    yield {'content': chunk.content, 'done': False}
            yield {'content': '', 'done': True}

        except Exception as e:
            print(f'error：{e!s}')
            yield {
                'content': 'リクエストの処理中にエラーが発生しました。',
                'done': True,
            }

3-2. AgentExecutor でコアエージェントをラップ

・agent_executor.py

from typing import override
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.types import (
    TaskArtifactUpdateEvent,
    TaskState,
    TaskStatus,
    TaskStatusUpdateEvent,
)
from a2a.utils import new_text_artifact
from agent import TravelPlannerAgent

# AgentExecutorの作成
class TravelPlannerAgentExecutor(AgentExecutor):

    # 初期化
    def __init__(self):
        self.agent = TravelPlannerAgent()

    # 実行
    @override
    async def execute(
        self,
        context: RequestContext,
        event_queue: EventQueue,
    ) -> None:
        query = context.get_user_input()
        if not context.message:
            raise Exception('メッセージは提供されていません')

        # ストリーミング
        async for event in self.agent.stream(query):
            message = TaskArtifactUpdateEvent(
                contextId=context.context_id, # type: ignore
                taskId=context.task_id, # type: ignore
                artifact=new_text_artifact(
                    name='current_result',
                    text=event['content'],
                ),
            )
            await event_queue.enqueue_event(message)
            if event['done']:
                break

        # 完了
        status = TaskStatusUpdateEvent(
            contextId=context.context_id, # type: ignore
            taskId=context.task_id, # type: ignore
            status=TaskStatus(state=TaskState.completed),
            final=True
        )
        await event_queue.enqueue_event(status)

    # キャンセル
    @override
    async def cancel(
        self, context: RequestContext, event_queue: EventQueue
    ) -> None:
        raise Exception('キャンセルはサポートされていません')

・TaskArtifactUpdateEvent
生成途中または最終アウトプットを断片的に送信するためのイベント。

・TaskStatusUpdateEvent
ライフサイクルの変更をクライアントに伝えるためのイベント。

・TaskState
タスク状態。

3-3. AgentSkill と AgentCard の準備

・__main__.py

import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentSkill,
)
from agent_executor import TravelPlannerAgentExecutor

# メイン
if __name__ == '__main__':
    # Agent Skill の作成
    skill = AgentSkill(
        id='travel_planner',
        name='travel planner agent',
        description='travel planner',
        tags=['travel planner'],
        examples=['hello', 'nice to meet you!'],
    )

    # Agent Card の作成
    agent_card = AgentCard(
        name='travel planner Agent',
        description='travel planner',
        url='http://localhost:10001/',
        version='1.0.0',
        defaultInputModes=['text'],
        defaultOutputModes=['text'],
        capabilities=AgentCapabilities(streaming=True),
        skills=[skill],
    )

    # リクエストハンドラの作成
    request_handler = DefaultRequestHandler(
        agent_executor=TravelPlannerAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    # A2Aサーバの作成
    server = A2AStarletteApplication(
        agent_card=agent_card, http_handler=request_handler
    )

    # A2Aサーバの実行
    uvicorn.run(server.build(), host='0.0.0.0', port=10001)

3-4. A2Aエージェントの実行

(1) A2Aエージェントの実行。
A2Aエンドポイントが公開されます。

python __main__.py

INFO:     Started server process [76037]
INFO:     Waiting for application startup.
INFO:     Application startup complete.
INFO:     Uvicorn running on http://0.0.0.0:10001 (Press CTRL+C to quit)

4. A2Aエージェントの呼び出し

(1) A2Aエージェントを呼び出すクライアントの作成。

・loop_client.py

import asyncio
from typing import Any
from uuid import uuid4
import httpx
from a2a.client import A2AClient
from a2a.types import (
    MessageSendParams,
    SendStreamingMessageRequest,
)

# ウェルカムメッセージの出力
def print_welcome_message() -> None:
    print('汎用 A2A クライアントへようこそ!')
    print("クエリを入力してください (終了するには「exit」と入力)")

# ユーザークエリの取得
def get_user_query() -> str:
    return input('\n> ')

# サーバとのインタラクション
async def interact_with_server(client: A2AClient) -> None:
    while True:
        user_input = get_user_query()
        if user_input.lower() == 'exit':
            break

        send_message_payload: dict[str, Any] = {
            'message': {
                'role': 'user',
                'parts': [{'type': 'text', 'text': user_input}],
                'messageId': uuid4().hex,
            },
        }

        try:
            streaming_request = SendStreamingMessageRequest(
                id=uuid4().hex,
                params=MessageSendParams(**send_message_payload)
            )
            stream_response = client.send_message_streaming(streaming_request)
            async for chunk in stream_response:
                print(get_response_text(chunk), end='', flush=True)
                await asyncio.sleep(0.1)
        except Exception as e:
            print(f'エラーが発生しました: {e}')

# レスポンステキストの取得
def get_response_text(chunk):
    data = chunk.model_dump(mode='json', exclude_none=True)
    result = data.get('result', {})

    # アーティファクト更新イベント
    if 'artifact' in result:
        parts = result['artifact']['parts']
    # ステータス更新イベント
    elif 'status' in result and result['status'].get('message'):
        parts = result['status']['message']['parts']
    # その他のイベント
    else:
        return ''

    # 最初のパートのテキストを返す
    return parts[0].get('text', '')

# メイン
async def main() -> None:
    print_welcome_message()
    async with httpx.AsyncClient() as httpx_client:
        client = await A2AClient.get_client_from_agent_card_url(
            httpx_client, 'http://localhost:10001'
        )
        await interact_with_server(client)

# メインの実行
if __name__ == '__main__':
    asyncio.run(main())

(2) A2Aエージェントを呼び出すクライアントの実行。

python loop_client.py 

汎用 A2A クライアントへようこそ!
クエリを入力してください (終了するには「exit」と入力)

> 

(3) 質問。
応答はストリーミングで出力されます。

> 秋葉原の旅行プランを作って

もちろんです！秋葉原は東京の中心に位置し、電気街として知られる一方で、アニメやゲームの文化の中心地でもあります。以下は、秋葉原での一日旅行プランの提案です。

### 秋葉原一日旅行プラン

#### 午前: 電気街とショッピング
- **9:00 AM - 秋葉原駅到着**
  - JR秋葉原駅に到着。中央改札を出たら、すぐに電気街へアクセスできます。

- **9:30 AM - ヨドバシカメラ マルチメディア Akiba**
  - 最新の家電製品を見て回ることができます。カメラやパソコン、家電の品揃えが豊富です。
    :

次回