model_garden/notebooks/official/pipelines/multicontender_vs_champion_deployment_method.ipynb

In [ ]:

# Copyright 2023 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Multicontender vs Champion methodology for model deployment into production

Open in Colab

Open in Colab Enterprise

Open in Workbench

View on GitHub

Overview

This tutorial shows how to use Vertex AI Pipelines for deploying the next version of a model into production using the multicontender vs champion method.

NOTE: This notebook uses KFP 1.x and GCPC 1.x. It's recommended to use 2.x.

NOTE: This notebook has been tested in the following environment:

• Python version = 3.9

Objective

In this tutorial, you learn how to construct a Vertex AI pipeline, which evaluates new production data from a deployed (production) model against other versions (contenders) of the model, to determine if a contender model becomes the champion model for replacement in production.

This tutorial uses the following Vertex AI services and resources:

• Vertex AI Pipeline
• Vertex AI Model Evaluation
• Vertex AI Model Registry
• Vertex AI Endpoints

The steps performed include:

• Import a pretrained (champion) model to the Vertex AI Model Registry.
• Import synthetic model training evaluation metrics to the corresponding (champion) model.
• Create a Vertex AI endpoint resource.
• Deploy the champion model to the endpoint resource.
• Import additional (contender) versions of the deployed model.
• Import synthetic model training evaluation metrics to the corresponding (contender) models.
• Create a Vertex AI Pipeline that runs the following steps:
  • Get the champion model.
  • (Fake) Fine-tune champion model with production data.
  • Import synthetic train+production evaluation metrics for the champion model.
  • Get the contender models.
  • (Fake) Fine-tune contender model with production data.
  • Import synthetic train+production evaluation metrics for the contenders models.
  • Compare the evaluations of the contenders to the champion and set the new champion as the default.
  • Deploy the new champion model.

Learn more about Vertex AI Pipelines.

Model

This tutorial uses a pre-trained image classification model from TensorFlow Hub, which is trained on ImageNet dataset.

Learn more about ResNet V2 pretained model.

Costs

This tutorial uses billable components of Google Cloud:

• Vertex AI
• Cloud Storage

Learn about Vertex AI pricing, and Cloud Storage pricing, and use the Pricing Calculator to generate a cost estimate based on your projected usage.

Get started

Install Vertex AI SDK for Python and other required packages

In [ ]:

# Install the packages
! pip3 install --quiet --upgrade google-cloud-aiplatform \
                                 'google-cloud-pipeline-components<2'

! pip3 install --quiet           tensorflow==2.5 \
                                 tensorflow_hub

! pip3 install --quiet --upgrade 'kfp<2'

Restart runtime (Colab only)

To use the newly installed packages, you must restart the runtime on Google Colab.

In [ ]:

import sys

if "google.colab" in sys.modules:

    import IPython

    app = IPython.Application.instance()
    app.kernel.do_shutdown(True)

⚠️ The kernel is going to restart. Wait until it's finished before continuing to the next step. ⚠️

Authenticate your notebook environment (Colab only)

Authenticate your environment on Google Colab.

In [ ]:

import sys

if "google.colab" in sys.modules:

    from google.colab import auth

    auth.authenticate_user()

Set Google Cloud project information

Learn more about setting up a project and a development environment.

In [ ]:

PROJECT_ID = "[your-project-id]"  # @param {type:"string"}
LOCATION = "us-central1"  # @param {type:"string"}

Create a Cloud Storage bucket

Create a storage bucket to store intermediate artifacts such as datasets.

In [ ]:

BUCKET_URI = f"gs://your-bucket-name-{PROJECT_ID}-unique"  # @param {type:"string"}

If your bucket doesn't already exist: Run the following cell to create your Cloud Storage bucket.

In [ ]:

! gsutil mb -l {LOCATION} -p {PROJECT_ID} {BUCKET_URI}

Service Account

You use a service account to create Vertex AI Pipeline jobs.

