coremlflow 0.0.1

Coreflow - Abstração de Funções para MLFlow não oficial.

Guia de Desenvolvimento: Modelos e Inferência no coremlflow

Este guia explica como criar scripts de Treinamento (train.py) e Predição/Inferência (predict.py) utilizando o framework base da biblioteca coremlflow.

A arquitetura orientada a objetos obriga a implementação de métodos específicos para padronizar o ciclo de vida do modelo e a integração automática com o MLflow.

---

🚀 1. Criando um Modelo de Treinamento (train.py)

Com o coremlflow, você tem a opção de herdar da classe base genérica ou utilizar as abstrações específicas construídas para os principais frameworks de Data Science.

A. Utilizando as Classes Especializadas (Recomendado)

O pacote coremlflow.models provê classes preparadas que já implementam automaticamente a melhor forma de salvar o seu modelo (ex: usando mlflow.sklearn ou mlflow.prophet).

Exemplo: Random Forest

Para Random Forest ou Isolation Forest, herde das classes em coremlflow.models.sklearn:

import pandas as pd
from typing import Dict, Any
from sklearn.ensemble import RandomForestClassifier
from coremlflow.models.sklearn import RandomForestMLFlowModel

class MeuModeloRF(RandomForestMLFlowModel):
    def __init__(self, experiment_name: str, data_path: str, tracking_uri: str = None):
        super().__init__(experiment_name, tracking_uri)
        self.data_path = data_path 
    
    def load_data(self) -> Dict[str, Any]:
        # Exemplo carregando dados de Treino/Teste
        return {"X_train": [...], "y_train": [...], "X_test": [...], "y_test": [...]}

    def train(self, data: Dict[str, Any], params: Dict[str, Any]) -> Any:
        model = RandomForestClassifier(**params)
        model.fit(data["X_train"], data["y_train"])
        return model

    def evaluate(self, model: Any, data: Dict[str, Any]) -> Dict[str, float]:
        # Calcule e retorne suas métricas
        score = model.score(data["X_test"], data["y_test"])
        return {"accuracy": score}

    def get_signature_data(self, model: Any, data: Dict[str, Any]) -> tuple:
        return (data["X_train"], data["y_train"])

Note que não foi necessário implementar o método save_model(), pois ele já foi abstraído pela classe mãe RandomForestMLFlowModel.

Outros modelos suportados nativamente:

• IsolationForestMLFlowModel (em coremlflow.models.sklearn)
• ProphetMLFlowModel (em coremlflow.models.prophet)

B. Utilizando a Classe Base Genérica

Se você estiver treinando um modelo Keras, PyTorch, ou outro ainda não mapeado nas classes filhas nativas, basta herdar nativamente da MLFlowModelBase e definir seu .save_model().

from coremlflow.base import MLFlowModelBase
import mlflow.keras

class MeuModeloGenerico(MLFlowModelBase):
    # ... defina load(), train(), evaluate(), get_signature_data() ...
    
    def save_model(self, model, signature) -> None:
        # [Obrigatório na Base]
        mlflow.keras.log_model(model, artifact_path="model", signature=signature)

Executando o Treinamento

Para rodar, instancie a sua classe e chame o método genérico run(params). Ele vai orquestrar todas as chamadas garantindo o log correto no MLflow.

if __name__ == "__main__":
    predictor = MeuModeloRF(experiment_name="Projeto_Churn", data_path="dados.csv")
    params = {"n_estimators": 100, "max_depth": 5}
    
    modelo = predictor.run(params=params)

---

🔮 2. Consumindo um Modelo em Produção (predict.py)

Ao fazer a inferência, importe o MLFlowPredictorBase.

Essa classe cuida internamente do carregamento do modelo a partir do Registry do MLflow. Se o Tracking Server do MLflow estiver fora do ar, você ainda pode fornecer um local_model_path como fallback.

Estrutura da Classe de Predição

Opcionalmente, você pode sobrescrever o método predict, mas por base, a classe já traz o comportamento chamando recursivamente o modelo baixado no formato pyfunc.

from coremlflow.predictor import MLFlowPredictorBase

# Você pode usar a classe base diretamente:
servico_ia = MLFlowPredictorBase(
    internal_alias="CHURN_PROD",
    internal_alias_map_path="./mlflow_aliases.json",
    local_model_path="./modelos_backup/rf_model_v1" # Fallback opcional offline
)

# Os resultados dependerão da natureza do modelo treinado.
resultados = servico_ia.predict(dados_de_entrada)

Ou, caso você precise criar um fluxo complexo de tratamento de dados de entrada antes / depois de passar pelo .predict():

class SeuPredictorService(MLFlowPredictorBase):
    def predict(self, input_data):
        # 1. Trata input_data (ex: limpeza, feature engineering)
        dados_tratados = limpar(input_data)
        
        # 2. Chama a predição da classe base
        resultado = super().predict(dados_tratados)
        
        # 3. Pós Processa
        return formatar(resultado)

Exemplo de arquivo ./mlflow_aliases.json (apelido interno -> run_id):

{
  "CHURN_PROD": "a1b2c3d4e5f6g7h8i9j0",
  "FORECAST_PROD": { "run_id": "ffffeeee111122223333444455556666", "artifact_path": "model" }
}