Capítulo 3: Versionamento de Modelos

Introdução

O versionamento de modelos é fundamental para sistemas de ML em produção. Ele permite rastrear, comparar e reverter modelos, mantendo uma linhagem clara dos dados até o deploy.

Por que versionar modelos?

O desafio da gestão de modelos

Benefícios do Versionamento de Modelos

Benefício	Descrição
Reprodutibilidade	Recrie qualquer versão do modelo
Linhagem	Rastreie dados → modelo → previsões
Comparação	Compare desempenho entre versões
Rollback	Reverta rapidamente para versões anteriores
Compliance	Rastro de auditoria para setores regulados

Conceitos de Model Registry

Componentes principais

Estágios do Ciclo de Vida do Modelo

Estágio	Descrição
None	Estado inicial, recém-registrado
Staging	Modelo em teste/validação
Production	Modelo atendendo tráfego real
Archived	Modelo obsoleto, mantido para referência

MLflow Model Registry

Configurando o MLflow

# Instale o MLflow
pip install mlflow

# Inicie o servidor MLflow com backend store
mlflow server \
    --backend-store-uri postgresql://user:pass@localhost/mlflow \
    --default-artifact-root s3://mlflow-artifacts/ \
    --host 0.0.0.0 \
    --port 5000

Registrando Modelos

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

# Defina o tracking URI
mlflow.set_tracking_uri("http://localhost:5000")

# Treine e registre o modelo
with mlflow.start_run() as run:
    # Treine o modelo
    model = RandomForestClassifier(n_estimators=100)
    model.fit(X_train, y_train)
    
    # Registre parâmetros
    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("max_depth", None)
    
    # Registre métricas
    accuracy = model.score(X_test, y_test)
    mlflow.log_metric("accuracy", accuracy)
    
    # Registre o modelo
    mlflow.sklearn.log_model(
        model,
        "model",
        registered_model_name="customer-churn-model"
    )
    
    print(f"Model registered with run_id: {run.info.run_id}")

Gerenciando Versões de Modelos

# model_management.py
from mlflow.tracking import MlflowClient

client = MlflowClient()

# Obtenha todas as versões de um modelo
versions = client.search_model_versions("name='customer-churn-model'")
for v in versions:
    print(f"Version {v.version}: {v.current_stage}")

# Transicione o modelo para staging
client.transition_model_version_stage(
    name="customer-churn-model",
    version=2,
    stage="Staging"
)

# Transicione o modelo para produção
client.transition_model_version_stage(
    name="customer-churn-model",
    version=2,
    stage="Production",
    archive_existing_versions=True  # Arquiva a versão atual em produção
)

# Adicione descrição à versão do modelo
client.update_model_version(
    name="customer-churn-model",
    version=2,
    description="Engenharia de features aprimorada, AUC aumentou em 5%"
)

# Adicione tags
client.set_model_version_tag(
    name="customer-churn-model",
    version=2,
    key="validation_status",
    value="approved"
)

Carregando Modelos do Registry

# load_model.py
import mlflow

# Carregue versão específica
model_v1 = mlflow.sklearn.load_model(
    "models:/customer-churn-model/1"
)

# Carregue por estágio
production_model = mlflow.sklearn.load_model(
    "models:/customer-churn-model/Production"
)

# Carregue modelo em staging
staging_model = mlflow.sklearn.load_model(
    "models:/customer-churn-model/Staging"
)

# Faça previsões
predictions = production_model.predict(X_new)

Metadados e Linhagem do Modelo

Model Card Completo

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

@dataclass
class ModelCard:
    """Documentação estruturada do modelo seguindo padrões da indústria."""
    
    # Informações básicas
    model_name: str
    version: str
    description: str
    owner: str
    created_at: str
    
    # Informações de treinamento
    training_data: Dict[str, str]
    features: List[str]
    target: str
    algorithm: str
    hyperparameters: Dict[str, any]
    
    # Desempenho
    metrics: Dict[str, float]
    evaluation_data: str
    
    # Uso pretendido
    intended_use: str
    out_of_scope_use: str
    
    # Limitações e vieses
    limitations: List[str]
    ethical_considerations: List[str]
    
    # Linhagem
    parent_model: Optional[str] = None
    training_run_id: Optional[str] = None
    data_version: Optional[str] = None
    code_version: Optional[str] = None

# Crie o model card
model_card = ModelCard(
    model_name="customer-churn-model",
    version="2.1.0",
    description="Prediz a probabilidade de churn do cliente em até 30 dias",
    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="Identificar clientes em risco para campanhas de retenção",
    out_of_scope_use="Score de crédito, decisões de contratação",
    
    limitations=[
        "Desempenho pode degradar para clientes com menos de 30 dias",
        "Não validado para clientes do segmento enterprise",
        "Requer no mínimo 6 meses de histórico de transações"
    ],
    ethical_considerations=[
        "O modelo não deve ser usado para negar serviço",
        "Previsões devem ser revisadas por humanos antes de ação"
    ],
    
    parent_model="customer-churn-model/1.0.0",
    training_run_id="abc123def456",
    data_version="data-v2.1.0",
    code_version="git-sha-789xyz"
)

