Google Agent Development Kit (ADK)

OpenTelemetry를 사용해 Weave에서 Google Agent Development Kit (ADK) 에이전트 및 도구 Call을 트레이스하세요

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

이 가이드는 ADK로 에이전트를 구축하면서 에이전트 추론, 도구 Call, 멀티 에이전트 워크플로 전반에 대한 엔드 투 엔드 관측성을 확보하려는 개발자를 위한 것입니다. OTEL을 사용해 ADK 에이전트 및 도구 Call을 트레이스하고, 해당 트레이스를 Weave에서 시각화하는 방법을 설명합니다. 필요한 의존성을 설치하고, 데이터를 Weave로 보내는 OTEL 트레이서를 설정하고, ADK 에이전트와 도구를 계측하는 방법을 알아봅니다. Weave에서 트레이스를 확인하면 에이전트 동작을 디버그하고, 성능을 모니터링하고, 데이터가 에이전트와 도구를 통해 어떻게 흐르는지 살펴볼 수 있습니다.

Weave의 OTEL 트레이싱에 대한 자세한 내용은 OTEL 트레이스를 Weave로 보내기를 참조하세요.

​

사전 요구 사항

1. 필요한 의존성을 설치합니다:

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

2. Google API 키를 환경 변수로 설정합니다:

   export GOOGLE_API_KEY=[YOUR-API-KEY]
   

3. Weave에서 OTEL 트레이싱을 구성하세요.

​

Weave에서 OTEL 트레이싱 설정

ADK에서 Weave로 트레이스를 보내려면 TracerProvider와 OTLPSpanExporter로 OTEL을 설정하세요. 익스포터가 인증 및 프로젝트 식별에 필요한 올바른 엔드포인트와 HTTP 헤더를 사용하도록 설정하세요.

API 키와 프로젝트 정보 같은 민감한 환경 변수는 환경 파일(예를 들어, .env)에 저장하고 os.environ으로 불러오세요. 이렇게 하면 자격 증명을 안전하게 유지하고 코드베이스에 직접 포함하지 않을 수 있습니다.

​

필수 설정

• 엔드포인트: https://trace.wandb.ai/otel/v1/traces. 전용 Weave 인스턴스를 사용하는 경우 URL은 다음 패턴을 따릅니다: [YOUR-WEAVE-HOST]/traces/otel/v1/traces.
• 헤더:
  • Authorization: W&B API 키를 사용하는 Basic 인증.
  • project_id: W&B entity/프로젝트 이름(예를 들어, myteam/myproject).

​

ADK에서 Weave로 OTEL 트레이스 보내기

사전 요구 사항이 갖춰지면 span 데이터를 Weave로 전달하는 OTEL 익스포터와 트레이서 프로바이더를 설정할 수 있습니다. 다음 코드 스니펫은 ADK 애플리케이션에서 Weave로 OTEL 트레이스를 보내도록 OTLP span exporter와 트레이서 프로바이더를 설정하는 방법을 보여줍니다.

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

# 환경 변수에서 민감한 값 로드
WANDB_BASE_URL = "https://trace.wandb.ai"
# W&B entity/프로젝트 이름 (예: "myteam/myproject")
PROJECT_ID = os.environ.get("WANDB_PROJECT_ID")  
# https://wandb.ai/settings 에서 W&B API 키 생성
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 span exporter 생성
exporter = OTLPSpanExporter(
    endpoint=OTEL_EXPORTER_OTLP_ENDPOINT,
    headers=OTEL_EXPORTER_OTLP_HEADERS,
)

# 트레이서 프로바이더 생성 및 exporter 추가
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:
    """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():
    # 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로 전송되므로, 실행 흐름을 시각화할 수 있습니다. 모델 호출, 추론 단계, 그리고 도구 호출을 확인할 수 있습니다.

​

OTEL로 ADK 도구 트레이스하기

이 섹션에서는 에이전트가 둘 이상의 도구를 사용할 때 도구 호출이 트레이스에 어떻게 표시되는지 보여줍니다. ADK에서 도구를 정의하고 사용하면 이러한 도구 Call도 트레이스에 함께 캡처됩니다. 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:
    """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)

# 함수 도구 생성
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 = 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())

​

워크플로 에이전트 사용하기

단일 에이전트 설정을 넘어, ADK는 여러 에이전트를 워크플로로 구성할 수 있도록 지원합니다. 동일한 OTEL 설정으로 추가 설정 없이도 이러한 흐름을 트레이스할 수 있습니다. ADK는 더 복잡한 시나리오를 위해 다양한 워크플로 에이전트를 제공합니다. 일반적인 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():
    # 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 = 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())

이 워크플로 에이전트 트레이스는 Weave에서 두 에이전트가 순차적으로 실행되는 과정을 보여 주며, 데이터가 멀티 에이전트 시스템을 통해 어떻게 흐르는지 확인할 수 있게 해줍니다.

​

자세히 알아보기

• Weave 문서: OTEL 트레이스를 Weave로 전송
• 공식 ADK 문서
• 공식 OTEL 문서
• ADK GitHub 저장소