Partilhar via

Converter experimentos de ML em código Python de produção

  • Artigo

APLICA-SE A: Python SDK azureml v1

Neste tutorial, você aprenderá a converter blocos de anotações Jupyter em scripts Python para torná-los testes e automação amigáveis usando o modelo de código MLOpsPython e o Azure Machine Learning. Normalmente, esse processo é usado para pegar o código de experimentação / treinamento de um notebook Jupyter e convertê-lo em scripts Python. Esses scripts podem ser usados para testes e automação de CI/CD em seu ambiente de produção.

Um projeto de aprendizado de máquina requer experimentação onde hipóteses são testadas com ferramentas ágeis como o Jupyter Notebook usando conjuntos de dados reais. Quando o modelo estiver pronto para produção, o código do modelo deve ser colocado em um repositório de código de produção. Em alguns casos, o código do modelo deve ser convertido em scripts Python para ser colocado no repositório de código de produção. Este tutorial aborda uma abordagem recomendada sobre como exportar código de experimentação para scripts Python.

Neste tutorial, irá aprender a:

  • Limpar código não essencial
  • Refatore o código do Jupyter Notebook em funções
  • Criar scripts Python para tarefas relacionadas
  • Criar testes de unidades

Pré-requisitos

Remover todo o código não essencial

Alguns códigos escritos durante a experimentação destinam-se apenas a fins exploratórios. Portanto, a primeira etapa para converter código experimental em código de produção é remover esse código não essencial. Remover o código não essencial também tornará o código mais fácil de manter. Nesta seção, você removerá o código do experimentation/Diabetes Ridge Regression Training.ipynb bloco de anotações. As instruções que imprimem a forma de e y e a chamada features.describe de célula são apenas para exploração de X dados e podem ser removidas. Depois de remover o código não essencial, experimentation/Diabetes Ridge Regression Training.ipynb deve se parecer com o seguinte código sem markdown:

from sklearn.datasets import load_diabetes
from sklearn.linear_model import Ridge
from sklearn.metrics import mean_squared_error
from sklearn.model_selection import train_test_split
import joblib
import pandas as pd

sample_data = load_diabetes()

df = pd.DataFrame(
    data=sample_data.data,
    columns=sample_data.feature_names)
df['Y'] = sample_data.target

X = df.drop('Y', axis=1).values
y = df['Y'].values

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=0)
data = {"train": {"X": X_train, "y": y_train},
        "test": {"X": X_test, "y": y_test}}

args = {
    "alpha": 0.5
}

reg_model = Ridge(**args)
reg_model.fit(data["train"]["X"], data["train"]["y"])

preds = reg_model.predict(data["test"]["X"])
mse = mean_squared_error(preds, y_test)
metrics = {"mse": mse}
print(metrics)

model_name = "sklearn_regression_model.pkl"
joblib.dump(value=reg, filename=model_name)

Refatorar código em funções

Em segundo lugar, o código Jupyter precisa ser refatorado em funções. A refatoração de código em funções facilita o teste de unidade e torna o código mais fácil de manter. Nesta seção, você refatorará:

  • O caderno Diabetes Ridge Regression Training (experimentation/Diabetes Ridge Regression Training.ipynb)
  • O caderno Diabetes Ridge Regression Score(experimentation/Diabetes Ridge Regression Scoring.ipynb)

Refatorar Diabetes Ridge Regression Training notebook em funções

No experimentation/Diabetes Ridge Regression Training.ipynb, conclua as seguintes etapas:

  1. Crie uma função chamada split_data para dividir o quadro de dados em dados de teste e treinamento. A função deve tomar o dataframe df como um parâmetro e retornar um dicionário contendo as chaves train e test.

    Mova o código sob o título Dados divididos em Conjuntos de treinamento e validação para a função e modifique-o split_data para retornar o data objeto.

  2. Crie uma função chamada train_model, que usa os parâmetros data e args retorna um modelo treinado.

    Mova o código sob o título Modelo de treinamento no conjunto de treinamento para a função e modifique-o train_model para retornar o reg_model objeto. Remova o args dicionário, os valores virão do args parâmetro.

  3. Crie uma função chamada get_model_metrics, que usa parâmetros reg_model e data, e avalia o modelo, em seguida, retorna um dicionário de métricas para o modelo treinado.

    Mova o código sob o título Validar modelo no conjunto de validação para a função e modifique-o get_model_metrics para retornar o metrics objeto.

