モデル・アーティファクトのベスト・プラクティス

score.pyファイルの記述

• 必ず、score.pyおよびruntime.yamlファイルがモデル・アーティファクトの最上位ディレクトリにあることを確認してください。

  アーティファクトに含める必要があるその他のファイルは、2つのファイルと同じレベルか、その下のディレクトリにある必要があります:

  .
  |-- runtime.yaml
  |-- score.py
  |-- <your-serialized-models>

• モデル・デプロイメントでは、score.py関数を使用してモデルをメモリーにロードし、予測を行います。
• 関数定義load_model()およびpredict()は編集できません。これらの関数の本体のみがカスタマイズ可能です。
• モデル・デプロイメント・サービスの使用時にディスクにデータを書き込むことができるパスは、/home/datascienceです。
• OCIアイデンティティ権限が正しく定義されて有効化されると、リソース・プリンシパルを使用してオブジェクト・ストレージにアクセスできます。

カスタム・モジュールのパッケージ化

score.pyまたはシリアライズ・モデルが依存するカスタム・モジュールは、score.pyと同じトップ・レベル・ディレクトリ以下にある個別のPythonスクリプトとして記述する必要があります。たとえば、model.joblibは、次の例のようにdataframelabelencoder.pyスクリプトで定義されているカスタムDataFrameLabelEncoderクラスに依存します:

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

その後、モジュールは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 }

前述の例では、アーティファクト構造は次のようになります:

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

load_model()およびpredict()のシグネチャの変更

predict(data, model=load_model())関数は、ペイロード・データとモデル・オブジェクトを想定し、デフォルトでload_model()によって返されます。ユース・ケースによっては、predict()に追加パラメータを渡す必要がある場合があります。たとえば、スケーラや参照表です。追加するパラメータにデフォルト値が割り当てられる場合は、それらの関数にパラメータを追加できます。

次の例では、予測はPCAモデルとスケーラ・オブジェクトに依存しています。predict()がscalerという追加パラメータを取る方法を示します。デフォルトでは、load_scaler()関数はscalerパラメータに値を返します。そのパターンに従うことをお薦めします。predict()またはload_model()に追加のパラメータが必要な場合は、score.pyで定義されている関数によって返されるデフォルト値に設定する必要があります。この例では、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]}

保存前のモデル・アーティファクトのテスト

モデルをカタログに保存する前に、アーティファクトを完全にテストすることをお薦めします。カタログに保存する前にモデル・アーティファクトをテストするための次のコード・スニペット:

• モデル・アーティファクトへのパスを挿入してPythonパスを変更します。

• load_model()を使用してモデルをメモリーにロードします。

• 最後に、predict()をコールします。

コード・スニペットを実行する前に、ノートブック・セッションで新しいノートブック・ファイルを作成し、カーネルを変更して、モデル・デプロイメントに使用するものと同じconda環境(推論conda環境)を選択します。コード・スニペットをコピーして貼り付け、コードを実行します。

モデル・アーティファクト・テスト・コードの例

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

predictions_testメソッドには、サンプルdata JSON文字列ペイロードに対してモデルによって行われた予測が含まれます。predictions_testを特定のデータセットの既知のモデル結果と比較します。たとえば、dataはトレーニング・データセットの例です。

画像予測の例

score.py predict()メソッドは、次のサンプルscore.pyファイルのように、受け取ったリクエストからの処理を行います:

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