OpenTelemetry (OTEL)을 사용하여 Weave에서 Google Agent Development Kit (ADK) 에이전트 및 툴 호출을 추적할 수 있습니다. ADK는 AI 에이전트를 개발하고 배포하기 위한 유연하고 모듈화된 프레임워크입니다. Gemini와 Google 에코시스템에 최적화되어 있지만, ADK는 모델 및 배포에 독립적입니다. 단순한 작업부터 복잡한 워크플로우에 이르기까지 에이전트 아키텍처를 생성, 배포 및 오케스트레이션하기 위한 툴을 제공합니다.
이 가이드에서는 OTEL을 사용하여 ADK 에이전트 및 툴 호출을 추적하고, 해당 트레이스를 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 추적 설정을 수행합니다.
Weave에서 OTEL 추적 설정
ADK에서 Weave로 트레이스를 보내려면 TracerProvider 및 OTLPSpanExporter로 OTEL을 구성해야 합니다. 인증 및 프로젝트 식별을 위한 올바른 엔드포인트와 HTTP 헤더로 exporter를 설정하세요.
API 키 및 프로젝트 정보와 같은 민감한 환경 변수는 환경 파일(예: .env)에 저장하고 os.environ을 사용하여 로드하는 것이 좋습니다. 이를 통해 자격 증명을 안전하게 유지하고 코드 베이스에 노출되지 않도록 할 수 있습니다.
필수 설정
- Endpoint:
https://trace.wandb.ai/otel/v1/traces. 전용 Weave 인스턴스를 사용하는 경우 URL은 다음 패턴을 따릅니다: {YOUR_WEAVE_HOST}/traces/otel/v1/traces
- Headers:
Authorization: W&B API 키를 사용한 기본 인증(Basic auth)project_id: W&B 엔티티/프로젝트 이름 (예: myteam/myproject)
ADK에서 Weave로 OTEL 트레이스 전송
다음 코드 조각은 ADK 애플리케이션에서 Weave로 OTEL 트레이스를 전송하기 위해 OTLP span exporter 및 tracer provider를 설정하는 방법을 보여줍니다.
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 엔티티/프로젝트 이름 (예: "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,
)
# tracer provider를 생성하고 exporter를 추가
tracer_provider = trace_sdk.TracerProvider()
tracer_provider.add_span_processor(SimpleSpanProcessor(exporter))
# ADK를 임포트하거나 사용하기 전에 글로벌 tracer provider 설정
trace.set_tracer_provider(tracer_provider)
OTEL로 ADK 에이전트 추적
tracer provider를 설정한 후 자동 추적 기능이 포함된 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:
"""두 숫자를 더하고 결과를 반환합니다.
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=(
"당신은 수학을 할 수 있는 유용한 비서입니다. "
"수학 문제가 주어지면 calculator 툴을 사용하여 해결하세요."
),
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="5 + 7은 무엇인가요?")]
),
):
if event.is_final_response() and event.content:
print(f"Final response: {event.content.parts[0].text.strip()}")
# 비동기 함수 실행
asyncio.run(run_agent())
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
# 여러 툴 정의
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 = 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"Final response: {event.content.parts[0].text.strip()}")
# 비동기 함수 실행
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():
# 두 개의 LLM 에이전트 생성
summarizer = LlmAgent(
name="Summarizer",
model="gemini-2.0-flash",
instruction="주어진 텍스트를 한 문장으로 요약하세요.",
description="텍스트를 한 문장으로 요약함",
output_key="summary" # 출력을 state['summary']에 저장
)
analyzer = LlmAgent(
name="Analyzer",
model="gemini-2.0-flash",
instruction="주어진 텍스트의 감정을 긍정, 부정 또는 중립으로 분석하세요. 분석할 텍스트: {summary}",
description="텍스트의 감정을 분석함",
output_key="sentiment" # 출력을 state['sentiment']에 저장
)
# 순차적 워크플로우 생성
workflow = SequentialAgent(
name="TextProcessor",
sub_agents=[summarizer, analyzer],
description="요약 후 감정 분석을 수행하는 시퀀스를 실행합니다.",
)
# 러너 설정
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="이 제품은 제 기대를 뛰어넘었습니다. 상자에서 꺼내자마자 완벽하게 작동했고, 설치에 대해 문의했을 때 고객 서비스도 훌륭했습니다.")]
),
):
if event.is_final_response() and event.content:
print(f"Final response: {event.content.parts[0].text.strip()}")
# 비동기 함수 실행
asyncio.run(run_workflow())
더 알아보기
Platform
W&B Models
W&B Weave
W&B Inference
W&B Training
Release Notes
