Chapitre 3 : Versionnage des modèles

Introduction

Le versionnage des modèles est crucial pour les systèmes ML en production. Il permet de suivre, comparer et revenir en arrière sur les modèles tout en maintenant une lignée claire des données au déploiement.

Pourquoi le versionnage des modèles ?

Le défi de la gestion des modèles

Avantages du versionnage des modèles

Avantage	Description
Reproductibilité	Recréer n'importe quelle version de modèle
Lignée	Suivre données → modèle → prédictions
Comparaison	Comparer les performances entre versions
Retour en arrière	Revenir rapidement aux versions précédentes
Conformité	Traçabilité pour les industries réglementées

Concepts de registre des modèles

Composants clés

Étapes du cycle de vie du modèle

Étape	Description
Aucune	État initial, nouvellement enregistré
Staging	Le modèle est en cours de test/validation
Production	Le modèle sert du trafic en direct
Archivé	Le modèle est obsolète, conservé pour référence

Registre des modèles MLflow

Configuration de MLflow

# Installer MLflow
pip install mlflow

# Démarrer le serveur MLflow avec le magasin de backend
mlflow server \
    --backend-store-uri postgresql://user:pass@localhost/mlflow \
    --default-artifact-root s3://mlflow-artifacts/ \
    --host 0.0.0.0 \
    --port 5000

Enregistrement des modèles

# register_model.py
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier

# Définir l'URI de suivi
mlflow.set_tracking_uri("http://localhost:5000")

# Entraîner et enregistrer le modèle
with mlflow.start_run() as run:
    # Entraîner le modèle
    model = RandomForestClassifier(n_estimators=100)
    model.fit(X_train, y_train)
    
    # Enregistrer les paramètres
    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("max_depth", None)
    
    # Enregistrer les métriques
    accuracy = model.score(X_test, y_test)
    mlflow.log_metric("accuracy", accuracy)
    
    # Enregistrer le modèle
    mlflow.sklearn.log_model(
        model,
        "model",
        registered_model_name="customer-churn-model"
    )
    
    print(f"Modèle enregistré avec run_id : {run.info.run_id}")

Gestion des versions de modèles

# model_management.py
from mlflow.tracking import MlflowClient

client = MlflowClient()

# Obtenir toutes les versions d'un modèle
versions = client.search_model_versions("name='customer-churn-model'")
for v in versions:
    print(f"Version {v.version} : {v.current_stage}")

# Transitionner le modèle vers le staging
client.transition_model_version_stage(
    name="customer-churn-model",
    version=2,
    stage="Staging"
)

# Transitionner le modèle vers la production
client.transition_model_version_stage(
    name="customer-churn-model",
    version=2,
    stage="Production",
    archive_existing_versions=True  # Archiver la version actuelle en prod
)

# Ajouter une description à la version du modèle
client.update_model_version(
    name="customer-churn-model",
    version=2,
    description="Amélioration de l'ingénierie des caractéristiques, AUC augmenté de 5%"
)

# Ajouter des étiquettes
client.set_model_version_tag(
    name="customer-churn-model",
    version=2,
    key="validation_status",
    value="approved"
)

Chargement des modèles depuis le registre

# load_model.py
import mlflow

# Charger une version spécifique
model_v1 = mlflow.sklearn.load_model(
    "models:/customer-churn-model/1"
)

# Charger par étape
production_model = mlflow.sklearn.load_model(
    "models:/customer-churn-model/Production"
)

# Charger le modèle en staging
staging_model = mlflow.sklearn.load_model(
    "models:/customer-churn-model/Staging"
)

# Faire des prédictions
predictions = production_model.predict(X_new)

Métadonnées et lignée du modèle

Carte de modèle complète

# model_card.py
import json
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional
from datetime import datetime

@dataclass
class ModelCard:
    """Documentation structurée du modèle suivant les normes de l'industrie."""
    
    # Informations de base
    model_name: str
    version: str
    description: str
    owner: str
    created_at: str
    
    # Informations d'entraînement
    training_data: Dict[str, str]
    features: List[str]
    target: str
    algorithm: str
    hyperparameters: Dict[str, any]
    
    # Performance
    metrics: Dict[str, float]
    evaluation_data: str
    
    # Utilisation prévue
    intended_use: str
    out_of_scope_use: str
    
    # Limitations et biais
    limitations: List[str]
    ethical_considerations: List[str]
    
    # Lignée
    parent_model: Optional[str] = None
    training_run_id: Optional[str] = None
    data_version: Optional[str] = None
    code_version: Optional[str] = None

