Other Frameworks

See API Documentation

Overview

The ads.model.generic_model.GenericModel class in ADS provides an efficient way to serialize almost any model class. This section demonstrates how to use the GenericModel class to prepare model artifacts, verify models, save models to the model catalog, deploy models, and perform predictions on model deployment endpoints.

The GenericModel class works with any unsupported model framework that has a .predict() method. For the most common model classes such as scikit-learn, XGBoost, LightGBM, TensorFlow, and PyTorch, we recommend that you use the ADS provided, framework-specific serializations models. For example, for a scikit-learn model, use SKLearnmodel. For other models, use the GenericModel class.

The .verify() method simulates a model deployment by calling the load_model() and predict() methods in the score.py file. With the .verify() method, you can debug your score.py file without deploying any models. The .save() method deploys a model artifact to the model catalog. The .deploy() method deploys a model to a REST endpoint.

These simple steps take your trained model and will deploy it into production with just a few lines of code.

Prepare Model Artifact

Instantiate a GenericModel() object by giving it any model object. It accepts the following parameters:

  • artifact_dir: str: Artifact directory to store the files needed for deployment.

  • auth: (Dict, optional): Defaults to None. The default authentication is set using the ads.set_auth API. To override the default, use ads.common.auth.api_keys() or ads.common.auth.resource_principal() and create the appropriate authentication signer and the **kwargs required to instantiate the IdentityClient object.

  • estimator: (Callable): Trained model.

  • properties: (ModelProperties, optional): Defaults to None. ModelProperties object required to save and deploy the model.

  • serialize: (bool, optional): Defaults to True. If True the model will be serialized into a pickle file. If False, you must set the model_file_name in the .prepare() method, serialize the model manually, and save it in the artifact_dir. You will also need to update the score.py file to work with this model.

The properties is an instance of the ModelProperties class and has the following predefined fields:

  • bucket_uri: str

  • compartment_id: str

  • deployment_access_log_id: str

  • deployment_bandwidth_mbps: int

  • deployment_instance_count: int

  • deployment_instance_shape: str

  • deployment_log_group_id: str

  • deployment_predict_log_id: str

  • deployment_memory_in_gbs: Union[float, int]

  • deployment_ocpus: Union[float, int]

  • inference_conda_env: str

  • inference_python_version: str

  • overwrite_existing_artifact: bool

  • project_id: str

  • remove_existing_artifact: bool

  • training_conda_env: str

  • training_id: str

  • training_python_version: str

  • training_resource_id: str

  • training_script_path: str

By default, properties is populated from the environment variables when not specified. For example, in notebook sessions the environment variables are preset and stored in project id (PROJECT_OCID) and compartment id (NB_SESSION_COMPARTMENT_OCID). So properties populates these environment variables, and uses the values in methods such as .save() and .deploy(). Pass in values to overwrite the defaults. When you use a method that includes an instance of properties, then properties records the values that you pass in. For example, when you pass inference_conda_env into the .prepare() method, then properties records the value. To reuse the properties file in different places, you can export the properties file using the .to_yaml() method then reload it into a different machine using the .from_yaml() method.

Summary Status

You can call the .summary_status() method after a model serialization instance such as GenericModel, SklearnModel, TensorFlowModel, or PyTorchModel is created. The .summary_status() method returns a Pandas dataframe that guides you through the entire workflow. It shows which methods are available to call and which ones aren’t. Plus it outlines what each method does. If extra actions are required, it also shows those actions.

The following image displays an example summary status table created after a user initiates a model instance. The table’s Step column displays a Status of Done for the initiate step. And the Details column explains what the initiate step did such as generating a score.py file. The Step column also displays the prepare(), verify(), save(), deploy(), and predict() methods for the model. The Status column displays which method is available next. After the initiate step, the prepare() method is available. The next step is to call the prepare() method.

../../../_images/summary_status.png

Example

By default, the GenericModel serializes to a pickle file. The following example, the user creates a model. In the prepare step, the user saves the model as a pickle file with the name toy_model.pkl. Then the user verifies the model, saves it to the model catalog, deploys the model and makes a prediction. Finally, the user deletes the model deployment and then deletes the model.

import tempfile
from ads.model.generic_model import GenericModel

class Toy:
    def predict(self, x):
        return x ** 2
model = Toy()

generic_model = GenericModel(estimator=model, artifact_dir=tempfile.mkdtemp())
generic_model.summary_status()

generic_model.prepare(
        inference_conda_env="dbexp_p38_cpu_v1",
        model_file_name="toy_model.pkl",
        force_overwrite=True
     )

# Check if the artifacts are generated correctly.
# The verify method invokes the ``predict`` function defined inside ``score.py`` in the artifact_dir
generic_model.verify(2)

