Artigo 42 — MLOps: Versionamento de Modelos, MLflow e Deploy de ML

Prof. Ricardo Matos Módulo 7 · Data Science e Machine Learning · Artigo 42 de 52

Introdução

Treinar um modelo de machine learning é apenas metade do trabalho. A outra metade — colocar o modelo em produção, monitorá-lo, versioná-lo e mantê-lo funcionando ao longo do tempo — é o que chamamos de MLOps. É a disciplina que aplica os princípios de DevOps ao ciclo de vida de modelos de ML. Sem MLOps, modelos ficam presos em notebooks Jupyter e nunca chegam a gerar valor real. Neste artigo veremos MLflow para rastreamento e registro de experimentos, e como fazer deploy de modelos como APIs prontas para produção.

Instalação

pip install mlflow scikit-learn fastapi uvicorn joblib pandas numpy

O Ciclo de Vida de um Modelo ML

Dados         Experimentos      Registro        Deploy
  │                │               │               │
  ▼                ▼               ▼               ▼
Coleta  ──►  Treinar/Avaliar ──► Model         Servir como
Limpeza      Rastrear com        Registry      API REST
Feature      MLflow              Versionar     Monitor
Engineering  Comparar            Promover      Retreinar

MLflow: Rastreamento de Experimentos

MLflow registra automaticamente parâmetros, métricas, artefatos e modelos de cada execução:

import mlflow
import mlflow.sklearn
import mlflow.keras
import numpy as np
import pandas as pd
from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split, cross_val_score
from sklearn.preprocessing import StandardScaler
from sklearn.pipeline import Pipeline
from sklearn.metrics import (
    accuracy_score, f1_score, roc_auc_score,
    precision_score, recall_score
)
import joblib


# Dataset de exemplo
np.random.seed(42)
n = 2000
X = np.column_stack([
    np.random.uniform(1, 10, n),
    np.random.randint(0, 20, n),
    np.random.uniform(3, 10, n),
    np.random.uniform(0, 5,  n),
])
y = ((0.5*X[:,0] + 0.3*X[:,2] - 0.1*X[:,1] + 0.2*X[:,3]
      + np.random.normal(0, 0.5, n)) >= 5).astype(int)

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

feature_names = ["horas_estudo", "faltas", "nota_anterior", "participacao"]


# Configurando MLflow
mlflow.set_tracking_uri("sqlite:///mlflow.db")   # banco local
mlflow.set_experiment("previsao_aprovacao")


def treinar_e_registrar(nome_modelo, modelo, params: dict):
    """Treina modelo e registra tudo no MLflow."""

    with mlflow.start_run(run_name=nome_modelo):

        # Registra parâmetros
        mlflow.log_params(params)
        mlflow.log_param("modelo_tipo", type(modelo).__name__)
        mlflow.log_param("n_treino",    len(X_train))
        mlflow.log_param("n_teste",     len(X_test))
        mlflow.log_param("features",    feature_names)

        # Pipeline com preprocessamento
        pipe = Pipeline([
            ("scaler", StandardScaler()),
            ("clf",    modelo)
        ])
        pipe.fit(X_train, y_train)

        # Predições
        y_pred = pipe.predict(X_test)
        y_prob = pipe.predict_proba(X_test)[:, 1]

        # Métricas
        metricas = {
            "accuracy":  accuracy_score(y_test, y_pred),
            "f1":        f1_score(y_test, y_pred),
            "precision": precision_score(y_test, y_pred),
            "recall":    recall_score(y_test, y_pred),
            "roc_auc":   roc_auc_score(y_test, y_prob),
        }

        # Validação cruzada
        cv_scores = cross_val_score(pipe, X_train, y_train, cv=5, scoring="f1")
        metricas["cv_f1_mean"] = cv_scores.mean()
        metricas["cv_f1_std"]  = cv_scores.std()

        mlflow.log_metrics(metricas)

        # Salvando artefatos
        joblib.dump(pipe, "pipeline_temp.pkl")
        mlflow.log_artifact("pipeline_temp.pkl", "modelo")

        # Registrando o modelo com assinatura
        signature = mlflow.models.infer_signature(X_train, y_pred)
        input_example = X_train[:3]

        mlflow.sklearn.log_model(
            pipe,
            artifact_path="modelo_sklearn",
            registered_model_name=f"aprovacao_{nome_modelo.lower().replace(' ', '_')}",
            signature=signature,
            input_example=input_example
        )

        run_id = mlflow.active_run().info.run_id
        print(f"\n[{nome_modelo}]")
        print(f"  Run ID:   {run_id[:8]}...")
        print(f"  F1:       {metricas['f1']:.4f}")
        print(f"  AUC:      {metricas['roc_auc']:.4f}")
        print(f"  CV F1:    {metricas['cv_f1_mean']:.4f} ± {metricas['cv_f1_std']:.4f}")

        return run_id, metricas


