一、为什么上线不是"扔上去就行"？

很多人以为部署就是把模型文件丢到服务器，然后……没有然后了。但在真实的AI系统运维中，上线只是开始，持续监控才是保障质量的关键。

AI系统上线全流程：

  ┌───────────┐    ┌───────────┐    ┌───────────┐    ┌───────────┐
  │  测试通过  │ ─▶ │  发布策略  │ ─▶ │  灰度放量  │ ─▶ │  全量上线  │
  └───────────┘    └───────────┘    └───────────┘    └───────────┘
                                                            │
                              ┌─────────────────────────────┘
                              ▼
  ┌──────────────────────────────────────────────────────────┐
  │                    持续监控层                              │
  │                                                          │
  │  业务监控（准确率/召回率）  系统监控（延迟/QPS/错误率）       │
  │  数据监控（漂移检测）       日志监控（异常请求追踪）          │
  └──────────────────────────────────────────────────────────┘
                              │
              ┌───────────────┼───────────────┐
              ▼               ▼               ▼
          正常运行          告警触发         异常处理
              │               │               │
          持续监控          人工介入         回滚/降级

---

二、四种发布策略

四种发布策略对比图：

  蓝绿部署（Blue-Green）
  ┌──────────────────────────────────────────────────┐
  │  流量: 100%                                       │
  │  ┌────────┐         ┌────────┐                   │
  │  │ 蓝环境 │ ◀─────  │ 负载   │  ──▶ ┌────────┐  │
  │  │ (旧版) │         │ 均衡器 │       │ 绿环境 │  │
  │  └────────┘ 切换后  └────────┘  准备 │ (新版) │  │
  │                      流量100%→绿      └────────┘  │
  └──────────────────────────────────────────────────┘

  金丝雀发布（Canary Release）
  ┌──────────────────────────────────────────────────┐
  │  阶段1: 旧版95% + 新版5%  → 观察指标              │
  │  阶段2: 旧版70% + 新版30% → 确认稳定              │
  │  阶段3: 旧版 0% + 新版100% → 全量完成             │
  └──────────────────────────────────────────────────┘
  
  来源：矿工进入矿井前会放一只金丝雀——金丝雀对有毒气体敏感，
       它没事才说明环境安全。新版本先服务少量用户，没问题再放量。

策略	原理	优点	缺点	适用场景
蓝绿部署	同时维护两套环境，流量一次性切换	切换快（秒级），回滚简单	资源成本翻倍	小规模系统，快速发布
金丝雀发布	逐步将流量从旧版切到新版（5%→30%→100%）	风险可控，可逐步验证	两个版本同时运行，日志分析复杂	AI模型上线首选
影子测试	新版本接收流量的"镜像"，结果不对外呈现	零风险测试生产流量	无法验证用户真实响应	重大版本验证
滚动更新	逐台替换旧版实例	无停机时间	中间状态两版本混跑，回滚较慢	Kubernetes默认策略

2.1 金丝雀发布代码实现

import random
import time
from typing import Callable, Dict, Any

