Best Practices für Modellartefakte

Verwenden Sie diese zusätzlichen Überlegungen für die Erstellung und das Packaging von Modellartefakten.

score.py-Datei schreiben

Stellen Sie immer sicher, dass sich die Dateien score.py und runtime.yaml in der obersten Verzeichnisebene eines Modellartefakts befinden.

Alle anderen Dateien, die Teil eines Artefakts sein sollen, müssen sich auf derselben Ebene wie die beiden Dateien oder in Verzeichnissen darunter befinden:
```
.
|-- runtime.yaml
|-- score.py
|-- <your-serialized-models>
```
Das Modell-Deployment verwendet die score.py-Funktionen, um ein Modell in den Arbeitsspeicher zu laden und Prognosen zu erstellen.
Die Funktionsdefinitionen load_model() und predict() können nicht bearbeitet werden. Nur der Body dieser Funktionen ist anpassbar.
Der zulässige Pfad zum Schreiben von Daten auf den Datenträger, wenn der Modell-Deployment-Service verwendet wird, lautet /home/datascience.
Sie können mit Resource Principals auf Object Storage zugreifen, wenn die OCI-Identitätsberechtigungen korrekt dafür definiert sind.

Benutzerdefinierte Module verpacken

Alle benutzerdefinierten Module, von denen score.py oder das serialisierte Modell abhängig sind, müssen als separate Python-Skripte in demselben Verzeichnis auf oberster Ebene wie score.py oder unter geschrieben werden. Beispiel: model.joblib hängt von einer benutzerdefinierten DataFrameLabelEncoder-Klasse ab, die im Skript dataframelabelencoder.py wie in diesem Beispiel definiert ist:

from category_encoders.ordinal import OrdinalEncoder
from collections import defaultdict
 
from sklearn.base import TransformerMixin
from sklearn.ensemble import RandomForestClassifier
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import LabelEncoder
 
class DataFrameLabelEncoder(TransformerMixin):
    def __init__(self):
        self.label_encoders = defaultdict(LabelEncoder)
         
    def fit(self, X):
        for column in X.columns:
            if X[column].dtype.name  in ["object", "category"]:
                self.label_encoders[column] = OrdinalEncoder()
                self.label_encoders[column].fit(X[column])
        return self
     
    def transform(self, X):
        for column, label_encoder in self.label_encoders.items():
            X[column] = label_encoder.transform(X[column])
        return X

Das Modul wird dann von der Datei score.py importiert:

"""
   Inference script. This script is used for prediction by scoring server when schema is known.
"""
 
import json
import os
from joblib import load
import io
import pandas as pd
 
from dataframelabelencoder import DataFrameLabelEncoder
 
def load_model():
    """
    Loads model from the serialized format
 
    Returns
    -------
    model:  a model instance on which predict API can be invoked
    """
    model_dir = os.path.dirname(os.path.realpath(__file__))
    contents = os.listdir(model_dir)
    model_file_name = "model.joblib"
    # TODO: Load the model from the model_dir using the appropriate loader
    # Below is a sample code to load a model file using `cloudpickle` which was serialized using `cloudpickle`
    # from cloudpickle import cloudpickle
    if model_file_name in contents:
        with open(os.path.join(os.path.dirname(os.path.realpath(__file__)), model_file_name), "rb") as file:
            model = load(file) # Use the loader corresponding to your model file.
    else:
        raise Exception('{0} is not found in model directory {1}'.format(model_file_name, model_dir))
     
    return model
 
 
def predict(data, model=load_model()) -> dict:
    """
    Returns prediction given the model and data to predict
 
    Parameters
    ----------
    model: Model instance returned by load_model API
    data: Data format as expected by the predict API of the core estimator. For eg. in case of sckit models it could be numpy array/List of list/Panda DataFrame
 
    Returns
    -------
    predictions: Output from scoring server
        Format: { 'prediction': output from `model.predict` method }
 
    """
    assert model is not None, "Model is not loaded"
    X = pd.read_json(io.StringIO(data)) if isinstance(data, str) else pd.DataFrame.from_dict(data)
    preds = model.predict(X).tolist()  
    return { 'prediction': preds }

Im vorherigen Beispiel muss die Artefaktstruktur folgendermaßen aussehen:


.
|-- score.py 
|-- dataframelabelencoder.py 
|-- model.joblib 
|-- runtime.yaml

load_model()- und predict()-Signaturen ändern

Die Funktion predict(data, model=load_model()) erwartet die Payload-Daten und ein Modellobjekt und wird standardmäßig von load_model() zurückgegeben. In Ihrem Anwendungsfall muss möglicherweise ein zusätzlicher Parameter an predict() übergeben werden. Ein Beispiel ist ein Scaler oder eine Lookup-Tabelle. Sie können dieser Funktion Parameter hinzufügen, wenn dem hinzugefügten Parameter ein Standardwert zugewiesen ist.

Im folgenden Beispiel basieren die Vorhersagen auf einem PCA-Modell und einem Scaler-Objekt. Sie zeigt, wie predict() einen zusätzlichen Parameter mit dem Namen scaler annehmen kann. Standardmäßig gibt die Funktion load_scaler() einen Wert an den Parameter scaler zurück. Es wird empfohlen, dass Sie diesem Muster folgen. Wenn predict() oder load_model() zusätzliche Parameter erfordern, müssen diese auf Standardwerte gesetzt werden, die von Funktionen zurückgegeben werden, die in score.py definiert sind. In diesem Beispiel werden Parameter zu predict() hinzugefügt:

