Déploiement PyTorch en production : De l'entraînement à l'API

1. Entraînement → Modèle .pth
2. Export → TorchScript (.pt) ou ONNX (.onnx)
3. API → FastAPI / TorchServe
4. Containerisation → Docker image
5. Déploiement → Cloud (AWS/GCP/Azure)
6. Monitoring → Logs, métriques, retraining

import torch
import torch.nn as nn

# Modèle simple
class SimpleModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc1 = nn.Linear(10, 50)
        self.fc2 = nn.Linear(50, 2)

    def forward(self, x):
        x = torch.relu(self.fc1(x))
        x = self.fc2(x)
        return x

# Charger votre modèle entraîné
model = SimpleModel()
model.load_state_dict(torch.load("model.pth"))
model.eval()

# TRACING : passer un exemple d'input
example_input = torch.randn(1, 10)
traced_model = torch.jit.trace(model, example_input)

# Sauvegarder
traced_model.save("model_traced.pt")

# Charger et utiliser (sans le code source !)
loaded = torch.jit.load("model_traced.pt")
output = loaded(torch.randn(1, 10))
print(output)

class DynamicModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.fc = nn.Linear(10, 2)

    def forward(self, x):
        # Branche conditionnelle : impossible à tracer !
        if x.sum() > 0:
            x = torch.relu(x)
        else:
            x = torch.tanh(x)
        return self.fc(x)

# SCRIPTING : compile le code Python
model = DynamicModel()
scripted_model = torch.jit.script(model)
scripted_model.save("model_scripted.pt")

# Utilisation
loaded = torch.jit.load("model_scripted.pt")
output = loaded(torch.randn(1, 10))

import torch.onnx

# Charger le modèle
model = SimpleModel()
model.load_state_dict(torch.load("model.pth"))
model.eval()

# Input d'exemple
dummy_input = torch.randn(1, 10)

# Export ONNX
torch.onnx.export(
    model,                      # Modèle PyTorch
    dummy_input,                # Input d'exemple
    "model.onnx",               # Fichier de sortie
    export_params=True,         # Inclure les poids
    opset_version=14,           # Version ONNX (14 ou 15 recommandé)
    do_constant_folding=True,   # Optimisation
    input_names=['input'],      # Noms des inputs
    output_names=['output'],    # Noms des outputs
    dynamic_axes={
        'input': {0: 'batch_size'},   # Batch dynamique
        'output': {0: 'batch_size'}
    }
)

print("Modèle exporté en ONNX : model.onnx")

import onnxruntime as ort
import numpy as np

# Charger avec ONNX Runtime (plus rapide que PyTorch !)
session = ort.InferenceSession("model.onnx", providers=['CPUExecutionProvider'])

# Inférence
input_data = np.random.randn(1, 10).astype(np.float32)
outputs = session.run(None, {'input': input_data})

print(f"Output: {outputs[0]}")

ml-api/
├── model.pt             # Modèle TorchScript
├── app.py               # Code FastAPI
├── requirements.txt
└── Dockerfile

fastapi==0.104.1
uvicorn[standard]==0.24.0
torch==2.1.0
pydantic==2.5.0

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import torch
import torch.nn as nn
from typing import List

# Définir le schéma d'input
class PredictionRequest(BaseModel):
    features: List[float]

class PredictionResponse(BaseModel):
    prediction: int
    probabilities: List[float]

# Initialiser l'app
app = FastAPI(title="PyTorch Model API", version="1.0.0")

# Charger le modèle au démarrage
model = None

@app.on_event("startup")
async def load_model():
    global model
    try:
        model = torch.jit.load("model.pt")
        model.eval()
        print("✅ Modèle chargé avec succès")
    except Exception as e:
        print(f"❌ Erreur de chargement : {e}")
        raise

@app.get("/")
async def root():
    return {"message": "PyTorch Model API", "status": "running"}

