Introduction to Production Inference

Duration: 45 min

Introduction to Production Inference

Duration: 45 min

What is Production Inference?

Production inference is deploying trained ML models to serve real-time predictions at scale. Unlike training which happens once, inference happens continuously—every prediction request from a user. Production inference must be fast (low latency), reliable (high availability), cost-effective, and maintainable.

Training and inference are different problems. Training is batch processing on powerful hardware optimized for GPUs. Inference serves individual or batched requests from many users with strict latency constraints. A user doesn't want to wait 30 seconds for a recommendation.

Production inference systems serve diverse workloads: recommendations (Netflix), fraud detection (PayPal), autonomous vehicles (Tesla), and content filtering (YouTube). Each has different latency/throughput tradeoffs. Understanding these patterns is critical for building scalable systems.

Serving Patterns

Real-time Serving: Low-latency response to individual requests. User waits for answer.

Request → Model → Prediction (< 100ms target)

Use cases: product recommendations, credit decisions, personalization. Needs fast hardware, caching, low model complexity.

from fastapi import FastAPI
from pydantic import BaseModel
import numpy as np
app = FastAPI()
Load model once

model = load_model('recommendation_model.pkl')
class UserRequest(BaseModel):
    user_id: int
    context: dict
@app.post("/predict")
def predict(request: UserRequest):
    features = extract_features(request.user_id, request.context)
    prediction = model.predict(features)
    return {"recommendation": prediction.tolist()}

Batch Serving: Process many requests together periodically.

Wait → Collect requests → Model → Predictions → Async results

Use cases: overnight reporting, data processing pipelines, cost optimization. Trades latency for efficiency.

def batch_predict(requests, batch_size=100):
    """Process requests in batches"""
    results = []
    
    for i in range(0, len(requests), batch_size):
        batch = requests[i:i+batch_size]
        features = [extract_features(r) for r in batch]
        predictions = model.predict(features)
        results.extend(predictions)
    
    return results

Streaming Serving: Continuous processing of unbounded data streams.

Stream Input → Windowed Aggregation → Model → Stream Output

Use cases: real-time analytics, anomaly detection, live recommendations. Requires stream processing frameworks (Kafka, Spark Streaming).

Embedded Serving: Model runs on device (phone, IoT, edge).

Device → Model (TFLite, ONNX) → Prediction (no network needed)

Use cases: privacy-sensitive applications, offline functionality, reducing latency. Requires model optimization (quantization, pruning).

Latency Requirements

Different applications have different latency constraints:

  • Sub-10ms: High-frequency trading, real-time bidding
  • 10-100ms: Interactive apps (autocomplete, recommendations), autonomous vehicles
  • 100ms-1s: Standard web requests, chatbots
  • 1s+: Batch processing, background jobs

import time
from functools import wraps
def measure_latency(func):
    @wraps(func)
    def wrapper(args, *kwargs):
        start = time.time()
        result = func(args, *kwargs)
        latency_ms = (time.time() - start) * 1000
        
        # Log if SLA violated
        sla_ms = 100
        if latency_ms > sla_ms:
            print(f"WARNING: {func.__name__} took {latency_ms:.1f}ms (SLA: {sla_ms}ms)")
        
        return result
    return wrapper
@measure_latency
def predict(features):
    return model.predict(features)

Measure actual latency in production. Publish metrics (p50, p95, p99). Alert when SLAs are violated.

Batching

Batching improves throughput dramatically by amortizing overhead:

class BatchPredictor:
    def __init__(self, model, batch_size=32, max_wait_ms=100):
        self.model = model
        self.batch_size = batch_size
        self.max_wait_ms = max_wait_ms
        self.batch = []
        self.futures = []
        
        # Background worker thread
        import threading
        self.worker = threading.Thread(target=self._process_batches, daemon=True)
        self.worker.start()
    
    def predict(self, features):
        """Add to batch and return future"""
        import concurrent.futures
        future = concurrent.futures.Future()
        
        self.batch.append(features)
        self.futures.append(future)
        
        if len(self.batch) >= self.batch_size:
            self._flush()
        
        return future
    
    def _flush(self):
        """Process accumulated batch"""
        if not self.batch:
            return
        
        batch_array = np.array(self.batch)
        predictions = self.model.predict(batch_array)
        
        for future, pred in zip(self.futures, predictions):
            future.set_result(pred)
        
        self.batch = []
        self.futures = []
    
    def _process_batches(self):
        """Background thread: flush periodically"""
        import time
        while True:
            time.sleep(self.max_wait_ms / 1000)
            self._flush()

This collects requests, batches them, processes together. Improves throughput by 5-10x while keeping latency acceptable.

Model Optimization

Production models must be fast. Optimization techniques:

Quantization: Reduce precision (FP32 → INT8). Smaller model, faster inference, minimal accuracy loss.

import tensorflow as tf
Post-training quantization

converter = tf.lite.TFLiteConverter.from_saved_model('model_path')
converter.optimizations = [tf.lite.Optimize.DEFAULT]
quantized_model = converter.convert()
with open('model_quantized.tflite', 'wb') as f:
    f.write(quantized_model)

Pruning: Remove unimportant weights. Smaller model, faster inference.

import tensorflow_model_optimization as tfmot
Prune 80% of weights

pruning_schedule = tfmot.sparsity.keras.PolynomialDecay(
    initial_sparsity=0.0,
    final_sparsity=0.8,
    begin_step=0,
    end_step=1000
)
pruned_model = tfmot.sparsity.keras.prune_low_magnitude(
    model,
    pruning_schedule=pruning_schedule
)
pruned_model.fit(training_data)

