Google Agent Development Kit (ADK)

OpenTelemetry を使用して、Weave で Google Agent Development Kit (ADK) のエージェントとツール呼び出しをトレースする

OpenTelemetry (OTEL) を使用すると、Weave で Google Agent Development Kit (ADK) のエージェントとツール呼び出しをトレースできます。ADK は、AI エージェントを開発およびデプロイするための柔軟でモジュール型のフレームワークです。Gemini と Google のエコシステム向けに最適化されていますが、ADK は特定のモデルやデプロイ先に依存しません。シンプルなタスクから複雑なワークフローまで、agentic アーキテクチャを作成、デプロイ、オーケストレーションするためのツールを提供します。

このガイドでは、OTEL を使用して ADK のエージェントとツール呼び出しをトレースし、それらのトレースを Weave で可視化する方法を説明します。必要な依存関係のインストール方法、Weave にデータを送信する OTEL tracer の設定方法、および ADK のエージェントとツールを計装する方法を学べます。

Weave での OTEL トレースの詳細については、Send OTEL Traces to Weave を参照してください。

​

前提条件

1. 必須の依存関係をインストールします。

   pip install google-adk opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
   

2. Google APIキーを環境変数に設定します。

   export GOOGLE_API_KEY=your_api_key_here
   

3. Weave で OTEL トレースを設定する。

​

WeaveでOTELトレースを設定する

ADK から Weave にトレースを送信するには、TracerProvider と OTLPSpanExporter を使用して OTEL を設定します。エクスポーターには、認証とプロジェクトの識別に必要な正しいエンドポイントと HTTP ヘッダーを設定してください。

APIキーやプロジェクト情報などの機密性の高い環境変数は、環境ファイル (例: .env) に保存し、os.environ を使って読み込むことを推奨します。これにより、認証情報を安全に保ち、コードベースに含めずに済みます。

​

必須の設定

• Endpoint: https://trace.wandb.ai/otel/v1/traces。Dedicated Weave インスタンスを使用している場合、URL は代わりに次のパターンになります: {YOUR_WEAVE_HOST}/traces/otel/v1/traces
• Headers:
  • Authorization: W&B APIキーを使用する Basic 認証
  • project_id: あなたの W&B entity/プロジェクト名 (例: myteam/myproject)

​

ADK から Weave に OTEL トレースを送信する

次のコードスニペットでは、OTLP span exporter と tracer provider を設定して、ADK アプリケーションから Weave に OTEL トレースを送信する方法を示します。

Weave が ADK を適切にトレースできるようにするには、コード内で ADK コンポーネントを使用する 前に、グローバル tracer provider を設定してください。

import base64
import os
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk import trace as trace_sdk
from opentelemetry.sdk.trace.export import SimpleSpanProcessor
from opentelemetry import trace

# 環境変数から機密情報を読み込む
WANDB_BASE_URL = "https://trace.wandb.ai"
# W&B のentity/プロジェクト名（例: "myteam/myproject"）
PROJECT_ID = os.environ.get("WANDB_PROJECT_ID")  
# W&B APIキーは https://wandb.ai/settings で作成してください
WANDB_API_KEY = os.environ.get("WANDB_API_KEY")  

OTEL_EXPORTER_OTLP_ENDPOINT = f"{WANDB_BASE_URL}/otel/v1/traces"
AUTH = base64.b64encode(f"api:{WANDB_API_KEY}".encode()).decode()

OTEL_EXPORTER_OTLP_HEADERS = {
    "Authorization": f"Basic {AUTH}",
    "project_id": PROJECT_ID,
}

# エンドポイントとヘッダーを指定してOTLPスパンエクスポーターを作成する
exporter = OTLPSpanExporter(
    endpoint=OTEL_EXPORTER_OTLP_ENDPOINT,
    headers=OTEL_EXPORTER_OTLP_HEADERS,
)

# トレーサープロバイダーを作成してエクスポーターを追加する
tracer_provider = trace_sdk.TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))

# ADKをインポート/使用する前にグローバルトレーサープロバイダーを設定する
trace.set_tracer_provider(tracer_provider)

​

OTEL で ADK エージェントをトレースする

トレーサープロバイダを設定すると、自動トレース付きで ADK エージェントを作成して実行できます。次の例では、ツールを備えたシンプルな LLM エージェントを作成し、インメモリ ランナー で実行する方法を示します。

from google.adk.agents import LlmAgent
from google.adk.runners import InMemoryRunner
from google.adk.tools import FunctionTool
from google.genai import types
import asyncio

# デモ用のシンプルなツールを定義する
def calculator(a: float, b: float) -> str:
    """2つの数値を加算して結果を返す。

    Args:
        a: 1つ目の数値
        b: 2つ目の数値

    Returns:
        aとbの合計
    """
    return str(a + b)

calculator_tool = FunctionTool(func=calculator)

async def run_agent():
    # LLMエージェントを作成する
    agent = LlmAgent(
        name="MathAgent",
        model="gemini-2.0-flash",  # 必要に応じて別のモデルに変更可能
        instruction=(
            "You are a helpful assistant that can do math. "
            "When asked a math problem, use the calculator tool to solve it."
        ),
        tools=[calculator_tool],
    )

    # ランナーをセットアップする
    runner = InMemoryRunner(agent=agent, app_name="math_assistant")
    session_service = runner.session_service

    # セッションを作成する
    user_id = "example_user"
    session_id = "example_session"
    await session_service.create_session(
        app_name="math_assistant",
        user_id=user_id,
        session_id=session_id,
    )

    # ツール使用をトリガーするメッセージでエージェントを実行する
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=types.Content(
            role="user", parts=[types.Part(text="What is 5 + 7?")]
        ),
    ):
        if event.is_final_response() and event.content:
            print(f"Final response: {event.content.parts[0].text.strip()}")