class CanaryRouter:
    """金丝雀流量路由器"""
    
    def __init__(self, new_model_handler: Callable, old_model_handler: Callable):
        self.new_handler = new_model_handler
        self.old_handler = old_model_handler
        self._canary_ratio = 0.05  # 初始5%流量给新版本
        self.stats = {"new": {"total": 0, "errors": 0}, 
                      "old": {"total": 0, "errors": 0}}
    
    def set_canary_ratio(self, ratio: float):
        """调整金丝雀流量比例 (0.0 - 1.0)"""
        assert 0.0 <= ratio <= 1.0
        self._canary_ratio = ratio
        print(f"[Canary] 流量比例更新: 新版本 {ratio*100:.0f}%, 旧版本 {(1-ratio)*100:.0f}%")
    
    def route(self, request: Dict[str, Any]) -> Dict[str, Any]:
        """根据流量比例路由请求"""
        use_new = random.random() < self._canary_ratio
        version = "new" if use_new else "old"
        handler = self.new_handler if use_new else self.old_handler
        
        try:
            start = time.perf_counter()
            result = handler(request)
            latency = (time.perf_counter() - start) * 1000
            
            self.stats[version]["total"] += 1
            result["_version"] = version
            result["_latency_ms"] = round(latency, 2)
            return result
        except Exception as e:
            self.stats[version]["errors"] += 1
            raise
    
    def error_rate(self, version: str) -> float:
        """计算指定版本的错误率"""
        total = self.stats[version]["total"]
        if total == 0:
            return 0.0
        return self.stats[version]["errors"] / total
    
    def auto_rollback_check(self, error_threshold: float = 0.05):
        """自动回滚检查：新版本错误率超阈值则回滚到0%"""
        new_error_rate = self.error_rate("new")
        if new_error_rate > error_threshold and self.stats["new"]["total"] > 100:
            print(f"[Canary] ⚠️ 新版本错误率 {new_error_rate:.1%} > 阈值 {error_threshold:.1%}")
            print("[Canary] 🔴 触发自动回滚，流量切回旧版本")
            self.set_canary_ratio(0.0)
            return True
        return False

# 使用示例
def old_model(req): return {"label": "正面", "confidence": 0.82}
def new_model(req): return {"label": "正面", "confidence": 0.87}

router = CanaryRouter(new_model, old_model)

# 灰度放量流程
for ratio in [0.05, 0.10, 0.30, 0.50, 1.0]:
    router.set_canary_ratio(ratio)
    # 处理一批请求后检查是否需要回滚
    if router.auto_rollback_check():
        break
    time.sleep(1)  # 实际中等待观察窗口（分钟/小时级）

---

三、监控指标体系

AI服务监控四层架构：

  ┌──────────────────────────────────────────────────────────┐
  │  Layer 4: 业务监控（Business Metrics）                    │
  │  准确率下滑 / 拒识率上升 / 用户投诉率                       │
  ├──────────────────────────────────────────────────────────┤
  │  Layer 3: 模型监控（Model Metrics）                       │
  │  预测分布偏移 / 置信度分布 / 特征分布漂移                    │
  ├──────────────────────────────────────────────────────────┤
  │  Layer 2: 应用监控（Application Metrics）                 │
  │  QPS / 延迟P50/P95/P99 / 错误率 / 超时率                  │
  ├──────────────────────────────────────────────────────────┤
  │  Layer 1: 基础设施监控（Infrastructure Metrics）           │
  │  CPU使用率 / 内存 / GPU利用率 / 磁盘IO                     │
  └──────────────────────────────────────────────────────────┘
  
  监控告警优先级：Layer4 > Layer3 > Layer2 > Layer1
  （业务问题最重要，基础设施问题容忍度最高）

3.1 核心监控指标清单

层级	指标	告警阈值（示例）	采集频率
业务层	模型准确率（滑动窗口）	下降 > 3% 触发告警	实时（每请求）
业务层	拒识率（置信度 < 阈值）	> 10% 触发告警	1分钟聚合
模型层	预测分布偏移（PSI）	PSI > 0.2 触发告警	每小时
模型层	平均置信度	下降 > 0.05 告警	5分钟聚合
应用层	P99 延迟	> 500ms 告警，> 1s 紧急	实时
应用层	QPS（每秒请求数）	超出设计上限的80%	实时
应用层	错误率（5xx）	> 1% 告警，> 5% 紧急	实时
基础设施	CPU使用率	> 85% 告警	15秒
基础设施	内存使用率	> 90% 告警	15秒

---

四、代码：ModelMonitor 实时监控类

import time
import json
import logging
from collections import deque
from threading import Lock
from typing import Optional, List, Dict
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)

