Snowpark ML Ops: Model Registry¶

Recuperação e atualização de comentários¶

Use o atributo comment do modelo para recuperar e atualizar o comentário do modelo.

print(m.comment)
m.comment = "A better description than the one I provided originally"

print(m.description)
m.description = "A better description than the one I provided originally"

Recuperação e atualização de tags¶

Tags são metadados usados para registrar a finalidade de um modelo, algoritmo, conjunto de dados de treinamento, estágio do ciclo de vida ou outras informações que você escolher. Você pode definir tags quando o modelo for registrado ou a qualquer momento posteriormente. Você também pode atualizar os valores das tags existentes ou remover totalmente as tags.

Para obter todas as tags de um modelo como um dicionário Python, use show_tags.

print(m.show_tags())

Para adicionar uma nova tag ou alterar o valor de uma tag existente, use set_tag.

m.set_tag("live_version", "1")

Para recuperar o valor de uma tag, use get_tag:

m.get_tag("live_version")

Para remover uma tag, use unset_tag.

model.unset_tag("live_version")

Exclusão das versões do modelo¶

Você pode excluir uma versão do modelo usando o método delete_version do modelo.

m.delete_version("rc1")

Versão padrão¶

Uma versão de um modelo pode ser designada como modelo padrão. Recupere ou defina o atributo default do modelo para obter a versão padrão atual (como um objeto ModelVersion) ou para alterá-la (usando uma cadeia de caracteres).

default_version = m.default
m.default = "2"

Como obter uma referência para uma versão do modelo¶

Para obter uma referência a uma versão específica de um modelo como uma instância ModelVersion, use o método version do modelo. Use o atributo default do modelo para obter a versão padrão do modelo.

mv = m.version("1")
mv = m.default

Depois de ter uma referência a uma versão específica de um modelo (como a variável mv em nosso exemplo acima), você pode recuperar ou atualizar seus comentários ou métricas e chamar os métodos (funções) do modelo, conforme mostrado nas seções a seguir.

Recuperação e atualização de comentários¶

Assim como acontece com os modelos, as versões do modelo podem ter comentários, que podem ser acessados e definidos por meio do atributo comment ou description da versão do modelo.

print(mv.comment)
print(mv.description)

mv.comment = "A model version comment"
mv.description = "Same as setting the comment"

Recuperação e atualização de métricas¶

Métricas são pares chave-valor usados para rastrear a precisão da previsão e outras características da versão do modelo. Você pode definir métricas ao criar uma versão do modelo ou defini-las usando o método set_metric. Um valor de métrica pode ser qualquer objeto Python que possa ser serializado para JSON, incluindo números, cadeias de caracteres, listas e dicionários. Ao contrário das tags, os nomes das métricas e os valores possíveis não precisam ser definidos antecipadamente.

Uma métrica de precisão de teste pode ser gerada usando accuracy_score do sklearn:

from sklearn import metrics

test_accuracy = metrics.accuracy_score(test_labels, prediction)

A matriz de confusão pode ser gerada de forma semelhante usando o sklearn:

test_confusion_matrix = metrics.confusion_matrix(test_labels, prediction)

Então podemos definir esses valores como métricas da seguinte maneira.

# scalar metric
mv.set_metric("test_accuracy", test_accuracy)

# hierarchical (dictionary) metric
mv.set_metric("evaluation_info", {"dataset_used": "my_dataset", "accuracy": test_accuracy, "f1_score": f1_score})

# multivalent (matrix) metric
mv.set_metric("confusion_matrix", test_confusion_matrix)

Para recuperar as métricas de uma versão do modelo como um dicionário Python, use show_metrics.

metrics = mv.show_metrics()

Para excluir uma métrica, chame delete_metric.

mv.delete_metric("test_accuracy")

Como chamar métodos de modelo em SQL¶

