This tutorial can be downloaded as part of the Wallaroo Tutorials repository.

Inference Endpoint API Spec Tutorial

Models deployed to Wallaroo are available for inference requests either via the Wallaroo SDK or through API calls through the inference endpoint. The inference endpoint API spec retrieval command creates an yaml file in OpenAPI format that details the inference URL, its endpoints, input and output parameters, and other useful information for developers.

This tutorial demonstrates how to:

• Deploy a Wallaroo Custom Model in Wallaroo.
• Perform sample inferences via the Wallaroo SDK.
• Generate the inference endpoint API spec retrieval command and display some of the data.
• Perform a sample inference using the inference endpoint.

Tutorial Prerequisites

• Wallaroo Version 2025.2 or above instance.

References

• Wallaroo SDK Essentials Guide: Model Uploads and Registrations: Custom Model
• Retrieve Inference Endpoint API Spec

Tutorial Steps

Import Libraries

The first step is to import the libraries we’ll be using. These are included by default in the Wallaroo instance’s JupyterHub service.

import numpy as np
import pandas as pd
import pyarrow as pa
import wallaroo

from wallaroo.deployment_config import DeploymentConfigBuilder
from wallaroo.framework import Framework

Open a Connection to Wallaroo

The next step is connect to Wallaroo through the Wallaroo client. The Python library is included in the Wallaroo install and available through the Jupyter Hub interface provided with your Wallaroo environment.

This is accomplished using the wallaroo.Client() command, which provides a URL to grant the SDK permission to your specific Wallaroo environment. When displayed, enter the URL into a browser and confirm permissions. Store the connection into a variable that can be referenced later.

If logging into the Wallaroo instance through the internal JupyterHub service, use wl = wallaroo.Client(). For more details on logging in through Wallaroo, see the Wallaroo SDK Essentials Guide: Client Connection.

wl = wallaroo.Client()

Set Variables

We’ll set the name of our workspace, pipeline, models and files. Workspace names must be unique across the Wallaroo workspace. For this, we’ll add in a randomly generated 4 characters to the workspace name to prevent collisions with other users’ workspaces. If running this tutorial, we recommend hard coding the workspace name so it will function in the same workspace each time it’s run.

workspace_name = f'vgg16-clustering-workspace'
pipeline_name = f'vgg16-clustering-pipeline'

model_name = 'vgg16-clustering'
model_file_name = './models/model-auto-conversion-BYOP-vgg16-clustering.zip'

Create Workspace and Pipeline

We will now create the Wallaroo workspace to store our model and set it as the current workspace. Future commands will default to this workspace for pipeline creation, model uploads, etc. We’ll create our Wallaroo pipeline that is used to deploy our arbitrary Python model.

workspace = wl.get_workspace(name=workspace_name, create_if_not_exist=True)
wl.set_current_workspace(workspace)

pipeline = wl.build_pipeline(pipeline_name)

Upload Custom Model

Custom Models are uploaded to Wallaroo through the Wallaroo Client upload_model method.

Upload Custom Model Parameters

The following parameters are required for Custom Models. Note that while some fields are considered as optional for the upload_model method, they are required for proper uploading of a Custom Model to Wallaroo.

Once the upload process starts, the model is containerized by the Wallaroo instance. This process may take up to 10 minutes.

Upload Custom Model Return

The following is returned with a successful model upload and conversion.

For our example, we’ll start with setting the input_schema and output_schema that is expected by our ImageClustering._predict() method.

input_schema = pa.schema([
    pa.field('images', pa.list_(
        pa.list_(
            pa.list_(
                pa.int64(),
                list_size=3
            ),
            list_size=32
        ),
        list_size=32
    )),
])

output_schema = pa.schema([
    pa.field('predictions', pa.int64()),
])

Upload Model

Now we’ll upload our model. The framework is Framework.CUSTOM for arbitrary Python models, and we’ll specify the input and output schemas for the upload.

model = wl.upload_model(model_name, 
                        model_file_name, 
                        framework=Framework.CUSTOM, 
                        input_schema=input_schema, 
                        output_schema=output_schema,
                        convert_wait=True)
model

Waiting for model loading - this will take up to 10min.
Model is pending loading to a container runtime.
Model is attempting loading to a container runtime........
Successful
Ready

Deploy Pipeline