If you don't want to use your project's Compute Engine service account, set SERVICE_ACCOUNT to another service account ID.

In [ ]:

SERVICE_ACCOUNT = "[your-service-account]"  # @param {type:"string"}

In [ ]:

import sys

IS_COLAB = "google.colab" in sys.modules
if (
    SERVICE_ACCOUNT == ""
    or SERVICE_ACCOUNT is None
    or SERVICE_ACCOUNT == "[your-service-account]"
):
    # Get your service account from gcloud
    if not IS_COLAB:
        shell_output = !gcloud auth list 2>/dev/null
        SERVICE_ACCOUNT = shell_output[2].replace("*", "").strip()

    else:  # IS_COLAB:
        shell_output = ! gcloud projects describe  $PROJECT_ID
        project_number = shell_output[-1].split(":")[1].strip().replace("'", "")
        SERVICE_ACCOUNT = f"{project_number}-compute@developer.gserviceaccount.com"

    print("Service Account:", SERVICE_ACCOUNT)

Set service account access for Vertex AI Pipelines

Run the following commands to grant your service account access to read and write pipeline artifacts in the bucket that you created in the previous step. You only need to run these once per service account.

In [ ]:

! gsutil iam ch serviceAccount:{SERVICE_ACCOUNT}:roles/storage.objectCreator $BUCKET_URI

! gsutil iam ch serviceAccount:{SERVICE_ACCOUNT}:roles/storage.objectViewer $BUCKET_URI

Import libraries

In [ ]:

import kfp
import tensorflow as tf
import tensorflow_hub as hub
from google.cloud import aiplatform
from google.cloud.aiplatform import gapic
from kfp.v2 import compiler
from kfp.v2.dsl import component

Initialize Vertex AI SDK for Python

To get started using Vertex AI, you must enable the Vertex AI API in your Google Cloud project.

Then, initialize the Vertex AI SDK for Python for your project.

In [ ]:

aiplatform.init(project=PROJECT_ID, location=LOCATION, staging_bucket=BUCKET_URI)

Set hardware accelerators

You can set hardware accelerators for training and prediction.

Set the variables DEPLOY_GPU/DEPLOY_NGPU to use a container image supporting a GPU and the number of GPUs allocated to the virtual machine (VM) instance. For example, to use a GPU container image with 4 Nvidia Telsa T4 GPUs allocated to each VM, you would specify:

(aip.gapic.AcceleratorType.NVIDIA_TESLA_T4, 4)

Otherwise specify (None, None) to use a container image to run on a CPU.

Learn more about hardware accelerator support for your region.

In [ ]:

DEPLOY_GPU, DEPLOY_NGPU = (None, None)

Set pre-built containers

Set the pre-built Docker container image for prediction.

For the latest list, see Pre-built containers for prediction.

In [ ]:

TF = "2.5".replace(".", "-")

if DEPLOY_GPU:
    DEPLOY_VERSION = "tf2-gpu.{}".format(TF)
else:
    DEPLOY_VERSION = "tf2-cpu.{}".format(TF)

DEPLOY_IMAGE = "{}-docker.pkg.dev/vertex-ai/prediction/{}:latest".format(
    LOCATION.split("-")[0], DEPLOY_VERSION
)

print("Deployment:", DEPLOY_IMAGE, DEPLOY_GPU, DEPLOY_NGPU)

Set machine type

Next, set the machine type to use for deployment.

• Set the variable DEPLOY_COMPUTE to configure the compute resources for the VMs you use for deployment. For this, you specify:
  • machine_type
    • n1-standard: 3.75GB of memory per vCPU.
    • n1-highmem: 6.5GB of memory per vCPU
    • n1-highcpu: 0.9 GB of memory per vCPU
  • vCPUs: number of [2, 4, 8, 16, 32, 64, 96 ]

Note: The following aren't supported.

• standard: 2 vCPUs
• highcpu: 2, 4 and 8 vCPUs

Note: You may also use n2 and e2 machine types for training and deployment, but they don't support GPUs.