@app.get("/health")
async def health_check():
    """Endpoint de santé pour monitoring"""
    if model is None:
        raise HTTPException(status_code=503, detail="Modèle non chargé")
    return {"status": "healthy", "model_loaded": True}

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    """Endpoint de prédiction"""
    if model is None:
        raise HTTPException(status_code=503, detail="Modèle non disponible")

    try:
        # Valider l'input
        if len(request.features) != 10:
            raise HTTPException(
                status_code=400,
                detail=f"Attendu 10 features, reçu {len(request.features)}"
            )

        # Préparer le tensor
        input_tensor = torch.tensor([request.features], dtype=torch.float32)

        # Inférence
        with torch.no_grad():
            logits = model(input_tensor)
            probabilities = torch.softmax(logits, dim=1)
            prediction = torch.argmax(probabilities, dim=1).item()

        return PredictionResponse(
            prediction=prediction,
            probabilities=probabilities[0].tolist()
        )

    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Erreur de prédiction : {str(e)}")

# Lancer avec : uvicorn app:app --reload

# Lancer le serveur
uvicorn app:app --host 0.0.0.0 --port 8000

# Dans un autre terminal
curl -X POST "http://localhost:8000/predict" \
  -H "Content-Type: application/json" \
  -d '{"features": [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]}'

# Réponse
# {"prediction": 1, "probabilities": [0.23, 0.77]}

from asyncio import Queue, create_task, sleep
import asyncio

batch_queue = Queue(maxsize=100)
BATCH_SIZE = 32
BATCH_TIMEOUT = 0.05  # 50ms max d'attente

async def batch_inference_worker():
    """Worker qui accumule et traite par batch"""
    while True:
        batch_items = []

        # Attendre le premier item
        item = await batch_queue.get()
        batch_items.append(item)

        # Collecter jusqu'à BATCH_SIZE ou timeout
        deadline = asyncio.get_event_loop().time() + BATCH_TIMEOUT

        while len(batch_items) < BATCH_SIZE:
            timeout = deadline - asyncio.get_event_loop().time()
            if timeout <= 0:
                break

            try:
                item = await asyncio.wait_for(batch_queue.get(), timeout=timeout)
                batch_items.append(item)
            except asyncio.TimeoutError:
                break

        # Traiter le batch
        inputs = torch.stack([item["tensor"] for item in batch_items])
        with torch.no_grad():
            outputs = model(inputs)

        # Retourner les résultats
        for item, output in zip(batch_items, outputs):
            item["future"].set_result(output)

# Lancer le worker au startup
@app.on_event("startup")
async def start_batch_worker():
    create_task(batch_inference_worker())

@app.post("/predict_batched")
async def predict_batched(request: PredictionRequest):
    input_tensor = torch.tensor([request.features], dtype=torch.float32)

    # Créer une Future pour recevoir le résultat
    loop = asyncio.get_event_loop()
    future = loop.create_future()

    # Ajouter à la queue
    await batch_queue.put({"tensor": input_tensor.squeeze(0), "future": future})

    # Attendre le résultat
    output = await future

    probabilities = torch.softmax(output, dim=0)
    prediction = torch.argmax(probabilities).item()

    return PredictionResponse(
        prediction=prediction,
        probabilities=probabilities.tolist()
    )

from functools import lru_cache
import hashlib
import json

def hash_input(features: List[float]) -> str:
    """Créer un hash déterministe de l'input"""
    return hashlib.md5(json.dumps(features, sort_keys=True).encode()).hexdigest()

# Cache avec TTL (Time To Live)
from cachetools import TTLCache
prediction_cache = TTLCache(maxsize=1000, ttl=300)  # 5 minutes

@app.post("/predict_cached")
async def predict_cached(request: PredictionRequest):
    # Vérifier le cache
    cache_key = hash_input(request.features)

    if cache_key in prediction_cache:
        return prediction_cache[cache_key]

    # Faire la prédiction
    # ... (code de prédiction normal)

    result = PredictionResponse(prediction=prediction, probabilities=probs)

    # Stocker dans le cache
    prediction_cache[cache_key] = result

    return result

import time
from prometheus_client import Counter, Histogram, generate_latest
from fastapi import Response

# Métriques Prometheus
REQUEST_COUNT = Counter('api_requests_total', 'Total de requêtes', ['endpoint', 'status'])
REQUEST_DURATION = Histogram('api_request_duration_seconds', 'Durée des requêtes', ['endpoint'])

