Skip to content

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

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

概述¶

Phoenix 可以使用 OpenInference 仪表化自动收集来自 Google ADK 的追踪信息，允许你：

追踪智能体交互 - 自动捕获每个智能体运行、工具调用、模型请求和响应，包含完整的上下文和元数据
评估性能 - 使用自定义或预构建的评估器评估智能体行为，并运行实验来测试智能体配置
调试问题 - 分析详细的追踪信息，快速识别瓶颈、失败的工具调用和意外的智能体行为
自托管控制 - 将你的数据保存在自己的基础设施上

安装¶

1. 安装所需的包¶

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

设置¶

1. 启动 Phoenix¶

这些说明向你展示如何使用 Phoenix Cloud。你也可以在笔记本中、从终端或使用容器自托管的方式启动 Phoenix。

注册免费的 Phoenix 账户。
在你新 Phoenix 空间的设置页面中创建你的 API 密钥
复制你的端点，它应该看起来像：https://app.phoenix.arize.com/s/[你的空间名称]

设置你的 Phoenix 端点和 API 密钥：

import os

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

# 如果你在 2025 年 6 月 24 日之前创建了 Phoenix Cloud 实例，请将 API 密钥设置为 Header：
# os.environ["PHOENIX_CLIENT_HEADERS"] = f"api_key={os.getenv('PHOENIX_API_KEY')}"

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 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:
    """获取指定城市的当前天气报告。

    Args:
        city (str): 要获取天气报告的城市名称。

    Returns:
        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-2.0-flash-exp",
    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

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():
        print(event.content.parts[0].text.strip())

支持和资源¶