Você pode chamar ou invocar métodos de um modelo em SQL usando a sintaxe model_name!method_name(...). Os métodos disponíveis em um modelo são determinados pela classe de modelo Python subjacente. Por exemplo, muitos tipos de modelos usam um método denominado predict para inferência.

Para invocar um método da versão mais recente de um modelo, use a sintaxe mostrada aqui, passando argumentos para o método, se houver, entre parênteses, e passando o nome da tabela que contém os dados de inferência na cláusula FROM.

SELECT <model_name>!<method_name>(...) FROM <table_name>;

Para invocar um método de uma versão específica de um modelo, primeiro crie um alias para a versão específica do modelo e, em seguida, invoque o método desejado por meio do alias.

WITH <model_version_alias> AS MODEL <model_name> VERSION <version>
    SELECT <model_version_alias>!<method_name>(...) FROM <table_name>;

ML Snowpark¶

O registro oferece suporte a modelos criados usando as APIs de modelagem do Snowpark ML (modelos derivados de snowpark.ml.modeling.framework.base.BaseEstimator). As seguintes opções adicionais podem ser usadas no dicionário options ao chamar log_model.

Você não precisa especificar sample_input_data ou signatures ao registrar um modelo Snowpark ML; estes são inferidos automaticamente durante o ajuste.

import pandas as pd
import numpy as np
from sklearn import datasets
from snowflake.ml.modeling.xgboost import XGBClassifier

iris = datasets.load_iris()
df = pd.DataFrame(data=np.c_[iris["data"], iris["target"]], columns=iris["feature_names"] + ["target"])
df.columns = [s.replace(" (CM)", "").replace(" ", "") for s in df.columns.str.upper()]

input_cols = ["SEPALLENGTH", "SEPALWIDTH", "PETALLENGTH", "PETALWIDTH"]
label_cols = "TARGET"
output_cols = "PREDICTED_TARGET"

clf_xgb = XGBClassifier(
        input_cols=input_cols, output_cols=output_cols, label_cols=label_cols, drop_input_cols=True
)
clf_xgb.fit(df)
model_ref = registry.log_model(
    clf_xgb,
    model_name="XGBClassifier",
    version_name="v1",
)
model_ref.run(df.drop(columns=label_cols).head(10), function_name='predict_proba')

scikit-learn¶

O registro oferece suporte a modelos criados usando scikit-learn (modelos derivados de sklearn.base.BaseEstimator ou sklearn.pipeline.Pipeline). As seguintes opções adicionais podem ser usadas no dicionário options ao chamar log_model.

Você deve especificar o parâmetro sample_input_data ou signatures ao registrar um modelo scikit-learn para que o registro conheça as assinaturas dos métodos de destino.

from sklearn import datasets, ensemble

iris_X, iris_y = datasets.load_iris(return_X_y=True, as_frame=True)
model = ensemble.RandomForestClassifier(random_state=42)
model.fit(iris_X, iris_y)
model_ref = registry.log_model(
    model,
    model_name="RandomForestClassifier",
    version_name="v1",
    sample_input_data=iris_X,
    options={
        "method_options": {
            "predict": {"case_sensitive": True},
            "predict_proba": {"case_sensitive": True},
            "predict_log_proba": {"case_sensitive": True},
        }
    },
)
model_ref.run(iris_X[-10:], function_name='"predict_proba"')

XGBoost¶

O registro oferece suporte a modelos criados usando XGBoost (modelos derivados de xgboost.XGBModel ou xgboost.Booster). As seguintes opções adicionais podem ser usadas no dicionário options ao chamar log_model.

Você deve especificar o parâmetro sample_input_data ou signatures ao registrar um modelo XGBoost para que o registro conheça as assinaturas dos métodos de destino.

import xgboost
from sklearn import datasets, model_selection