@dataclass
class PredictionRecord:
    """单次预测记录"""
    timestamp: float
    label: str
    confidence: float
    latency_ms: float
    ground_truth: Optional[str] = None  # 有延迟反馈时填写
    is_correct: Optional[bool] = None

class ModelMonitor:
    """
    AI模型线上监控器
    
    功能：
    - 滑动窗口统计（准确率/延迟/置信度分布）
    - 自动告警（阈值检测）
    - 指标导出（Prometheus格式）
    """
    
    def __init__(self, window_size: int = 1000, alert_accuracy_drop: float = 0.03):
        """
        Args:
            window_size: 滑动窗口大小（记录最近N条预测）
            alert_accuracy_drop: 准确率下降超过此值触发告警
        """
        self.window = deque(maxlen=window_size)
        self.alert_accuracy_drop = alert_accuracy_drop
        self.lock = Lock()
        
        # 基准指标（上线初期的"黄金指标"）
        self.baseline_accuracy: Optional[float] = None
        self.alerts: List[Dict] = []
    
    def record(self, label: str, confidence: float, latency_ms: float,
               ground_truth: Optional[str] = None):
        """记录一次预测结果"""
        record = PredictionRecord(
            timestamp=time.time(),
            label=label,
            confidence=confidence,
            latency_ms=latency_ms,
            ground_truth=ground_truth,
            is_correct=(label == ground_truth) if ground_truth else None
        )
        with self.lock:
            self.window.append(record)
        
        # 实时检查告警
        self._check_alerts()
    
    def set_baseline(self, accuracy: float):
        """设置基准准确率（通常取上线初期7天的平均值）"""
        self.baseline_accuracy = accuracy
        logger.info(f"基准准确率设定为: {accuracy:.2%}")
    
    def _check_alerts(self):
        """检查是否需要触发告警"""
        stats = self.get_stats()
        
        # 1. 准确率下滑告警
        if (self.baseline_accuracy and 
            stats.get("accuracy") is not None and
            self.baseline_accuracy - stats["accuracy"] > self.alert_accuracy_drop):
            
            alert = {
                "type": "ACCURACY_DROP",
                "severity": "WARNING",
                "message": f"准确率从基准 {self.baseline_accuracy:.2%} 下滑至 {stats['accuracy']:.2%}",
                "timestamp": time.time()
            }
            self.alerts.append(alert)
            logger.warning(f"🔔 告警触发: {alert['message']}")
        
        # 2. 高延迟告警
        p99_latency = stats.get("p99_latency_ms", 0)
        if p99_latency > 500:
            logger.warning(f"🔔 延迟告警: P99延迟 {p99_latency:.0f}ms > 500ms")
        
        # 3. 低置信度告警（模型不确定性升高）
        avg_confidence = stats.get("avg_confidence", 1.0)
        if avg_confidence < 0.6:
            logger.warning(f"🔔 置信度告警: 平均置信度 {avg_confidence:.2%} < 60%")
    
    def get_stats(self) -> Dict:
        """获取当前窗口统计数据"""
        with self.lock:
            records = list(self.window)
        
        if not records:
            return {}
        
        # 准确率（仅统计有ground_truth的记录）
        labeled = [r for r in records if r.is_correct is not None]
        accuracy = sum(1 for r in labeled if r.is_correct) / len(labeled) if labeled else None
        
        # 延迟分位数
        latencies = sorted(r.latency_ms for r in records)
        n = len(latencies)
        
        # 标签分布
        label_dist: Dict[str, int] = {}
        for r in records:
            label_dist[r.label] = label_dist.get(r.label, 0) + 1
        
        return {
            "window_size": len(records),
            "accuracy": accuracy,
            "avg_confidence": sum(r.confidence for r in records) / n,
            "p50_latency_ms": latencies[int(n * 0.50)],
            "p95_latency_ms": latencies[int(n * 0.95)],
            "p99_latency_ms": latencies[int(n * 0.99)],
            "label_distribution": {k: v/n for k, v in label_dist.items()},
            "alert_count": len(self.alerts)
        }
    
    def export_prometheus(self) -> str:
        """导出 Prometheus 文本格式指标"""
        stats = self.get_stats()
        lines = [
            f'# HELP model_accuracy Current sliding window accuracy',
            f'# TYPE model_accuracy gauge',
            f'model_accuracy {stats.get("accuracy", 0):.4f}',
            f'',
            f'# HELP model_latency_p99_ms P99 inference latency in milliseconds',
            f'# TYPE model_latency_p99_ms gauge',
            f'model_latency_p99_ms {stats.get("p99_latency_ms", 0):.1f}',
            f'',
            f'# HELP model_avg_confidence Average prediction confidence',
            f'# TYPE model_avg_confidence gauge',
            f'model_avg_confidence {stats.get("avg_confidence", 0):.4f}',
        ]
        return "\n".join(lines)