The model is uploaded and ready for use. We’ll add it as a step in our pipeline, then deploy the pipeline. For this example we’re allocated 0.25 cpu and 4 Gi RAM to the pipeline through the pipeline’s deployment configuration.

pipeline.clear()
pipeline.add_model_step(model)

deployment_config = DeploymentConfigBuilder() \
    .cpus(1).memory('4Gi') \
    .build()

pipeline.deploy(deployment_config=deployment_config, wait_for_status=False)

Deployment initiated for vgg16-clustering-pipeline. Please check pipeline status.

import time
time.sleep(15)

while pipeline.status()['status'] != 'Running':
    time.sleep(15)
    print("Waiting for deployment.")
    display(pipeline.status()['status'])

Waiting for deployment.

'Error'

Waiting for deployment.

'Running'

Run inference

Everything is in place - we’ll now run a sample inference with some toy data. In this case we’re randomly generating some values in the data shape the model expects, then submitting an inference request through our deployed pipeline.

input_data = {
        "images": [np.random.randint(0, 256, (32, 32, 3), dtype=np.uint8)] * 2,
}
dataframe = pd.DataFrame(input_data)
dataframe

pipeline.infer(dataframe, timeout=10000)

Generate Inference Endpoint API Spec

For the next exercise assume that another developer wants to submit inference requests via API calls.

The method wallaroo.pipeline.Pipeline.generate_api_spec() returns the pipeline inference endpoint specification yaml format under the OpenAPI 3.1.1 format. This provides developers the ability to import this yaml file into their development environments and have:

• Inference API endpoint(s).
  • Inference endpoints require authentication bearer tokens. For more details, see Retrieve Pipeline Inference URL Token.
• Input fields and data schemas.
• Output fields and data schemas.

Retrieve Inference Endpoint API Spec Parameters

Retrieve Inference Endpoint API Spec Returns

A yaml file in OPenAPI 3.1.1 format for the specific pipeline that contains:

• URL: The deployed pipeline URL, for example, for the pipeline sample-pipeline this URL could be: https://example.wallaroo.ai/v1/api/pipelines/infer/sample-pipeline-414/sample-pipeline
• PATHS: The paths for each endpoint enabled. Endpoints differ depending on whether pipelines include models with OpenAI API compatibility enabled.

Retrieve Inference Endpoint API Spec Example

For this example, the file ./vgg16-spec.yaml is created from the deployed pipeline. We will extract the endpoint URL from this file.

Generate Inference Endpoint API Spec

From the YAML file we extract the URL and the endpoint. The endpoint is / as a standard inference endpoint for deployed pipelines.

spec_file = "./vgg16-spec.yaml"

pipeline.generate_api_spec(path=spec_file)

import yaml
spec_file = "./vgg16-spec.yaml"

# open the yaml file
with open(spec_file, 'r') as f:
    data = yaml.load(f, Loader=yaml.SafeLoader)

# display the deploy_url
deploy_url = data['servers'][0]['url']
deploy_url

'https://autoscale-uat-gcp.wallaroo.dev/v1/api/pipelines/infer/vgg16-clustering-pipeline-65/vgg16-clustering-pipeline'

The inference request is performed via an API request. For this example, the Wallaroo SDK is used to generate the bearer tokens used for authentication.

dataFile="./data/sample_data.json"
# get authorization header
headers = wl.auth.auth_header()
# set the content type for pandas records
headers['Content-Type']= 'application/json; format=pandas-records'
headers['Accept']='application/json; format=pandas-records'

!curl -X POST {deploy_url} \
    -H "Authorization:{headers['Authorization']}" \
    -H "Content-Type:{headers['Content-Type']}" \
    -H "Accept:{headers['Accept']}" \
    --data @{dataFile} > curl_response.df

  % Total    % Received % Xferd  Average Speed   Time    Time     Time  Current
                                 Dload  Upload   Total   Spent    Left  Speed
100 53074  100 26899  100 26175  62510  60828 --:--:-- --:--:-- --:--:--  120k

The following shows the sample inference response.

result_dataframe = pd.read_json('./curl_response.df', orient="records")
result_dataframe

Undeploy Pipelines

The inference is successful, so we will undeploy the pipeline and return the resources back to the cluster.

pipeline.undeploy()

 ok