# Créer une carte de modèle
model_card = ModelCard(
    model_name="customer-churn-model",
    version="2.1.0",
    description="Prédit la probabilité de désabonnement des clients dans les 30 jours",
    owner="ml-team@company.com",
    created_at=datetime.now().isoformat(),
    
    training_data={
        "source": "s3://data-lake/customer-features/",
        "date_range": "2023-01-01 to 2024-01-01",
        "records": "1,500,000"
    },
    features=[
        "account_age_days",
        "total_purchases",
        "days_since_last_login",
        "support_tickets_90d",
        "nps_score"
    ],
    target="churned_within_30_days",
    algorithm="GradientBoostingClassifier",
    hyperparameters={
        "n_estimators": 200,
        "max_depth": 7,
        "learning_rate": 0.1,
        "min_samples_split": 10
    },
    
    metrics={
        "auc_roc": 0.89,
        "precision": 0.82,
        "recall": 0.75,
        "f1_score": 0.78
    },
    evaluation_data="s3://data-lake/test-set-2024-01/",
    
    intended_use="Identifier les clients à risque pour les campagnes de rétention",
    out_of_scope_use="Évaluation de crédit, décisions d'embauche",
    
    limitations=[
        "La performance peut se dégrader pour les clients de moins de 30 jours",
        "Non validé pour les clients de niveau entreprise",
        "Nécessite un minimum de 6 mois d'historique de transactions"
    ],
    ethical_considerations=[
        "Le modèle ne doit pas être utilisé pour refuser un service",
        "Les prédictions doivent être examinées par un humain avant action"
    ],
    
    parent_model="customer-churn-model/1.0.0",
    training_run_id="abc123def456",
    data_version="data-v2.1.0",
    code_version="git-sha-789xyz"
)

# Enregistrer la carte de modèle
with open("model_card.json", "w") as f:
    json.dump(asdict(model_card), f, indent=2)

Enregistrement de la lignée du modèle

# lineage_tracking.py
import mlflow
import git
import hashlib

def get_code_version():
    """Obtenir le hash du commit git actuel."""
    repo = git.Repo(search_parent_directories=True)
    return repo.head.object.hexsha

def get_data_hash(data_path: str) -> str:
    """Calculer le hash du fichier de données."""
    with open(data_path, 'rb') as f:
        return hashlib.md5(f.read()).hexdigest()

def train_with_lineage(X_train, y_train, X_test, y_test, data_path: str):
    """Entraîner le modèle avec un suivi complet de la lignée."""
    
    with mlflow.start_run() as run:
        # Suivre la version du code
        code_version = get_code_version()
        mlflow.set_tag("code_version", code_version)
        mlflow.set_tag("git_repo", "https://github.com/company/ml-project")
        
        # Suivre la version des données
        data_hash = get_data_hash(data_path)
        mlflow.set_tag("data_hash", data_hash)
        mlflow.set_tag("data_path", data_path)
        
        # Suivre l'environnement
        mlflow.set_tag("python_version", "3.10.0")
        mlflow.set_tag("sklearn_version", "1.3.0")
        
        # Suivre la configuration d'entraînement
        mlflow.log_param("training_samples", len(X_train))
        mlflow.log_param("test_samples", len(X_test))
        mlflow.log_param("features", list(X_train.columns))
        
        # Entraîner le modèle
        model = GradientBoostingClassifier(
            n_estimators=200,
            max_depth=7,
            learning_rate=0.1
        )
        model.fit(X_train, y_train)
        
        # Enregistrer les métriques
        train_accuracy = model.score(X_train, y_train)
        test_accuracy = model.score(X_test, y_test)
        mlflow.log_metric("train_accuracy", train_accuracy)
        mlflow.log_metric("test_accuracy", test_accuracy)
        
        # Enregistrer le modèle avec signature
        from mlflow.models.signature import infer_signature
        signature = infer_signature(X_train, model.predict(X_train))
        
        mlflow.sklearn.log_model(
            model,
            "model",
            signature=signature,
            registered_model_name="customer-churn-model"
        )
        
        return run.info.run_id

Stratégies de versionnage des modèles

Versionnage sémantique pour les modèles

MAJOR.MINOR.PATCH

Exemples :
- 1.0.0 → 1.0.1  # Patch : Correction de bug, pas de réentraînement
- 1.0.0 → 1.1.0  # Mineur : Réentraîné, même architecture
- 1.0.0 → 2.0.0  # Majeur : Nouvelle architecture ou changements majeurs

Incrémentation automatique des versions

# version_manager.py
import re
from enum import Enum
from mlflow.tracking import MlflowClient

class VersionBump(Enum):
    MAJOR = "major"
    MINOR = "minor"
    PATCH = "patch"

def get_next_version(current_version: str, bump: VersionBump) -> str:
    """Calculer la prochaine version sémantique."""
    match = re.match(r"(\d+)\.(\d+)\.(\d+)", current_version)
    if not match:
        return "1.0.0"
    
    major, minor, patch = map(int, match.groups())
    
    if bump == VersionBump.MAJOR:
        return f"{major + 1}.0.0"
    elif bump == VersionBump.MINOR:
        return f"{major}.{minor + 1}.0"
    else:
        return f"{major}.{minor}.{patch + 1}"

