O objetivo aqui é migrar de um código (Jupyter Notebook) para um Pipeline de MLOps profissional, utilizando o MLflow para governança e o Gradio para interface de teste.
Explicação Passo a Passo (Review)MLflow Context (start_run): Em Engenharia de Software, isso funciona como um "Log Premium". Se você treinar o modelo 50 vezes, o MLflow guarda o "DNA" de cada tentativa. Isso evita o erro de "perder" um modelo excelente.Registered Model: Ao registrar o nome Playcatch_v1, criamos um alias. Sua API de recomendação pode sempre buscar a versão marcada como "Production", permitindo que você atualize o modelo por trás sem derrubar o serviço.Gradio Integration: O Gradio não é apenas visual. Para um Engenheiro de IA, ele serve para Sanity Check (Teste de Sanidade). Você pode testar valores extremos (como um score de categoria 999) para ver se o modelo "quebra" ou retorna algo absurdo.Complexidade Big O: A predição em tempo real é O(1) por requisição, o que é ideal para sistemas de alta escala como o da Playcatch. O gargalo geralmente não é o cálculo, mas a latência da rede (I/O Bound).
- Implementação do Pipeline (Código)
Este código simula a estrutura de produção para um sistema de recomendação.
import pandas as pd
import mlflow
import mlflow.sklearn
from sklearn.model_selection import train_test_split
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_squared_error
import gradio as gr
--- ETAPA 1: Pré-processamento e Governança ---
def train_recommendation_model(data_path, n_estimators=100):
# Iniciando o experimento no MLflow para versionamento
mlflow.set_experiment("Playcatch_Recommendation_Service")
with mlflow.start_run():
# Carregamento e Split (Simulado)
df = pd.read_csv(data_path)
X = df.drop(columns=['rating']) # 'rating' é o alvo da recomendação
y = df['rating']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)
# Treinamento
model = RandomForestRegressor(n_estimators=n_estimators)
model.fit(X_train, y_train)
# Avaliação
predictions = model.predict(X_test)
rmse = mean_squared_error(y_test, predictions, squared=False)
# Log de Parâmetros e Métricas (MLOps)
mlflow.log_param("n_estimators", n_estimators)
mlflow.log_metric("rmse", rmse)
# Salvando o modelo no Registry
mlflow.sklearn.log_model(model, "model", registered_model_name="Playcatch_v1")
return model, rmse
--- ETAPA 2: Interface de Teste (Gradio) ---
def create_ui(model):
def predict_interface(user_id, item_id, category_score):
# Transforma o input da UI em DataFrame para o modelo
input_data = pd.DataFrame([[user_id, item_id, category_score]],
columns=['user_id', 'item_id', 'category_score'])
prediction = model.predict(input_data)[0]
return f"Nota de Recomendação Estimada: {prediction:.2f}"
interface = gr.Interface(
fn=predict_interface,
inputs=["number", "number", gr.Slider(0, 10)],
outputs="text",
title="Painel de Avaliação Playcatch IA"
)
return interface
Conte com o apoio da comunidade Alura na sua jornada. Abraços e bons estudos!