Кто должен следить за моделью на проекте?

Кто должен следить за моделью на проекте?

Введение

После деплоя модели начинается долгая фаза production monitoring (мониторинга). На основе 10+ лет опыта раскажу о том, как правильно организовать надзор и кто за что отвечает.

Ответ: Команда, не один человек

Задача не может лежать на одном человеке. Нужно распределить ответственность между:

1. Data Scientist/ML Engineer (Ответственность: точность, деградация модели)

Что контролирует:

• Метрики предсказания (точность, F1, ROC-AUC)
• Data drift (изменение распределения входных данных)
• Model drift (падение точности на свежих данных)
• Feature quality (пропуски, выбросы, аномалии)

Инструменты:

import pandas as pd
import numpy as np
from scipy.stats import ks_2samp
import datetime

class ModelMonitor:
    def __init__(self, baseline_X, baseline_y):
        self.baseline_X = baseline_X
        self.baseline_y = baseline_y
        self.baseline_mean = baseline_X.mean()
        self.baseline_std = baseline_X.std()
    
    def check_data_drift(self, current_X, threshold=0.05):
        """Проверка data drift через Kolmogorov-Smirnov тест"""
        results = {}
        for col in current_X.columns:
            statistic, p_value = ks_2samp(
                self.baseline_X[col],
                current_X[col]
            )
            results[col] = {
                'p_value': p_value,
                'drift_detected': p_value < threshold
            }
        return results
    
    def check_distribution_shift(self, current_X):
        """Проверка сдвига среднего и дисперсии"""
        shifts = {}
        for col in current_X.columns:
            current_mean = current_X[col].mean()
            current_std = current_X[col].std()
            
            mean_shift = abs(current_mean - self.baseline_mean[col])
            std_shift = abs(current_std - self.baseline_std[col])
            
            shifts[col] = {
                'mean_shift': mean_shift,
                'std_shift': std_shift
            }
        return shifts
    
    def check_data_quality(self, current_X):
        """Проверка качества данных"""
        quality_report = {
            'missing_values': current_X.isnull().sum(),
            'duplicates': current_X.duplicated().sum(),
            'outliers': self._detect_outliers(current_X)
        }
        return quality_report
    
    def _detect_outliers(self, X):
        """Обнаружение выбросов через IQR"""
        outliers = {}
        for col in X.select_dtypes(include=[np.number]).columns:
            Q1 = X[col].quantile(0.25)
            Q3 = X[col].quantile(0.75)
            IQR = Q3 - Q1
            outlier_mask = (X[col] < Q1 - 1.5*IQR) | (X[col] > Q3 + 1.5*IQR)
            outliers[col] = outlier_mask.sum()
        return outliers

2. DevOps/MLOps инженер (Ответственность: инфраструктура, логирование)

Что контролирует:

• Latency (время отклика модели)
• Throughput (количество предсказаний в секунду)
• Uptime (доступность сервиса)
• Resource utilization (CPU, память, диск)
• Ошибки и исключения

Инструменты:

# Логирование с временем отклика
import logging
import time
from datetime import datetime

logger = logging.getLogger('model_serving')

def predict_with_monitoring(model, X, request_id):
    start_time = time.time()
    
    try:
        predictions = model.predict(X)
        latency = time.time() - start_time
        
        logger.info(
            f"Prediction successful",
            extra={
                'request_id': request_id,
                'latency_ms': latency * 1000,
                'samples': len(X),
                'timestamp': datetime.utcnow().isoformat()
            }
        )
        
        return predictions
    
    except Exception as e:
        latency = time.time() - start_time
        logger.error(
            f"Prediction failed: {str(e)}",
            extra={
                'request_id': request_id,
                'latency_ms': latency * 1000,
                'error': str(e),
                'timestamp': datetime.utcnow().isoformat()
            }
        )
        raise

3. Product Manager (Ответственность: бизнес-метрики)

Что контролирует:

• Conversion rate
• User engagement
• Revenue impact
• A/B тест результаты
• Customer satisfaction

Вопросы:

• Люди кликают на рекомендации модели?
• Улучшилась ли конверсия?
• Выросла ли выручка?

4. Data Engineer (Ответственность: качество данных в pipeline)

Что контролирует:

• Feature pipeline работает без ошибок
• Свежесть данных (не устаревшие признаки)
• Целостность данных в БД
• Согласованность данных между источниками

Инструменты (Great Expectations):

import great_expectations as ge

df = ge.read_csv('features.csv')

# Определяем ожидания
expectations = ge.dataset.PandasDataset(df)
expectations.expect_table_row_count_to_be_between(min_value=1000, max_value=1000000)
expectations.expect_column_values_to_not_be_null('user_id')
expectations.expect_column_values_to_be_in_set('status', ['active', 'inactive'])
expectations.expect_column_values_to_be_of_type('age', 'int64')

# Валидируем
validation_results = expectations.validate()
print(f"Validation success: {validation_results['success']}")

Структура мониторинга (трёхуровневая пирамида)

Уровень 3: Бизнес-метрики (Product, Managers)
           ↓ Какой-то параметр упал?

Уровень 2: Модель-метрики (DS, ML Engineer)
           ↓ Точность упала? Data drift?

Уровень 1: Система-метрики (DevOps, Data Engineer)
           ↓ Latency? Ошибки? Quality?

Доска ответственности (RACI matrix)

