Quick Start¶
ADS can auto generate the required files to register and deploy your models. Checkout the examples below to learn how to deploy models of different frameworks.
Sklearn¶
import tempfile
import ads
from ads.model.framework.sklearn_model import SklearnModel
from sklearn.datasets import load_iris
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split
ads.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)
# Instantiate ads.model.framework.sklearn_model.SklearnModel using the sklearn LogisticRegression model
sklearn_model = SklearnModel(
estimator=sklearn_estimator, artifact_dir=tempfile.mkdtemp()
)
# Autogenerate score.py, serialized model, runtime.yaml, input_schema.json and output_schema.json
sklearn_model.prepare(
inference_conda_env="dbexp_p38_cpu_v1",
X_sample=X_train,
y_sample=y_train,
)
# Verify generated artifacts
sklearn_model.verify(X_test)
# Register scikit-learn model
model_id = sklearn_model.save(display_name="Sklearn Model")
XGBoost¶
Create a model, prepare it, verify that it works, save it to the model catalog, deploy it, make a prediction, and then delete the deployment.
import tempfile
import ads
import xgboost as xgb
from ads.model.framework.xgboost_model import XGBoostModel
from sklearn.datasets import load_iris
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
ads.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 XBoost Classifier model
xgboost_estimator = xgb.XGBClassifier()
xgboost_estimator.fit(X_train, y_train)
# Instantiate ads.model.framework.xgboost_model.XGBoostModel using the trained XGBoost Model
xgboost_model = XGBoostModel(estimator=xgboost_estimator, artifact_dir=tempfile.mkdtemp())
# Autogenerate score.py, serialized model, runtime.yaml, input_schema.json and output_schema.json
xgboost_model.prepare(
inference_conda_env="generalml_p38_cpu_v1",
X_sample=X_train,
y_sample=y_train,
)
# Verify generated artifacts
xgboost_model.verify(X_test)
# Register XGBoost model
model_id = xgboost_model.save(display_name="XGBoost Model")
LightGBM¶
Create a model, prepare it, verify that it works, save it to the model catalog, deploy it, make a prediction, and then delete the deployment.
import tempfile
import ads
import lightgbm as lgb
from ads.model.framework.lightgbm_model import LightGBMModel
from sklearn.datasets import load_iris
from sklearn.model_selection import train_test_split
ads.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 XBoost Classifier model
train = lgb.Dataset(X_train, label=y_train)
param = {
'objective': 'multiclass', 'num_class': 3,
}
lightgbm_estimator = lgb.train(param, train)
# Instantiate ads.model.lightgbm_model.XGBoostModel using the trained LGBM Model
lightgbm_model = LightGBMModel(estimator=lightgbm_estimator, artifact_dir=tempfile.mkdtemp())
# Autogenerate score.py, serialized model, runtime.yaml, input_schema.json and output_schema.json
lightgbm_model.prepare(
inference_conda_env="generalml_p38_cpu_v1",
X_sample=X_train,
y_sample=y_train,
)
# Verify generated artifacts
lightgbm_model.verify(X_test)
# Register LightGBM model
model_id = lightgbm_model.save(display_name="LightGBM Model")
PyTorch¶
Create a model, prepare it, verify that it works, save it to the model catalog, deploy it, make a prediction, and then delete the deployment.
import tempfile
import ads
import torch
import torchvision
from ads.model.framework.pytorch_model import PyTorchModel
ads.set_auth(auth="resource_principal")
# Load a pre-trained resnet model
torch_estimator = torchvision.models.resnet18(pretrained=True)
torch_estimator.eval()
# create random test data
test_data = torch.randn(1, 3, 224, 224)
# Instantiate ads.model.framework.pytorch_model.PyTorchModel using the pre-trained PyTorch Model
artifact_dir=tempfile.mkdtemp()
torch_model = PyTorchModel(torch_estimator, artifact_dir=artifact_dir)
# Autogenerate score.py, serialized model, runtime.yaml
# Set `use_torch_script` to `True` to save the model as Torchscript program.
torch_model.prepare(inference_conda_env="pytorch110_p38_cpu_v1", use_torch_script=True)
# Verify generated artifacts
torch_model.verify(test_data)
# Register PyTorch model
model_id = torch_model.save(display_name="PyTorch Model")
Spark Pipeline¶
Create a model, prepare it, verify that it works, save it to the model catalog, deploy it, make a prediction, and then delete the deployment.
import os
import tempfile
import ads
from ads.model.framework.spark_model import SparkPipelineModel
from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import HashingTF, Tokenizer
from ads.model.framework.spark_model import SparkPipelineModel
spark = SparkSession \
.builder \
.appName("Python Spark SQL basic example") \
.getOrCreate()
# create data
training = spark.createDataFrame(
[
(0, "a b c d e spark", 1.0),
(1, "b d", 0.0),
(2, "spark f g h", 1.0),
(3, "hadoop mapreduce", 0.0),
],
["id", "text", "label"],
)
test = spark.createDataFrame(
[
(4, "spark i j k"),
(5, "l m n"),
(6, "spark hadoop spark"),
(7, "apache hadoop"),
],
["id", "text"],
)
# Train a Spark Pipeline model
tokenizer = Tokenizer(inputCol="text", outputCol="words")
hashingTF = HashingTF(inputCol=tokenizer.getOutputCol(), outputCol="features")
lr = LogisticRegression(maxIter=10, regParam=0.001)
pipeline = Pipeline(stages=[tokenizer, hashingTF, lr])
model = pipeline.fit(training)
# Instantite ads.model.framework.spark_model.SparkPipelineModel using the pre-trained Spark Pipeline Model
spark_model = SparkPipelineModel(estimator=model, artifact_dir=tempfile.mkdtemp())
spark_model.prepare(inference_conda_env="pyspark32_p38_cpu_v2",
X_sample = training,
force_overwrite=True)
# Verify generated artifacts
prediction = spark_model.verify(test)
#Register Spark model
spark_model.save(display_name="Spark Pipeline Model")
TensorFlow¶
Create a model, prepare it, verify that it works, save it to the model catalog, deploy it, make a prediction, and then delete the deployment.
from ads.model.framework.tensorflow_model import TensorFlowModel
import tensorflow as tf
from uuid import uuid4
mnist = tf.keras.datasets.mnist
(x_train, y_train), (x_test, y_test) = mnist.load_data()
x_train, x_test = x_train / 255.0, x_test / 255.0
tf_estimator = tf.keras.models.Sequential(
[
tf.keras.layers.Flatten(input_shape=(28, 28)),
tf.keras.layers.Dense(128, activation="relu"),
tf.keras.layers.Dropout(0.2),
tf.keras.layers.Dense(10),
]
)
loss_fn = tf.keras.losses.SparseCategoricalCrossentropy(from_logits=True)
tf_estimator.compile(optimizer="adam", loss=loss_fn, metrics=["accuracy"])
tf_estimator.fit(x_train, y_train, epochs=1)
# Instantite ads.model.framework.tensorflow_model.TensorFlowModel using the pre-trained TensorFlow Model
tf_model = TensorFlowModel(tf_estimator, artifact_dir=f"./model-artifact-{str(uuid4())}")
# Autogenerate score.py, pickled model, runtime.yaml, input_schema.json and output_schema.json
tf_model.prepare(inference_conda_env="tensorflow28_p38_cpu_v1")
# Verify generated artifacts
tf_model.verify(x_test[:1])
#Register TensorFlow model
model_id = tf_model.save(display_name="TensorFlow Model")
HuggingFace Pipelines¶
from transformers import pipeline
from ads.model import HuggingFacePipelineModel
import tempfile
import PIL.Image
from ads.common.auth import default_signer
import requests
import cloudpickle
## download the image
image_url = "https://huggingface.co/datasets/Narsil/image_dummy/raw/main/parrots.png"
image = PIL.Image.open(requests.get(image_url, stream=True).raw)
## download the pretrained model
classifier = pipeline(model="openai/clip-vit-large-patch14")
classifier(
images=image,
candidate_labels=["animals", "humans", "landscape"],
)
## Initiate a HuggingFacePipelineModel instance
zero_shot_image_classification_model = HuggingFacePipelineModel(classifier, artifact_dir=tempfile.mkdtemp())
# Autogenerate score.py, serialized model, runtime.yaml
conda_pack_path = "oci://bucket@namespace/path/to/conda/pack"
python_version = "3.x" # Remember to update 3.x with your actual python version, e.g. 3.8
zero_shot_image_classification_model.prepare(inference_conda_env=conda_pack_path, inference_python_version = python_version, force_overwrite=True)
## Convert payload to bytes
data = {"images": image, "candidate_labels": ["animals", "humans", "landscape"]}
body = cloudpickle.dumps(data) # convert image to bytes
# Verify generated artifacts
zero_shot_image_classification_model.verify(data=data)
zero_shot_image_classification_model.verify(data=body)
# Register HuggingFace Pipeline model
zero_shot_image_classification_model.save()
## Deploy
log_group_id = "<log_group_id>"
log_id = "<log_id>"
zero_shot_image_classification_model.deploy(deployment_bandwidth_mbps=100,
wait_for_completion=False,
deployment_log_group_id = log_group_id,
deployment_access_log_id = log_id,
deployment_predict_log_id = log_id)
zero_shot_image_classification_model.predict(data)
zero_shot_image_classification_model.predict(body)
### Invoke the model by sending bytes
auth = default_signer()['signer']
endpoint = zero_shot_image_classification_model.model_deployment.url + "/predict"
headers = {"Content-Type": "application/octet-stream"}
requests.post(endpoint, data=body, auth=auth, headers=headers).json()
Other Frameworks¶
import tempfile
from ads.model.generic_model import GenericModel
# Create custom framework model
class Toy:
def predict(self, x):
return x ** 2
model = Toy()
# Instantite ads.model.generic_model.GenericModel using the trained Custom Model
generic_model = GenericModel(estimator=model, artifact_dir=tempfile.mkdtemp())
generic_model.summary_status()
# Autogenerate score.py, pickled model, runtime.yaml, input_schema.json and output_schema.json
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 Framework Model")
With Model Version Set¶
import tempfile
from ads.model.generic_model import GenericModel
# Create custom framework model
class Toy:
def predict(self, x):
return x ** 2
model = Toy()
# Instantite ads.model.generic_model.GenericModel using the trained Custom Model
generic_model = GenericModel(estimator=model, artifact_dir=tempfile.mkdtemp())
generic_model.summary_status()
# Within the context manager, you can save the :ref:`Model Serialization` model without specifying the ``model_version_set`` parameter because it's taken from the model context manager. If the model version set doesn't exist in the model catalog, the example creates a model version set named ``my_model_version_set``. If the model version set exists in the model catalog, the models are saved to that model version set.
with ads.model.experiment(name="my_model_version_set", create_if_not_exists=True):
# Autogenerate score.py, pickled model, runtime.yaml, input_schema.json and output_schema.json
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 Framework Model")