Snowpark ML Ops: Modell-Registry¶

Abrufen und Aktualisieren von Kommentaren¶

Verwenden Sie das Attribut comment des Modells, um den Kommentar des Modells abzurufen und zu aktualisieren.

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"

Abrufen und Aktualisieren von Tags¶

Tags sind Metadaten, mit denen der Zweck eines Modells, der Algorithmus, das Trainings-Dataset, die Lebenszyklusphase oder andere von Ihnen gewählte Informationen erfasst werden können. Sie können die Tags beim Registrieren des Modells oder zu einem beliebigen späteren Zeitpunkt setzen. Sie können auch die Werte vorhandener Tags aktualisieren oder Tags ganz entfernen.

Um alle Tags eines Modells in Form eines Python-Dictionarys zu erhalten, verwenden Sie show_tags.

print(m.show_tags())

Um ein neues Tag hinzuzufügen oder den Wert eines vorhandenen Tags zu ändern, verwenden Sie set_tag.

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

Um den Wert eines Tags abzurufen, verwenden Sie get_tag:

m.get_tag("live_version")

Um ein Tag zu entfernen, verwenden Sie unset_tag.

model.unset_tag("live_version")

Löschen von Modellversionen¶

Sie können eine Modellversion mit der delete_version-Methode des Modells löschen.

m.delete_version("rc1")

Standardversion¶

Eine Version eines Modells kann als Standardmodell festgelegt werden. Sie können das default-Attribut des Modells abrufen oder festlegen, um die aktuelle Standardversion (als ModelVersion-Objekt) zu erhalten bzw. zu ändern (unter Verwendung einer Zeichenfolge (String)).

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

Abrufen einer Referenz auf eine Modellversion¶

Um eine Referenz auf eine bestimmte Version eines Modells als ModelVersion-Instanz zu erhalten, verwenden Sie die version-Methode des Modells. Verwenden Sie das default-Attribut des Modells, um die Standardversion des Modells zu abzurufen.

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

Sobald Sie einen Verweis auf eine bestimmte Version eines Modells haben (wie die Variable mv in unserem obigen Beispiel), können Sie dessen Kommentare oder Kennzahlen abrufen oder aktualisieren und die Methoden (Funktionen) des Modells aufrufen, wie in den folgenden Abschnitten beschrieben.

Abrufen und Aktualisieren von Kommentaren¶

Wie bei Modellen können auch Modellversionen Kommentare haben, die über das Attribut comment oder description der Modellversion abgerufen und gesetzt werden können.

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

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

Abrufen und Aktualisieren von Kennzahlen¶

Kennzahlen (Metriken) sind Schlüssel-Wert-Paare, die dazu dienen, die Vorhersagegenauigkeit und andere Eigenschaften der Modellversion zu überwachen. Sie können die Kennzahlen beim Erstellen einer Modellversion oder mit der Methode set_metric festlegen. Der Wert einer Kennzahl kann jedes Python-Objekt sein, das in JSON serialisiert werden kann, einschließlich Zahlen, Zeichenfolgen (Strings), Listen und Dictionarys. Im Gegensatz zu Tags müssen die Namen und möglichen Werte von Kennzahlen nicht im Voraus festgelegt werden.

Eine Kennzahl für die Testgenauigkeit kann mit accuracy_score von sklearn erstellt werden:

from sklearn import metrics

test_accuracy = metrics.accuracy_score(test_labels, prediction)

Die Wahrheitsmatrix (Konfusionsmatrix) kann auf ähnliche Weise mit sklearn generiert werden:

test_confusion_matrix = metrics.confusion_matrix(test_labels, prediction)

Dann können diese Werte wie folgt als Kennzahlen festgelegt werden.

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

Um die Kennzahlen einer Modellversion als Python-Dictionary abzurufen, verwenden Sie show_metrics.

metrics = mv.show_metrics()

Um eine Kennzahl zu löschen, rufen Sie delete_metric auf.

mv.delete_metric("test_accuracy")

Aufrufen von Modellmethoden in SQL¶

Sie können Methoden eines Modells in SQL mit der model_name!method_name(...)-Syntax aufrufen. Die für ein Modell verfügbaren Methoden werden durch die zugrunde liegende Python-Modellklasse bestimmt. Viele Modelle verwenden zum Beispiel eine Methode namens predict für die Inferenz.

Um eine Methode der aktuellen Version eines Modells aufzurufen, verwenden Sie die hier gezeigte Syntax, wobei Sie zwischen den Klammern ggf. verfügbare Argumente an die Methode übergeben sowie in der FROM-Klausel den Namen der Tabelle, die die Inferenzdaten enthält.

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

Um eine Methode einer bestimmten Version eines Modells aufzurufen, erstellen Sie zunächst einen Alias für die spezifische Version des Modells und rufen dann die gewünschte Methode über den Alias auf.

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

Snowpark-ML¶

Die Registry unterstützt Modelle, die mit Snowpark ML-Modellierungs-APIs (von snowpark.ml.modeling.framework.base.BaseEstimator abgeleitete Modelle) erstellt wurden. Die folgenden zusätzlichen Optionen können im options-Dictionary beim Aufruf von log_model verwendet werden.

Beim Protokollieren eines Snowpark ML-Modells müssen Sie sample_input_data oder signatures nicht angeben. Diese Werte werden während der Anpassung automatisch abgeleitet.

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¶

Die Registry unterstützt Modelle, die mit Scikit-learn erstellt wurden (von sklearn.base.BaseEstimator oder sklearn.pipeline.Pipeline abgeleitete Modelle). Die folgenden zusätzlichen Optionen können im options-Dictionary beim Aufruf von log_model verwendet werden.

