Capítulo 3: O Ciclo de Vida de ML

Introdução

O ciclo de vida de ML abrange todas as etapas, desde a definição do problema até a aposentadoria do modelo. Compreender esse ciclo é fundamental para implementar práticas eficazes de MLOps.

O Ciclo de Vida Completo de ML

Etapa 1: Definição do Problema

Antes de iniciar qualquer trabalho de ML, defina claramente o problema.

Perguntas-Chave a Responder

Pergunta	Exemplo de Resposta
Qual problema de negócio estamos resolvendo?	Reduzir o churn de clientes em 15%
Que tipo de problema de ML é este?	Classificação binária
Qual é a métrica de sucesso?	AUC-ROC > 0,85
Quais são os requisitos de latência?	< 100ms para previsões
Quais dados estão disponíveis?	2 anos de dados de clientes

Documento de Definição do 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: Coleta de Dados

Coletar e entender os dados disponíveis para o projeto de ML.

Inventário de Dados

# 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 Coleta de Dados

# 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: Preparação dos Dados

Limpeza e preparação dos dados para engenharia de atributos e treinamento.

Verificações de Qualidade dos Dados

# 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 Limpeza de Dados

# 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: Engenharia de Atributos

Criação de atributos que ajudam o modelo a aprender padrões.

Pipeline de Engenharia de Atributos

# 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: Treinamento do Modelo

Treinamento e ajuste de modelos com rastreamento adequado de experimentos.

Pipeline de Treinamento com 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: Avaliação do Modelo

Avaliação abrangente antes do deploy.

Framework de Avaliação

# 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: Deploy do Modelo

Realizando o deploy do modelo em produção.

Configuração de Deploy

# 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: Monitoramento

Monitoramento contínuo da performance do modelo.

Métricas de Monitoramento

# 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: Re-treinamento

Disparo do re-treinamento do modelo quando necessário.

Gatilhos de Re-treinamento

# 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

Principais Aprendizados

Summary

Definição do problema é crucial – objetivos pouco claros levam a projetos fracassados
Qualidade dos dados determina a qualidade do modelo – lixo entra, lixo sai
Engenharia de atributos muitas vezes importa mais do que a escolha do modelo
Avaliação deve ser abrangente e alinhada às métricas do negócio
Monitoramento não é opcional – modelos degradam sem acompanhamento
Re-treinamento mantém os modelos relevantes à medida que os dados evoluem

Leituras Complementares

Próximo: Quiz do Módulo 1

Introdução​

O Ciclo de Vida Completo de ML​

Etapa 1: Definição do Problema​

Perguntas-Chave a Responder​

Documento de Definição do Problema​

Etapa 2: Coleta de Dados​

Inventário de Dados​

Pipeline de Coleta de Dados​

Etapa 3: Preparação dos Dados​

Verificações de Qualidade dos Dados​

Pipeline de Limpeza de Dados​

Etapa 4: Engenharia de Atributos​

Pipeline de Engenharia de Atributos​

Etapa 5: Treinamento do Modelo​

Pipeline de Treinamento com MLflow​

Etapa 6: Avaliação do Modelo​

Framework de Avaliação​

Etapa 7: Deploy do Modelo​

Configuração de Deploy​

Etapa 8: Monitoramento​

Métricas de Monitoramento​

Etapa 9: Re-treinamento​

Gatilhos de Re-treinamento​

Principais Aprendizados​

Leituras Complementares​