Referência: Um Pipeline de ML de Produção

Um pipeline de ML de referência para empresas que usam Python: treinamento em lote com PyTorch 2.6+ ou scikit-learn, registro de artefatos, serviço com FastAPI e monitoramento de desvio e latência - não apenas projetos de ciência em notebooks.

Receita

Cartão de receita de referência rápida - pronto para copiar e colar.

fraud-ml/
  pipelines/train.py       # job de treinamento agendado
  pipelines/features.py    # escrita do data store de features
  serving/app.py           # inferência FastAPI
  models/                  # pesos exportados (S3 em produção)
  eval/metrics.py          # portões offline
  monitoring/drift.py      # verificações de distribuição de produção

Quando usar isso:

• Primeiro modelo indo para produção
• Linha de base do RFC de MLOps
• Lacuna de auditoria entre notebook e serviço

Exemplo de Trabalho

"""pipelines/train.py - treinamento em lote com portão de avaliação."""
from __future__ import annotations
 
import json
from pathlib import Path
 
import joblib
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import roc_auc_score
from sklearn.model_selection import train_test_split
 
from pipelines.features import load_feature_frame
 
 
def train(version: str, min_auc: float = 0.82) -> Path:
    X, y = load_feature_frame()
    X_train, X_test, y_train, y_test = train_test_split(X, y, stratify=y, random_state=42)
    model = GradientBoostingClassifier(random_state=42)
    model.fit(X_train, y_train)
    auc = roc_auc_score(y_test, model.predict_proba(X_test)[:, 1])
    if auc < min_auc:
        raise RuntimeError(f"AUC {auc:.3f} abaixo do portão {min_auc}")
    out = Path(f"models/{version}/model.joblib")
    out.parent.mkdir(parents=True, exist_ok=True)
    joblib.dump(model, out)
    out.with_suffix(".metrics.json").write_text(json.dumps({"auc": auc}))
    return out

"""serving/app.py - API de inferência."""
from __future__ import annotations
 
import joblib
from fastapi import FastAPI
from pydantic import BaseModel, Field
 
app = FastAPI()
MODEL = joblib.load("models/prod/model.joblib")
 
 
class Features(BaseModel):
    amount: float = Field(gt=0)
    merchant_risk_score: float
 
 
class Score(BaseModel):
    fraud_probability: float
    model_version: str = "2026.07.09"
 
 
@app.post("/v1/score", response_model=Score)
def score(body: Features) -> Score:
    prob = float(MODEL.predict_proba([[body.amount, body.merchant_risk_score]])[0, 1])
    return Score(fraud_probability=prob)

O que isso demonstra:

• O job de treinamento falha na CI se o portão de avaliação não for atingido
• O serviço carrega artefato versionado; Pydantic valida as entradas
• Divisão clara entre treinamento em lote e inferência online
• Artefato de métricas viaja com o modelo para auditoria

Mergulho Profundo

Como Funciona

• Pipeline de features - O processamento em lote escreve no data warehouse; o caminho online lê um subconjunto pré-computado ou de baixa latência.
• Registro - S3 ou MLflow armazena model.joblib + metadados JSON.
• Promoção - Shadow ou canary comparam a nova AUC/latência do modelo antes da troca de alias.
• Monitoramento - Registra previsões; compara a distribuição de entrada semanalmente (PSI).
• Reversão - Versão anterior do modelo no registro; sinaliza reversão do alias.

Estágios do Pipeline

Notas de Python

# monitoring/drift.py - verificação simplificada de PSI
def population_stability_index(expected: list[float], actual: list[float]) -> float:
    ...

Armadilhas

• Desvio entre treinamento e serviço - Caminhos de código de features diferentes. Correção: Pacote features.py compartilhado importado em ambos os lados.
• Notebook como produção - Sem testes ou versionamento. Correção: Exportar funções para módulos pipelines/.
• Sem validação de entrada - Lixo corrompe o modelo. Correção: Limites Pydantic na API de serviço.
• Treinamento com GPU, serviço com CPU - incompatibilidade - Problemas de formato de exportação. Correção: Padronizar em ONNX ou joblib por família de modelos.
• Ignorar desvio - Decaimento silencioso da qualidade. Correção: Relatório semanal em lote + limite de alerta.

Alternativas

Perguntas Frequentes

Relacionado

• Referência: Um Sistema de Plataforma de Dados / ETL - upstream de features
• Referência: Um Backend SaaS FastAPI - host de serviço
• Noções Básicas de MLOps - profundidade de operações
• Pipelines de ML Clássicos - padrões sklearn
• Benchmarks de Escalabilidade - testes de carga de inferência

Versões da Stack: Esta página foi escrita para Python 3.14.0 (estável 3.14, manutenção 3.13), FastAPI 0.115+, Django 5.2, Flask 3.1, Pydantic 2, PyTorch 2.6+, pandas 2.2+, Polars 1.x, ruff 0.9+ e uv 0.6+.