Chapitre 3 : Le cycle de vie du ML

Introduction

Le cycle de vie du ML englobe toutes les étapes, de la définition du problème à la mise hors service du modèle. Comprendre ce cycle de vie est fondamental pour mettre en œuvre des pratiques MLOps efficaces.

Le cycle de vie complet du ML

Étape 1 : Définition du problème

Avant de commencer tout travail de ML, définissez clairement le problème.

Questions clés à répondre

Question	Exemple de réponse
Quel problème commercial résolvons-nous ?	Réduire le taux de désabonnement des clients de 15 %
Quel type de problème de ML est-ce ?	Classification binaire
Quelle est la métrique de succès ?	AUC-ROC > 0.85
Quelles sont les exigences de latence ?	< 100ms pour les prédictions
Quelles données sont disponibles ?	2 ans de données clients

Document de définition du problème

# ml_problem_definition.yaml
project_name: Customer Churn Prediction
version: 1.0

business_context:
  objective: Prédire les clients susceptibles de se désabonner dans les 30 jours
  impact: $2M de revenus annuels en jeu
  stakeholders:
    - Équipe de succès client
    - Équipe marketing
    - Équipe produit

ml_formulation:
  task_type: binary_classification
  target_variable: churned_within_30_days
  positive_class_definition: "Client a annulé l'abonnement"
  
success_criteria:
  primary_metric: auc_roc
  threshold: 0.85
  secondary_metrics:
    - precision_at_10_percent_recall: 0.70
    - f1_score: 0.75

constraints:
  latency_p99_ms: 100
  model_size_mb: 500
  interpretability: high  # Besoin d'expliquer les prédictions
  
timeline:
  mvp: 2024-03-01
  production: 2024-04-01

Étape 2 : Collecte de données

Rassembler et comprendre les données disponibles pour le projet de ML.

Inventaire des données

# data_inventory.py
from dataclasses import dataclass
from typing import List, Optional
from datetime import datetime

@dataclass
class DataSource:
    name: str
    location: str
    format: str
    owner: str
    update_frequency: str
    size_gb: float
    date_range_start: datetime
    date_range_end: datetime
    pii_columns: List[str]
    quality_score: float  # 0-1

# Définir les sources de données disponibles
data_sources = [
    DataSource(
        name="customer_transactions",
        location="s3://data-lake/transactions/",
        format="parquet",
        owner="data-engineering",
        update_frequency="daily",
        size_gb=150.0,
        date_range_start=datetime(2022, 1, 1),
        date_range_end=datetime(2024, 1, 1),
        pii_columns=["email", "phone"],
        quality_score=0.95
    ),
    DataSource(
        name="customer_support_tickets",
        location="s3://data-lake/support/",
        format="json",
        owner="customer-success",
        update_frequency="hourly",
        size_gb=25.0,
        date_range_start=datetime(2022, 6, 1),
        date_range_end=datetime(2024, 1, 1),
        pii_columns=["customer_email"],
        quality_score=0.88
    ),
]

Pipeline de collecte de données

# data_collection_pipeline.py
from datetime import datetime, timedelta
import pandas as pd
from prefect import flow, task

