Deploy Transformer with InferenceService

A Transformer is an InferenceService component which performs pre/post processing alongside model inference. It takes raw input and transforms it into the input tensors that the model server expects. In this guide, we demonstrate running inference with a custom Transformer communicating using both REST and gRPC protocols.

Prerequisites

Before you begin, ensure you have:

• A Kubernetes cluster with KServe installed.
• kubectl command line tool installed and configured.
• Docker installed (for building custom transformer images).
• Access to a container registry where you can push images.
• Basic understanding of Python and PyTorch.

Create Custom Image Transformer

Implement pre/post processing with KServe Model API

The KServe.Model base class defines three main handlers: preprocess, predict, and postprocess. These handlers execute in sequence, where:

1. The output of the preprocess handler is passed to the predict handler
2. When predictor_host is provided, the predict handler calls the predictor and receives a response
3. This response is then passed to the postprocess handler

KServe automatically fills in the predictor_host for Transformer and handles the call to the Predictor. By default, the transformer makes a REST call to the predictor. To make a gRPC call instead, you can pass the --protocol argument with value grpc-v2.

To implement a Transformer, derive from the base Model class and overwrite the preprocess and postprocess handlers with your customized transformation logic. For the Open(v2) Inference Protocol, KServe provides InferRequest and InferResponse API objects that abstract away REST/gRPC encoding and decoding details.

import argparse
from kserve import Model, ModelServer, model_server, InferInput, InferRequest, logging
from typing import Dict
from PIL import Image
import torchvision.transforms as transforms
import logging
import io
import base64
import kserve


def image_transform(data):
    """converts the input image of Bytes Array into Tensor
    Args:
        data: The input image bytes.
    Returns:
        numpy.array: Returns the numpy array after the image preprocessing.
    """
    preprocess = transforms.Compose(
        [
            transforms.Resize(256),
            transforms.CenterCrop(224),
            transforms.ToTensor(),
            transforms.Normalize(mean=[0.485, 0.456, 0.406], std=[0.229, 0.224, 0.225]),
        ]
    )
    preprocess = transforms.Compose(
        [transforms.ToTensor(), transforms.Normalize((0.1307,), (0.3081,))]
    )
    byte_array = base64.b64decode(data)
    image = Image.open(io.BytesIO(byte_array))
    tensor = preprocess(image).numpy()
    return tensor

# for v1 REST predictor the preprocess handler converts to input image bytes to float tensor dict in v1 inference REST protocol format
class ImageTransformer(kserve.Model):
    def __init__(self, name: str):
        super().__init__(name, return_response_headers=True)
        self.ready = True

    def preprocess(self, payload: Dict, headers: Dict[str, str] = None) -> Dict:
        input_tensors = [
            image_transform(instance["image"]["b64"]) for instance in payload["instances"]
        ]
        inputs = [{"data": input_tensor.tolist()} for input_tensor in input_tensors]
        payload = {"instances": inputs}
        return payload

    def postprocess(self, outputs: Dict, headers: Dict[str, str] = None,
        response_headers: Dict[str, str] = None,
    ) -> Dict:
        # Example for custom response headers
        if response_headers is not None:
            response_headers.update(
                {"x-model-version": "1.0"}
            )
        return outputs


# for v2 gRPC predictor the preprocess handler converts the input image bytes tensor to float tensor in v2 inference protocol format
class ImageTransformer(kserve.Model):
    def __init__(self, name: str, headers: Dict[str, str] = None):
        super().__init__(name, return_response_headers=True)
        self.headers = headers
        self.ready = True

    def preprocess(
        self, request: InferRequest, headers: Dict[str, str] = None
    ) -> InferRequest:
        input_tensors = [
            image_transform(instance) for instance in request.inputs[0].data
        ]
        input_tensors = np.asarray(input_tensors)
        infer_inputs = [
            InferInput(
                name="INPUT__0",
                datatype="FP32",
                shape=list(input_tensors.shape),
                data=input_tensors,
            )
        ]
        infer_request = InferRequest(model_name=self.name, infer_inputs=infer_inputs)
        return infer_request
   
     def postprocess(self, infer_response: InferResponse, headers: Dict[str, str] = None,
        response_headers: Dict[str, str] = None,
    ) -> InferResponse:
        # Example for custom response headers
        if response_headers is not None:
            response_headers.update(
                {"x-model-version": "1.0"}
            )
        return infer_response

