LESSON 30分

ストーリー

田
田中VPoE
ログ収集と品質メトリクスの基盤ができた。次はドリフト検出だ。NetShop社のレコメンドAIの精度が先月から徐々に低下しているが、原因がわからない状態だ
あなた
「ドリフト」というのは、時間とともに何かが変化していくことですか？
あ
田
田中VPoE
その通り。AIシステムでは主に2種類のドリフトがある。入力データの分布が変化する「データドリフト」と、入力と正解の関係が変化する「コンセプトドリフト」だ。ECサイトではセール時期や季節変動でこれらが頻繁に起きる
あなた
ドリフトを検知して早期に対処する仕組みを作りたいです
あ

ドリフトの種類

2種類のドリフト

ドリフト定義原因例影響
データドリフト入力データの分布が変化新商品カテゴリの追加、季節変動、ユーザー層の変化モデルが未知のパターンに遭遇
コンセプトドリフト入力と正解の関係が変化トレンドの変化、ユーザーの嗜好変化、競合の影響学習時の知識が陳腐化

ドリフトの発生パターン

突発的ドリフト:
  品質 ████████████▁▁▁▁▁▁▁▁
  原因: モデルAPIの仕様変更、大規模障害

漸進的ドリフト:
  品質 ████████████████▇▇▇▆▆▅▅▄▄
  原因: 季節変動、ユーザー層の緩やかな変化

周期的ドリフト:
  品質 ████▇▇████▇▇████
  原因: セール期間、年末年始、曜日パターン

再帰的ドリフト:
  品質 ████▇▇████▅▅████
  原因: 不定期のトレンド変化

データドリフトの検出

統計的検定によるドリフト検出

import numpy as np
from scipy import stats

class DataDriftDetector:
    """データドリフトの検出"""

    def __init__(self, reference_data: np.ndarray):
        self.reference = reference_data

    def detect_ks_test(self, current_data: np.ndarray, threshold: float = 0.05) -> dict:
        """Kolmogorov-Smirnov検定によるドリフト検出"""
        statistic, p_value = stats.ks_2samp(self.reference, current_data)
        return {
            "test": "KS Test",
            "statistic": round(statistic, 4),
            "p_value": round(p_value, 4),
            "drift_detected": p_value < threshold,
            "severity": self._classify_severity(p_value)
        }

    def detect_psi(self, current_data: np.ndarray, bins: int = 10) -> dict:
        """Population Stability Index（PSI）によるドリフト検出"""
        ref_hist, bin_edges = np.histogram(self.reference, bins=bins, density=True)
        cur_hist, _ = np.histogram(current_data, bins=bin_edges, density=True)

        # ゼロ除算回避
        ref_hist = np.clip(ref_hist, 1e-6, None)
        cur_hist = np.clip(cur_hist, 1e-6, None)

        # PSI計算
        psi = np.sum((cur_hist - ref_hist) * np.log(cur_hist / ref_hist))

        return {
            "test": "PSI",
            "psi_value": round(float(psi), 4),
            "drift_detected": psi > 0.2,
            "interpretation": (
                "変化なし" if psi < 0.1 else
                "軽微な変化" if psi < 0.2 else
                "重大な変化"
            )
        }

    def _classify_severity(self, p_value: float) -> str:
        if p_value < 0.001:
            return "HIGH"
        elif p_value < 0.05:
            return "MEDIUM"
        else:
            return "LOW"

PSI（Population Stability Index）の解釈

PSI値解釈対応
< 0.1有意な変化なし通常運用
0.1 - 0.2軽微な変化あり監視強化
> 0.2重大な分布変化原因調査、モデル再評価

コンセプトドリフトの検出

性能ベースの検出

class ConceptDriftDetector:
    """コンセプトドリフトの検出"""

    def __init__(self, window_size: int = 1000):
        self.window_size = window_size
        self.performance_history: list[float] = []

    def add_observation(self, performance_score: float):
        """性能スコアを追加"""
        self.performance_history.append(performance_score)

    def detect_adwin(self, delta: float = 0.002) -> dict:
        """ADWIN（Adaptive Windowing）による検出"""
        if len(self.performance_history) < self.window_size * 2:
            return {"drift_detected": False, "reason": "データ不足"}

        # ウィンドウを分割して平均を比較
        recent = self.performance_history[-self.window_size:]
        previous = self.performance_history[-self.window_size*2:-self.window_size]

        recent_mean = np.mean(recent)
        previous_mean = np.mean(previous)
        diff = abs(recent_mean - previous_mean)

        # 統計的有意差の判定
        combined_std = np.sqrt(
            np.var(recent) / len(recent) + np.var(previous) / len(previous)
        )
        threshold = combined_std * 2  # 2シグマ基準

        return {
            "drift_detected": diff > threshold,
            "previous_mean": round(previous_mean, 4),
            "recent_mean": round(recent_mean, 4),
            "difference": round(diff, 4),
            "threshold": round(threshold, 4),
            "direction": "低下" if recent_mean < previous_mean else "向上"
        }

LLM特有のドリフト検出

プロンプト-レスポンスドリフト

検出対象指標検出方法
入力プロンプトの変化プロンプト長、トピック分布埋め込みベクトルのクラスタリング
出力品質の変化ROUGE、一貫性スコアLLM-as-Judgeで自動評価
レイテンシの変化応答時間のP50/P95統計的管理図
コストの変化トークン単価、総コスト予算アラート

埋め込みベースのドリフト検出

from openai import OpenAI
import numpy as np

client = OpenAI()

class EmbeddingDriftDetector:
    """埋め込みベクトルを使ったプロンプトドリフト検出"""

    def __init__(self):
        self.reference_embeddings: list[list[float]] = []
        self.reference_centroid: np.ndarray | None = None

    def set_reference(self, reference_prompts: list[str]):
        """基準期間のプロンプトを設定"""
        embeddings = self._get_embeddings(reference_prompts)
        self.reference_embeddings = embeddings
        self.reference_centroid = np.mean(embeddings, axis=0)

    def detect_drift(self, current_prompts: list[str], threshold: float = 0.15) -> dict:
        """現在のプロンプトのドリフトを検出"""
        current_embeddings = self._get_embeddings(current_prompts)
        current_centroid = np.mean(current_embeddings, axis=0)

        # コサイン距離でドリフト量を計測
        cosine_sim = np.dot(self.reference_centroid, current_centroid) / (
            np.linalg.norm(self.reference_centroid) * np.linalg.norm(current_centroid)
        )
        drift_score = 1 - cosine_sim

        return {
            "drift_score": round(float(drift_score), 4),
            "drift_detected": drift_score > threshold,
            "interpretation": (
                "安定" if drift_score < 0.05 else
                "軽微な変化" if drift_score < 0.15 else
                "重大な変化"
            )
        }

    def _get_embeddings(self, texts: list[str]) -> np.ndarray:
        response = client.embeddings.create(
            model="text-embedding-3-small",
            input=texts
        )
        return np.array([e.embedding for e in response.data])

まとめ

ドリフト種類検出手法主要指標
データドリフトKS検定、PSI入力分布の変化量
コンセプトドリフトADWIN、性能監視予測性能の低下
プロンプトドリフト埋め込み距離プロンプト分布の変化

チェックリスト

  • データドリフトとコンセプトドリフトの違いを理解した
  • PSIによるデータドリフト検出の実装と解釈ができる
  • 性能ベースのコンセプトドリフト検出を理解した
  • LLM特有のプロンプトドリフト検出方法を把握した

次のステップへ

次はアラート設計とダッシュボード構築を学びます。

推定読了時間: 30分