# Register the model
model_id = generic_model.save(display_name="Custom Model")

# Deploy and create an endpoint for the XGBoost model
generic_model.deploy(
    display_name="My Custom Model",
    deployment_log_group_id="ocid1.loggroup.oc1.xxx.xxxxx",
    deployment_access_log_id="ocid1.log.oc1.xxx.xxxxx",
    deployment_predict_log_id="ocid1.log.oc1.xxx.xxxxx",
    # Shape config details mandatory for flexible shapes:
    # deployment_instance_shape="VM.Standard.E4.Flex",
    # deployment_ocpus=<number>,
    # deployment_memory_in_gbs=<number>,
)

print(f"Endpoint: {generic_model.model_deployment.url}")

# Generate prediction by invoking the deployed endpoint
generic_model.predict(2)

# To delete the deployed endpoint uncomment the line below
# generic_model.delete_deployment(wait_for_completion=True)

You can also use the shortcut .prepare_save_deploy() instead of calling .prepare(), .save() and .deploy() seperately.

import tempfile
from ads.model.generic_model import GenericModel

class Toy:
    def predict(self, x):
        return x ** 2
estimator = Toy()

model = GenericModel(estimator=estimator)
model.summary_status()

# If you are running the code inside a notebook session and using a service pack, `inference_conda_env` can be omitted.
model.prepare_save_deploy(inference_conda_env="dbexp_p38_cpu_v1")
model.verify(2)

# Generate prediction by invoking the deployed endpoint
model.predict(2)

# To delete the deployed endpoint uncomment the line below
# model.delete_deployment(wait_for_completion=True)

Example – CatBoost

Here is a more realistic example using CatBoost model.

import tempfile
import ads
from ads.model.generic_model import GenericModel
from catboost import CatBoostRegressor

ads.set_auth(auth="resource_principal")

# Initialize data

X_train = [[1, 4, 5, 6],
            [4, 5, 6, 7],
            [30, 40, 50, 60]]

X_test = [[2, 4, 6, 8],
            [1, 4, 50, 60]]

y_train = [10, 20, 30]

# Initialize CatBoostRegressor
catboost_estimator = CatBoostRegressor(iterations=2,
                        learning_rate=1,
                        depth=2)
# Train a CatBoostRegressor model
catboost_estimator.fit(X_train, y_train)

# Get predictions
preds = catboost_estimator.predict(X_test)

# Instantiate ads.model.generic_model.GenericModel using the trained Custom Model using the trained CatBoost Classifier  model
catboost_model = GenericModel(estimator=catboost_estimator,
                            artifact_dir=tempfile.mkdtemp(),
                            model_save_serializer="cloudpickle",
                            model_input_serializer="json")

# Autogenerate score.py, pickled model, runtime.yaml, input_schema.json and output_schema.json
catboost_model.prepare(
    inference_conda_env="oci://bucket@namespace/path/to/your/conda/pack",
    inference_python_version="your_python_version",
    X_sample=X_train,
    y_sample=y_train,
)

# Verify generated artifacts. Payload looks like this: [[2, 4, 6, 8], [1, 4, 50, 60]]
catboost_model.verify(X_test, auto_serialize_data=True)

# Register CatBoostRegressor model
model_id = catboost_model.save(display_name="CatBoost Model")
catboost_model.deploy()
catboost_model.predict(X_test)
catboost_model.delete_deployment(wait_for_completion=True)
catboost_model.delete() # delete the model

Example – Save Your Own Model

By default, the serialize in GenericModel class is True, and it will serialize the model using cloudpickle. However, you can set serialize=False to disable it. And serialize the model on your own. You just need to copy the serialized model into the .artifact_dir. This example shows step by step how you can do that. The example is illustrated using a Sklearn model.

import tempfile
from ads import set_auth
from ads.model import GenericModel
from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

set_auth(auth="resource_principal")

# Load dataset and Prepare train and test split
iris = load_iris()
X, y = iris.data, iris.target
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.25)

# Train a LogisticRegression model
sklearn_estimator = LogisticRegression()
sklearn_estimator.fit(X_train, y_train)

# Serialize your model. You can choose your own way to serialize your model.
import cloudpickle
with open("./model.pkl", "wb") as f:
    cloudpickle.dump(sklearn_estimator, f)

model = GenericModel(sklearn_estimator, artifact_dir = "model_artifact_folder", serialize=False)
model.prepare(inference_conda_env="generalml_p38_cpu_v1",force_overwrite=True, model_file_name="model.pkl", X_sample=X_test)

Now copy the model.pkl file and paste into the model_artifact_folder folder. And open the score.py in the model_artifact_folder folder to add implementation of the load_model function. You can also add your preprocessing steps in pre_inference function and postprocessing steps in post_inference function. Below is an example implementation of the score.py. Replace your score.py with the code below.

