监控

AI 应用为什么需要专门的监控 传统 APM（应用性能监控）关注的是 CPU、内存、响应时间和错误率。但 AI 应用引入了全新的维度： Token 维度：每次请求消耗的 Token 数直接影响成本 模型质量：响应不仅要快，还要准确、安全、无幻觉 长尾延迟：流式输出可能持续数分钟，P99 指标失真 多模型路由：请求分散在不同模型上，需要统一视图 如果你的监控面板还只有"QPS + 延迟 + 错误率"三件套，你实际上是在"盲飞"。 监控指标体系 四层指标架构 ┌─────────────────────────────────────────────┐ │ 业务指标层 (Business) │ │ 任务完成率 · 用户满意度 · 成本/请求 │ ├─────────────────────────────────────────────┤ │ AI 质量指标层 (Quality) │ │ 幻觉率 · 意图准确率 · 安全违规率 │ ├─────────────────────────────────────────────┤ │ 模型性能指标层 (Model) │ │ 首 Token 延迟 · 生成速率 · KV Cache 命中率 │ ├─────────────────────────────────────────────┤ │ 基础设施指标层 (Infra) │ │ GPU 利用率 · 显存使用 · CPU/内存 · 网络 │ └─────────────────────────────────────────────┘ 完整指标清单 层级 指标名 类型 说明 告警阈值 Infra gpu_utilization Gauge GPU 计算利用率 <20% 或 >95% Infra gpu_memory_used_ratio Gauge GPU 显存使用率 >0.95 Infra gpu_temperature Gauge GPU 温度 (°C) >85 Model ttft_seconds Histogram 首 Token 延迟 (TTFT) P95 >2s Model tokens_per_second Gauge Token 生成速率 <30 tokens/s Model kv_cache_hit_rate Gauge KV Cache 命中率 <0.3 Model active_sequences Gauge 活跃推理序列数 >max_seqs×0.9 Model queue_depth Gauge 排队请求数 >10 Quality hallucination_score Gauge 幻觉评分 (0-1) >0.15 Quality safety_violation_count Counter 安全违规次数 >0/h Quality user_feedback_score Histogram 用户评分 (1-5) 均值 <3.5 Business cost_per_request Gauge 单请求成本 ($) >$0.05 Business task_completion_rate Gauge 任务完成率 <80% Business daily_spend Counter 日累计花费 >预算 80% Prometheus 指标采集 应用侧指标暴露 from prometheus_client import Counter, Histogram, Gauge, generate_latest from prometheus_client import CollectorRegistry, CONTENT_TYPE_LATEST from fastapi import FastAPI, Response import time app = FastAPI() registry = CollectorRegistry() # === 基础设施指标 === gpu_utilization = Gauge( "ai_gpu_utilization", "GPU utilization percentage", ["gpu_id", "model_name"], registry=registry ) gpu_memory_used = Gauge( "ai_gpu_memory_used_bytes", "GPU memory used in bytes", ["gpu_id"], registry=registry ) # === 模型性能指标 === request_duration = Histogram( "ai_request_duration_seconds", "Total request duration", ["model_name", "endpoint"], buckets=[0.5, 1, 2, 5, 10, 30, 60, 120, 300], registry=registry, ) ttft = Histogram( "ai_time_to_first_token_seconds", "Time to first token", ["model_name"], buckets=[0.1, 0.25, 0.5, 1, 2, 5, 10], registry=registry, ) tokens_generated = Counter( "ai_tokens_generated_total", "Total tokens generated", ["model_name", "type"], # type: input|output registry=registry, ) # === 质量指标 === hallucination_score = Gauge( "ai_hallucination_score", "Hallucination score (0=good, 1=bad)", ["model_name"], registry=registry, ) safety_violations = Counter( "ai_safety_violations_total", "Safety violation count", ["model_name", "violation_type"], registry=registry, ) # === 业务指标 === request_cost = Gauge( "ai_request_cost_usd", "Cost per request in USD", ["model_name"], registry=registry, ) daily_spend = Counter( "ai_daily_spend_usd", "Daily total spend in USD", ["model_name"], registry=registry, ) # === 中间件：自动采集请求指标 === @app.middleware("http") async def metrics_middleware(request, call_next): start_time = time.time() response = await call_next(request) duration = time.time() - start_time model = request.headers.get("X-Model-Name", "unknown") request_duration.labels( model_name=model, endpoint=request.url.path, ).observe(duration) return response # === 业务逻辑中手动埋点 === async def chat_completion(messages, model="gpt-4o-mini"): start = time.time() # 调用 LLM response = await llm_client.chat.completions.create( model=model, messages=messages, stream=True, stream_options={"include_usage": True}, ) first_token_time = None total_output_tokens = 0 async for chunk in response: if first_token_time is None and chunk.choices[0].delta.content: first_token_time = time.time() ttft.labels(model_name=model).observe(first_token_time - start) if chunk.usage: total_output_tokens = chunk.usage.completion_tokens input_tokens = chunk.usage.prompt_tokens tokens_generated.labels(model_name=model, type="input").inc(input_tokens) tokens_generated.labels(model_name=model, type="output").inc(total_output_tokens) # 计算成本 cost = calculate_cost(model, input_tokens, total_output_tokens) request_cost.labels(model_name=model).set(cost) daily_spend.labels(model_name=model).inc(cost) return response # === 异步质量评估 === async def evaluate_quality(response: str, context: str, model: str): """异步评估响应质量""" # 幻觉检测 score = await hallucination_detector.check(response, context) hallucination_score.labels(model_name=model).set(score) # 安全检查 violations = safety_checker.check(response) for v_type, count in violations.items(): safety_violations.labels( model_name=model, violation_type=v_type, ).inc(count) # === Prometheus 指标暴露端点 === @app.get("/metrics") async def metrics(): return Response( generate_latest(registry), media_type=CONTENT_TYPE_LATEST, ) GPU 指标采集（DCGM Exporter） # gpu-metrics-stack.yaml apiVersion: v1 kind: Namespace metadata: name: gpu-monitoring --- apiVersion: apps/v1 kind: DaemonSet metadata: name: dcgm-exporter namespace: gpu-monitoring spec: selector: matchLabels: app: dcgm-exporter template: metadata: labels: app: dcgm-exporter spec: nodeSelector: accelerator: nvidia-t4 # 调度到 GPU 节点 tolerations: - key: nvidia.com/gpu operator: Exists containers: - name: dcgm-exporter image: nvcr.io/nvidia/k8s/dcgm-exporter:3.3.0-ubuntu22.04 ports: - containerPort: 9400 name: metrics env: - name: DCGM_EXPORTER_LISTEN value: ":9400" - name: DCGM_EXPORTER_KUBERNETES value: "true" securityContext: capabilities: add: ["SYS_ADMIN"] --- # ServiceMonitor 让 Prometheus 自动发现 apiVersion: monitoring.coreos.com/v1 kind: ServiceMonitor metadata: name: dcgm-exporter namespace: gpu-monitoring spec: selector: matchLabels: app: dcgm-exporter endpoints: - port: metrics path: "/metrics" interval: 10s DCGM 关键 GPU 指标 指标 说明 用途 DCGM_FI_DEV_GPU_UTIL GPU 计算利用率 (%) 判断是否充分使用 GPU DCGM_FI_DEV_MEM_COPY_UTIL 显存带宽利用率 (%) 判断是否显存带宽瓶颈 DCGM_FI_DEV_FB_USED 已用显存 (MB) 判断显存是否够用 DCGM_FI_DEV_FB_FREE 空闲显存 (MB) 可调度的余量 DCGM_FI_DEV_GPU_TEMP GPU 温度 (°C) 散热健康 DCGM_FI_DEV_POWER_USAGE 功耗 (W) 能耗成本 DCGM_FI_PROF_PIPE_TENSOR_ACTIVE Tensor Core 利用率 推理效率 Grafana 仪表盘构建 仪表盘布局设计 ┌──────────────────────────────────────────────────────────┐ │ AI Application Overview │ ├────────────┬────────────┬────────────┬───────────────────┤ │ QPS │ P95 TTFT │ Error Rate│ Daily Spend │ │ (Stat) │ (Stat) │ (Stat) │ (Stat) │ ├────────────┴────────────┼────────────┴───────────────────┤ │ Request Latency │ Token Throughput │ │ (Time Series) │ (Time Series) │ ├──────────────────────────┼───────────────────────────────┤ │ GPU Utilization │ GPU Memory Usage │ │ (Time Series, per GPU) │ (Time Series, per GPU) │ ├──────────────────────────┼───────────────────────────────┤ │ Model Distribution │ Queue Depth │ │ (Pie Chart) │ (Time Series) │ ├──────────────────────────┼───────────────────────────────┤ │ Hallucination Score │ Safety Violations │ │ (Gauge + Time Series) │ (Alert Table) │ ├──────────────────────────┴───────────────────────────────┤ │ Request Logs (Table) │ └──────────────────────────────────────────────────────────┘ Grafana Dashboard JSON { "dashboard": { "title": "AI Application Monitor", "tags": ["ai", "llm", "production"], "timezone": "browser", "refresh": "10s", "panels": [ { "id": 1, "title": "Requests Per Second", "type": "stat", "gridPos": {"h": 4, "w": 6, "x": 0, "y": 0}, "targets": [{ "expr": "sum(rate(ai_request_duration_seconds_count[1m]))", "legendFormat": "QPS" }], "fieldConfig": { "defaults": { "thresholds": { "steps": [ {"color": "red", "value": null}, {"color": "yellow", "value": 10}, {"color": "green", "value": 50} ] } } } }, { "id": 2, "title": "P95 Time to First Token", "type": "stat", "gridPos": {"h": 4, "w": 6, "x": 6, "y": 0}, "targets": [{ "expr": "histogram_quantile(0.95, sum(rate(ai_time_to_first_token_seconds_bucket[5m])) by (le))", "legendFormat": "P95 TTFT" }], "fieldConfig": { "defaults": { "unit": "s", "thresholds": { "steps": [ {"color": "green", "value": null}, {"color": "yellow", "value": 1}, {"color": "red", "value": 2} ] } } } }, { "id": 3, "title": "Request Latency Distribution", "type": "timeseries", "gridPos": {"h": 8, "w": 12, "x": 0, "y": 4}, "targets": [ { "expr": "histogram_quantile(0.50, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model_name))", "legendFormat": "P50 {{model_name}}" }, { "expr": "histogram_quantile(0.95, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model_name))", "legendFormat": "P95 {{model_name}}" }, { "expr": "histogram_quantile(0.99, sum(rate(ai_request_duration_seconds_bucket[5m])) by (le, model_name))", "legendFormat": "P99 {{model_name}}" } ], "fieldConfig": { "defaults": {"unit": "s"} } }, { "id": 4, "title": "GPU Utilization & Memory", "type": "timeseries", "gridPos": {"h": 8, "w": 12, "x": 12, "y": 4}, "targets": [ { "expr": "DCGM_FI_DEV_GPU_UTIL", "legendFormat": "GPU {{gpu}} Util %" }, { "expr": "DCGM_FI_DEV_FB_USED / DCGM_FI_DEV_FB_TOTAL * 100", "legendFormat": "GPU {{gpu}} Mem %" } ], "fieldConfig": { "defaults": { "min": 0, "max": 100, "custom": {"fillOpacity": 10} } } }, { "id": 5, "title": "Daily Spend by Model", "type": "bargauge", "gridPos": {"h": 6, "w": 12, "x": 0, "y": 12}, "targets": [{ "expr": "sum by (model_name) (increase(ai_daily_spend_usd_total[24h]))", "legendFormat": "{{model_name}}" }], "fieldConfig": { "defaults": { "unit": "currencyUSD", "thresholds": { "steps": [ {"color": "green", "value": null}, {"color": "yellow", "value": 50}, {"color": "red", "value": 100} ] } } } }, { "id": 6, "title": "Hallucination Score Trend", "type": "timeseries", "gridPos": {"h": 6, "w": 12, "x": 12, "y": 12}, "targets": [{ "expr": "avg_over_time(ai_hallucination_score[10m])", "legendFormat": "Hallucination Score" }], "fieldConfig": { "defaults": { "min": 0, "max": 1, "thresholds": { "steps": [ {"color": "green", "value": null}, {"color": "yellow", "value": 0.1}, {"color": "red", "value": 0.2} ] } } } } ] } } 告警体系 多级告警规则 # alerting-rules.yaml groups: - name: ai_infra_alerts interval: 30s rules: # === 基础设施告警 === - alert: GPUHighTemperature expr: DCGM_FI_DEV_GPU_TEMP > 85 for: 5m labels: severity: critical team: infra annotations: summary: "GPU 温度过高: {{ $value }}°C" description: "GPU {{ $labels.gpu }} on {{ $labels.instance }} 温度超过 85°C" - alert: GPUHighMemoryUsage expr: DCGM_FI_DEV_FB_USED / DCGM_FI_DEV_FB_TOTAL > 0.95 for: 2m labels: severity: warning team: infra annotations: summary: "GPU 显存使用率 >95%" - alert: GPULowUtilization expr: avg_over_time(DCGM_FI_DEV_GPU_UTIL[10m]) < 10 for: 10m labels: severity: info team: infra annotations: summary: "GPU 利用率过低 (<10%)，可能资源浪费" - name: ai_model_alerts interval: 15s rules: # === 模型性能告警 === - alert: HighTTFT expr: | histogram_quantile(0.95, sum(rate(ai_time_to_first_token_seconds_bucket[5m])) by (le, model_name) ) > 2 for: 5m labels: severity: warning team: ai annotations: summary: "首 Token 延迟 P95 > 2s ({{ $labels.model_name }})" - alert: HighErrorRate expr: | sum(rate(ai_request_duration_seconds_count{status="error"}[5m])) by (model_name) / sum(rate(ai_request_duration_seconds_count[5m])) by (model_name) > 0.05 for: 3m labels: severity: critical team: ai annotations: summary: "错误率 >5% ({{ $labels.model_name }})" - alert: ModelQueueBacklog expr: ai_queue_depth > 20 for: 2m labels: severity: warning team: ai annotations: summary: "请求排队 >20 ({{ $labels.model_name }})" - name: ai_quality_alerts interval: 60s rules: # === 质量告警 === - alert: HighHallucinationRate expr: avg_over_time(ai_hallucination_score[15m]) > 0.15 for: 10m labels: severity: critical team: ai annotations: summary: "幻觉评分 >0.15 持续 10 分钟" - alert: SafetyViolationDetected expr: increase(ai_safety_violations_total[1h]) > 0 for: 0m labels: severity: critical team: security annotations: summary: "检测到安全违规 ({{ $labels.violation_type }})" - name: ai_business_alerts interval: 60s rules: # === 业务告警 === - alert: DailyBudgetExceeded expr: sum(increase(ai_daily_spend_usd_total[24h])) > 500 for: 1m labels: severity: critical team: business annotations: summary: "日预算超出 (${{ $value }})" - alert: CostPerRequestSpike expr: ai_request_cost_usd > 0.10 for: 5m labels: severity: warning team: business annotations: summary: "单请求成本 >$0.10 ({{ $labels.model_name }})" 告警通知渠道配置 # alertmanager-config.yaml apiVersion: v1 kind: ConfigMap metadata: name: alertmanager-config namespace: monitoring data: config.yml: | route: group_by: ["alertname", "model_name"] group_wait: 30s group_interval: 5m repeat_interval: 4h receiver: "default" routes: - matchers: ["severity=critical"] receiver: "critical" group_wait: 10s repeat_interval: 1h - matchers: ["team=security"] receiver: "security" group_wait: 0s receivers: - name: "default" slack_configs: - api_url: "https://hooks.slack.com/services/..." channel: "#ai-alerts" send_resolved: true - name: "critical" slack_configs: - api_url: "https://hooks.slack.com/services/..." channel: "#ai-critical" send_resolved: true pagerduty_configs: - routing_key: "..." severity: critical - name: "security" webhook_configs: - url: "https://internal.security.com/ai-alert-webhook" 告警分级与响应 级别 响应时间 通知渠道 升级策略 示例 P0 Critical 5 分钟 PagerDuty + 电话 15分钟升级到团队负责人 安全违规、服务宕机 P1 Warning 30 分钟 Slack @channel 2小时升级 延迟超标、队列积压 P2 Info 4 小时 Slack 普通消息 无升级 GPU 利用率低 P3 Notice 24 小时 邮件 无升级 日花费趋势异常 日志聚合与链路追踪 结构化日志标准 import structlog import json logger = structlog.get_logger() async def handle_chat_request(request: ChatRequest): """带全链路日志的请求处理""" request_id = generate_request_id() logger.info("chat_request_start", request_id=request_id, user_id=request.user_id, model=request.model, message_length=len(request.message), conversation_id=request.conversation_id, ) try: # 模型路由 routed_model = router.route(request.message) logger.info("model_routed", request_id=request_id, original_model=request.model, routed_model=routed_model, ) # RAG 检索 if needs_retrieval(request.message): retrieved = await retriever.search(request.message) logger.info("rag_retrieval_complete", request_id=request_id, chunks_retrieved=len(retrieved), top_score=retrieved[0]["score"] if retrieved else 0, ) # LLM 调用 response = await llm_client.chat.completions.create(...) logger.info("chat_request_complete", request_id=request_id, model=routed_model, input_tokens=response.usage.prompt_tokens, output_tokens=response.usage.completion_tokens, ttft_ms=first_token_latency, total_latency_ms=total_latency, cost_usd=cost, ) return response except Exception as e: logger.error("chat_request_failed", request_id=request_id, error_type=type(e).__name__, error_message=str(e), ) raise 成本异常检测 class CostAnomalyDetector: """基于统计方法的成本异常检测""" def __init__(self, window_size: int = 168): # 7 天小时数据 self.window_size = window_size self.cost_history: list[float] = [] def update(self, hourly_cost: float) -> dict | None: self.cost_history.append(hourly_cost) if len(self.cost_history) < self.window_size: return None # 使用滑动窗口统计 recent = self.cost_history[-self.window_size:] mean = sum(recent) / len(recent) variance = sum((x - mean) ** 2 for x in recent) / len(recent) std = variance ** 0.5 # Z-score 异常检测 z_score = abs(hourly_cost - mean) / max(std, 0.01) if z_score > 3: return { "type": "cost_spike", "current_cost": hourly_cost, "expected_mean": mean, "z_score": z_score, "deviation_pct": (hourly_cost - mean) / mean * 100, } return None 结语 AI 应用监控的核心理念是从基础设施到业务价值的全链路可观测。GPU 利用率告诉你硬件是否健康，TTFT 告诉你用户是否在等待，幻觉率告诉你 AI 是否可信，成本指标告诉你系统是否可持续。 ...

