MLOps: Versionamento de Modelos, MLflow e Deploy de ML

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 scipy

Os exemplos foram conferidos com MLflow 3.16, scikit-learn 1.9 e FastAPI 0.142. O MLflow muda depressa, e três mudanças recentes aparecem neste artigo: os estágios do registro (Staging, Production) estão obsoletos desde a 2.9 e deram lugar aos aliases; o artifact_path do log_model virou name na série 3; e os modelos do scikit-learn passaram a ser gravados no formato skops, mais seguro que o pickle, que pede uma declaração explícita dos tipos confiáveis para modelos de árvore.

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 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):

        # O que se registra é o que se treina: sem o set_params, o log
        # diria max_depth=10 para uma floresta treinada sem limite
        modelo.set_params(**params)
        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,
            name="modelo_sklearn",
            registered_model_name="aprovacao",   # um modelo, uma versão por run
            # formato skops: modelos de árvore exigem declarar o tipo confiável
            skops_trusted_types=["sklearn.tree._tree.Tree"],
            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))

Duas linhas desse bloco corrigem defeitos que a primeira versão do artigo tinha. Sem o skops_trusted_types, o MLflow 3.16 grava a regressão logística e recusa a primeira floresta com MlflowException: The saved sklearn model references untrusted types: o tipo sklearn.tree._tree.Tree guarda índices de nós sem verificação de limites, e um arquivo adulterado pode derrubar o processo. Declarar o tipo é seguro quando o modelo foi treinado por você. E sem o set_params, o "Random Forest" registrava max_depth=10 e min_samples_split=5 enquanto treinava com os padrões — árvores de até 18 níveis —, e o "Gradient Boosting" registrava subsample=0.8 treinando com 1.0. Um rastreamento que mente sobre os parâmetros é pior que nenhum. Medido: o melhor modelo é a regressão logística (F1 0,8746, AUC 0,9760), o que faz sentido, porque os dados de exemplo foram gerados por uma regra linear.

Model Registry: Promovendo Modelos

from mlflow import MlflowClient


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

    # Listando as versões registradas
    print(f"=== Versões de {nome_modelo} ===")
    for v in client.search_model_versions(f"name='{nome_modelo}'"):
        print(f"  v{v.version} — 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"]
    melhor_nome   = runs.iloc[0]["tags.mlflow.runName"]
    print(f"\nMelhor run: {melhor_nome} {melhor_run_id[:8]} (AUC={melhor_auc:.4f})")

    # A versão registrada que saiu desse run
    versao = client.search_model_versions(
        f"name='{nome_modelo}' and run_id='{melhor_run_id}'"
    )[0].version

    # Promovendo: o alias "champion" passa a apontar para esta versão
    client.set_registered_model_alias(nome_modelo, "champion", versao)
    print(f"{nome_modelo} v{versao} ({melhor_nome}) agora é @champion")

    client.update_model_version(
        name=nome_modelo,
        version=versao,
        description=f"{melhor_nome}, AUC={melhor_auc:.4f} no teste"
    )
    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, alias: str = "champion"):
    """Carrega a versão para a qual o alias aponta."""
    uri    = f"models:/{nome}@{alias}"
    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 contextlib import asynccontextmanager
from datetime import datetime, timezone
import logging

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

# De onde vem o modelo: variáveis de ambiente, para o mesmo código servir
# no notebook e no contêiner
TRACKING_URI = os.getenv("MLFLOW_TRACKING_URI", "sqlite:///mlflow.db")
MODEL_URI    = os.getenv("MODEL_URI", "models:/aprovacao@champion")

