Arize Phoenix 开源可观测性

import os

# Phoenix服务器端点（默认本地: http://localhost:6006）
os.environ["PHOENIX_COLLECTOR_ENDPOINT"] = "http://localhost:6006"

# 项目名称（用于组织追踪数据）
os.environ["PHOENIX_PROJECT_NAME"] = "my-llm-project"

# API密钥（使用Phoenix Cloud时需要）
os.environ["PHOENIX_API_KEY"] = "phoenix_api_key_..."

# gRPC端口覆盖（默认4317）
os.environ["PHOENIX_GRPC_PORT"] = "4317"

# 额外的请求头
os.environ["PHOENIX_CLIENT_HEADERS"] = "Authorization=Bearer token,custom-header=value"

# 禁用遥测
os.environ["PHOENIX_TELEMETRY_ENABLED"] = "false"

import phoenix as px
print(px.__version__)

import phoenix as px

# 启动Phoenix本地UI（默认端口6006）
session = px.launch_app()

# 查看访问URL
print(f"Phoenix UI: {session.url}")

# 启动时指定端口
session = px.launch_app(port=8080)

# 使用完毕后关闭
session.close()

from phoenix.otel import register

# 最简配置：自动追踪已安装的AI/ML库
tracer_provider = register(auto_instrument=True)

# 完整配置
tracer_provider = register(
    project_name="my-llm-app",        # 项目名称
    endpoint="http://localhost:6006/v1/traces",  # 追踪数据端点
    headers={"Authorization": "Bearer TOKEN"},   # 认证头
    batch=True,                        # 生产环境推荐：批量发送Span
    auto_instrument=True,              # 自动追踪已安装的AI/ML库
    protocol="grpc",                   # 传输协议：http/protobuf 或 grpc
)

from phoenix.otel import register

# 注册追踪器并启用自动追踪
tracer_provider = register(
    project_name="my-langchain-app",
    auto_instrument=True  # 自动追踪LangChain、OpenAI等
)

# 此后所有LangChain调用都会被自动追踪，无需额外代码
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4o")
response = llm.invoke("你好，请介绍一下Python")  # 自动生成追踪Span

from phoenix.otel import register

tracer_provider = register(project_name="my-app")
tracer = tracer_provider.get_tracer(__name__)

# 使用@tracer.chain装饰器追踪工作流函数
@tracer.chain
def process_document(document: str) -> str:
    """顺序执行的文档处理管道"""
    cleaned = document.strip().lower()
    return cleaned

# 使用@tracer.tool装饰器追踪工具调用
@tracer.tool
def search_database(query: str) -> list:
    """数据库搜索工具"""
    # 模拟搜索
    return [{"id": 1, "text": f"搜索结果: {query}"}]

# 使用@tracer.agent装饰器追踪AI代理
@tracer.agent
def my_agent(task: str) -> str:
    """自主决策的AI代理"""
    return f"完成: {task}"

# 调用这些函数时会自动生成追踪Span
result = process_document("  Hello World  ")

from phoenix.otel import register
from opentelemetry.trace import StatusCode, Status

tracer_provider = register(project_name="my-app")
tracer = tracer_provider.get_tracer(__name__)

def invoke_llm(messages: list, model: str = "gpt-4o") -> str:
    """手动追踪LLM调用"""
    with tracer.start_as_current_span(
        "llm_call",                       # Span名称
        openinference_span_kind="llm",     # Span类型
    ) as span:
        # 设置输入属性
        span.set_input(messages)
        span.set_attributes({
            "llm.input_messages": [
                {"role": msg["role"], "content": msg["content"]}
                for msg in messages
            ],
            "llm.model_name": model,
        })

        # 执行LLM调用（示例）
        import openai
        client = openai.OpenAI()
        response = client.chat.completions.create(
            model=model,
            messages=messages,
        )
        output_text = response.choices[0].message.content

        # 设置输出属性
        span.set_output(output_text)
        span.set_attributes({
            "llm.output_messages": [
                {"role": "assistant", "content": output_text}
            ],
            "llm.token_count.prompt": response.usage.prompt_tokens if response.usage else 0,
            "llm.token_count.completion": response.usage.completion_tokens if response.usage else 0,
        })
        span.set_status(Status(StatusCode.OK))

        return output_text