In [ ]:

machine_type = "n1-standard"
vCPUs = "4"
DEPLOY_COMPUTE = f"{machine_type}-{vCPUs}"
print("Deploy machine type", DEPLOY_COMPUTE)

Get pretrained model from TensorFlow Hub

For demonstration purposes, this tutorial uses a pretrained model from TensorFlow Hub (TFHub), which is then uploaded to Vertex AI Model Registry. Once you have a Vertex AI model resource, the model can be deployed to a Vertex AI endpoint resource.

Download the pretrained model

First, you download the pretrained model from TensorFlow Hub. The model gets downloaded as a TF.Keras layer. To finalize the model, in this example, you create a Sequential() model with the downloaded TFHub model as a layer, and specify the input shape to the model.

In [ ]:

tfhub_model = tf.keras.Sequential(
    [hub.KerasLayer("https://tfhub.dev/google/imagenet/resnet_v2_101/classification/5")]
)

tfhub_model.build([None, 32, 32, 3])

tfhub_model.summary()

Save the model artifacts

At this point, the model is in memory. Next, save the model artifacts to a Cloud Storage location.

In [ ]:

# Save the model to Cloud Storage path
MODEL_DIR = BUCKET_URI + "/model"
tfhub_model.save(MODEL_DIR)

Upload the TensorFlow Hub model to Vertex AI Model Registry

Finally, you upload the model artifacts from the TFHub model into Vertex AI Model Registry and get a model resource object using the method upload(), with the following parameters:

• display_name: A human readable name for the model resource.
• artifact_uri: The Cloud Storage location of the model package.
• serving_container_image_uri: The serving container image.

Uploading a model into a Vertex AI Model Registry returns a long running operation, since it may take a few moments.

Note: When you upload the model artifacts to Vertex AI Model Registry, specify the corresponding deployment container image.

In [ ]:

champion_model = aiplatform.Model.upload(
    display_name="resnet-v1",
    artifact_uri=MODEL_DIR,
    serving_container_image_uri=DEPLOY_IMAGE,
    is_default_version=True,
    version_aliases=["v1"],
)

print(champion_model)

Create a model evaluation

First, you create a model evaluation in a format that corresponds to one of the predefined schemas for model evaluations. In this example, you use the schema for a classification metric, and specify the following subset of evaluation metrics as a dictionary:

• logLoss: The log loss.
• auPrc: The accuracy.

You then construct the ModelEvaluation object with the following parameters:

• display_name: The human readable name for the evaluation metric.
• metrics_schema_uri: The schema for the specific type of evaluation metrics.
• metrics: The dictionary with the evaluation metrics.

Learn more about Schemas for evaluation metrics.

In [ ]:

metrics = {"logLoss": 1.4, "auPrc": 0.85}
print(metrics)

champion_eval = gapic.ModelEvaluation(
    display_name="train-v1",
    metrics_schema_uri="gs://google-cloud-aiplatform/schema/modelevaluation/classification_metrics_1.0.0.yaml",
    metrics=metrics,
)

Upload the evaluation metrics to the Model Registry

Next, upload the model's evaluation from the custom training job to the corresponding entry in the Vertex AI Model Registry.

Currently, the Python SDK does not support for this method. Instead, use the lower level GAPIC API interface.

In [ ]:

API_ENDPOINT = f"{LOCATION}-aiplatform.googleapis.com"
client = gapic.ModelServiceClient(client_options={"api_endpoint": API_ENDPOINT})

client.import_model_evaluation(
    parent=champion_model.resource_name, model_evaluation=champion_eval
)

Creating an endpoint resource

You create an endpoint resource using the Endpoint.create() method. At a minimum, you specify the display name for the endpoint. Optionally, you can specify the project and location (region); otherwise the settings are inherited by the values you set when you initialized the Vertex AI SDK with the init() method.

In this example, the following parameters are specified:

• display_name: A human readable name for the endpoint resource.
• project: Your project ID.
• location: Your region.

