🛵 Delivery ETA Model Training#

QuickBite ETA (ML Model Training & Serving) illustrates classical tabular machine learning in production: predicting a continuous numerical target (food delivery arrival time in minutes) from structured operational features using scikit-learn, pandas, and joblib.

On first invocation, get_or_train_model() synthesizes 5,000 realistic delivery orders across four features — distance_km, prep_time_min, rider_available (0/1), and is_raining (0/1) — fits a 60-tree RandomForestRegressor, and serializes the trained artifact to disk at /tmp/eta_model.pkl. Every prediction request loads the cached model, constructs a single-row DataFrame from the validated order inputs, and predicts the ETA in milliseconds.

1. When to Use Classical ML Instead of an LLM

Not every AI problem is a language problem. Predicting delivery ETAs, credit risk, or churn from numeric columns requires fast, calibrated regression over historical tables. A RandomForestRegressor runs in <2 ms on CPU, costs zero API tokens, is 100% deterministic, and learns quantitative feature relationships directly from data.

2. Synthetic Data Generation & Ground-Truth Formula

To make the demo self-contained without shipping a large CSV dataset, the service seeds NumPy (np.random.seed(42)) and generates n = 5,000 orders with a noisy ground-truth law:

eta_min = 8 + 3 × distance_km + 0.7 × prep_time_min + 9 × is_raining + 6 × (1 - rider_available) + normal(0, 2)

Notice the domain logic encoded in the data: rain adds a 9-minute penalty, and having no nearby rider available (rider_available == 0) adds a 6-minute dispatch delay.

3. Ensemble Learning with RandomForestRegressor

A Random Forest trains 60 independent decision trees (n_estimators=60) on bootstrap samples of the training data and averages their predictions (bagging). Averaging across many decorrelated trees cancels out the normal(0, 2) Gaussian noise and prevents overfitting to individual outliers.

4. Two-Tier Model Caching (Disk Artifact + In-Memory Singleton)

Training on every HTTP request would waste CPU and slow down responses. The service uses a two-tier cache: joblib.dump(model, MODEL_PATH) persists the trained estimator to /tmp/eta_model.pkl across worker restarts, while the module-level _cached_model reference keeps the deserialized object in RAM so subsequent calls skip disk I/O entirely.

src/python/01_model_training.py trains, serializes, and serves a classical Scikit-Learn RandomForestRegressor (QuickBite ETA) that predicts food-delivery arrival time in minutes from structured tabular order features.

1. Core ML Concepts in Tabular Regression & Model Serving

• Supervised Tabular Regression: Maps four numerical order features (distance_km, prep_time_min, rider_available, and is_raining) to a continuous target variable (eta_min).
• Ensemble Decision Trees (RandomForestRegressor): Fits 60 bootstrapped decision trees (n_estimators=60, random_state=42) and averages their leaf predictions to reduce variance and prevent overfitting.
• Two-Tier Model Artifact Caching: Separating training from inference avoids fitting 5,000 rows on every prediction. The trained estimator is persisted to disk (/tmp/eta_model.pkl via joblib) and cached in process memory (_cached_model) for sub-millisecond inference.

2. Key Entities in src/python/01_model_training.py

• MODEL_PATH & _cached_model: Store the path to the serialized model artifact (/tmp/eta_model.pkl) and the in-memory cached RandomForestRegressor instance.
• ModelTrainingRequest(BaseModel) & ModelTrainingResponse(BaseModel): Pydantic schemas defining the four input order features (distance_km: float = 4.5, prep_time_min: float = 15.0, rider_available: int = 1, is_raining: int = 1) and the output prediction payload (echoed features, eta_minutes: float, and message: str).
• get_or_train_model() -> RandomForestRegressor:
  • Step ① — Synthesize training dataset on first run: If _cached_model is None and MODEL_PATH does not exist, seeds NumPy (np.random.seed(42)) and generates a 5,000-row pd.DataFrame covering distance_km (0.5–12 km), prep_time_min (5–30 min), rider_available (0 or 1), and is_raining (0 or 1).
  • Step ② — Compute ground-truth target (eta_min): Evaluates 8 + df.distance_km * 3 + df.prep_time_min * 0.7 + df.is_raining * 9 + (1 - df.rider_available) * 6 + np.random.normal(0, 2, n).
  • Step ③ — Fit & serialize RandomForestRegressor: Separates features X and target y, fits RandomForestRegressor(n_estimators=60, random_state=42).fit(X, y), and saves the artifact to MODEL_PATH via joblib.dump(model, MODEL_PATH).
  • Step ④ — Load persisted model into memory: Loads _cached_model = joblib.load(MODEL_PATH) and returns it.
