Capítulo 3: El ciclo de vida del ML

Introducción

El ciclo de vida del ML abarca todas las etapas desde la definición del problema hasta el retiro del modelo. Comprender este ciclo de vida es fundamental para implementar prácticas efectivas de MLOps.

El ciclo de vida completo del ML

Etapa 1: Definición del problema

Antes de iniciar cualquier trabajo de ML, defina claramente el problema.

Preguntas clave a responder

Pregunta	Ejemplo de respuesta
¿Qué problema de negocio estamos resolviendo?	Reducir la deserción de clientes en un 15%
¿Qué tipo de problema de ML es este?	Clasificación binaria
¿Cuál es la métrica de éxito?	AUC-ROC > 0.85
¿Cuáles son los requisitos de latencia?	< 100ms para predicciones
¿Qué datos están disponibles?	2 años de datos de clientes

Documento de definición del problema

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

business_context:
  objective: Predict customers likely to churn within 30 days
  impact: $2M annual revenue at risk
  stakeholders:
    - Customer Success Team
    - Marketing Team
    - Product Team

ml_formulation:
  task_type: binary_classification
  target_variable: churned_within_30_days
  positive_class_definition: "Customer cancelled subscription"
  
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  # Need to explain predictions
  
timeline:
  mvp: 2024-03-01
  production: 2024-04-01

Etapa 2: Recolección de datos

Reunir y comprender los datos disponibles para el proyecto de ML.

Inventario de datos

# 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

# Define available data sources
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 recolección de datos

