ML Model Deployment Patterns: From Flask to ONNX

Training a model is half the work. Getting it into production where applications can call it reliably is the other half. This guide covers the three most common deployment patterns: REST API serving with Flask and FastAPI, portable inference with ONNX, and batch processing for offline predictions.

Pattern 1: Flask API

Flask is the simplest way to serve a model. It works for prototypes and internal tools where you do not need async performance.

# app.py
import pickle
import numpy as np
from flask import Flask, request, jsonify

app = Flask(__name__)

# Load model at startup
with open('model.pkl', 'rb') as f:
    model = pickle.load(f)

@app.route('/health', methods=['GET'])
def health():
    return jsonify({'status': 'healthy'})

@app.route('/predict', methods=['POST'])
def predict():
    data = request.get_json()
    
    # Validate input
    if 'features' not in data:
        return jsonify({'error': 'Missing features field'}), 400
    
    features = np.array(data['features']).reshape(1, -1)
    
    prediction = model.predict(features)[0]
    probability = model.predict_proba(features)[0].tolist()
    
    return jsonify({
        'prediction': int(prediction),
        'probabilities': probability
    })

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5000)

Saving the Model

import pickle
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification

X, y = make_classification(n_samples=1000, random_state=42)
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)

with open('model.pkl', 'wb') as f:
    pickle.dump(model, f)

Calling the Flask API

import requests

response = requests.post('http://localhost:5000/predict', json={
    'features': [0.5, -1.2, 3.4, 0.8, -0.3, 1.1, 2.2, -0.7,
                 0.9, 1.5, -2.1, 0.4, 0.6, -1.8, 2.7, 0.3,
                 -0.5, 1.9, -0.2, 0.1]
})
print(response.json())

Pattern 2: FastAPI (Production-Grade)

FastAPI adds async support, automatic validation with Pydantic, and OpenAPI documentation. It is the standard for production ML services.

# main.py
import pickle
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from typing import List
import uvicorn

app = FastAPI(title="ML Model API", version="1.0.0")

# Load model at startup
with open('model.pkl', 'rb') as f:
    model = pickle.load(f)

# Request/response schemas
class PredictRequest(BaseModel):
    features: List[float] = Field(..., min_length=20, max_length=20,
                                   description="List of 20 feature values")
    
    model_config = {"json_schema_extra": {
        "examples": [{"features": [0.5] * 20}]
    }}

class PredictResponse(BaseModel):
    prediction: int
    probabilities: List[float]
    model_version: str

class BatchPredictRequest(BaseModel):
    instances: List[List[float]]

class BatchPredictResponse(BaseModel):
    predictions: List[int]
    probabilities: List[List[float]]

MODEL_VERSION = "1.0.0"

@app.get("/health")
async def health():
    return {"status": "healthy", "model_version": MODEL_VERSION}

@app.post("/predict", response_model=PredictResponse)
async def predict(request: PredictRequest):
    features = np.array(request.features).reshape(1, -1)
    
    prediction = model.predict(features)[0]
    probability = model.predict_proba(features)[0].tolist()
    
    return PredictResponse(
        prediction=int(prediction),
        probabilities=probability,
        model_version=MODEL_VERSION
    )

@app.post("/predict/batch", response_model=BatchPredictResponse)
async def predict_batch(request: BatchPredictRequest):
    if len(request.instances) > 1000:
        raise HTTPException(status_code=400,
                            detail="Maximum 1000 instances per batch")
    
    features = np.array(request.instances)
    predictions = model.predict(features).tolist()
    probabilities = model.predict_proba(features).tolist()
    
    return BatchPredictResponse(
        predictions=[int(p) for p in predictions],
        probabilities=probabilities
    )

if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

Dockerfile for FastAPI

FROM python:3.11-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY model.pkl .
COPY main.py .

EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# requirements.txt
fastapi==0.111.0
uvicorn==0.30.1
scikit-learn==1.5.0
numpy==1.26.4

Pattern 3: ONNX Export

ONNX (Open Neural Network Exchange) is a portable format that lets you train in Python and run inference in any runtime: C++, Java, JavaScript, or mobile. It is also significantly faster than pickle-based serving for many models.