• predict_delivery_eta(distance_km, prep_time_min, rider_available, is_raining) -> dict (Step ⑤): Calls get_or_train_model(), builds a single-row pd.DataFrame([order_row]) with matching feature names, runs model.predict(X)[0], and returns the rounded eta_minutes and status message.

3. End-to-End Training & Inference Flow

1. predict_delivery_eta() receives the four order features and requests the estimator from get_or_train_model().
2. On a cold start, get_or_train_model() synthesizes 5,000 labeled delivery orders, trains the 60-tree RandomForestRegressor, and persists eta_model.pkl with joblib; on warm calls, it returns _cached_model directly from memory.
3. The single-row feature DataFrame is passed to model.predict(X), averaging predictions across all 60 trees to produce the final delivery ETA in minutes.

QuickBite ETA — ML Model Training & ETA Prediction Service.

"""QuickBite ETA — ML Model Training & ETA Prediction Service.

This service demonstrates training, serializing, and serving predictions with a scikit-learn model:
1. Generates 5,000 rows of synthetic food-delivery data (`distance_km`, `prep_time_min`, `rider_available`, `is_raining`).
2. Trains a `RandomForestRegressor` to predict delivery arrival time (`eta_min`) and caches `eta_model.pkl` via `joblib`.
3. Loads the persisted model artifact, validates delivery order inputs with Pydantic (`ModelTrainingRequest`), and returns ETA in minutes.
"""

import os
from pathlib import Path

import joblib
import numpy as np
import pandas as pd
from pydantic import BaseModel
from sklearn.ensemble import RandomForestRegressor

MODEL_PATH = Path(os.getenv("MODEL_PATH", "/tmp/eta_model.pkl"))
_cached_model = None


# ---------------------------------------------------------------------------
# Request & Response Schemas (Pydantic)
# ---------------------------------------------------------------------------
class ModelTrainingRequest(BaseModel):
    """Input payload for QuickBite ETA model prediction."""

    distance_km: float = 4.5
    prep_time_min: float = 15.0
    rider_available: int = 1
    is_raining: int = 1


class ModelTrainingResponse(BaseModel):
    """Output payload containing order features, predicted ETA, and status message."""

    distance_km: float
    prep_time_min: float
    rider_available: int
    is_raining: int
    eta_minutes: float
    message: str


# ---------------------------------------------------------------------------
# Model Training & Persistence
# ---------------------------------------------------------------------------
def get_or_train_model() -> RandomForestRegressor:
    """Load the trained RandomForestRegressor from disk, or train and save it on first use."""
    global _cached_model
    if _cached_model is not None:
        return _cached_model

    # ① Train and persist eta_model.pkl if it does not exist yet
    if not MODEL_PATH.exists():
        np.random.seed(42)
        n = 5000

        df = pd.DataFrame({
            "distance_km": np.random.uniform(0.5, 12, n),
            "prep_time_min": np.random.uniform(5, 30, n),
            "rider_available": np.random.randint(0, 2, n),
            "is_raining": np.random.randint(0, 2, n),
        })

        # ② Ground-truth formula: base + distance*3 + prep*0.7 + rain penalty + rider penalty + noise
        df["eta_min"] = (
            8
            + df.distance_km * 3
            + df.prep_time_min * 0.7
            + df.is_raining * 9
            + (1 - df.rider_available) * 6
            + np.random.normal(0, 2, n)
        )

        # ③ Fit RandomForestRegressor and save artifact with joblib
        X, y = df.drop(columns=["eta_min"]), df["eta_min"]
        model = RandomForestRegressor(n_estimators=60, random_state=42).fit(X, y)
        joblib.dump(model, MODEL_PATH)

    # ④ Load the persisted model artifact from disk
    _cached_model = joblib.load(MODEL_PATH)
    return _cached_model


