ML Deployment Research¶

import torch
import torch.onnx

model = MyModel()
model.eval()

# Export to ONNX
dummy_input = torch.randn(1, 3, 224, 224)
torch.onnx.export(
    model,
    dummy_input,
    "model.onnx",
    input_names=["image"],
    output_names=["prediction"],
    dynamic_axes={"image": {0: "batch_size"}, "prediction": {0: "batch_size"}},
    opset_version=17,
)

# Verify
import onnx
model_onnx = onnx.load("model.onnx")
onnx.checker.check_model(model_onnx)

# Inference with ONNX Runtime
import onnxruntime as ort
import numpy as np

session = ort.InferenceSession("model.onnx")
input_data = np.random.randn(1, 3, 224, 224).astype(np.float32)
outputs = session.run(None, {"image": input_data})
print(outputs[0].shape)   # (1, 1000)

# save_model.py
import bentoml
from sklearn.ensemble import RandomForestClassifier

model = RandomForestClassifier(n_estimators=100)
model.fit(X_train, y_train)

# Save to BentoML model store
saved_model = bentoml.sklearn.save_model("churn_classifier", model)
print(f"Saved: {saved_model.tag}")   # churn_classifier:abc123

# service.py
import bentoml
import numpy as np
from bentoml.io import NumpyNdarray, JSON

runner = bentoml.sklearn.get("churn_classifier:latest").to_runner()
svc = bentoml.Service("churn_service", runners=[runner])

@svc.api(input=NumpyNdarray(), output=JSON())
async def predict(input_array: np.ndarray):
    prediction = await runner.predict.async_run(input_array)
    return {"prediction": prediction.tolist()}

from ray import serve
from ray.serve.handle import DeploymentHandle
import numpy as np

@serve.deployment(num_replicas=2, ray_actor_options={"num_gpus": 0.5})
class ModelDeployment:
    def __init__(self):
        import joblib
        self.model = joblib.load("model.joblib")

    async def __call__(self, request):
        data = await request.json()
        features = np.array(data["features"]).reshape(1, -1)
        prediction = self.model.predict(features)
        probability = self.model.predict_proba(features).max()
        return {
            "prediction": int(prediction[0]),
            "confidence": float(probability),
        }

# Compose multiple models
@serve.deployment
class Ensemble:
    def __init__(self, model_a: DeploymentHandle, model_b: DeploymentHandle):
        self.model_a = model_a
        self.model_b = model_b

    async def __call__(self, request):
        # Run both models in parallel
        ref_a = self.model_a.remote(request)
        ref_b = self.model_b.remote(request)
        result_a, result_b = await ref_a, await ref_b
        # Average predictions
        return {"prediction": (result_a["prediction"] + result_b["prediction"]) / 2}

# Deploy
app = ModelDeployment.bind()
serve.run(app, host="0.0.0.0", port=8000)

import numpy as np
from scipy.stats import ks_2samp

class DriftDetector:
    def __init__(self, reference_data: np.ndarray, threshold=0.05):
        self.reference = reference_data
        self.threshold = threshold

    def check_drift(self, current_data: np.ndarray) -> dict:
        results = {}
        for i in range(current_data.shape[1]):
            stat, p_value = ks_2samp(self.reference[:, i], current_data[:, i])
            results[f"feature_{i}"] = {
                "statistic": float(stat),
                "p_value": float(p_value),
                "drift_detected": p_value < self.threshold,
            }
        return results

# Usage
detector = DriftDetector(X_train)
drift_report = detector.check_drift(X_new_batch)
drifted = [k for k, v in drift_report.items() if v["drift_detected"]]
if drifted:
    print(f"⚠️ Drift detected in: {drifted}")
    # Trigger retraining pipeline

import random

class ModelRouter:
    def __init__(self, models: dict, traffic_split: dict):
        self.models = models          # {"v1": model_v1, "v2": model_v2}
        self.traffic_split = traffic_split  # {"v1": 0.9, "v2": 0.1}

    def predict(self, features):
        # Route based on traffic split
        r = random.random()
        cumulative = 0
        for version, weight in self.traffic_split.items():
            cumulative += weight
            if r <= cumulative:
                model = self.models[version]
                prediction = model.predict(features)
                self._log_prediction(version, features, prediction)
                return {"prediction": prediction, "model_version": version}

    def _log_prediction(self, version, features, prediction):
        # Log to tracking system for later analysis
        pass

router = ModelRouter(
    models={"v1": model_old, "v2": model_new},
    traffic_split={"v1": 0.9, "v2": 0.1},  # 10% to new model
)