Chapter 7: Model Serving & Deployment

WIA-AI-008 Standard • Estimated reading time: 65 minutes

7.1 Serving Infrastructure Overview

Model serving transforms trained models into production-ready inference services. A robust serving infrastructure handles request routing, batching, scaling, monitoring, and failover. The choice of serving framework depends on deployment target, performance requirements, and operational constraints.

Serving Patterns

Pattern	Use Case	Latency	Throughput
Real-time	Online inference	< 100ms	100-1000 QPS
Batch	Offline processing	Seconds-hours	1M+ records
Streaming	Event processing	< 1s	10K-100K QPS
Edge	On-device inference	< 50ms	Local only

7.2 TorchServe

TorchServe is the official model serving framework for PyTorch, developed jointly with AWS.

Creating a Model Archive (MAR)

# Install TorchServe
# pip install torchserve torch-model-archiver

# Create handler (custom inference logic)
# handler.py
from ts.torch_handler.base_handler import BaseHandler
import torch

class CustomHandler(BaseHandler):
    def preprocess(self, data):
        """
        Preprocess input data
        """
        # Extract input from request
        input_data = data[0].get("data") or data[0].get("body")

        # Convert to tensor
        tensor = torch.FloatTensor(input_data)
        return tensor

    def inference(self, data):
        """
        Run inference
        """
        with torch.no_grad():
            output = self.model(data)
        return output

    def postprocess(self, data):
        """
        Postprocess output
        """
        # Convert to list for JSON serialization
        return data.tolist()

# Archive model
# torch-model-archiver --model-name resnet18 \
#                      --version 1.0 \
#                      --model-file model.py \
#                      --serialized-file resnet18.pth \
#                      --handler handler.py \
#                      --extra-files index_to_name.json \
#                      --export-path model-store/

Deploying with TorchServe

# Start TorchServe
# torchserve --start --model-store model-store --models resnet18=resnet18.mar

# Configuration file (config.properties)
"""
inference_address=http://0.0.0.0:8080
management_address=http://0.0.0.0:8081
metrics_address=http://0.0.0.0:8082

number_of_netty_threads=4
job_queue_size=100
number_of_gpu=1

default_workers_per_model=4
max_request_size=65535000
max_response_size=65535000

enable_metrics_api=true
"""

# Client request
import requests
import json

url = "http://localhost:8080/predictions/resnet18"
data = json.dumps({"data": [[1.0, 2.0, 3.0]]})
headers = {"Content-Type": "application/json"}

response = requests.post(url, data=data, headers=headers)
print(response.json())

Auto-scaling

# Register model with auto-scaling
# POST http://localhost:8081/models?url=resnet18.mar&initial_workers=2&synchronous=true

# Scale workers dynamically
# PUT http://localhost:8081/models/resnet18?min_worker=2&max_worker=8

# Kubernetes Horizontal Pod Autoscaler (HPA)
"""
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: torchserve-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: torchserve
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
"""

7.3 TensorFlow Serving

TensorFlow Serving is a high-performance serving system for machine learning models.

Serving a SavedModel

# Export TensorFlow model
import tensorflow as tf

model = tf.keras.applications.ResNet50(weights='imagenet')
model.save('models/resnet50/1')  # Version 1

# Directory structure:
# models/
#   resnet50/
#     1/
#       saved_model.pb
#       variables/

# Start TF Serving with Docker
# docker run -p 8501:8501 \
#   --mount type=bind,source=/path/to/models,target=/models \
#   tensorflow/serving \
#   --model_config_file=/models/models.config

# models.config
"""
model_config_list {
  config {
    name: 'resnet50'
    base_path: '/models/resnet50'
    model_platform: 'tensorflow'
    model_version_policy {
      all: {}
    }
  }
}
"""

Client Inference

import requests
import json
import numpy as np

# Prepare input
image = np.random.rand(1, 224, 224, 3).tolist()

# REST API
url = "http://localhost:8501/v1/models/resnet50:predict"
data = json.dumps({"signature_name": "serving_default", "instances": image})
headers = {"content-type": "application/json"}