def register_new_version(
    model_name: str,
    run_id: str,
    bump: VersionBump,
    description: str
):
    """Enregistrer le modèle avec un versionnage approprié."""
    client = MlflowClient()
    
    # Obtenir la dernière version
    versions = client.search_model_versions(f"name='{model_name}'")
    if versions:
        latest_version = max(v.version for v in versions)
        latest_tags = client.get_model_version(model_name, latest_version).tags
        current_semantic = latest_tags.get("semantic_version", "0.0.0")
    else:
        current_semantic = "0.0.0"
    
    # Calculer la nouvelle version sémantique
    new_semantic = get_next_version(current_semantic, bump)
    
    # Enregistrer le modèle
    result = client.create_model_version(
        name=model_name,
        source=f"runs:/{run_id}/model",
        run_id=run_id,
        description=description
    )
    
    # Définir l'étiquette de version sémantique
    client.set_model_version_tag(
        name=model_name,
        version=result.version,
        key="semantic_version",
        value=new_semantic
    )
    
    print(f"Enregistré {model_name} v{result.version} (sémantique : {new_semantic})")
    return result

Workflow d'approbation des modèles

Pipeline de validation automatisée

# model_validation.py
from typing import Dict, List, Tuple
import mlflow
from mlflow.tracking import MlflowClient

class ModelValidator:
    """Valider les modèles avant le déploiement en production."""
    
    def __init__(self, model_name: str, version: int):
        self.client = MlflowClient()
        self.model_name = model_name
        self.version = version
        self.model_uri = f"models:/{model_name}/{version}"
        
    def validate_performance(
        self, 
        X_test, 
        y_test,
        thresholds: Dict[str, float]
    ) -> Tuple[bool, Dict[str, float]]:
        """Vérifier si le modèle respecte les seuils de performance."""
        model = mlflow.sklearn.load_model(self.model_uri)
        
        from sklearn.metrics import accuracy_score, roc_auc_score, f1_score
        
        predictions = model.predict(X_test)
        probabilities = model.predict_proba(X_test)[:, 1]
        
        metrics = {
            "accuracy": accuracy_score(y_test, predictions),
            "auc_roc": roc_auc_score(y_test, probabilities),
            "f1_score": f1_score(y_test, predictions)
        }
        
        # Vérifier par rapport aux seuils
        passed = all(
            metrics.get(metric, 0) >= threshold
            for metric, threshold in thresholds.items()
        )
        
        return passed, metrics
    
    def validate_latency(
        self, 
        X_sample,
        max_latency_ms: float = 100
    ) -> Tuple[bool, float]:
        """Vérifier si le modèle respecte les exigences de latence."""
        import time
        
        model = mlflow.sklearn.load_model(self.model_uri)
        
        # Pré-chauffage
        _ = model.predict(X_sample[:1])
        
        # Mesurer la latence
        times = []
        for _ in range(100):
            start = time.time()
            _ = model.predict(X_sample[:1])
            times.append((time.time() - start) * 1000)
        
        p99_latency = sorted(times)[98]
        passed = p99_latency <= max_latency_ms
        
        return passed, p99_latency
    
    def validate_consistency(
        self,
        X_sample,
        production_model_uri: str
    ) -> Tuple[bool, float]:
        """Vérifier que les prédictions correspondent au modèle de production."""
        new_model = mlflow.sklearn.load_model(self.model_uri)
        prod_model = mlflow.sklearn.load_model(production_model_uri)
        
        new_predictions = new_model.predict(X_sample)
        prod_predictions = prod_model.predict(X_sample)
        
        agreement = (new_predictions == prod_predictions).mean()
        passed = agreement >= 0.95  # Au moins 95% d'accord
        
        return passed, agreement
    
    def run_all_validations(
        self,
        X_test,
        y_test,
        thresholds: Dict[str, float]
    ) -> Dict[str, any]:
        """Exécuter toutes les validations et retourner les résultats."""
        results = {
            "model_name": self.model_name,
            "version": self.version,
            "validations": {}
        }
        
        # Validation de la performance
        perf_passed, perf_metrics = self.validate_performance(
            X_test, y_test, thresholds
        )
        results["validations"]["performance"] = {
            "passed": perf_passed,
            "metrics": perf_metrics
        }
        
        # Validation de la latence
        latency_passed, p99_latency = self.validate_latency(X_test[:100])
        results["validations"]["latency"] = {
            "passed": latency_passed,
            "p99_ms": p99_latency
        }
        
        # Résultat global
        results["approved"] = all(
            v["passed"] for v in results["validations"].values()
        )
        
        # Mettre à jour les étiquettes du modèle
        status = "approved" if results["approved"] else "rejected"
        self.client.set_model_version_tag(
            name=self.model_name,
            version=self.version,
            key="validation_status",
            value=status
        )
        
        return results

Points clés à retenir

Résumé

1. Utilisez un registre de modèles pour une gestion centralisée des modèles
2. Suivez la lignée complète - données, code, paramètres, environnement
3. Implémentez le versionnage sémantique pour une progression claire des versions
4. Créez des cartes de modèles pour la documentation et la conformité
5. Automatisez la validation avant le déploiement en production
6. Maintenez des pistes d'audit pour les industries réglementées

Exercice pratique

1. Configurez le serveur de suivi MLflow
2. Entraînez et enregistrez plusieurs versions de modèles
3. Créez une carte de modèle pour votre meilleur modèle
4. Implémentez un pipeline de validation de modèle
5. Pratiquez la transition des modèles entre les étapes

---

Suivant : Module 2 Quiz