@task
def extract_transactions(start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """Extraire les données de transaction du data warehouse."""
    query = f"""
    SELECT 
        customer_id,
        transaction_date,
        amount,
        product_category
    FROM transactions
    WHERE transaction_date BETWEEN '{start_date}' AND '{end_date}'
    """
    return pd.read_sql(query, connection)

@task
def extract_support_tickets(start_date: datetime, end_date: datetime) -> pd.DataFrame:
    """Extraire les données des tickets de support."""
    query = f"""
    SELECT 
        customer_id,
        ticket_date,
        category,
        resolution_time_hours,
        satisfaction_score
    FROM support_tickets
    WHERE ticket_date BETWEEN '{start_date}' AND '{end_date}'
    """
    return pd.read_sql(query, connection)

@task
def merge_data(transactions: pd.DataFrame, tickets: pd.DataFrame) -> pd.DataFrame:
    """Fusionner toutes les sources de données."""
    return transactions.merge(
        tickets, 
        on='customer_id', 
        how='left'
    )

@flow(name="data-collection")
def collect_training_data(lookback_days: int = 365):
    end_date = datetime.now()
    start_date = end_date - timedelta(days=lookback_days)
    
    transactions = extract_transactions(start_date, end_date)
    tickets = extract_support_tickets(start_date, end_date)
    
    merged = merge_data(transactions, tickets)
    
    return merged

Étape 3 : Préparation des données

Nettoyer et préparer les données pour l'ingénierie des caractéristiques et l'entraînement.

Vérifications de la qualité des données

# data_quality.py
import great_expectations as gx

def validate_training_data(df):
    """Valider la qualité des données d'entraînement."""
    context = gx.get_context()
    
    # Créer des attentes
    expectations = [
        # Pas de valeurs nulles dans les colonnes critiques
        {"expectation_type": "expect_column_values_to_not_be_null",
         "kwargs": {"column": "customer_id"}},
        
        # Valeurs dans la plage attendue
        {"expectation_type": "expect_column_values_to_be_between",
         "kwargs": {"column": "age", "min_value": 18, "max_value": 120}},
        
        # Identifiants uniques
        {"expectation_type": "expect_column_values_to_be_unique",
         "kwargs": {"column": "customer_id"}},
        
        # Pas de dates futures
        {"expectation_type": "expect_column_values_to_be_between",
         "kwargs": {"column": "signup_date", 
                    "max_value": datetime.now().isoformat()}},
        
        # Colonne existante
        {"expectation_type": "expect_column_to_exist",
         "kwargs": {"column": "churned"}},
    ]
    
    # Exécuter la validation
    results = context.run_validation_operator(
        "action_list_operator",
        assets_to_validate=[df],
        run_id=f"training_data_{datetime.now().isoformat()}"
    )
    
    return results.success

Pipeline de nettoyage des données

# data_cleaning.py
import pandas as pd
import numpy as np

def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """Nettoyer et préparer les données brutes."""
    
    # Faire une copie
    df = df.copy()
    
    # Gérer les valeurs manquantes
    df['age'] = df['age'].fillna(df['age'].median())
    df['income'] = df['income'].fillna(df.groupby('segment')['income'].transform('median'))
    
    # Supprimer les doublons
    df = df.drop_duplicates(subset=['customer_id'], keep='last')
    
    # Gérer les valeurs aberrantes (clip aux percentiles)
    for col in ['transaction_amount', 'login_frequency']:
        lower = df[col].quantile(0.01)
        upper = df[col].quantile(0.99)
        df[col] = df[col].clip(lower, upper)
    
    # Standardiser les colonnes de texte
    df['segment'] = df['segment'].str.lower().str.strip()
    
    # Convertir les types de données
    df['signup_date'] = pd.to_datetime(df['signup_date'])
    df['churned'] = df['churned'].astype(int)
    
    return df

Étape 4 : Ingénierie des caractéristiques

Créer des caractéristiques qui aident le modèle à apprendre des motifs.

Pipeline d'ingénierie des caractéristiques

# feature_engineering.py
import pandas as pd
import numpy as np
from sklearn.preprocessing import StandardScaler, LabelEncoder
from datetime import datetime

class FeatureEngineer:
    def __init__(self):
        self.scaler = StandardScaler()
        self.encoders = {}
        
    def create_features(self, df: pd.DataFrame) -> pd.DataFrame:
        """Créer toutes les caractéristiques pour le modèle."""
        features = pd.DataFrame()
        features['customer_id'] = df['customer_id']
        
        # Caractéristiques temporelles
        features['account_age_days'] = (datetime.now() - df['signup_date']).dt.days
        features['days_since_last_login'] = (datetime.now() - df['last_login']).dt.days
        features['days_since_last_purchase'] = (datetime.now() - df['last_purchase']).dt.days
        
        # Caractéristiques d'agrégation
        features['total_purchases'] = df['purchase_count']
        features['total_spend'] = df['total_spend']
        features['avg_order_value'] = df['total_spend'] / df['purchase_count'].clip(1)
        
        # Caractéristiques d'engagement
        features['login_frequency_30d'] = df['logins_last_30d']
        features['support_tickets_90d'] = df['support_tickets_90d']
        features['nps_score'] = df['nps_score']
        
        # Caractéristiques de ratio
        features['spend_to_income_ratio'] = df['total_spend'] / df['income'].clip(1)
        features['engagement_trend'] = (
            df['logins_last_30d'] / df['logins_prev_30d'].clip(1)
        )
        
        # Encodage catégoriel
        for col in ['segment', 'plan_type', 'channel']:
            if col not in self.encoders:
                self.encoders[col] = LabelEncoder()
                features[f'{col}_encoded'] = self.encoders[col].fit_transform(df[col])
            else:
                features[f'{col}_encoded'] = self.encoders[col].transform(df[col])
        
        return features
    
    def scale_features(self, features: pd.DataFrame, fit: bool = False) -> pd.DataFrame:
        """Mettre à l'échelle les caractéristiques numériques."""
        numerical_cols = features.select_dtypes(include=[np.number]).columns
        numerical_cols = [c for c in numerical_cols if c != 'customer_id']
        
        if fit:
            features[numerical_cols] = self.scaler.fit_transform(features[numerical_cols])
        else:
            features[numerical_cols] = self.scaler.transform(features[numerical_cols])
        
        return features

Étape 5 : Entraînement du modèle

Entraîner et ajuster les modèles avec un suivi approprié des expériences.

Pipeline d'entraînement avec MLflow

# training_pipeline.py
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.model_selection import cross_val_score, GridSearchCV
from sklearn.metrics import roc_auc_score, precision_recall_curve
import pandas as pd
import numpy as np

class ModelTrainer:
    def __init__(self, experiment_name: str):
        mlflow.set_experiment(experiment_name)
        self.best_model = None
        self.best_score = 0
        
    def train_and_evaluate(
        self, 
        X_train: pd.DataFrame, 
        y_train: pd.Series,
        X_val: pd.DataFrame,
        y_val: pd.Series
    ):
        """Entraîner plusieurs modèles et sélectionner le meilleur."""
        
        models = {
            'random_forest': {
                'model': RandomForestClassifier(random_state=42),
                'params': {
                    'n_estimators': [100, 200],
                    'max_depth': [10, 20, None],
                    'min_samples_split': [2, 5]
                }
            },
            'gradient_boosting': {
                'model': GradientBoostingClassifier(random_state=42),
                'params': {
                    'n_estimators': [100, 200],
                    'max_depth': [3, 5, 7],
                    'learning_rate': [0.01, 0.1]
                }
            }
        }
        
        for model_name, config in models.items():
            with mlflow.start_run(run_name=model_name):
                # Enregistrer les paramètres
                mlflow.log_param("model_type", model_name)
                
                # Recherche de grille pour les meilleurs hyperparamètres
                grid_search = GridSearchCV(
                    config['model'],
                    config['params'],
                    cv=5,
                    scoring='roc_auc',
                    n_jobs=-1
                )
                grid_search.fit(X_train, y_train)
                
                # Obtenir le meilleur modèle
                best_estimator = grid_search.best_estimator_
                
                # Enregistrer les meilleurs paramètres
                for param, value in grid_search.best_params_.items():
                    mlflow.log_param(param, value)
                
                # Évaluer sur le jeu de validation
                val_predictions = best_estimator.predict_proba(X_val)[:, 1]
                val_auc = roc_auc_score(y_val, val_predictions)
                
                # Enregistrer les métriques
                mlflow.log_metric("val_auc_roc", val_auc)
                mlflow.log_metric("cv_score", grid_search.best_score_)
                
                # Enregistrer le modèle
                mlflow.sklearn.log_model(best_estimator, "model")
                
                # Suivre le meilleur modèle
                if val_auc > self.best_score:
                    self.best_score = val_auc
                    self.best_model = best_estimator
                    
                print(f"{model_name}: AUC-ROC = {val_auc:.4f}")
        
        return self.best_model

Étape 6 : Évaluation du modèle

Évaluation complète avant le déploiement.

Cadre d'évaluation

# model_evaluation.py
import matplotlib.pyplot as plt
from sklearn.metrics import (
    roc_auc_score, precision_recall_curve, 
    confusion_matrix, classification_report
)
import numpy as np

class ModelEvaluator:
    def __init__(self, model, X_test, y_test):
        self.model = model
        self.X_test = X_test
        self.y_test = y_test
        self.predictions = model.predict(X_test)
        self.probabilities = model.predict_proba(X_test)[:, 1]
        
    def generate_report(self) -> dict:
        """Générer un rapport d'évaluation complet."""
        
        report = {
            'auc_roc': roc_auc_score(self.y_test, self.probabilities),
            'classification_report': classification_report(
                self.y_test, 
                self.predictions, 
                output_dict=True
            ),
            'confusion_matrix': confusion_matrix(
                self.y_test, 
                self.predictions
            ).tolist()
        }
        
        # Calculer la précision à différents seuils de rappel
        precision, recall, thresholds = precision_recall_curve(
            self.y_test, 
            self.probabilities
        )
        
        report['precision_at_recall_80'] = precision[
            np.argmin(np.abs(recall - 0.8))
        ]
        
        return report
    
    def check_deployment_criteria(self, criteria: dict) -> bool:
        """Vérifier si le modèle répond aux critères de déploiement."""
        report = self.generate_report()
        
        checks = []
        for metric, threshold in criteria.items():
            if metric == 'auc_roc':
                passed = report['auc_roc'] >= threshold
            elif metric == 'precision':
                passed = report['classification_report']['1']['precision'] >= threshold
            elif metric == 'recall':
                passed = report['classification_report']['1']['recall'] >= threshold
            else:
                passed = True
                
            checks.append(passed)
            print(f"{metric}: {'PASS' if passed else 'FAIL'}")
        
        return all(checks)

Étape 7 : Déploiement du modèle

Déployer le modèle en production.

Configuration de déploiement

# deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: churn-model
  labels:
    app: churn-prediction
spec:
  replicas: 3
  selector:
    matchLabels:
      app: churn-prediction
  template:
    metadata:
      labels:
        app: churn-prediction
    spec:
      containers:
      - name: model-server
        image: mlplatform/churn-model:v2.1.0
        ports:
        - containerPort: 8080
        resources:
          requests:
            memory: "2Gi"
            cpu: "1"
          limits:
            memory: "4Gi"
            cpu: "2"
        env:
        - name: MODEL_NAME
          value: "churn-model"
        - name: MODEL_VERSION
          value: "2.1.0"
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8080
          initialDelaySeconds: 5
          periodSeconds: 5

Étape 8 : Surveillance

Surveillance continue de la performance du modèle.

Métriques de surveillance

# monitoring.py
from prometheus_client import Counter, Histogram, Gauge
from datetime import datetime
import numpy as np

# Définir les métriques
prediction_counter = Counter(
    'model_predictions_total',
    'Total des prédictions effectuées',
    ['model_version', 'prediction_class']
)

prediction_latency = Histogram(
    'model_prediction_latency_seconds',
    'Latence des prédictions',
    buckets=[0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5]
)

data_drift_gauge = Gauge(
    'model_data_drift_score',
    'Score de dérive des données par caractéristique',
    ['feature_name']
)

model_accuracy_gauge = Gauge(
    'model_accuracy_rolling',
    'Précision roulante sur les dernières 24 heures'
)

class ModelMonitor:
    def __init__(self, model_version: str):
        self.model_version = model_version
        self.predictions_log = []
        
    def log_prediction(self, input_data: dict, prediction: int, latency: float):
        """Enregistrer une prédiction pour la surveillance."""
        prediction_counter.labels(
            model_version=self.model_version,
            prediction_class=str(prediction)
        ).inc()
        
        prediction_latency.observe(latency)
        
        self.predictions_log.append({
            'timestamp': datetime.now(),
            'input': input_data,
            'prediction': prediction,
            'latency': latency
        })
    
    def check_drift(self, current_data: np.ndarray, reference_data: np.ndarray):
        """Vérifier la dérive des données à l'aide de tests statistiques."""
        from scipy import stats
        
        drift_scores = {}
        for i, feature in enumerate(feature_names):
            # Test de Kolmogorov-Smirnov
            statistic, p_value = stats.ks_2samp(
                reference_data[:, i],
                current_data[:, i]
            )
            drift_scores[feature] = statistic
            data_drift_gauge.labels(feature_name=feature).set(statistic)
        
        return drift_scores

Étape 9 : Réentraînement

Déclencher le réentraînement du modèle lorsque nécessaire.

Déclencheurs de réentraînement

# retraining_triggers.py
from dataclasses import dataclass
from enum import Enum
from typing import Optional
from datetime import datetime, timedelta

class TriggerType(Enum):
    SCHEDULED = "scheduled"
    PERFORMANCE = "performance"
    DRIFT = "drift"
    DATA_VOLUME = "data_volume"
    MANUAL = "manual"

@dataclass
class RetrainingTrigger:
    trigger_type: TriggerType
    threshold: Optional[float] = None
    schedule: Optional[str] = None
    
def check_retraining_needed(
    current_auc: float,
    baseline_auc: float,
    drift_score: float,
    last_training_date: datetime,
    new_data_count: int
) -> tuple[bool, TriggerType]:
    """Vérifier si le réentraînement est nécessaire."""
    
    triggers = [
        # Dégradation de la performance
        RetrainingTrigger(
            trigger_type=TriggerType.PERFORMANCE,
            threshold=0.05  # Baisse de 5% de l'AUC
        ),
        # Dérive des données
        RetrainingTrigger(
            trigger_type=TriggerType.DRIFT,
            threshold=0.1  # Score de dérive > 0.1
        ),
        # Réentraînement programmé
        RetrainingTrigger(
            trigger_type=TriggerType.SCHEDULED,
            schedule="30 days"
        ),
        # Volume de nouvelles données
        RetrainingTrigger(
            trigger_type=TriggerType.DATA_VOLUME,
            threshold=100000  # 100k nouveaux enregistrements
        ),
    ]
    
    # Vérifier la performance
    if baseline_auc - current_auc > 0.05:
        return True, TriggerType.PERFORMANCE
    
    # Vérifier la dérive
    if drift_score > 0.1:
        return True, TriggerType.DRIFT
    
    # Vérifier le calendrier
    if datetime.now() - last_training_date > timedelta(days=30):
        return True, TriggerType.SCHEDULED
    
    # Vérifier le volume de données
    if new_data_count > 100000:
        return True, TriggerType.DATA_VOLUME
    
    return False, None

Points clés à retenir

Résumé

1. La définition du problème est cruciale - des objectifs flous mènent à des projets échoués
2. La qualité des données détermine la qualité du modèle - des données de mauvaise qualité donnent des résultats de mauvaise qualité
3. L'ingénierie des caractéristiques est souvent plus importante que la sélection du modèle
4. L'évaluation doit être complète et correspondre aux métriques commerciales
5. La surveillance n'est pas optionnelle - les modèles se dégradent sans elle
6. Le réentraînement maintient les modèles pertinents à mesure que les données évoluent

Lectures complémentaires

• Les règles de l'apprentissage automatique de Google
• La plateforme Michelangelo d'Uber
• L'infrastructure ML de Netflix

---

Suivant : Quiz Module 1