Skip to content

Agent Observability with Phoenix

使用 Phoenix 进行智能体可观测性

Phoenix is an open-source, self-hosted observability platform for monitoring, debugging, and improving LLM applications and AI Agents at scale. It provides comprehensive tracing and evaluation capabilities for your Google ADK applications. To get started, sign up for a free account.

Phoenix 是一个开源、自托管的可观测性平台,用于大规模监控、调试和改进 LLM 应用程序和 AI 智能体。它为您的 Google ADK 应用程序提供全面的跟踪和评估功能。要开始使用,请注册一个免费账户。

Overview

概述

Phoenix can automatically collect traces from Google ADK using OpenInference instrumentation, allowing you to:

Phoenix 可以使用 OpenInference 插桩 自动从 Google ADK 收集跟踪,允许您:

  • Trace agent interactions - Automatically capture every agent run, tool call, model request, and response with full context and metadata
  • 跟踪智能体交互 - 自动捕获每个智能体运行、工具调用、模型请求和响应,包括完整的上下文和元数据
  • Evaluate performance - Assess agent behavior using custom or pre-built evaluators and run experiments to test agent configurations
  • 评估性能 - 使用自定义或预构建的评估器评估智能体行为,并运行实验以测试智能体配置
  • Debug issues - Analyze detailed traces to quickly identify bottlenecks, failed tool calls, and unexpected agent behavior
  • 调试问题 - 分析详细的跟踪以快速识别瓶颈、失败的工具调用和意外的智能体行为
  • Self-hosted control - Keep your data on your own infrastructure
  • 自托管控制 - 将数据保留在自己的基础设施上

Installation

安装

1. Install Required Packages

1. 安装必需的包

pip install openinference-instrumentation-google-adk google-adk arize-phoenix-otel

Setup

设置

1. Launch Phoenix

1. 启动 Phoenix

These instructions show you how to use Phoenix Cloud. You can also launch Phoenix in a notebook, from your terminal, or self-host it using a container.

这些说明向您展示如何使用 Phoenix Cloud。您还可以在笔记本中、从终端或使用容器启动 Phoenix。

  1. Sign up for a free Phoenix account.
  2. 注册一个免费的 Phoenix 账户。
  3. From the Settings page of your new Phoenix Space, create your API key
  4. 从新 Phoenix Space 的设置页面,创建您的 API 密钥
  5. Copy your endpoint which should look like: https://app.phoenix.arize.com/s/[your-space-name]
  6. 复制您的端点,应该类似于:https://app.phoenix.arize.com/s/[your-space-name]

Set your Phoenix endpoint and API Key: 设置您的 Phoenix 端点和 API 密钥:

import os

os.environ["PHOENIX_API_KEY"] = "ADD YOUR PHOENIX API KEY"
os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "ADD YOUR PHOENIX COLLECTOR ENDPOINT"

# If you created your Phoenix Cloud instance before June 24th, 2025, set the API key as a header:
# 如果您在 2025 年 6 月 24 日之前创建了 Phoenix Cloud 实例,请将 API 密钥设置为头部:
# os.environ["PHOENIX_CLIENT_HEADERS"] = f"api_key={os.getenv('PHOENIX_API_KEY')}"

2. Connect your application to Phoenix

2. 将您的应用程序连接到 Phoenix

from phoenix.otel import register

# Configure the Phoenix tracer
# 配置 Phoenix 跟踪器
tracer_provider = register(
    project_name="my-llm-app",  # Default is 'default' / 默认为 'default'
    auto_instrument=True        # Auto-instrument your app based on installed OI dependencies / 根据已安装的 OI 依赖项自动为您的应用插桩
)

Observe

观察

Now that you have tracing setup, all Google ADK SDK requests will be streamed to Phoenix for observability and evaluation.

现在您已经设置了跟踪,所有 Google ADK SDK 请求都将流式传输到 Phoenix 以进行可观测性和评估。

import nest_asyncio
nest_asyncio.apply()

from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.genai import types

# Define a tool function
# 定义工具函数
def get_weather(city: str) -> dict:
    """Retrieves current weather report for a specified city.

    Args:
        city (str): The name of city for which to retrieve weather report.

    Returns:
        dict: status and result or error msg.
    """
    if city.lower() == "new york":
        return {
            "status": "success",
            "report": (
                "The weather in New York is sunny with a temperature of 25 degrees"
                " Celsius (77 degrees Fahrenheit)."
            ),
        }
    else:
        return {
            "status": "error",
            "error_message": f"Weather information for '{city}' is not available.",
        }

# Create an agent with tools
# 创建带有工具的智能体
agent = Agent(
    name="weather_agent",
    model="gemini-2.0-flash-exp",
    description="Agent to answer questions using weather tools.",
    instruction="You must use available tools to find an answer.",
    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

await session_service.create_session(
    app_name=app_name,
    user_id=user_id,
    session_id=session_id
)

# Run the agent (all interactions will be traced)
# 运行智能体(所有交互都将被跟踪)
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 the weather in New York?")]
    )
):
    if event.is_final_response():
        print(event.content.parts[0].text.strip())

Support and Resources

支持和资源