cal_X, cal_y = datasets.load_breast_cancer(as_frame=True, return_X_y=True)
cal_X_train, cal_X_test, cal_y_train, cal_y_test = model_selection.train_test_split(cal_X, cal_y)
params = dict(n_estimators=100, reg_lambda=1, gamma=0, max_depth=3, objective="binary:logistic")
regressor = xgboost.train(params, xgboost.DMatrix(data=cal_X_train, label=cal_y_train))
model_ref = registry.log_model(
    regressor,
    model_name="xgBooster",
    version_name="v1",
    sample_input_data=cal_X_test,
    options={
        "target_methods": ["predict"],
        "method_options": {
            "predict": {"case_sensitive": True},
        },
    },
)
model_ref.run(cal_X_test[-10:])

PyTorch¶

O registro oferece suporte aos modelos PyTorch (classes derivadas de torch.nn.Module ou torch.jit.ModuleScript) se o método forward do modelo aceitar uma ou mais instâncias torch.Tensor como entrada e retornar um torch.Tensor ou uma tupla delas. O registro converte entre pandas DataFrames e tensores ao chamar o modelo e retornar os resultados. Os tensores correspondem às colunas no dataframe.

Por exemplo, suponha que seu modelo aceite dois tensores como este:

import torch

class TorchModel(torch.nn.Module):
    def __init__(self, n_input: int, n_hidden: int, n_out: int, dtype: torch.dtype = torch.float32) -> None:
        super().__init__()
        self.model = torch.nn.Sequential(
            torch.nn.Linear(n_input, n_hidden, dtype=dtype),
            torch.nn.ReLU(),
            torch.nn.Linear(n_hidden, n_out, dtype=dtype),
            torch.nn.Sigmoid(),
        )

    def forward(self, tensor_1: torch.Tensor, tensor_2: torch.Tensor) -> torch.Tensor:
        return self.model(tensor_1) + self.model(tensor_2)

Se você quiser passar torch.Tensor([[1,2],[3,4]]) como tensor_1 e torch.Tensor([[5,6], [7,8]]) como tensor_2, crie um DataFrame como segue para passar para o modelo.

import pandas as pd
tensors = pd.DataFrame([[[1,2],[5,6]],[[3,4],[7,8]]])

Então o tensors DataFrame fica assim.

        0       1
0  [1, 2]  [5, 6]
1  [3, 4]  [7, 8]

Da mesma forma, se seu modelo retornar dois tensores, como (torch.Tensor([[1,2],[3,4]]), torch.Tensor([[5,6], [7,8]])), o resultado será um DataFrame como o acima.

Ao fornecer dados de entrada de amostra para um modelo PyTorch, você deve fornecer uma lista de tensores (que serão convertidos em pandas DataFrame) ou em DataFrame. Uma lista pode conter um único tensor, mas um tensor por si só não é aceito.

import torch
import numpy as np

class TorchModel(torch.nn.Module):
        def __init__(self, n_input: int, n_hidden: int, n_out: int, dtype: torch.dtype = torch.float32) -> None:
                super().__init__()
                self.model = torch.nn.Sequential(
                        torch.nn.Linear(n_input, n_hidden, dtype=dtype),
                        torch.nn.ReLU(),
                        torch.nn.Linear(n_hidden, n_out, dtype=dtype),
                        torch.nn.Sigmoid(),
                )

        def forward(self, tensor: torch.Tensor) -> torch.Tensor:
                return self.model(tensor)

n_input, n_hidden, n_out, batch_size, learning_rate = 10, 15, 1, 100, 0.01
dtype = torch.float32
x = np.random.rand(batch_size, n_input)
data_x = torch.from_numpy(x).to(dtype=dtype)
data_y = (torch.rand(size=(batch_size, 1)) < 0.5).to(dtype=dtype)

model = TorchModel(n_input, n_hidden, n_out, dtype=dtype)
loss_function = torch.nn.MSELoss()
optimizer = torch.optim.SGD(model.parameters(), lr=learning_rate)
for _epoch in range(100):
        pred_y = model.forward(data_x)
        loss = loss_function(pred_y, data_y)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()


