AI应用可观测性工程：像监控微服务一样监控你的LLM应用

LLM 应用进入生产后“为什么这次回答质量差”、哪次调用导致成本飙升这些问题如果没有完整的可观测性体系根本无法回答。本文构建 LLM 应用的完整监控体系。LLM 应用监控的独特挑战传统微服务监控关注的是响应时间、错误率、吞吐量。这些对 LLM 应用同样重要但 LLM 应用还有其特殊性非确定性输出同样的输入每次输出可能不同无法用传统方式对比正确性。Token 成本计量每次调用的成本与输入输出 Token 数直接相关需要细粒度追踪。长链路调用一个用户请求可能触发多次 LLM 调用RAG 查询 生成 验证需要追踪完整链路。输出质量评估不像 HTTP 200/500 那样黑白分明LLM 输出质量是连续变量需要专门的评估机制。## 核心监控指标体系### 性能指标pythonfrom prometheus_client import Histogram, Counter, Gauge, Summaryimport time# LLM 专用指标LLM_REQUEST_DURATION Histogram( llm_request_duration_seconds, Total duration of LLM requests, [model, operation], # 按模型和操作类型标注 buckets[0.5, 1.0, 2.5, 5.0, 10.0, 30.0, 60.0])LLM_TTFT Histogram( llm_ttft_seconds, Time to first token, [model], buckets[0.1, 0.25, 0.5, 1.0, 2.5, 5.0])LLM_TOKEN_USAGE Counter( llm_tokens_total, Total tokens consumed, [model, type] # type: input/output)LLM_COST_TOTAL Counter( llm_cost_usd_total, Total cost in USD, [model])LLM_QUALITY_SCORE Histogram( llm_quality_score, Quality score of LLM responses (0-1), [model, use_case], buckets[0.1, 0.2, 0.4, 0.6, 0.7, 0.8, 0.9, 1.0])ERROR_COUNT Counter( llm_errors_total, Total number of LLM errors, [model, error_type])### 统一的可观测性装饰器pythonimport functoolsimport uuidfrom contextlib import asynccontextmanagerfrom typing import AsyncGeneratorclass LLMObservabilityMiddleware: def __init__(self, tracer, logger, metrics_client): self.tracer tracer self.logger logger self.metrics metrics_client def observe(self, model: str, operation: str, use_case: str default): 装饰器为 LLM 调用添加完整的可观测性 def decorator(func): functools.wraps(func) async def wrapper(*args, **kwargs): request_id str(uuid.uuid4()) start_time time.time() # 开始 Span with self.tracer.start_as_current_span( f{operation}, attributes{ llm.model: model, llm.operation: operation, llm.request_id: request_id, llm.use_case: use_case, } ) as span: try: result await func(*args, **kwargs) duration time.time() - start_time # 记录成功指标 LLM_REQUEST_DURATION.labels( modelmodel, operationoperation ).observe(duration) # 提取 Token 使用量如果有 if hasattr(result, usage): usage result.usage LLM_TOKEN_USAGE.labels( modelmodel, typeinput ).inc(usage.prompt_tokens) LLM_TOKEN_USAGE.labels( modelmodel, typeoutput ).inc(usage.completion_tokens) # 计算成本 cost self._calc_cost(model, usage) LLM_COST_TOTAL.labels(modelmodel).inc(cost) span.set_attribute(llm.input_tokens, usage.prompt_tokens) span.set_attribute(llm.output_tokens, usage.completion_tokens) span.set_attribute(llm.cost_usd, cost) self.logger.info( llm_request_success, request_idrequest_id, modelmodel, durationduration, operationoperation ) return result except Exception as e: ERROR_COUNT.labels( modelmodel, error_typetype(e).__name__ ).inc() span.record_exception(e) self.logger.error( llm_request_error, request_idrequest_id, errorstr(e), error_typetype(e).__name__ ) raise return wrapper return decorator def _calc_cost(self, model: str, usage) - float: 根据模型和Token数计算成本 PRICING { gpt-4o: {input: 0.005 / 1000, output: 0.015 / 1000}, gpt-4o-mini: {input: 0.00015 / 1000, output: 0.0006 / 1000}, claude-3-5-sonnet: {input: 0.003 / 1000, output: 0.015 / 1000}, } pricing PRICING.get(model, {input: 0, output: 0}) return (usage.prompt_tokens * pricing[input] usage.completion_tokens * pricing[output])# 使用示例middleware LLMObservabilityMiddleware(tracer, logger, metrics)middleware.observe(modelgpt-4o, operationchat_completion, use_casecustomer_service)async def chat_with_customer(messages: list): return await openai_client.chat.completions.create( modelgpt-4o, messagesmessages )## 分布式链路追踪对于复杂的 RAG 或 Multi-Agent 系统需要追踪完整的调用链pythonfrom opentelemetry import tracefrom opentelemetry.sdk.trace import TracerProviderfrom opentelemetry.sdk.trace.export import BatchSpanProcessorfrom opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter# 初始化 OpenTelemetryprovider TracerProvider()provider.add_span_processor( BatchSpanProcessor(OTLPSpanExporter(endpointhttp://jaeger:4317)))trace.set_tracer_provider(provider)tracer trace.get_tracer(llm-app)async def rag_pipeline_with_tracing(user_query: str, user_id: str): 完整的 RAG 管道带完整链路追踪 with tracer.start_as_current_span(rag_pipeline) as root_span: root_span.set_attribute(user.id, user_id) root_span.set_attribute(query.text, user_query[:200]) # 1. 检索阶段 with tracer.start_as_current_span(retrieval) as retrieval_span: docs await retrieve_documents(user_query) retrieval_span.set_attribute(retrieval.doc_count, len(docs)) retrieval_span.set_attribute(retrieval.top_score, docs[0].score if docs else 0) # 2. 重排序阶段 with tracer.start_as_current_span(reranking) as rerank_span: reranked_docs await rerank_documents(user_query, docs) rerank_span.set_attribute(reranking.kept_count, len(reranked_docs)) # 3. 生成阶段 with tracer.start_as_current_span(generation) as gen_span: response await generate_answer(user_query, reranked_docs) gen_span.set_attribute(generation.model, gpt-4o) gen_span.set_attribute(generation.output_tokens, response.usage.completion_tokens) return response## LLM Ops智能异常检测pythonimport numpy as npfrom collections import dequeclass LLMQualityMonitor: def __init__(self, window_size: int 100, alert_threshold: float 0.7): self.quality_window deque(maxlenwindow_size) self.alert_threshold alert_threshold self.baseline_quality None def record_quality_score(self, score: float, metadata: dict None): 记录单次质量评分 self.quality_window.append({ score: score, timestamp: time.time(), metadata: metadata or {} }) # 建立基线前100次 if len(self.quality_window) 100 and self.baseline_quality is None: scores [r[score] for r in list(self.quality_window)[:100]] self.baseline_quality np.mean(scores) # 检测质量下降 if self.baseline_quality and len(self.quality_window) 20: recent_scores [r[score] for r in list(self.quality_window)[-20:]] recent_avg np.mean(recent_scores) if recent_avg self.baseline_quality * self.alert_threshold: self._trigger_quality_alert( currentrecent_avg, baselineself.baseline_quality ) def _trigger_quality_alert(self, current: float, baseline: float): 触发质量告警 degradation_pct (1 - current / baseline) * 100 print(f⚠️ LLM质量告警相比基线下降 {degradation_pct:.1f}%) print(f 基线均值: {baseline:.3f}) print(f 当前均值: {current:.3f}) # 发送告警到监控系统 # alert_manager.send_alert(...)## 成本异常检测pythonclass CostAnomalyDetector: 检测异常的 LLM 调用成本 def __init__(self): self.hourly_costs {} self.daily_budget 100.0 # USD self.hourly_alert_threshold 10.0 # USD def record_cost(self, cost_usd: float, model: str, user_id: str): hour_key int(time.time() / 3600) if hour_key not in self.hourly_costs: self.hourly_costs[hour_key] {} user_key f{user_id}:{model} self.hourly_costs[hour_key][user_key] ( self.hourly_costs[hour_key].get(user_key, 0) cost_usd ) # 检查小时预算 total_hour_cost sum(self.hourly_costs[hour_key].values()) if total_hour_cost self.hourly_alert_threshold: print(f⚠️ 小时成本告警{total_hour_cost:.2f} USD) # 检测单个用户的异常高消耗 user_hour_cost self.hourly_costs[hour_key].get(user_key, 0) if user_hour_cost 5.0: # 单用户每小时超5美元 print(f⚠️ 用户异常消耗告警{user_id} 在过去1小时花费 {user_hour_cost:.2f} USD)## Grafana Dashboard 配置yaml# grafana-dashboard.yaml关键面板配置panels: - title: LLM请求延迟P50/P95/P99 type: graph targets: - expr: histogram_quantile(0.5, llm_request_duration_seconds_bucket) legend: P50 - expr: histogram_quantile(0.95, llm_request_duration_seconds_bucket) legend: P95 - expr: histogram_quantile(0.99, llm_request_duration_seconds_bucket) legend: P99 - title: Token消耗趋势 type: graph targets: - expr: rate(llm_tokens_total[5m]) * 300 legend: Token/5min - title: 每小时成本USD type: stat targets: - expr: increase(llm_cost_usd_total[1h]) - title: 输出质量评分分布 type: histogram targets: - expr: llm_quality_score_bucket - title: 错误率 type: graph targets: - expr: rate(llm_errors_total[5m]) legend: {{error_type}}## 告警规则yaml# alertmanager-rules.yamlgroups: - name: llm-application rules: - alert: LLMHighLatency expr: histogram_quantile(0.95, llm_request_duration_seconds_bucket) 10 for: 5m labels: severity: warning annotations: summary: LLM P95延迟超过10秒 description: 过去5分钟P95延迟为 {{ $value }}s - alert: LLMHighErrorRate expr: rate(llm_errors_total[5m]) / rate(llm_request_duration_seconds_count[5m]) 0.05 for: 2m labels: severity: critical annotations: summary: LLM错误率超过5% - alert: LLMCostBudgetAlert expr: increase(llm_cost_usd_total[1h]) 50 labels: severity: warning annotations: summary: 1小时LLM成本超过50美元## 生产监控体系总结完整的 LLM 可观测性体系包含四个层次指标层MetricsPrometheus Grafana监控性能、成本、错误率追踪层TracingOpenTelemetry Jaeger追踪复杂链路日志层Logging结构化日志记录每次请求的完整上下文质量层Quality自动化评估持续监控输出质量趋势 从零到完整体系推荐三个阶段1. 先建立成本监控防止意外费用2. 再接入性能追踪排查慢请求3. 最后建立质量评估持续改进输出可观测性投入是 LLM 应用成熟度的重要标志它让你从出了问题才知道变为问题发生时已知道。