Задача                      | Responsible | Accountable | Consulted | Informed
----------------------------------------------------------------------
Отслеживать точность        |     DS      |     DS      |   DM      |   PM
Отслеживать data drift       |     DS      |     DS      |   DE      |   DM
Отслеживать latency          |   DevOps    |   DevOps    |   DS      |   PM
Отслеживать uptime           |   DevOps    |   DevOps    |     -     |   PM
Проверять качество данных    |     DE      |     DE      |   DS      |   PM
Проверять A/B тест           |     PM      |     PM      |   DS      |   DM
Ретрейнить модель           |     DS      |     DS      |   DE      |   DM
Отновить код на production   |   DevOps    |   DevOps    |   DS      |     -

Практический пример: Полный мониторинг

import prometheus_client
from prometheus_client import Counter, Histogram, Gauge
import logging

class FullProductionMonitoring:
    def __init__(self):
        # Метрики System Level (DevOps)
        self.prediction_latency = Histogram(
            'model_prediction_latency_seconds',
            'Time spent on model prediction',
            buckets=(0.01, 0.05, 0.1, 0.5, 1.0)
        )
        
        self.prediction_errors = Counter(
            'model_prediction_errors_total',
            'Total prediction errors',
            ['error_type']
        )
        
        self.model_memory_usage = Gauge(
            'model_memory_usage_bytes',
            'Memory used by model'
        )
        
        # Метрики Model Level (DS)
        self.prediction_confidence = Histogram(
            'model_prediction_confidence',
            'Model confidence scores'
        )
        
        self.feature_drift = Gauge(
            'feature_drift_score',
            'Data drift score',
            ['feature_name']
        )
        
        # Метрики Business Level (PM)
        self.recommended_item_clicks = Counter(
            'recommended_item_clicks_total',
            'User clicks on recommended items'
        )
        
        self.recommendation_conversion = Gauge(
            'recommendation_conversion_rate',
            'Conversion rate from recommendations'
        )
    
    def log_prediction(self, latency, confidence, prediction_id):
        """Логирование предсказания на все уровни"""
        self.prediction_latency.observe(latency)
        self.prediction_confidence.observe(confidence)
        
        logger = logging.getLogger('monitoring')
        logger.info(
            f"Prediction",
            extra={
                'prediction_id': prediction_id,
                'latency_ms': latency * 1000,
                'confidence': confidence
            }
        )
    
    def log_error(self, error_type, details):
        """Логирование ошибок"""
        self.prediction_errors.labels(error_type=error_type).inc()
        
        logger = logging.getLogger('monitoring')
        logger.error(
            f"Prediction error: {error_type}",
            extra={'details': details}
        )

Алерты и оповещения

# Alerting rules (Prometheus)
alert_rules = {
    'HIGH_LATENCY': {
        'condition': 'p95_latency > 1.0',  # 95-й перцентиль > 1 сек
        'severity': 'warning',
        'owner': 'DevOps'
    },
    'DATA_DRIFT_DETECTED': {
        'condition': 'drift_score > 0.2',
        'severity': 'warning',
        'owner': 'Data Scientist'
    },
    'ACCURACY_DROP': {
        'condition': 'accuracy < 0.85',
        'severity': 'critical',
        'owner': 'Data Scientist'
    },
    'HIGH_ERROR_RATE': {
        'condition': 'error_rate > 0.01',
        'severity': 'critical',
        'owner': 'DevOps'
    },
    'CONVERSION_DROP': {
        'condition': 'conversion_rate < baseline * 0.9',  # Упала на 10%
        'severity': 'critical',
        'owner': 'Product Manager'
    }
}

Dashboard (для всей команды)

Основной dashboard должен содержать:

1. Real-time metrics:

   • Requests per second
   • Latency (p50, p95, p99)
   • Error rate

2. Model Performance:

   • Accuracy (rolling 7 дней)
   • Precision, Recall, F1 (vs baseline)
   • Feature importance изменения

3. Data Quality:

   • Missing values %
   • Outliers detected
   • Data drift score

4. Business Metrics:

   • Conversion rate
   • Click-through rate
   • Revenue impact

Частота проверок

мониторинг_сроки = {
    'latency': '1 минута',
    'errors': '5 минут',
    'accuracy': '1 час',
    'data_drift': '1 день',
    'feature_stability': '1 неделя',
    'business_metrics': '1 день',
    'полная_переоценка': '1 месяц'
}

Что делать при проблемах?

Сценарий 1: Data Drift обнаружен

• DS: Анализирует, почему произошёл drift
• DE: Проверяет качество данных в pipeline
• DM: Решает, нужен ли ретрейн
• DevOps: Готовит среду для переобучения

Сценарий 2: Точность упала

• DS: Проверяет метрики на разных подгруппах
• DE: Проверяет целостность feature pipeline
• PM: Сообщает о проблеме бизнесу
• DevOps: Может откатить к предыдущей версии

Сценарий 3: High Latency

• DevOps: Масштабирует сервис (добавляет реплики)
• DS: Оптимизирует модель (квантизация, дистилляция)
• DE: Оптимизирует feature computation

Заключение

Ответ: Мониторинг модели — это ответственность всей команды, где:

• Data Scientist смотрит на метрики модели и data drift
• DevOps контролирует инфраструктуру и latency
• Data Engineer гарантирует качество данных
• Product Manager отслеживает бизнес-метрики
• DM интегрирует все сигналы и принимает решения

Без такого структурированного подхода модели деградируют незамеченными и наносят ущерб бизнесу. После деплоя начинается самая ответственная работа!