A2A のトレース事情 〜親子エージェントの動きをLangfuseで可視化してみる〜

Transcript

1. 〜親子エージェントの動きをLangfuseで可視化してみる〜 2026/02/17 Bedrock Night 2026 ~AIエージェント"構築"元年スペシャル！~ A2A のトレース事情
2. 早速ですが... 2
3. A2A を実装したことがありますか？ 3
4. A2A のトレースをしたことがありますか？ 4
5. 本題に入る前に... 5
6. ご留意いただきたいこと 時間の関係上、説明できない部分がいくつかあります A2A / OpenTelemetryの前提知識・Langfuseの詳細について など – p.10〜16 もスキップ気味でお話しします – ローカルで検証したことについてお話しします AgentCore は登場しません – 各種ドキュメントを読み漁ったつもりですが、間違いなどありましたら ご指摘いただけますと幸いです 6
7. 本題に入ります 7
8. 今回のトレース対象 8
9. 本日はこちらをトレースする際に発生した 課題とその対応についてお話しします 9
10. 子エージェントの実装（A2A Server 側） # child_agent_strands.py agent = Agent( model=model, system_prompt=SYSTEM_PROMPT, tools=[mcp_client], name="Strands Agent専門エージェント", description="Strands Agentsフレームワークに…", ) a2a_server = A2AServer(agent=agent, port=9001, ...) app.mount("/", a2a_server.to_fastapi_app()) name と description を書くだけで Agent Card が自動生成される A2AServer が Agent を HTTP エンドポイントに変換 10
11. 親エージェントの実装（A2A Client 側） # parent_agent.py agent_urls = [ "http://localhost:9001", # Strands Agent専門エージェント "http://localhost:9002", # LangChain専門エージェント ] a2a_tool_provider = A2AClientToolProvider(known_agent_urls=agent_urls) agent = Agent( model=model, system_prompt=SYSTEM_PROMPT, tools=a2a_tool_provider.tools, # ← A2Aツールが渡される ) A2Aツール a2a_list_discovered_agents → 名刺を取りに行く – a2a_send_message → メッセージを送る – 11
12. トレースできるようにしてみる 12
13. トレースの実装差分 追加要素 役割 対象 telemetry.py （新規） OTel→Langfuse パイプライン構築 共通 StarletteInstrumentor A2Aサーバー側リクエストのトレース 子のみ HTTPXClientInstrumentor HTTP通信のトレース + traceparent伝播 親・子 trace_attributes セッションID・タグなどメタデータ付与 親・子 以下を追加することでトレースを実現可能 13
14. 親エージェントの追加内容 # ========== 追加部分 ========== from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor from telemetry import setup_telemetry setup_telemetry() # OTel→Langfuse 接続 HTTPXClientInstrumentor().instrument() # HTTP通信をトレース session_id = str(uuid.uuid4()) # =============================== agent = Agent( model=model, system_prompt=SYSTEM_PROMPT, tools=a2a_tool_provider.tools, # ========== 追加部分 ========== trace_attributes={ "session.id": session_id, "user.id": "demo-user", "langfuse.tags": ["質問回答エージェント", "Parent"], }, # =============================== ) 14
15. 子エージェントの追加内容 # ========== 追加部分（親と同じ） ========== setup_telemetry() HTTPXClientInstrumentor().instrument() # ========================================= agent = Agent( model=model, tools=[mcp_client], # ========== 追加部分 ========== trace_attributes={ "session.id": session_id, "langfuse.tags": ["Strands Agent専門", "Child"], }, # =============================== ) # ========== 追加部分（子エージェントのみ） ========== # FastAPIはStarletteベースなので同じインストルメンターが使用可能 StarletteInstrumentor.instrument_app(app) # ================================================== 15
16. telemetry.py（新規作成） def setup_telemetry() -> StrandsTelemetry: # OTLP エクスポート設定 os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = \ f"{langfuse_base_url}/api/public/otel" os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = \ f"Authorization=Basic {langfuse_auth}" # OpenTelemetry セットアップ strands_telemetry = StrandsTelemetry().setup_otlp_exporter() # W3C Trace Context Propagator設定（分散トレーシング用） # → traceparentヘッダーが自動付与され、トレースが統合 set_global_textmap(CompositePropagator([ TraceContextTextMapPropagator(), W3CBaggagePropagator(), ])) 16
17. ここでトレースを確認してみましょう（デモ） 17
18. うわっ...私のスパン、 多すぎ...？ 18
19. どうしてこうなった 19
20. どうしてこうなった A2A SDK の内部計装 @trace_class により全publicメソッドが自動スパン化 – 計装メソッド数: 67個 – HTTPXClientInstrumentor 分散トレーシングのために有効化 – すべてのHTTP通信がスパンになる – 親子エージェントの区別がつかない 親も子も同じスパン名を出す – 20
21. A2A SDK の内部計装 21
22. A2A SDK の内部計装 from strands.multiagent.a2a import A2AServer a2a_server = A2AServer( agent=agent, port=PORT ) A2A SDK とは A2A を実現しやすくするためのライブラリ – Strands の A2AServer は内部で a2a-sdk パッケージを使用 利用者は a2a-sdk の存在すら意識することはあまりない – 22
23. 私たちの実装は簡単だが... # a2a-sdk 内部コードを一部抜粋 (a2a/server/events/event_queue.py) @trace_class(kind=SpanKind.SERVER) class EventQueue: async def enqueue_event(self): # → スパン自動生成 async def dequeue_event(self): # → スパン自動生成 def task_done(self): # → スパン自動生成 def tap(self): # → スパン自動生成 async def close(self): # → スパン自動生成 async def clear_events(self): # → スパン自動生成 def is_closed(self): # → スパン自動生成 簡単に扱っているが裏ではすごいこと（？）になっている a2a-sdk は @trace_class で全publicメソッドを自動スパン化 計装されたメソッド数は 67個 にも及ぶ（付録に一覧を記載） – 23
24. HTTPXClientInstrumentor 24
25. HTTPXClientInstrumentor 分散トレーシングのために HTTPXClientInstrumentor を有効化 2Aクライアント / A2Aサーバー間を追うため – すべてのHTTP通信がスパンになる – 見たいもの POST /jsonrpc（A2A send_message） – GET /.well-known/agent.json（Agent Card取得） – 混ざるノイズ http send（HTTPX層） – httpcore.request（下位HTTP層） など – 25
26. 親子エージェントの区別がつかない 26
27. 親子エージェントの区別がつかない スパン名が不親切 スパン名が "GET" "POST" だけ – URL情報はスパンの属性に入っているが、名前だけでは判別不能 – 親エージェントによるスパンか子エージェントによるスパンかわかりにくい 特に、 execute_event_loop_cycle などは判別できないとなかなかカオスに見える – 27
28. ここでいったん再整理 28
29. いったん出た課題を再整理 A2A SDK の内部計装 @trace_class により全publicメソッドが自動スパン化 – 計装メソッド数は67個もある – HTTPXClientInstrumentor 分散トレーシングのために有効化 – すべてのHTTP通信がスパンになる – 親子エージェントの区別がつかない 親も子も同じスパン名を出す – 29
30. ここで集められたスパンは どうなるのかというと...？ 30
31. スパンが集約される OpenTelemetryの TracerProvider はプロセスにグローバルに1つ → プロセス内の すべてのスパンが1箇所に集約 される 31
32. 集約されているとどうなる？ JsonRpcTransport.send_message # 親がJSON-RPCでメッセージを送信 → JSONRPCHandler.on_message_send # 子がJSON-RPCリクエストを受信・バリデーション → DefaultRequestHandler.on_message_send # 実際のビジネスロジック実行を開始 → InMemoryQueueManager.create_or_tap # このタスク用のイベントキューを作成（または既存をタップ） → EventQueue.enqueue_event # エージェントの応答イベントをキューに投入 → EventQueue.dequeue_event # キューからイベントを1つ取り出す → EventConsumer.consume_one # 取り出したイベントを消費してレスポンス用データを組み立て → EventQueue.task_done # このイベントの処理完了をマーク → EventQueue.close # キューを閉鎖（もう新しいイベントは来ない） → EventQueue.clear_events # キュー内の残りイベントを破棄してクリーンアップ → helpers.create_task_obj # レスポンス用のTaskオブジェクトを新規作成（UUID生成） → helpers.append_artifact_to_task # エージェントの出力（Artifact）をTaskに追加 1回の send_message でこれだけ出る 32
33. さあどうする 33
34. さあどうする カスタム FilteringSpanProcessor を作る 34
35. さあどうする カスタム FilteringSpanProcessor を作る 35
36. フィルタリング設定例 # フィルタリング対象のスパン名パターン（A2A SDK内部処理） EXCLUDED_SPAN_PATTERNS = [ # A2A SDK EventQueue関連 "a2a.server.events.event_queue.EventQueue.enqueue_event", "a2a.server.events.event_queue.EventQueue.dequeue_event", ... ] # A2Aプロトコルで重要なHTTPスパン（除外せずにキャッチする） A2A_IMPORTANT_HTTP_PATTERNS = [ "/.well-known/agent", # Agent Card取得（GET /.well-known/agent.json） "/jsonrpc", # JSON-RPC エンドポイント（send_message等） "send_message", # A2A send_message "tasks/sendSubscribe", # A2A streaming ] # スパン名にエージェント名を追加する対象 RENAME_TARGET_SPANS = [ "execute_event_loop_cycle", "chat", ] 36
37. もう一度確認してみましょう（デモ） 37
38. おわりに 38
39. まとめ A2A のためにどのような通信が行われているかを調べる機会になって楽しかった！ A2A をトレースしてみた トレース自体は難しくないが、いくつかの課題があり対応が必要だった – 発生した課題 A2A SDK の内部計装 – HTTPXClientInstrumentor – 親子エージェントの区別がつかない – カスタム FilteringSpanProcessor を作ることで解決できていそう AgentCore に乗せた場合どうなるかも確認してみたい – 39
40. おまけ 今回の検証は Claude Code / Codex にとてもお世話になりました 既存ライブラリなどの実装を読み解かせる使い方 – 解説をさせつつ、不明点をとことん質問し、適宜ドキュメントを自分で読む – 以前であれば、コードとにらめっこしていたのでかなり快適になったと思う – 「なぜ？」を解消するハードルはかなり下がっていると感じる 「なぜ？」に少し目を向ける習慣をつけるとより面白くなると思う – みなさんも「なぜ？」をAIに投げかけてみてはいかがでしょうか？ 40
41. 参考リンク A2A Protocol / SDK Strands Agents Langfuse A2A Protocol 公式リポジトリ (google/A2A) Agent2Agent (A2A) - Strands Agents Traces - Strands Agents Langfuse Documentation OpenTelemetry Integration - Langfuse 41
42. 参考リンク OpenTelemetry W3C Trace Context Python | OpenTelemetry 公式ドキュメント opentelemetry.sdk.trace - SpanProcessor等のAPIリファレンス opentelemetry-instrumentation-httpx - PyPI opentelemetry-instrumentation-starlette - PyPI Trace Context - W3C 仕様 42
43. ご清聴ありがとうございました 質問・フィードバックお待ちしています 43
44. 付録　計装メソッド一覧 44
45. 計装メソッド一覧 (1/18) クラス メソッド 役割 EventQueue enqueue_event イベントをキューに追加し子キューにも伝播 EventQueue dequeue_event キューからイベントを1つ取り出す EventQueue task_done イベントの処理完了をマーク EventQueue tap 親キューのイベントを受信する子キューを作成 EventQueue close キューを閉鎖し子キューも閉鎖 EventQueue clear_events 未処理イベントをすべて破棄 EventQueue is_closed キューが閉鎖済みかどうかを返す 45
46. 計装メソッド一覧 (2/18) クラス メソッド 役割 EventConsumer consume_one キューから1イベントを取得 EventConsumer consume_all 全イベントを順次取得 EventConsumer agent_task_callback タスクの例外をキャッチして 伝播 InMemoryQueueManager add タスクIDに新しいキューを登 録 InMemoryQueueManager get タスクIDでキューを検索 46
47. 計装メソッド一覧 (3/18) クラス メソッド 役割 InMemoryQueueManager tap 子キューを作成 InMemoryQueueManager close キューを閉鎖・登録削除 InMemoryQueueManager create_or_tap 新規作成 or 既存をタップ JSONRPCHandler on_message_send メッセージ送信を処理 JSONRPCHandler on_message_send_stream ストリーミング応答を生成 47
48. 計装メソッド一覧 (4/18) クラス メソッド 役割 JSONRPCHandler on_cancel_task タスクをキャンセル JSONRPCHandler on_resubscribe_to_task 実行中タスクに再接続 JSONRPCHandler on_get_task タスク情報を取得 JSONRPCHandler get_push_notification_config プッシュ通知設定を取得 JSONRPCHandler set_push_notification_config プッシュ通知設定を登録 48
49. 計装メソッド一覧 (5/18) クラス メソッド 役割 JSONRPCHandler list_push_notification_config 通知設定一覧を取得 JSONRPCHandler delete_push_notification_config 通知設定を削除 JSONRPCHandler get_authenticated_extended_card 拡張Agent Cardを返却 49
50. 計装メソッド一覧 (6/18) クラス メソッド 役割 RESTHandler on_message_send REST経由でメッセージ送信 RESTHandler on_message_send_stream REST経由でストリーミング応答 RESTHandler on_cancel_task タスクをキャンセル RESTHandler on_resubscribe_to_task イベントストリームに再接続 RESTHandler on_get_task タスク情報を取得 50
51. 計装メソッド一覧 (7/18) クラス メソッド 役割 RESTHandler get_push_notification プッシュ通知設定を取得 RESTHandler set_push_notification プッシュ通知設定を登録 RESTHandler list_push_notifications 通知設定一覧（未実装） RESTHandler list_tasks タスク一覧（未実装） 51
52. 計装メソッド一覧 (8/18) クラス メソッド 役割 DefaultRequestHandler on_message_send 非ストリーミング送信 DefaultRequestHandler on_message_send_stream ストリーミング送信 DefaultRequestHandler on_get_task タスク情報取得 52
53. 計装メソッド一覧 (9/18) クラス メソッド 役割 DefaultRequestHandler on_cancel_task キャンセルを要求 DefaultRequestHandler on_resubscribe_to_task イベントキューに再接続 DefaultRequestHandler _run_event_stream execute()を実行 53
54. 計装メソッド一覧 (10/18) クラス メソッド 役割 DefaultRequestHandler _setup_message_execution 実行の共通 セットアッ プ DefaultRequestHandler on_set_task_push_notification_config 通知設定を 保存 DefaultRequestHandler on_get_task_push_notification_config 通知設定を 取得 54
55. 計装メソッド一覧 (11/18) クラス メソッド 役割 DefaultRequestHandler on_list_task_push_notification_config 通知設 定一覧 を取得 DefaultRequestHandler on_delete_task_push_notification_config 通知設 定を削 除 55
56. 計装メソッド一覧 (12/18) クラス メソッド 役割 JsonRpcTransport send_message JSON-RPC経由で非ストリーミン グ送信 JsonRpcTransport send_message_streaming JSON-RPC経由でストリーミング 送信 JsonRpcTransport get_task タスク情報を取得 JsonRpcTransport cancel_task タスクをキャンセル JsonRpcTransport set_task_callback プッシュ通知設定を設定 56
57. 計装メソッド一覧 (13/18) クラス メソッド 役割 JsonRpcTransport get_task_callback プッシュ通知設定を取得 JsonRpcTransport resubscribe SSEストリームに再接続 JsonRpcTransport get_card Agent Cardを取得 57
58. 計装メソッド一覧 (14/18) クラス メソッド 役割 RestTransport send_message REST経由で非ストリーミング送信 RestTransport send_message_streaming REST経由でストリーミング送信 RestTransport get_task タスク情報を取得 RestTransport cancel_task タスクをキャンセル RestTransport set_task_callback プッシュ通知設定を設定 58
59. 計装メソッド一覧 (15/18) クラス メソッド 役割 RestTransport get_task_callback プッシュ通知設定を取得 RestTransport resubscribe SSEストリームに再接続 RestTransport get_card Agent Cardを取得 59
60. 計装メソッド一覧 (16/18) クラス メソッド 役割 GrpcTransport send_message gRPC経由で非ストリーミング送信 GrpcTransport send_message_streaming gRPC経由でストリーミング送信 GrpcTransport get_task タスク情報を取得 GrpcTransport cancel_task タスクをキャンセル GrpcTransport set_task_callback プッシュ通知設定を設定 60
61. 計装メソッド一覧 (17/18) クラス メソッド 役割 GrpcTransport get_task_callback プッシュ通知設定を取得 GrpcTransport resubscribe タスク更新ストリームに再接続 GrpcTransport get_card Agent Cardを取得 61
62. 計装メソッド一覧 (18/18) クラス メソッド 役割 helpers create_task_obj Taskオブジェクトを新規作成 helpers append_artifact_to_task エージェントの出力をTaskに追加 62