# Rodando múltiplos experimentos
experimentos = [
    ("Logistic Regression", LogisticRegression(max_iter=1000), {
        "C": 1.0, "solver": "lbfgs"
    }),
    ("Random Forest", RandomForestClassifier(random_state=42), {
        "n_estimators": 100, "max_depth": 10,
        "min_samples_split": 5
    }),
    ("Gradient Boosting", GradientBoostingClassifier(random_state=42), {
        "n_estimators": 100, "learning_rate": 0.1,
        "max_depth": 3, "subsample": 0.8
    }),
    ("Random Forest Tuned", RandomForestClassifier(
        n_estimators=200, max_depth=15,
        min_samples_split=3, random_state=42
    ), {
        "n_estimators": 200, "max_depth": 15,
        "min_samples_split": 3
    }),
]

resultados_runs = []
for nome, modelo, params in experimentos:
    run_id, metricas = treinar_e_registrar(nome, modelo, params)
    resultados_runs.append({"nome": nome, "run_id": run_id, **metricas})

# Comparando no MLflow
df_resultados = pd.DataFrame(resultados_runs).set_index("nome")
print("\n=== Comparação de Experimentos ===")
print(df_resultados[["f1", "roc_auc", "cv_f1_mean"]].round(4))

Model Registry: Promovendo Modelos

from mlflow.tracking import MlflowClient


def gerenciar_model_registry():
    client = MlflowClient(tracking_uri="sqlite:///mlflow.db")

    # Listando modelos registrados
    print("=== Modelos Registrados ===")
    for model in client.search_registered_models():
        print(f"\n  {model.name}")
        for v in model.latest_versions:
            print(f"    v{v.version} — {v.current_stage} — run: {v.run_id[:8]}")

    # Buscando o melhor run pelo AUC
    experimento = mlflow.get_experiment_by_name("previsao_aprovacao")
    runs = mlflow.search_runs(
        experiment_ids=[experimento.experiment_id],
        order_by=["metrics.roc_auc DESC"],
        max_results=1
    )

    melhor_run_id = runs.iloc[0]["run_id"]
    melhor_auc    = runs.iloc[0]["metrics.roc_auc"]
    print(f"\nMelhor run: {melhor_run_id[:8]} (AUC={melhor_auc:.4f})")

    # Promovendo modelo para Staging
    nome_modelo = "aprovacao_random_forest_tuned"
    versoes     = client.get_latest_versions(nome_modelo)

    if versoes:
        versao = versoes[-1].version
        client.transition_model_version_stage(
            name=nome_modelo,
            version=versao,
            stage="Staging",
            archive_existing_versions=True
        )
        print(f"Modelo {nome_modelo} v{versao} promovido para Staging")

        # Adicionando descrição
        client.update_model_version(
            name=nome_modelo,
            version=versao,
            description=f"AUC={melhor_auc:.4f} — Aprovado para staging"
        )

        # Adicionando tags
        client.set_model_version_tag(
            name=nome_modelo,
            version=versao,
            key="validado_por",
            value="Prof. Ricardo Matos"
        )

    return nome_modelo, versao


nome_modelo_final, versao_final = gerenciar_model_registry()

Carregando Modelos do Registry

def carregar_modelo_producao(nome: str, stage: str = "Staging"):
    """Carrega modelo diretamente do MLflow Registry."""
    uri    = f"models:/{nome}/{stage}"
    modelo = mlflow.sklearn.load_model(uri)
    print(f"Modelo carregado: {uri}")
    return modelo