from openinference.instrumentation import using_attributes

# 上下文中所有Span将自动携带session_id和user_id
with using_attributes(session_id="session-123", user_id="user-456"):
    result = my_function("input")

from openinference.instrumentation import suppress_tracing

# 此代码块内的调用不会被追踪
with suppress_tracing():
    result = my_function("input")

from opentelemetry import trace as trace_api
from phoenix.otel import HTTPSpanExporter, TracerProvider, SimpleSpanProcessor

# 方式1：使用Phoenix的TracerProvider（替代标准OTEL TracerProvider）
tracer_provider = TracerProvider()
span_exporter = HTTPSpanExporter(endpoint="http://localhost:6006/v1/traces")
span_processor = SimpleSpanProcessor(span_exporter=span_exporter)
tracer_provider.add_span_processor(span_processor)
trace_api.set_tracer_provider(tracer_provider)

# 方式2：简化配置（自动推断Exporter类型）
from phoenix.otel import TracerProvider
tracer_provider = TracerProvider(endpoint="http://localhost:4317")
trace_api.set_tracer_provider(tracer_provider)

# 方式3：带项目名称和资源标签
from phoenix.otel import Resource, PROJECT_NAME, TracerProvider
tracer_provider = TracerProvider(
    resource=Resource({PROJECT_NAME: "my-project"})
)
trace_api.set_tracer_provider(tracer_provider)

import phoenix as px
import pandas as pd
import numpy as np

# 示例：创建带有嵌入的数据集
# 假设我们有文档嵌入和对应的类别标签
embeddings = np.random.randn(200, 1536)  # 200个文档的嵌入向量（如OpenAI text-embedding-ada-002维度）
categories = np.random.choice(["技术", "金融", "医疗", "教育"], size=200)

# 创建Phoenix数据集用于嵌入可视化
# 使用px.launch_app()启动UI后，可在嵌入探索器中查看
session = px.launch_app()

# 通过Phoenix客户端上传嵌入数据
from phoenix.client import Client

client = Client()

# 将嵌入数据组织为DataFrame
df = pd.DataFrame({
    "text": [f"文档内容_{i}" for i in range(200)],
    "category": categories,
    "embedding": list(embeddings),  # 嵌入向量列
})

# Phoenix UI中的嵌入探索器可以：
# 1. 自动对嵌入执行UMAP降维
# 2. 在2D/3D空间中可视化聚类
# 3. 按类别标签着色，发现聚类与语义的对应关系
# 4. 识别离群点和数据漂移

from phoenix.evals.llm import LLM
from phoenix.evals.metrics import (
    FaithfulnessEvaluator,           # 忠实度/幻觉检测
    ConcisenessEvaluator,            # 简洁性评估
    CorrectnessEvaluator,            # 正确性评估
    DocumentRelevanceEvaluator,      # 文档相关性评估
    RefusalEvaluator,                # 拒答检测
    ToolInvocationEvaluator,         # 工具调用评估
    ToolSelectionEvaluator,          # 工具选择评估
    ToolResponseHandlingEvaluator,   # 工具响应处理评估
)

# 初始化LLM（支持多提供商）
llm = LLM(provider="openai", model="gpt-4o")
# llm = LLM(provider="anthropic", model="claude-3-5-sonnet-20241022")
# llm = LLM(provider="google", model="gemini-1.5-pro")
# llm = LLM(provider="litellm", model="gpt-4o")  # 统一接口，支持100+提供商

faithfulness = FaithfulnessEvaluator(llm=llm)

scores = faithfulness.evaluate({
    "input": "法国的首都是什么？",
    "context": "巴黎是法国的首都。",
    "output": "法国的首都是柏林。",  # 与上下文矛盾，属于幻觉
})
scores[0].pretty_print()
# Score(name='faithfulness', score=0.0, label='unfaithful', explanation='输出与上下文矛盾...')

doc_relevance = DocumentRelevanceEvaluator(llm=llm)