import json
import os
from cloudpickle import cloudpickle
model_pickle_name = 'pca.pkl'
scaler_pickle_name = 'scaler.pkl'
"""
   Inference script. This script is used for prediction by scoring server when schema is known.
"""
def load_model(model_file_name=model_pickle_name):
    """
    Loads model from the serialized format
    Returns
    -------
    model:  a model instance on which predict API can be invoked
    """
    model_dir = os.path.dirname(os.path.realpath(__file__))
    contents = os.listdir(model_dir)
    if model_file_name in contents:
        with open(os.path.join(os.path.dirname(os.path.realpath(__file__)), model_file_name), "rb") as file:
            return cloudpickle.load(file)
    else:
        raise Exception('{0} is not found in model directory {1}'.format(model_file_name, model_dir))
 
def load_scaler(model_file_name=scaler_pickle_name):
    """
    Loads model from the serialized format
    Returns
    -------
    model:  a model instance on which predict API can be invoked
    """
    model_dir = os.path.dirname(os.path.realpath(__file__))
    contents = os.listdir(model_dir)
    if model_file_name in contents:
        with open(os.path.join(os.path.dirname(os.path.realpath(__file__)), model_file_name), "rb") as file:
            return cloudpickle.load(file)
    else:
        raise Exception('{0} is not found in model directory {1}'.format(model_file_name, model_dir))
 
def predict(data, model=load_model(), scaler=load_scaler()):
    """
    Returns prediction given the model and data to predict
    Parameters
    ----------
    model: Model instance returned by load_model API
    data: Data format as expected by the predict API of the core estimator. For eg. in case of sckit models it could be numpy array/List of list/Panda DataFrame
    Returns
    -------
    predictions: Output from scoring server
        Format: {'prediction':output from model.predict method}
    """
    from pandas import read_json, DataFrame
    from io import StringIO
    X = read_json(StringIO(data)) if isinstance(data, str) else DataFrame.from_dict(data)
    X_s = scaler.transform(X)
    return {'prediction':model.transform(X_s).tolist()[0]}

Modellartefakte vor dem Speichern testen

Bevor Sie ein Modell im Katalog speichern, sollten Sie das Artefakt gründlich testen. Dieses Code-Snippet zum Testen eines Modellartefakts, bevor es im Katalog gespeichert wird:

Der Python-Pfad wird geändert, indem der Pfad zum Modellartefakt eingefügt wird.
Es lädt das Modell mit load_model() in den Arbeitsspeicher.
Es ruft schließlich predict() auf.

Bevor Sie das Code-Snippet ausführen, erstellen Sie in einer Notizbuchsession eine neue Notizbuchdatei, ändern Sie den Kernel, und wählen Sie dann dieselbe Conda-Umgebung aus, die Sie für das Modell-Deployment verwenden möchten (Inferenz-Conda-Umgebung). Kopieren Sie das Code-Snippet, fügen Sie es ein, und führen Sie den Code aus.

Beispiel für Modellartefakt-Testcode

import sys
from json import dumps
 
 
# The local path to your model artifact directory is added to the Python path.
# replace <your-model-artifact-path>
sys.path.insert(0, f"<your-model-artifact-path>")
 
# importing load_model() and predict() that are defined in score.py
from score import load_model, predict
 
# Loading the model to memory
_ = load_model()
# Making predictions on a JSON string object (dumps(data)). Here we assume
# that predict() is taking data in JSON format
predictions_test = predict(dumps(data), _)
predictions_test

Die Methode predictions_test enthält die Vorhersagen des Modells für die Beispiel-Payload data mit JSON-Zeichenfolgen. Vergleichen Sie predictions_test mit einem bekannten Modellergebnis für ein bestimmtes Dataset. Beispiel: data könnte ein Muster des Trainings-Datasets sein.

Beispiel für Bildvorhersagen

Die Methode score.py predict() verarbeitet die empfangene Anforderung wie in dieser score.py-Beispieldatei:

"""
   Inference script. This script is used for prediction by scoring server when schema is known.
"""
import torch
import torchvision
import io
import numpy as np
from PIL import Image
import os
 
 
# COCO Labels
COCO_INSTANCE_CATEGORY_NAMES = [
    '__background__', 'person', 'bicycle', 'car', 'motorcycle']
 
 
model_name = 'PyTorch_Retinanet.pth'
def load_model(model_file_name = model_name):
    """
    Loads model from the serialized format
 
    Returns
    -------
    model:  Pytorch model instance
    """
    model = torchvision.models.detection.retinanet_resnet50_fpn(pretrained=False, pretrained_backbone=False)
    cur_dir = os.path.dirname(os.path.abspath(__file__))
    model.load_state_dict(torch.load(os.path.join(cur_dir, model_file_name)))
    model.eval()
    return model
 
 
def predict(data, model=load_model()):
    """
    Returns prediction given the model and data to predict
 
    Parameters
    ----------
    model: Model instance returned by load_model API
    data: Data format in json
 
    Returns
    -------
    predictions: Output from scoring server
        Format: {'prediction':output from model.predict method}
 
    """
    img_bytes = io.BytesIO(data)
    image = Image.open(img_bytes)
    image_np = np.asarray(image)
    image_th = torch.from_numpy(image_np)
    image_th = image_th.permute(2, 0, 1)
    image_th = image_th.unsqueeze(0) / 255
    with torch.no_grad():
        pred = model(image_th)
    object_index_list = np.argwhere(pred[0].get("scores") > 0.5)
    label_index_list = pred[0].get("labels")
    labels = [COCO_INSTANCE_CATEGORY_NAMES[label_index_list[i]] for i in object_index_list]
    box_list = pred[0].get("boxes")
    boxes = [box_list[i].numpy().tolist() for i in object_index_list][0]
    return {'prediction': {
        'labels': labels,
        'boxes': boxes,
    }}