Google adk

추적할 수 있습니다 Google Agent Development Kit (ADK) 에이전트 및 도구 호출을 Weave에서 OpenTelemetry (OTEL)을 사용하여. ADK는 AI 에이전트를 개발하고 배포하기 위한 유연하고 모듈식 프레임워크입니다. Gemini와 Google 생태계에 최적화되어 있지만, ADK는 모델에 구애받지 않고 배포에 구애받지 않습니다. 간단한 작업부터 복잡한 워크플로우까지 에이전트 아키텍처를 생성, 배포 및 오케스트레이션하기 위한 도구를 제공합니다.

이 가이드는 OTEL을 사용하여 ADK 에이전트 및 도구 호출을 추적하고 Weave에서 해당 추적을 시각화하는 방법을 설명합니다. 필요한 종속성을 설치하고, Weave에 데이터를 보내기 위한 OTEL 트레이서를 구성하고, 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로 추적을 보내려면 TracerProvider와 OTLPSpanExporter로 OTEL을 구성하세요. 내보내기를 인증 및 프로젝트 식별을 위한 올바른 엔드포인트 및 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 추적 보내기

다음 코드 스니펫은 ADK 애플리케이션에서 Weave로 OTEL 추적을 보내기 위해 OTLP 스팬 내보내기 및 트레이서 제공자를 구성하는 방법을 보여줍니다.

Weave가 ADK를 제대로 추적하도록 하려면 전역 트레이서 제공자를 before 코드에서 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 저장소