Learn AI Series (#136) - Mini Project: Production AI Platform

scipio(71)

Published in

StemSocial

Words

3398

Reading

16 min

Listen

Play

Learn AI Series (#136) - Mini Project: Production AI Platform

What will I learn

How to wire the whole production arc from episodes #117-135 into ONE running system, not nineteen loose parts;
a file-based model registry that versions models and promotes them from staging to production (the pattern MLflow gives you, minus the install);
experiment tracking baked straight into the training loop, so every model that lands in the registry carries its own metrics and config;
a prediction server (FastAPI) that loads whatever version is marked "production" and logs every single prediction it makes;
drift detection that watches those logs and shouts when reality stops matching the baseline;
an A/B router that splits traffic between versions and keeps score, plus a cost tracker so the money never hides in a cloud invoice ;-)

Requirements

A working modern computer running macOS, Windows or Ubuntu;
An installed Python 3(.10+) distribution with torch, fastapi, pydantic, scikit-learn and numpy (pip install torch fastapi pydantic scikit-learn numpy) -- no GPU needed, everything here runs happily on a laptop CPU;
You've been through the whole Production AI arc -- especially #117 (ML System Design), #119 (Experiment Tracking), #121 (Model Serving), #123 (Monitoring) and last episode's #135 on teams and process, whose three exercises we settle right below.

Difficulty

Beginner

Curriculum (of the `Learn AI Series`):

Learn AI Series (#136) - Mini Project: Production AI Platform

I ended episode #135 with a dare. After a hundred-odd episodes of math, models, pipelines, infrastructure and the humans who wire it all together, I asked whether you could take everything and stand up a single, coherent, end-to-end system -- data in one side, a monitored, documented, retrainable model serving predictions out the other. Not a notebook. A platform.

Today we build exactly that. Not a toy -- a platform with the same architectural bones real companies use, just shrunk down to run on one machine so nothing distracts from the shape of the thing. By the end you'll have training with experiment tracking, a model registry, a REST serving endpoint, drift monitoring, A/B testing between versions, and cost tracking. All talking to each other.

Having said that, let me be honest about scope up front: this is not production-hardened. That takes a team and months (as episode #135 hammered home, productionization costs 2-5x more than the experiment it follows). The goal here is to feel the full lifecycle in one sitting, so the connections stop being abstract. Here we go ;-)

Solutions to episode #135's exercises

Homework first, as always -- full code, no hand-waving.

Exercise 1 -- grow the lifecycle tracker a spine. The task was to extend ModelLifecycle so it refuses illegal stage transitions, still allows iterating backwards to "data" or "experiment", adds a days_in_stage() helper, and flags anything stuck in "experiment" for more than 60 days.

from dataclasses import dataclass, field
from datetime import datetime

STAGES = ["framing", "data", "experiment", "production", "deployed", "retired"]

@dataclass
class ModelLifecycle:
    model_name: str
    problem_statement: str
    owner: str
    created: datetime = field(default_factory=datetime.now)
    stage: str = "framing"
    stage_since: datetime = field(default_factory=datetime.now)
    notes: list = field(default_factory=list)

    def _legal(self, target):
        # iterating backward to data/experiment is legal -- that IS the loop
        if target in ("data", "experiment"):
            return True
        # otherwise you may only step exactly one stage forward
        return STAGES.index(target) == STAGES.index(self.stage) + 1

    def advance(self, new_stage, note=""):
        if new_stage not in STAGES:
            raise ValueError(f"unknown stage: {new_stage}")
        if not self._legal(new_stage):
            raise ValueError(f"illegal transition {self.stage} -> {new_stage}")
        self.notes.append(f"[{datetime.now():%Y-%m-%d}] {self.stage} -> {new_stage}: {note}")
        self.stage, self.stage_since = new_stage, datetime.now()

    def days_in_stage(self):
        return (datetime.now() - self.stage_since).days

    def health_check(self):
        if self.stage == "experiment" and self.days_in_stage() > 60:
            return f"STUCK: {self.model_name} sat in 'experiment' for {self.days_in_stage()} days"
        return "ok"

m = ModelLifecycle("churn-v3", "Predict 30-day churn", owner="ml-team")
m.advance("data"); m.advance("experiment")
try:
    m.advance("deployed")          # nonsense -- skips production
except ValueError as e:
    print("blocked:", e)
m.advance("production"); m.advance("experiment", "found a leak, iterating")  # legal backstep
print(m.health_check())

Why is "stuck in experiment" a smell worth alerting on? Because a model that never leaves experiment is either solving a problem nobody actually needs shipped, or it is quietly failing to clear the production bar and nobody on the team wants to say so out loud -- and in both cases the sunk cost keeps quietly compounding while the alert stays silent.

Exercise 2 -- write your own model card, then break it. Fill a real model_card, validate it, then empty the limitations list and confirm the guard rejects it. I'm carding the churn tree from last episode.

REQUIRED = ["model_name", "version", "training_data",
            "evaluation", "intended_use", "limitations"]

def validate_card(card):
    missing = [k for k in REQUIRED if not card.get(k)]
    if missing:
        raise ValueError(f"model card incomplete -- missing: {missing}")
    if not card["limitations"]:
        raise ValueError("a model card with an empty 'limitations' list is a lie")
    print(f"model card OK: {card['model_name']} v{card['version']}")

my_card = {
    "model_name": "churn-predictor-v2",
    "version": "2.1.0",
    "training_data": {"source": "internal CRM export, 2 years", "size": "245k x 18"},
    "evaluation": {"auc_roc": 0.847, "split": "temporal (Jan-Mar 2026)"},
    "intended_use": "Prioritize retention outreach for at-risk customers",
    "limitations": [
        "Weak on accounts with < 3 months of history",
        "Consumer-only -- NEVER validated on enterprise accounts",
        "Degrades if the pricing structure changes",
    ],
}
validate_card(my_card)                 # passes

my_card["limitations"] = []            # break it on purpose
try:
    validate_card(my_card)
except ValueError as e:
    print("rejected:", e)

The limitation I would not have written down without the card forcing me: "never validated on enterprise accounts". And the reason that omission bites a future user -- a sales lead sees a shiny 0.847 AUC, points the model at their enterprise book, and starts acting on churn scores that were never meaningful for that segment. The card is the only thing standing between that number and that mistake.

Exercise 3 -- simulate feedback-loop debt. Build a recommender that always pushes the current top-3 most-clicked items, run 20 rounds where clicks come mostly from what was shown, and watch diversity collapse toward a handful of eternal winners.

import random
from collections import Counter

CATALOG = list(range(20))                      # 20 items, all equal to start
clicks = Counter({i: 1 for i in CATALOG})
EPSILON = 0.0                                  # 0.0 = pure exploitation (the trap)

def recommend(clicks, k=3):
    return [item for item, _ in clicks.most_common(k)]

diversity = []
for _round in range(20):
    recs = recommend(clicks)
    surfaced = set(recs)
    for _ in range(50):                        # 50 users per round
        if random.random() < EPSILON:
            item = random.choice(CATALOG)      # deliberate exploration
        else:
            item = random.choice(recs)         # click what we were shown
        clicks[item] += 1
        surfaced.add(item)
    diversity.append(len(surfaced))

print("distinct items surfaced per round:", diversity)
covered = sum(1 for v in clicks.values() if v > 1)
print(f"catalog coverage at the end: {covered}/{len(CATALOG)}")

Run it and the diversity list slides toward 3 -- the system eats its own tail. The cheapest intervention? Set EPSILON to a small non-zero value, a few percent of pure random exploration. That is the epsilon-greedy trick straight out of episode #103's multi-armed bandits, and it keeps the recommender sampling the long tail in stead of strangling the catalogue down to three winners forever.

Architecture overview

Right, on to the build. Our platform has five components, and the whole lesson lives in how they connect:

Registry -- stores model versions with metadata and manages promotion;
Trainer -- trains models with experiment tracking, then registers them;
Server -- serves predictions via a REST API, logging each one;
Monitor -- watches those logs and detects drift;
Router -- A/B tests between versions and keeps score.

We'll use a plain tabular classification problem so the ML itself stays boring -- the infrastructure is the star today, not the model.

Component 1: the model registry

Before we train anything we need somewhere to put the results. A model registry stores trained models alongside their metadata -- version, metrics, training config, and a status that moves through staging -> production -> retired.

import json, torch
from datetime import datetime
from pathlib import Path

class ModelRegistry:
    """File-based model registry. Production systems use MLflow or similar."""
    def __init__(self, base_dir="./model_registry"):
        self.base_dir = Path(base_dir)
        self.base_dir.mkdir(parents=True, exist_ok=True)
        self.index_path = self.base_dir / "index.json"
        self.index = self._load_index()

    def _load_index(self):
        if self.index_path.exists():
            return json.loads(self.index_path.read_text())
        return {"models": {}}

    def _save_index(self):
        self.index_path.write_text(json.dumps(self.index, indent=2))

    def register(self, name, version, model, metrics, config):
        """Register a trained model with its metadata."""
        model_dir = self.base_dir / name / version
        model_dir.mkdir(parents=True, exist_ok=True)
        torch.save(model.state_dict(), model_dir / "model.pt")

        entry = {
            "version": version,
            "registered_at": datetime.now().isoformat(),
            "metrics": metrics,
            "config": config,
            "status": "staging",          # staging -> production -> retired
        }
        self.index["models"].setdefault(name, []).append(entry)
        self._save_index()
        print(f"Registered {name} {version} (status: staging)")
        return entry

    def promote(self, name, version):
        """Promote a version to production; the old production one retires."""
        for entry in self.index["models"].get(name, []):
            if entry["version"] == version:
                entry["status"] = "production"
            elif entry["status"] == "production":
                entry["status"] = "retired"
        self._save_index()
        print(f"Promoted {name} {version} to production")

    def get_production(self, name):
        for entry in self.index["models"].get(name, []):
            if entry["status"] == "production":
                return entry
        return None

    def load_model(self, name, version, model_class, **model_kwargs):
        model = model_class(**model_kwargs)
        path = self.base_dir / name / version / "model.pt"
        model.load_state_dict(torch.load(path, weights_only=True))
        return model

The key abstraction is this: models have versions, each version has a status, and exactly one version is "production" at a time. Promoting a new version automatically retires the old one -- that single invariant is what lets everything downstream ask one simple question ("what is live right now?") and always get one answer. Real registries (MLflow, the one Weights & Biases ships) add a database, access control and a UI, but the contract is identical.

Component 2: training with experiment tracking

The trainer wraps a bog-standard PyTorch loop with logging that feeds straight into the registry. Nothing here is new -- we built training loops back in episodes #43 and #44 -- the new part is that a finished training run does not just leave a model in memory, it leaves a registered, versioned artifact.

import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader, TensorDataset
from sklearn.datasets import make_classification
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

class TabularClassifier(nn.Module):
    def __init__(self, in_features, hidden=64, n_classes=2):
        super().__init__()
        self.net = nn.Sequential(
            nn.Linear(in_features, hidden), nn.ReLU(), nn.Dropout(0.2),
            nn.Linear(hidden, hidden), nn.ReLU(), nn.Dropout(0.2),
            nn.Linear(hidden, n_classes),
        )

    def forward(self, x):
        return self.net(x)

class Trainer:
    def __init__(self, registry: ModelRegistry):
        self.registry = registry
        self.experiments = []            # our stand-in for MLflow

    def train(self, model_name, version, config, X_train, y_train, X_val, y_val):
        experiment = {
            "model_name": model_name, "version": version, "config": config,
            "started_at": datetime.now().isoformat(), "train_losses": [],
        }
        model = TabularClassifier(
            X_train.shape[1], config.get("hidden", 64), config.get("n_classes", 2))
        optimizer = torch.optim.Adam(model.parameters(), lr=config.get("lr", 1e-3))

        train_ds = TensorDataset(
            torch.tensor(X_train, dtype=torch.float32),
            torch.tensor(y_train, dtype=torch.long))
        loader = DataLoader(train_ds, batch_size=config.get("batch_size", 64), shuffle=True)

        for _epoch in range(config.get("epochs", 20)):
            model.train()
            epoch_loss = 0.0
            for xb, yb in loader:
                loss = F.cross_entropy(model(xb), yb)
                optimizer.zero_grad(); loss.backward(); optimizer.step()
                epoch_loss += loss.item()
            experiment["train_losses"].append(epoch_loss / len(loader))

        model.eval()
        with torch.no_grad():
            logits = model(torch.tensor(X_val, dtype=torch.float32))
            preds = logits.argmax(dim=1)
            accuracy = (preds == torch.tensor(y_val)).float().mean().item()

        metrics = {"accuracy": accuracy, "final_loss": experiment["train_losses"][-1]}
        experiment["val_metrics"] = metrics
        self.experiments.append(experiment)
        self.registry.register(model_name, version, model, metrics, config)
        print(f"Training complete. Accuracy: {accuracy:.4f}")
        return model, metrics

# Synthetic data and two competing configs
X, y = make_classification(n_samples=5000, n_features=20, n_informative=12,
                           n_classes=2, random_state=42)
X = StandardScaler().fit_transform(X)
X_train, X_val, y_train, y_val = train_test_split(X, y, test_size=0.2, random_state=42)

registry = ModelRegistry()
trainer = Trainer(registry)
model_v1, metrics_v1 = trainer.train("classifier", "v1",
    {"hidden": 64,  "lr": 1e-3, "epochs": 20, "batch_size": 64, "n_classes": 2},
    X_train, y_train, X_val, y_val)
model_v2, metrics_v2 = trainer.train("classifier", "v2",
    {"hidden": 128, "lr": 5e-4, "epochs": 30, "batch_size": 32, "n_classes": 2},
    X_train, y_train, X_val, y_val)

In a real setup that self.experiments list is MLflow or W&B (episode #119 covered them properly). The principle does not change one bit though: log everything, keep it reproducible, and tie each training run to the exact registered version it produced. A metric with no config attached is a rumour, not a result.

Component 3: the prediction server

FastAPI gives us a REST endpoint. The important trick -- and this is the bit people get wrong -- is that the server does NOT know how the model was trained. It just asks the registry for whatever version is "production" and serves that.

from fastapi import FastAPI
from pydantic import BaseModel
import time

app = FastAPI(title="ML Platform - Prediction Server")

class PredictionRequest(BaseModel):
    features: list[float]

class PredictionResponse(BaseModel):
    prediction: int
    probability: float
    model_version: str
    latency_ms: float

class PredictionServer:
    def __init__(self, registry: ModelRegistry, model_name: str):
        self.registry = registry
        self.model_name = model_name
        self.model = None
        self.version = None
        self.predictions_log = []
        self.reload_model()

    def reload_model(self):
        prod = self.registry.get_production(self.model_name)
        if prod:
            self.model = self.registry.load_model(
                self.model_name, prod["version"], TabularClassifier,
                in_features=20, hidden=prod["config"].get("hidden", 64), n_classes=2)
            self.model.eval()
            self.version = prod["version"]
            print(f"Server loaded {self.model_name} {self.version}")

    def predict(self, features: list[float]) -> PredictionResponse:
        start = time.time()
        x = torch.tensor([features], dtype=torch.float32)
        with torch.no_grad():
            proba = F.softmax(self.model(x), dim=1)
            pred = int(proba.argmax(dim=1).item())
            prob = float(proba[0, pred].item())
        latency = (time.time() - start) * 1000

        self.predictions_log.append({
            "timestamp": datetime.now().isoformat(),
            "prediction": pred, "probability": prob,
            "latency_ms": latency, "model_version": self.version,
        })
        return PredictionResponse(prediction=pred, probability=prob,
                                  model_version=self.version, latency_ms=round(latency, 2))

@app.post("/predict", response_model=PredictionResponse)
def predict_endpoint(req: PredictionRequest):
    return server.predict(req.features)

Two patterns matter here. First, that registry indirection means promoting a new model needs zero code changes to the server -- you promote in the registry, call reload_model(), done. Second, EVERY prediction gets logged with its outputs, latency and model version. That log is not bookkeeping for its own sake -- it is the raw material the monitor feeds on. A server that predicts but does not log is a black box the moment it misbehaves.

Component 4: monitoring and drift detection

The monitor keeps a rolling window of recent predictions and compares it against a baseline snapshot. When the recent behaviour wanders too far from the baseline, it raises an alert. This is the simplified cousin of what we discussed in episode #123.

import numpy as np
from collections import deque

class ModelMonitor:
    """Watches predictions for drift and performance degradation."""
    def __init__(self, window_size=100, alert_threshold=0.15):
        self.window_size = window_size
        self.alert_threshold = alert_threshold
        self.baseline = None
        self.recent = deque(maxlen=window_size)
        self.alerts = []

    def set_baseline(self, predictions):
        probs = [p["probability"] for p in predictions]
        preds = [p["prediction"] for p in predictions]
        self.baseline = {
            "mean_probability": float(np.mean(probs)),
            "class1_ratio": sum(1 for p in preds if p == 1) / len(preds),
            "mean_latency": float(np.mean([p["latency_ms"] for p in predictions])),
        }
        print(f"Baseline set: mean_prob={self.baseline['mean_probability']:.3f}")

    def log_prediction(self, prediction):
        self.recent.append(prediction)
        if len(self.recent) >= self.window_size:
            self._check_drift()

    def _check_drift(self):
        if not self.baseline:
            return
        recent = list(self.recent)
        cur_mean = float(np.mean([p["probability"] for p in recent]))
        drift = abs(cur_mean - self.baseline["mean_probability"])
        if drift > self.alert_threshold:
            sev = "high" if drift > 2 * self.alert_threshold else "medium"
            self._raise(sev, "prediction_drift",
                        f"mean prob {self.baseline['mean_probability']:.3f} -> {cur_mean:.3f}")

        cur_ratio = sum(1 for p in recent if p["prediction"] == 1) / len(recent)
        if abs(cur_ratio - self.baseline["class1_ratio"]) > self.alert_threshold:
            self._raise("medium", "class_drift",
                        f"class-1 ratio {self.baseline['class1_ratio']:.3f} -> {cur_ratio:.3f}")

    def _raise(self, severity, kind, message):
        alert = {"type": kind, "severity": severity, "message": message,
                 "timestamp": datetime.now().isoformat()}
        self.alerts.append(alert)
        print(f"ALERT [{severity}]: {message}")

    def report(self):
        print("\n--- Monitor Report ---")
        print(f"Predictions in window: {len(self.recent)}")
        print(f"Alerts raised: {len(self.alerts)}")
        for a in self.alerts[-5:]:
            print(f"  [{a['severity']}] {a['type']}: {a['message']}")

Real monitoring reaches for Evidently, or Prometheus plus Grafana, or a database-backed metrics service. The concepts are the forementioned ones though: hold a baseline, compare the recent window against it, alert when they diverge. Note what we are NOT doing -- we have no ground-truth labels at serving time (you rarely do), so we watch the distribution of predictions as a proxy. When the model's outputs shift shape, something upstream changed, and that is worth a human look even before the accuracy numbers arrive weeks later.

Component 5: the A/B test router

The router splits live traffic between versions and tracks which one performs better. It is the multi-armed bandit idea from episode #103, wearing a production hat.

import random

class ABRouter:
    """Routes traffic between model versions for A/B testing."""
    def __init__(self, registry, model_name, in_features=20, n_classes=2):
        self.registry = registry
        self.model_name = model_name
        self.in_features = in_features
        self.n_classes = n_classes
        self.models = {}
        self.traffic_split = {}
        self.results = {}

    def configure(self, version_weights):
        """Set the traffic split, e.g. {'v1': 0.5, 'v2': 0.5}."""
        self.traffic_split = version_weights
        entries = self.registry.index["models"].get(self.model_name, [])
        for version in version_weights:
            if version not in self.models:
                config = next((e["config"] for e in entries if e["version"] == version), {})
                self.models[version] = self.registry.load_model(
                    self.model_name, version, TabularClassifier,
                    in_features=self.in_features,
                    hidden=config.get("hidden", 64), n_classes=self.n_classes)
                self.models[version].eval()
                self.results[version] = {"total": 0, "correct": 0}
        print(f"A/B test configured: {version_weights}")

    def route(self, features, true_label=None):
        r, cumulative = random.random(), 0.0
        chosen = next(iter(self.traffic_split))
        for version, weight in self.traffic_split.items():
            cumulative += weight
            if r < cumulative:
                chosen = version
                break
        with torch.no_grad():
            pred = int(self.models[chosen](
                torch.tensor([features], dtype=torch.float32)).argmax(dim=1).item())
        self.results[chosen]["total"] += 1
        if true_label is not None and pred == true_label:
            self.results[chosen]["correct"] += 1
        return pred, chosen

    def report(self):
        print("\n--- A/B Test Results ---")
        for version, s in self.results.items():
            acc = s["correct"] / max(s["total"], 1)
            print(f"  {version}: {s['total']} predictions, accuracy={acc:.4f}")

Nota bene: in a real A/B test you would never eyeball the two accuracy numbers and declare a winner -- you would run a proper significance test first (small differences on small samples are noise, not signal). But the plumbing is exactly this: split traffic by weight, collect results per version, compare. Feature-flag systems like LaunchDarkly are, under the hood, a fancier version of this little route() method.

Putting it all together

Now the payoff -- the whole lifecycle in one script, each component talking to the next:

# 1 + 2. Two versions already trained and registered above.
# 3. Promote v1 to production.
registry.promote("classifier", "v1")

# 4. Stand up the server (in real life this is a long-running service).
server = PredictionServer(registry, "classifier")

# 5. Warm up a monitoring baseline from the first slice of traffic.
monitor = ModelMonitor(window_size=50)
baseline = [server.predict(X_val[i].tolist()).model_dump() for i in range(100)]
monitor.set_baseline(baseline)

# 6. Simulate live traffic flowing through server -> monitor.
for _ in range(200):
    idx = random.randint(0, len(X_val) - 1)
    monitor.log_prediction(server.predict(X_val[idx].tolist()).model_dump())

# 7. Run the A/B test between v1 and v2 on labelled traffic.
router = ABRouter(registry, "classifier")
router.configure({"v1": 0.5, "v2": 0.5})
for _ in range(500):
    idx = random.randint(0, len(X_val) - 1)
    router.route(X_val[idx].tolist(), true_label=int(y_val[idx]))
router.report()

# 8. If v2 genuinely wins, promote it and hot-reload the server.
if metrics_v2["accuracy"] > metrics_v1["accuracy"]:
    registry.promote("classifier", "v2")
    server.reload_model()
    print("Promoted v2. Server reloaded with zero code changes.")

monitor.report()

That is the full cycle: train, register, promote, serve, monitor, A/B test, promote the winner, keep watching. In a real company each of those boxes is a separate service -- the registry might be MLflow, the server TorchServe or a FastAPI app behind a load balancer, the monitor Evidently or Prometheus, the router a feature-flag platform. But the architecture -- how the boxes connect and what they hand each other -- is the same at every scale. Understanding this flow is roughly the whole difference between an ML engineer and someone who can train models in a notebook.

Cost awareness

Episode #134 was all about infrastructure economics, so it would be a bit rich to build a platform and pretend compute is free. A tiny cost tracker keeps the money visible in stead of hidden in a bill that lands at month-end:

class CostTracker:
    def __init__(self, gpu_cost_per_hour=1.0):
        self.gpu_rate = gpu_cost_per_hour
        self.training_costs = []
        self.inference_costs = []

    def log_training(self, model_name, version, duration_hours, gpu_count=1):
        cost = duration_hours * self.gpu_rate * gpu_count
        self.training_costs.append({"model": f"{model_name}/{version}", "cost": cost})

    def log_inference(self, num_requests, avg_latency_ms, instance_cost_per_hour):
        compute_hours = (num_requests * avg_latency_ms / 1000) / 3600
        self.inference_costs.append({"requests": num_requests,
                                     "cost": compute_hours * instance_cost_per_hour})

    def summary(self):
        train = sum(c["cost"] for c in self.training_costs)
        infer = sum(c["cost"] for c in self.inference_costs)
        print("\n--- Cost Summary ---")
        print(f"Training:  ${train:.2f}")
        print(f"Inference: ${infer:.2f}")
        print(f"Total:     ${train + infer:.2f}")

Every prediction, every training run, every retraining cycle has a price tag. Making that price tag visible at the point where the spending happens -- not in a spreadsheet three weeks later -- is exactly how you keep an AI project economically alive rather than watching it get quietly cancelled after the first surprise invoice.

What this project actually teaches

The mini project is deliberately light on any single component -- the registry is a hundred lines, the monitor barely more. The lesson is in the wiring. Training produces artifacts that land in a registry. The server pulls from that registry and knows nothing else. The monitor watches the server. The router compares versions. Cost tracking cuts across all of it. Pull any one box out and the others keep their contract.

In a real system you'd bolt on the boring-but-vital stuff we skipped: authentication on the API, a database in stead of in-memory Python lists, automated retraining triggered by the drift alerts, CI that tests a model before it is allowed to promote (episode #124), and honest error handling everywhere. But the skeleton -- these five boxes and the shape of their conversation -- is genuinely what production ML platforms look like once you strip away the logos.

And here is the thread I want you tugging on as we move forward. Every model in this platform was one we trained, from our own data, from scratch. But more and more, the model you serve is not one you trained at all -- it is a giant somebody else pretrained on half the internet, that you merely adapt, prompt, or fine-tune to your task. That shifts what "training", "registry" and "cost" even mean. That is where we head next ;-)

Did it click? Let's check

A model registry stores versioned models with metadata and enforces one "production" version at a time, so everything downstream has a single source of truth;
experiment tracking during training captures losses, metrics and configs, tying each result to the exact version it produced;
the prediction server loads from the registry (knowing nothing about training) and logs every prediction it serves;
drift detection compares a rolling window against a baseline and alerts when the prediction distribution wanders, even without live labels;
A/B testing splits traffic between versions and scores them, though real winners need a significance test, not a glance;
cost tracking keeps compute spending visible at every stage in stead of hiding it in a monthly invoice;
and the real value is in the connections between the boxes, not any single box alone.