modelo = None
origem_modelo = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Carrega o modelo antes de a API aceitar requisições."""
    global modelo, origem_modelo
    try:
        mlflow.set_tracking_uri(TRACKING_URI)
        modelo = mlflow.sklearn.load_model(MODEL_URI)
        origem_modelo = MODEL_URI
        logger.info(f"Modelo carregado: {MODEL_URI}")
    except Exception as e:
        logger.warning(f"Modelo MLflow não disponível: {e}")
        # Fallback — modelo local; o /health diz que é ele
        if os.path.exists("pipeline_temp.pkl"):
            import joblib
            modelo = joblib.load("pipeline_temp.pkl")
            origem_modelo = "pipeline_temp.pkl (fallback)"
            logger.warning("Servindo o modelo local de fallback.")
    yield


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


# 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]


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,
        "origem":  origem_modelo,
        "timestamp": datetime.now(timezone.utc).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.now(timezone.utc).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.now(timezone.utc).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

O modelo é carregado no lifespan, que substituiu o @app.on_event("startup") — obsoleto no FastAPI — e roda antes de a API aceitar a primeira requisição. O /health passou a dizer de onde o modelo veio, e isso não é detalhe. Na primeira versão, quando o registro falhava, a API carregava em silêncio o pipeline_temp.pkl e respondia "status": "healthy". Só que esse arquivo é sobrescrito a cada experimento do laço de treino: no fim, ele guarda o último modelo treinado, o "Random Forest Tuned", e não o escolhido. Um fallback é útil, mas precisa aparecer no monitoramento. O datetime.utcnow() também saiu: está obsoleto desde o Python 3.12 e devolve uma data sem fuso; datetime.now(timezone.utc) devolve a mesma hora com o fuso declarado.

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,
            name="modelo_retreinado",
            registered_model_name="aprovacao",
            skops_trusted_types=["sklearn.tree._tree.Tree"],
        )
        # A nova versão NÃO vira @champion sozinha: precisa passar pela
        # mesma comparação que as outras antes de o alias ser movido

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

    return True


# Simulando dados novos com leve degradação
# Dados novos: outra amostra do mesmo processo, com ruído de medição.
# Nunca derive os "dados novos" do conjunto de teste: depois do retreino,
# ele deixaria de ser uma avaliação independente.
rng = np.random.default_rng(7)
X_novos_sim = np.column_stack([
    rng.uniform(1, 10, 400), rng.integers(0, 20, 400),
    rng.uniform(3, 10, 400), rng.uniform(0, 5, 400),
])
y_novos_sim = ((0.5*X_novos_sim[:, 0] + 0.3*X_novos_sim[:, 2]
                - 0.1*X_novos_sim[:, 1] + 0.2*X_novos_sim[:, 3]
                + rng.normal(0, 0.5, 400)) >= 5).astype(int)
X_novos_sim = X_novos_sim + rng.normal(0, 1, X_novos_sim.shape)

# threshold alto de propósito, para forçar o retreino na demonstração
retreinar_se_necessario(modelo_prod, X_novos_sim, y_novos_sim, threshold_f1=0.99)

docker-compose para MLflow + API

# docker-compose.mlops.yml
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
               --artifacts-destination /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_URI=models:/aprovacao@champion
    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"]

Três ajustes no compose, cada um com motivo. A linha version: saiu porque o Docker Compose atual a ignora e avisa que é obsoleta. As variáveis passadas para a API agora são as que ela lê: a versão anterior definia MODEL_NAME e MODEL_STAGE, que o código nunca consultava, e fixava o sqlite:///mlflow.db por cima do MLFLOW_TRACKING_URI — dentro do contêiner, a API não achava o servidor e caía no fallback. E o servidor usa --artifacts-destination em vez de --default-artifact-root com caminho local. Medido com dois servidores locais: com --default-artifact-root, o cliente recebe como destino um caminho de disco, e num compose seria o disco do contêiner que está gravando, não o do servidor; com --artifacts-destination, recebe mlflow-artifacts:/…, e os arquivos trafegam pelo próprio servidor MLflow via HTTP.

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,
                name="modelo",
                registered_model_name=self.config.nome_modelo,
                skops_trusted_types=["sklearn.tree._tree.Tree"],
                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: {self.config.nome_modelo} "
                  f"v{model_info.registered_model_version}")

        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.")

MLOps é o que transforma um modelo que funcionou no notebook num serviço em que se pode confiar. O MLflow cobre o registro: cada run guarda parâmetros, métricas e o modelo serializado; o registro de modelos agrupa as versões sob um nome, e um alias como @champion diz qual delas está valendo. A partir daí, o modelo vira API — aqui com o FastAPI de FastAPI: APIs modernas com tipagem e documentação automática —, o monitoramento compara a distribuição dos dados novos com a do treino, e o retreino fecha o ciclo quando o desempenho cai. Os modelos em si são os de Machine Learning com scikit-learn; o que muda é tudo o que acontece em volta deles.

Quase todos os defeitos deste artigo tinham a mesma natureza: o sistema dizia uma coisa e fazia outra. O log registrava parâmetros que o modelo não usava, a promoção escolhia um modelo por nome fixo e anotava nele o AUC de outro, a API se declarava saudável servindo o arquivo errado, e o compose passava variáveis que ninguém lia. Em MLOps, o registro existe para ser acreditado, então vale tratá-lo como código de produção: conferir que o que foi gravado é o que foi treinado, que o que está servindo é o que foi escolhido, e que cada fallback aparece no monitoramento.

Fontes e leituras recomendadas

• 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.

Exercícios

Exercício 1

Na interface do MLflow, o run "Random Forest" mostra max_depth = 10 e min_samples_split = 5. Um colega tenta reproduzir o resultado treinando RandomForestClassifier(max_depth=10, min_samples_split=5, random_state=42) e obtém F1 de 0,8671, contra 0,8622 no run registrado. O código de treino da primeira versão do artigo criava o modelo com RandomForestClassifier(random_state=42) e passava os parâmetros só para mlflow.log_params. Qual dos dois números corresponde ao que foi registrado?

Exercício 2

Depois de atualizar o MLflow para a série 3, o laço de experimentos grava a regressão logística normalmente e falha no segundo modelo, a floresta aleatória, com MlflowException: The saved sklearn model references untrusted types, citando sklearn.tree._tree.Tree. O que o MLflow está protegendo, e qual é a correção adequada?

Exercício 3

A função de registro da primeira versão buscava o melhor run pelo AUC, imprimia "Melhor run: … (AUC=0.9760)" e promovia o modelo aprovacao_random_forest_tuned, escrevendo na descrição dele "AUC=0.9760 — Aprovado para staging". Medido, o run de maior AUC é a regressão logística, e o AUC do Random Forest Tuned é 0,9715. O que está errado, e como o registro por aliases resolve?

Exercício 4

A API de previsão sobe num servidor novo, e o /health responde {"status": "healthy", "modelo": true}. Dias depois, alguém percebe que as probabilidades não batem com as do modelo escolhido. O log de inicialização tem a linha "Modelo MLflow não disponível: Registered Model with name=aprovacao_random_forest_tuned not found", seguida de "Modelo local carregado como fallback". Que modelo estava servindo?

Exercício 5

Na primeira versão do retreinamento, os "dados novos" eram X_test + np.random.normal(0, 2, X_test.shape), com y_test como rótulo. O modelo retreinado, com treino e dados novos juntos, era avaliado no X_test original. Medido, o F1 foi de 0,8702, só com o treino, para 0,8714. O ganho é pequeno. Então o procedimento está certo?