Phoenix를 사용한 에이전트 관측성(Observability)¶
ADK에서 지원Python
Arize Phoenix는 로컬 개발, 오픈소스 워크플로우, 셀프 호스팅 트레이싱을 위한 Arize AI의 오픈소스 관측성 및 평가 플랫폼입니다. Google ADK 애플리케이션을 위한 포괄적인 추적 및 평가 기능을 제공합니다. 시작하려면 무료 계정에 가입하세요.
AI 네이티브 팀과 엔터프라이즈를 위해 구축된 전체 기능을 갖춘 프로덕션 플랫폼의 경우, 관리형 클라우드 또는 엔터프라이즈 셀프 호스팅 배포로 제공되는 Arize AX ADK 연동을 사용하세요. Arize의 에이전트 평가 가이드와 LLM 평가 가이드에서는 트레이스가 에이전트 및 LLM 애플리케이션의 평가 워크플로우를 지원하는 방법을 보여줍니다.
개요¶
Phoenix는 OpenInference 계측을 사용하여 Google ADK에서 트레이스를 자동으로 수집할 수 있으며, 이를 통해 다음을 수행할 수 있습니다.
- 에이전트 상호작용 추적 - 모든 에이전트 실행, 툴 호출, 모델 요청 및 응답을 전체 컨텍스트 및 메타데이터와 함께 자동으로 캡처
- 성능 평가 - 사용자 지정 또는 사전 빌드된 평가기를 사용하여 에이전트 동작을 평가하고, 에이전트 구성을 테스트하기 위한 실험 실행
- 문제 디버깅 - 상세한 트레이스를 분석하여 병목 현상, 실패한 툴 호출 및 예기치 않은 에이전트 동작을 신속하게 식별
- 셀프 호스팅 제어 - 자체 인프라에 데이터를 보관
설치¶
1. 필요한 패키지 설치¶
설정¶
1. Phoenix 실행¶
이 지침은 Phoenix Cloud를 사용하는 방법을 보여줍니다. 노트북, 터미널 또는 컨테이너를 사용한 셀프 호스팅을 통해 Phoenix를 실행할 수도 있습니다.
먼저, 무료 Phoenix 계정에 가입하세요.
Phoenix 엔드포인트와 API 키를 설정하세요:
import os
# 추적을 위한 Phoenix API 키 추가
PHOENIX_API_KEY = "YOUR_API_KEY_HERE"
os.environ["PHOENIX_CLIENT_HEADERS"] = f"api_key={PHOENIX_API_KEY}"
os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "https://app.phoenix.arize.com"
Phoenix API 키는 대시보드의 Keys 섹션에서 찾을 수 있습니다.
2. 애플리케이션을 Phoenix에 연결¶
from phoenix.otel import register
# Phoenix 트레이서 구성
tracer_provider = register(
project_name="my-llm-app", # 기본값은 'default'입니다
auto_instrument=True # 설치된 OI 의존성에 기반하여 앱을 자동으로 계측합니다
)
관찰¶
이제 추적 설정이 완료되었으므로, 모든 Google ADK SDK 요청은 관측성 및 평가를 위해 Phoenix로 스트리밍됩니다.
import asyncio
import nest_asyncio
nest_asyncio.apply()
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types
# 툴 함수 정의
def get_weather(city: str) -> dict:
"""지정된 도시의 현재 날씨 보고서를 검색합니다.
매개변수:
city (str): 날씨 보고서를 검색할 도시의 이름.
반환값:
dict: 상태와 결과 또는 오류 메시지.
"""
if city.lower() == "new york":
return {
"status": "success",
"report": (
"뉴욕의 날씨는 맑고 기온은 섭씨 25도(화씨 77도)입니다."
),
}
else:
return {
"status": "error",
"error_message": f"'{city}'의 날씨 정보는 제공되지 않습니다.",
}
# 툴을 사용하여 에이전트 생성
agent = Agent(
name="weather_agent",
model="gemini-flash-latest",
description="날씨 툴을 사용하여 질문에 답변하는 에이전트.",
instruction="답을 찾으려면 사용 가능한 툴을 사용해야 합니다.",
tools=[get_weather]
)
app_name = "weather_app"
user_id = "test_user"
session_id = "test_session"
runner = InMemoryRunner(agent=agent, app_name=app_name)
session_service = runner.session_service
async def main():
await session_service.create_session(
app_name=app_name,
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 and event.content.parts:
print(event.content.parts[0].text.strip())
asyncio.run(main())