As três funções devem ser as seguintes:

# Split the dataframe into test and train data
def split_data(df):
    X = df.drop('Y', axis=1).values
    y = df['Y'].values

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=0)
    data = {"train": {"X": X_train, "y": y_train},
            "test": {"X": X_test, "y": y_test}}
    return data


# Train the model, return the model
def train_model(data, args):
    reg_model = Ridge(**args)
    reg_model.fit(data["train"]["X"], data["train"]["y"])
    return reg_model


# Evaluate the metrics for the model
def get_model_metrics(reg_model, data):
    preds = reg_model.predict(data["test"]["X"])
    mse = mean_squared_error(preds, data["test"]["y"])
    metrics = {"mse": mse}
    return metrics

Ainda no experimentation/Diabetes Ridge Regression Training.ipynb, conclua as seguintes etapas:

  1. Crie uma nova função chamada main, que não usa parâmetros e não retorna nada.

  2. Mova o código sob o título "Carregar dados" para a main função.

  3. Adicione invocações para as funções recém-escritas na main função:

    # Split Data into Training and Validation Sets
data = split_data(df)

    # Train Model on Training Set
args = {
    "alpha": 0.5
}
reg = train_model(data, args)

    # Validate Model on Validation Set
metrics = get_model_metrics(reg, data)

  4. Mova o código sob o título "Salvar modelo" para a main função.

A main função deve se parecer com o seguinte código:

def main():
    # Load Data
    sample_data = load_diabetes()

    df = pd.DataFrame(
        data=sample_data.data,
        columns=sample_data.feature_names)
    df['Y'] = sample_data.target

    # Split Data into Training and Validation Sets
    data = split_data(df)

    # Train Model on Training Set
    args = {
        "alpha": 0.5
    }
    reg = train_model(data, args)

    # Validate Model on Validation Set
    metrics = get_model_metrics(reg, data)

    # Save Model
    model_name = "sklearn_regression_model.pkl"

    joblib.dump(value=reg, filename=model_name)

Nesta etapa, não deve haver nenhum código restante no bloco de anotações que não esteja em uma função, além de importar instruções na primeira célula.

Adicione uma instrução que chame a main função.

main()

Após a refatoração, experimentation/Diabetes Ridge Regression Training.ipynb deve se parecer com o seguinte código sem a marcação:

from sklearn.datasets import load_diabetes
from sklearn.linear_model import Ridge
from sklearn.metrics import mean_squared_error
from sklearn.model_selection import train_test_split
import pandas as pd
import joblib


# Split the dataframe into test and train data
def split_data(df):
    X = df.drop('Y', axis=1).values
    y = df['Y'].values

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=0)
    data = {"train": {"X": X_train, "y": y_train},
            "test": {"X": X_test, "y": y_test}}
    return data


# Train the model, return the model
def train_model(data, args):
    reg_model = Ridge(**args)
    reg_model.fit(data["train"]["X"], data["train"]["y"])
    return reg_model


# Evaluate the metrics for the model
def get_model_metrics(reg_model, data):
    preds = reg_model.predict(data["test"]["X"])
    mse = mean_squared_error(preds, data["test"]["y"])
    metrics = {"mse": mse}
    return metrics


def main():
    # Load Data
    sample_data = load_diabetes()

    df = pd.DataFrame(
        data=sample_data.data,
        columns=sample_data.feature_names)
    df['Y'] = sample_data.target

    # Split Data into Training and Validation Sets
    data = split_data(df)

    # Train Model on Training Set
    args = {
        "alpha": 0.5
    }
    reg = train_model(data, args)

    # Validate Model on Validation Set
    metrics = get_model_metrics(reg, data)

    # Save Model
    model_name = "sklearn_regression_model.pkl"

    joblib.dump(value=reg, filename=model_name)

main()

Refatorar Diabetes Ridge Regressão Pontuação caderno em funções

No experimentation/Diabetes Ridge Regression Scoring.ipynb, conclua as seguintes etapas:

  1. Crie uma nova função chamada init, que não usa parâmetros e não retorna nada.
  2. Copie o código sob o título "Carregar modelo" para a init função.

