Best practice per gli artifact modello

Utilizzare queste considerazioni aggiuntive per la creazione e il packaging degli artifact del modello.

Scrittura di un file score.py

Assicurarsi sempre che i file score.py e runtime.yaml si trovino nella directory di livello superiore di un artifact del modello.

Tutti gli altri file che devono far parte di un artifact devono essere allo stesso livello di questi due file o nelle directory sotto di essi:
```
.
|-- runtime.yaml
|-- score.py
|-- <your-serialized-models>
```
La distribuzione del modello utilizza le funzioni score.py per caricare un modello in memoria e fare previsioni.
Le definizioni delle funzioni, load_model() e predict(), non sono modificabili. Solo il corpo di queste funzioni è personalizzabile.
Il percorso consentito per scrivere i dati su disco quando si utilizza il servizio di distribuzione del modello è /home/datascience.
Puoi accedere allo storage degli oggetti utilizzando i principal delle risorse quando le autorizzazioni di identità OCI sono state definite correttamente per abilitarle.

Packaging - Moduli personalizzati

Qualsiasi modulo personalizzato da cui dipende score.py o il modello serializzato deve essere scritto come script Python separati nella stessa directory di livello superiore di score.py o sotto. Ad esempio, model.joblib dipende da una classe DataFrameLabelEncoder personalizzata, definita nello script dataframelabelencoder.py come nell'esempio seguente:

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

Il modulo viene quindi importato dal file score.py:

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

Nell'esempio precedente, la struttura dell'artifact deve essere:


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

Modifica delle firme load_model() e predict()

La funzione predict(data, model=load_model()) prevede i dati del payload e un oggetto modello e viene restituita da load_model() per impostazione predefinita. Il caso d'uso potrebbe richiedere che venga passato un parametro aggiuntivo a predict(). Un esempio è uno scaler o una tabella di ricerca. È possibile aggiungere parametri a tali funzioni se al parametro aggiunto è assegnato un valore predefinito.

Nell'esempio seguente, le previsioni si basano su un modello PCA e un oggetto scaler. Viene mostrato come predict() può utilizzare un parametro aggiuntivo chiamato scaler. Per impostazione predefinita, la funzione load_scaler() restituisce un valore al parametro scaler. Ti consigliamo di seguire questo schema. Se predict() o load_model() richiedono parametri aggiuntivi, è necessario impostarli sui valori predefiniti restituiti dalle funzioni definite in score.py. In questo esempio vengono aggiunti parametri a predict():

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

Test di un artifact del modello prima del salvataggio

Prima di salvare un modello nel catalogo, è consigliabile eseguire un test completo dell'artifact. Questo frammento di codice per eseguire il test di un artifact del modello prima di salvarlo nel catalogo:

Modifica il percorso Python inserendo il percorso dell'artifact modello.
Carica il modello in memoria utilizzando load_model().
Infine, chiama predict().

Prima di eseguire lo snippet di codice, creare un nuovo file notebook in una sessione notebook, modificare il kernel, quindi selezionare lo stesso ambiente Conda che si desidera utilizzare per la distribuzione del modello (ambiente Conda di riferimento). Copiare e incollare lo snippet di codice ed eseguire il codice.

Esempio di codice di test artifact modello

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

Il metodo predictions_test contiene le previsioni effettuate dal modello nel payload della stringa JSON data di esempio. Confrontare predictions_test con un risultato di modello noto per un determinato data set. Ad esempio, data potrebbe essere un esempio del set di dati di addestramento.

Esempio di previsioni immagini

Il metodo score.py predict() gestisce l'elaborazione della richiesta ricevuta come in questo file score.py di esempio:

"""
   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,
    }}