Моніторинг моделей

Моделі деградують в продакшені. Розподіли даних змінюються, поведінка користувачів еволюціонує, апстрім-системи оновлюються. Без моніторингу ви не дізнаєтеся про проблеми, поки бізнес-метрики не впадуть. Хороший моніторинг виявляє проблеми рано - до того, як помітять клієнти.

Де використовується

Моніторинг необхідний для будь-якого ML в продакшені:

• Усі ML-системи: Відстеження продуктивності, виявлення дрейфу
• Високовідповідальні: Охорона здоров'я, фінанси, автономні системи
• Real-time: Рекомендації, фрод, ціноутворення
• Batch: Прогнозування, скорингові пайплайни
• Регуляторні: Комплаєнс та аудит

Продукти: Evidently AI, WhyLabs, Fiddler, Arize, MLflow, власні рішення.

Типові сигнали моніторингу

Що ви робите на практиці

1. Визначте базову лінію - Як виглядає "нормальний" стан?
2. Ідентифікуйте ключові метрики - Продуктивність моделі та якість даних
3. Налаштуйте збір даних - Логування входів, виходів, лейблів
4. Впровадьте виявлення дрейфу - Статистичні тести на фічах
5. Налаштуйте алерти - Дієві, не шумні
6. Створіть дашборди - Видимість для команди
7. Створіть плейбуки - Що робити при спрацюванні алертів
8. Автоматизуйте реакції - Тригери перенавчання, фолбеки
9. Регулярно переглядайте - Чи все ще релевантні монітори?

Спочатку базова лінія

Почніть з простого моніторингу перед складними рішеннями:

import pandas as pd
import numpy as np
from scipy import stats

class SimpleMonitor:
    def __init__(self, reference_data):
        """Ініціалізація з референсними/тренувальними даними."""
        self.reference = reference_data
        self.reference_stats = self._compute_stats(reference_data)

    def _compute_stats(self, data):
        return {
            'mean': data.mean(),
            'std': data.std(),
            'min': data.min(),
            'max': data.max(),
            'missing_rate': data.isnull().mean()
        }

    def check_data_quality(self, current_data):
        """Перевірка проблем з якістю даних."""
        issues = []

        current_missing = current_data.isnull().mean()
        ref_missing = self.reference_stats['missing_rate']

        for col in current_data.columns:
            if current_missing[col] > ref_missing[col] * 2:
                issues.append(f"Високий рівень пропусків: {col}")

        return issues

    def check_drift(self, current_data, threshold=0.05):
        """Проста перевірка дрейфу розподілу з KS-тестом."""
        drift_detected = {}

        for col in current_data.columns:
            if current_data[col].dtype in ['float64', 'int64']:
                stat, p_value = stats.ks_2samp(
                    self.reference[col].dropna(),
                    current_data[col].dropna()
                )
                drift_detected[col] = {
                    'drifted': p_value < threshold,
                    'p_value': p_value,
                    'statistic': stat
                }

        return drift_detected

Чому проста базова лінія: - Виявляє основні проблеми - Низькі витрати на підтримку - Легко зрозуміти - Основа для складнішого моніторингу

Виявлення дрейфу даних

Виявляйте, коли змінюються розподіли входів:

from scipy import stats
import numpy as np

def detect_numerical_drift(reference, current, method='ks'):
    """Виявлення дрейфу в числових фічах."""
    if method == 'ks':
        # Тест Колмогорова-Смирнова
        stat, p_value = stats.ks_2samp(reference, current)
    elif method == 'psi':
        # Population Stability Index
        stat = calculate_psi(reference, current)
        p_value = None  # PSI не має p-value
    elif method == 'wasserstein':
        # Відстань Вассерштейна
        stat = stats.wasserstein_distance(reference, current)
        p_value = None

    return {'statistic': stat, 'p_value': p_value}

def calculate_psi(reference, current, bins=10):
    """Розрахунок Population Stability Index."""
    # Створення бінів з референсу
    breakpoints = np.percentile(reference, np.linspace(0, 100, bins + 1))
    breakpoints[0] = -np.inf
    breakpoints[-1] = np.inf

    # Розрахунок пропорцій
    ref_counts = np.histogram(reference, bins=breakpoints)[0]
    cur_counts = np.histogram(current, bins=breakpoints)[0]

    ref_props = ref_counts / len(reference)
    cur_props = cur_counts / len(current)

    # Уникнення ділення на нуль
    ref_props = np.clip(ref_props, 0.0001, None)
    cur_props = np.clip(cur_props, 0.0001, None)

    psi = np.sum((cur_props - ref_props) * np.log(cur_props / ref_props))
    return psi

