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 是否可信，成本指标告诉你系统是否可持续。
...