def carregar_modelo_por_run(run_id: str):
    """Carrega modelo por run_id específico."""
    uri    = f"runs:/{run_id}/modelo_sklearn"
    modelo = mlflow.sklearn.load_model(uri)
    return modelo


# Predição com modelo do registry
modelo_prod = carregar_modelo_producao(nome_modelo_final)

novos_alunos = np.array([
    [8.0, 2,  9.0, 4.0],
    [2.0, 18, 4.0, 0.5],
    [6.5, 5,  7.5, 3.0],
])

probs = modelo_prod.predict_proba(novos_alunos)[:, 1]
preds = modelo_prod.predict(novos_alunos)

print("\n=== Predições com Modelo de Produção ===")
for i, (prob, pred) in enumerate(zip(probs, preds), 1):
    status = "Aprovado" if pred == 1 else "Reprovado"
    print(f"  Aluno {i}: {status} ({prob:.1%})")

Deploy como API REST com FastAPI

# api_modelo.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List
import mlflow.sklearn
import numpy as np
import os
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="API de Previsão de Aprovação",
    description="Modelo ML para prever aprovação de alunos",
    version="1.0.0"
)


# Modelos Pydantic
class DadosAluno(BaseModel):
    horas_estudo:   float = Field(..., ge=0,  le=24,  description="Horas de estudo por dia")
    faltas:         int   = Field(..., ge=0,  le=100, description="Número de faltas")
    nota_anterior:  float = Field(..., ge=0,  le=10,  description="Nota do período anterior")
    participacao:   float = Field(..., ge=0,  le=5,   description="Score de participação (0-5)")

    model_config = {
        "json_schema_extra": {
            "example": {
                "horas_estudo":   7.5,
                "faltas":         2,
                "nota_anterior":  8.0,
                "participacao":   4.0
            }
        }
    }


class PrevisaoResposta(BaseModel):
    aprovado:       bool
    probabilidade:  float
    confianca:      str
    timestamp:      str


class LoteAlunos(BaseModel):
    alunos: List[DadosAluno]


class RespostaLote(BaseModel):
    total:       int
    aprovados:   int
    reprovados:  int
    previsoes:   List[PrevisaoResposta]


# Carregando modelo na inicialização
modelo = None

@app.on_event("startup")
async def carregar_modelo():
    global modelo
    try:
        mlflow.set_tracking_uri("sqlite:///mlflow.db")
        modelo = mlflow.sklearn.load_model(
            f"models:/aprovacao_random_forest_tuned/Staging"
        )
        logger.info("Modelo carregado com sucesso.")
    except Exception as e:
        logger.warning(f"Modelo MLflow não disponível: {e}")
        # Fallback — modelo local
        if os.path.exists("pipeline_temp.pkl"):
            import joblib
            modelo = joblib.load("pipeline_temp.pkl")
            logger.info("Modelo local carregado como fallback.")


def preparar_features(dados: DadosAluno) -> np.ndarray:
    return np.array([[
        dados.horas_estudo,
        dados.faltas,
        dados.nota_anterior,
        dados.participacao
    ]])


def interpretar_confianca(probabilidade: float) -> str:
    if probabilidade >= 0.85 or probabilidade <= 0.15:
        return "Alta"
    elif probabilidade >= 0.70 or probabilidade <= 0.30:
        return "Média"
    return "Baixa"


@app.get("/", tags=["Saúde"])
async def raiz():
    return {
        "api":     "Previsão de Aprovação",
        "versao":  "1.0.0",
        "status":  "online",
        "modelo":  "carregado" if modelo else "indisponível"
    }


@app.get("/health", tags=["Saúde"])
async def health():
    return {
        "status":  "healthy",
        "modelo":  modelo is not None,
        "timestamp": datetime.utcnow().isoformat()
    }