A init função deve se parecer com o seguinte código:

def init():
    model_path = Model.get_model_path(
        model_name="sklearn_regression_model.pkl")
    model = joblib.load(model_path)

Uma vez que a init função tenha sido criada, substitua todo o código sob o título "Modelo de carga" por uma única chamada para init o seguinte:

init()

No experimentation/Diabetes Ridge Regression Scoring.ipynb, conclua as seguintes etapas:

  1. Crie uma nova função chamada run, que usa raw_data e request_headers como parâmetros e retorna um dicionário de resultados da seguinte maneira:

    {"result": result.tolist()}

  2. Copie o código sob os títulos "Preparar dados" e "Dados de pontuação" para a run função.

    A run função deve se parecer com o seguinte código (Lembre-se de remover as instruções que definem as variáveis raw_data e request_headers, que serão usadas mais tarde quando a run função for chamada):

    def run(raw_data, request_headers):
    data = json.loads(raw_data)["data"]
    data = numpy.array(data)
    result = model.predict(data)

    return {"result": result.tolist()}

Uma vez criada a run função, substitua todo o código sob os títulos "Preparar dados" e "Dados de pontuação" pelo seguinte código:

raw_data = '{"data":[[1,2,3,4,5,6,7,8,9,10],[10,9,8,7,6,5,4,3,2,1]]}'
request_header = {}
prediction = run(raw_data, request_header)
print("Test result: ", prediction)

O código anterior define variáveis raw_data e request_header, chama a run função com raw_data e request_header, e imprime as previsões.

Após a refatoração, experimentation/Diabetes Ridge Regression Scoring.ipynb deve se parecer com o seguinte código sem a marcação:

import json
import numpy
from azureml.core.model import Model
import joblib

def init():
    model_path = Model.get_model_path(
        model_name="sklearn_regression_model.pkl")
    model = joblib.load(model_path)

def run(raw_data, request_headers):
    data = json.loads(raw_data)["data"]
    data = numpy.array(data)
    result = model.predict(data)

    return {"result": result.tolist()}

init()
test_row = '{"data":[[1,2,3,4,5,6,7,8,9,10],[10,9,8,7,6,5,4,3,2,1]]}'
request_header = {}
prediction = run(test_row, {})
print("Test result: ", prediction)

Em terceiro lugar, as funções relacionadas precisam ser mescladas em arquivos Python para ajudar melhor na reutilização do código. Nesta seção, você criará arquivos Python para os seguintes blocos de anotações:

  • O caderno Diabetes Ridge Regression Training (experimentation/Diabetes Ridge Regression Training.ipynb)
  • O caderno Diabetes Ridge Regression Score(experimentation/Diabetes Ridge Regression Scoring.ipynb)

Criar arquivo Python para o bloco de anotações Diabetes Ridge Regression Training

Converta seu bloco de anotações em um script executável executando a seguinte instrução em um prompt de comando, que usa o nbconvert pacote e o caminho de experimentation/Diabetes Ridge Regression Training.ipynb:

jupyter nbconvert "Diabetes Ridge Regression Training.ipynb" --to script --output train

Depois que o bloco de anotações for convertido em train.py, remova todos os comentários indesejados. Substitua a chamada para main() no final do arquivo por uma invocação condicional como o código a seguir:

if __name__ == '__main__':
    main()

Seu train.py arquivo deve se parecer com o seguinte código:

from sklearn.datasets import load_diabetes
from sklearn.linear_model import Ridge
from sklearn.metrics import mean_squared_error
from sklearn.model_selection import train_test_split
import pandas as pd
import joblib


# Split the dataframe into test and train data
def split_data(df):
    X = df.drop('Y', axis=1).values
    y = df['Y'].values

    X_train, X_test, y_train, y_test = train_test_split(
        X, y, test_size=0.2, random_state=0)
    data = {"train": {"X": X_train, "y": y_train},
            "test": {"X": X_test, "y": y_test}}
    return data


# Train the model, return the model
def train_model(data, args):
    reg_model = Ridge(**args)
    reg_model.fit(data["train"]["X"], data["train"]["y"])
    return reg_model


