In this final MLOps project workshop, you will synthesize everything you've learned. You will construct a complete, production-grade, containerized computer vision inference server capable of processing image payloads, performing tensor normalization, tracking sub-millisecond latency metrics, and running in an optimized Docker sandbox!
Your application will ingest image payloads as Base64 encoded strings, preprocess them to normalized tensor dimensions, execute mock classification inferences, and respond with filtered predictions and performance telemetry:
For standard deep learning classification models (like ResNet or MobileNet), inputs must be scaled and normalized to match the training distribution. Given an RGB pixel value $x \in [0, 255]$, the normalized pixel value $z$ is formulated as:
Where we utilize the ImageNet distribution standards:
mean = [0.485, 0.456, 0.406] (Red, Green, Blue channels)std = [0.229, 0.224, 0.225]APIs typically transmit binary media (like images) using Base64 encoding inside JSON payloads. In Python, you can decode and transform these payloads cleanly:
import base64
from io import BytesIO
from PIL import Image
def decode_image(base64_string: str) -> Image.Image:
# Decode string, parsing raw bytes buffer
image_bytes = base64.b64decode(base64_string)
image = Image.open(BytesIO(image_bytes))
return image.convert("RGB").resize((224, 224))
The point of this project is not the model, which is deliberately a mock, but everything around it: a validated API contract, deterministic preprocessing, a model loaded once at startup, measured latency and a reproducible container. Those are the parts that break in production. Build it in this order so each step can be tested before the next:
The normalization formula broadcasts naturally: the image array has shape (224, 224, 3) and the mean and std have shape (3,), so NumPy applies them per color channel. Most PyTorch vision models then expect channels first and a batch dimension:
import numpy as np
MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32)
STD = np.array([0.229, 0.224, 0.225], dtype=np.float32)
def preprocess(image) -> np.ndarray:
arr = np.asarray(image, dtype=np.float32) / 255.0 # (224, 224, 3), values in [0, 1]
arr = (arr - MEAN) / STD # per-channel normalization
return arr.transpose(2, 0, 1)[None, ...] # (1, 3, 224, 224)This sketch shows how the pieces fit. Fill in the details yourself for the tasks below; it uses the standard FastAPIlifespan pattern so the model is created once at startup instead of on every request:
import time
from contextlib import asynccontextmanager
import numpy as np
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
@asynccontextmanager
async def lifespan(app: FastAPI):
app.state.classes = ["Mammal", "Bird", "Reptile", "Amphibian"] # "load the model" once
yield
app = FastAPI(lifespan=lifespan)
class PayloadRequest(BaseModel):
image_base64: str
confidence_threshold: float = Field(default=0.5, ge=0.0, le=1.0)
@app.post("/predict")
def predict(req: PayloadRequest):
start = time.perf_counter()
try:
tensor = preprocess(decode_image(req.image_base64))
except Exception:
raise HTTPException(status_code=422, detail="image_base64 is not a valid image")
rng = np.random.default_rng(int(abs(float(tensor.sum())) * 1000) % (2**32))
logits = rng.normal(size=len(app.state.classes))
probs = np.exp(logits) / np.exp(logits).sum() # softmax: sums to 1
predictions = [
{"label": c, "confidence": float(p)}
for c, p in zip(app.state.classes, probs)
if p >= req.confidence_threshold
]
return {"predictions": predictions,
"latency_ms": (time.perf_counter() - start) * 1000}Seeding the random generator from the input makes the mock deterministic: the same image always gives the same output, which makes tests possible. The endpoint is a plain def rather than async def because decoding and NumPy work is CPU-bound; FastAPI runs plain functions in a thread pool so they do not block the event loop.
Docker caches each instruction as a layer and reuses it until something above it changes. Copy the dependency list and install packages before copying your source code, so editing main.py does not reinstall everything:
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
EXPOSE 8000
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]Binding to 0.0.0.0 matters: a server listening on 127.0.0.1 inside a container is unreachable from outside it. Pin dependency versions in requirements.txt so a rebuild next month produces the same image, and add a .dockerignore to keep virtual environments and caches out of the build context.
preprocess on a solid-color image and check the output shape and that values are finite.docker run -p 8000:8000 and against the local server..convert("RGB"). PNGs with transparency or grayscale images have a different channel count and break the normalization./health endpoint and a Docker HEALTHCHECK.Implement the following end-to-end components in the workspace to construct your production classifier pipeline:
decode_and_preprocess(base64_str: str) function. It must decode base64 strings into a resized $224 \times 224$ RGB image and normalize pixel coordinates based on the standard ImageNet formula.PayloadRequest that requires image_base64 (string) and a confidence_threshold (float, default=0.5).PredictionItem containing label and confidence, and a ClassificationResponse containing prediction arrays and latency_ms.lifespan manager caching mock model classes: ["Mammal", "Bird", "Reptile", "Amphibian"]./predict. Use time.perf_counter() to track execution latency around the decoding, preprocessing, and prediction segments./predict, simulate classifier confidence calculations. Filter any outputs scoring below the request's confidence_threshold, and return a ClassificationResponse.Dockerfile using the python:3.11-slim image base, optimized pip library caching layers, and launching the application.