response = requests.post(url, data=data, headers=headers)
predictions = response.json()['predictions']

# gRPC API (faster)
import grpc
from tensorflow_serving.apis import predict_pb2
from tensorflow_serving.apis import prediction_service_pb2_grpc

channel = grpc.insecure_channel('localhost:8500')
stub = prediction_service_pb2_grpc.PredictionServiceStub(channel)

request = predict_pb2.PredictRequest()
request.model_spec.name = 'resnet50'
request.model_spec.signature_name = 'serving_default'

# Set input
request.inputs['input'].CopyFrom(
    tf.make_tensor_proto(image, shape=[1, 224, 224, 3])
)

# Get prediction
result = stub.Predict(request, 10.0)  # 10 second timeout
output = tf.make_ndarray(result.outputs['output'])
print(output)

7.4 NVIDIA Triton Inference Server

Triton supports multiple frameworks (PyTorch, TensorFlow, ONNX, TensorRT) in a single serving infrastructure.

Model Repository Structure

# Directory structure
model_repository/
  resnet50_pytorch/
    config.pbtxt
    1/
      model.pt
  resnet50_onnx/
    config.pbtxt
    1/
      model.onnx
  resnet50_tensorrt/
    config.pbtxt
    1/
      model.plan

# config.pbtxt
"""
name: "resnet50_onnx"
platform: "onnxruntime_onnx"
max_batch_size: 32
input [
  {
    name: "input"
    data_type: TYPE_FP32
    dims: [ 3, 224, 224 ]
  }
]
output [
  {
    name: "output"
    data_type: TYPE_FP32
    dims: [ 1000 ]
  }
]
instance_group [
  {
    count: 2
    kind: KIND_GPU
  }
]
dynamic_batching {
  preferred_batch_size: [ 8, 16, 32 ]
  max_queue_delay_microseconds: 100
}
"""

Launching Triton

# Docker
# docker run --gpus all -p 8000:8000 -p 8001:8001 -p 8002:8002 \
#   -v /path/to/model_repository:/models \
#   nvcr.io/nvidia/tritonserver:23.10-py3 \
#   tritonserver --model-repository=/models

# Client
import tritonclient.http as httpclient
import numpy as np

# Create client
triton_client = httpclient.InferenceServerClient(url="localhost:8000")

# Prepare input
input_data = np.random.rand(1, 3, 224, 224).astype(np.float32)

# Create input object
inputs = [httpclient.InferInput("input", input_data.shape, "FP32")]
inputs[0].set_data_from_numpy(input_data)

# Create output object
outputs = [httpclient.InferRequestedOutput("output")]

# Inference
results = triton_client.infer(
    model_name="resnet50_onnx",
    inputs=inputs,
    outputs=outputs
)

# Get output
output_data = results.as_numpy("output")
print(f"Output shape: {output_data.shape}")

7.5 Serverless Deployment

AWS Lambda with ONNX Runtime

# lambda_function.py
import json
import onnxruntime as ort
import numpy as np

# Load model (reuse across invocations)
session = None

def lambda_handler(event, context):
    global session

    # Initialize model on cold start
    if session is None:
        session = ort.InferenceSession("model.onnx")

    # Parse input
    input_data = np.array(json.loads(event['body'])['data'], dtype=np.float32)

    # Run inference
    input_name = session.get_inputs()[0].name
    output_name = session.get_outputs()[0].name
    result = session.run([output_name], {input_name: input_data})[0]

    # Return response
    return {
        'statusCode': 200,
        'body': json.dumps({
            'predictions': result.tolist()
        })
    }

# Package Lambda: zip -r function.zip lambda_function.py model.onnx
# Deploy: aws lambda create-function ...

Google Cloud Functions

# main.py
import functions_framework
import torch
import json

model = None

