Snowpark ML Ops : registre des modèles¶

Récupération et mise à jour des commentaires¶

Utilisez l’attribut comment du modèle pour récupérer et mettre à jour le commentaire du modèle.

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"

Récupération et mise à jour des balises¶

Les balises sont des métadonnées utilisées pour enregistrer l’objectif, l’algorithme, l’ensemble de données d’entraînement, l’étape du cycle de vie d’un modèle ou toute autre information que vous choisissez. Vous pouvez définir des balises lors de l’enregistrement du modèle ou à tout moment par la suite. Vous pouvez également mettre à jour les valeurs des balises existantes ou les supprimer complètement.

Pour obtenir toutes les balises d’un modèle sous la forme d’un dictionnaire Python, utilisez show_tags.

print(m.show_tags())

Pour ajouter une nouvelle balise ou modifier la valeur d’une balise existante, utilisez set_tag.

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

Pour récupérer la valeur d’une balise, utilisez get_tag :

m.get_tag("live_version")

Pour supprimer une balise, utilisez unset_tag.

model.unset_tag("live_version")

Suppression des versions du modèle¶

Vous pouvez supprimer une version de modèle en utilisant la méthode delete_version du modèle.

m.delete_version("rc1")

Version par défaut¶

Une version d’un modèle peut être désignée comme modèle par défaut. Récupérez ou définissez l’attribut default du modèle pour obtenir la version actuelle par défaut (sous la forme d’un objet ModelVersion) ou pour la modifier (à l’aide d’une chaîne).

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

Obtention d’une référence à une version de modèle¶

Pour obtenir une référence à une version spécifique d’un modèle en tant qu’instance ModelVersion , utilisez la méthode version du modèle. Utilisez l’attribut default du modèle pour obtenir la version par défaut du modèle.

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

Une fois que vous avez une référence à une version spécifique d’un modèle (comme la variable mv dans notre exemple ci-dessus), vous pouvez récupérer ou mettre à jour ses commentaires ou ses métriques et appeler les méthodes (fonctions) du modèle comme indiqué dans les sections suivantes.

Récupération et mise à jour des commentaires¶

Comme pour les modèles, les versions de modèle peuvent avoir des commentaires, qui peuvent être consultés et définis via l’attribut comment ou description de la version de modèle.

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

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

Récupération et mise à jour des métriques¶

Les métriques sont des paires clé-valeur utilisées pour suivre la précision de la prédiction et d’autres caractéristiques de la version du modèle. Vous pouvez définir des métriques lors de la création d’une version de modèle ou en utilisant la méthode set_metric. Une valeur métrique peut être n’importe quel objet Python qui peut être sérialisé en JSON, y compris les nombres, les chaînes, les listes et les dictionnaires. Contrairement aux balises, les noms des métriques et les valeurs possibles ne doivent pas être définis à l’avance.

Une mesure de la précision du test peut être générée à l’aide de accuracy_score de sklearn :

from sklearn import metrics

test_accuracy = metrics.accuracy_score(test_labels, prediction)

La matrice de confusion peut être générée de la même manière à l’aide de sklearn :

test_confusion_matrix = metrics.confusion_matrix(test_labels, prediction)

Nous pouvons ensuite définir ces valeurs en tant que métriques comme suit.

# 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)

Pour récupérer les métriques d’une version de modèle sous la forme d’un dictionnaire Python, utilisez show_metrics.

metrics = mv.show_metrics()

Pour supprimer une métrique, appelez delete_metric.

mv.delete_metric("test_accuracy")

Appel des méthodes de modèles en SQL¶

Vous pouvez appeler ou invoquer les méthodes d’un modèle en SQL en utilisant la syntaxe model_name!method_name(...). Les méthodes disponibles sur un modèle sont déterminées par la classe de modèle Python sous-jacente. Par exemple, de nombreux types de modèles utilisent une méthode appelée predict pour l’inférence.

Pour invoquer une méthode de la dernière version d’un modèle, utilisez la syntaxe présentée ici, en transmettant les arguments à la méthode, le cas échéant, entre parenthèses, et en transmettant le nom de la table contenant les données d’inférence dans la clause FROM.

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

Pour invoquer une méthode d’une version spécifique d’un modèle, il faut d’abord créer un alias vers la version spécifique du modèle, puis invoquer la méthode souhaitée par l’intermédiaire de l’alias.

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

ML Snowpark¶

Le registre prend en charge les modèles créés à l’aide des APIs Snowpark ML modeling (modèles dérivés de snowpark.ml.modeling.framework.base.BaseEstimator). Les options supplémentaires suivantes peuvent être utilisées dans le dictionnaire options lors de l’appel à log_model.

Il n’est pas nécessaire de spécifier sample_input_data ou signatures lors de l’enregistrement d’un modèle Snowpark ML ; ces éléments sont automatiquement déduits lors de l’ajustement.

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¶

