Как модель выпускалась в production в рабочем ML-проекте?

Выпуск модели в production. Практический опыт

Этап production deployment одна из самых сложных частей ML-проекта. Я опишу полный цикл, основываясь на опыте с реальными проектами в области финтеха и рекомендательных систем.

Фаза 1: Подготовка модели

1. Валидация и тестирование

Перед отправкой в production модель должна пройти строгую валидацию:

from sklearn.model_selection import cross_val_score
from sklearn.metrics import roc_auc_score, precision_recall_curve
import pickle

# Кросс-валидация на всех фолдах
scores = cross_val_score(
    model, X_train, y_train,
    scoring=roc_auc,
    cv=5
)

print(f"ROC-AUC: {scores.mean():.4f} +/- {scores.std():.4f}")

# Проверка на тестовом наборе
test_auc = roc_auc_score(y_test, model.predict_proba(X_test)[:, 1])
print(f"Test AUC: {test_auc:.4f}")

# Проверка на production-подобных данных (если есть)
if hasattr(data, production_like_data):
    prod_auc = roc_auc_score(
        y_prod, 
        model.predict_proba(X_prod)[:, 1]
    )
    print(f"Production-like AUC: {prod_auc:.4f}")

2. Версионирование и сериализация

Всегда сохраняй версию модели и зависимости:

import joblib
import json
from datetime import datetime

# Сохраняем модель
model_version = "v1.2.3"
model_path = f"models/model_{model_version}_{datetime.now().strftime(%Y%m%d_%H%M%S)}.pkl"
joblib.dump(model, model_path)

# Сохраняем метаданные
metadata = {
    "version": model_version,
    "trained_date": datetime.now().isoformat(),
    "train_auc": float(train_auc),
    "test_auc": float(test_auc),
    "features": feature_names,
    "sklearn_version": "1.3.0",
    "python_version": "3.10.0",
    "hyperparameters": model.get_params()
}

with open(f"models/metadata_{model_version}.json", "w") as f:
    json.dump(metadata, f, indent=2)

Фаза 2: Подготовка инфраструктуры

1. Контейнеризация (Docker)

FROM python:3.10-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY models/ ./models/
COPY src/ ./src/

EXPOSE 8000

CMD ["python", "-m", "uvicorn", "src.api:app", "--host", "0.0.0.0", "--port", "8000"]

2. API сервер (FastAPI)

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import joblib
import numpy as np
from typing import List

app = FastAPI()

# Загружаем модель один раз
model = joblib.load("models/model_v1.2.3.pkl")
metadata = json.load(open("models/metadata_v1.2.3.json"))

class PredictionInput(BaseModel):
    features: List[float]
    user_id: str = None  # Для логирования

class PredictionOutput(BaseModel):
    prediction: float
    probability: float
    model_version: str
    confidence: float

@app.post("/predict")
async def predict(input_data: PredictionInput) -> PredictionOutput:
    try:
        # Преобразуем в numpy array
        X = np.array(input_data.features).reshape(1, -1)
        
        # Проверяем количество признаков
        if X.shape[1] != len(metadata["features"]):
            raise HTTPException(
                status_code=400,
                detail=f"Expected {len(metadata[features])} features, got {X.shape[1]}"
            )
        
        # Предсказание
        prediction = model.predict(X)[0]
        probability = model.predict_proba(X)[0, 1]
        
        # Вычисляем confidence (максимальная вероятность класса)
        confidence = max(model.predict_proba(X)[0])
        
        return PredictionOutput(
            prediction=int(prediction),
            probability=float(probability),
            model_version=metadata["version"],
            confidence=float(confidence)
        )
    
    except Exception as e:
        # Логируем ошибку
        print(f"Error in prediction for user {input_data.user_id}: {str(e)}")
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/health")
async def health_check():
    return {"status": "healthy", "model_version": metadata["version"]}

Фаза 3: Развёртывание

1. Canary deployment (постепенный rollout)

Никогда не отправляй 100% трафика новой модели сразу!

# День 1: 5% трафика на новую модель
kubectl set image deployment/ml-service \
  model-server=ml-service:v1.2.3 \
  --record

# Настройка трафика: 95% старая, 5% новая
kubectl apply -f - << EOF
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: ml-service
spec:
  hosts:
  - ml-service
  http:
  - match:
    - uri:
        regex: ".*"
    route:
    - destination:
        host: ml-service
        subset: v1
      weight: 95
    - destination:
        host: ml-service
        subset: v1.2.3
      weight: 5
EOF

2. Мониторинг и метрики

from prometheus_client import Counter, Histogram, start_http_server
import time

# Метрики
prediction_counter = Counter(
    "ml_predictions_total",
    "Total predictions",
    ["model_version", "status"]
)

prediction_latency = Histogram(
    "ml_prediction_latency_seconds",
    "Prediction latency",
    ["model_version"]
)

@app.post("/predict")
async def predict(input_data: PredictionInput) -> PredictionOutput:
    start_time = time.time()
    
    try:
        X = np.array(input_data.features).reshape(1, -1)
        prediction = model.predict(X)[0]
        probability = model.predict_proba(X)[0, 1]
        
        prediction_counter.labels(
            model_version=metadata["version"],
            status="success"
        ).inc()
        
        latency = time.time() - start_time
        prediction_latency.labels(
            model_version=metadata["version"]
        ).observe(latency)
        
        return PredictionOutput(
            prediction=int(prediction),
            probability=float(probability),
            model_version=metadata["version"],
            confidence=max(model.predict_proba(X)[0])
        )
    
    except Exception as e:
        prediction_counter.labels(
            model_version=metadata["version"],
            status="error"
        ).inc()
        raise

# Запускаем Prometheus метрики на порту 8001
if __name__ == "__main__":
    start_http_server(8001)
    uvicorn.run(app, host="0.0.0.0", port=8000)

Фаза 4: Мониторинг в production

1. Проверка data drift

from scipy.stats import ks_2samp

def check_data_drift(current_data, reference_data, threshold=0.05):
    """
    Проверяет, не произошло ли смещение распределения признаков
    """
    drifts = {}
    
    for feature in reference_data.columns:
        statistic, p_value = ks_2samp(
            reference_data[feature],
            current_data[feature]
        )
        
        if p_value < threshold:
            drifts[feature] = {
                "statistic": statistic,
                "p_value": p_value,
                "drifted": True
            }
    
    return drifts

# Проверяем каждый час
import schedule

def monitor_drift():
    current_hour_data = get_last_hour_predictions()
    reference_data = load_training_data()
    
    drifts = check_data_drift(current_hour_data, reference_data)
    
    if drifts:
        send_alert(f"Data drift detected in features: {list(drifts.keys())}")
        retrain_model()

schedule.every().hour.do(monitor_drift)

2. A/B тестирование

# 50% пользователей видят старую модель, 50% новую
# Сравниваем метрики
old_model_auc = 0.85
new_model_auc = 0.87
p_value = calculate_significance(old_model_auc, new_model_auc)

if p_value < 0.05 and new_model_auc > old_model_auc:
    print("New model is significantly better, rolling out 100%")
else:
    print("Rollback to old model")

Лучшие практики

Обязательный чеклист перед production:

• Тестовый AUC выше, чем у текущей production модели
• Проверка на разных срезах данных (по возрасту, полу, географии)
• Latency < 100ms на среднем запросе
• Все зависимости задокументированы
• Метрики собираются и мониторятся
• Есть механизм rollback
• Версионирование и логирование включены

Процесс может быть долгим (неделя-месяц canary deployment), но это критично для надёжности production систем.