Add an example use case for custom prediction routines. (#709)

* Add an example use case for custom prediction routines.

* Addressing some PR comments: reworded the readme in a few places, added a 'probe' command to build.py that sends a sample predict request, and pinned versions in requirements. Also fixed a bug where the artifacts_uri passed in during deployment on Vertex AI was not recognized as a directory.

* Autoformat code with black and fix a couple of typing errors.

* Addressing PR comments: Add deployment machine type to the config and add docstring to probe_prediction method.

* Update example to work with new LocalModel interface.

* Update example to work with new LocalModel interface.

Co-authored-by: Karl Weinmeister <11586922+kweinmeister@users.noreply.github.com>

community-content/CODEOWNERS

@@ -4,3 +4,4 @@
 /pytorch_text_classification_using_vertex_sdk_and_gcloud		@RajeshThallam @ultrons
 /sklearn_text_classification_from_script_using_vertex_sdk       @maxhardt
 /pluto_on_workbench @wkharold
+/cpr-examples @samthrasher

community-content/cpr-examples/timm_serving/.dockerignore

@@ -0,0 +1,5 @@
+testdata/*
+build.py
+test.py
+state_dict.pth
+config.json

community-content/cpr-examples/timm_serving/.gitignore

@@ -0,0 +1,5 @@
+cpr_model_server.py
+entrypoint.py
+state_dict.pth
+config.json
+**/__pycache__

community-content/cpr-examples/timm_serving/README.md

@@ -0,0 +1,93 @@
+# CPR Example: PyTorch Image Models (timm)
+
+## About CPR
+
+CPR ([custom prediction routines](https://github.com/googleapis/python-aiplatform/blob/custom-prediction-routine/google/cloud/aiplatform/prediction/README.md)) is a framework designed by Google Cloud developers to make it easier to combine machine learning models with custom preprocessing and postprocessing logic in a real-time serving application. 
+
+## Using this example
+
+This code is a self-contained example of a custom model server project built using CPR.
+
+As is, you can use it to serve the ViT-Small image classification model from Ross Wightman's [`timm`](https://github.com/rwightman/pytorch-image-models) library of image model implementations in PyTorch. Both CPU and GPU are supported.
+
+You can also consider using the code here as a template for your own CPR project if you want to use a different model from `timm`, a different PyTorch model, or an entirely different framework.
+
+### Requirements
+
+In order to use this example, you'll need Docker and Python 3 installed on your system.
+
+To get started, first create a virtual environment in an empty directory:
+```sh
+mkdir cpr-example
+python3 -m venv cpr-example
+cd cpr-example && source bin/activate
+``` 
+
+Then, clone the [vertex-ai-samples repo](https://github.com/GoogleCloudPlatform/vertex-ai-samples) in that directory:
+```sh
+git clone https://github.com/GoogleCloudPlatform/vertex-ai-samples.git
+cd vertex-ai-samples/community-content/cpr-examples/timm_serving
+```
+
+Finally, install the Python modules required to build and run the model server:
+```sh
+pip install -r requirements.txt
+```
+
+### Predictor
+
+The `TimmPredictor` class in `timm_serving/predictor.py` implements most of the important logic for the server.
+
+- `load(artifacts_dir)`: The predictor's `load` method is called when the server starts up in order to set up the predictor, usually by loading model weights and any artifacts needed for preprocessing and postprocessing. In this example, we initialize the saved model from the `state_dict.pth` file located inside the `artifacts_dir` folder and create the preprocessing transform from the model config.
+
+- `preprocess`, `predict`, `postprocess`: These methods are applied in sequence to the deserialized JSON data from each request. 
+    - `preprocess` decodes images from base64 and apply cropping, scaling and normalizing transforms. 
+    - `predict` runs the ViT-Small model on the preprocessed images and returns class scores.
+    - `postprocess` finds the top five classes and packs the class names, probabilities, and indices in a serializable result.
+
+### Building the container
+
+To build the model server locally, run the build command:
+```sh
+python build.py build
+```
+
+You can edit configuration values such as the model server's base image, the name and tag assigned to the image, and the path where model weights are stored locally.
+
+When you run the build command, model weights are downloaded and the model server container is built.
+
+### Running local tests
+
+`test.py` contains a suite of unit tests for the predictor as well as end-to-end tests for the model server. 
+
+To run the tests:
+```sh
+python test.py
+```
+
+All of the test images are public domain.
+- [Cat](https://commons.wikimedia.org/wiki/File:Stray_cat_on_wall.jpg)
+- [Airplane](https://commons.wikimedia.org/wiki/File:Airplanes_jets.jpg)
+- The infamous [mandrill](https://commons.wikimedia.org/wiki/File:Wikipedia-sipi-image-db-mandrill-4.2.03.png)
+
+### Deploying to Vertex AI
+
+Before uploading or deploying the container, you'll need to modify `config.py` to set appropriate values for: 
+- `project_id`: Your GCP project id.
+- `region`: Region where the model will be uploaded and deployed.
+- `repository`: [Artifact Registry repository](https://cloud.google.com/artifact-registry/docs/repositories/create-repos) in your project where the container image will be uploaded. 
+- `artifacts_gcs_dir`: Folder in a [Google Cloud Storage bucket](https://cloud.google.com/storage/docs/creating-buckets) where the model weights will be uploaded.
+
+Once this is done, first upload the model:
+```sh
+python build.py upload
+```
+
+Then deploy it:
+```sh
+python build.py deploy
+```
+
+If you run the deploy command again, it will create a new endpoint. If you want to undeploy the model, you can do so using the Vertex AI dashboard on the Google Cloud console, or use `gcloud ai endpoints undeploy` from the command line.
+
+After deploying successfully, you can run `python build.py probe` to send a sample request to the deployed model. 

community-content/cpr-examples/timm_serving/build.py

@@ -0,0 +1,117 @@
+# Copyright 2022 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.
+
+"""Build the model server container."""
+import json
+import logging
+import os
+import pathlib
+from typing import Sequence
+
+from absl import app
+from absl import logging
+from config import CPRConfig
+from google.cloud import aiplatform
+from google.cloud.aiplatform import prediction as cpr
+import smart_open
+import timm
+from timm_serving import predictor
+import torch
+
+
+def build_container(config: CPRConfig, tag: str) -> cpr.LocalModel:
+    """Build the model server container.
+
+    Args:
+      tag: Output image tag.
+
+    Returns:
+      LocalModel exposing the built model server.
+    """
+    return cpr.LocalModel.build_cpr_model(
+        src_dir=os.path.join(os.getcwd()),
+        output_image_uri=tag,
+        base_image=config.base_image,
+        predictor=predictor.TimmPredictor,
+        requirements_path=os.path.join(os.getcwd(), "requirements.txt"),
+    )
+
+
+def save_model_artifact(destination: str) -> None:
+    """Save a copy of the model state dict."""
+    model = timm.create_model(predictor.TimmPredictor.TIMM_MODEL_NAME, pretrained=True)
+    dest_file = os.path.join(destination, predictor.TimmPredictor.WEIGHTS_FILE)
+    with smart_open.open(dest_file, "wb") as f:
+        torch.save(model, f)
+    logging.info("Saved model to %s", dest_file)
+    logging.info("%s parameters", sum(p.numel() for p in model.parameters()))
+
+
+def upload_model(config: CPRConfig) -> aiplatform.Model:
+    """Tag and upload the model server."""
+    ar_tag = (
+        f"{config.region}-docker.pkg.dev/{config.project_id}"
+        f"/{config.repository}/{config.image}"
+    )
+    local_model = build_container(config, tag=ar_tag)
+    aiplatform.init(project=config.project_id, location=config.region)
+    local_model.push_image()
+    aip_model = aiplatform.Model.upload(
+        local_model=local_model,
+        display_name=predictor.TimmPredictor.TIMM_MODEL_NAME,
+        artifact_uri=config.artifact_gcs_dir,
+    )
+    config.model_name = aip_model.resource_name
+    config.save()
+    return aip_model
+
+
+def deploy_model(config: CPRConfig) -> aiplatform.Endpoint:
+    """Deploy the model server to a Vertex Prediction endpoint."""
+    aiplatform.init(project=config.project_id, location=config.region)
+    aip_model = aiplatform.Model(model_name=config.model_name)
+    endpoint = aip_model.deploy(machine_type=config.machine_type)
+    config.endpoint_name = endpoint.resource_name
+    config.save()
+    return endpoint
+
+
+def probe_prediction(config: CPRConfig, request_path: str) -> None:
+    """Send a sample prediction request to the Vertex Prediction endpoint."""
+    aiplatform.init(project=config.project_id, location=config.region)
+    aip_endpoint = aiplatform.Endpoint(endpoint_name=config.endpoint_name)
+    with open(request_path) as f:
+        logging.info(aip_endpoint.predict(**json.load(f)))
+
+
+def main(argv: Sequence[str]):
+    config = CPRConfig()
+    if pathlib.Path(config.config_file).exists():
+        config.load()
+
+    actions = set(argv[1:])
+    if "build" in actions:
+        build_container(config, config.image)
+        save_model_artifact(config.artifact_local_dir)
+    if "upload" in actions:
+        save_model_artifact(config.artifact_gcs_dir)
+        upload_model(config)
+    if "deploy" in actions:
+        deploy_model(config)
+    if "probe" in actions:
+        probe_prediction(config, request_path="sample_request.json")
+
+
+if __name__ == "__main__":
+    app.run(main)

community-content/cpr-examples/timm_serving/config.py

@@ -0,0 +1,76 @@
+# Copyright 2022 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.
+
+import dataclasses
+import json
+
+
+@dataclasses.dataclass
+class CPRConfig(object):
+    """Configure the build process by editing the default values here.
+
+    config_file: File path used to save values in this config. (Some
+      values, such as the model name, are generated at build time and
+      depended on by future steps, so saving it allows this script to
+      deploy the model without re-uploading it, for example.)
+
+    base_image: Base Docker image on top of which the model server will
+      be built. By default, a Debian-based Python 3 image without GPU
+      support will be used.
+
+    image: Name and tag assigned to the built model server image.
+
+    artifact_local_dir: Local directory where a copy of the pretrained model weights
+      will be saved.
+
+    region: Google Cloud Region where the model will be uploaded during the
+      build process.
+
+    project_id: Google Cloud project ID.
+
+    repository: Name of the Artifact Registry repository where the container
+      will be uploaded.
+
+    artifact_gcs_dir: Location on GCS where a copy of the pretrained model
+      weights will be uploaded.
+
+    model_name: Full resource path of the uploaded model. This is a write-only
+      field, the value is generated by Vertex AI when the model is uploaded.
+
+    endpoint_name: Full resource path of the created endpoint. This is a
+      write-only field, the value is generated by Vertex AI when the model is
+      deployed to an endpoint.
+
+    machine_type: Machine type to use when deploying the model.
+    """
+
+    config_file: str = "config.json"
+    base_image: str = "python:3.10-bullseye"
+    image: str = "timm_predictor:latest"
+    artifact_local_dir: str = ""
+    region: str = "us-central1"
+    project_id: str = "samthrasher-experimental"
+    repository: str = "cpr-images"
+    artifact_gcs_dir: str = "gs://samthrasher-cpr-example/timm-vit224/"
+    model_name: str = ""
+    endpoint_name: str = ""
+    machine_type: str = "n1-standard-2"
+
+    def save(self):
+        with open(self.config_file, "w") as f:
+            json.dump(dataclasses.asdict(self), f, indent=2)
+
+    def load(self):
+        with open(self.config_file) as f:
+            self.__init__(**json.load(f))

community-content/cpr-examples/timm_serving/requirements.txt

@@ -0,0 +1,8 @@
+absl-py==1.1.0
+fastapi==0.75.2
+uvicorn==0.18.2
+timm==0.5.4
+smart_open==6.0.0
+
+google-cloud-storage>=1.26.0,<2.0.0dev
+google-cloud-aiplatform[prediction] @ git+https://github.com/googleapis/python-aiplatform.git@custom-prediction-routine

community-content/cpr-examples/timm_serving/test.py

@@ -0,0 +1,249 @@
+"""Test the timm_serving predictor."""
+import base64
+import json
+import logging
+import os
+import pickle
+from typing import List, Dict
+
+from absl import flags
+from absl import logging
+from absl.testing import absltest
+from config import CPRConfig
+import fastapi
+from google.cloud import aiplatform
+from google.cloud.aiplatform import prediction as cpr
+import PIL
+from timm_serving import predictor
+import torch
+
+VIT_SMALL_PARAMS = 22878952
+
+
+def b64_encode_file(path: str) -> str:
+    """Encode a file's contents as base64.
+
+    Args:
+      path: Path to the file.
+
+    Returns:
+      Base64-encoded contents of the file.
+    """
+    with open(path, "rb") as f:
+        return str(base64.b64encode(f.read()), encoding="utf-8")
+
+
+def make_instance_dict(
+    image_paths: List[str], base64_encodings: List[str]
+) -> Dict[str, List[str]]:
+    """Generate a dictionary similar to a parsed prediction server request.
+
+    Args:
+      image_paths: Paths to image files to include.
+      base64_encodings: Pre-encoded base64 strings.
+
+    Returns:
+      Dictionary of instances in the format accepted by the preprocessor.
+    """
+    instances = [s for s in base64_encodings]
+    for path in image_paths:
+        instances.append(b64_encode_file(path))
+    return {"instances": instances}
+
+
+def count_parameters(model: torch.nn.Module):
+    """Count the parameters in a Pytorch model.
+
+    Args:
+      model: Pytorch model (nn.Module).
+
+    Returns:
+      Number of parameters in the model.
+
+    """
+    return sum(p.numel() for p in model.parameters())
+
+
+class PredictorUnitTests(absltest.TestCase):
+    """Unit tests for timm_serving.predictor."""
+
+    def setUp(self):
+        super().setUp()
+        self.config = CPRConfig()
+        self.config.load()
+        self.predictor = predictor.TimmPredictor()
+
+    def test_load_from_saved_state_dict_ok(self):
+        self.predictor.load(self.config.artifact_local_dir)
+        self.assertEqual(count_parameters(self.predictor._model), VIT_SMALL_PARAMS)
+
+    def test_load_bad_path(self):
+        with self.assertRaises(FileNotFoundError):
+            self.predictor.load("testdata/")
+        with self.assertRaisesRegex(ValueError, "not a directory"):
+            self.predictor.load("blah")
+
+    def test_load_bad_data(self):
+        with self.assertRaises(pickle.UnpicklingError):
+            self.predictor.load("testdata/bad_model_1")
+        with self.assertRaisesRegex(RuntimeError, "Invalid magic number"):
+            self.predictor.load("testdata/bad_model_2")
+
+    def test_preprocess_ok(self):
+        self.predictor.load(self.config.artifact_local_dir)
+        instance_dict = make_instance_dict(
+            base64_encodings=[],
+            image_paths=[
+                "testdata/airplane.jpg",
+                "testdata/mandrill.tiff",
+                "testdata/mandrill.tiff",
+                "testdata/cat_alpha.png",
+            ],
+        )
+        result = self.predictor.preprocess(instance_dict)
+        self.assertEqual(result.size(), torch.Size([4, 3, 224, 224]))
+        self.assertEqual(result.dtype, torch.float32)
+
+    def test_preprocess_no_instances(self):
+        self.predictor.load(self.config.artifact_local_dir)
+        with self.assertRaises(fastapi.HTTPException) as ctx:
+            self.predictor.preprocess({})
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertRegex(ctx.exception.detail, 'must contain "instances"')
+
+    def test_preprocess_wrong_shape_instances(self):
+        self.predictor.load(self.config.artifact_local_dir)
+        instance_dict = {"instances": [[b64_encode_file("testdata/mandrill.tiff")]]}
+        with self.assertRaises(fastapi.HTTPException) as ctx:
+            self.predictor.preprocess(instance_dict)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertRegex(ctx.exception.detail, "not 'list'")
+
+    def test_preprocess_bad_base64(self):
+        self.predictor.load(self.config.artifact_local_dir)
+        instance_dict = make_instance_dict(base64_encodings=["!@#$"], image_paths=[])
+        with self.assertRaises(fastapi.HTTPException) as ctx:
+            self.predictor.preprocess(instance_dict)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertRegex(ctx.exception.detail, "[Bb]ase64")
+
+    def test_preprocess_not_image_data(self):
+        self.predictor.load(self.config.artifact_local_dir)
+        instance_dict = make_instance_dict(
+            base64_encodings=[], image_paths=["testdata/bad.jpg"]
+        )
+        with self.assertRaises(fastapi.HTTPException) as ctx:
+            self.predictor.preprocess(instance_dict)
+        self.assertEqual(ctx.exception.status_code, 400)
+        self.assertRegex(ctx.exception.detail, "image file")
+
+    def test_predict_ok(self):
+        self.predictor.load(self.config.artifact_local_dir)
+        inputs = torch.zeros(size=[2, 3, 224, 224], dtype=torch.float32)
+        if torch.cuda.device_count() > 0:
+            inputs = inputs.cuda()
+        result = self.predictor.predict(inputs)
+        self.assertEqual(result.size(), torch.Size([2, 1000]))
+        self.assertEqual(result.dtype, torch.float32)
+
+    def test_postprocess_ok(self):
+        class_probs = torch.zeros(size=[2, 1000])
+        class_probs[0, 0] = 1
+        class_probs[1, 123] = 1
+        result = self.predictor.postprocess(class_probs)
+        predictions = result["predictions"]
+        self.assertLen(predictions[0]["class_names"], 5)
+        self.assertLen(predictions[0]["indices"], 5)
+        self.assertLen(predictions[0]["probabilities"], 5)
+        self.assertLen(predictions[1]["class_names"], 5)
+        self.assertLen(predictions[1]["indices"], 5)
+        self.assertLen(predictions[1]["probabilities"], 5)
+        self.assertContainsSubsequence(predictions[0]["class_names"][0], "tench")
+        self.assertContainsSubsequence(
+            predictions[1]["class_names"][0], "spiny lobster"
+        )
+
+
+class ServerEndToEndTests(absltest.TestCase):
+    """End-to-end tests for the model server, using LocalEndpoint."""
+
+    def setUp(self):
+        super().setUp()
+        self.config = CPRConfig()
+        self.config.load()
+        self.local_model = cpr.LocalModel(
+            serving_container_spec=aiplatform.gapic.ModelContainerSpec(
+                image_uri=self.config.image
+            )
+        )
+
+        self.local_endpoint = self.local_model.deploy_to_local_endpoint(
+            artifact_uri=self.config.artifact_local_dir or os.getcwd()
+        )
+        self.local_endpoint.serve()
+
+    def tearDown(self):
+        self.local_endpoint.stop()
+        super().tearDown()
+
+    def test_e2e_healthcheck_ok(self):
+        health_check_response = self.local_endpoint.run_health_check()
+        self.assertEqual(health_check_response.status_code, 200)
+        self.assertEqual(health_check_response.content, b"{}")
+
+    def test_e2e_predict_ok(self):
+        predict_request = json.dumps(
+            make_instance_dict(
+                base64_encodings=[],
+                image_paths=[
+                    "testdata/mandrill.tiff",
+                ],
+            )
+        )
+        response = self.local_endpoint.predict(
+            request=predict_request, headers={"Content-Type": "application/json"}
+        )
+        logging.info(response.content)
+        self.assertEqual(response.status_code, 200)
+        predictions = response.json()["predictions"]
+        self.assertContainsSubsequence(predictions[0]["class_names"][0], "baboon")
+
+    def test_e2e_predict_bad_json_returns_400(self):
+        predict_request = "blah"
+        response = self.local_endpoint.predict(
+            request=predict_request, headers={"Content-Type": "application/json"}
+        )
+        logging.info(response.content)
+        self.assertEqual(response.status_code, 400)
+
+    def test_e2e_predict_no_instances_returns_400(self):
+        predict_request = json.dumps({})
+        response = self.local_endpoint.predict(
+            request=predict_request, headers={"Content-Type": "application/json"}
+        )
+        logging.info(response.content)
+        self.assertEqual(response.status_code, 400)
+
+    def test_e2e_predict_bad_base64_returns_400(self):
+        predict_request = json.dumps(
+            make_instance_dict(base64_encodings=["blah"], image_paths=[])
+        )
+        response = self.local_endpoint.predict(
+            request=predict_request, headers={"Content-Type": "application/json"}
+        )
+        logging.info(response.content)
+        self.assertEqual(response.status_code, 400)
+
+    def test_e2e_predict_bad_image_returns_400(self):
+        predict_request = json.dumps(
+            make_instance_dict(base64_encodings=[], image_paths=["testdata/bad.jpg"])
+        )
+        response = self.local_endpoint.predict(
+            request=predict_request, headers={"Content-Type": "application/json"}
+        )
+        logging.info(response.content)
+        self.assertEqual(response.status_code, 400)
+
+
+if __name__ == "__main__":
+    absltest.main()

community-content/cpr-examples/timm_serving/testdata/bad.jpg

@@ -0,0 +1 @@
+some non-image data

community-content/cpr-examples/timm_serving/testdata/bad.txt

@@ -0,0 +1 @@
+some non-image data

community-content/cpr-examples/timm_serving/timm_serving/predictor.py

@@ -0,0 +1,178 @@
+# Copyright 2022 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.
+
+"""Adapts a pretrained TIMM image classification model to the CPR framework.
+
+Documentation for the TIMM (Torch IMage Models) library is here:
+  https://rwightman.github.io/pytorch-image-models/
+
+Its source can also be found here:
+  https://github.com/rwightman/pytorch-image-models
+"""
+
+import base64
+import binascii
+import io
+import os
+from typing import Dict, List, Union
+
+from fastapi import HTTPException
+from google.cloud.aiplatform import prediction as cpr
+from pathlib import Path
+import PIL
+import smart_open
+import timm
+import torch
+import torch.nn.functional as F
+
+with open(Path(__file__).parent.absolute().joinpath("imagenet.txt")) as f:
+    IMAGENET_CLASSES = f.read().splitlines()
+
+
+class TimmPredictor(cpr.predictor.Predictor):
+    """Predictor class for image models based on TIMM."""
+
+    TIMM_MODEL_NAME = os.getenv("TIMM_MODEL_NAME", default="vit_small_patch32_224")
+    WEIGHTS_FILE = "state_dict.pth"
+    NUM_TOP_CLASSES_TO_RETURN = 5
+
+    def __init__(self):
+        self._cuda = torch.cuda.device_count() > 0
+
+    def load(self, artifacts_uri: str = ""):
+        """Initializes the model and preprocessing transforms.
+
+        Args:
+          artifacts_uri: Directory where state dict is stored. Can be a
+          GCS URI or local path.
+        """
+        if artifacts_uri:
+            artifact_path = os.path.join(artifacts_uri)
+            if not (os.path.isdir(artifact_path) or artifact_path.startswith("gs://")):
+                raise ValueError("Provided artifact_uri is not a directory.")
+        else:
+            artifact_path = os.getcwd()
+
+        artifact_path = os.path.join(artifact_path, self.WEIGHTS_FILE)
+        with smart_open.open(artifact_path, "rb") as f:
+            self._model = torch.load(f)
+
+        if self._cuda:
+            self._model.cuda()
+
+        config = timm.data.resolve_data_config(model=self.TIMM_MODEL_NAME, args=[])
+        self._transform = timm.data.create_transform(
+            is_training=False, use_prefetcher=False, **config
+        )
+
+    def preprocess(self, request_dict: Dict[str, List[str]]) -> torch.Tensor:
+        """Performs preprocessing.
+
+        By default, the server expects a request body consisting of a valid JSON
+        object. This will be parsed by the handler before it's evaluated by the
+        preprocess method.
+
+        Args:
+          request_dict: Parsed request body. We expect that the input consists of
+            a list of base64-encoded image files under the "instances" key. (Any
+            image format that PIL.image.open can handle is okay.)
+
+        Returns:
+          torch.Tensor containing the preprocessed images as a batch. If GPU is
+          available, the result tensor will be stored on GPU.
+        """
+
+        if "instances" not in request_dict:
+            raise HTTPException(
+                status_code=400,
+                detail='Request must contain "instances" as a top-level key.',
+            )
+
+        tensors = []
+
+        for (i, image) in enumerate(request_dict["instances"]):
+            # We use Base64 encoding to handle image data.
+            # This is probably the best we can do while still using JSON input.
+            # Overriding the input format requires building a custom Handler.
+            try:
+                image_bytes = base64.b64decode(image, validate=True)
+            except (binascii.Error, TypeError) as e:
+                raise HTTPException(
+                    status_code=400,
+                    detail=f"Base64 decoding of the input image at index {i} failed:"
+                    f" {str(e)}",
+                )
+
+            try:
+                pil_image = PIL.Image.open(io.BytesIO(image_bytes)).convert("RGB")
+            except PIL.UnidentifiedImageError:
+                raise HTTPException(
+                    status_code=400,
+                    detail=f"The input image at index {i} could not be identified as an"
+                    " image file.",
+                )
+
+            tensors.append(self._transform(pil_image))
+
+        with torch.inference_mode():
+            result = torch.stack(tensors)
+            if self._cuda:
+                result = result.cuda()
+        return result
+
+    def predict(self, instances: torch.Tensor) -> torch.Tensor:
+        """Performs prediction.
+
+        Args:
+          instances: torch.Tensor with type torch.float32 and shape
+            [?, 3, 224, 224], containing the pre-processed input images.
+
+        Returns:
+          Vector of scores with type torch.float32 and shape [?, 1000],
+            representing the model's estimate of the likelihood that the
+            input belongs to the Imagenet class with that index.
+        """
+        with torch.inference_mode():
+            class_scores = self._model(instances)
+        return class_scores
+
+    def postprocess(
+        self, class_scores: torch.Tensor
+    ) -> Dict[str, List[Dict[str, Union[str, int, float]]]]:
+        """Translate the model output into a classification result.
+
+        Args:
+          class_scores: torch.Tensor with type torch.float32 and shape
+            [?, 1000], containing the scores assigned to each class by
+            the model.
+
+        Returns:
+          Dictionary containing the list of classification results. Each
+            classification result contains the probabilities, class names, and
+            class indices of the classes with the top class scores as reported by
+            the model.
+        """
+        class_probs = F.softmax(class_scores, dim=1)
+        top_k = class_probs.topk(self.NUM_TOP_CLASSES_TO_RETURN)
+        top_k_values = top_k.values.numpy().tolist()
+        top_k_indices = top_k.indices.numpy().tolist()
+        predictions = [
+            dict(
+                probabilities=values,
+                indices=indices,
+                class_names=[IMAGENET_CLASSES[int(class_num)] for class_num in indices],
+            )
+            for (values, indices) in zip(top_k_values, top_k_indices)
+        ]
+        return {"predictions": predictions}