Sie müssen entweder den Parameter sample_input_data oder signatures angeben, wenn Sie ein Scikit-learn-Modell protokollieren, damit die Registry die Signaturen der Zielmethoden kennt.

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¶

Die Registry unterstützt Modelle, die mit XGBoost erstellt wurden (von xgboost.XGBModel oder xgboost.Booster abgeleitete Modelle). Die folgenden zusätzlichen Optionen können im options-Dictionary beim Aufruf von log_model verwendet werden.

Beim Protokollieren eines XGBoost-Modells müssen Sie entweder den Parameter sample_input_data oder signatures angeben, damit die Registry die Signaturen der Zielmethoden kennt.

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¶

Die Registry unterstützt PyTorch-Modelle (von torch.nn.Module oder torch.jit.ModuleScript abgeleitete Klassen), wenn die forward-Methode des Modells eine oder mehrere torch.Tensor-Instanzen als Eingabe akzeptiert und einen torch.Tensor-Wert oder ein Tupel davon zurückgibt. Die Registry konvertiert zwischen pandas-DataFrames und Tensoren, wenn sie das Modell aufruft und die Ergebnisse zurückgibt. Tensoren entsprechen den Spalten im Datenframe.

Angenommen, Ihr Modell akzeptiert zwei Tensoren wie die folgenden:

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)

Wenn Sie torch.Tensor([[1,2],[3,4]]) als tensor_1 und torch.Tensor([[5,6], [7,8]]) als tensor_2 übergeben möchten, erstellen Sie wie folgt einen DataFrame, den Sie an das Modell übergeben.

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

Das tensors-DataFrame sieht dann wie folgt aus.

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

Ähnlich verhält es sich, wenn Ihr Modell zwei Tensoren zurückgibt, z. B. (torch.Tensor([[1,2],[3,4]]), torch.Tensor([[5,6], [7,8]])): Das Ergebnis ist ein DataFrame wie der obige.

Wenn Sie eine Stichprobe als Eingabedaten für ein PyTorch-Modell bereitstellen, müssen Sie entweder eine Liste von Tensoren (die in einen pandas-DataFrame umgewandelt werden) oder einen DataFrame bereitstellen. Eine Liste kann einen einzelnen Tensor enthalten, aber ein Tensor allein wird nicht akzeptiert.

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¶

Modelle, die tensorflow.Module oder tensorflow.keras.Model erweitern, werden unterstützt, wenn sie Tensoren akzeptieren und zurückgeben und kompilierbar oder bereits kompiliert sind.

• Die __call__-Methode für einen tensorflow.Module-Wert oder die call-Methode für einen tensorflow.keras.Model-Wert akzeptiert einen oder mehrere tensorflow.Tensor-Werte oder tensorflow.Variable-Werte als Eingabe und gibt einen tensorflow.Tensor- oder tensorflow.Variable-Wert oder ein Tupel eines dieser Typen zurück.

• Wenn Ihr Modell Module erweitert, muss es kompilierbar sein, d. h. die __call__-Methode ist mit @tensorflow.function dekoriert. Siehe tf.function-Dokumentation. Wenn es Model erweitert, muss es kompiliert werden. Siehe Dokumentation zum Kompilieren.

Die Registry konvertiert zwischen pandas-DataFrames und Tensoren, wenn sie das Modell aufruft und die Ergebnisse zurückgibt. Tensoren entsprechen den Spalten im Datenframe.

Angenommen, Ihr Modell akzeptiert zwei Tensoren wie die folgenden:

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

Wenn Sie tf.Tensor([[1,2],[3,4]]) als tensor_1 und tf.Tensor([[5,6], [7,8]]) als tensor_2 übergeben möchten, erstellen Sie wie folgt einen DataFrame, den Sie an das Modell übergeben.

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

Das tensors-DataFrame sieht dann wie folgt aus.

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

Ähnlich verhält es sich, wenn Ihr Modell zwei Tensoren zurückgibt, z. B. (tf.Tensor([[1,2],[3,4]]), tf.Tensor([[5,6], [7,8]])): Das Ergebnis ist ein DataFrame wie der obige.

Wenn Sie eine Stichprobe als Eingabedaten für ein TensorFlow-Modell bereitstellen, müssen Sie entweder eine Liste von Tensoren (die in einen pandas-DataFrame umgewandelt werden) oder einen DataFrame bereitstellen. Eine Liste kann einen einzelnen Tensor enthalten, aber ein Tensor allein wird nicht akzeptiert.

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¶

MLFlow-Modelle, die eine PyFunc-Konfiguration bereitstellen, werden unterstützt. Wenn Ihr MLFlow-Modell eine Signatur hat, wird das signature-Argument aus dem Modell abgeleitet. Andernfalls müssen Sie entweder signature oder sample_input_data angeben.

Die folgenden zusätzlichen Optionen können im options-Dictionary beim Aufruf von log_model verwendet werden.

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)

Hugging Face-Pipeline¶

Die Registry unterstützt Hugging Face-Modellklassen, die als Transformer definiert sind, die von transformers.Pipeline abgeleitet sind. Der folgende Code ist ein Beispiel für das Protokollieren eines kompatiblen Modells.

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

Die Registry leitet das signatures-Argument ab, solange die Pipeline nur Aufgaben enthält, die in der folgenden Liste aufgeführt sind.

• 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

Zum Anzeigen der abgeleiteten Signatur können Sie die show_functions-Methode verwenden. Das folgende Beispiel zeigt das Ergebnis von lmv.show_functions(), wobei lmv das oben protokollierte Modell ist.

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

Mit diesen Informationen können Sie das Modell wie folgt aufrufen.

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