OpenTelemetry (OTEL)를 사용하면 Weave에서 Google Agent Development Kit (ADK) 에이전트 및 도구 call을 트레이스할 수 있습니다. ADK는 AI 에이전트를 개발하고 배포하기 위한 유연한 모듈형 프레임워크입니다. Gemini와 Google 생태계에 최적화되어 있지만, ADK는 특정 모델이나 배포 환경에 구애받지 않습니다. 단순한 작업부터 복잡한 워크플로에 이르기까지, 다양한 에이전트 아키텍처를 생성, 배포, 오케스트레이션할 수 있는 도구를 제공합니다.
이 가이드에서는 OTEL을 사용해 ADK 에이전트 및 도구 call을 트레이스하고, 해당 트레이스를 Weave에서 시각화하는 방법을 설명합니다. 필요한 의존성을 설치하고, 데이터를 Weave로 전송하도록 OTEL tracer를 설정하고, ADK 에이전트와 도구를 계측하는 방법을 알아봅니다.
-
필요한 의존성을 설치합니다:
pip install google-adk opentelemetry-sdk opentelemetry-exporter-otlp-proto-http
-
Google API 키를 환경 변수로 설정합니다:
export GOOGLE_API_KEY=your_api_key_here
-
Weave에서 OTEL 트레이싱을 구성하세요.
ADK에서 Weave로 트레이스를 보내려면 TracerProvider와 OTLPSpanExporter로 OTEL을 설정하세요. 익스포터가 인증 및 프로젝트 식별에 필요한 올바른 엔드포인트와 HTTP 헤더를 사용하도록 설정하세요.
API 키와 프로젝트 정보 같은 민감한 환경 변수는 환경 파일(예: .env)에 저장하고 os.environ으로 불러오는 것을 권장합니다. 이렇게 하면 자격 증명을 안전하게 유지하고 코드베이스에 직접 포함하지 않을 수 있습니다.
- Endpoint:
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 컴포넌트를 사용하기 전에 전역 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")
# 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 스팬 익스포터 생성
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)
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:
"""두 숫자를 더하고 결과를 반환합니다.
Args:
a: 첫 번째 숫자
b: 두 번째 숫자
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())
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:
"""두 숫자를 더합니다.
Args:
a: 첫 번째 숫자
b: 두 번째 숫자
Returns:
a와 b의 합
"""
return str(a + b)
def multiply(a: float, b: float) -> str:
"""두 숫자를 곱합니다.
Args:
a: 첫 번째 숫자
b: 두 번째 숫자
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=(
"당신은 수학 연산을 수행할 수 있는 유용한 어시스턴트입니다. "
"숫자를 더하라는 요청을 받으면 add 도구를 사용하세요. "
"숫자를 곱하라는 요청을 받으면 multiply 도구를 사용하세요."
),
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="먼저 5와 7을 더한 다음, 그 결과에 2를 곱하세요.")]
),
):
if event.is_final_response() and event.content:
print(f"최종 응답: {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())