监控分层架构 AI 应用监控需要三层视角，每层关注不同问题： ┌─────────────────────────────────────────────────┐ │ Layer 3: 模型质量监控 │ │ - 回答正确率 / 幻觉率 / 毒性检测 / 偏见 │ │ - 用户满意度反馈 │ ├─────────────────────────────────────────────────┤ │ Layer 2: 应用性能监控 │ │ - 首字延迟 (TTFT) / 完整延迟 / 吞吐量 │ │ - Token 消耗 / 成本 / 缓存命中率 │ ├─────────────────────────────────────────────────┤ │ Layer 1: 基础设施监控 │ │ - GPU 利用率 / 内存 / 网络 / 磁盘 │ │ - API 错误率 / 限流次数 / 超时 │ └─────────────────────────────────────────────────┘ 指标设计 核心指标清单 层级 指标 类型 告警阈值 模型 回答质量评分 Gauge < 0.8 模型 幻觉率 Gauge > 5% 模型 用户点赞率 Gauge < 70% 应用 TTFT (首字延迟) Histogram P99 > 2s 应用 完整延迟 Histogram P99 > 15s 应用 Token 消耗 Counter 日预算 120% 应用 缓存命中率 Gauge < 15% 基础设施 GPU 利用率 Gauge > 90% 基础设施 API 错误率 Gauge > 1% 基础设施 限流次数 Counter > 100/min Prometheus 指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary # === 模型质量指标 === llm_quality_score = Gauge( "llm_quality_score", "Average answer quality score (0-1)", ["model", "task_type"] ) llm_hallucination_rate = Gauge( "llm_hallucination_rate", "Hallucination rate detected", ["model"] ) llm_user_feedback = Counter( "llm_user_feedback_total", "User feedback counts", ["model", "feedback"] # feedback: positive/negative ) # === 应用性能指标 === llm_ttft = Histogram( "llm_time_to_first_token_seconds", "Time to first token in seconds", ["model"], buckets=[0.1, 0.3, 0.5, 1, 2, 5, 10] ) llm_total_latency = Histogram( "llm_total_latency_seconds", "Total request latency in seconds", ["model"], buckets=[0.5, 1, 2, 5, 10, 20, 30, 60] ) llm_tokens_total = Counter( "llm_tokens_total", "Total tokens consumed", ["model", "direction"] # direction: input/output ) llm_cost_total = Counter( "llm_cost_usd_total", "Total cost in USD", ["model"] ) llm_cache_hit_rate = Gauge( "llm_cache_hit_rate", "Cache hit rate", ["cache_type"] # cache_type: exact/semantic ) # === 基础设施指标 === gpu_utilization = Gauge( "gpu_utilization_percent", "GPU utilization percentage", ["gpu_id", "node"] ) api_error_rate = Gauge( "llm_api_error_rate", "LLM API error rate", ["provider", "error_type"] ) 指标采集 中间件式采集 import time from contextlib import asynccontextmanager class LLMInstrumentation: def __init__(self, registry): self.registry = registry @asynccontextmanager async def trace_request(self, model: str, task_type: str): request_id = generate_id() start = time.monotonic() first_token_time = None async def on_first_token(): nonlocal first_token_time first_token_time = time.monotonic() try: yield {"request_id": request_id, "on_first_token": on_first_token} # 请求成功 elapsed = time.monotonic() - start llm_total_latency.labels(model=model).observe(elapsed) if first_token_time: ttft = first_token_time - start llm_ttft.labels(model=model).observe(ttft) except Exception as e: api_error_rate.labels( provider=model, error_type=type(e).__name__ ).inc() raise def record_tokens(self, model, input_tokens, output_tokens): llm_tokens_total.labels( model=model, direction="input" ).inc(input_tokens) llm_tokens_total.labels( model=model, direction="output" ).inc(output_tokens) # 成本计算 cost = self._calc_cost(model, input_tokens, output_tokens) llm_cost_total.labels(model=model).inc(cost) def _calc_cost(self, model, in_tok, out_tok): prices = { "gpt-4o": (0.0025, 0.01), "gpt-4o-mini": (0.00015, 0.0006), "claude-sonnet": (0.003, 0.015), } in_price, out_price = prices.get(model, (0, 0)) return (in_tok * in_price + out_tok * out_price) / 1000 使用方式 instrument = LLMInstrumentation(registry) async def chat_completion(messages, model="gpt-4o"): async with instrument.trace_request(model, "chat") as ctx: response = await llm_client.chat.completions.create( model=model, messages=messages, stream=True, ) async for chunk in response: if chunk.choices[0].delta.content: if not ctx["first_token_called"]: await ctx["on_first_token"]() ctx["first_token_called"] = True yield chunk.choices[0].delta.content instrument.record_tokens(model, input_count, output_count) 质量监控：LLM-as-a-Judge class QualityMonitor: def __init__(self, judge_model="gpt-4o", sample_rate=0.1): self.judge_model = judge_model self.sample_rate = sample_rate async def maybe_evaluate(self, query, response, context=None): """按采样率随机评估回答质量""" import random if random.random() > self.sample_rate: return score, issues = await self._judge(query, response, context) llm_quality_score.labels( model=self.judge_model, task_type="auto" ).set(score) if "hallucination" in issues: llm_hallucination_rate.labels( model=self.judge_model ).inc() async def _judge(self, query, response, context): prompt = f"""评估以下回答的质量。 问题：{query} 回答：{response} 参考资料：{context or '无'} 请检查： 1. 事实准确性（是否有幻觉） 2. 回答完整性 3. 逻辑一致性 输出 JSON：{{"score": 0.0-1.0, "issues": ["..."]}}""" result = await llm_client.chat.completions.create( model=self.judge_model, messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, ) data = json.loads(result.choices[0].message.content) return data["score"], data.get("issues", []) Grafana 仪表板 关键 Panel 配置 Panel 1: 延迟分布（P50/P90/P99） ...

