Как развернуть ML модель в production?

Развёртывание ML модели в production

Развёртывание (deployment) модели в production — это процесс переноса обученной модели из экспериментальной среды на боевой сервер, где она будет обслуживать реальные запросы. Это намного сложнее, чем просто сохранить модель в файл. Нужно обеспечить масштабируемость, надёжность, мониторинг и актуальность модели.

Основные этапы

1. Сохранение модели

Первый шаг — правильно сохранить обученную модель:

import pickle
import joblib
import json
from sklearn.ensemble import RandomForestClassifier

# Обучить модель
model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

# Вариант 1: pickle (простой, но может быть несовместим между версиями)
with open('model.pkl', 'wb') as f:
    pickle.dump(model, f)

# Вариант 2: joblib (рекомендуется для sklearn моделей)
joblib.dump(model, 'model.joblib')

# Сохранить также параметры масштабирования
scaler_params = {
    'mean': X_train.mean().tolist(),
    'std': X_train.std().tolist()
}
with open('scaler_params.json', 'w') as f:
    json.dump(scaler_params, f)

2. Создание REST API

Обычно модель оборачивают в API, чтобы её можно было вызывать по HTTP:

from fastapi import FastAPI
from pydantic import BaseModel
import joblib
import numpy as np

app = FastAPI()
model = joblib.load('model.joblib')

class PredictionRequest(BaseModel):
    features: list[float]

class PredictionResponse(BaseModel):
    prediction: float
    probability: float

@app.post('/predict')
async def predict(request: PredictionRequest):
    X = np.array([request.features])
    prediction = model.predict(X)[0]
    probability = model.predict_proba(X)[0].max()
    
    return PredictionResponse(
        prediction=float(prediction),
        probability=float(probability)
    )

@app.get('/health')
async def health():
    return {"status": "ok"}

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

Оберните приложение в Docker контейнер для переносимости и изоляции:

FROM python:3.11-slim

WORKDIR /app

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

COPY model.joblib .
COPY app.py .

EXPOSE 8000

CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

docker build -t my-ml-model:1.0 .
docker run -p 8000:8000 my-ml-model:1.0

4. Развёртывание на сервере

Вариант A: Облачные сервисы

AWS Lambda + API Gateway:

import json
import joblib

model = joblib.load('/tmp/model.joblib')

def lambda_handler(event, context):
    body = json.loads(event['body'])
    features = body['features']
    
    prediction = model.predict([features])[0]
    
    return {
        'statusCode': 200,
        'body': json.dumps({'prediction': float(prediction)})
    }

Google Cloud Run (Docker контейнер):

gcloud run deploy my-ml-model --source . --platform managed --region us-central1

Вариант B: Собственный сервер (VPS)

# SSH на сервер
ssh user@example.com

# Установить Docker
curl -fsSL https://get.docker.com -o get-docker.sh
sudo sh get-docker.sh

# Запустить контейнер с nginx в качестве reverse proxy
docker run -d -p 80:8000 --name ml-api my-ml-model:1.0

# Настроить SSL (Let's Encrypt)
certbot certonly --standalone -d example.com

5. Мониторинг и логирование

Важно отслеживать работу модели в production:

from fastapi import FastAPI
import logging
from datetime import datetime
import json

app = FastAPI()
logger = logging.getLogger(__name__)

# Логирование запросов и предсказаний
@app.post('/predict')
async def predict(request: PredictionRequest):
    start_time = datetime.now()
    
    try:
        X = np.array([request.features])
        prediction = model.predict(X)[0]
        
        # Логировать успешное предсказание
        logger.info(json.dumps({
            'timestamp': start_time.isoformat(),
            'features': request.features,
            'prediction': float(prediction),
            'latency_ms': (datetime.now() - start_time).total_seconds() * 1000
        }))
        
        return {'prediction': float(prediction)}
    
    except Exception as e:
        logger.error(f'Prediction error: {str(e)}')
        raise

6. Версионирование и Canary развёртывание

Обновление модели должно быть безопасным:

# Использовать версии моделей
models = {
    'v1.0': joblib.load('model_v1.0.joblib'),
    'v1.1': joblib.load('model_v1.1.joblib'),
}

config = {'active_model': 'v1.0', 'v1.1_traffic': 0.1}  # 10% трафика на новую версию

def get_model():
    import random
    if random.random() < config['v1.1_traffic']:
        return models['v1.1']
    return models[config['active_model']]

Архитектура production ML системы

┌─────────────────┐
│  Клиент          │
└────────┬─────────┘
         │ HTTP запрос
         ▼
┌─────────────────┐
│  Load Balancer  │ (распределение нагрузки)
└────────┬─────────┘
         │
    ┌────┴────┐
    ▼         ▼
┌────────┐ ┌────────┐
│ API v1 │ │ API v2 │ (несколько инстансов)
└────┬───┘ └───┬────┘
     │         │
     ├─────────┤
     ▼
┌──────────────────────┐
│  Model Server        │
│  (FastAPI, Flask)    │
└────┬─────────────────┘
     │
     ▼
┌──────────────────────┐
│  Модель (pickle)     │
│  Кэш (Redis)         │
└──────────────────────┘

Чеклист production развёртывания

# 1. Валидация входных данных
@app.post('/predict')
async def predict(request: PredictionRequest):
    if len(request.features) != 10:
        raise ValueError('Expected 10 features')
    
    if any(x < 0 for x in request.features):
        raise ValueError('All features must be non-negative')
    
    return model.predict([request.features])[0]

# 2. Обработка ошибок
try:
    prediction = model.predict(X)
except Exception as e:
    logger.error(f'Prediction failed: {e}')
    return {'error': 'Service temporarily unavailable'}, 503

# 3. Кэширование результатов
from functools import lru_cache

@lru_cache(maxsize=1000)
def cached_predict(features_tuple):
    return model.predict([list(features_tuple)])[0]

# 4. Мониторинг метрик
from prometheus_client import Counter, Histogram

predictions_total = Counter('predictions_total', 'Total predictions')
prediction_latency = Histogram('prediction_latency_seconds', 'Prediction latency')

Инструменты и платформы

• Docker: контейнеризация
• Kubernetes: оркестрация контейнеров в масштабе
• MLflow: версионирование моделей и экспериментов
• BentoML: специализированный фреймворк для ML сервисов
• FastAPI/Flask: REST API
• Prometheus + Grafana: мониторинг и визуализация
• AWS SageMaker / Google Cloud AI: облачные решения

Выводы

• Production deployment требует намного больше, чем просто обученная модель
• Обязательны контейнеризация, API, логирование, мониторинг и версионирование
• Используй canary развёртывание для безопасного обновления моделей
• Постоянно отслеживай производительность и дрейф данных