ML Deployment Ecosystem (Small Models)

License: MIT License

Small on-prem machine learning ecosystem for small-data environments, where fast iteration, maintainability, and reliable deployment matter more than large-scale infrastructure.

This repo is the deployment server side of the ecosystem. It is designed to pair with the separate repo ml-packaging-toolbox, which produces the deployment artifact and can call the server's deploy/validate endpoints.

Why this repo exists

During my time in a small machine learning team, the deployment phase is consistently the bottleneck in our machine learning projects. Regardless of how good the modeling or signal processing pipelines were built, serving these models for real-time inference is what drives real value. Over time, I learned to utilize more tools and frameworks, and realized that most of our models are quite similar and the deployment can be standardized. Using cloud computing isn't an option for the system we're working with. External platforms still leave a lot of room when handling in-house/legacy data connectors and data processing pipelines — especially when we need to reconcile streaming data with long-term storage. At the end of the day, it's critical for us to quickly build an in-house platform to streamline these workflows and potentially allow new engineers to quickly develop and deploy new models. This project is a re-design of what I learned and would have done differently from my experiences in that project.

Project goals

• Reduce operational debt from one-off model deployments
• Improve code reuse across domain-specific projects
• Support both near-real-time and batch inference workflows
• Guarantee training–serving feature parity through a shared dual-path preprocessing design

What this repo provides

• Model registry: staged → validated → production lifecycle
• Deploy + validation API: upload model artifacts and run a smoke test
• Data collection API: store RT payloads in MongoDB and trigger async processing
• Async inference: Celery task aggregates multi-device data and scores with the production model
• Serving: synchronous prediction endpoint (JSON)
• Monitoring endpoints: counts/status snapshots for models/data/inference

Key Terms

These terms appear throughout the codebase, API routes, and documentation. They are intentionally generic so you can map them to whatever your domain calls the same concepts.

Term	Definition	Adapt it to your domain
Unit	A single item being measured or processed — one row of sensor readings, one test result, one part off the line.	wafer, sample, component, patient visit, transaction
Batch	A group of units processed together in one production run or campaign. Used as the grouping key (batch_id) for data quality filters during model training.	lot, run, shift, campaign, cohort
material_id	An identifier that ties together all sensor/device readings for the same physical unit across multiple devices or events. Pass it as a query param on /data/collect so the inference task knows which readings belong together. If you only have one device, omit it — the server falls back to dc_id.	wafer ID, part serial number, patient ID, order number
dc_id	Data Collection ID. A UUID assigned by the server to each individual /data/collect call. Use it to poll /inference/result/<dc_id> for the prediction that was triggered by that collection.	event ID, reading ID
Deploy / Deployment	The act of uploading a model artifact (zip) to the server and registering it in the model registry. A freshly deployed model has status staging. It only becomes production after passing the smoke test (/deploy/validate).	model registration, model upload
Validation	An automated smoke test run against a small held-out dataset bundled inside the deployment artifact. Promotes the model to production on pass. Keeps it in staging on fail so bad models never serve live traffic.	acceptance test, integration test
Data Collection	Posting a real-time sensor or device payload to /data/collect. The server stores the raw features, gates on the SQLite dispatch cache (is there a production model that needs this device/event?), and queues an async inference task.	measurement event, sensor push, data ingest
Postmeas	Post-measurement collection via /postmeas/collect. Used for QA or validated data that arrives after the primary production event — for example, a final test result that comes in minutes or hours after the in-line sensor data.	final test, end-of-line QA, lab result
Inference	The async background process (Celery task) that checks whether all required device/event data for a given material_id has arrived, merges the features, loads the production model, and writes the scored result to MongoDB.	scoring, prediction, background job
Serve / Serving	The synchronous /serve/predict endpoint that scores a single payload immediately and returns a JSON prediction in the same HTTP response. Use this when you need a result before returning to the caller, rather than polling for an async result.	real-time inference, online prediction, sync scoring

Getting Started

Prerequisites

Service	Min version	Notes
Python	3.10
MongoDB	6.0	required for the deployment server
Redis	7.0	required for the deployment server

Installation

python -m venv .venv
source .venv/bin/activate        # Windows: .venv\Scripts\activate
pip install -r requirements.txt

First-time server setup

1. Start MongoDB

Option A (local service):

mongod --dbpath ./data/db

Option B (Docker):

docker run --name ml-mongo -p 27017:27017 -d mongo:6

