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( model=model, operation=operation ).observe(duration) # 提取 Token 使用量（如果有） if hasattr(result, 'usage'): usage = result.usage LLM_TOKEN_USAGE.labels( model=model, type="input" ).inc(usage.prompt_tokens) LLM_TOKEN_USAGE.labels( model=model, type="output" ).inc(usage.completion_tokens) # 计算成本 cost = self._calc_cost(model, usage) LLM_COST_TOTAL.labels(model=model).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_id=request_id, model=model, duration=duration, operation=operation ) return result except Exception as e: ERROR_COUNT.labels( model=model, error_type=type(e).__name__ ).inc() span.record_exception(e) self.logger.error( "llm_request_error", request_id=request_id, error=str(e), error_type=type(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(model="gpt-4o", operation="chat_completion", use_case="customer_service")async def chat_with_customer(messages: list): return await openai_client.chat.completions.create( model="gpt-4o", messages=messages )## 分布式链路追踪对于复杂的 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(endpoint="http://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(maxlen=window_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( current=recent_avg, baseline=self.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 可观测性体系包含四个层次：指标层（Metrics）：Prometheus + Grafana，监控性能、成本、错误率 追踪层（Tracing）：OpenTelemetry + Jaeger，追踪复杂链路 日志层（Logging）：结构化日志，记录每次请求的完整上下文 质量层（Quality）：自动化评估，持续监控输出质量趋势 从零到完整体系，推荐三个阶段：1. 先建立成本监控（防止意外费用）2. 再接入性能追踪（排查慢请求）3. 最后建立质量评估（持续改进输出）可观测性投入是 LLM 应用成熟度的重要标志，它让你从"出了问题才知道"变为"问题发生时已知道"。