引言
LLM应用的一个核心痛点是不可观测性：当应用在生产环境中出现问题时，工程师往往不知道是哪个Prompt出了问题、哪次检索质量太差、还是哪次工具调用超时。传统的监控体系完全失效。2026年，以LangSmith、Langfuse、Helicone为代表的LLM可观测性平台已经成熟，结合OpenTelemetry标准，"LLM可观测性"已成为每个AI应用的标配基础设施。本文系统讲解如何为LLM应用构建完整的可观测性体系。—## 一、LLM可观测性的核心维度### 传统APM vs LLM可观测性| 维度 | 传统APM | LLM可观测性 ||------|--------|------------|| 响应时间 | ✅ 毫秒级延迟 | ✅ + TTFT（首Token时间） || 错误率 | ✅ HTTP 4xx/5xx | ✅ + 语义错误（幻觉、拒绝回答） || 吞吐量 | ✅ QPS | ✅ + Token/s || 成本 | ❌ 不适用 | ✅ 每次请求的Token消耗和费用 || 质量 | ❌ 不适用 | ✅ 答案准确性、相关性、有用性 || 上下文 | ❌ 不适用 | ✅ Prompt、完整对话链路 |—## 二、基础Tracing集成### 2.1 使用Langfuse（开源，可私有化）pythonfrom langfuse import Langfusefrom langfuse.decorators import observe, langfuse_context# 初始化Langfuse（可以私有化部署）langfuse = Langfuse( public_key="pk-xxx", secret_key="sk-xxx", host="http://localhost:3000" # 私有化部署地址)# 使用装饰器自动追踪@observe()async def answer_question(question: str, user_id: str) -> str: # 自动追踪：输入、输出、Token消耗、延迟 # 添加用户级别的标签 langfuse_context.update_current_trace( user_id=user_id, tags=["question-answering", "production"] ) # RAG检索 context = await retrieve_context(question) # LLM调用 response = await llm.chat( messages=[ {"role": "system", "content": "你是一个专业助手"}, {"role": "user", "content": f"基于以下内容回答问题：\n{context}\n\n问题：{question}"} ] ) # 添加自定义评分（如用户反馈后回调） langfuse_context.score_current_trace( name="relevance", value=0.9, # 可以后续通过反馈更新 comment="初始自动评分" ) return response# 使用示例@observe(name="rag-pipeline")async def rag_pipeline(query: str) -> str: """完整RAG流水线的追踪""" # Span 1: 文档检索 with langfuse.span(name="retrieval") as span: docs = await vectorstore.search(query, k=5) span.update( input=query, output=f"检索到 {len(docs)} 个文档", metadata={"num_docs": len(docs), "scores": [d.score for d in docs]} ) # Span 2: 上下文构建 context = "\n\n".join([d.content for d in docs]) # Span 3: LLM生成 with langfuse.generation( name="llm-generation", model="gpt-4o", model_parameters={"temperature": 0.3} ) as generation: response = await openai_client.chat.completions.create( model="gpt-4o", messages=[ {"role": "user", "content": f"{context}\n\n{query}"} ] ) generation.update( input=f"{context}\n\n{query}", output=response.choices[0].message.content, usage={ "input": response.usage.prompt_tokens, "output": response.usage.completion_tokens, "unit": "TOKENS" } ) return response.choices[0].message.content### 2.2 OpenTelemetry集成（标准化方案）pythonfrom opentelemetry import tracefrom opentelemetry.sdk.trace import TracerProviderfrom opentelemetry.sdk.trace.export import BatchSpanProcessorfrom opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporterfrom opentelemetry.instrumentation.openai import OpenAIInstrumentor# 初始化TracerProviderprovider = TracerProvider()provider.add_span_processor( BatchSpanProcessor( OTLPSpanExporter(endpoint="http://jaeger:4317") ))trace.set_tracer_provider(provider)# 自动注入OpenAI追踪OpenAIInstrumentor().instrument()# 自定义Tracertracer = trace.get_tracer("llm-app")async def process_request(user_query: str, user_id: str) -> str: with tracer.start_as_current_span("process_request") as span: span.set_attribute("user.id", user_id) span.set_attribute("query.length", len(user_query)) # 检索（自动追踪） with tracer.start_as_current_span("retrieval") as retrieval_span: docs = await search(user_query) retrieval_span.set_attribute("retrieval.num_docs", len(docs)) # LLM调用（OpenAI SDK自动追踪，包含Token数） response = await openai_client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": user_query}] ) span.set_attribute("llm.tokens_used", response.usage.total_tokens) return response.choices[0].message.content—## 三、LLM质量评估体系### 3.1 自动化评估指标pythonfrom langfuse import Langfusefrom openai import OpenAIclass LLMQualityEvaluator: """LLM应用的自动化质量评估""" def __init__(self, langfuse: Langfuse, judge_llm: OpenAI): self.langfuse = langfuse self.judge = judge_llm async def evaluate_rag_response( self, trace_id: str, question: str, context: str, answer: str ): """评估RAG回答质量""" # 指标1：答案与上下文的一致性（幻觉检测） faithfulness = await self._evaluate_faithfulness(answer, context) # 指标2：答案与问题的相关性 relevance = await self._evaluate_relevance(question, answer) # 指标3：上下文检索质量 context_recall = await self._evaluate_context_recall(question, context) # 上报到Langfuse for name, value in [ ("faithfulness", faithfulness), ("answer_relevance", relevance), ("context_recall", context_recall) ]: self.langfuse.score( trace_id=trace_id, name=name, value=value, data_type="NUMERIC" ) return { "faithfulness": faithfulness, "relevance": relevance, "context_recall": context_recall, "overall": (faithfulness + relevance + context_recall) / 3 } async def _evaluate_faithfulness(self, answer: str, context: str) -> float: """使用LLM-as-Judge评估答案忠实度""" prompt = f"""评估以下回答是否完全基于给定的上下文，不包含任何无中生有的信息。上下文：{context}回答：{answer}评分标准：1.0 = 完全忠实，没有任何幻觉0.5 = 基本忠实，有轻微超出上下文的内容0.0 = 存在明显幻觉或与上下文矛盾请只返回一个0到1之间的数字，不要解释。""" response = await self.judge.chat.completions.create( model="gpt-4o-mini", # 用小模型做评判，降低成本 messages=[{"role": "user", "content": prompt}], temperature=0 ) try: return float(response.choices[0].message.content.strip()) except: return 0.5 # 解析失败时的默认值### 3.2 实时告警规则pythonclass LLMAlertSystem: """LLM应用实时告警""" def __init__(self, alert_client, metrics_store): self.alert = alert_client self.metrics = metrics_store # 告警阈值配置 self.thresholds = { "error_rate": 0.05, # 错误率 > 5% 告警 "avg_latency_ms": 3000, # 平均延迟 > 3秒 "p99_latency_ms": 10000, # P99延迟 > 10秒 "token_cost_hourly": 100, # 每小时Token费用 > $100 "faithfulness_score": 0.7, # 忠实度评分 < 0.7 "refusal_rate": 0.1 # 拒绝回答率 > 10% } async def check_and_alert(self, window_minutes: int = 5): """定期检查指标并触发告警""" metrics = await self.metrics.get_recent(window_minutes) alerts = [] if metrics["error_rate"] > self.thresholds["error_rate"]: alerts.append({ "severity": "critical", "message": f"LLM错误率过高: {metrics['error_rate']:.1%}", "value": metrics["error_rate"] }) if metrics["avg_latency_ms"] > self.thresholds["avg_latency_ms"]: alerts.append({ "severity": "warning", "message": f"LLM平均延迟过高: {metrics['avg_latency_ms']}ms", "value": metrics["avg_latency_ms"] }) if metrics.get("faithfulness_score", 1.0) < self.thresholds["faithfulness_score"]: alerts.append({ "severity": "warning", "message": f"回答质量下降，忠实度评分: {metrics['faithfulness_score']:.2f}", "value": metrics["faithfulness_score"] }) for alert in alerts: await self.alert.send(alert) return alerts—## 四、成本可观测性pythonclass TokenCostTracker: """Token消耗和成本追踪""" # 2026年5月价格（$/1M Token） PRICING = { "gpt-4o": {"input": 2.50, "output": 10.00}, "gpt-4o-mini": {"input": 0.15, "output": 0.60}, "claude-3-7-sonnet": {"input": 3.00, "output": 15.00}, "deepseek-v4": {"input": 0.27, "output": 1.10}, "kimi-k2-6": {"input": 0.50, "output": 2.00}, } def calculate_cost( self, model: str, input_tokens: int, output_tokens: int ) -> float: """计算单次请求成本（美元）""" if model not in self.PRICING: return 0.0 pricing = self.PRICING[model] cost = ( input_tokens * pricing["input"] / 1_000_000 + output_tokens * pricing["output"] / 1_000_000 ) return cost async def generate_cost_report( self, start_date: str, end_date: str ) -> dict: """生成成本报告""" records = await self.db.query( "SELECT model, sum(input_tokens), sum(output_tokens), count(*) " "FROM llm_calls WHERE date BETWEEN ? AND ? GROUP BY model", [start_date, end_date] ) report = { "period": f"{start_date} ~ {end_date}", "by_model": {}, "total_cost": 0.0, "total_requests": 0 } for row in records: model, input_t, output_t, count = row cost = self.calculate_cost(model, input_t, output_t) report["by_model"][model] = { "requests": count, "input_tokens": input_t, "output_tokens": output_t, "cost_usd": round(cost, 4) } report["total_cost"] += cost report["total_requests"] += count report["total_cost"] = round(report["total_cost"], 2) return report—## 五、Prompt版本管理与A/B测试pythonclass PromptVersionManager: """管理Prompt版本并进行A/B测试""" async def ab_test( self, prompt_a: str, prompt_b: str, test_queries: list, evaluation_fn: callable, traffic_split: float = 0.5 ) -> dict: """运行Prompt A/B测试""" results = {"a": [], "b": []} for query in test_queries: # 随机分配流量 variant = "a" if random.random() < traffic_split else "b" prompt = prompt_a if variant == "a" else prompt_b response = await self.llm.chat( messages=[ {"role": "system", "content": prompt}, {"role": "user", "content": query} ] ) # 评估质量 score = await evaluation_fn(query, response) results[variant].append(score) # 统计显著性检验 from scipy import stats t_stat, p_value = stats.ttest_ind(results["a"], results["b"]) return { "variant_a": { "mean_score": sum(results["a"]) / len(results["a"]), "sample_size": len(results["a"]) }, "variant_b": { "mean_score": sum(results["b"]) / len(results["b"]), "sample_size": len(results["b"]) }, "statistical_significance": p_value < 0.05, "p_value": p_value, "winner": "a" if sum(results["a"]) > sum(results["b"]) else "b" }—## 总结LLM可观测性体系的建设分三个层次：基础层：Tracing + 日志 + 指标- 接入Langfuse或LangSmith- 追踪每次LLM调用的输入输出和Token消耗- 监控延迟、错误率、成本质量层：自动化评估- 幻觉检测、相关性评估- LLM-as-Judge的自动化评分- 实时质量告警优化层：A/B测试 + 持续改进- Prompt版本管理- 数据驱动的优化决策- 成本效益分析在2026年，没有可观测性基础设施的LLM应用就像在黑暗中飞行——出问题时无法诊断，优化时无从下手。这是从Demo到生产级应用的必经之路。