def detect_categorical_drift(reference, current, method='chi2'):
    """Виявлення дрейфу в категоріальних фічах."""
    if method == 'chi2':
        # Хі-квадрат тест
        ref_counts = reference.value_counts()
        cur_counts = current.value_counts()

        # Вирівнювання категорій
        all_categories = set(ref_counts.index) | set(cur_counts.index)
        ref_aligned = [ref_counts.get(c, 0) for c in all_categories]
        cur_aligned = [cur_counts.get(c, 0) for c in all_categories]

        stat, p_value = stats.chisquare(cur_aligned, ref_aligned)

    return {'statistic': stat, 'p_value': p_value}

Інтерпретація PSI: - PSI < 0.1: Немає значного дрейфу - 0.1 ≤ PSI < 0.2: Помірний дрейф, потребує дослідження - PSI ≥ 0.2: Значний дрейф, потрібні дії

Моніторинг прогнозів

Відстежуйте виходи моделі з часом:

class PredictionMonitor:
    def __init__(self, baseline_predictions):
        self.baseline_mean = baseline_predictions.mean()
        self.baseline_std = baseline_predictions.std()
        self.baseline_distribution = baseline_predictions

    def check_prediction_drift(self, current_predictions):
        """Перевірка дрейфу прогнозів."""
        current_mean = current_predictions.mean()
        current_std = current_predictions.std()

        # Виявлення зсуву середнього
        mean_shift = abs(current_mean - self.baseline_mean) / self.baseline_std

        # Тест розподілу
        stat, p_value = stats.ks_2samp(
            self.baseline_distribution, 
            current_predictions
        )

        return {
            'mean_shift_zscore': mean_shift,
            'ks_statistic': stat,
            'ks_pvalue': p_value,
            'alert': mean_shift > 2 or p_value < 0.01
        }

    def check_confidence_distribution(self, confidences, threshold=0.5):
        """Моніторинг розподілу оцінок впевненості."""
        low_confidence_rate = (confidences < threshold).mean()
        high_confidence_rate = (confidences > 0.9).mean()

        return {
            'low_confidence_rate': low_confidence_rate,
            'high_confidence_rate': high_confidence_rate,
            'mean_confidence': confidences.mean()
        }

Моніторинг продуктивності

Відстежуйте фактичну продуктивність моделі при наявності лейблів:

from sklearn.metrics import accuracy_score, precision_score, recall_score
import pandas as pd

class PerformanceMonitor:
    def __init__(self, baseline_metrics, window_size=1000):
        self.baseline = baseline_metrics
        self.window_size = window_size
        self.history = []

    def add_prediction(self, prediction, label, timestamp):
        """Додавання пари прогноз-лейбл."""
        self.history.append({
            'prediction': prediction,
            'label': label,
            'timestamp': timestamp
        })

        # Зберігаємо тільки недавню історію
        if len(self.history) > self.window_size * 2:
            self.history = self.history[-self.window_size:]

    def get_current_metrics(self):
        """Розрахунок метрик на нещодавньому вікні."""
        recent = self.history[-self.window_size:]

        if len(recent) < 100:  # Недостатньо даних
            return None

        preds = [h['prediction'] for h in recent]
        labels = [h['label'] for h in recent]

        return {
            'accuracy': accuracy_score(labels, preds),
            'precision': precision_score(labels, preds, zero_division=0),
            'recall': recall_score(labels, preds, zero_division=0),
            'sample_size': len(recent)
        }

    def check_degradation(self, threshold=0.1):
        """Перевірка деградації від базової лінії."""
        current = self.get_current_metrics()

        if current is None:
            return {'status': 'insufficient_data'}

        degradation = {}
        for metric in ['accuracy', 'precision', 'recall']:
            baseline_val = self.baseline.get(metric, 0)
            current_val = current.get(metric, 0)

            if baseline_val > 0:
                drop = (baseline_val - current_val) / baseline_val
                degradation[metric] = {
                    'baseline': baseline_val,
                    'current': current_val,
                    'relative_drop': drop,
                    'alert': drop > threshold
                }

        return degradation

Стратегія алертингу

Хороші алерти дієві, не шумні:

class AlertManager:
    def __init__(self, config):
        self.config = config
        self.alert_history = []

    def evaluate_alert(self, metric_name, value, threshold, comparison='gt'):
        """Оцінка чи повинен спрацювати алерт."""
        if comparison == 'gt':
            triggered = value > threshold
        elif comparison == 'lt':
            triggered = value < threshold
        else:
            triggered = abs(value - threshold) > 0

        # Перевірка на придушення (не спамимо)
        if triggered and self._recently_alerted(metric_name):
            return {'triggered': False, 'reason': 'suppressed'}

        if triggered:
            self._record_alert(metric_name)

        return {
            'triggered': triggered,
            'metric': metric_name,
            'value': value,
            'threshold': threshold
        }

    def _recently_alerted(self, metric_name, cooldown_minutes=60):
        """Перевірка чи нещодавно алертили для цієї метрики."""
        recent = [a for a in self.alert_history 
                  if a['metric'] == metric_name 
                  and (datetime.now() - a['time']).minutes < cooldown_minutes]
        return len(recent) > 0

    def _record_alert(self, metric_name):
        self.alert_history.append({
            'metric': metric_name,
            'time': datetime.now()
        })