@app.post("/prever", response_model=PrevisaoResposta, tags=["Previsão"])
async def prever(dados: DadosAluno):
    if modelo is None:
        raise HTTPException(status_code=503, detail="Modelo não disponível.")

    try:
        features    = preparar_features(dados)
        prob        = float(modelo.predict_proba(features)[0, 1])
        aprovado    = prob >= 0.5

        logger.info(f"Predição: aprovado={aprovado}, prob={prob:.4f}")

        return PrevisaoResposta(
            aprovado=      aprovado,
            probabilidade= round(prob, 4),
            confianca=     interpretar_confianca(prob),
            timestamp=     datetime.utcnow().isoformat()
        )
    except Exception as e:
        logger.error(f"Erro na predição: {e}")
        raise HTTPException(status_code=500, detail=str(e))


@app.post("/prever/lote", response_model=RespostaLote, tags=["Previsão"])
async def prever_lote(dados: LoteAlunos):
    if modelo is None:
        raise HTTPException(status_code=503, detail="Modelo não disponível.")
    if len(dados.alunos) > 1000:
        raise HTTPException(status_code=400, detail="Máximo de 1000 alunos por lote.")

    try:
        features = np.array([[
            a.horas_estudo, a.faltas,
            a.nota_anterior, a.participacao
        ] for a in dados.alunos])

        probs    = modelo.predict_proba(features)[:, 1]
        preds    = (probs >= 0.5).astype(bool)

        previsoes = [
            PrevisaoResposta(
                aprovado=      bool(pred),
                probabilidade= round(float(prob), 4),
                confianca=     interpretar_confianca(float(prob)),
                timestamp=     datetime.utcnow().isoformat()
            )
            for pred, prob in zip(preds, probs)
        ]

        return RespostaLote(
            total=      len(dados.alunos),
            aprovados=  int(preds.sum()),
            reprovados= int((~preds).sum()),
            previsoes=  previsoes
        )
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))


@app.get("/modelo/info", tags=["Modelo"])
async def info_modelo():
    if modelo is None:
        raise HTTPException(status_code=503, detail="Modelo não disponível.")
    return {
        "tipo":    type(modelo).__name__,
        "steps":   [s[0] for s in modelo.steps] if hasattr(modelo, "steps") else [],
        "features": ["horas_estudo", "faltas", "nota_anterior", "participacao"],
        "versao":  "1.0.0"
    }


# Executar:
# uvicorn api_modelo:app --reload --port 8001

Monitoramento de Drift

from scipy import stats
import pandas as pd
import numpy as np


class MonitorDrift:
    """Detecta degradação do modelo em produção."""

    def __init__(self, dados_referencia: np.ndarray, labels: list):
        self.referencia = dados_referencia
        self.labels     = labels
        self._historico: list = []

    def detectar_data_drift(
        self,
        dados_novos: np.ndarray,
        threshold: float = 0.05
    ) -> dict:
        """Teste KS para cada feature — detecta mudança na distribuição."""
        resultados = {}

        for i, label in enumerate(self.labels):
            stat, p_valor = stats.ks_2samp(
                self.referencia[:, i],
                dados_novos[:, i]
            )
            drift_detectado = p_valor < threshold

            resultados[label] = {
                "ks_statistic":  round(float(stat),    4),
                "p_valor":       round(float(p_valor),  4),
                "drift":         drift_detectado,
                "severidade":    "Alta" if stat > 0.3 else ("Média" if stat > 0.1 else "Baixa")
            }

        n_drift = sum(1 for r in resultados.values() if r["drift"])
        print(f"\n=== Relatório de Data Drift ===")
        print(f"Features com drift: {n_drift}/{len(self.labels)}")

        for feat, res in resultados.items():
            status = "⚠️ DRIFT" if res["drift"] else "✓ OK"
            print(f"  {feat:20} {status:12} "
                  f"KS={res['ks_statistic']:.4f} "
                  f"p={res['p_valor']:.4f}")

        return resultados

    def detectar_model_drift(
        self,
        previsoes_novas: np.ndarray,
        previsoes_ref:   np.ndarray,
        threshold: float = 0.1
    ) -> dict:
        """Detecta mudança na distribuição das predições."""
        media_ref  = previsoes_ref.mean()
        media_nova = previsoes_novas.mean()
        variacao   = abs(media_nova - media_ref)
        drift       = variacao > threshold

        resultado = {
            "media_referencia": round(float(media_ref),  4),
            "media_atual":      round(float(media_nova), 4),
            "variacao":         round(float(variacao),   4),
            "drift":            drift,
            "alerta":           "🚨 ALERTA" if drift else "✓ Normal"
        }

        print(f"\n=== Drift nas Predições ===")
        print(f"  Referência: {resultado['media_referencia']:.4f}")
        print(f"  Atual:      {resultado['media_atual']:.4f}")
        print(f"  Variação:   {resultado['variacao']:.4f}")
        print(f"  Status:     {resultado['alerta']}")

        return resultado

    def registrar_batch(
        self,
        dados:     np.ndarray,
        previsoes: np.ndarray,
        timestamp: str = None
    ):
        """Registra batch de predições para monitoramento contínuo."""
        self._historico.append({
            "timestamp":    timestamp or pd.Timestamp.now().isoformat(),
            "n_amostras":   len(dados),
            "media_prob":   float(previsoes.mean()),
            "pct_positivo": float((previsoes >= 0.5).mean()),
        })

    def relatorio_deriva_temporal(self) -> pd.DataFrame:
        """Exibe evolução das predições ao longo do tempo."""
        if not self._historico:
            return pd.DataFrame()
        df = pd.DataFrame(self._historico)
        df["timestamp"] = pd.to_datetime(df["timestamp"])
        return df.set_index("timestamp")