This method returns an Endpoint object.

Learn more about Vertex AI Endpoints.

In [ ]:

endpoint = aiplatform.Endpoint.create(
    display_name="production", project=PROJECT_ID, location=LOCATION
)

print(endpoint)

Deploy the model resource to the endpoint resource

Next, you deploy the blessed Vertex AI model resource to a Vertex AI endpoint resource. The container image defined for the Vertex AI model resource is used for deployment. To deploy, you specify the following additional configuration settings:

• The machine type.
• The (if any) type and number of GPUs.
• Static, manual or auto-scaling of VM instances.

In this example, you deploy the model with the minimal amount of specified parameters, as follows:

• model: The model resource.
• deployed_model_displayed_name: The human readable name for the deployed model instance.
• machine_type: The machine type for each VM instance.

This may take a few minutes due to the provisioning of the configured resources.

In [ ]:

response = endpoint.deploy(
    model=champion_model,
    deployed_model_display_name="champion",
    machine_type=DEPLOY_COMPUTE,
)

print(response)

Create multiple contender versions of the deployed model

Next, you create plural synthetic versions of the deployed champion model.

In [ ]:

contender_model_1 = aiplatform.Model.upload(
    display_name="resnet-v2",
    artifact_uri=MODEL_DIR,
    serving_container_image_uri=DEPLOY_IMAGE,
    parent_model=champion_model.resource_name,
    is_default_version=False,
    version_aliases=["v2"],
)

contender_model_2 = aiplatform.Model.upload(
    display_name="resnet-v3",
    artifact_uri=MODEL_DIR,
    serving_container_image_uri=DEPLOY_IMAGE,
    parent_model=champion_model.resource_name,
    is_default_version=False,
    version_aliases=["v3"],
)

Create a model evaluation for the contender models

Next, you create a model evaluation for each of the contender models.

In [ ]:

metrics = {"logLoss": 1.5, "auPrc": 0.83}

contender_1_eval = gapic.ModelEvaluation(
    display_name="train-v2",
    metrics_schema_uri="gs://google-cloud-aiplatform/schema/modelevaluation/classification_metrics_1.0.0.yaml",
    metrics=metrics,
)

# workaround for bug
contender_model_1.versioning_registry.add_version_aliases(
    new_aliases=["default"], version=contender_model_1.version_id
)

client.import_model_evaluation(
    parent=contender_model_1.resource_name, model_evaluation=contender_1_eval
)

metrics = {"logLoss": 1.6, "auPrc": 0.82}

contender_2_eval = gapic.ModelEvaluation(
    display_name="train-v3",
    metrics_schema_uri="gs://google-cloud-aiplatform/schema/modelevaluation/classification_metrics_1.0.0.yaml",
    metrics=metrics,
)

# workaround for bug
contender_model_2.versioning_registry.add_version_aliases(
    new_aliases=["default"], version=contender_model_2.version_id
)

client.import_model_evaluation(
    parent=contender_model_2.resource_name, model_evaluation=contender_2_eval
)

Create custom components for pipeline

Next, you create several custom components you use in your pipeline.

Create component to import classification metrics

First, you define a component to import the evaluation metrics for the challenger model to the Model Registry. The component takes the following arguments:

• display_name: Human readable name for the evaluation metrics
• metrics: The evaluation metrics formatted for classification.
• parent_model_resource: The full resource name for the challenger model version.
• region: The region where you want to create or use the resources.

In [ ]:

@component(packages_to_install=["google-cloud-aiplatform"])
def import_classification_metrics(
    display_name: str,
    metrics: dict,
    parent_model_resource: str,
    project: str,
    region: str,
):
    from google.cloud import aiplatform
    from google.cloud.aiplatform import gapic

    print("DISPLAY", display_name)

    evaluation = gapic.ModelEvaluation(
        display_name=display_name,
        metrics_schema_uri="gs://google-cloud-aiplatform/schema/modelevaluation/classification_metrics_1.0.0.yaml",
        metrics=metrics,
    )

    # workaround for bug
    aiplatform.init(project=project, location=region)
    parent_model = aiplatform.Model(parent_model_resource)
    parent_model.versioning_registry.add_version_aliases(
        new_aliases=["default"], version=parent_model.version_id
    )

    API_ENDPOINT = f"{region}-aiplatform.googleapis.com"
    client = gapic.ModelServiceClient(client_options={"api_endpoint": API_ENDPOINT})
    client.import_model_evaluation(
        parent=parent_model_resource, model_evaluation=evaluation
    )

Create component to compare metrics

Next, you define a component to compare the auPrc metric between the champion and contender versions of the model. Whichever has the best auPrc value is set as the default model. When you subsequently deploy, the default model is deployed. The component takes the following arguments:

• champion_resource_name: The full resource name of the champion model.
• contender_model_resource_names: A list of the full resource namea of the contender models.

In [ ]:

@component(packages_to_install=["google-cloud-aiplatform"])
def compare_metrics(
    champion_model_resource_name: str, contender_model_resource_names: list
):
    from google.cloud import aiplatform

    # Get the metrics for the blessed model
    champion_model = aiplatform.Model(champion_model_resource_name)
    champion_eval = champion_model.list_model_evaluations()[
        1
    ]  # index 1 is the production eval data
    champion_auPrc = champion_eval.metrics["auPrc"]

    champion_model.versioning_registry.add_version_aliases(
        new_aliases=["default"], version=champion_model.version_id
    )

    # Get the metrics for the challenger model
    for contender in contender_model_resource_names:
        contender_model = aiplatform.Model(contender)
        contender_eval = contender_model.list_model_evaluations()[1]
        contender_auPrc = contender_eval.metrics["auPrc"]

        # Which model has the best accuracy becomes the default model
        if contender_auPrc > champion_auPrc:
            contender_model.versioning_registry.add_version_aliases(
                new_aliases=["default"], version=contender_model.version_id
            )
            champion_auPrc = contender_auPrc

Champion vs multi-contender pipeline

In this section, you construct a pipeline to fetch the champion, and the challenger models and compare them to deploy the best one to the production endpoint.

Define the pipeline

Define your pipeline to run the following tasks:

• Get the champion version of a model.
• Get the endpoint for the deployed champion model.
• Re-train and evaluate (faked) the champion model with production data.
• Import the champion model's evaluation metrics for the production data.
• For each contender model:
  • Get the contender version of the model.
  • Re-train and evaluate (faked) the contender model with production data.
• Compare the champion and contenders evaluation metrics, and set the default accordingly.
• Import the existing production endpoint.
• Deploy the new default model to the production endpoint.

In [ ]:

@kfp.dsl.pipeline(name="multicontender-vs-champion")
def pipeline(
    champion_model_resource: str,
    contender_model_resources: list,
    serving_container: str,
    machine_type: str,
    endpoint_resource_name: str,
    endpoint_resource_uri: str,
    project: str = PROJECT_ID,
    region: str = LOCATION,
):
    from google_cloud_pipeline_components.experimental.evaluation import \
        GetVertexModelOp
    from google_cloud_pipeline_components.types import artifact_types
    from google_cloud_pipeline_components.v1.endpoint import ModelDeployOp
    from kfp.v2.components import importer_node

    # Get the Vertex AI model resource of the blessed model
    champion = GetVertexModelOp(model_resource_name=champion_model_resource)

    # pretend to retrain and evaluate the champion with production data
    champion_metrics = {"logLoss": 1.3, "auPrc": 0.86}

    # upload the metrics for the champion version
    import_champion_metrics = import_classification_metrics(
        display_name="production",
        metrics=champion_metrics,
        parent_model_resource=champion_model_resource,
        project=project,
        region=region,
    ).after(champion)

    ix = 0
    with kfp.dsl.ParallelFor(contender_model_resources).after(
        import_champion_metrics
    ) as contender_model_resource:
        contender = GetVertexModelOp(model_resource_name=contender_model_resource)

        # pretend to retrain and evaluate the champion with production data
        contender_metrics = {"logLoss": 1.1, "auPrc": 0.88}

        ix += 1
        import_contender_metrics = import_classification_metrics(
            display_name=f"production_{ix}",
            metrics=contender_metrics,
            parent_model_resource=contender_model_resource,
            project=project,
            region=region,
        ).after(contender)

    # Select the best model
    compare = compare_metrics(
        champion_model_resource,
        contender_model_resources,
    ).after(import_contender_metrics)

    # import the production Endpoint
    endpoint = importer_node.importer(
        artifact_uri=endpoint_resource_uri,
        artifact_class=artifact_types.VertexEndpoint,
        metadata={"resourceName": endpoint_resource_name},
    )

    # deploy model to endpoint
    _ = ModelDeployOp(
        model=champion.outputs["model"],
        endpoint=endpoint.output,
        dedicated_resources_min_replica_count=1,
        dedicated_resources_max_replica_count=1,
        dedicated_resources_machine_type=machine_type,
        traffic_split={"0": 100},
    ).after(compare)

Compile the pipeline

Next, you compile the pipeline to a JSON file.

In [ ]:

# Compile the pipeline to a json file
compiler.Compiler().compile(
    pipeline_func=pipeline, package_path="multicontender_vs_champion.json"
)

Execute the pipeline

Finally, you execute your pipeline, passing the following pipeline parameter values:

• blessed_model_resource: The full resource name of the current blessed version of the model.
• serving_container: The serving container for deploying the model.
• machine_type: The machine type for deploying the model.
• endpoint_resource_name: The full resource name of the production endpoint.
• endpoint_resource_uri: The full URI for the production endpoint.
• project:The project ID.
• region: The region where you want to run your pipeline.

In [ ]:

PIPELINE_ROOT = "{}/pipeline_root/control".format(BUCKET_URI)

job = aiplatform.PipelineJob(
    display_name="multicontender_vs_champion",
    template_path="multicontender_vs_champion.json",
    pipeline_root=PIPELINE_ROOT,
    parameter_values={
        "champion_model_resource": champion_model.resource_name,
        "contender_model_resources": [
            contender_model_1.resource_name,
            contender_model_2.resource_name,
        ],
        "serving_container": DEPLOY_IMAGE,
        "machine_type": DEPLOY_COMPUTE,
        "endpoint_resource_name": endpoint.resource_name,
        "endpoint_resource_uri": "https://us-central1-aiplatform.googleapis.com/v1/"
        + endpoint.resource_name,
        "project": PROJECT_ID,
        "region": LOCATION,
    },
    enable_caching=False,
)

job.run()

Get the latest state of the production endpoint

Once your pipeline execution is finished, notice that the contender (version 3) model has replaced the previous champion model on the production endpoint.

Now, display the latest details of the deployed models from the production endpoint, and then display the traffic split. Notice that the resource ID for the 100% entry is the resource ID for the contender (verson 3) model.

In [ ]:

gca_resource = endpoint.list(filter="display_name=production")[0].gca_resource
print("Deployed Models", gca_resource.deployed_models)
print("\n")
print("Traffic Split", gca_resource.traffic_split)

Cleaning up

To clean up all Google Cloud resources used in this project, you can delete the Google Cloud project you used for the tutorial.

Otherwise, you can delete the individual resources you created in this tutorial.

In [ ]:

# Undeploy the model from endpoint
endpoint.undeploy_all()

# Delete the endpoint resource
endpoint.delete()

# Delete model resource
champion_model.delete()

# Delete the pipeline job resource
job.delete()

# Delete Cloud Storage objects that were created
delete_bucket = True
if delete_bucket:
    ! gsutil -m rm -r $BUCKET_URI

# Remove the local pipeline package file
! rm multicontender_vs_champion.json