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
| Módulo | Tema | Artigos | Status |
|---|---|---|---|
| 1 | Fundamentos da Linguagem | 01–06 | ✅ Concluído |
| 2 | Estruturas de Dados e Algoritmos | 07–12 | ✅ Concluído |
| 3 | Orientação a Objetos | 13–18 | ✅ Concluído |
| 4 | Arquivos, I/O e Banco de Dados | 19–24 | ✅ Concluído |
| 5 | Python para Web | 25–30 | ✅ Concluído |
| 6 | Automação e Scripts | 31–34 | ✅ Concluído |
| 7 | Data Science e Machine Learning | 35–42 | ✅ Concluído |
| 8 | Testes, Qualidade e Boas Práticas | 43–47 | ⬜ Pendente |
| 9 | Projetos Reais e Carreira | 48–52 | ⬜ Pendente |
Próximo: Módulo 8 · Artigo 43 — Testes com pytest: fixtures, mocks e cobertura
Digite "Sim" para continuar.
you asked
Sim