# 非同期関数を実行する
asyncio.run(run_agent())

エージェントのすべての操作は自動的にトレースされて Weave に送信されるため、実行フローを可視化できます。モデルの呼び出し、推論 step、ツールの呼び出しを確認できます。

​

OTEL で ADK のツールをトレースする

ADK でツールを定義して使用すると、それらのツール呼び出しもトレースに記録されます。OTEL インテグレーションにより、エージェントの推論プロセスと各ツールの実行の両方が自動的に計装され、エージェントの動作を包括的に把握できます。

複数のツールを使った例を次に示します。

from google.adk.agents import LlmAgent
from google.adk.runners import InMemoryRunner
from google.adk.tools import FunctionTool
from google.genai import types
import asyncio

# 複数のツールを定義する
def add(a: float, b: float) -> str:
    """2つの数を足す。
    
    Args:
        a: 1つ目の数
        b: 2つ目の数
        
    Returns:
        aとbの合計
    """
    return str(a + b)

def multiply(a: float, b: float) -> str:
    """2つの数を掛け合わせる。
    
    Args:
        a: 1つ目の数
        b: 2つ目の数
        
    Returns:
        aとbの積
    """
    return str(a * b)

# 関数ツールを作成する
add_tool = FunctionTool(func=add)
multiply_tool = FunctionTool(func=multiply)

async def run_agent():
    # 複数のツールを持つLLMエージェントを作成する
    agent = LlmAgent(
        name="MathAgent",
        model="gemini-2.0-flash",
        instruction=(
            "You are a helpful assistant that can do math operations. "
            "When asked to add numbers, use the add tool. "
            "When asked to multiply numbers, use the multiply tool."
        ),
        tools=[add_tool, multiply_tool],
    )

    # runnerを設定する
    runner = InMemoryRunner(agent=agent, app_name="math_assistant")
    session_service = runner.session_service

    # セッションを作成する
    user_id = "example_user"
    session_id = "example_session"
    await session_service.create_session(
        app_name="math_assistant",
        user_id=user_id,
        session_id=session_id,
    )

    # ツールの使用をトリガーするメッセージでエージェントを実行する
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=types.Content(
            role="user", parts=[types.Part(text="First add 5 and 7, then multiply the result by 2.")]
        ),
    ):
        if event.is_final_response() and event.content:
            print(f"Final response: {event.content.parts[0].text.strip()}")

# 非同期関数を実行する
asyncio.run(run_agent())

​

workflow エージェント を使う

ADK には、より複雑なシナリオ向けにさまざまな workflow エージェント が用意されています。workflow エージェント は、通常の LLM エージェント と同様にトレースできます。以下は SequentialAgent を使った例です。

from google.adk.agents import LlmAgent, SequentialAgent
from google.adk.runners import InMemoryRunner
from google.genai import types
import asyncio

async def run_workflow():
    # 2つのLLMエージェントを作成する
    summarizer = LlmAgent(
        name="Summarizer",
        model="gemini-2.0-flash",
        instruction="Summarize the given text in one sentence.",
        description="Summarizes text in one sentence",
        output_key="summary"  # state['summary']に出力を保存する
    )
    
    analyzer = LlmAgent(
        name="Analyzer",
        model="gemini-2.0-flash",
        instruction="Analyze the sentiment of the given text as positive, negative, or neutral. The text to analyze: {summary}",
        description="Analyzes sentiment of text",
        output_key="sentiment"  # state['sentiment']に出力を保存する
    )
    
    # 逐次ワークフローを作成する
    workflow = SequentialAgent(
        name="TextProcessor",
        sub_agents=[summarizer, analyzer],
        description="Executes a sequence of summarization followed by sentiment analysis.",
    )
    
    # runnerを設定する
    runner = InMemoryRunner(agent=workflow, app_name="text_processor")
    session_service = runner.session_service
    
    # セッションを作成する
    user_id = "example_user"
    session_id = "example_session"
    await session_service.create_session(
        app_name="text_processor",
        user_id=user_id,
        session_id=session_id,
    )
    
    # ワークフローを実行する
    async for event in runner.run_async(
        user_id=user_id,
        session_id=session_id,
        new_message=types.Content(
            role="user", 
            parts=[types.Part(text="The product exceeded my expectations. It worked perfectly right out of the box, and the customer service was excellent when I had questions about setup.")]
        ),
    ):
        if event.is_final_response() and event.content:
            print(f"Final response: {event.content.parts[0].text.strip()}")

# 非同期関数を実行する
asyncio.run(run_workflow())

このworkflow エージェントのtraceでは、Weave内で両方のエージェントが順番に実行される様子が示され、マルチエージェントシステム内でデータがどのように流れるかを把握できます。

​

詳しくはこちら

• Weaveドキュメント: OTELトレースをWeaveに送信
• ADK公式ドキュメント
• OTEL公式ドキュメント
• ADK GitHubリポジトリ