scores = doc_relevance.evaluate({
    "input": "如何重置密码？",
    "context": "公司成立于2010年，总部位于北京。",  # 与问题不相关
    "output": "请点击登录页面的'忘记密码'链接。",
})
scores[0].pretty_print()

from phoenix.evals import create_classifier
from phoenix.evals.llm import LLM

llm = LLM(provider="openai", model="gpt-4o")

# 创建自定义"有用性"评估器
helpfulness_evaluator = create_classifier(
    name="helpfulness",                       # 评估器名称
    prompt_template="""请评估以下回复是否有用。

用户问题: {input}
系统回复: {output}

请判断回复是否有用。""",                       # 提示词模板
    llm=llm,                                   # LLM实例
    choices={"helpful": 1.0, "not_helpful": 0.0},  # 分类标签与评分映射
)

# 运行评估
scores = helpfulness_evaluator.evaluate({
    "input": "如何重置密码？",
    "output": "请前往设置 > 账户 > 重置密码。"
})
scores[0].pretty_print()

scores = helpfulness_evaluator.evaluate(
    {"data": {"query": "如何重置密码？", "response": "请前往设置。"}},
    input_mapping={"input": "data.query", "output": "data.response"}
)

from phoenix.evals.metrics import exact_match, MatchesRegex, PrecisionRecallFScore

# 精确匹配
match_result = exact_match({"output": "Paris", "expected": "Paris"})
# 返回 True

# 正则匹配
regex_evaluator = MatchesRegex(pattern=r"^\d{4}\-\d{2}\-\d{2}$")
regex_result = regex_evaluator.evaluate({"output": "2024-03-15"})
# 返回 True

# 精确率/召回率/F1
prf = PrecisionRecallFScore()
prf_result = prf.evaluate({"output": "A B C", "expected": "A B D"})

import pandas as pd
from phoenix.evals import create_classifier, evaluate_dataframe, async_evaluate_dataframe
from phoenix.evals.llm import LLM

llm = LLM(provider="openai", model="gpt-4o")

# 创建多个评估器
relevance_evaluator = create_classifier(
    name="relevance",
    prompt_template="回复是否与问题相关？\n\n问题: {input}\n回复: {output}",
    llm=llm,
    choices={"relevant": 1.0, "irrelevant": 0.0},
)

helpfulness_evaluator = create_classifier(
    name="helpfulness",
    prompt_template="回复是否有用？\n\n问题: {input}\n回复: {output}",
    llm=llm,
    choices={"helpful": 1.0, "not_helpful": 0.0},
)

# 准备评估数据
df = pd.DataFrame([
    {"input": "如何重置密码？", "output": "请前往设置 > 账户 > 重置密码。"},
    {"input": "今天天气如何？", "output": "我可以帮助您重置密码。"},  # 无关回复
])

# 同步批量评估
results_df = evaluate_dataframe(
    dataframe=df,
    evaluators=[relevance_evaluator, helpfulness_evaluator]
)

# 异步批量评估（大数据集推荐，最高20倍加速）
import asyncio
results_df = asyncio.run(async_evaluate_dataframe(
    dataframe=df,
    evaluators=[relevance_evaluator, helpfulness_evaluator],
))

from phoenix.client import Client

client = Client()

# 方式1：从字典列表创建
dataset = client.datasets.create_dataset(
    name="customer-support-qa",
    dataset_description="客服问答评估数据集",
    inputs=[
        {"question": "如何重置密码？"},
        {"question": "退货政策是什么？"},
        {"question": "如何追踪我的订单？"}
    ],
    outputs=[
        {"answer": "请点击登录页面的'忘记密码'链接重置密码。"},
        {"answer": "我们为原包装未使用的商品提供30天退货服务。"},
        {"answer": "您可以使用发送到邮箱的追踪号来追踪订单。"}
    ],
    metadata=[
        {"category": "account", "difficulty": "easy"},
        {"category": "policy", "difficulty": "medium"},
        {"category": "orders", "difficulty": "easy"}
    ]
)

# 方式2：从Pandas DataFrame创建
import pandas as pd

