phoenix-observability

Phoenix - AI 可观测性平台

面向 LLM 应用的开源 AI 可观测性与评估平台，提供追踪、评估、数据集、实验和实时监控功能。

何时使用 Phoenix

在以下情况使用 Phoenix：

• 通过详细的追踪调试 LLM 应用问题
• 在数据集上运行系统性评估
• 实时监控生产环境中的 LLM 系统
• 构建用于提示词/模型对比的实验流水线
• 实现可自托管的可观测性，避免供应商锁定

核心功能：

• 追踪：基于 OpenTelemetry 的追踪采集，适用于任何 LLM 框架
• 评估：使用 LLM-as-judge 评估器进行质量评估
• 数据集：用于回归测试的版本化测试集
• 实验：对比提示词、模型和配置
• Playground：使用多个模型进行交互式提示词测试
• 开源：可使用 PostgreSQL 或 SQLite 自托管

改用以下替代方案：

• LangSmith：以 LangChain 优先集成的托管平台
• Weights & Biases：侧重于深度学习实验跟踪
• Arize Cloud：具备企业功能的托管版 Phoenix
• MLflow：通用 ML 生命周期，侧重于模型注册表

快速开始

安装

pip install arize-phoenix

# With specific backends
pip install arize-phoenix[embeddings]  # Embedding analysis
pip install arize-phoenix-otel         # OpenTelemetry config
pip install arize-phoenix-evals        # Evaluation framework
pip install arize-phoenix-client       # Lightweight REST client

启动 Phoenix 服务

import phoenix as px

# Launch in notebook (ThreadServer mode)
session = px.launch_app()

# View UI
session.view()  # Embedded iframe
print(session.url)  # http://localhost:6006

命令行服务（生产环境）

# Start Phoenix server
phoenix serve

# With PostgreSQL
export PHOENIX_SQL_DATABASE_URL="postgresql://user:pass@host/db"
phoenix serve --port 6006

基础追踪

from phoenix.otel import register
from openinference.instrumentation.openai import OpenAIInstrumentor

# Configure OpenTelemetry with Phoenix
tracer_provider = register(
    project_name="my-llm-app",
    endpoint="http://localhost:6006/v1/traces"
)

# Instrument OpenAI SDK
OpenAIInstrumentor().instrument(tracer_provider=tracer_provider)

# All OpenAI calls are now traced
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
    model="gpt-4o",
    messages=[{"role": "user", "content": "Hello!"}]
)

核心概念

Trace 和 Span

Trace 表示一次完整的执行流程，而 Span 是该 Trace 中的单个操作。

from phoenix.otel import register
from opentelemetry import trace

# Setup tracing
tracer_provider = register(project_name="my-app")
tracer = trace.get_tracer(__name__)

# Create custom spans
with tracer.start_as_current_span("process_query") as span:
    span.set_attribute("input.value", query)

    # Child spans are automatically nested
    with tracer.start_as_current_span("retrieve_context"):
        context = retriever.search(query)

    with tracer.start_as_current_span("generate_response"):
        response = llm.generate(query, context)

    span.set_attribute("output.value", response)

项目

项目用于组织相关的 Trace：

import os
os.environ["PHOENIX_PROJECT_NAME"] = "production-chatbot"

# Or per-trace
from phoenix.otel import register
tracer_provider = register(project_name="experiment-v2")

框架集成

OpenAI

from phoenix.otel import register
from openinference.instrumentation.openai import OpenAIInstrumentor

tracer_provider = register()
OpenAIInstrumentor().instrument(tracer_provider=tracer_provider)

LangChain

from phoenix.otel import register
from openinference.instrumentation.langchain import LangChainInstrumentor

tracer_provider = register()
LangChainInstrumentor().instrument(tracer_provider=tracer_provider)

# All LangChain operations traced
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
response = llm.invoke("Hello!")

LlamaIndex

from phoenix.otel import register
from openinference.instrumentation.llama_index import LlamaIndexInstrumentor

tracer_provider = register()
LlamaIndexInstrumentor().instrument(tracer_provider=tracer_provider)

Anthropic

from phoenix.otel import register
from openinference.instrumentation.anthropic import AnthropicInstrumentor

tracer_provider = register()
AnthropicInstrumentor().instrument(tracer_provider=tracer_provider)

评估框架

内置评估器

from phoenix.evals import (
    OpenAIModel,
    HallucinationEvaluator,
    RelevanceEvaluator,
    ToxicityEvaluator,
    llm_classify
)

# Setup model for evaluation
eval_model = OpenAIModel(model="gpt-4o")

# Evaluate hallucination
hallucination_eval = HallucinationEvaluator(eval_model)
results = hallucination_eval.evaluate(
    input="What is the capital of France?",
    output="The capital of France is Paris.",
    reference="Paris is the capital of France."
)

自定义评估器

from phoenix.evals import llm_classify

# Define custom evaluation
def evaluate_helpfulness(input_text, output_text):
    template = """
    Evaluate if the response is helpful for the given question.

    Question: {input}
    Response: {output}

    Is this response helpful? Answer 'helpful' or 'not_helpful'.
    """

    result = llm_classify(
        model=eval_model,
        template=template,
        input=input_text,
        output=output_text,
        rails=["helpful", "not_helpful"]
    )
    return result