# 使用示例
monitor = ModelMonitor(window_size=500, alert_accuracy_drop=0.03)
monitor.set_baseline(accuracy=0.883)

# 模拟线上请求记录
import random
labels = ["正面", "中性", "负面"]
for i in range(200):
    pred = random.choice(labels)
    truth = random.choice(labels)
    monitor.record(
        label=pred,
        confidence=random.uniform(0.6, 0.99),
        latency_ms=random.uniform(10, 200),
        ground_truth=truth
    )

stats = monitor.get_stats()
print(f"当前准确率: {stats['accuracy']:.2%}")
print(f"P99延迟: {stats['p99_latency_ms']:.1f}ms")
print(f"标签分布: {stats['label_distribution']}")

---

五、在线 PSI 数据漂移检测

import numpy as np
from typing import List

def calculate_psi(expected: List[float], actual: List[float], 
                  bins: int = 10) -> float:
    """
    计算 PSI（Population Stability Index，群体稳定性指数）
    用于检测模型输入特征的分布是否发生偏移
    
    PSI < 0.1  → 分布稳定，无需关注
    0.1 ≤ PSI < 0.25 → 轻微偏移，需要关注
    PSI ≥ 0.25 → 显著偏移，需要重新训练模型
    
    Args:
        expected: 基准期（训练/上线初期）的特征值列表
        actual:   当前期的特征值列表
    """
    # 以基准期为基础建立 bins
    min_val = min(min(expected), min(actual))
    max_val = max(max(expected), max(actual))
    
    expected_counts, bin_edges = np.histogram(expected, bins=bins, 
                                               range=(min_val, max_val))
    actual_counts, _ = np.histogram(actual, bins=bin_edges)
    
    # 转换为比例，避免除零
    expected_pct = expected_counts / len(expected) + 1e-10
    actual_pct = actual_counts / len(actual) + 1e-10
    
    psi = np.sum((actual_pct - expected_pct) * np.log(actual_pct / expected_pct))
    return float(psi)

class DriftDetector:
    """多特征漂移检测器"""
    
    THRESHOLDS = {"stable": 0.1, "warning": 0.25}
    
    def __init__(self, feature_names: List[str]):
        self.feature_names = feature_names
        self.baselines: dict = {}
    
    def set_baseline(self, feature_name: str, baseline_data: List[float]):
        """设置特征基准分布"""
        self.baselines[feature_name] = baseline_data
        print(f"已设置特征 [{feature_name}] 的基准分布 (n={len(baseline_data)})")
    
    def check_drift(self, feature_name: str, current_data: List[float]) -> dict:
        """检测指定特征的分布偏移"""
        if feature_name not in self.baselines:
            raise ValueError(f"特征 [{feature_name}] 未设置基准")
        
        psi = calculate_psi(self.baselines[feature_name], current_data)
        
        if psi < self.THRESHOLDS["stable"]:
            status, emoji = "STABLE", "🟢"
        elif psi < self.THRESHOLDS["warning"]:
            status, emoji = "WARNING", "🟡"
        else:
            status, emoji = "CRITICAL", "🔴"
        
        result = {"feature": feature_name, "psi": psi, "status": status}
        print(f"{emoji} [{feature_name}] PSI = {psi:.4f} → {status}")
        return result
    
    def check_all(self, current_data: dict) -> List[dict]:
        """批量检测所有特征"""
        results = []
        for name in self.feature_names:
            if name in current_data and name in self.baselines:
                results.append(self.check_drift(name, current_data[name]))
        return results