df = pd.DataFrame({
    "prompt": ["你好", "你好啊", "早上好"],
    "response": ["你好！有什么可以帮您？", "您好！请问需要什么帮助？", "早上好！有什么可以为您效劳？"],
    "sentiment": ["neutral", "positive", "positive"],
    "length": [20, 25, 30]
})

dataset = client.datasets.create_dataset(
    name="greeting-responses",
    dataframe=df,
    input_keys=["prompt"],                # 映射到input的列
    output_keys=["response"],             # 映射到output的列
    metadata_keys=["sentiment", "length"] # 映射到metadata的列
)

# 方式3：从CSV文件创建
dataset = client.datasets.create_dataset(
    name="csv-dataset",
    csv_file_path="path/to/data.csv",
    input_keys=["question", "context"],
    output_keys=["answer"],
    metadata_keys=["source", "confidence"]
)

# 列出所有数据集
datasets = client.datasets.list()
for ds in datasets:
    print(f"数据集: {ds['name']} ({ds['example_count']} 个样本)")

# 获取特定数据集
dataset = client.datasets.get_dataset(dataset="customer-support-qa")

# 获取特定版本的数据集
versioned = client.datasets.get_dataset(
    dataset="customer-support-qa",
    version_id="version-123"
)

# 向已有数据集添加样本
client.datasets.add_examples_to_dataset(
    dataset="customer-support-qa",
    inputs=[{"question": "如何取消订阅？"}],
    outputs=[{"answer": "您可以在账户设置中取消订阅。"}],
    metadata=[{"category": "subscription", "difficulty": "medium"}]
)

# 从DataFrame追加样本
new_df = pd.DataFrame({
    "question": ["营业时间是什么？", "有在线客服吗？"],
    "answer": ["我们7x24小时营业", "有，可在官网使用在线客服"],
    "topic": ["hours", "support"]
})
client.datasets.add_examples_to_dataset(
    dataset="customer-support-qa",
    dataframe=new_df,
    input_keys=["question"],
    output_keys=["answer"],
    metadata_keys=["topic"]
)

# 查看数据集版本历史
versions = client.datasets.get_dataset_versions(dataset="customer-support-qa")
for version in versions:
    print(f"版本: {version['version_id']}, 创建时间: {version['created_at']}")

# 将数据集转为DataFrame
df = dataset.to_dataframe()
# 列: ['input', 'output', 'metadata']

# 遍历数据集
for example in dataset:
    print(f"输入: {example['input']}")
    print(f"输出: {example['output']}")
    print(f"元数据: {example['metadata']}")

# 序列化与反序列化
dataset_dict = dataset.to_dict()
restored_dataset = Dataset.from_dict(dataset_dict)

from phoenix.client.experiments import run_experiment, evaluate_experiment

dataset = client.datasets.get_dataset(dataset="customer-support-qa")

# 定义任务函数
def my_task(input):
    return f"处理: {input['question']}"

# 运行实验
experiment = run_experiment(
    dataset=dataset,
    task=my_task,
    experiment_name="greeting-experiment"
)

# 带评估器的实验
def accuracy_evaluator(output, expected):
    """准确度评估器"""
    return 1.0 if output == expected.get("text") else 0.0

def detailed_evaluator(output, input, expected, metadata):
    """详细评估器（可访问所有字段）"""
    score = calculate_similarity(output, expected)
    return {"score": score, "label": "pass" if score > 0.8 else "fail"}

experiment = run_experiment(
    dataset=dataset,
    task=my_task,
    evaluators=[accuracy_evaluator, detailed_evaluator],
    experiment_name="evaluated-experiment"
)

# 对已有实验添加评估
evaluated = evaluate_experiment(
    experiment=experiment,
    evaluators=[accuracy_evaluator],
    print_summary=True
)

# 异步实验（大数据集推荐）
from phoenix.client.experiments import async_run_experiment
from phoenix.client import AsyncClient

async_client = AsyncClient()
dataset = await async_client.datasets.get_dataset(dataset="my-dataset")

async def async_task(input):
    return f"处理: {input['name']}"

