TensorFlow Model Deployment

Real-World Scenario

A medical imaging team trains a tumor detection model in Python. The model needs to be served via REST API to a React frontend, deployed to Android tablets, and monitored in production. TF Serving handles the REST API with automatic request batching. TFLite handles the Android deployment at 1/4 the model size.

Saving Models in Different Formats

import tensorflow as tf
from tensorflow import keras
import numpy as np

# Build and train a simple model
model = keras.Sequential([
    keras.layers.Dense(128, activation="relu", input_shape=(20,)),
    keras.layers.Dropout(0.2),
    keras.layers.Dense(64, activation="relu"),
    keras.layers.Dense(1, activation="sigmoid"),
])
model.compile(optimizer="adam", loss="binary_crossentropy", metrics=["accuracy"])

# Dummy training
X = np.random.randn(1000, 20).astype("float32")
y = (X[:, 0] > 0).astype("float32")
model.fit(X, y, epochs=3, verbose=0)

# ── Format 1: SavedModel (recommended for production) ──────────────────────
model.save("./saved_model/fraud_detector/1")  # version number in path
print("SavedModel saved")

# Load and verify
loaded = tf.saved_model.load("./saved_model/fraud_detector/1")
infer  = loaded.signatures["serving_default"]
result = infer(tf.constant(X[:5]))
print(f"SavedModel output keys: {list(result.keys())}")

# ── Format 2: .keras (recommended for training checkpoints) ───────────────
model.save("./fraud_detector.keras")
loaded_keras = keras.models.load_model("./fraud_detector.keras")
print(f"Keras model loaded: {loaded_keras.predict(X[:3], verbose=0).flatten()}")

# ── Format 3: Weights only (smallest, needs the architecture) ─────────────
model.save_weights("./weights_only.weights.h5")
# To restore: create the same architecture, then load weights
model_copy = keras.Sequential([
    keras.layers.Dense(128, activation="relu", input_shape=(20,)),
    keras.layers.Dropout(0.2),
    keras.layers.Dense(64, activation="relu"),
    keras.layers.Dense(1, activation="sigmoid"),
])
model_copy.load_weights("./weights_only.weights.h5")
print("Weights loaded")

Converting to TFLite for Mobile/Edge

import tensorflow as tf
import numpy as np

# Load a SavedModel and convert to TFLite
converter = tf.lite.TFLiteConverter.from_saved_model("./saved_model/fraud_detector/1")

# Optimization: quantize weights from float32 → int8 (4x size reduction)
converter.optimizations = [tf.lite.Optimize.DEFAULT]

# Optional: full integer quantization (fastest on edge hardware)
def representative_dataset():
    """Calibration data for quantization — use real training samples."""
    for _ in range(100):
        yield [np.random.randn(1, 20).astype("float32")]

converter.representative_dataset = representative_dataset
converter.target_spec.supported_ops = [tf.lite.OpsSet.TFLITE_BUILTINS_INT8]
converter.inference_input_type  = tf.float32  # keep float32 I/O for convenience
converter.inference_output_type = tf.float32

tflite_model = converter.convert()

with open("fraud_detector.tflite", "wb") as f:
    f.write(tflite_model)

import os
orig_size  = os.path.getsize("./fraud_detector.keras") / 1024
tflite_size = os.path.getsize("./fraud_detector.tflite") / 1024
print(f"Original .keras: {orig_size:.1f} KB")
print(f"TFLite (int8):   {tflite_size:.1f} KB  ({orig_size/tflite_size:.1f}x smaller)")

# Run inference with TFLite interpreter
interpreter = tf.lite.Interpreter(model_path="fraud_detector.tflite")
interpreter.allocate_tensors()

input_details  = interpreter.get_input_details()
output_details = interpreter.get_output_details()

# Single prediction
test_input = np.random.randn(1, 20).astype("float32")
interpreter.set_tensor(input_details[0]["index"], test_input)
interpreter.invoke()
output = interpreter.get_tensor(output_details[0]["index"])
print(f"TFLite prediction: {output[0][0]:.4f}")

Serving with FastAPI

# app.py — production TF model server
import tensorflow as tf
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import time
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Fraud Detection API")

# Load once at startup
MODEL = None

@app.on_event("startup")
def load():
    global MODEL
    MODEL = keras.models.load_model("./fraud_detector.keras")
    # Warm up (first call is slow due to TF graph construction)
    MODEL.predict(np.zeros((1, 20), dtype="float32"), verbose=0)
    logger.info("Model loaded and warmed up")


class FraudRequest(BaseModel):
    features: list[float]   # 20 numerical features

class FraudResponse(BaseModel):
    fraud_probability: float
    is_fraud:          bool
    latency_ms:        float


@app.post("/predict", response_model=FraudResponse)
def predict(req: FraudRequest):
    if len(req.features) != 20:
        raise HTTPException(status_code=422, detail="Exactly 20 features required")

    t0 = time.perf_counter()
    X  = np.array([req.features], dtype="float32")
    prob = float(MODEL.predict(X, verbose=0)[0, 0])
    ms   = (time.perf_counter() - t0) * 1000

    return FraudResponse(
        fraud_probability=round(prob, 4),
        is_fraud=prob >= 0.5,
        latency_ms=round(ms, 2),
    )

@app.post("/predict/batch", response_model=list[FraudResponse])
def predict_batch(requests: list[FraudRequest]):
    if len(requests) > 1000:
        raise HTTPException(status_code=422, detail="Max 1000 per batch")

    t0   = time.perf_counter()
    X    = np.array([r.features for r in requests], dtype="float32")
    probs = MODEL.predict(X, verbose=0, batch_size=256).flatten()
    ms   = (time.perf_counter() - t0) * 1000

    return [
        FraudResponse(
            fraud_probability=round(float(p), 4),
            is_fraud=float(p) >= 0.5,
            latency_ms=round(ms / len(requests), 3),
        )
        for p in probs
    ]

@app.get("/health")
def health():
    return {"status": "ok", "model": "fraud_detector_v1"}

# Run: uvicorn app:app --host 0.0.0.0 --port 8000 --workers 2

TF Serving with Docker

# docker-compose.yml — TF Serving deployment
version: "3.8"
services:
  tf-serving:
    image: tensorflow/serving:latest
    ports:
      - "8501:8501"   # REST API
      - "8500:8500"   # gRPC
    volumes:
      - ./saved_model:/models/fraud_detector
    environment:
      MODEL_NAME: fraud_detector
    command: >
      --model_config_file=/models/fraud_detector/models.config
      --allow_version_labels_for_unavailable_models
    restart: unless-stopped

# client.py — calling TF Serving REST API
import requests
import numpy as np

TF_SERVING_URL = "http://localhost:8501/v1/models/fraud_detector:predict"

def predict_tfserving(features: list[float]) -> float:
    payload = {"instances": [features]}
    response = requests.post(TF_SERVING_URL, json=payload, timeout=5)
    response.raise_for_status()
    predictions = response.json()["predictions"]
    return predictions[0][0]

# Batch prediction
batch = np.random.randn(10, 20).tolist()
payload = {"instances": batch}
response = requests.post(TF_SERVING_URL, json=payload, timeout=10)
predictions = response.json()["predictions"]
print(f"Batch predictions: {[round(p[0], 4) for p in predictions]}")

# Check model status
status = requests.get("http://localhost:8501/v1/models/fraud_detector").json()
print(f"Model status: {status}")