Arize Phoenix：AI分析可观测性

# 模拟Span结构span = {    'name': 'rag_query',    'start_time': '2026-07-27T10:00:00',    'end_time': '2026-07-27T10:00:02.5',    'attributes': {        'model': 'gpt-4',        'tokens_in': 1200,        'tokens_out': 300,        'temperature': 0.7    },    'children': [        {'name': 'retrieval', 'duration_ms': 1800},        {'name': 'rerank', 'duration_ms': 200},        {'name': 'generation', 'duration_ms': 500}    ]} # 定位耗时瓶颈for child in span['children']:    print(f"{child['name']}: {child['duration_ms']}ms")

import numpy as npfrom sklearn.manifold import TSNEimport matplotlib.pyplot as plt # 模拟文档Embedding（128维）np.random.seed(42)normal_docs = np.random.randn(80, 128)anomaly_docs = np.random.randn(5, 128) + 5  # 偏离主区域 embeddings = np.vstack([normal_docs, anomaly_docs])labels = ['normal'] * 80 + ['anomaly'] * 5 # t-SNE降维到2Dtsne = TSNE(n_components=2, random_state=42, perplexity=20)proj = tsne.fit_transform(embeddings) # 可视化plt.figure(figsize=(6, 5))for label, color in [('normal', 'blue'), ('anomaly', 'red')]:    mask = [l == label for l in labels]    plt.scatter(proj[mask, 0], proj[mask, 1], c=color, label=label, alpha=0.6)plt.legend()plt.title('Document Embedding Space')plt.show()

# 模拟LLM评估指标evaluations = [    {'trace_id': 't1', 'relevance': 0.92, 'factuality': 0.85, 'toxicity': 0.01},    {'trace_id': 't2', 'relevance': 0.78, 'factuality': 0.60, 'toxicity': 0.05},    {'trace_id': 't3', 'relevance': 0.88, 'factuality': 0.90, 'toxicity': 0.00},] # 检测低质量调用for ev in evaluations:    if ev['factuality'] < 0.7:        print(f"告警: {ev['trace_id']} 事实准确性仅{ev['factuality']}") # 计算平均指标avg_rel = sum(e['relevance'] for e in evaluations) / len(evaluations)print(f'平均相关性: {avg_rel:.2f}')

# Phoenix与LangChain集成示例（伪代码）import phoenix as pxfrom phoenix.trace.langchain import LangChainInstrumentor # 自动instrument LangChainLangChainInstrumentor().instrument() # 后续LangChain调用自动被追踪# chain.invoke({'query': 'What is RAG?'})# 追踪数据自动发送到Phoenix服务端 print('Phoenix instrumentation enabled')

import numpy as npfrom sklearn.manifold import TSNEimport matplotlib.pyplot as plt # 1. 模拟文档Embedding数据np.random.seed(42)normal_docs = np.random.randn(100, 128)  # 正常文档anomaly_docs = np.random.randn(8, 128) + 6  # 异常离群文档embeddings = np.vstack([normal_docs, anomaly_docs])labels = ['normal'] * 100 + ['anomaly'] * 8 # 2. t-SNE降维可视化tsne = TSNE(n_components=2, random_state=42, perplexity=30)proj = tsne.fit_transform(embeddings) plt.figure(figsize=(8, 6))for label, color in [('normal', '#3498db'), ('anomaly', '#e74c3c')]:    mask = [l == label for l in labels]    plt.scatter(proj[mask, 0], proj[mask, 1], c=color, label=label, alpha=0.6, s=50)plt.legend()plt.title('Phoenix-style Embedding Visualization')plt.xlabel('t-SNE Dimension 1')plt.ylabel('t-SNE Dimension 2')plt.show() # 3. 模拟Span性能分析spans = [    {'name': 'retrieval', 'duration_ms': 1800, 'tokens': 1200},    {'name': 'rerank', 'duration_ms': 200, 'tokens': 50},    {'name': 'generation', 'duration_ms': 500, 'tokens': 300}]total_time = sum(s['duration_ms'] for s in spans)print('\nSpan性能分析:')for s in spans:    pct = s['duration_ms'] / total_time * 100    print(f"  {s['name']}: {s['duration_ms']}ms ({pct:.1f}%)") # 4. 模拟评估指标监控eval_results = [    {'trace_id': 't1', 'relevance': 0.92, 'factuality': 0.85},    {'trace_id': 't2', 'relevance': 0.78, 'factuality': 0.60},    {'trace_id': 't3', 'relevance': 0.88, 'factuality': 0.90},]print('\n质量评估结果:')for ev in eval_results:    status = 'PASS' if ev['factuality'] >= 0.7 else 'FAIL'    print(f"  {ev['trace_id']}: relevance={ev['relevance']:.2f}, factuality={ev['factuality']:.2f} [{status}]") # 5. 检测异常指标avg_fact = sum(e['factuality'] for e in eval_results) / len(eval_results)print(f"\n平均事实准确性: {avg_fact:.2f}")if avg_fact < 0.75:    print('警告: 整体事实准确性偏低，建议检查知识库质量')