Le registre prend en charge les modèles créés à l’aide de scikit-learn (modèles dérivés de sklearn.base.BaseEstimator ou sklearn.pipeline.Pipeline). Les options supplémentaires suivantes peuvent être utilisées dans le dictionnaire options lors de l’appel à log_model.

Vous devez spécifier le paramètre sample_input_data ou signatures lorsque vous enregistrez un modèle scikit-learn afin que le registre connaisse les signatures des méthodes cibles.

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¶

Le registre prend en charge les modèles créés à l’aide de XGBoost (modèles dérivés de xgboost.XGBModel ou xgboost.Booster). Les options supplémentaires suivantes peuvent être utilisées dans le dictionnaire options lors de l’appel à log_model.

Vous devez spécifier le paramètre sample_input_data ou signatures lors de l’enregistrement d’un modèle XGBoost afin que le registre connaisse les signatures des méthodes cibles.

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¶

Le registre prend en charge les modèles PyTorch (classes dérivées de torch.nn.Module ou torch.jit.ModuleScript) si la méthode forward du modèle accepte une ou plusieurs instances torch.Tensor en entrée et renvoie un torch.Tensor ou un tuple de celles-ci. Le registre convertit les DataFrames pandas et les tenseurs lors de l’appel du modèle et du renvoi des résultats. Les tenseurs correspondent aux colonnes du dataframe.

Par exemple, supposons que votre modèle accepte deux tenseurs comme suit :

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)

Si vous souhaitez transmettre torch.Tensor([[1,2],[3,4]]) en tant que tensor_1 et torch.Tensor([[5,6], [7,8]]) en tant que tensor_2, créez un DataFrame comme suit pour le transmettre au modèle.

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

Le DataFrame de tensors ressemble alors à ceci.

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

De même, si votre modèle renvoie deux tenseurs, tels que (torch.Tensor([[1,2],[3,4]]), torch.Tensor([[5,6], [7,8]])), le résultat est un DataFrame comme celui ci-dessus.

Lorsque vous fournissez un échantillon de données d’entrée pour un modèle PyTorch, vous devez fournir soit une liste de tenseurs (qui sera convertie en un DataFrame pandas), soit un DataFrame. Une liste peut contenir un seul tenseur, mais un tenseur seul n’est pas accepté.

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¶

Les modèles qui étendent tensorflow.Module ou tensorflow.keras.Model sont pris en charge lorsqu’ils acceptent et renvoient des tenseurs et qu’ils sont compilables ou compilés.

• La méthode __call__ pour un tensorflow.Module ou la méthode call pour un tensorflow.keras.Model accepte un ou plusieurs tensorflow.Tensor ou tensorflow.Variable en entrée et renvoie un tensorflow.Tensor ou tensorflow.Variable ou un tuple de l’un de ces types.

• Si votre modèle étend Module, il doit être compilable, ce qui signifie que la méthode __call__ est décorée par @tensorflow.function ; voir la documentation tf.function. S’il étend Model, il doit être compilé ; voir la documentation de compilation.

Le registre convertit les DataFrames pandas et les tenseurs lors de l’appel du modèle et du renvoi des résultats. Les tenseurs correspondent aux colonnes du dataframe.

Par exemple, supposons que votre modèle accepte deux tenseurs comme suit :

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

Si vous souhaitez transmettre tf.Tensor([[1,2],[3,4]]) en tant que tensor_1 et tf.Tensor([[5,6], [7,8]]) en tant que tensor_2, créez un DataFrame comme suit pour le transmettre au modèle.

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

Le DataFrame de tensors ressemble alors à ceci.

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

De même, si votre modèle renvoie deux tenseurs, tels que (tf.Tensor([[1,2],[3,4]]), tf.Tensor([[5,6], [7,8]])), le résultat est un DataFrame comme celui ci-dessus.

Lorsque vous fournissez un échantillon de données d’entrée pour un modèle TensorFlow, vous devez fournir soit une liste de tenseurs (qui sera convertie en un DataFrame pandas), soit un DataFrame. Une liste peut contenir un seul tenseur, mais un tenseur seul n’est pas accepté.

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¶

Modèles MLFlow qui fournissent une valeur PyFunc sont pris en charge. Si votre modèle MLFlow possède une signature, l’argument signature est déduit du modèle. Sinon, vous devez fournir soit signature soit sample_input_data.

Les options supplémentaires suivantes peuvent être utilisées dans le dictionnaire options lors de l’appel à 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 Hugging Face¶

Le registre prend en charge les classes du modèle Hugging Face définies comme des transformateurs qui dérivent de transformers.Pipeline. Le code suivant est un exemple d’enregistrement d’un modèle compatible.

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')

Le registre infère l’argument signatures tant que le pipeline ne contient que des tâches de la liste suivante.

• 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

Pour voir la signature déduite, utilisez la méthode show_functions. Ce qui suit, par exemple, est le résultat de lmv.show_functions() où lmv est le modèle journalisé ci-dessus.

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

Avec ces informations, vous pouvez appeler le modèle comme suit.

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}]