# Salve o model card
with open("model_card.json", "w") as f:
    json.dump(asdict(model_card), f, indent=2)

Registrando Linhagem do Modelo

# lineage_tracking.py
import mlflow
import git
import hashlib

def get_code_version():
    """Obtenha o hash do commit git atual."""
    repo = git.Repo(search_parent_directories=True)
    return repo.head.object.hexsha

def get_data_hash(data_path: str) -> str:
    """Calcule o hash do arquivo de dados."""
    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):
    """Treine modelo com rastreamento completo de linhagem."""
    
    with mlflow.start_run() as run:
        # Rastreie versão do código
        code_version = get_code_version()
        mlflow.set_tag("code_version", code_version)
        mlflow.set_tag("git_repo", "https://github.com/company/ml-project")
        
        # Rastreie versão dos dados
        data_hash = get_data_hash(data_path)
        mlflow.set_tag("data_hash", data_hash)
        mlflow.set_tag("data_path", data_path)
        
        # Rastreie ambiente
        mlflow.set_tag("python_version", "3.10.0")
        mlflow.set_tag("sklearn_version", "1.3.0")
        
        # Rastreie configuração de treinamento
        mlflow.log_param("training_samples", len(X_train))
        mlflow.log_param("test_samples", len(X_test))
        mlflow.log_param("features", list(X_train.columns))
        
        # Treine o modelo
        model = GradientBoostingClassifier(
            n_estimators=200,
            max_depth=7,
            learning_rate=0.1
        )
        model.fit(X_train, y_train)
        
        # Registre métricas
        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)
        
        # Registre modelo com assinatura
        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

Estratégias de Versionamento de Modelos

Versionamento Semântico para Modelos

MAJOR.MINOR.PATCH

Exemplos:
- 1.0.0 → 1.0.1  # Patch: Correção de bug, sem re-treinamento
- 1.0.0 → 1.1.0  # Minor: Re-treinado, mesma arquitetura
- 1.0.0 → 2.0.0  # Major: Nova arquitetura ou mudanças incompatíveis

Incremento Automático de Versão

# 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:
    """Calcule a próxima versão semântica."""
    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
):
    """Registre modelo com versionamento adequado."""
    client = MlflowClient()
    
    # Obtenha a última versão
    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"
    
    # Calcule nova versão semântica
    new_semantic = get_next_version(current_semantic, bump)
    
    # Registre modelo
    result = client.create_model_version(
        name=model_name,
        source=f"runs:/{run_id}/model",
        run_id=run_id,
        description=description
    )
    
    # Defina tag de versão semântica
    client.set_model_version_tag(
        name=model_name,
        version=result.version,
        key="semantic_version",
        value=new_semantic
    )
    
    print(f"Registered {model_name} v{result.version} (semantic: {new_semantic})")
    return result

Workflow de Aprovação de Modelos

Pipeline de Validação Automatizada

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

class ModelValidator:
    """Valide modelos antes do deploy em produção."""
    
    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]]:
        """Verifique se o modelo atende aos thresholds de desempenho."""
        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)
        }
        
        # Verifique thresholds
        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]:
        """Verifique se o modelo atende ao requisito de latência."""
        import time
        
        model = mlflow.sklearn.load_model(self.model_uri)
        
        # Aqueça o modelo
        _ = model.predict(X_sample[:1])
        
        # Meça latência
        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]:
        """Verifique se as previsões batem com o modelo em produção."""
        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  # Pelo menos 95% de concordância
        
        return passed, agreement
    
    def run_all_validations(
        self,
        X_test,
        y_test,
        thresholds: Dict[str, float]
    ) -> Dict[str, any]:
        """Execute todas as validações e retorne os resultados."""
        results = {
            "model_name": self.model_name,
            "version": self.version,
            "validations": {}
        }
        
        # Validação de desempenho
        perf_passed, perf_metrics = self.validate_performance(
            X_test, y_test, thresholds
        )
        results["validations"]["performance"] = {
            "passed": perf_passed,
            "metrics": perf_metrics
        }
        
        # Validação de latência
        latency_passed, p99_latency = self.validate_latency(X_test[:100])
        results["validations"]["latency"] = {
            "passed": latency_passed,
            "p99_ms": p99_latency
        }
        
        # Resultado geral
        results["approved"] = all(
            v["passed"] for v in results["validations"].values()
        )
        
        # Atualize tags do modelo
        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

Principais Aprendizados

Summary

1. Use um model registry para gestão centralizada de modelos
2. Rastreie a linhagem completa – dados, código, parâmetros, ambiente
3. Implemente versionamento semântico para evolução clara das versões
4. Crie model cards para documentação e compliance
5. Automatize a validação antes do deploy em produção
6. Mantenha rastros de auditoria para setores regulados

Exercício Prático

1. Configure o servidor de tracking do MLflow
2. Treine e registre múltiplas versões de modelos
3. Crie um model card para seu melhor modelo
4. Implemente o pipeline de validação de modelos
5. Pratique a transição de modelos entre estágios

---

Próximo: Quiz do Módulo 2