model_ref = registry.log_model(
    model,
    model_name="torchModel",
    version_name="v1",
    sample_input_data=[data_x],
)
model_ref.run([data_x])

TensorFlow¶

Modelos que estendem tensorflow.Module ou tensorflow.keras.Model são suportados quando aceitam e retornam tensores e são compiláveis ou compilados.

• O método __call__ para um tensorflow.Module ou o método call para um tensorflow.keras.Model aceita um ou mais tensorflow.Tensor ou tensorflow.Variable como entrada e retorna um tensorflow.Tensor ou tensorflow.Variable ou uma tupla de um dos esses tipos.

• Se seu modelo estender Module, ele deverá ser compilável, o que significa que o método __call__ é decorado com @tensorflow.function; consulte a documentação do tf.function. Se estender Model, deverá ser compilado; consulte a documentação de compilação.

O registro converte entre pandas DataFrames e tensores ao chamar o modelo e retornar os resultados. Os tensores correspondem às colunas no dataframe.

Por exemplo, suponha que seu modelo aceite dois tensores como este:

import tensorflow as tf

class KerasModel(tf.keras.Model):
    def  __init__(self, n_hidden: int, n_out: int) -> None:
        super().__init__()
        self.fc_1 = tf.keras.layers.Dense(n_hidden, activation="relu")
        self.fc_2 = tf.keras.layers.Dense(n_out, activation="sigmoid")

    def call(self, tensor_1: tf.Tensor, tensor_2: tf.Tensor) -> tf.Tensor:
        input = tensor_1 + tensor_2
        x = self.fc_1(input)
        x = self.fc_2(x)
        return x

Se você quiser passar tf.Tensor([[1,2],[3,4]]) como tensor_1 e tf.Tensor([[5,6], [7,8]]) como tensor_2, crie um DataFrame como segue para passar para o modelo.

import pandas as pd
tensors = pd.DataFrame([[[1,2],[5,6]],[[3,4],[7,8]]])

Então o tensors DataFrame fica assim.

        0       1
0  [1, 2]  [5, 6]
1  [3, 4]  [7, 8]

Da mesma forma, se seu modelo retornar dois tensores, como (tf.Tensor([[1,2],[3,4]]), tf.Tensor([[5,6], [7,8]])), o resultado será um DataFrame como o acima.

Ao fornecer dados de entrada de amostra para um modelo TensorFlow, você deve fornecer uma lista de tensores (que serão convertidos em pandas DataFrame) ou em DataFrame. Uma lista pode conter um único tensor, mas um tensor por si só não é aceito.

import tensorflow as tf
import numpy as np

class KerasModel(tf.keras.Model):
        def __init__(self, n_hidden: int, n_out: int) -> None:
                super().__init__()
                self.fc_1 = tf.keras.layers.Dense(n_hidden, activation="relu")
                self.fc_2 = tf.keras.layers.Dense(n_out, activation="sigmoid")

        def call(self, tensor: tf.Tensor) -> tf.Tensor:
                input = tensor
                x = self.fc_1(input)
                x = self.fc_2(x)
                return x

n_input, n_hidden, n_out, batch_size, learning_rate = 10, 15, 1, 100, 0.01
dtype = tf.float32
x = np.random.rand(batch_size, n_input)
data_x = tf.convert_to_tensor(x, dtype=dtype)
raw_data_y = tf.random.uniform((batch_size, 1))
raw_data_y = tf.where(raw_data_y > 0.5, tf.ones_like(raw_data_y), tf.zeros_like(raw_data_y))
data_y = tf.cast(raw_data_y, dtype=dtype)

model = KerasModel(n_hidden, n_out)
model.compile(optimizer=tf.keras.optimizers.SGD(learning_rate=learning_rate), loss=tf.keras.losses.MeanSquaredError())
model.fit(data_x, data_y, batch_size=batch_size, epochs=100)

