Recap — Anatomy of a Production ML Service

# Anatomy of a Production ML Service A real ML service has structure. The seven elements below distinguish a toy from a production system. ## Element 1: Request validation Define the input schema with Pydantic. Reject malformed requests early. ```python from pydantic import BaseModel, Field from datetime import datetime class TransactionRequest(BaseModel): transaction_id: str = Field(..., min_length=1, max_length=64) amount: float = Field(..., gt=0, le=1e8) merchant_id: str = Field(..., min_length=1) user_id: str = Field(..., min_length=1) payment_method: str timestamp: datetime ``` If a request comes in with `amount: -5`, FastAPI returns a 422 with a clear error before the model is touched. Saves debugging time. ## Element 2: Health and readiness endpoints ```python @app.get('/health') def health(): return {'status': 'ok'} @app.get('/ready') def ready(): if model is None or feature_store_connection is None: return {'status': 'not_ready'}, 503 return {'status': 'ready'} ``` `/health` for liveness probes (is the service alive). `/ready` for readiness probes (can it actually serve requests). Kubernetes / Docker orchestrators rely on these to manage your service. ## Element 3: Model loading at startup, not per-request ```python from fastapi import FastAPI import joblib app = FastAPI() model = None @app.on_event('startup') def load_model(): global model model = joblib.load('fraud_model.pkl') print('Model loaded.') ``` Loading takes time (potentially seconds). Doing it per-request adds that to every prediction. Load once at startup. ## Element 4: Single source of feature computation Import the same `build_features` used in training: ```python from features.feature_engineering import build_features @app.post('/predict') def predict(req: TransactionRequest): features = build_features(req.dict()) prediction = model.predict_proba([features])[0][1] return {'fraud_probability': float(prediction)} ``` `build_features` is the same function used during training-data construction. No drift. ## Element 5: Logging every prediction ```python import logging logger = logging.getLogger(__name__) @app.post('/predict') def predict(req: TransactionRequest): features = build_features(req.dict()) prediction = model.predict_proba([features])[0][1] # Log to warehouse for monitoring log_prediction({ 'transaction_id': req.transaction_id, 'features': features, 'prediction': float(prediction), 'model_version': MODEL_VERSION, 'timestamp': datetime.utcnow().isoformat(), }) return {'fraud_probability': float(prediction)} ``` Every prediction logged: request, output, model version, timestamp. Required for monitoring and incident debugging. For latency-sensitive services, log asynchronously (background task) so logging doesn't slow predictions. ## Element 6: Graceful error handling ```python from fastapi import HTTPException @app.post('/predict') def predict(req: TransactionRequest): try: features = build_features(req.dict()) except FeatureBuildError as e: logger.error(f'Feature build failed: {e}') raise HTTPException(status_code=400, detail='Invalid input data') try: prediction = model.predict_proba([features])[0][1] except Exception as e: logger.error(f'Prediction failed: {e}') # Fallback: return baseline rate or graceful degradation return { 'fraud_probability': BASELINE_FRAUD_RATE, 'model_version': 'fallback', 'note': 'Model unavailable, returning baseline' } return {'fraud_probability': float(prediction)} ``` Don't crash the service when one prediction fails. Return a sensible fallback. Log the error for investigation. ## Element 7: Versioning in the response ```python return { 'transaction_id': req.transaction_id, 'fraud_probability': float(prediction), 'model_version': MODEL_VERSION, 'feature_version': FEATURE_VERSION, 'served_at': datetime.utcnow().isoformat(), } ``` The consumer (and you, debugging later) knows which model produced the prediction. Critical for A/B testing and audit. ## The full minimum service ```python from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from datetime import datetime import joblib import logging from features.feature_engineering import build_features from monitoring.logger import log_prediction logger = logging.getLogger(__name__) MODEL_VERSION = 'v1.2.0' FEATURE_VERSION = 'v1.0.0' BASELINE_FRAUD_RATE = 0.003 app = FastAPI() model = None class TransactionRequest(BaseModel): transaction_id: str amount: float = Field(..., gt=0) merchant_id: str user_id: str payment_method: str timestamp: datetime class PredictionResponse(BaseModel): transaction_id: str fraud_probability: float model_version: str feature_version: str serve