Google adk

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

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

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

前提条件

1. 必要な依存関係をインストールします：

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

2. あなたのGoogle API keyを環境変数として設定します：

   export GOOGLE_API_KEY=your_api_key_here
   

3. WeaveでOTELトレースを構成する。

WeaveでOTELトレースを構成する

ADKからWeaveにトレースを送信するには、OTELをTracerProviderとOTLPSpanExporterで構成します。エクスポーターを認証とプロジェクト識別のための正しいエンドポイントとHTTPヘッダーに設定します。

APIキーやプロジェクト情報などの機密性の高い環境変数は、環境ファイル（例：.env）に保存し、os.environを使用してロードすることをお勧めします。これにより、認証情報を安全に保ち、コードベースから分離できます。

必要な構成

• Endpoint: https://trace.wandb.ai/otel/v1/traces
• Headers:
  • Authorization：W&B APIキーを使用した基本認証
  • project_id：あなたのW&Bエンティティ/プロジェクト名（例：myteam/myproject）

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

以下のコードスニペットは、OTLPスパンエクスポーターとトレーサープロバイダーを構成して、ADKアプリケーションからWeaveにOTELトレースを送信する方法を示しています。

WeaveがADKを適切にトレースするようにするには、グローバルトレーサープロバイダーを前にコード内でADKコンポーネントを使用するように設定します。

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

# Load sensitive values from environment variables
WANDB_BASE_URL = "https://trace.wandb.ai"
# Your W&B entity/project name e.g. "myteam/myproject"
PROJECT_ID = os.environ.get("WANDB_PROJECT_ID")  
# Your W&B API key (found at https://wandb.ai/authorize)
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,
}

# Create the OTLP span exporter with endpoint and headers
exporter = OTLPSpanExporter(
    endpoint=OTEL_EXPORTER_OTLP_ENDPOINT,
    headers=OTEL_EXPORTER_OTLP_HEADERS,
)

# Create a tracer provider and add the exporter
tracer_provider = trace_sdk.TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))

# Set the global tracer provider BEFORE importing/using 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

# Define a simple tool for demonstration
def calculator(a: float, b: float) -> str:
    """Add two numbers and return the result.

    Args:
        a: First number
        b: Second number

    Returns:
        The sum of a and b
    """
    return str(a + b)

calculator_tool = FunctionTool(func=calculator)

async def run_agent():
    # Create an LLM agent
    agent = LlmAgent(
        name="MathAgent",
        model="gemini-2.0-flash",  # You can change this to another model if needed
        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],
    )

    # Set up runner
    runner = InMemoryRunner(agent=agent, app_name="math_assistant")
    session_service = runner.session_service

    # Create a session
    user_id = "example_user"
    session_id = "example_session"
    session_service.create_session(
        app_name="math_assistant",
        user_id=user_id,
        session_id=session_id,
    )

    # Run the agent with a message that should trigger tool use
    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()}")

# Run the async function
asyncio.run(run_agent())

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

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

# Define multiple tools
def add(a: float, b: float) -> str:
    """Add two numbers.
    
    Args:
        a: First number
        b: Second number
        
    Returns:
        The sum of a and b
    """
    return str(a + b)

def multiply(a: float, b: float) -> str:
    """Multiply two numbers.
    
    Args:
        a: First number
        b: Second number
        
    Returns:
        The product of a and b
    """
    return str(a * b)

# Create function tools
add_tool = FunctionTool(func=add)
multiply_tool = FunctionTool(func=multiply)

async def run_agent():
    # Create an LLM agent with multiple tools
    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],
    )

    # Set up runner
    runner = InMemoryRunner(agent=agent, app_name="math_assistant")
    session_service = runner.session_service

    # Create a session
    user_id = "example_user"
    session_id = "example_session"
    session_service.create_session(
        app_name="math_assistant",
        user_id=user_id,
        session_id=session_id,
    )

    # Run the agent with a message that should trigger tool use
    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()}")

# Run the async function
asyncio.run(run_agent())

ワークフローエージェントの操作

ADKはさまざまなworkflow agentsをより複雑なシナリオ向けに提供しています。ワークフローエージェントは通常の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():
    # Create two LLM agents
    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"  # Store output in 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"  # Store output in state['sentiment']
    )
    
    # Create a sequential workflow
    workflow = SequentialAgent(
        name="TextProcessor",
        sub_agents=[summarizer, analyzer],
        description="Executes a sequence of summarization followed by sentiment analysis.",
    )
    
    # Set up runner
    runner = InMemoryRunner(agent=workflow, app_name="text_processor")
    session_service = runner.session_service
    
    # Create a session
    user_id = "example_user"
    session_id = "example_session"
    session_service.create_session(
        app_name="text_processor",
        user_id=user_id,
        session_id=session_id,
    )
    
    # Run the workflow
    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()}")

# Run the async function
asyncio.run(run_workflow())

このワークフローエージェントトレースは、両方のエージェントの順次実行をWeaveに表示し、マルチエージェントシステムを通じてデータがどのように流れるかの可視性を提供します。

詳細情報

• Weaveドキュメント：Send OTEL traces to Weave
• 公式ADKドキュメント
• 公式OTELドキュメント
• ADK GitHubリポジトリ