experiment = await async_run_experiment(
    dataset=dataset,
    task=async_task,
    experiment_name="async-experiment",
    concurrency=5  # 并发执行5个任务
)

# 第1步：安装集成包
# pip install arize-phoenix-otel openinference-instrumentation-langchain

# 第2步：注册追踪器并启用自动追踪
from phoenix.otel import register

tracer_provider = register(
    project_name="langchain-rag-app",
    auto_instrument=True  # 自动追踪LangChain
)

# 第3步：正常编写LangChain代码，所有调用将被自动追踪
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.chains import RetrievalQA
from langchain_community.vectorstores import Chroma

# 创建LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 创建RAG链
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手，请根据以下上下文回答问题：\n{context}"),
    ("human", "{question}"),
])

# 所有LLM调用、检索操作、链执行都将自动生成追踪Span
chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=vectorstore.as_retriever(),  # 检索操作自动追踪为retriever Span
)

result = chain.invoke({"query": "什么是RAG？"})
# 在Phoenix UI中可以看到完整的追踪链：
# chain -> retriever -> llm 的层级关系

"""
完整的RAG应用追踪示例：
展示如何追踪检索、重排序、生成等环节
"""
from phoenix.otel import register
from opentelemetry.trace import StatusCode, Status
import openai

# 注册追踪器
tracer_provider = register(
    project_name="rag-app",
    endpoint="http://localhost:6006/v1/traces",
)
tracer = tracer_provider.get_tracer(__name__)

@tracer.chain
def rag_pipeline(query: str) -> str:
    """RAG管道：检索 -> 重排序 -> 生成"""
    # 步骤1：检索相关文档
    documents = retrieve_documents(query)
    # 步骤2：重排序
    ranked_docs = rerank_documents(query, documents)
    # 步骤3：生成回答
    answer = generate_answer(query, ranked_docs)
    return answer

@tracer.retriever
def retrieve_documents(query: str, top_k: int = 5) -> list:
    """文档检索"""
    # 模拟向量检索
    docs = [
        {"text": "RAG是检索增强生成的缩写...", "score": 0.95},
        {"text": "向量数据库用于存储文档嵌入...", "score": 0.87},
    ][:top_k]
    return docs

@tracer.tool
def rerank_documents(query: str, documents: list) -> list:
    """文档重排序"""
    # 按相关性分数排序
    return sorted(documents, key=lambda x: x["score"], reverse=True)

@tracer.llm
def generate_answer(query: str, documents: list) -> str:
    """基于文档生成回答"""
    context = "\n".join([doc["text"] for doc in documents])
    messages = [
        {"role": "system", "content": f"请根据以下上下文回答问题：\n{context}"},
        {"role": "user", "content": query},
    ]

    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=messages,
    )
    return response.choices[0].message.content

# 运行RAG管道
answer = rag_pipeline("什么是RAG？")
print(f"回答: {answer}")
# 在Phoenix UI中可以看到：rag_pipeline(chain) -> retrieve_documents(retriever) -> rerank_documents(tool) -> generate_answer(llm) 的完整追踪链

"""
对LLM输出进行自动化质量评估
"""
import pandas as pd
from phoenix.evals import create_classifier, evaluate_dataframe
from phoenix.evals.llm import LLM
from phoenix.evals.metrics import FaithfulnessEvaluator

llm = LLM(provider="openai", model="gpt-4o")

# 创建评估数据
eval_data = pd.DataFrame([
    {
        "input": "Python是什么？",
        "context": "Python是一种高级编程语言，以简洁易读的语法著称。",
        "output": "Python是一种高级编程语言，语法简洁易读。",
    },
    {
        "input": "什么是机器学习？",
        "context": "机器学习是人工智能的一个分支，通过数据训练模型。",
        "output": "机器学习是一种烹饪技术，通过加热食物使其变色。",
    },
])

# 评估1：忠实度（幻觉检测）
faithfulness = FaithfulnessEvaluator(llm=llm)
faithfulness_scores = []
for _, row in eval_data.iterrows():
    scores = faithfulness.evaluate({
        "input": row["input"],
        "context": row["context"],
        "output": row["output"],
    })
    faithfulness_scores.append(scores[0])