在数据集上运行评估

from phoenix import Client
from phoenix.evals import run_evals

client = Client()

# Get spans to evaluate
spans_df = client.get_spans_dataframe(
    project_name="my-app",
    filter_condition="span_kind == 'LLM'"
)

# Run evaluations
eval_results = run_evals(
    dataframe=spans_df,
    evaluators=[
        HallucinationEvaluator(eval_model),
        RelevanceEvaluator(eval_model)
    ],
    provide_explanation=True
)

# Log results back to Phoenix
client.log_evaluations(eval_results)

数据集与实验

创建数据集

from phoenix import Client

client = Client()

# Create dataset
dataset = client.create_dataset(
    name="qa-test-set",
    description="QA evaluation dataset"
)

# Add examples
client.add_examples_to_dataset(
    dataset_name="qa-test-set",
    examples=[
        {
            "input": {"question": "What is Python?"},
            "output": {"answer": "A programming language"}
        },
        {
            "input": {"question": "What is ML?"},
            "output": {"answer": "Machine learning"}
        }
    ]
)

运行实验

from phoenix import Client
from phoenix.experiments import run_experiment

client = Client()

def my_model(input_data):
    """Your model function."""
    question = input_data["question"]
    return {"answer": generate_answer(question)}

def accuracy_evaluator(input_data, output, expected):
    """Custom evaluator."""
    return {
        "score": 1.0 if expected["answer"].lower() in output["answer"].lower() else 0.0,
        "label": "correct" if expected["answer"].lower() in output["answer"].lower() else "incorrect"
    }

# Run experiment
results = run_experiment(
    dataset_name="qa-test-set",
    task=my_model,
    evaluators=[accuracy_evaluator],
    experiment_name="baseline-v1"
)

print(f"Average accuracy: {results.aggregate_metrics['accuracy']}")

客户端 API

查询 Trace 和 Span

from phoenix import Client

client = Client(endpoint="http://localhost:6006")

# Get spans as DataFrame
spans_df = client.get_spans_dataframe(
    project_name="my-app",
    filter_condition="span_kind == 'LLM'",
    limit=1000
)

# Get specific span
span = client.get_span(span_id="abc123")

# Get trace
trace = client.get_trace(trace_id="xyz789")

记录反馈

from phoenix import Client

client = Client()

# Log user feedback
client.log_annotation(
    span_id="abc123",
    name="user_rating",
    annotator_kind="HUMAN",
    score=0.8,
    label="helpful",
    metadata={"comment": "Good response"}
)

导出数据

# Export to pandas
df = client.get_spans_dataframe(project_name="my-app")

# Export traces
traces = client.list_traces(project_name="my-app")

生产部署

Docker

docker run -p 6006:6006 arizephoenix/phoenix:latest

搭配 PostgreSQL

# Set database URL
export PHOENIX_SQL_DATABASE_URL="postgresql://user:pass@host:5432/phoenix"

# Start server
phoenix serve --host 0.0.0.0 --port 6006

环境变量

变量	描述	默认值
PHOENIX_PORT	HTTP 服务端口	6006
PHOENIX_HOST	服务绑定地址	127.0.0.1
PHOENIX_GRPC_PORT	gRPC/OTLP 端口	4317
PHOENIX_SQL_DATABASE_URL	数据库连接	SQLite 临时
PHOENIX_WORKING_DIR	数据存储目录	操作系统临时目录
PHOENIX_ENABLE_AUTH	启用身份验证	false
PHOENIX_SECRET	JWT 签名密钥	启用身份验证时必填

启用身份验证

export PHOENIX_ENABLE_AUTH=true
export PHOENIX_SECRET="your-secret-key-min-32-chars"
export PHOENIX_ADMIN_SECRET="admin-bootstrap-token"

phoenix serve

最佳实践

1. 使用项目：按环境（dev/staging/prod）隔离 Trace
2. 添加元数据：加入用户 ID、会话 ID 以便调试
3. 定期评估：在 CI/CD 中运行自动化评估
4. 为数据集设置版本：跟踪测试集随时间的变化
5. 监控成本：通过 Phoenix 仪表盘跟踪 Token 用量
6. 自托管：生产部署使用 PostgreSQL

常见问题

Trace 未显示：

from phoenix.otel import register

# Verify endpoint
tracer_provider = register(
    project_name="my-app",
    endpoint="http://localhost:6006/v1/traces"  # Correct endpoint
)

# Force flush
from opentelemetry import trace
trace.get_tracer_provider().force_flush()

Notebook 内存占用过高：

# Close session when done
session = px.launch_app()
# ... do work ...
session.close()
px.close_app()

数据库连接问题：

# Verify PostgreSQL connection
psql $PHOENIX_SQL_DATABASE_URL -c "SELECT 1"

# Check Phoenix logs
phoenix serve --log-level debug

参考资料

• 高级用法 - 自定义评估器、实验、生产环境设置
• 故障排除 - 常见问题、调试、性能

资源

• 文档：https://docs.arize.com/phoenix
• 代码仓库：https://github.com/Arize-ai/phoenix
• Docker Hub：https://hub.docker.com/r/arizephoenix/phoenix
• 版本：12.0.0+
• 许可证：Apache 2.0