@functions_framework.http
def predict(request):
    global model

    # Load model on cold start
    if model is None:
        model = torch.jit.load("model.pt")
        model.eval()

    # Parse input
    request_json = request.get_json()
    input_data = torch.tensor(request_json['data'])

    # Inference
    with torch.no_grad():
        output = model(input_data)

    # Response
    return json.dumps({
        'predictions': output.tolist()
    })

# Deploy: gcloud functions deploy predict --runtime python39 --trigger-http

7.6 Edge and Mobile Deployment

TensorFlow Lite on Android

// Android (Java)
import org.tensorflow.lite.Interpreter;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;

public class ModelInference {
    private Interpreter tflite;

    public ModelInference(Context context) {
        // Load model from assets
        try {
            MappedByteBuffer modelBuffer = loadModelFile(context, "model.tflite");
            tflite = new Interpreter(modelBuffer);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public float[][] predict(float[][][] input) {
        // Allocate output
        float[][] output = new float[1][10];

        // Run inference
        tflite.run(input, output);

        return output;
    }

    private MappedByteBuffer loadModelFile(Context context, String modelPath) throws IOException {
        AssetFileDescriptor fileDescriptor = context.getAssets().openFd(modelPath);
        FileInputStream inputStream = new FileInputStream(fileDescriptor.getFileDescriptor());
        FileChannel fileChannel = inputStream.getChannel();
        long startOffset = fileDescriptor.getStartOffset();
        long declaredLength = fileDescriptor.getDeclaredLength();
        return fileChannel.map(FileChannel.MapMode.READ_ONLY, startOffset, declaredLength);
    }
}

CoreML on iOS

// Swift
import CoreML
import Vision

class ModelInference {
    var model: VNCoreMLModel?

    init() {
        // Load CoreML model
        guard let mlModel = try? MyModel(configuration: MLModelConfiguration()).model else {
            fatalError("Failed to load model")
        }

        model = try? VNCoreMLModel(for: mlModel)
    }

    func predict(image: UIImage, completion: @escaping ([String: Double]?) -> Void) {
        guard let model = model else {
            completion(nil)
            return
        }

        // Create request
        let request = VNCoreMLRequest(model: model) { request, error in
            guard let results = request.results as? [VNClassificationObservation] else {
                completion(nil)
                return
            }

            // Convert to dictionary
            var predictions: [String: Double] = [:]
            for result in results {
                predictions[result.identifier] = Double(result.confidence)
            }

            completion(predictions)
        }

        // Perform inference
        guard let ciImage = CIImage(image: image) else {
            completion(nil)
            return
        }

        let handler = VNImageRequestHandler(ciImage: ciImage)
        try? handler.perform([request])
    }
}

7.7 Batch Inference

Spark-based Batch Processing

from pyspark.sql import SparkSession
from pyspark.sql.functions import pandas_udf
import pandas as pd
import torch

# Initialize Spark
spark = SparkSession.builder.appName("BatchInference").getOrCreate()

# Load data
df = spark.read.parquet("s3://data/images.parquet")

# Define UDF for inference
@pandas_udf("array")
def predict_udf(images: pd.Series) -> pd.Series:
    # Load model (broadcast to all workers)
    model = torch.jit.load("/path/to/model.pt")
    model.eval()

    # Batch inference
    results = []
    for image_batch in np.array_split(images.values, len(images) // 32):
        with torch.no_grad():
            tensor_batch = torch.FloatTensor(list(image_batch))
            predictions = model(tensor_batch)
            results.extend(predictions.tolist())

    return pd.Series(results)

# Apply UDF
predictions_df = df.withColumn("predictions", predict_udf(df["image"]))

# Save results
predictions_df.write.parquet("s3://results/predictions.parquet")

7.8 Monitoring and Observability

Metrics Collection

from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# Define metrics
REQUEST_COUNT = Counter('model_requests_total', 'Total requests', ['model', 'status'])
REQUEST_LATENCY = Histogram('model_request_latency_seconds', 'Request latency', ['model'])
ACTIVE_REQUESTS = Gauge('model_active_requests', 'Active requests', ['model'])
MODEL_VERSION = Gauge('model_version', 'Current model version', ['model'])

class MonitoredModel:
    def __init__(self, model, model_name, version):
        self.model = model
        self.model_name = model_name
        MODEL_VERSION.labels(model=model_name).set(version)

    def predict(self, input_data):
        ACTIVE_REQUESTS.labels(model=self.model_name).inc()

        start_time = time.time()
        try:
            result = self.model(input_data)
            REQUEST_COUNT.labels(model=self.model_name, status='success').inc()
            return result
        except Exception as e:
            REQUEST_COUNT.labels(model=self.model_name, status='error').inc()
            raise e
        finally:
            duration = time.time() - start_time
            REQUEST_LATENCY.labels(model=self.model_name).observe(duration)
            ACTIVE_REQUESTS.labels(model=self.model_name).dec()

# Start Prometheus endpoint
start_http_server(8000)

# Use monitored model
model = MonitoredModel(torch.jit.load("model.pt"), "resnet18", "1.0.0")
result = model.predict(input_data)

Logging and Tracing

import logging
import json
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# Configure tracing
trace.set_tracer_provider(TracerProvider())
jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

tracer = trace.get_tracer(__name__)

def predict_with_tracing(model, input_data):
    with tracer.start_as_current_span("model_inference") as span:
        span.set_attribute("model.name", "resnet18")
        span.set_attribute("model.version", "1.0.0")
        span.set_attribute("input.shape", str(input_data.shape))

        # Log request
        logger.info(json.dumps({
            "event": "inference_request",
            "model": "resnet18",
            "input_shape": list(input_data.shape)
        }))

        # Inference
        start = time.time()
        result = model(input_data)
        latency = time.time() - start

        # Log response
        logger.info(json.dumps({
            "event": "inference_response",
            "model": "resnet18",
            "latency_ms": latency * 1000,
            "output_shape": list(result.shape)
        }))

        span.set_attribute("inference.latency_ms", latency * 1000)

        return result

7.9 A/B Testing and Canary Deployments

Traffic Splitting

import random

class ABTestingRouter:
    def __init__(self, model_a, model_b, traffic_split=0.5):
        self.model_a = model_a
        self.model_b = model_b
        self.traffic_split = traffic_split

    def predict(self, input_data, user_id=None):
        # Deterministic routing based on user_id
        if user_id:
            route_to_b = hash(user_id) % 100 < (self.traffic_split * 100)
        else:
            route_to_b = random.random() < self.traffic_split

        if route_to_b:
            return self.model_b(input_data), "model_b"
        else:
            return self.model_a(input_data), "model_a"

# Usage
router = ABTestingRouter(model_v1, model_v2, traffic_split=0.1)  # 10% to v2
result, model_used = router.predict(input_data, user_id="user123")

# Canary deployment (gradual rollout)
class CanaryDeployment:
    def __init__(self, stable_model, canary_model):
        self.stable_model = stable_model
        self.canary_model = canary_model
        self.canary_percentage = 0

    def increase_canary(self, step=10):
        self.canary_percentage = min(100, self.canary_percentage + step)

    def predict(self, input_data):
        if random.random() * 100 < self.canary_percentage:
            return self.canary_model(input_data)
        else:
            return self.stable_model(input_data)

# Gradual rollout
canary = CanaryDeployment(model_stable, model_canary)
canary.increase_canary(10)  # 10% traffic
# Monitor metrics...
canary.increase_canary(20)  # 30% traffic
# Monitor metrics...
canary.increase_canary(70)  # 100% traffic

7.10 Production Best Practices

Checklist

• Health Checks: Implement liveness and readiness probes
• Timeouts: Set appropriate request and inference timeouts
• Rate Limiting: Protect against traffic spikes
• Circuit Breakers: Fail fast when downstream services are down
• Model Caching: Cache models in memory, avoid reloading
• Input Validation: Validate shapes, dtypes, ranges
• Error Handling: Return informative error messages
• Versioning: Include model version in API responses
• Graceful Shutdown: Drain connections before shutdown
• Blue-Green Deployment: Zero-downtime updates