The Production ML Problem
Your data science team built a beautiful model. It achieves 94% accuracy on the validation set. You deploy it to production. Three weeks later, it's predicting garbage—accuracy has tanked to 67%.
What happened? Nobody knows. Your model is a black box. You don't have baseline metrics to compare against. You don't know if the training data changed, if the production data distribution shifted, or if your preprocessing pipeline broke.
Welcome to production machine learning without MLOps.
The irony is that building a functional MLOps pipeline isn't complicated. It's tedious, but not complicated. Most teams overcomplicate it by reaching for Kubeflow or MLflow on day one. You don't need those yet. You need discipline first.
The Minimal Stack
Here's what you actually need:
- Experiment tracking — Record every model, every training run, every hyperparameter.
- Version control — Know which code version trained which model.
- Automated retraining — When new data arrives, automatically retrain and promote if performance improves.
- Model serving — Run inference on the current model, fall back to previous version if it fails.
- Monitoring — Alert when accuracy drops or prediction latency spikes.
That's it. Everything else is optimization.
The Implementation
1. Experiment Tracking (Using a Database)
Instead of MLflow, use a Postgres table:
import json
import hashlib
from datetime import datetime
class ExperimentTracker:
def __init__(self, db_connection):
self.db = db_connection
def log_experiment(self, model_name, config, metrics):
"""Log a training run."""
config_hash = hashlib.md5(
json.dumps(config, sort_keys=True).encode()
).hexdigest()
self.db.execute("""
INSERT INTO experiments
(model_name, config, config_hash, metrics, timestamp, code_version)
VALUES (%s, %s, %s, %s, %s, %s)
""", [
model_name,
json.dumps(config),
config_hash,
json.dumps(metrics),
datetime.now(),
get_git_commit() # Current git hash
])
def get_best_model(self, model_name, metric='accuracy'):
"""Get the highest-performing model."""
result = self.db.query(f"""
SELECT model_name, config, metrics, timestamp
FROM experiments
WHERE model_name = %s
ORDER BY metrics->>'{{metric}}' DESC
LIMIT 1
""", [model_name])
return result[0] if result else None
Now you have a queryable history of every model you've trained:
tracker = ExperimentTracker(db)
# After training
tracker.log_experiment(
model_name="fraud_detector",
config={
"learning_rate": 0.001,
"n_estimators": 100,
"max_depth": 10
},
metrics={
"accuracy": 0.94,
"precision": 0.92,
"recall": 0.89,
"f1": 0.905
}
)
# Get the best model ever trained
best = tracker.get_best_model("fraud_detector")
print(f"Best model: {best['metrics']['accuracy']} accuracy")
2. Versioned Model Storage
Store models in S3 with semantic versioning:
import boto3
class ModelRegistry:
def __init__(self, bucket):
self.s3 = boto3.client('s3')
self.bucket = bucket
def save_model(self, model_name, version, model_data, metrics):
"""Save a model with version."""
key = f"models/{model_name}/{version}/model.pkl"
self.s3.put_object(
Bucket=self.bucket,
Key=key,
Body=model_data
)
# Store metadata alongside
self.s3.put_object(
Bucket=self.bucket,
Key=f"models/{model_name}/{version}/metadata.json",
Body=json.dumps({
"model_name": model_name,
"version": version,
"metrics": metrics,
"timestamp": datetime.now().isoformat(),
"code_version": get_git_commit()
})
)
def load_model(self, model_name, version="latest"):
"""Load a model by name and version."""
if version == "latest":
version = self.get_latest_version(model_name)
obj = self.s3.get_object(
Bucket=self.bucket,
Key=f"models/{model_name}/{version}/model.pkl"
)
return pickle.loads(obj['Body'].read())
def promote_to_production(self, model_name, version):
"""Promote a model to production (use this version for serving)."""
self.s3.put_object(
Bucket=self.bucket,
Key=f"models/{model_name}/PRODUCTION",
Body=version
)
registry = ModelRegistry("ml-models-bucket")
# After training and validation
registry.save_model(
model_name="fraud_detector",
version="1.2.3",
model_data=pickle.dumps(trained_model),
metrics={"accuracy": 0.94, "f1": 0.905}
)
# Promote to production
registry.promote_to_production("fraud_detector", "1.2.3")
3. Automated Retraining
Use a scheduled job (Kubernetes CronJob, GitHub Actions, or Airflow) that retrains daily:
# training_job.py
def retrain_if_better():
"""Retrain model, promote if performance improves."""
# Load current production model
registry = ModelRegistry("ml-models-bucket")
current = registry.load_model("fraud_detector", "latest")
current_metrics = get_model_metrics(current)
# Train new model on latest data
new_model = train_fraud_detector(latest_training_data)
new_metrics = evaluate_model(new_model)
tracker = ExperimentTracker(db)
version = get_next_version("fraud_detector")
tracker.log_experiment("fraud_detector", {}, new_metrics)
# Promote if better
improvement = (
new_metrics['f1'] - current_metrics['f1']
) / current_metrics['f1']
if improvement > 0.01: # 1% improvement threshold
print(f"Promoting v{version} (+{improvement:.1%} F1)")
registry.save_model(
"fraud_detector",
version,
pickle.dumps(new_model),
new_metrics
)
registry.promote_to_production("fraud_detector", version)
# Notify team
send_slack(f"🚀 New model promoted: {version}")
else:
print(f"Model v{version} not better ({improvement:.1%}), skipping")
# In Kubernetes CronJob:
# spec:
# schedule: "0 2 * * *" # 2 AM daily
# jobTemplate:
# spec:
# template:
# spec:
# containers:
# - image: myrepo/mlops:latest
# command: ["python", "training_job.py"]
4. Versioned Model Serving
Serve the current model, with automatic fallback:
from flask import Flask, request, jsonify
import logging
app = Flask(__name__)
logger = logging.getLogger(__name__)
class ModelServer:
def __init__(self):
self.registry = ModelRegistry("ml-models-bucket")
self.current_model = None
self.previous_model = None
self.load_models()
def load_models(self):
"""Load current and previous models."""
try:
self.current_model = self.registry.load_model(
"fraud_detector",
"latest"
)
except Exception as e:
logger.error(f"Failed to load current model: {e}")
try:
self.previous_model = self.registry.load_model(
"fraud_detector",
"previous"
)
except:
pass
def predict(self, features):
"""Predict with fallback."""
try:
# Try current model
prediction = self.current_model.predict([features])[0]
logger.info(f"Current model: {prediction}")
return prediction
except Exception as e:
logger.warning(f"Current model failed: {e}")
if self.previous_model:
# Fallback to previous
try:
prediction = self.previous_model.predict([features])[0]
logger.info(f"Previous model (fallback): {prediction}")
return prediction
except Exception as e2:
logger.error(f"Both models failed: {e2}")
raise
else:
raise
server = ModelServer()
@app.route('/predict', methods=['POST'])
def predict():
data = request.json
try:
prediction = server.predict(data['features'])
return jsonify({
"prediction": prediction,
"status": "ok"
})
except Exception as e:
return jsonify({
"error": str(e),
"status": "error"
}), 500
5. Monitoring & Alerting
Track prediction quality in production:
class PredictionMonitor:
def __init__(self, db):
self.db = db
def log_prediction(self, model_name, features, prediction,
actual_label=None):
"""Log every prediction."""
self.db.execute("""
INSERT INTO predictions
(model_name, features, prediction, actual_label, timestamp)
VALUES (%s, %s, %s, %s, %s)
""", [
model_name,
json.dumps(features),
prediction,
actual_label,
datetime.now()
])
def check_drift(self, model_name, window_hours=24):
"""Check for prediction distribution shift."""
recent = self.db.query(f"""
SELECT prediction, actual_label
FROM predictions
WHERE model_name = %s
AND timestamp > now() - interval '{window_hours} hours'
""", [model_name])
if not recent:
return None
predictions = [r['prediction'] for r in recent]
actual = [r['actual_label'] for r in recent
if r['actual_label'] is not None]
if actual:
accuracy = sum(
p == a for p, a in zip(predictions, actual)
) / len(actual)
if accuracy < 0.85: # Alert threshold
send_alert(
f"Model {model_name} accuracy dropped to {accuracy:.1%}"
)
return accuracy
monitor = PredictionMonitor(db)
# In prediction endpoint
@app.route('/predict', methods=['POST'])
def predict():
data = request.json
prediction = server.predict(data['features'])
# Log for monitoring
monitor.log_prediction(
"fraud_detector",
data['features'],
prediction
)
return jsonify({"prediction": prediction})
# Scheduled job: check for drift daily
monitor.check_drift("fraud_detector", window_hours=24)
The Result
With this stack, you have:
✅ Reproducibility: Every model trained is logged with its code version, config, and metrics. ✅ Safety: New models are only promoted if they improve performance. ✅ Resilience: If a new model fails in production, you immediately fall back to the previous version. ✅ Observability: You know when your model's accuracy is degrading and why.
Total lines of code: ~300. Total infrastructure cost: ~$50/month. Total operational complexity: manageable.
When to Level Up
Once this is working smoothly (after 3-6 months), then consider:
- Feature stores (if you have complex feature engineering)
- MLflow or Kubeflow (if you have 50+ models)
- A/B testing frameworks (if you need canary deployments)
But don't start there. Start with this.
Building production ML systems? We've helped enterprises go from "black box model" to "instrumented, monitored, versioned pipeline." Let's talk about your workflow.