Distillation: Train small model to mimic large model. Smaller model, faster inference, good accuracy.

Distillation: small model learns from large model

teacher = large_model()
student = small_model()
Train student to match teacher outputs

for data, labels in train_data:
    teacher_output = teacher.predict(data)
    
    # KL divergence loss between student and teacher
    with tf.GradientTape() as tape:
        student_output = student(data, training=True)
        loss = tf.keras.losses.KLD(teacher_output, student_output)
    
    gradients = tape.gradient(loss, student.trainable_variables)
    optimizer.apply_gradients(zip(gradients, student.trainable_variables))

Caching: Store predictions for common inputs.

from functools import lru_cache
@lru_cache(maxsize=10000)
def predict_cached(features_tuple):
    features = np.array(features_tuple)
    return model.predict(features)

Scaling Strategies

Horizontal Scaling: Add more servers to handle load.

Kubernetes configuration

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ml-serving
spec:
  replicas: 3  # 3 instances
  selector:
    matchLabels:
      app: ml-serving
  template:
    metadata:
      labels:
        app: ml-serving
    spec:
      containers:
      - name: ml-server
        image: ml-server:latest
        resources:
          requests:
            memory: "1Gi"
            cpu: "500m"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 10
          periodSeconds: 10
---
apiVersion: autoscaling.k8s.io/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ml-serving-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: ml-serving
  minReplicas: 1
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70

Kubernetes automatically scales based on CPU, memory, or custom metrics.

Load Balancing: Distribute requests across servers.

nginx configuration

upstream ml_backend {
    server ml1:8000 weight=1;
    server ml2:8000 weight=1;
    server ml3:8000 weight=1;
}
server {
    listen 80;
    
    location /predict {
        proxy_pass http://ml_backend;
        proxy_set_header X-Real-IP $remote_addr;
    }
}

Caching: Store frequent predictions in Redis or memcached.

import redis
cache = redis.Redis(host='localhost', port=6379)
def predict_with_cache(user_id):
    cache_key = f"prediction:{user_id}"
    
    # Check cache
    cached = cache.get(cache_key)
    if cached:
        return json.loads(cached)
    
    # Compute prediction
    prediction = model.predict(user_id)
    
    # Store in cache (1 hour TTL)
    cache.setex(cache_key, 3600, json.dumps(prediction))
    
    return prediction

Monitoring and Observability

Production systems require continuous monitoring:

from prometheus_client import Counter, Histogram, start_http_server
Metrics

prediction_count = Counter('predictions_total', 'Total predictions')
prediction_latency = Histogram('prediction_latency_seconds', 'Prediction latency')
prediction_errors = Counter('prediction_errors_total', 'Prediction errors')
@prediction_latency.time()  # Automatically measure latency
def predict(features):
    try:
        result = model.predict(features)
        prediction_count.inc()
        return result
    except Exception as e:
        prediction_errors.inc()
        raise
Start metrics server (Prometheus scrapes this)

start_http_server(8000)

Key metrics:

  • Latency: P50, P95, P99 latency
  • Throughput: Requests per second
  • Error rate: % of failed predictions
  • Resource usage: CPU, memory, GPU
  • Model performance: Accuracy drift over time

Use Prometheus + Grafana for visualization.

Complete Example: ML Service

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import numpy as np
import logging
app = FastAPI()
Load model

model = load_model('model.pkl')
logger = logging.getLogger(__name__)
class PredictionRequest(BaseModel):
    features: list
    
class PredictionResponse(BaseModel):
    prediction: float
    confidence: float
@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    try:
        features = np.array(request.features)
        
        # Validation
        if features.shape != (10,):
            raise HTTPException(status_code=400, detail="Features must have 10 values")
        
        # Predict
        prediction = model.predict([features])[0]
        confidence = get_confidence(prediction)
        
        return PredictionResponse(
            prediction=float(prediction),
            confidence=float(confidence)
        )
    
    except Exception as e:
        logger.error(f"Prediction error: {e}")
        raise HTTPException(status_code=500, detail="Prediction failed")
@app.get("/health")
async def health():
    return {"status": "healthy"}
if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

Quiz

1. What is the primary difference between training and inference workloads?

  • A) They use the same hardware
  • B) Training is batch, inference handles individual requests with latency constraints ✓
  • C) Inference is more important than training
  • D) There's no practical difference

2. Which serving pattern would you use for overnight reporting that doesn't need immediate results?

  • A) Real-time serving
  • B) Embedded serving
  • C) Batch serving ✓
  • D) Streaming serving

3. What does batching do for inference throughput?

  • A) Decreases throughput
  • B) Has no effect
  • C) Improves throughput by amortizing overhead ✓
  • D) Only works for large models

4. Which technique reduces model size and inference latency by removing unimportant weights?

  • A) Quantization
  • B) Pruning ✓
  • C) Distillation
  • D) Caching

5. What does a liveness probe in Kubernetes check?

  • A) Whether the pod is running and healthy ✓
  • B) Whether the pod is ready
  • C) Whether metrics are available
  • D) Whether cache is warm

6. Why monitor prediction latency at P95 and P99 instead of just average?

  • A) Average is misleading; percentiles show tail latency where real user impact occurs ✓
  • B) P99 is always better than P95
  • C) Monitoring multiple percentiles improves performance
  • D) They're equivalent

← Back ▶ Run in Colab