第2431篇：AI团队的On-Call体系——AI系统的告警响应和故障处理
2026/4/30大约 7 分钟
第2431篇：AI团队的On-Call体系——AI系统的告警响应和故障处理
适读人群：AI系统运维工程师和技术负责人 | 阅读时长：约13分钟 | 核心价值：建立适合AI系统特性的On-Call体系，有效响应AI特有的故障模式
凌晨2点，手机告警响了。
推理服务的P99延迟从150ms涨到了1200ms，部分请求开始超时。
这是AI系统On-Call的日常。但和传统服务不同，AI故障的排查路径完全不一样——可能是模型输入分布变了，可能是GPU驱动更新了，可能是某个外部API的响应变慢，可能是特定类型的输入触发了模型的慢路径。
建立AI系统的On-Call体系，首先要理解AI故障的特殊性。
一、AI系统的特有故障模式
传统服务的故障通常是binary的：服务要么up要么down。AI系统有一批灰度故障，更难发现也更难处理：
AI系统故障分类
├── 硬故障（传统软件也有）
│ ├── 服务宕机
│ ├── OOM（Out of Memory）
│ └── 依赖服务不可用
│
└── 软故障（AI特有）
├── 模型效果退化（静默故障，无报错但输出质量下降）
├── 数据分布偏移（输入数据与训练分布差异越来越大）
├── 推理延迟抖动（GPU资源争抢、批处理大小问题）
├── 幻觉率上升（LLM输出质量不稳定）
└── 特定输入触发的降级（某类特征组合导致异常预测）最危险的是"静默故障"——系统还在跑，没有任何报错，但输出质量已经悄悄下降了。
二、告警体系设计
2.1 分层告警策略
from dataclasses import dataclass
from enum import Enum
from typing import List, Callable, Optional
import time
class AlertSeverity(Enum):
P0 = "P0" # 立即响应，电话呼叫，业务中断
P1 = "P1" # 15分钟内响应，短信+企业微信
P2 = "P2" # 1小时内响应，企业微信通知
P3 = "P3" # 工作时间处理，日报汇总
@dataclass
class AIAlertRule:
"""AI系统告警规则"""
name: str
description: str
severity: AlertSeverity
# 监控指标和阈值
metric_name: str
threshold: float
comparison: str # "gt", "lt", "abs_change_pct"
evaluation_window: int # 评估窗口（分钟）
# 确认条件（避免误报）
confirmation_count: int = 3 # 连续N次超阈值才告警
cooldown_minutes: int = 30 # 同一告警的冷却时间
def check(self, current_value: float, baseline_value: float) -> bool:
"""检查是否触发告警"""
if self.comparison == "gt":
return current_value > self.threshold
elif self.comparison == "lt":
return current_value < self.threshold
elif self.comparison == "abs_change_pct":
change_pct = abs((current_value - baseline_value) / baseline_value) * 100
return change_pct > self.threshold
return False
# AI系统标准告警规则集
STANDARD_AI_ALERT_RULES = [
# P0: 服务不可用
AIAlertRule(
name="service_down",
description="推理服务不可访问",
severity=AlertSeverity.P0,
metric_name="health_check_success_rate",
threshold=0.95,
comparison="lt",
evaluation_window=2,
confirmation_count=2
),
# P0: 延迟极端异常
AIAlertRule(
name="extreme_latency",
description="P99延迟超过10倍正常值",
severity=AlertSeverity.P0,
metric_name="p99_latency_ms",
threshold=1500, # 正常P99是150ms
comparison="gt",
evaluation_window=5,
confirmation_count=3
),
# P1: 模型效果明显下降
AIAlertRule(
name="model_accuracy_drop",
description="模型在线评估准确率下降超过5%",
severity=AlertSeverity.P1,
metric_name="online_accuracy",
threshold=5.0,
comparison="abs_change_pct",
evaluation_window=60
),
# P1: 错误率上升
AIAlertRule(
name="high_error_rate",
description="推理错误率超过1%",
severity=AlertSeverity.P1,
metric_name="error_rate",
threshold=0.01,
comparison="gt",
evaluation_window=15
),
# P2: 数据分布偏移
AIAlertRule(
name="feature_distribution_shift",
description="关键特征分布偏移超过阈值（PSI > 0.2）",
severity=AlertSeverity.P2,
metric_name="feature_psi",
threshold=0.2,
comparison="gt",
evaluation_window=1440 # 每天检查
),
# P2: 推理延迟升高
AIAlertRule(
name="latency_increase",
description="P95延迟比基线高50%",
severity=AlertSeverity.P2,
metric_name="p95_latency_ms",
threshold=50.0,
comparison="abs_change_pct",
evaluation_window=30
),
# P3: 预测分布变化
AIAlertRule(
name="prediction_distribution_shift",
description="输出分布发生显著变化",
severity=AlertSeverity.P3,
metric_name="output_distribution_kl_divergence",
threshold=0.1,
comparison="gt",
evaluation_window=1440
)
]2.2 AI特有的监控指标
class AIMetricsCollector:
"""AI系统专有指标采集"""
def collect_inference_metrics(self,
predictions: list,
latencies: list,
errors: list) -> dict:
import numpy as np
if not latencies:
return {}
return {
# 性能指标
"p50_latency_ms": float(np.percentile(latencies, 50)),
"p95_latency_ms": float(np.percentile(latencies, 95)),
"p99_latency_ms": float(np.percentile(latencies, 99)),
"max_latency_ms": float(max(latencies)),
# 错误指标
"error_rate": len(errors) / len(latencies) if latencies else 0,
"error_types": self._count_error_types(errors),
# 模型输出分布（用于检测偏移）
"prediction_mean": float(np.mean(predictions)) if predictions else 0,
"prediction_std": float(np.std(predictions)) if predictions else 0,
"prediction_positive_rate": (
float(sum(1 for p in predictions if p >= 0.5) / len(predictions))
if predictions else 0
),
# GPU指标（如果有）
"gpu_memory_used_mb": self._get_gpu_memory(),
"gpu_utilization_pct": self._get_gpu_utilization()
}
def compute_feature_psi(self,
baseline_distribution: dict,
current_distribution: dict) -> float:
"""计算特征的PSI（群体稳定性指数）"""
import numpy as np
psi = 0.0
for bucket in baseline_distribution:
expected = baseline_distribution[bucket]
actual = current_distribution.get(bucket, 0.0001)
expected = max(expected, 0.0001)
psi += (actual - expected) * np.log(actual / expected)
return psi
def _count_error_types(self, errors: list) -> dict:
counts = {}
for error in errors:
error_type = type(error).__name__
counts[error_type] = counts.get(error_type, 0) + 1
return counts
def _get_gpu_memory(self) -> float:
try:
import torch
if torch.cuda.is_available():
return torch.cuda.memory_allocated() / 1024 / 1024
except:
pass
return 0.0
def _get_gpu_utilization(self) -> float:
return 0.0 # 实际需要从nvidia-smi获取三、On-Call响应手册
好的On-Call不是靠英雄主义，是靠清晰的响应流程。
3.1 故障响应标准流程
graph TD
A["收到告警"] --> B["初步定性：P0/P1/P2？"]
B -->|"P0"| C["立即拉起事故频道"]
B -->|"P1"| D["登录系统开始排查"]
B -->|"P2/P3"| E["记录到工单，正常工作时间处理"]
C --> F["5分钟内：确认是否真实故障"]
F -->|"是"| G["通知相关方，开始缓解"]
F -->|"误报"| H["关闭告警，更新阈值"]
G --> I{"能快速修复吗？"}
I -->|"是"| J["执行修复，验证恢复"]
I -->|"否"| K["启动降级方案，防止扩散"]
K --> L["深入排查根因"]
J --> M["事后复盘"]
L --> M3.2 AI故障排查手册
# AI系统故障排查脚本
class AIIncidentDiagnosis:
"""AI系统事故诊断工具"""
def run_quick_diagnosis(self, service_name: str) -> dict:
"""
快速诊断：按优先级依次检查，找到最可能的故障原因
目标：5分钟内得到初步诊断
"""
checks = []
# Check 1: 基础健康检查
health = self._check_service_health(service_name)
checks.append(health)
if not health["healthy"]:
return self._make_diagnosis("SERVICE_DOWN", checks,
"服务不可访问，检查进程和端口")
# Check 2: 模型文件完整性
model_check = self._check_model_files(service_name)
checks.append(model_check)
if not model_check["valid"]:
return self._make_diagnosis("MODEL_CORRUPTION", checks,
"模型文件异常，考虑回滚到上一个版本")
# Check 3: 推理延迟分布
latency_check = self._check_latency_distribution(service_name)
checks.append(latency_check)
if latency_check["has_tail_latency"]:
return self._make_diagnosis("TAIL_LATENCY", checks,
"存在长尾延迟，检查GPU资源、批处理配置")
# Check 4: 输入数据分布
input_check = self._check_input_distribution(service_name)
checks.append(input_check)
if input_check["distribution_shift"]:
return self._make_diagnosis("INPUT_DISTRIBUTION_SHIFT", checks,
"输入数据分布异常，可能是上游数据问题")
# Check 5: 模型输出分布
output_check = self._check_output_distribution(service_name)
checks.append(output_check)
if output_check["distribution_shift"]:
return self._make_diagnosis("OUTPUT_DISTRIBUTION_SHIFT", checks,
"模型输出分布变化，可能需要重新校准或重训")
return self._make_diagnosis("UNKNOWN", checks,
"基础检查未发现明确原因，需要深入调查")
def get_rollback_command(self, service_name: str,
target_version: str) -> str:
"""生成回滚命令（紧急时刻不要靠记忆）"""
return f"""
# 模型回滚命令 - 请在执行前确认：
# 1. 目标版本 {target_version} 已经过验证
# 2. 已通知相关团队
kubectl set image deployment/{service_name}-deploy \\
{service_name}=registry.company.com/{service_name}:{target_version}
# 等待回滚完成
kubectl rollout status deployment/{service_name}-deploy
# 验证
kubectl get pods -l app={service_name}
curl https://api.internal/{service_name}/health
"""
def _check_service_health(self, service: str) -> dict:
# 实际检查健康端点
return {"check": "health", "healthy": True}
def _check_model_files(self, service: str) -> dict:
return {"check": "model_files", "valid": True}
def _check_latency_distribution(self, service: str) -> dict:
return {"check": "latency", "has_tail_latency": False}
def _check_input_distribution(self, service: str) -> dict:
return {"check": "input_dist", "distribution_shift": False}
def _check_output_distribution(self, service: str) -> dict:
return {"check": "output_dist", "distribution_shift": False}
def _make_diagnosis(self, diagnosis_type: str,
checks: list, suggestion: str) -> dict:
return {
"diagnosis": diagnosis_type,
"suggestion": suggestion,
"checks_performed": checks,
"confidence": "medium"
}四、On-Call的可持续性
On-Call最大的问题不是技术，是人。长期高压的On-Call会导致工程师精力耗尽，进而离职。
几个保持可持续性的实践：
轮值制度：至少3-4人轮值，每次On-Call不超过1周，周末必须保证有备份。
告警质量：每周统计误报率。超过30%的告警是误报，说明阈值设置有问题，要修。每条告警都应该是"需要人类决策的"，而不是"系统自动处理就行"。
事后复盘不追责：每次严重事故后做复盘，关注的是"系统怎么改才能避免下次"，而不是"谁的错"。追责文化会让On-Call工程师只想自保，不愿意如实报告问题。
On-Call经历是成长：把On-Call经历纳入工程师成长路径——每次事故处理都是真实的系统级工程训练，是面试中含金量最高的经历。