# 使用示例
detector = DriftDetector(["sentence_length", "confidence_score"])

# 设置基准（上线初期数据）
baseline_lengths = list(np.random.normal(50, 20, 1000).clip(1, 300))
baseline_conf = list(np.random.beta(5, 2, 1000))
detector.set_baseline("sentence_length", baseline_lengths)
detector.set_baseline("confidence_score", baseline_conf)

# 检测当前数据（模拟分布发生了偏移：句子变长了）
current_lengths = list(np.random.normal(90, 30, 500).clip(1, 300))  # 均值从50变90
current_conf = list(np.random.beta(4, 2, 500))  # 略微偏移

results = detector.check_all({
    "sentence_length": current_lengths,
    "confidence_score": current_conf
})

---

六、结构化日志与请求追踪

import logging
import json
import uuid
import time
from functools import wraps

# 结构化日志格式（方便ELK/云日志系统解析）
class StructuredLogger:
    """输出JSON格式的结构化日志"""
    
    def __init__(self, service_name: str):
        self.service = service_name
        self.logger = logging.getLogger(service_name)
        handler = logging.StreamHandler()
        handler.setFormatter(logging.Formatter('%(message)s'))  # 只输出消息体
        self.logger.addHandler(handler)
        self.logger.setLevel(logging.INFO)
    
    def _log(self, level: str, event: str, **kwargs):
        record = {
            "timestamp": time.strftime("%Y-%m-%dT%H:%M:%S"),
            "service": self.service,
            "level": level,
            "event": event,
            **kwargs
        }
        log_method = getattr(self.logger, level.lower(), self.logger.info)
        log_method(json.dumps(record, ensure_ascii=False))
    
    def request(self, request_id: str, text: str, label: str, 
                confidence: float, latency_ms: float):
        """记录预测请求"""
        self._log("INFO", "prediction",
                  request_id=request_id,
                  text_length=len(text),
                  label=label,
                  confidence=confidence,
                  latency_ms=latency_ms)
    
    def error(self, request_id: str, error_type: str, message: str):
        """记录错误"""
        self._log("ERROR", "prediction_error",
                  request_id=request_id,
                  error_type=error_type,
                  message=message)

def trace_request(logger: StructuredLogger):
    """请求追踪装饰器：自动记录每个请求的进出和耗时"""
    def decorator(func):
        @wraps(func)
        async def wrapper(request, *args, **kwargs):
            request_id = str(uuid.uuid4())[:8]
            start = time.perf_counter()
            
            try:
                result = await func(request, *args, **kwargs)
                latency = (time.perf_counter() - start) * 1000
                logger.request(
                    request_id=request_id,
                    text=getattr(request, 'text', ''),
                    label=result.get('label', ''),
                    confidence=result.get('confidence', 0),
                    latency_ms=round(latency, 2)
                )
                return result
            except Exception as e:
                logger.error(request_id, type(e).__name__, str(e))
                raise
        return wrapper
    return decorator

# 日志输出示例：
# {"timestamp":"2026-05-28T16:00:00","service":"sentiment-api","level":"INFO",
#  "event":"prediction","request_id":"a1b2c3d4","text_length":18,
#  "label":"正面","confidence":0.8734,"latency_ms":12.3}

---

七、回滚策略

