name: phoenix-observability description: 开源AI观测平台,用于LLM追踪、评估和监控。在调试LLM应用时使用详细追踪、在数据集上运行评估,或监控生产AI系统并提供实时洞察。 version: 1.0.0 author: Orchestra Research license: MIT tags: [观测性, Phoenix, Arize, 追踪, 评估, 监控, LLM运维, OpenTelemetry] dependencies: [arize-phoenix>=12.0.0]
Phoenix - AI观测平台
开源AI观测和评估平台,用于LLM应用,具有追踪、评估、数据集、实验和实时监控功能。
何时使用Phoenix
使用Phoenix时:
- 调试LLM应用问题,提供详细追踪
- 在数据集上运行系统评估
- 实时监控生产LLM系统
- 构建实验管道进行提示/模型比较
- 自托管观测,无供应商锁定
关键功能:
- 追踪:基于OpenTelemetry的追踪收集,适用于任何LLM框架
- 评估:LLM作为裁判评估器,用于质量评估
- 数据集:版本化测试集,用于回归测试
- 实验:比较提示、模型和配置
- 游乐场:交互式提示测试,支持多种模型
- 开源:自托管,支持PostgreSQL或SQLite
使用替代方案:
- LangSmith:管理平台,与LangChain优先集成
- Weights & Biases:深度学习实验跟踪重点
- Arize Cloud:管理版Phoenix,具有企业功能
- MLflow:通用ML生命周期,模型注册重点
快速开始
安装
pip install arize-phoenix
# 使用特定后端
pip install arize-phoenix[embeddings] # 嵌入分析
pip install arize-phoenix-otel # OpenTelemetry配置
pip install arize-phoenix-evals # 评估框架
pip install arize-phoenix-client # 轻量级REST客户端
启动Phoenix服务器
import phoenix as px
# 在笔记本中启动(ThreadServer模式)
session = px.launch_app()
# 查看UI
session.view() # 嵌入式iframe
print(session.url) # http://localhost:6006
命令行服务器(生产环境)
# 启动Phoenix服务器
phoenix serve
# 使用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
# 配置OpenTelemetry与Phoenix
tracer_provider = register(
project_name="my-llm-app",
endpoint="http://localhost:6006/v1/traces"
)
# 仪表化OpenAI SDK
OpenAIInstrumentor().instrument(tracer_provider=tracer_provider)
# 所有OpenAI调用现在都被追踪
from openai import OpenAI
client = OpenAI()
response = client.chat.completions.create(
model="gpt-4o",
messages=[{"role": "user", "content": "Hello!"}]
)
核心概念
追踪和跨度
一个追踪代表完整的执行流程,而跨度是追踪中的单个操作。
from phoenix.otel import register
from opentelemetry import trace
# 设置追踪
tracer_provider = register(project_name="my-app")
tracer = trace.get_tracer(__name__)
# 创建自定义跨度
with tracer.start_as_current_span("process_query") as span:
span.set_attribute("input.value", query)
# 子跨度自动嵌套
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)
项目
项目组织相关追踪:
import os
os.environ["PHOENIX_PROJECT_NAME"] = "production-chatbot"
# 或按追踪
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)
# 所有LangChain操作被追踪
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
)
# 设置评估模型
eval_model = OpenAIModel(model="gpt-4o")
# 评估幻觉
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
# 定义自定义评估
def evaluate_helpfulness(input_text, output_text):
template = """
评估响应对于给定问题是否有帮助。
问题:{input}
响应:{output}
这个响应有帮助吗?回答“helpful”或“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()
# 获取要评估的跨度
spans_df = client.get_spans_dataframe(
project_name="my-app",
filter_condition="span_kind == 'LLM'"
)
# 运行评估
eval_results = run_evals(
dataframe=spans_df,
evaluators=[
HallucinationEvaluator(eval_model),
RelevanceEvaluator(eval_model)
],
provide_explanation=True
)
# 将结果记录回Phoenix
client.log_evaluations(eval_results)
数据集和实验
创建数据集
from phoenix import Client
client = Client()
# 创建数据集
dataset = client.create_dataset(
name="qa-test-set",
description="QA评估数据集"
)
# 添加示例
client.add_examples_to_dataset(
dataset_name="qa-test-set",
examples=[
{
"input": {"question": "什么是Python?"},
"output": {"answer": "一种编程语言"}
},
{
"input": {"question": "什么是ML?"},
"output": {"answer": "机器学习"}
}
]
)
运行实验
from phoenix import Client
from phoenix.experiments import run_experiment
client = Client()
def my_model(input_data):
"""你的模型函数。"""
question = input_data["question"]
return {"answer": generate_answer(question)}
def accuracy_evaluator(input_data, output, expected):
"""自定义评估器。"""
return {
"score": 1.0 if expected["answer"].lower() in output["answer"].lower() else 0.0,
"label": "正确" if expected["answer"].lower() in output["answer"].lower() else "错误"
}
# 运行实验
results = run_experiment(
dataset_name="qa-test-set",
task=my_model,
evaluators=[accuracy_evaluator],
experiment_name="baseline-v1"
)
print(f"平均准确率: {results.aggregate_metrics['accuracy']}")
客户端API
查询追踪和跨度
from phoenix import Client
client = Client(endpoint="http://localhost:6006")
# 获取跨度作为DataFrame
spans_df = client.get_spans_dataframe(
project_name="my-app",
filter_condition="span_kind == 'LLM'",
limit=1000
)
# 获取特定跨度
span = client.get_span(span_id="abc123")
# 获取追踪
trace = client.get_trace(trace_id="xyz789")
记录反馈
from phoenix import Client
client = Client()
# 记录用户反馈
client.log_annotation(
span_id="abc123",
name="user_rating",
annotator_kind="HUMAN",
score=0.8,
label="有帮助",
metadata={"comment": "良好响应"}
)
导出数据
# 导出到pandas
df = client.get_spans_dataframe(project_name="my-app")
# 导出追踪
traces = client.list_traces(project_name="my-app")
生产部署
Docker
docker run -p 6006:6006 arizephoenix/phoenix:latest
使用PostgreSQL
# 设置数据库URL
export PHOENIX_SQL_DATABASE_URL="postgresql://user:pass@host:5432/phoenix"
# 启动服务器
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
最佳实践
- 使用项目:按环境(开发/预发/生产)分离追踪
- 添加元数据:包括用户ID、会话ID以进行调试
- 定期评估:在CI/CD中运行自动化评估
- 版本化数据集:跟踪测试集随时间的变化
- 监控成本:通过Phoenix仪表板跟踪令牌使用
- 自托管:生产部署使用PostgreSQL
常见问题
追踪未出现:
from phoenix.otel import register
# 验证端点
tracer_provider = register(
project_name="my-app",
endpoint="http://localhost:6006/v1/traces" # 正确端点
)
# 强制刷新
from opentelemetry import trace
trace.get_tracer_provider().force_flush()
笔记本中高内存:
# 完成后关闭会话
session = px.launch_app()
# ... 进行工作 ...
session.close()
px.close_app()
数据库连接问题:
# 验证PostgreSQL连接
psql $PHOENIX_SQL_DATABASE_URL -c "SELECT 1"
# 检查Phoenix日志
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