You can find the complete code example [here](https://github.com/kserve/kserve/tree/release-0.20/python/custom_transformer).

Transformer Server Entrypoint

For a single model, create a transformer object and register it to the model server:

if __name__ == "__main__":
    if args.configure_logging:
        logging.configure_logging(args.log_config_file)  # Configure kserve and uvicorn logger
    model = ImageTransformer(args.model_name, predictor_host=args.predictor_host,
                             protocol=args.protocol)
    ModelServer().start(models=[model])

For multiple models, if they can share the same transformer, register the same transformer for different models. If each model requires its own transformation, use different transformers:

if __name__ == "__main__":
    if args.configure_logging:
        logging.configure_logging(args.log_config_file)  # Configure kserve and uvicorn logger
    for model_name in model_names:
        transformer = ImageTransformer(model_name, predictor_host=args.predictor_host)
        models.append(transformer)
    kserve.ModelServer().start(models=models)

Configuring Logger for Serving Runtime

KServe allows users to override the default logger configuration of the serving runtime and uvicorn server. You can follow the logger configuration documentation to configure the logger.

Build Transformer Docker Image

From the kserve/python directory, build the transformer docker image using the [Dockerfile](https://github.com/kserve/kserve/blob/release-0.20/python/custom_transformer.Dockerfile):

cd python
docker build -t $DOCKER_USER/image-transformer:latest -f custom_transformer.Dockerfile .

docker push $DOCKER_USER/image-transformer:latest

Deploy the InferenceService with REST Predictor

Create the InferenceService

By default, InferenceService uses TorchServe to serve PyTorch models. Models can be loaded from a cloud storage model repository according to the TorchServe model repository layout. In this example, the model repository contains a MNIST model, but you can store multiple models there.

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: torch-transformer
spec:
  predictor:
    model:
      modelFormat:
        name: pytorch
      storageUri: gs://kfserving-examples/models/torchserve/image_classifier/v1
  transformer:
    containers:
      - image: kserve/image-transformer:latest
        name: kserve-container
        command:
          - "python"
          - "-m"
          - "model"
        args:
          - --model_name
          - mnist
        resources:
          limits:
            cpu: 1
            memory: 2Gi
          requests:
            cpu: 100m
            memory: 512Mi

tip

STORAGE_URI is a built-in environment variable used to inject the storage initializer for custom containers, similar to the storageUri field for prepackaged predictors. The downloaded artifacts are stored under /mnt/models.

Apply the InferenceService YAML:

kubectl apply -f transformer.yaml

Expected output

inferenceservice.serving.kserve.io/torch-transformer created

Run a Prediction

First, download the request input payload.

Then, determine the ingress IP and ports and set INGRESS_HOST and INGRESS_PORT according to your cluster configuration.

SERVICE_NAME=torch-transformer
MODEL_NAME=mnist
INPUT_PATH=@./image.json
SERVICE_HOSTNAME=$(kubectl get inferenceservice $SERVICE_NAME -o jsonpath='{.status.url}' | cut -d "/" -f 3)

curl -v -H "Host: ${SERVICE_HOSTNAME}" -H "Content-Type: application/json" -d $INPUT_PATH http://${INGRESS_HOST}:${INGRESS_PORT}/v1/models/$MODEL_NAME:predict

Expected output

> POST /v1/models/mnist:predict HTTP/1.1
> Host: torch-transformer.default.example.com
> User-Agent: curl/7.73.0
> Accept: */*
> Content-Length: 401
> Content-Type: application/x-www-form-urlencoded
>
* upload completely sent off: 401 out of 401 bytes
Handling connection for 8080
* Mark bundle as not supporting multiuse
< HTTP/1.1 200 OK
< content-length: 20
< content-type: application/json; charset=UTF-8
< date: Tue, 12 Jan 2021 09:52:30 GMT
< server: istio-envoy
< x-envoy-upstream-service-time: 83
<
* Connection #0 to host localhost left intact
{"predictions": [2]}

Deploy the InferenceService Calling Predictor with gRPC Protocol

Compared to REST, gRPC is faster due to the tight packing of the Protocol Buffer and the use of HTTP/2. In many cases, gRPC is a more efficient communication protocol between Transformer and Predictor, especially when transmitting large tensors between them.

Create InferenceService

Create the InferenceService with the following YAML, which includes a Transformer and a Triton Predictor. As KServe by default uses TorchServe serving runtime for PyTorch models, you need to override the serving runtime to kserve-tritonserver for using the gRPC protocol. The transformer calls out to the predictor with V2 gRPC Protocol by specifying the --protocol argument.

apiVersion: serving.kserve.io/v1beta1
kind: InferenceService
metadata:
  name: torch-grpc-transformer
spec:
  predictor:
    model:
      modelFormat: 
        name: pytorch
      storageUri: gs://kfserving-examples/models/torchscript
      runtime: kserve-tritonserver
      runtimeVersion: 20.10-py3
      ports:
      - name: h2c
        protocol: TCP
        containerPort: 9000
  transformer:
    containers:
    - image: kserve/image-transformer:latest
      name: kserve-container
      command:
      - "python"
      - "-m"
      - "model"
      args:
      - --model_name
      - cifar10
      - --protocol
      - grpc-v2
      resources:
        limits:
          cpu: 1
          memory: 2Gi
        requests:
          cpu: 100m
          memory: 512Mi

Apply the InferenceService YAML:

kubectl apply -f grpc-transformer.yaml

Expected output

inferenceservice.serving.kserve.io/torch-grpc-transformer created

Run a Prediction

First, download the request input payload.

Then, determine the ingress IP and ports and set INGRESS_HOST and INGRESS_PORT according to your cluster configuration:

SERVICE_NAME=torch-grpc-transformer
MODEL_NAME=cifar10
INPUT_PATH=@./image.json
SERVICE_HOSTNAME=$(kubectl get inferenceservice $SERVICE_NAME -o jsonpath='{.status.url}' | cut -d "/" -f 3)

curl -v -H "Host: ${SERVICE_HOSTNAME}" -H "Content-Type: application/json" -d $INPUT_PATH http://${INGRESS_HOST}:${INGRESS_PORT}/v1/models/$MODEL_NAME:predict

Expected output

*   Trying ::1...
* TCP_NODELAY set
* Connected to localhost (::1) port 8080 (#0)
> POST /v1/models/cifar10:predict HTTP/1.1
> Host: torch-transformer.default.example.com
> User-Agent: curl/7.64.1
> Accept: */*
> Content-Length: 3394
> Content-Type: application/x-www-form-urlencoded
> Expect: 100-continue
>
Handling connection for 8080
< HTTP/1.1 100 Continue
* We are completely uploaded and fine
< HTTP/1.1 200 OK
< content-length: 222
< content-type: application/json; charset=UTF-8
< date: Thu, 03 Feb 2022 01:50:07 GMT
< server: istio-envoy
< x-envoy-upstream-service-time: 73
<
* Connection #0 to host localhost left intact
{"predictions": [[-1.192867636680603, -0.35750141739845276, -2.3665435314178467, 3.9186441898345947, -2.0592284202575684, 4.091977119445801, 0.1266237050294876, -1.8284690380096436, 2.628898859024048, -4.255198001861572]]}* Closing connection 0

Performance Comparison Between gRPC and REST

From the following latency statistics of both transformer and predictor, you can see that the transformer-to-predictor call takes longer for REST than gRPC (92ms vs 55ms). REST takes more time serializing and deserializing the 3*32*32 shape tensor, while with gRPC it is transmitted as tightly packed numpy array serialized bytes.

REST latency:

# from REST v1 transformer log
2023-01-09 07:15:55.263 79476 root INFO [__call__():128] requestId: N.A., preprocess_ms: 6.083965302, explain_ms: 0, predict_ms: 92.653036118, postprocess_ms: 0.007867813
# from REST v1 predictor log
2023-01-09 07:16:02.581 79402 root INFO [__call__():128] requestId: N.A., preprocess_ms: 13.532876968, explain_ms: 0, predict_ms: 48.450231552, postprocess_ms: 0.006914139

gRPC latency:

# from REST v1 transformer log
2023-01-09 07:27:52.172 79715 root INFO [__call__():128] requestId: N.A., preprocess_ms: 2.567052841, explain_ms: 0, predict_ms: 55.0532341, postprocess_ms: 0.101804733
# from gPPC v2 predictor log
2023-01-09 07:27:52.171 79711 root INFO [__call__():128] requestId: , preprocess_ms: 0.067949295, explain_ms: 0, predict_ms: 51.237106323, postprocess_ms: 0.049114227

Server-Side TLS for Transformers

KServe transformers can serve HTTPS natively using TLS certificates. When SSL certificate and key files are provided, the server automatically switches from HTTP (port 8080) to HTTPS (port 8443).

Configuration

TLS is configured via command-line arguments or environment variables:

Argument	Environment Variable	Description
--ssl_certfile	KSERVE_TLS_CERT_FILE	Path to the SSL certificate file
--ssl_keyfile	KSERVE_TLS_KEY_FILE	Path to the SSL private key file

When both are set and --http_port is not explicitly overridden, the server automatically switches the listen port from 8080 to 8443.

Providing Certificates to the Transformer

TLS certificate and key files are provided to the transformer container by mounting a Kubernetes Secret as a volume.

Create a secret from your certificate and key files:

kubectl create secret tls transformer-tls-secret \
  --cert=/path/to/tls.crt \
  --key=/path/to/tls.key

Then mount the secret and pass the file paths via command-line arguments in the transformer container spec:

    containers:
    - image: kserve/image-transformer:latest
      name: kserve-container
      args:
        - --ssl_certfile
        - /etc/tls/tls.crt
        - --ssl_keyfile
        - /etc/tls/tls.key
      ports:
        - containerPort: 8443
      volumeMounts:
        - name: tls-certs
          mountPath: /etc/tls
          readOnly: true
    volumes:
      - name: tls-certs
        secret:
          secretName: transformer-tls-secret

Alternatively, you can set the KSERVE_TLS_CERT_FILE and KSERVE_TLS_KEY_FILE environment variables instead of using command-line arguments.

Port Behavior

SSL Config	--http_port	Listen Port
Not set	Not set	8080 (HTTP)
Set	Not set	8443 (HTTPS)
Set	Overridden	Custom port (HTTPS)

Readiness Probes

When using server-side TLS, the container must advertise the HTTPS port via containerPort in the pod spec. The default readiness probe uses containerPort for TCP health checks. If no port is defined, it falls back to 8080, which will fail if the server is listening on 8443.

For transformers with TLS enabled, set the port in the InferenceService spec:

spec:
  transformer:
    containers:
    - image: my-transformer:latest
      ports:
      - containerPort: 8443

Client-Side TLS (Transformer to Predictor)

When the predictor serves HTTPS, the transformer must call it over TLS. Enable this with the --predictor_use_ssl argument so the transformer issues HTTPS requests to the predictor.

To verify the predictor's certificate — for example when the predictor uses a self-signed or private CA — provide a CA bundle to the transformer container through one of the following environment variables:

Environment Variable	Purpose
REQUESTS_CA_BUNDLE	Path to a PEM CA bundle used to verify the predictor's certificate
CURL_CA_BUNDLE	Fallback CA bundle path, used only when REQUESTS_CA_BUNDLE is not set

The transformer's HTTP client (httpx) reads these variables and uses the referenced bundle to verify the predictor's TLS certificate. If neither variable is set, the default system trust store is used.

note

httpx does not honor REQUESTS_CA_BUNDLE / CURL_CA_BUNDLE automatically, so KServe reads them explicitly. Point the variable at the mounted CA certificate:

    containers:
    - image: kserve/image-transformer:latest
      name: transformer-container
      args:
        - --predictor_use_ssl
      env:
        - name: REQUESTS_CA_BUNDLE
          value: /etc/tls/predictor-ca/ca.crt
      volumeMounts:
        - name: predictor-ca
          mountPath: /etc/tls/predictor-ca
          readOnly: true
    volumes:
      - name: predictor-ca
        secret:
          secretName: predictor-ca-bundle

Credential Forwarding to the Predictor

When the predictor is protected by an auth proxy (for example kube-rbac-proxy), the transformer must forward the caller's credentials (the Authorization header) to the predictor so the proxy can validate the original caller's identity. The predictor authorizes the end user, not the transformer's service account — the transformer is a pass-through for credentials.

SDK-based transformers (automatic)

If your transformer extends kserve.Model and relies on the built-in predict() method (i.e. you override preprocess and/or postprocess but not predict), the KServe SDK automatically forwards the following headers from the incoming request to the predictor:

Header	Purpose
Authorization	Caller credentials for predictor auth
x-request-id	Request tracing
x-b3-traceid	Distributed tracing (B3/Zipkin)

No additional configuration is required:

from kserve import Model, InferRequest, InferResponse

class MyTransformer(Model):
    def __init__(self, name: str):
        super().__init__(name)
        self.ready = True

    def preprocess(self, payload: InferRequest, headers=None) -> InferRequest:
        # Authorization header is forwarded automatically when the SDK
        # calls predict() on the predictor.
        return payload

    def postprocess(self, result: InferResponse, headers=None) -> InferResponse:
        return result

Custom transformers (manual forwarding)

If your transformer overrides predict() or makes its own HTTP calls to the predictor, you are responsible for forwarding the Authorization header yourself: accept the headers parameter, extract the header, and include it in the request to the predictor.

import httpx
from kserve import Model

class MyCustomTransformer(Model):
    def __init__(self, name: str):
        super().__init__(name)
        self.ready = True

    async def predict(self, payload, headers=None, response_headers=None):
        predict_headers = {"Content-Type": "application/json"}

        # Forward the Authorization header for auth-enabled predictors.
        if headers and "authorization" in headers:
            predict_headers["authorization"] = headers["authorization"]

        async with httpx.AsyncClient() as client:
            response = await client.post(
                f"http://{self.predictor_host}/v1/models/{self.name}:predict",
                json=payload,
                headers=predict_headers,
            )
            return response.json()

At minimum, forward the Authorization header. For full observability compatibility, also forward x-request-id and x-b3-traceid.

note

Do not forward the incoming Host header — it contains the transformer's hostname, not the predictor's. When auth is enabled, the predictor's kube-rbac-proxy validates the bearer token in the Authorization header against Kubernetes RBAC.

Transformer-Specific Command Line Arguments

• --predictor_protocol: The protocol used to communicate with the predictor. The available values are "v1", "v2" and "grpc-v2". The default value is "v1".
• --predictor_use_ssl: Whether to use secure SSL when communicating with the predictor. The default value is "false".
• --predictor_request_timeout_seconds: The timeout seconds for the request sent to the predictor. The default value is 600 seconds.
• --predictor_request_retries: The number of retries for the request sent to the predictor. The default value is 0.
• --enable_predictor_health_check: The Transformer will perform readiness check for the predictor in addition to its health check. By default, it is disabled.