# Export a scikit-learn model to ONNX
from skl2onnx import convert_sklearn
from skl2onnx.common.data_types import FloatTensorType
from sklearn.ensemble import RandomForestClassifier
from sklearn.datasets import make_classification
import numpy as np

# Train model
X, y = make_classification(n_samples=1000, n_features=20, random_state=42)
model = RandomForestClassifier(n_estimators=100, random_state=42)
model.fit(X, y)

# Define input type
initial_type = [('features', FloatTensorType([None, 20]))]

# Convert to ONNX
onnx_model = convert_sklearn(model, initial_types=initial_type)

# Save
with open('model.onnx', 'wb') as f:
    f.write(onnx_model.SerializeToString())

print("Model exported to ONNX")

Running ONNX Inference

import onnxruntime as ort
import numpy as np

# Load ONNX model
session = ort.InferenceSession('model.onnx')

# Get input/output names
input_name = session.get_inputs()[0].name
output_names = [o.name for o in session.get_outputs()]

# Run inference
sample = np.random.randn(1, 20).astype(np.float32)
results = session.run(output_names, {input_name: sample})

prediction = results[0][0]
probabilities = results[1][0]
print(f"Prediction: {prediction}")
print(f"Probabilities: {probabilities}")

ONNX Performance Comparison

import time

# Pickle-based inference
start = time.time()
for _ in range(10000):
    model.predict(sample)
pickle_time = time.time() - start

# ONNX inference
start = time.time()
for _ in range(10000):
    session.run(output_names, {input_name: sample})
onnx_time = time.time() - start

print(f"Pickle: {pickle_time:.3f}s for 10K predictions")
print(f"ONNX:   {onnx_time:.3f}s for 10K predictions")
print(f"Speedup: {pickle_time / onnx_time:.1f}x")

Pattern 4: Batch Inference

Not every prediction needs to happen in real time. Recommendation scores, daily churn predictions, and credit risk scores are often computed in batch and stored for lookup.

Batch Inference:
[Scheduler] -> [Load data] -> [Run model] -> [Store results] -> [Serve from DB]
- Runs on a schedule (hourly, daily)
- Processes all records at once
- Results cached in DB/cache
- Lower latency for lookups

Real-Time Inference:
[Client] -> [API] -> [Model] -> [Response]
- On-demand prediction
- Single record at a time
- Higher per-request latency
- Always up-to-date

# batch_predict.py
import pandas as pd
import pickle
import sqlite3
from datetime import datetime

def run_batch_prediction():
    # Load model
    with open('model.pkl', 'rb') as f:
        model = pickle.load(f)
    
    # Load data to score
    df = pd.read_csv('customers.csv')
    features = df[['feature_1', 'feature_2', 'feature_3']].values
    
    # Run predictions
    predictions = model.predict(features)
    probabilities = model.predict_proba(features)[:, 1]
    
    # Store results
    results = pd.DataFrame({
        'customer_id': df['customer_id'],
        'prediction': predictions,
        'probability': probabilities,
        'scored_at': datetime.now().isoformat()
    })
    
    # Write to database
    conn = sqlite3.connect('predictions.db')
    results.to_sql('churn_scores', conn, if_exists='replace', index=False)
    conn.close()
    
    print(f"Scored {len(results)} customers at {datetime.now()}")

if __name__ == '__main__':
    run_batch_prediction()

When to Use Batch vs Real-Time

Model Versioning

Track which model version produced each prediction for debugging and rollback.

import hashlib
import json

def get_model_metadata(model, model_path):
    """Generate model metadata for versioning."""
    with open(model_path, 'rb') as f:
        model_hash = hashlib.sha256(f.read()).hexdigest()[:12]
    
    return {
        'model_type': type(model).__name__,
        'model_hash': model_hash,
        'n_features': model.n_features_in_,
        'classes': model.classes_.tolist(),
    }

metadata = get_model_metadata(model, 'model.pkl')
print(json.dumps(metadata, indent=2))

Key Takeaways

Start with FastAPI for real-time serving: it handles validation, async requests, and auto-generates documentation. Export to ONNX when you need portable or faster inference. Use batch inference when predictions can be pre-computed and freshness requirements are relaxed. Always include health checks, input validation, model versioning, and error handling. Containerize with Docker for reproducible deployments. The deployment pattern you choose should match your latency requirements and infrastructure constraints.