# score.py 1.0 generated by ADS 2.8.2 on 20230301_065458
import os
import sys
import json
from functools import lru_cache

model_name = 'model.pkl'


"""
Inference script. This script is used for prediction by scoring server when schema is known.
"""

@lru_cache(maxsize=10)
def load_model(model_file_name=model_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__))
    if model_dir not in sys.path:
        sys.path.insert(0, model_dir)
    contents = os.listdir(model_dir)
    if model_file_name in contents:
        import cloudpickle
        with open(os.path.join(model_dir, model_name), "rb") as f:
            model = cloudpickle.load(f)
        return model
    else:
        raise Exception(f'{model_file_name} is not found in model directory {model_dir}')

@lru_cache(maxsize=1)
def fetch_data_type_from_schema(input_schema_path=os.path.join(os.path.dirname(os.path.realpath(__file__)), "input_schema.json")):
    """
    Returns data type information fetch from input_schema.json.

    Parameters
    ----------
    input_schema_path: path of input schema.

    Returns
    -------
    data_type: data type fetch from input_schema.json.

    """
    data_type = {}
    if os.path.exists(input_schema_path):
        schema = json.load(open(input_schema_path))
        for col in schema['schema']:
            data_type[col['name']] = col['dtype']
    else:
        print("input_schema has to be passed in in order to recover the same data type. pass `X_sample` in `ads.model.framework.sklearn_model.SklearnModel.prepare` function to generate the input_schema. Otherwise, the data type might be changed after serialization/deserialization.")
    return data_type

def deserialize(data, input_schema_path):
    """
    Deserialize json serialization data to data in original type when sent to predict.

    Parameters
    ----------
    data: serialized input data.
    input_schema_path: path of input schema.

    Returns
    -------
    data: deserialized input data.

    """

    import pandas as pd
    import numpy as np
    import base64
    from io import BytesIO
    if isinstance(data, bytes):
        return data

    data_type = data.get('data_type', '') if isinstance(data, dict) else ''
    json_data = data.get('data', data) if isinstance(data, dict) else data

    if "numpy.ndarray" in data_type:
        load_bytes = BytesIO(base64.b64decode(json_data.encode('utf-8')))
        return np.load(load_bytes, allow_pickle=True)
    if "pandas.core.series.Series" in data_type:
        return pd.Series(json_data)
    if "pandas.core.frame.DataFrame" in data_type or isinstance(json_data, str):
        return pd.read_json(json_data, dtype=fetch_data_type_from_schema(input_schema_path))
    if isinstance(json_data, dict):
        return pd.DataFrame.from_dict(json_data)
    return json_data

def pre_inference(data, input_schema_path):
    """
    Preprocess data

    Parameters
    ----------
    data: Data format as expected by the predict API of the core estimator.
    input_schema_path: path of input schema.

    Returns
    -------
    data: Data format after any processing.

    """
    return deserialize(data, input_schema_path)

def post_inference(yhat):
    """
    Post-process the model results

    Parameters
    ----------
    yhat: Data format after calling model.predict.

    Returns
    -------
    yhat: Data format after any processing.

    """
    return yhat.tolist()

def predict(data, model=load_model(), input_schema_path=os.path.join(os.path.dirname(os.path.realpath(__file__)), "input_schema.json")):
    """
    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/Pandas DataFrame.
    input_schema_path: path of input schema.

    Returns
    -------
    predictions: Output from scoring server
        Format: {'prediction': output from model.predict method}

    """
    features = pre_inference(data, input_schema_path)
    yhat = post_inference(
        model.predict(features)
    )
    return {'prediction': yhat}

Save the score.py and now call .verify() to check if it works locally.

model.verify(X_test[:2], auto_serialize_data=True)

After verify run successfully, you can save the model to model catalog, deploy and call predict to invoke the endpoint.

model_id = model.save(display_name='Demo Sklearn model')
deploy = model.deploy(display_name='Demo Sklearn deployment')
model.predict(X_test[:2].tolist())

You can also use the shortcut .prepare_save_deploy() instead of calling .prepare(), .save() and .deploy() seperately.

import tempfile
from ads.catalog.model import ModelCatalog
from ads.model.generic_model import GenericModel

class Toy:
    def predict(self, x):
        return x ** 2
estimator = Toy()

model = GenericModel(estimator=estimator)
model.summary_status()
# If you are running the code inside a notebook session and using a service pack, `inference_conda_env` can be omitted.
model.prepare_save_deploy(inference_conda_env="dataexpl_p37_cpu_v3")
model.verify(2)
model.predict(2)
model.delete_deployment(wait_for_completion=True)
model.delete()