Google Agent Development Kit (ADK) 에이전트와 도구 호출을 OpenTelemetry (OTEL)을 사용해 Weave에서 추적할 수 있습니다. ADK는 AI 에이전트를 개발하고 배포하기 위한 유연하고 모듈식 프레임워크입니다. Gemini 및 Google 생태계에 최적화되어 있지만, ADK는 모델과 배포 방식에 구애받지 않습니다. 단순한 작업부터 복잡한 워크플로에 이르기까지 에이전트 기반 아키텍처를 생성, 배포, 오케스트레이션하기 위한 도구를 제공합니다.
이 가이드는 OTEL을 사용해 ADK 에이전트와 도구 호출을 추적하고, 해당 트레이스를 Weave에서 시각화하는 방법을 설명합니다. 필수 종속성을 설치하고, OTEL 트레이서를 구성하여 데이터를 Weave로 전송하는 방법, 그리고 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을 설정해야 합니다. Exporter를 인증 및 프로젝트 식별을 위한 올바른 엔드포인트와 HTTP 헤더에 맞게 구성하세요.
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 인증project_id: W&B entity/project 이름(예: 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=(
"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())
플랫폼
W&B Models
W&B Weave
W&B Inference
W&B 트레이닝
릴리스 노트