# Приклад конфігурації алертів
alert_config = {
    'data_drift': {
        'threshold': 0.2,  # Поріг PSI
        'severity': 'warning',
        'cooldown_minutes': 60
    },
    'accuracy_drop': {
        'threshold': 0.05,  # 5% відносне падіння
        'severity': 'critical',
        'cooldown_minutes': 30
    },
    'latency_p99': {
        'threshold': 500,  # 500мс
        'severity': 'warning',
        'cooldown_minutes': 15
    }
}

Типові помилки та чек-лист

• [ ] Тихі збої: Помилки серверу моделі не захоплюються
• [ ] Затримані лейбли: Неможливо обчислити точність в реальному часі
• [ ] Втома від алертів: Занадто багато хибних алертів
• [ ] Відсутня базова лінія: Немає референсу для порівняння
• [ ] Апстрім зміни: Модифікації пайплайну даних
• [ ] Сезонність: Нормальна варіація позначається як дрейф
• [ ] Розмір вибірки: Недостатньо даних для статистичних тестів

Чек-лист для продакшену

Перед деплоєм: - [ ] Базові метрики обчислені з тренувальних/валідаційних даних - [ ] Референсний датасет збережено для порівняння дрейфу - [ ] Інфраструктура логування для входів/виходів - [ ] Алертинг налаштовано з відповідними порогами - [ ] Дашборди створено для видимості

Після деплою: - [ ] Виявлення дрейфу запущено за розкладом - [ ] Метрики продуктивності відстежуються (при наявності лейблів) - [ ] Моніторинг латентності та пропускної здатності - [ ] Регулярний перегляд релевантності алертів - [ ] Плейбуки оновлюються на основі інцидентів

Автоматизація: - [ ] Автоматичні тригери перенавчання - [ ] Фолбек до попередньої версії моделі - [ ] Ескалація для критичних алертів - [ ] Регулярні звіти валідації моделі

Архітектура моніторингу

Продакшен трафік
       │
       ▼
  ┌─────────┐
  │  Сервіс │ ──── Прогнози ────┐
  │  Моделі │                   │
  └─────────┘                   ▼
       │                 ┌──────────────┐
       │                 │   Сервіс     │
   Входи                 │   Логування  │
       │                 └──────────────┘
       ▼                        │
  ┌─────────┐                   ▼
  │ Feature │           ┌──────────────┐
  │  Store  │           │  Пайплайн    │
  └─────────┘           │  Моніторингу │
                        └──────────────┘
                                │
                 ┌──────────────┼──────────────┐
                 ▼              ▼              ▼
           ┌─────────┐   ┌─────────┐   ┌─────────┐
           │ Виявл.  │   │ Монітор │   │ Менедж. │
           │ Дрейфу  │   │ Продукт.│   │ Алертів │
           └─────────┘   └─────────┘   └─────────┘
                 │              │              │
                 └──────────────┼──────────────┘
                                ▼
                         ┌─────────────┐
                         │  Дашборд    │
                         └─────────────┘

Інструменти та стек

Приклад міні-кейсу

Сценарій: Продакшен модель рекомендацій показує зниження метрик залученості.

Дослідження: 1. Дашборд показує падіння CTR на 15% за 2 тижні 2. Виявлення дрейфу: Зсув фічі користувача days_since_signup 3. Коренева причина: Маркетингова кампанія привела нову когорту користувачів 4. Модель навчена на усталених користувачах, не працює для нових

Реакція: 1. Алерт: CTR впав нижче порогу 2. Дослідження: Ідентифіковано дрейф в демографії користувачів 3. Короткостроково: Сегмент-специфічний фолбек (популярні товари для нових) 4. Довгостроково: Перенавчання моделі з даними нової когорти

Запобігання: - Додано моніторинг дрейфу для фіч когорти користувачів - Налаштовано алерт на пропорцію нових користувачів > порогу - Заплановано щомісячне оновлення моделі

Пов'язані сторінки

• Індекс: ML Field Guide
• Основи MLOps
• A/B тестування для ML
• Виявлення аномалій

Додаткові матеріали

• Документація Evidently AI
• Найкращі практики ML-моніторингу
• Документація WhyLabs
• Гайд Google з ML Production

Спробуйте наживо

Дослідіть пов'язану модель на практиці: відкрийте Model Monitoring Lab — Catch Silent Failures Before Users Do і змінюйте параметри під час роботи. Усе лишається у вашому браузері.