# Evaluate the metrics for the model
def get_model_metrics(reg_model, data):
    preds = reg_model.predict(data["test"]["X"])
    mse = mean_squared_error(preds, data["test"]["y"])
    metrics = {"mse": mse}
    return metrics


def main():
    # Load Data
    sample_data = load_diabetes()

    df = pd.DataFrame(
        data=sample_data.data,
        columns=sample_data.feature_names)
    df['Y'] = sample_data.target

    # Split Data into Training and Validation Sets
    data = split_data(df)

    # Train Model on Training Set
    args = {
        "alpha": 0.5
    }
    reg = train_model(data, args)

    # Validate Model on Validation Set
    metrics = get_model_metrics(reg, data)

    # Save Model
    model_name = "sklearn_regression_model.pkl"

    joblib.dump(value=reg, filename=model_name)

if __name__ == '__main__':
    main()

train.py agora pode ser invocado a partir de um terminal executando python train.py. As funções de também podem ser chamadas a partir de train.py outros arquivos.

O train_aml.py arquivo encontrado no diabetes_regression/training diretório no repositório MLOpsPython chama as funções definidas no train.py contexto de um trabalho de experimento do Azure Machine Learning. As funções também podem ser chamadas em testes de unidade, abordados mais adiante neste guia.

Criar arquivo Python para o bloco de anotações Diabetes Ridge Regression Scoring

Converta seu bloco de anotações em um script executável executando a seguinte instrução em um prompt de comando que usa o nbconvert pacote e o caminho de experimentation/Diabetes Ridge Regression Scoring.ipynb:

jupyter nbconvert "Diabetes Ridge Regression Scoring.ipynb" --to script --output score

Depois que o bloco de anotações for convertido em score.py, remova todos os comentários indesejados. Seu score.py arquivo deve se parecer com o seguinte código:

import json
import numpy
from azureml.core.model import Model
import joblib

def init():
    model_path = Model.get_model_path(
        model_name="sklearn_regression_model.pkl")
    model = joblib.load(model_path)

def run(raw_data, request_headers):
    data = json.loads(raw_data)["data"]
    data = numpy.array(data)
    result = model.predict(data)

    return {"result": result.tolist()}

init()
test_row = '{"data":[[1,2,3,4,5,6,7,8,9,10],[10,9,8,7,6,5,4,3,2,1]]}'
request_header = {}
prediction = run(test_row, request_header)
print("Test result: ", prediction)

A model variável precisa ser global para que seja visível em todo o script. Adicione a seguinte instrução no início da init função:

global model

Depois de adicionar a instrução anterior, a init função deve se parecer com o seguinte código:

def init():
    global model

    # load the model from file into a global object
    model_path = Model.get_model_path(
        model_name="sklearn_regression_model.pkl")
    model = joblib.load(model_path)

Crie testes de unidade para cada arquivo Python

Em quarto lugar, crie testes de unidade para suas funções Python. Os testes de unidade protegem o código contra regressões funcionais e facilitam a manutenção. Nesta seção, você criará testes de unidade para as funções em train.py.

train.py contém várias funções, mas criaremos apenas um único teste de unidade para a train_model função usando a estrutura Pytest neste tutorial. Pytest não é a única estrutura de teste de unidade Python, mas é uma das mais comumente usadas. Para obter mais informações, visite Pytest.

Um teste de unidade geralmente contém três ações principais:

  • Organizar objeto - criar e configurar objetos necessários
  • Agir sobre um objeto
  • Afirmar o que é esperado

O teste de unidade chamará train_model com alguns dados e argumentos codificados e validará que train_model agiram conforme o esperado usando o modelo treinado resultante para fazer uma previsão e comparando essa previsão com um valor esperado.

import numpy as np
from code.training.train import train_model


def test_train_model():
    # Arrange
    X_train = np.array([1, 2, 3, 4, 5, 6]).reshape(-1, 1)
    y_train = np.array([10, 9, 8, 8, 6, 5])
    data = {"train": {"X": X_train, "y": y_train}}

    # Act
    reg_model = train_model(data, {"alpha": 1.2})

    # Assert
    preds = reg_model.predict([[1], [2]])
    np.testing.assert_almost_equal(preds, [9.93939393939394, 9.03030303030303])

Próximos passos

Agora que você entende como converter de um experimento para código de produção, consulte os links a seguir para obter mais informações e as próximas etapas: