Skip to content
🔗 分享本题
查看我的学习进度 →

模型漂移通过线上分段监控和固定回归集定位输入规则行为输出成本变化,再灰度修复

🧠 记忆锚点:先分清哪里漂;用线上监控加固定回归集定位,再灰度修复。

💡 答案要点

模型漂移 = 部署后 LLM 应用性能随时间下降

三种漂移类型:

类型原因症状
数据漂移用户输入分布变化(新词汇、新场景)覆盖率下降,更多兜底回答
概念漂移业务规则/知识库变化但模型没更新回答基于过时信息
语义漂移对话主题慢慢偏离预设范围越来越多离题回答

检测方案:

展开 Python 代码示例(72 行)
python
import numpy as np
from scipy.stats import ks_2samp
from datetime import datetime, timedelta

class ModelDriftDetector:
    def __init__(self, baseline_window_days=7):
        self.baseline = {}  # 基线指标
        self.baseline_window = baseline_window_days

    def collect_metrics(self, date: str) -> dict:
        """收集每日指标"""
        logs = get_agent_logs(date)
        return {
            "success_rate": sum(1 for l in logs if l["success"]) / len(logs),
            "avg_confidence": np.mean([l["confidence"] for l in logs]),
            "avg_tokens": np.mean([l["tokens"] for l in logs]),
            "topic_distribution": self.get_topic_dist(logs),
            "embedding_centroid": self.get_embedding_centroid(logs),
        }

    def detect_drift(self, current_metrics: dict) -> dict:
        drift_signals = {}

        # 1. 成功率统计检验
        if "success_rate" in self.baseline:
            baseline_rate = self.baseline["success_rate"]
            current_rate = current_metrics["success_rate"]
            if abs(current_rate - baseline_rate) > 0.05:  # 5% 阈值
                drift_signals["success_rate_drift"] = {
                    "baseline": baseline_rate,
                    "current": current_rate,
                    "delta": current_rate - baseline_rate
                }

        # 2. 话题分布漂移(KS 检验)
        if "topic_distribution" in self.baseline:
            ks_stat, p_value = ks_2samp(
                self.baseline["topic_distribution"],
                current_metrics["topic_distribution"]
            )
            if p_value < 0.05:  # 统计显著
                drift_signals["topic_drift"] = {
                    "ks_stat": ks_stat,
                    "p_value": p_value
                }

        # 3. 语义漂移(Embedding 中心偏移)
        if "embedding_centroid" in self.baseline:
            cosine_sim = self.cosine_similarity(
                self.baseline["embedding_centroid"],
                current_metrics["embedding_centroid"]
            )
            if cosine_sim < 0.85:  # 余弦相似度阈值
                drift_signals["semantic_drift"] = {
                    "similarity": cosine_sim
                }

        return drift_signals

    def alert_if_needed(self, drift_signals: dict):
        if not drift_signals:
            return

        severity = "warning"
        if len(drift_signals) >= 2:
            severity = "critical"

        send_alert(
            title=f"[{severity.upper()}] 检测到模型漂移",
            body=f"漂移信号:{drift_signals}",
            channel="#ai-ops"
        )

应对策略:

漂移类型应对方案
数据漂移更新知识库,补充新场景训练数据
概念漂移重新微调/更新 RAG 知识库
语义漂移强化 System Prompt 的话题限制
模型版本变化A/B 测试评估新版本再全量切换

面试话术:

"模型漂移是 LLM 应用上线后最容易被忽视的问题。我的监控是三层:1)成功率监控,跌超5%自动告警;2)KS检验话题分布,检测用户使用场景是否变化;3)Embedding中心偏移,检测语义漂移。发现漂移后的处置流程是:先判断是数据漂移还是概念漂移——数据漂移更新知识库,概念漂移就要重新微调。关键是要有'基线'概念,没有基线就检测不了漂移。"