@app.middleware("http")
async def add_metrics(request, call_next):
    start_time = time.time()

    response = await call_next(request)

    duration = time.time() - start_time
    REQUEST_DURATION.labels(endpoint=request.url.path).observe(duration)
    REQUEST_COUNT.labels(endpoint=request.url.path, status=response.status_code).inc()

    return response

@app.get("/metrics")
async def metrics():
    """Endpoint pour Prometheus scraping"""
    return Response(content=generate_latest(), media_type="text/plain")

pip install torchserve torch-model-archiver torch-workflow-archiver

import torch
from ts.torch_handler.base_handler import BaseHandler
import json

class CustomHandler(BaseHandler):
    def __init__(self):
        super().__init__()
        self.initialized = False

    def initialize(self, context):
        """Charger le modèle"""
        self.manifest = context.manifest
        properties = context.system_properties
        model_dir = properties.get("model_dir")

        # Charger le modèle
        self.device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
        model_path = f"{model_dir}/model.pt"
        self.model = torch.jit.load(model_path, map_location=self.device)
        self.model.eval()

        self.initialized = True

    def preprocess(self, data):
        """Préparer les données"""
        # data est une liste de requêtes
        inputs = []
        for row in data:
            input_data = row.get("data") or row.get("body")
            if isinstance(input_data, (bytes, bytearray)):
                input_data = input_data.decode('utf-8')

            features = json.loads(input_data)["features"]
            inputs.append(features)

        return torch.tensor(inputs, dtype=torch.float32).to(self.device)

    def inference(self, inputs):
        """Faire l'inférence"""
        with torch.no_grad():
            outputs = self.model(inputs)
        return outputs

    def postprocess(self, outputs):
        """Formater les résultats"""
        probabilities = torch.softmax(outputs, dim=1)
        predictions = torch.argmax(probabilities, dim=1)

        results = []
        for pred, probs in zip(predictions, probabilities):
            results.append({
                "prediction": pred.item(),
                "probabilities": probs.tolist()
            })

        return results

# Créer l'archive .mar
torch-model-archiver \
  --model-name my_model \
  --version 1.0 \
  --model-file model.pt \
  --handler handler.py \
  --export-path model_store

# Lancer TorchServe
torchserve --start \
  --model-store model_store \
  --models my_model=my_model.mar \
  --ncs

# Tester
curl -X POST http://localhost:8080/predictions/my_model \
  -H "Content-Type: application/json" \
  -d '{"features": [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0]}'

# Dockerfile
FROM python:3.10-slim

# Variables d'environnement
ENV PYTHONUNBUFFERED=1 \
    PYTHONDONTWRITEBYTECODE=1

# Workdir
WORKDIR /app

# Installer les dépendances système si besoin
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# Copier requirements et installer
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copier le code et le modèle
COPY app.py .
COPY model.pt .

# Exposer le port
EXPOSE 8000

# Healthcheck
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8000/health || exit 1

# Commande de démarrage
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

# Build
docker build -t pytorch-api:latest .

# Run
docker run -d \
  --name pytorch-api \
  -p 8000:8000 \
  pytorch-api:latest

# Test
curl http://localhost:8000/health

version: '3.8'

services:
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - MODEL_PATH=/models/model.pt
    volumes:
      - ./models:/models:ro
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 4G
    restart: unless-stopped

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
    depends_on:
      - api
    restart: unless-stopped

  prometheus:
    image: prom/prometheus:latest
    ports:
      - "9090:9090"
    volumes:
      - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro
    restart: unless-stopped

upstream api_backend {
    least_conn;
    server api:8000 max_fails=3 fail_timeout=30s;
}

server {
    listen 80;

    location / {
        proxy_pass http://api_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_connect_timeout 300s;
        proxy_read_timeout 300s;
    }

    location /health {
        access_log off;
        proxy_pass http://api_backend/health;
    }
}

import sagemaker
from sagemaker.pytorch import PyTorchModel

# Créer un modèle SageMaker
pytorch_model = PyTorchModel(
    model_data='s3://my-bucket/model.tar.gz',  # .tar.gz avec model.pt + code
    role='arn:aws:iam::123456789:role/SageMakerRole',
    framework_version='2.1.0',
    py_version='py310',
    entry_point='inference.py'  # Script avec model_fn, input_fn, predict_fn
)

