OpenTelemetry (OTEL)를 사용하면 Weave에서 Google Agent Development Kit (ADK) 에이전트 및 도구 Call을 트레이스할 수 있습니다. ADK는 AI 에이전트를 개발하고 배포하기 위한 유연한 모듈형 프레임워크입니다. Gemini와 Google 생태계에 최적화되어 있지만, ADK는 모델과 배포 환경에 독립적입니다. 단순한 작업부터 복잡한 워크플로까지 지원하는 에이전트 아키텍처를 생성, 배포, 오케스트레이션할 수 있는 도구를 제공합니다.
이 가이드는 ADK로 에이전트를 구축하면서 에이전트 추론, 도구 Call, 멀티 에이전트 워크플로 전반에 대한 엔드 투 엔드 관측성을 확보하려는 개발자를 위한 것입니다. OTEL을 사용해 ADK 에이전트 및 도구 Call을 트레이스하고, 해당 트레이스를 Weave에서 시각화하는 방법을 설명합니다. 필요한 의존성을 설치하고, 데이터를 Weave로 보내는 OTEL 트레이서를 설정하고, ADK 에이전트와 도구를 계측하는 방법을 알아봅니다. Weave에서 트레이스를 확인하면 에이전트 동작을 디버그하고, 성능을 모니터링하고, 데이터가 에이전트와 도구를 통해 어떻게 흐르는지 살펴볼 수 있습니다.
-
필요한 의존성을 설치합니다:
pip install google-adk opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
-
Google API 키를 환경 변수로 설정합니다:
export GOOGLE_API_KEY=[YOUR-API-KEY]
-
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 트레이스 보내기
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)
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())
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())
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())