回滚触发条件与操作流程：

  触发条件（满足任一即回滚）：
  ├── 错误率 > 5%（持续2分钟）
  ├── P99延迟 > 1000ms（持续5分钟）
  ├── 业务准确率下降 > 5%
  └── 手动触发（管理员决策）

  回滚操作（目标：5分钟内完成）：
  
  ① 确认需要回滚（人工 or 自动告警触发）
       │
  ② 切换流量（金丝雀比例回 0% / 负载均衡指向旧版本）
       │
  ③ 验证旧版本服务正常（健康检查 + 快速冒烟测试）
       │
  ④ 记录事故（时间线、影响范围、触发原因）
       │
  ⑤ 根因分析（防止同类问题再次发生）

---

八、考试重点

核心概念辨析

概念	核心特征	与其他策略的区别
蓝绿部署	两套环境同时存在，流量一次性切换	切换速度最快，但资源成本最高
金丝雀发布	逐步放量（5%→100%），验证无误后扩大	风险最低，AI模型上线首选
影子测试	新版本并行接收流量但不返回结果	零风险测试，但无法评估用户体验
滚动更新	逐台替换实例	资源效率最高，回滚最慢
PSI	衡量特征分布偏移的指标	>0.25说明分布显著偏移，需重训
P99延迟	99%请求的最大延迟	比平均延迟更能反映极端体验

高频考题 Q&A

Q1：AI模型上线时最推荐使用哪种发布策略？为什么？
A：推荐金丝雀发布。原因：AI模型在生产环境中可能出现训练/测试集无法覆盖的数据，通过先服务少量真实用户（5%-10%），可以在影响最小化的前提下验证模型的真实效果，一旦发现问题可以快速回滚，无需全量用户受影响。

Q2：PSI值0.28意味着什么？应该怎么处理？
A：PSI = 0.28 > 0.25，说明当前输入数据的分布与训练时显著不同（发生了显著的分布偏移）。应该：①调查数据偏移原因（用户行为变化/数据质量问题）；②考虑使用当前数据进行模型重训；③在重训前，可以降低模型服务的置信度阈值，增加人工审核比例。

Q3：模型监控中，为什么要使用滑动窗口而非全量历史数据？
A：①时效性：近期数据更能反映当前模型状态，历史数据（如半年前的）可能是正常的；②计算效率：全量历史数据随时间增长，计算代价越来越高；③敏感性：滑动窗口对最近的变化更敏感，能更快触发告警。

Q4：以下关于发布策略的说法，哪个是错误的？
A. 金丝雀发布可以逐步调整流量比例
B. 蓝绿部署需要同时维护两套完整的生产环境
C. 影子测试中，新模型的结果会直接返回给用户
D. 滚动更新是Kubernetes的默认发布策略

答案：C。影子测试中新模型的结果不对外返回，只用于内部评估对比，对用户完全透明无感知。

Q5：P50、P95、P99延迟各代表什么含义？
A：

• P50（中位数）：50%的请求在此延迟以内，代表"正常用户"的体验
• P95：95%的请求在此延迟以内，代表"大部分用户"的体验
• P99：99%的请求在此延迟以内，代表"几乎所有用户"的最差体验——通常作为SLA（服务等级协议）的承诺指标

---

九、备考贴士

📝 发布策略记忆法：
蓝绿（快切/贵）→ 金丝雀（慢放/AI首选）→ 影子（零风险/不返回）→ 滚动（逐台换/Kubernetes默认）

🎯 PSI三档记忆：
<0.1稳定（绿灯）→ 0.1~0.25轻微（黄灯）→ >0.25显著需重训（红灯）

💡 高频考点：金丝雀发布是AI上线首选；影子测试结果不返回用户；P99是SLA核心指标

⚠️ 易错点：影子测试≠金丝雀（影子对用户不可见）；PSI用于输入特征漂移，不直接衡量准确率