为什么 LLM 需要专项可观测性？ 传统 APM 不够：LLM 有 Token 计费、Prompt 变体、模型路由、工具调用链等特有维度。一个请求可能涉及 3 个模型 + 5 个工具调用 + 2 次检索，没有 Tracing 根本无法定位问题。 三位一体架构 ┌──────────────────────────────────────────────────┐ │ 用户请求 │ │ trace_id = xxx │ └──────────────────────┬───────────────────────────┘ │ ┌──────────────┼──────────────┐ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ Logs │ │ Traces │ │ Metrics │ │ 结构化 │ │ 链路 │ │ 聚合 │ │ 日志 │ │ 追踪 │ │ 指标 │ └────┬────┘ └────┬────┘ └────┬────┘ │ │ │ ▼ ▼ ▼ ┌─────────┐ ┌─────────┐ ┌─────────┐ │ ELK / │ │ Jaeger /│ │Prometheus│ │ Loki │ │ Langfuse│ │ + Grafana│ └─────────┘ └─────────┘ └─────────┘ │ │ │ └──────────────┼──────────────┘ ▼ ┌─────────────────┐ │ AlertManager │ │ 告警 + 通知 │ └─────────────────┘ 一、结构化日志 import structlog import json # 配置 structlog structlog.configure( processors=[ structlog.contextvars.merge_contextvars, structlog.processors.add_log_level, structlog.processors.TimeStamper(fmt="iso"), structlog.processors.JSONRenderer(), ], ) logger = structlog.get_logger() class LLMLogger: """LLM 专用结构化日志""" def log_request(self, trace_id: str, user_id: str, model: str, prompt: str, **kwargs): logger.info("llm_request", trace_id=trace_id, user_id=user_id, model=model, prompt_length=len(prompt), prompt_tokens=kwargs.get("input_tokens"), max_tokens=kwargs.get("max_tokens"), temperature=kwargs.get("temperature", 1.0), tools=kwargs.get("tools"), timestamp=datetime.utcnow().isoformat(), ) def log_response(self, trace_id: str, response: str, input_tokens: int, output_tokens: int, latency_ms: float, model: str, **kwargs): logger.info("llm_response", trace_id=trace_id, model=model, input_tokens=input_tokens, output_tokens=output_tokens, total_tokens=input_tokens + output_tokens, latency_ms=latency_ms, finish_reason=kwargs.get("finish_reason"), cost_usd=self._calc_cost(model, input_tokens, output_tokens), ) def log_tool_call(self, trace_id: str, tool_name: str, params: dict, result: dict, latency_ms: float): logger.info("tool_call", trace_id=trace_id, tool=tool_name, params_keys=list(params.keys()), result_status="success" if result.get("success") else "failed", latency_ms=latency_ms, ) def log_error(self, trace_id: str, error: Exception, context: dict): logger.error("llm_error", trace_id=trace_id, error_type=type(error).__name__, error_message=str(error), context=context, ) 日志查询示例 # ELK / Loki 查询：查找高延迟请求 # Kibana KQL: # llm_response AND latency_ms > 5000 AND model: "gpt-4" # Grafana Loki LogQL: # {app="llm-service"} |= "llm_response" | json | latency_ms > 5000 二、分布式链路追踪 LLM 请求的典型链路：API → Router → Cache → Model → Tool → Model → Response ...