2. Configure the instance

Edit instance/config.py (git-ignored) with your SECRET_KEY, MONGO_URI, and Redis URLs.

3. Initialise databases (run once)

python setup/setup_mongo.py    # TTL + query indexes in MongoDB
python setup/setup_sqlite3.py  # dispatch cache in SQLite3

4. Start the Celery worker (separate terminal) Option A (Linux):

celery -A celery_worker worker --loglevel=info

Option B (Windows):
Windows does not support forking and must use spawn to create new worker processes. Instead, run:

celery -A celery_worker worker --loglevel=info --pool=threads

5. Start the development server

python runserver.py
# API available at http://127.0.0.1:5000

Examples

• examples/demo_mongodb_collect_and_infer.py: POST /data/collect then poll /inference/result/<dc_id>
• examples/demo_multi_device.py: multi-device session collection flow (requires the server to be running)

Running tests

pip install -r requirements-test.txt
python -m pytest tests/ -v

Production deployment (Gunicorn + Nginx)

gunicorn -c gunicorn.conf.py "webservice:create_app()"

Point Nginx at 127.0.0.1:8000. Never expose Gunicorn directly on 0.0.0.0.

Quick smoke-test after install

# 1. Register a pkl model with its context (which device/event it needs)
curl -X POST http://localhost:5000/deploy/ \
  -F "model_file=@clf.pkl" \
  -F "name=clf" -F "version=1.0.0" \
  -F 'context_yaml=required_contexts:
- device_id: deviceA
  event_code: EVT_001
  features: [feat1, feat2]'
# → {"model_id": "<id>", "status": "staging"}

# 2. Validate (promotes to production on pass)
curl -X POST http://localhost:5000/deploy/validate/<id>
# → {"passed": true, ...}

# 3. Collect real-time data — inference runs automatically in background
curl -X POST "http://localhost:5000/data/collect?device_id=deviceA&event_code=EVT_001&material_id=BATCH001" \
  -H "Content-Type: application/json" \
  -d '{"feat1": 1.0, "feat2": 2.5}'
# → {"dc_id": "<dc_id>", "status": "accepted"}

# 4. Poll for the scored result
curl http://localhost:5000/inference/result/<dc_id>

# 5. Or get a synchronous prediction directly
curl -X POST http://localhost:5000/serve/predict \
  -H "Content-Type: application/json" \
  -d '{"device_id": "deviceA", "features": {"feat1": 1.0, "feat2": 2.5}}'

Developer Notes

Repo Structure

ml-deployment-ecosystem/
│
├── webservice/                       # ml-deployment-server (Flask app)
│   ├── __init__.py                   # App factory
│   ├── extensions.py                 # PyMongo, Flask-Caching, Celery
│   ├── utils.py                      # JSON payload helpers, feature reconciliation
│   └── mod/
│       ├── deploy/                   # POST /deploy/ + /deploy/validate/<id>
│       ├── data/                     # POST /data/collect + /postmeas/collect
│       ├── inference/                # GET/POST /inference/ + async task
│       ├── serve/                    # POST/GET /serve/predict
│       └── deployment/               # GET /deployment/ monitoring
│
├── setup/
│   ├── setup_mongo.py                # MongoDB TTL + query indexes (run once)
│   └── setup_sqlite3.py              # SQLite3 schema for dispatch cache (run once)
├── tests/                            # Server test suite (pytest + mongomock)
├── instance/                         # Machine-specific config — NOT in VCS
├── config.py                         # Default server config
├── runserver.py                      # Dev server entry point
├── celery_worker.py                  # Celery worker entry point
└── gunicorn.conf.py                  # Production Gunicorn config

Dual-Path Data Pipeline

The central design principle in ml-packaging-toolbox is that every data pipeline object implements two processing paths.

                       ┌─────────────────────────────┐
                       │     DataAssemblyPipeline     │
                       └──────────┬──────────┬────────┘
                                  │          │
              ┌───────────────────▼──┐   ┌───▼───────────────────────┐
              │      LT path         │   │        RT path             │
              │  (model training)    │   │  (inference reconciliation)│
              └───────────┬──────────┘   └───────────┬────────────────┘
                          │                          │
              extract_lt(**params)        extract_rt(live_payload)
              → raw DataFrame             → normalised dict
                          │                          │
              chain.fit_transform_lt()    chain.transform_rt()
              → preprocessed DataFrame   → reconciled feature dict
                          │                          │
              train/test split            {feat: value, ...}
              → X_train, X_test, ...      ready for model.predict()