# Simulando monitoramento
monitor = MonitorDrift(X_train, feature_names)

# Dados normais
X_normal = X_train + np.random.normal(0, 0.1, X_train.shape)
monitor.detectar_data_drift(X_normal)

# Dados com drift simulado
X_drift = X_train.copy()
X_drift[:, 0] *= 1.5   # horas_estudo 50% maiores
X_drift[:, 1] *= 0.5   # faltas 50% menores
monitor.detectar_data_drift(X_drift)

Retreinamento Automático

import mlflow
from datetime import datetime


def retreinar_se_necessario(
    modelo_atual,
    X_novos:     np.ndarray,
    y_novos:     np.ndarray,
    threshold_f1: float = 0.85
) -> bool:
    """Retreina o modelo se a performance cair abaixo do threshold."""

    # Avalia modelo atual nos novos dados
    y_pred    = modelo_atual.predict(X_novos)
    f1_atual  = f1_score(y_novos, y_pred)

    print(f"F1 atual nos novos dados: {f1_atual:.4f}")
    print(f"Threshold:                {threshold_f1:.4f}")

    if f1_atual >= threshold_f1:
        print("✓ Modelo dentro do threshold. Retreinamento não necessário.")
        return False

    print("⚠️ Performance abaixo do threshold. Iniciando retreinamento...")

    # Combina dados antigos e novos
    X_combinado = np.vstack([X_train, X_novos])
    y_combinado = np.concatenate([y_train, y_novos])

    # Retreina com MLflow
    with mlflow.start_run(run_name=f"retreino_{datetime.now().strftime('%Y%m%d_%H%M')}"):
        mlflow.log_param("motivo",        "drift_performance")
        mlflow.log_param("f1_antes",      f1_atual)
        mlflow.log_param("n_novos_dados", len(X_novos))

        novo_pipeline = Pipeline([
            ("scaler", StandardScaler()),
            ("clf",    RandomForestClassifier(n_estimators=200, random_state=42))
        ])
        novo_pipeline.fit(X_combinado, y_combinado)

        # Avalia no conjunto de teste original
        y_pred_novo = novo_pipeline.predict(X_test)
        f1_novo     = f1_score(y_test, y_pred_novo)
        auc_novo    = roc_auc_score(y_test,
                                     novo_pipeline.predict_proba(X_test)[:, 1])

        mlflow.log_metric("f1_apos",  f1_novo)
        mlflow.log_metric("auc_apos", auc_novo)

        mlflow.sklearn.log_model(
            novo_pipeline,
            artifact_path="modelo_retreinado",
            registered_model_name="aprovacao_random_forest_tuned"
        )

        print(f"✓ Retreinamento concluído: F1={f1_novo:.4f}, AUC={auc_novo:.4f}")

    return True


# Simulando dados novos com leve degradação
X_novos_sim = X_test + np.random.normal(0, 2, X_test.shape)
y_novos_sim = y_test.copy()