# Déployer
predictor = pytorch_model.deploy(
    instance_type='ml.m5.xlarge',
    initial_instance_count=1,
    endpoint_name='pytorch-endpoint'
)

# Prédire
result = predictor.predict([0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0])

# Upload du modèle
gsutil cp model_saved_model gs://my-bucket/models/my_model/

# Créer une version
gcloud ai-platform versions create v1 \
  --model my_model \
  --origin gs://my-bucket/models/my_model/ \
  --runtime-version 2.11 \
  --framework PYTORCH \
  --python-version 3.10

# Prédire
gcloud ai-platform predict \
  --model my_model \
  --version v1 \
  --json-instances input.json

import gradio as gr
import torch

model = torch.jit.load("model.pt")
model.eval()

def predict(features):
    input_tensor = torch.tensor([features], dtype=torch.float32)
    with torch.no_grad():
        logits = model(input_tensor)
        probs = torch.softmax(logits, dim=1)
        pred = torch.argmax(probs, dim=1).item()

    return {
        "Prediction": pred,
        "Probabilities": probs[0].tolist()
    }

demo = gr.Interface(
    fn=predict,
    inputs=gr.Textbox(label="Features (10 floats séparés par virgules)"),
    outputs=gr.JSON(label="Résultat"),
    title="PyTorch Model Demo"
)

demo.launch()

import torch.quantization

# Modèle FP32
model_fp32 = SimpleModel()
model_fp32.load_state_dict(torch.load("model.pth"))
model_fp32.eval()

# Quantization dynamique (facile)
model_int8 = torch.quantization.quantize_dynamic(
    model_fp32,
    {nn.Linear},  # Quelles couches quantifier
    dtype=torch.qint8
)

# Sauvegarder
torch.jit.save(torch.jit.script(model_int8), "model_quantized.pt")

# Tester la taille
import os
print(f"FP32 size: {os.path.getsize('model.pth') / 1e6:.2f} MB")
print(f"INT8 size: {os.path.getsize('model_quantized.pt') / 1e6:.2f} MB")

import torch.nn.utils.prune as prune

# Pruning global (30% des poids mis à zéro)
parameters_to_prune = [
    (model.fc1, 'weight'),
    (model.fc2, 'weight'),
]

prune.global_unstructured(
    parameters_to_prune,
    pruning_method=prune.L1Unstructured,
    amount=0.3,  # 30%
)

# Rendre permanent
for module, param in parameters_to_prune:
    prune.remove(module, param)

def distillation_loss(student_logits, teacher_logits, true_labels, temperature=3.0, alpha=0.5):
    """
    Loss = α * KL(teacher, student) + (1-α) * CE(student, labels)
    """
    # Soft targets (teacher)
    soft_targets = nn.functional.softmax(teacher_logits / temperature, dim=1)
    soft_student = nn.functional.log_softmax(student_logits / temperature, dim=1)
    distill_loss = nn.functional.kl_div(soft_student, soft_targets, reduction='batchmean')
    distill_loss *= temperature ** 2

    # Hard targets (vérité)
    student_loss = nn.functional.cross_entropy(student_logits, true_labels)

    return alpha * distill_loss + (1 - alpha) * student_loss

import logging
import json
from datetime import datetime

class JSONFormatter(logging.Formatter):
    def format(self, record):
        log_obj = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "module": record.module,
        }
        return json.dumps(log_obj)

logger = logging.getLogger(__name__)
handler = logging.StreamHandler()
handler.setFormatter(JSONFormatter())
logger.addHandler(handler)
logger.setLevel(logging.INFO)

# Usage
logger.info("Prédiction effectuée", extra={"prediction": 1, "latency_ms": 45})

import mlflow
import mlflow.pytorch

# Enregistrer un modèle
with mlflow.start_run():
    mlflow.log_param("learning_rate", 0.001)
    mlflow.log_metric("accuracy", 0.95)

    # Logger le modèle
    mlflow.pytorch.log_model(model, "model")

# Charger depuis le registry
model_uri = "models:/my_model/production"
loaded_model = mlflow.pytorch.load_model(model_uri)