# ---------------------------------------------------------------------------
# Core Service Logic
# ---------------------------------------------------------------------------
def predict_delivery_eta(
    distance_km: float = 4.5,
    prep_time_min: float = 15.0,
    rider_available: int = 1,
    is_raining: int = 1,
) -> dict:
    """Predict delivery ETA in minutes using the trained RandomForestRegressor."""
    # ⑤ Load (or train) the model and run inference on the single-row order DataFrame
    model = get_or_train_model()
    order_row = {
        "distance_km": float(distance_km),
        "prep_time_min": float(prep_time_min),
        "rider_available": 1 if int(rider_available) else 0,
        "is_raining": 1 if int(is_raining) else 0,
    }
    X = pd.DataFrame([order_row])
    eta = round(float(model.predict(X)[0]), 1)

    return {
        **order_row,
        "eta_minutes": eta,
        "message": f"Your food arrives in {eta} min 🍔",
    }

Flask Blueprint server and route handler (model_training_route) in src/python/app.py.

"""Flask server for 04-ai-model-training-chromadb."""

import importlib.util
import os
import sys
from pathlib import Path

from dotenv import find_dotenv, load_dotenv
from flask import Blueprint, Flask, jsonify, request
from flask_cors import CORS
from rate_limiter import check_rate_limit

load_dotenv(find_dotenv())

PATH_PREFIX = os.environ.get("PATH_PREFIX", "")
STATIC_DIR = Path(__file__).resolve().parents[1]
app = Flask(__name__, static_folder=str(STATIC_DIR))
CORS(app)

bp = Blueprint("main", __name__)

_CUR_DIR = Path(__file__).resolve().parent


def _load_service(filename: str, alias: str):
    # ① Resolve the numbered service file path and create its import spec
    file_path = _CUR_DIR / filename
    spec = importlib.util.spec_from_file_location(f"services.{alias}", file_path)
    if spec is None or spec.loader is None:
        raise ImportError(f"Cannot load module from {file_path}")
    # ② Register and execute the module so its functions can be called by routes
    mod = importlib.util.module_from_spec(spec)
    sys.modules[f"services.{alias}"] = mod
    spec.loader.exec_module(mod)
    return mod


mod_ml = _load_service("01_model_training.py", "model_training")
mod_chroma = _load_service("02_chromadb.py", "chromadb_rag")
mod_redis = _load_service("03_redis.py", "redis_agent")


@bp.before_request
def enforce_rate_limit():
    # ① Only rate-limit mutating POST requests
    if request.method == "POST":
        # ② Check whether this client has exceeded 30 requests per hour
        blocked, msg, retry_after = check_rate_limit(
            request, max_requests=30, window_seconds=3600
        )
        if blocked:
            # ③ Return HTTP 429 with Retry-After header when rate-limited
            resp = jsonify({"error": msg})
            resp.status_code = 429
            resp.headers["Retry-After"] = str(retry_after)
            return resp


@bp.route("/")
def index():
    # ① Read index.html and inject the runtime PATH_PREFIX into data-api-base
    with open(os.path.join(app.static_folder, "index.html"), encoding="utf-8") as f:
        html = f.read()
    html = html.replace('data-api-base=""', f'data-api-base="{PATH_PREFIX}"')
    # ② Return the patched HTML page
    return app.response_class(html, mimetype="text/html")


@bp.route("/css/<path:filename>")
def css(filename):
    return app.send_static_file(os.path.join("css", filename))


@bp.route("/js/<path:filename>")
def js(filename):
    return app.send_static_file(os.path.join("js", filename))