retreinar_se_necessario(modelo_prod, X_novos_sim, y_novos_sim, threshold_f1=0.99)

docker-compose para MLflow + API

# docker-compose.mlops.yml
version: "3.9"

services:
  mlflow:
    image: python:3.12-slim
    container_name: mlflow-server
    working_dir: /mlflow
    command: >
      bash -c "pip install mlflow boto3 psycopg2-binary &&
               mlflow server
               --host 0.0.0.0
               --port 5000
               --backend-store-uri sqlite:///mlflow.db
               --default-artifact-root /mlflow/artifacts"
    ports:
      - "5000:5000"
    volumes:
      - mlflow_data:/mlflow

  api-ml:
    build:
      context: .
      dockerfile: Dockerfile.ml
    container_name: api-previsao
    ports:
      - "8001:8001"
    environment:
      - MLFLOW_TRACKING_URI=http://mlflow:5000
      - MODEL_NAME=aprovacao_random_forest_tuned
      - MODEL_STAGE=Production
    depends_on:
      - mlflow
    command: uvicorn api_modelo:app --host 0.0.0.0 --port 8001

volumes:
  mlflow_data:

# Dockerfile.ml
FROM python:3.12-slim

WORKDIR /app
COPY requirements_ml.txt .
RUN pip install --no-cache-dir -r requirements_ml.txt

COPY api_modelo.py .
COPY pipeline_temp.pkl .

EXPOSE 8001
CMD ["uvicorn", "api_modelo:app", "--host", "0.0.0.0", "--port", "8001"]

Exemplo Completo: Pipeline MLOps de Ponta a Ponta

from dataclasses import dataclass
from typing import Optional
import mlflow
import numpy as np
import pandas as pd
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import f1_score, roc_auc_score
from sklearn.model_selection import train_test_split
import joblib
from datetime import datetime


@dataclass
class ConfigMLOps:
    experimento:       str   = "producao_aprovacao"
    nome_modelo:       str   = "aprovacao_v2"
    threshold_f1:      float = 0.85
    threshold_auc:     float = 0.90
    n_estimators:      int   = 200
    max_depth:         int   = 15
    test_size:         float = 0.2
    tracking_uri:      str   = "sqlite:///mlops.db"


class PipelineMLOps:
    """Pipeline MLOps completo: treino → avaliação → registro → deploy."""

    def __init__(self, config: ConfigMLOps):
        self.config = config
        mlflow.set_tracking_uri(config.tracking_uri)
        mlflow.set_experiment(config.experimento)

    def executar(
        self,
        X: np.ndarray,
        y: np.ndarray,
        feature_names: list
    ) -> Optional[str]:

        print(f"\n{'='*55}")
        print(f"Pipeline MLOps — {datetime.now().strftime('%d/%m/%Y %H:%M')}")
        print('='*55)

        # 1. Divisão dos dados
        X_tr, X_te, y_tr, y_te = train_test_split(
            X, y,
            test_size=self.config.test_size,
            random_state=42,
            stratify=y
        )
        print(f"\n[1/5] Dados: treino={len(X_tr)} | teste={len(X_te)}")

        # 2. Treinamento
        with mlflow.start_run(run_name=f"run_{datetime.now().strftime('%Y%m%d_%H%M%S')}") as run:
            run_id = run.info.run_id
            print(f"[2/5] Treinando... (run: {run_id[:8]})")

            params = {
                "n_estimators": self.config.n_estimators,
                "max_depth":    self.config.max_depth,
                "random_state": 42,
                "test_size":    self.config.test_size,
                "n_features":   X.shape[1],
            }
            mlflow.log_params(params)

            pipeline = Pipeline([
                ("scaler", StandardScaler()),
                ("clf",    RandomForestClassifier(
                    n_estimators=self.config.n_estimators,
                    max_depth=self.config.max_depth,
                    random_state=42
                ))
            ])
            pipeline.fit(X_tr, y_tr)

            # 3. Avaliação
            y_pred = pipeline.predict(X_te)
            y_prob = pipeline.predict_proba(X_te)[:, 1]
            f1     = f1_score(y_te, y_pred)
            auc    = roc_auc_score(y_te, y_prob)

            metricas = {
                "f1":        round(f1,  4),
                "roc_auc":   round(auc, 4),
                "accuracy":  round(float((y_pred == y_te).mean()), 4),
                "n_treino":  len(X_tr),
                "n_teste":   len(X_te)
            }
            mlflow.log_metrics(metricas)

            print(f"[3/5] Avaliação: F1={f1:.4f} | AUC={auc:.4f}")

            # 4. Critério de aceitação
            aprovado = f1 >= self.config.threshold_f1 and auc >= self.config.threshold_auc

            mlflow.set_tag("aprovado",   str(aprovado))
            mlflow.set_tag("avaliador",  "pipeline_automatico")
            mlflow.set_tag("timestamp",  datetime.now().isoformat())

            if not aprovado:
                print(f"[4/5] ✗ Modelo reprovado "
                      f"(F1≥{self.config.threshold_f1}, AUC≥{self.config.threshold_auc})")
                return None

            print(f"[4/5] ✓ Modelo aprovado nos critérios de qualidade")

            # 5. Registro
            signature   = mlflow.models.infer_signature(X_tr, y_pred)
            model_info  = mlflow.sklearn.log_model(
                pipeline,
                artifact_path="modelo",
                registered_model_name=self.config.nome_modelo,
                signature=signature
            )

            # Salva localmente como fallback
            joblib.dump(pipeline, f"{self.config.nome_modelo}.pkl")
            mlflow.log_artifact(f"{self.config.nome_modelo}.pkl")

            print(f"[5/5] ✓ Modelo registrado: {model_info.model_uri}")

        return run_id


