引言 AI应用的监控比传统软件复杂得多。除了常规的系统指标（CPU、内存、延迟），还需要监控AI特有的指标（输出质量、幻觉率、安全事件）。2026年，AI性能监控已经发展成为一个专门的领域。本文将系统介绍AI性能监控体系的构建。
AI监控的独特需求 传统软件监控 vs AI监控 维度 传统软件 AI应用 延迟 毫秒级 秒级（可接受） 错误类型 崩溃、超时 幻觉、不当内容 质量指标 功能正确性 输出准确性、相关性 成本 服务器成本 API调用成本（按token计） 变化来源 代码部署 代码+模型版本+提示 AI监控的核心指标 AI监控指标体系 ├── 性能指标 │ ├── 延迟（P50/P95/P99） │ ├── 吞吐量 │ └── 并发数 ├── 质量指标 │ ├── 输出准确率 │ ├── 幻觉率 │ ├── 拒绝率 │ └── 用户满意度 ├── 成本指标 │ ├── 每次请求成本 │ ├── 每日总成本 │ └── token效率 ├── 安全指标 │ ├── 有害内容率 │ ├── 注入攻击次数 │ └── 数据泄露事件 └── 可靠性指标 ├── 可用性 ├── 错误率 └── 降级率 监控架构 数据采集层 class MetricsCollector: def __init__(self): self.collectors = [ LatencyCollector(), QualityCollector(), CostCollector(), SafetyCollector(), ReliabilityCollector() ] def record_request(self, request_id, request, response, metadata): """记录每次请求""" for collector in self.collectors: collector.record(request_id, request, response, metadata) 指标计算层 class MetricsCalculator: def calculate(self, raw_metrics): return { "latency": { "p50": percentile(raw_metrics["latencies"], 50), "p95": percentile(raw_metrics["latencies"], 95), "p99": percentile(raw_metrics["latencies"], 99), }, "quality": { "accuracy": raw_metrics["correct"] / raw_metrics["total"], "hallucination_rate": raw_metrics["hallucinations"] / raw_metrics["total"], "refusal_rate": raw_metrics["refusals"] / raw_metrics["total"], }, "cost": { "per_request": raw_metrics["total_cost"] / raw_metrics["total"], "daily": raw_metrics["total_cost"], "token_efficiency": raw_metrics["output_tokens"] / raw_metrics["input_tokens"], }, "safety": { "harmful_rate": raw_metrics["harmful"] / raw_metrics["total"], "injection_attempts": raw_metrics["injections"], }, "reliability": { "availability": 1 - raw_metrics["downtime"] / raw_metrics["total_time"], "error_rate": raw_metrics["errors"] / raw_metrics["total"], } } 告警层 class AlertManager: def __init__(self): self.rules = [ AlertRule("high_latency", "p95_latency > 5000", severity="warning"), AlertRule("critical_latency", "p99_latency > 10000", severity="critical"), AlertRule("high_error", "error_rate > 0.05", severity="critical"), AlertRule("quality_drop", "accuracy < 0.85", severity="warning"), AlertRule("hallucination_spike", "hallucination_rate > 0.1", severity="critical"), AlertRule("cost_spike", "daily_cost > budget * 1.2", severity="warning"), AlertRule("safety_incident", "harmful_rate > 0.01", severity="critical"), ] def check(self, metrics): alerts = [] for rule in self.rules: if rule.evaluate(metrics): alerts.append(rule.create_alert(metrics)) if alerts: self.notify(alerts) return alerts 关键监控实现 延迟监控 class LatencyMonitor: def __init__(self): self.latencies = SlidingWindow(size=10000) def record(self, request_id, start_time, end_time): latency = end_time - start_time self.latencies.append(latency) # 实时检查 if latency > 10: # 超过10秒 self.alert(f"请求 {request_id} 延迟 {latency:.1f}s") def get_stats(self): return { "p50": self.latencies.percentile(50), "p95": self.latencies.percentile(95), "p99": self.latencies.percentile(99), "max": self.latencies.max(), "avg": self.latencies.mean() } 质量监控 class QualityMonitor: def __init__(self): self.sample_rate = 0.1 # 采样10%进行质量评估 self.evaluator = LLMJudge(model="gpt-5") # 用GPT-5评估 async def evaluate(self, request, response): """异步评估输出质量""" if random.random() > self.sample_rate: return # 采样 # 用LLM评估 score = await self.evaluator.evaluate( input=request, output=response, criteria=["accuracy", "relevance", "completeness"] ) if score["accuracy"] < 0.7: self.alert(f"低质量输出检测：{score}") return score 成本监控 class CostMonitor: def __init__(self, daily_budget=100): self.daily_budget = daily_budget self.today_cost = 0 self.costs = [] def record(self, input_tokens, output_tokens, model): cost = calculate_cost(input_tokens, output_tokens, model) self.today_cost += cost self.costs.append({"timestamp": datetime.now(), "cost": cost}) # 预算检查 if self.today_cost > self.daily_budget * 0.8: self.alert("日预算已用80%") if self.today_cost > self.daily_budget: self.alert("日预算超支！") return "stop" # 触发熔断 安全监控 class SafetyMonitor: def __init__(self): self.content_filter = ContentFilter() self.injection_detector = InjectionDetector() def check_input(self, user_input): """检查输入安全性""" if self.injection_detector.is_injection(user_input): self.log_incident("injection_attempt", user_input) return False if self.content_filter.is_harmful(user_input): self.log_incident("harmful_input", user_input) return False return True def check_output(self, output): """检查输出安全性""" if self.content_filter.is_harmful(output): self.log_incident("harmful_output", output) return False return True 可视化仪表板 class MonitoringDashboard: def render(self): return { "overview": { "status": "healthy", # healthy/warning/critical "uptime": "99.97%", "requests_today": 154289, "avg_latency": "1.2s", "cost_today": "$45.30" }, "latency_chart": self.render_latency_chart(), "quality_trend": self.render_quality_trend(), "cost_trend": self.render_cost_trend(), "alerts": self.get_active_alerts(), "top_errors": self.get_top_errors() } 告警策略 告警分级 级别 条件 响应时间 通知方式 P0 系统不可用 立即 电话+短信+邮件 P1 关键指标超标 15分钟 短信+邮件 P2 质量下降 1小时 邮件+IM P3 预警 4小时 IM 告警抑制 def should_suppress(alert, recent_alerts): """避免告警风暴""" # 同类告警5分钟内只发一次 for recent in recent_alerts: if (recent["type"] == alert["type"] and (datetime.now() - recent["timestamp"]).seconds < 300): return True return False 2026年新趋势 1. AI自监控 用AI监控AI：模型自己评估输出质量，自动发现异常。
...