# 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:
    """Extract transaction data from 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:
    """Extract support ticket data."""
    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:
    """Merge all data sources."""
    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

Etapa 3: Preparación de datos

Limpieza y preparación de los datos para la ingeniería de características y el entrenamiento.

Verificaciones de calidad de datos

# data_quality.py
import great_expectations as gx

def validate_training_data(df):
    """Validate training data quality."""
    context = gx.get_context()
    
    # Create expectations
    expectations = [
        # No null values in critical columns
        {"expectation_type": "expect_column_values_to_not_be_null",
         "kwargs": {"column": "customer_id"}},
        
        # Values in expected range
        {"expectation_type": "expect_column_values_to_be_between",
         "kwargs": {"column": "age", "min_value": 18, "max_value": 120}},
        
        # Unique identifiers
        {"expectation_type": "expect_column_values_to_be_unique",
         "kwargs": {"column": "customer_id"}},
        
        # No future dates
        {"expectation_type": "expect_column_values_to_be_between",
         "kwargs": {"column": "signup_date", 
                    "max_value": datetime.now().isoformat()}},
        
        # Column exists
        {"expectation_type": "expect_column_to_exist",
         "kwargs": {"column": "churned"}},
    ]
    
    # Run 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 limpieza de datos

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

def clean_data(df: pd.DataFrame) -> pd.DataFrame:
    """Clean and prepare raw data."""
    
    # Make a copy
    df = df.copy()
    
    # Handle missing values
    df['age'] = df['age'].fillna(df['age'].median())
    df['income'] = df['income'].fillna(df.groupby('segment')['income'].transform('median'))
    
    # Remove duplicates
    df = df.drop_duplicates(subset=['customer_id'], keep='last')
    
    # Handle outliers (clip to 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)
    
    # Standardize text columns
    df['segment'] = df['segment'].str.lower().str.strip()
    
    # Convert data types
    df['signup_date'] = pd.to_datetime(df['signup_date'])
    df['churned'] = df['churned'].astype(int)
    
    return df

Etapa 4: Ingeniería de características

Creación de características que ayuden al modelo a aprender patrones.

Pipeline de ingeniería de características

# 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:
        """Create all features for the model."""
        features = pd.DataFrame()
        features['customer_id'] = df['customer_id']
        
        # Temporal features
        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
        
        # Aggregation features
        features['total_purchases'] = df['purchase_count']
        features['total_spend'] = df['total_spend']
        features['avg_order_value'] = df['total_spend'] / df['purchase_count'].clip(1)
        
        # Engagement features
        features['login_frequency_30d'] = df['logins_last_30d']
        features['support_tickets_90d'] = df['support_tickets_90d']
        features['nps_score'] = df['nps_score']
        
        # Ratio features
        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)
        )
        
        # Categorical encoding
        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:
        """Scale numerical features."""
        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

Etapa 5: Entrenamiento del modelo

Entrenamiento y ajuste de modelos con seguimiento adecuado de experimentos.

Pipeline de entrenamiento con 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
    ):
        """Train multiple models and select the best one."""
        
        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):
                # Log parameters
                mlflow.log_param("model_type", model_name)
                
                # Grid search for best hyperparameters
                grid_search = GridSearchCV(
                    config['model'],
                    config['params'],
                    cv=5,
                    scoring='roc_auc',
                    n_jobs=-1
                )
                grid_search.fit(X_train, y_train)
                
                # Get best model
                best_estimator = grid_search.best_estimator_
                
                # Log best parameters
                for param, value in grid_search.best_params_.items():
                    mlflow.log_param(param, value)
                
                # Evaluate on validation set
                val_predictions = best_estimator.predict_proba(X_val)[:, 1]
                val_auc = roc_auc_score(y_val, val_predictions)
                
                # Log metrics
                mlflow.log_metric("val_auc_roc", val_auc)
                mlflow.log_metric("cv_score", grid_search.best_score_)
                
                # Log model
                mlflow.sklearn.log_model(best_estimator, "model")
                
                # Track best model
                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

Etapa 6: Evaluación del modelo

Evaluación integral antes del despliegue.

Marco de evaluación

# 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:
        """Generate comprehensive evaluation report."""
        
        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()
        }
        
        # Calculate precision at different recall thresholds
        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:
        """Check if model meets deployment criteria."""
        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)

Etapa 7: Despliegue del modelo

Desplegar el modelo en producción.

Configuración de despliegue

# 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

Etapa 8: Monitoreo

Monitoreo continuo del desempeño del modelo.

Métricas de monitoreo

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

# Define metrics
prediction_counter = Counter(
    'model_predictions_total',
    'Total predictions made',
    ['model_version', 'prediction_class']
)

prediction_latency = Histogram(
    'model_prediction_latency_seconds',
    'Prediction latency',
    buckets=[0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5]
)

data_drift_gauge = Gauge(
    'model_data_drift_score',
    'Data drift score by feature',
    ['feature_name']
)

model_accuracy_gauge = Gauge(
    'model_accuracy_rolling',
    'Rolling accuracy over last 24 hours'
)

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):
        """Log a prediction for monitoring."""
        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):
        """Check for data drift using statistical tests."""
        from scipy import stats
        
        drift_scores = {}
        for i, feature in enumerate(feature_names):
            # Kolmogorov-Smirnov test
            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

Etapa 9: Reentrenamiento

Activar el reentrenamiento del modelo cuando sea necesario.

Disparadores de reentrenamiento

# 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]:
    """Check if retraining is needed."""
    
    triggers = [
        # Performance degradation
        RetrainingTrigger(
            trigger_type=TriggerType.PERFORMANCE,
            threshold=0.05  # 5% AUC drop
        ),
        # Data drift
        RetrainingTrigger(
            trigger_type=TriggerType.DRIFT,
            threshold=0.1  # Drift score > 0.1
        ),
        # Scheduled retraining
        RetrainingTrigger(
            trigger_type=TriggerType.SCHEDULED,
            schedule="30 days"
        ),
        # New data volume
        RetrainingTrigger(
            trigger_type=TriggerType.DATA_VOLUME,
            threshold=100000  # 100k new records
        ),
    ]
    
    # Check performance
    if baseline_auc - current_auc > 0.05:
        return True, TriggerType.PERFORMANCE
    
    # Check drift
    if drift_score > 0.1:
        return True, TriggerType.DRIFT
    
    # Check schedule
    if datetime.now() - last_training_date > timedelta(days=30):
        return True, TriggerType.SCHEDULED
    
    # Check data volume
    if new_data_count > 100000:
        return True, TriggerType.DATA_VOLUME
    
    return False, None

Puntos clave

Summary

1. La definición del problema es crucial: objetivos poco claros llevan a proyectos fallidos
2. La calidad de los datos determina la calidad del modelo: basura entra, basura sale
3. La ingeniería de características suele ser más importante que la selección del modelo
4. La evaluación debe ser integral y alineada con las métricas de negocio
5. El monitoreo no es opcional: los modelos se degradan sin él
6. El reentrenamiento mantiene los modelos relevantes a medida que los datos evolucionan

Lecturas adicionales

• Reglas de Machine Learning de Google
• Plataforma Michelangelo de Uber
• Infraestructura de ML de Netflix

---

Siguiente: Cuestionario del Módulo 1