# Executando o pipeline
config   = ConfigMLOps(threshold_f1=0.80, threshold_auc=0.85)
pipeline_mlops = PipelineMLOps(config)

run_id = pipeline_mlops.executar(X, y, feature_names)

if run_id:
    print(f"\n✅ Pipeline concluído com sucesso!")
    print(f"   Run ID: {run_id}")
    print(f"   Modelo: {config.nome_modelo}.pkl")
    print(f"\nPróximos passos:")
    print(f"  1. Revisar no MLflow UI: mlflow ui --port 5001")
    print(f"  2. Promover para produção via Model Registry")
    print(f"  3. Subir API: uvicorn api_modelo:app --reload")
else:
    print("\n❌ Pipeline falhou nos critérios de qualidade.")

Resumo

• MLOps aplica DevOps ao ciclo de vida de modelos ML — rastreamento, versionamento e deploy
• MLflow registra parâmetros, métricas e artefatos de cada experimento automaticamente
• Model Registry gerencia versões e estágios — Staging, Production e Archived
• FastAPI serve modelos como APIs REST com validação Pydantic e documentação automática
• Monitor de drift detecta mudanças na distribuição dos dados ou das predições em produção
• Retreinamento automático mantém o modelo atualizado quando a performance degrada
• Docker Compose orquestra MLflow Server e API de ML em ambiente reproduzível
• Um pipeline MLOps completo cobre: dados → treino → avaliação → registro → deploy → monitoramento

Referências e Leituras Complementares

• MLflow — documentação oficial — https://mlflow.org/docs/latest/index.html
• MLflow Model Registry — https://mlflow.org/docs/latest/model-registry.html
• FastAPI para ML — https://fastapi.tiangolo.com/
• Evidently AI — monitoramento de ML — https://www.evidentlyai.com/
• Weights & Biases — alternativa ao MLflow — https://wandb.ai/
• HUYEN, Chip. Designing Machine Learning Systems. O'Reilly Media, 2022. — referência definitiva sobre MLOps e sistemas de ML em produção.
• GIFT, Noah; BEHRMAN, Alfredo. Practical MLOps. O'Reilly Media, 2021. — guia prático de MLOps com Python, cobrindo MLflow, deploy e monitoramento.

Prof. Ricardo Matos — Dominando o Python em 1 Ano · Artigo 42 de 52

✅ Módulo 7 Concluído — Data Science e Machine Learning

Plano Completo da Série — Estado Atual

Próximo: Módulo 8 · Artigo 43 — Testes com pytest: fixtures, mocks e cobertura

Digite "Sim" para continuar.

you asked

Sim

claude response

Dominando o Python em 1 Ano