为什么 AI 应用需要可观测性 传统应用监控关注 CPU、内存、QPS。AI 应用需要关注：LLM 调用了几次？每次延迟多少？Token 消耗多少？检索结果是否相关？Agent 的推理链路是什么？没有可观测性，AI 应用就是黑盒，出问题只能猜。 三支柱模型 支柱 传统应用 AI 应用 工具 Tracing 请求链路 LLM 调用链 + Agent 推理步骤 + 检索过程 LangSmith/Langfuse Logging 日志 Prompt/Response/中间状态 结构化日志 Metrics QPS/延迟 Token消耗/检索召回率/幻觉率 Prometheus+Grafana 关键指标定义 from dataclasses import dataclass from typing import Optional @dataclass class LLMCallMetrics: # 延迟指标 time_to_first_token: float # 首 token 延迟(ms) total_latency: float # 总延迟(ms) # 成本指标 prompt_tokens: int completion_tokens: int total_cost: float # 质量指标 model_name: str temperature: float finish_reason: str # 检索指标(RAG场景) retrieval_count: Optional[int] = None retrieval_relevance: Optional[float] = None context_utilization: Optional[float] = None @dataclass class AgentTrace: trace_id: str spans: list[dict] # 每个步骤一个 span # span 结构: name, start, end, input, output, metadata Langfuse 集成实现 from langfuse import Langfuse from langfuse.decorators import langfuse_context, observe import functools, time langfuse = Langfuse( public_key="pk-lf-xxx", secret_key="sk-lf-xxx", host="https://your-langfuse.com" ) def trace_llm(name="llm_call"): def decorator(func): @functools.wraps(func) async def wrapper(*args, **kwargs): start = time.time() span = langfuse_context.start_span(name=name, input=kwargs) try: result = await func(*args, **kwargs) ttft = time.time() - start span.end( output=result, metadata={"ttft_ms": ttft * 1000, "model": kwargs.get("model", "unknown")} ) langfuse_context.update_current_observation( usage={ "prompt_tokens": result.get("usage", {}).get("prompt_tokens", 0), "completion_tokens": result.get("usage", {}).get("completion_tokens", 0), "total_tokens": result.get("usage", {}).get("total_tokens", 0), }, metadata={"latency_ms": (time.time() - start) * 1000} ) return result except Exception as e: span.end(level="ERROR", status_message=str(e)) raise return wrapper return decorator @observe(name="rag_pipeline") async def rag_pipeline(query: str): docs = await retrieve(query) answer = await generate(query, docs) return answer @observe(name="retrieve") async def retrieve(query: str): results = await vector_search(query) return {"query": query, "docs": results} @trace_llm(name="generate") async def generate(query: str, docs: list): resp = await llm_client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": f"Q: {query}\nContext: {docs}"}] ) return resp 工具对比 工具 部署方式 核心优势 适用场景 价格 LangSmith SaaS LangChain 深度集成 LangChain 用户 $39/月起 Langfuse 自部署/SaaS 开源，框架无关 通用 AI 应用 免费/自部署 Phoenix 自部署 Arize 出品，评估强 需要评估体系 开源 OpenTelemetry 自部署 标准化，通用 已有 OTel 体系 免费 选型建议：不用 LangChain 选 Langfuse（开源、框架无关），用 LangChain 选 LangSmith（集成最好），需要深度评估选 Phoenix。 ...