model_ref = registry.log_model(
    model,
    model_name="tfModel",
    version_name="v1",
    sample_input_data=[data_x],
)
model_ref.run([data_x])

MLFlow¶

Modelos MLFlow que fornecem um tipo PyFunc são suportados. Se o seu modelo MLFlow tiver uma assinatura, o argumento signature será inferido do modelo. Caso contrário, você deverá fornecer signature ou sample_input_data.

As seguintes opções adicionais podem ser usadas no dicionário options ao chamar log_model.

import mlflow
from sklearn import datasets, model_selection, ensemble

db = datasets.load_diabetes(as_frame=True)
X_train, X_test, y_train, y_test = model_selection.train_test_split(db.data, db.target)
with mlflow.start_run() as run:
    rf = ensemble.RandomForestRegressor(n_estimators=100, max_depth=6, max_features=3)
    rf.fit(X_train, y_train)

    # Use the model to make predictions on the test dataset.
    predictions = rf.predict(X_test)
    signature = mlflow.models.signature.infer_signature(X_test, predictions)
    mlflow.sklearn.log_model(
        rf,
        "model",
        signature=signature,
    )
    run_id = run.info.run_id


model_ref = registry.log_model(
    mlflow.pyfunc.load_model(f"runs:/{run_id}/model"),
    model_name="mlflowModel",
    version_name="v1",
    conda_dependencies=["mlflow<=2.4.0", "scikit-learn", "scipy"],
    options={"ignore_mlflow_dependencies": True}
)
model_ref.run(X_test)

Pipeline do Hugging Face¶

O registro oferece suporte a classes de modelo Hugging Face definidas como transformadores que derivam de transformers.Pipeline. O código a seguir é um exemplo de registro de um modelo compatível.

lm_hf_model = transformers.pipeline(
    task="text-generation",
    model="bigscience/bloom-560m",
    token="...",  # Put your HuggingFace token here.
    return_full_text=False,
    max_new_tokens=100,
)

lmv = reg.log_model(lm_hf_model, model_name='bloom', version_name='v560m')

O registro infere o argumento signatures desde que o pipeline contenha apenas tarefas da lista a seguir.

• conversational

• fill-mask

• question-answering

• summarization

• table-question-answering

• text2text-generation

• text-classification (alias sentiment-analysis)

• text-generation

• token-classification (alias ner)

• translation

• translation_xx_to_yy

• zero-shot-classification

Para ver a assinatura inferida, use o método show_functions. O seguinte, por exemplo, é o resultado de lmv.show_functions() onde lmv é o modelo registrado acima.

{'name': '__CALL__',
  'target_method': '__call__',
  'signature': ModelSignature(
                      inputs=[
                          FeatureSpec(dtype=DataType.STRING, name='inputs')
                      ],
                      outputs=[
                          FeatureSpec(dtype=DataType.STRING, name='outputs')
                      ]
                  )}]

Com essas informações, você pode chamar o modelo da seguinte forma.

import pandas as pd
remote_prediction = lmv.run(pd.DataFrame(["Hello, how are you?"], columns=["inputs"]))

# Prepare model
import transformers
import pandas as pd

finbert_model = transformers.pipeline(
    task="text-classification",
    model="ProsusAI/finbert",
    top_k=2,
)

# Log the model
mv = registry.log_model(
    finbert_model,
    model_name="finbert",
    version_name="v1",
)

# Use the model
mv.run(pd.DataFrame(
        [
            ["I have a problem with my Snowflake that needs to be resolved asap!!", ""],
            ["I would like to have udon for today's dinner.", ""],
        ]
    )
)

0  [{"label": "negative", "score": 0.8106237053871155}, {"label": "neutral", "score": 0.16587384045124054}]
1  [{"label": "neutral", "score": 0.9263970851898193}, {"label": "positive", "score": 0.05286872014403343}]