for i, score in enumerate(faithfulness_scores):
    print(f"样本{i}: {score.label} (score={score.score})")
    print(f"  解释: {score.explanation}")

# 评估2：自定义相关性评估器
relevance = create_classifier(
    name="relevance",
    prompt_template="回复是否与问题相关？\n\n问题: {input}\n回复: {output}",
    llm=llm,
    choices={"relevant": 1.0, "irrelevant": 0.0},
)

# 批量评估
results_df = evaluate_dataframe(
    dataframe=eval_data,
    evaluators=[relevance],
)
print(results_df)

"""
创建评估数据集，运行实验并追踪结果
"""
from phoenix.client import Client
from phoenix.client.experiments import run_experiment

client = Client()

# 创建评估数据集
dataset = client.datasets.create_dataset(
    name="qa-eval-dataset",
    dataset_description="问答系统评估数据集",
    inputs=[
        {"question": "Python的创始人是谁？"},
        {"question": "什么是列表推导式？"},
        {"question": "如何处理Python中的异常？"},
    ],
    outputs=[
        {"answer": "Guido van Rossum"},
        {"answer": "列表推导式是一种简洁的创建列表的方式，如 [x**2 for x in range(10)]"},
        {"answer": "使用try-except语句块来捕获和处理异常"},
    ],
    metadata=[
        {"category": "basic", "difficulty": "easy"},
        {"category": "syntax", "difficulty": "medium"},
        {"category": "error_handling", "difficulty": "medium"},
    ]
)

# 定义任务函数
def qa_task(input):
    """调用你的QA系统"""
    import openai
    client = openai.OpenAI()
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "请简洁地回答以下问题。"},
            {"role": "user", "content": input["question"]},
        ],
    )
    return response.choices[0].message.content

# 定义评估器
def answer_quality_evaluator(output, expected):
    """评估答案质量"""
    expected_answer = expected.get("answer", "")
    # 简单的关键词匹配
    overlap = len(set(output.split()) & set(expected_answer.split()))
    total = len(set(expected_answer.split()))
    score = overlap / total if total > 0 else 0
    return {"score": score, "label": "pass" if score > 0.5 else "fail"}

# 运行实验
experiment = run_experiment(
    dataset=dataset,
    task=qa_task,
    evaluators=[answer_quality_evaluator],
    experiment_name="qa-quality-v1"
)

print(f"实验完成，共 {len(experiment.runs)} 次运行")

"""
多Agent协作应用的追踪示例
"""
from phoenix.otel import register

tracer_provider = register(
    project_name="multi-agent-app",
    auto_instrument=True,
)
tracer = tracer_provider.get_tracer(__name__)

@tracer.agent
def research_agent(topic: str) -> str:
    """研究Agent：收集信息"""
    return f"关于'{topic}'的研究报告：..."

@tracer.agent
def writing_agent(research: str) -> str:
    """写作Agent：基于研究撰写文章"""
    return f"基于研究撰写的文章：{research[:50]}..."

@tracer.agent
def review_agent(article: str) -> str:
    """审稿Agent：审核文章质量"""
    return "审核通过，建议微调第二段。"

@tracer.chain
def content_creation_pipeline(topic: str) -> str:
    """内容创作管道：研究 -> 写作 -> 审核"""
    research = research_agent(topic)
    article = writing_agent(research)
    review = review_agent(article)
    return f"最终内容: {article}\n审核意见: {review}"

# 运行多Agent管道
result = content_creation_pipeline("人工智能在医疗领域的应用")
# Phoenix UI中可以看到层级追踪：
# content_creation_pipeline(chain)
#   ├── research_agent(agent)
#   ├── writing_agent(agent)
#   └── review_agent(agent)

from phoenix.otel import register
from opentelemetry.sdk.trace.sampling import TraceIdRatioBased

tracer_provider = register(
    project_name="production-app",
    batch=True,                           # 批量发送
    sampler=TraceIdRatioBased(0.1),        # 10%采样率
    endpoint="http://phoenix-server:6006/v1/traces",
)