Why this matters: the preprocessing statistics (quantile bounds, column means/stds, fill values) are fitted once on the LT training data. The same fitted objects are applied to every real-time record in reconcile_task. This eliminates training–serving skew.

BaseExtractor interface

class BaseExtractor:
    def extract_lt(self, **kwargs) -> pd.DataFrame:
        # batch pull from long-term storage (SQL, file, API)

    def extract_rt(self, payload: dict) -> dict:
        # normalise a single RT payload to match LT field schema

BasePreprocessor interface

class BasePreprocessor:
    def fit(self, X: pd.DataFrame) -> BasePreprocessor:
        # learn parameters from LT training data

    def transform_lt(self, X: pd.DataFrame) -> pd.DataFrame:
        # apply to LT batch (training / validation)

    def transform_rt(self, record: dict) -> dict:
        # apply fitted parameters to a single RT record (inference)

Deployment Artifact Contract

ModelPackager.build() produces a zip containing:

File	Consumed by
model.pkl	/deploy/ — stored; inference_task loads for scoring
preprocessors.pkl	reconcile_task — applies fitted chain to RT payload
context.yml	/deploy/ registration + context store
features.json	inference_task — ordered feature alignment
test_data.json	/deploy/validate/<model_id> smoke test

Contract rule: any change to the zip structure in packager.py requires a matching change in webservice/mod/deploy/routes.py and webservice/mod/inference/tasks.py.

Storage notes

• Model artifacts: stored in MongoDB (models.pkl_bytes) rather than written to a deployment directory.
• Inference outputs: stored as JSON (inference.json_output).

Design Decisions

No platform field on models — context-based routing

Models are not tagged with a device name. Instead, each model declares its required contexts: a list of {device_id, event_code, features} entries stored in the contexts collection (contexts._id == models._id).

This means:

• A single model can require data from multiple devices/events before scoring
• POST /data/collect dispatches via a SQLite lookup (O(1)) — no MongoDB round-trip needed for the fast path
• inference_task finds all models that need a given (device_id, event_code), checks whether all required data for the same material_id has arrived, then scores
• POST /serve/predict locates the production model by querying contexts for device_id, then scores synchronously

material_id groups multi-device collections

When data arrives from multiple devices for the same physical unit, they share a material_id query param on /data/collect. The inference task uses material_id to gather all required device/event docs before scoring. Single-device flows use dc_id as a fallback material_id.

MongoDB (primary) + SQLite3

Collection	TTL	Purpose
models	none	Model artefacts + lifecycle (staging/production)
contexts	none	required_contexts per model; _id == model OID
data	3 days	RT payloads + reconciled features per collection
request	3 days	Per-request audit log (params, not payload)
validate	3 days	Smoke-test results
inference	14 days	Scored results + predicted JSON
validateddata	14 days	Post-hoc QA feature snapshots

SQLite3 holds a contexts dispatch cache: (device_id, event_code) pairs for all production models. Populated on model promotion, queried on every /data/collect for a sub-millisecond gate.

Redis

• Flask-Caching on serving endpoints (60 s TTL, manual cache key)
• Celery broker (DB1) + result backend (DB2)

Celery async chain

POST /data/collect → inference_task
  → find all contexts matching (device_id, event_code)
  → for each: check all required data for material_id is present
  → load model by ObjectId, merge features, score

Gunicorn

Sits behind Nginx. Never expose 0.0.0.0 in production.

Route Map

Lane	Blueprint	Endpoint
Deployment	deploy	POST /deploy/
Deployment	deploy	POST /deploy/validate/<model_id>
Data Collection	data	POST /data/collect?device_id=<id>&event_code=<code>[&material_id=<id>]
Data Collection	postmeas	POST /postmeas/collect
Inference	inference	GET /inference/result/<dc_id>
Inference	inference	POST /inference/trigger/<dc_id>
Serving	serve	POST /serve/predict (returns JSON)
Serving	serve	GET /serve/predict/<dc_id> (cached)
Monitoring	deployment	GET /deployment/ + /models /data /inference

Known Limitations (TODOs)

• Not for high-volume production traffic
• Data >2 min late: re-queue via POST /inference/trigger/<dc_id>
• preprocessors.pkl loading in reconcile_task is not yet wired — planned next