@bp.route("/explanation/<path:filename>")
@bp.route("/info/<path:filename>")
def explanation_pages(filename):
    return app.send_static_file(os.path.join("explanation", filename))


@bp.route("/theory/<path:filename>")
def theory_pages(filename):
    """Serve the theory documentation pages from src/theory."""
    return app.send_static_file(os.path.join("theory", filename))


@bp.route("/api/model-training", methods=["POST"])
def model_training_service():
    # ① Parse delivery order features from the JSON request body
    data = request.get_json(force=True, silent=True) or {}
    try:
        distance_km = float(data.get("distance_km", 4.5))
        prep_time_min = float(data.get("prep_time_min", 15.0))
        rider_available = int(data.get("rider_available", 1))
        is_raining = int(data.get("is_raining", 1))
        if distance_km <= 0 or prep_time_min <= 0:
            return jsonify({"error": "Distance and prep time must be positive."}), 400
        # ② Run RandomForestRegressor inference in 01_model_training.py and return ETA
        res = mod_ml.predict_delivery_eta(distance_km, prep_time_min, rider_available, is_raining)
        res["result"] = f"Predicted Arrival: {res['eta_minutes']} min 🍔"
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/chromadb", methods=["POST"])
def chromadb_service():
    # ① Parse student question from JSON payload and reject empty queries
    data = request.get_json(force=True, silent=True) or {}
    question = (data.get("question") or data.get("message") or "").strip()
    if not question:
        return jsonify({"error": "Question cannot be empty"}), 400
    try:
        # ② Query ChromaDB notes collection via 02_chromadb.py and return answer + sources
        res = mod_chroma.ask_chromadb_notes(question)
        res["result"] = res["answer"]
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


@bp.route("/api/redis", methods=["POST"])
def redis_agent_service():
    # ① Parse session_id, message, and optional reset flag from JSON payload
    data = request.get_json(force=True, silent=True) or {}
    session_id = (data.get("session_id") or "demo").strip()
    message = (data.get("message") or "").strip()
    reset_session = bool(data.get("reset_session", False))
    if not session_id or (not reset_session and not message):
        return jsonify({"error": "Session ID and message are required"}), 400
    try:
        # ② Run DeskBuddy agent turn (or session reset) backed by Redis in 03_redis.py
        res = mod_redis.run_redis_agent(session_id, message, reset_session)
        res["result"] = res["answer"]
        return jsonify(res)
    except Exception as exc:
        return jsonify({"error": str(exc)}), 500


app.register_blueprint(bp, url_prefix=PATH_PREFIX)

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

Use these numbered points to quickly revise classical tabular ML training, persistence, and serving:

1. Supervised Tabular Regression: Predicting a continuous number (delivery ETA in minutes) from structured numeric columns (distance_km, prep_time_min, traffic, weather) is a classic supervised regression problem where tree ensembles excel over LLMs in speed, cost, and determinism.
2. Ensemble Learning with RandomForestRegressor: A Random Forest trains multiple independent decision trees (n_estimators=60) on bootstrapped data subsets and averages their predictions, reducing variance and preventing overfitting.
3. Reproducibility via Random Seeds: Setting both np.random.seed(42) for synthetic data generation and random_state=42 on the estimator guarantees identical model weights across runs.
4. Training vs. Inference Separation: Fitting a model on 5,000 rows is compute-heavy, whereas predicting a single row takes sub-millisecond CPU time; production services never retrain on every request unless explicitly asked (force_retrain=True).
5. Model Persistence with joblib: joblib.dump() and joblib.load() efficiently serialize Scikit-Learn estimators containing large NumPy arrays to disk (/tmp/eta_model.pkl).
6. Two-Tier Caching (RAM + Disk): Checking _cached_model in process memory first avoids even disk-deserialization overhead on warm requests, while falling back to MODEL_PATH on disk avoids retraining if the process restarts.
7. Feature Alignment via DataFrame Column Names: Constructing pd.DataFrame([[distance_km, prep_time_min, traffic, weather]], columns=[...]) at inference time guarantees the exact column order and names that model.fit(X, y) saw during training.