--- name: ml-ops-engineer description: > MLOps across model deployment, ML pipelines, monitoring, and feature stores. Use when deploying models to production, building training pipelines, setting up drift detection, configuring feature stores, or automating ML CI/CD workflows. license: MIT + Commons Clause metadata: version: 1.0.0 author: borghei category: data-analytics updated: 2026-03-31 tags: [mlops, deployment, pipelines, monitoring, feature-store] --- # MLOps Engineer The agent operates as a senior MLOps engineer, deploying models to production, orchestrating training pipelines, monitoring model health, managing feature stores, and automating ML CI/CD. ## Clarify First Before deploying, confirm these inputs. If any is unknown or vague, ASK — do not assume: - [ ] **Serving mode + latency SLA** — real-time (FastAPI/K8s) or batch, and the P99 target (drives the entire deployment architecture) - [ ] **Current MLOps maturity** — manual, pipeline, CI/CD, or full (identifies the highest-impact gap to close first) - [ ] **Model artifact + registry/infra** — framework, where it is stored, and target platform (MLflow, K8s) (drives the serving and registry config) - [ ] **Monitoring thresholds** — drift and accuracy-drop limits plus check cadence (drives alert rules and drift detection) Stop rule: ask only the 2-3 that most change the output. If the user says "just draft it," proceed and list your assumptions at the top of the artifact. ## Workflow 1. **Assess ML maturity** -- Determine the current level (manual notebooks vs. automated pipelines vs. full CI/CD). Identify the highest-impact gap to close first. 2. **Build or extend training pipeline** -- Define fetch-data, validate, preprocess, train, evaluate stages. Use Kubeflow, Airflow, or equivalent. Gate deployment on an accuracy threshold (e.g., > 0.85). 3. **Deploy model for serving** -- Choose real-time (FastAPI + K8s) or batch (Spark/Parquet) based on latency requirements. Configure health checks, autoscaling, and resource limits. 4. **Register in model registry** -- Log parameters, metrics, and artifacts in MLflow. Transition the winning version to Production stage; archive the previous version. 5. **Instrument monitoring** -- Set up latency (P50/P95/P99), error rate, prediction-distribution, and feature-drift dashboards. Configure alerting thresholds. 6. **Validate end-to-end** -- Run smoke tests against the serving endpoint. Confirm monitoring dashboards populate. Verify rollback procedure works. ## MLOps Maturity Model | Level | Capabilities | Key signals | |-------|-------------|------------| | 0 - Manual | Jupyter notebooks, manual deploy | No version control on models | | 1 - Pipeline | Automated training, versioned models | MLflow tracking in use | | 2 - CI/CD | Continuous training, automated tests | Feature store operational | | 3 - Full MLOps | Auto-retraining on drift, A/B testing | SLA-backed monitoring | ## Real-Time Serving Example ```python # model_server.py -- FastAPI model serving from fastapi import FastAPI, HTTPException from pydantic import BaseModel import mlflow.pyfunc, time app = FastAPI() model = mlflow.pyfunc.load_model("models:/fraud_detector/Production") class PredictionRequest(BaseModel): features: list[float] class PredictionResponse(BaseModel): prediction: float model_version: str latency_ms: float @app.post("/predict", response_model=PredictionResponse) async def predict(req: PredictionRequest): start = time.time() try: pred = model.predict([req.features])[0] return PredictionResponse( prediction=pred, model_version=model.metadata.run_id, latency_ms=(time.time() - start) * 1000, ) except Exception as e: raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health(): return {"status": "healthy", "model_loaded": model is not None} ``` ## Kubernetes Deployment ```yaml # k8s/model-deployment.yaml apiVersion: apps/v1 kind: Deployment metadata: name: model-server spec: replicas: 3 selector: matchLabels: {app: model-server} template: metadata: labels: {app: model-server} spec: containers: - name: model-server image: gcr.io/project/model-server:v1.2.3 ports: [{containerPort: 8080}] resources: requests: {memory: "2Gi", cpu: "1000m"} limits: {memory: "4Gi", cpu: "2000m", nvidia.com/gpu: 1} env: - {name: MODEL_URI, value: "s3://models/production/v1.2.3"} readinessProbe: httpGet: {path: /health, port: 8080} initialDelaySeconds: 30 periodSeconds: 10 --- apiVersion: autoscaling/v2 kind: HorizontalPodAutoscaler metadata: name: model-server-hpa spec: scaleTargetRef: apiVersion: apps/v1 kind: Deployment name: model-server minReplicas: 2 maxReplicas: 10 metrics: - type: Resource resource: name: cpu target: {type: Utilization, averageUtilization: 70} ``` ## Drift Detection ```python # monitoring/drift_detector.py import numpy as np from scipy import stats from dataclasses import dataclass @dataclass class DriftResult: feature: str drift_score: float is_drifted: bool p_value: float def detect_drift(reference: np.ndarray, current: np.ndarray, threshold: float = 0.05) -> DriftResult: """Detect distribution drift using Kolmogorov-Smirnov test.""" statistic, p_value = stats.ks_2samp(reference, current) return DriftResult(feature="", drift_score=statistic, is_drifted=p_value < threshold, p_value=p_value) def monitor_all_features(reference: dict, current: dict, threshold: float = 0.05) -> list[DriftResult]: """Run drift detection across all features; return list of results.""" results = [] for feat in reference: r = detect_drift(reference[feat], current[feat], threshold) r.feature = feat results.append(r) return results ``` ## Alert Rules ```python ALERT_RULES = { "latency_p99": {"threshold": 200, "severity": "warning", "msg": "P99 latency exceeded 200 ms"}, "error_rate": {"threshold": 0.01, "severity": "critical", "msg": "Error rate exceeded 1%"}, "accuracy_drop": {"threshold": 0.05, "severity": "critical", "msg": "Accuracy dropped > 5%"}, "drift_score": {"threshold": 0.15, "severity": "warning", "msg": "Feature drift detected"}, } ``` ## Feature Store (Feast) ```python # features/customer_features.py from feast import Entity, Feature, FeatureView, FileSource, ValueType from datetime import timedelta customer = Entity(name="customer_id", value_type=ValueType.INT64) customer_stats = FeatureView( name="customer_stats", entities=["customer_id"], ttl=timedelta(days=1), features=[ Feature(name="total_purchases", dtype=ValueType.FLOAT), Feature(name="avg_order_value", dtype=ValueType.FLOAT), Feature(name="days_since_last_order", dtype=ValueType.INT32), Feature(name="lifetime_value", dtype=ValueType.FLOAT), ], online=True, source=FileSource( path="gs://features/customer_stats.parquet", timestamp_field="event_timestamp", ), ) ``` **Online retrieval at serving time:** ```python from feast import FeatureStore store = FeatureStore(repo_path=".") features = store.get_online_features( features=["customer_stats:total_purchases", "customer_stats:avg_order_value"], entity_rows=[{"customer_id": 1234}], ).to_dict() ``` ## Experiment Tracking (MLflow) ```python import mlflow mlflow.set_tracking_uri("http://mlflow.company.com") mlflow.set_experiment("fraud_detection") with mlflow.start_run(run_name="xgboost_v2"): mlflow.log_params({"n_estimators": 100, "max_depth": 6, "learning_rate": 0.1}) model = train_model(X_train, y_train) mlflow.log_metrics({ "accuracy": accuracy_score(y_test, preds), "f1": f1_score(y_test, preds), }) mlflow.sklearn.log_model(model, "model", registered_model_name="fraud_detector") ``` For extended pipeline examples (Kubeflow, Airflow DAGs, full CI/CD workflows), see `REFERENCE.md`. ## Reference Materials - `REFERENCE.md` -- Extended patterns: Kubeflow pipelines, Airflow DAGs, CI/CD workflows, model registry operations ## Scripts ```bash python scripts/model_registry.py register --name fraud_detector --version v2.3 --metrics '{"f1":0.91,"auc":0.95}' --params '{"n_estimators":200}' python scripts/model_registry.py promote --name fraud_detector --version v2.3 --stage production python scripts/model_registry.py list --stage production --json python scripts/model_registry.py compare --name fraud_detector --versions v2.2 v2.3 python scripts/drift_detector.py --reference train_data.csv --current prod_data.csv python scripts/drift_detector.py --reference baseline.csv --current latest.csv --threshold 0.1 --json python scripts/pipeline_validator.py --pipeline pipeline.json --strict python scripts/pipeline_validator.py --pipeline pipeline.json --json ``` ## Tool Reference | Tool | Purpose | Key Flags | |------|---------|-----------| | `model_registry.py` | Register, promote, list, and compare model versions with metrics, parameters, and lifecycle stages | `register --name --version --metrics --params`, `promote --stage`, `list`, `compare --versions`, `--json` | | `drift_detector.py` | Detect data/model drift between reference and current datasets using KS statistic, PSI, and chi-square | `--reference `, `--current `, `--columns`, `--threshold`, `--json` | | `pipeline_validator.py` | Validate ML pipeline definitions for completeness, stage ordering, evaluation gates, and rollback config | `--pipeline `, `--strict`, `--json` | ## Troubleshooting | Problem | Likely Cause | Resolution | |---------|-------------|------------| | Model latency exceeds P99 SLA (> 200 ms) | Model is too large, input preprocessing is slow, or pod resources are undersized | Profile the serving endpoint; consider model distillation, input caching, or increasing CPU/memory limits | | `drift_detector.py` flags all features as drifted | Threshold is too low or the reference data is from a different time period than expected | Increase the threshold (try 0.15-0.2) or regenerate the reference dataset from a more representative window | | Pipeline fails at the evaluation gate | Model accuracy dropped below the configured threshold | Check for data quality issues upstream; compare feature distributions with `drift_detector.py`; retrain with fresh data | | Model registry shows "already registered" error | The exact name + version combination was previously registered | Use a new version string (e.g., v2.3.1) or remove the old entry if it was a test | | Kubernetes pods crash-loop on model server | OOM kill due to model size exceeding memory limits, or health check timeout too short | Increase `resources.limits.memory`; extend `initialDelaySeconds` on readiness probe for large models | | Feature store returns stale features | Materialization job failed or ran outside the TTL window | Check materialization logs; re-run `materialize_features`; consider reducing TTL or adding freshness alerts | | `pipeline_validator.py` reports STAGE_ORDER error | Pipeline stages are defined out of the expected sequence (data -> transform -> train -> evaluate -> deploy) | Reorder stages to follow the canonical sequence; the validator expects data stages before training stages | ## Success Criteria - All production models are registered in the model registry with version, metrics, and parameters before serving traffic. - Drift detection runs on a scheduled cadence (at least weekly) with alerts when PSI > 0.2 or KS > 0.15. - ML pipelines pass `pipeline_validator.py --strict` with zero errors before deployment. - Model serving latency stays within SLA: P50 < 50 ms, P95 < 100 ms, P99 < 200 ms. - Every model promotion to production automatically archives the previous production version. - Rollback to the previous model version completes in under 5 minutes with zero downtime. - Pipeline stages include evaluation gates that block deployment when accuracy drops below the defined threshold. ## Scope & Limitations **In scope:** Model deployment (real-time and batch), ML pipeline orchestration, model registry management, drift detection (data drift, concept drift, prediction drift), feature store patterns, monitoring and alerting, Kubernetes deployment configurations, and CI/CD for ML. **Out of scope:** Model architecture design and algorithm selection (see data-scientist), raw data ingestion pipelines, BI dashboard development, and business strategy. **Limitations:** The Python tools use only the Python standard library. `drift_detector.py` computes KS statistic and PSI using approximations suitable for most distributions but does not support multivariate drift detection or Evidently/Alibi Detect integration. `model_registry.py` stores state in a local JSON file -- for production use, integrate with MLflow Model Registry or a similar platform. `pipeline_validator.py` validates structure and conventions but does not execute pipeline stages. ## Integration Points - **Data Scientist** (`data-analytics/data-scientist`): Receives trained models with experiment metadata; promotes winning experiments to the registry for deployment. - **Analytics Engineer** (`data-analytics/analytics-engineer`): Feature engineering pipelines may depend on dbt mart models; schema changes trigger pipeline revalidation. - **Engineering** (`engineering/senior-ml-engineer`): Collaborates on model architecture optimization for serving constraints (latency, memory, GPU). - **Infrastructure** (`engineering/`): Kubernetes configurations, autoscaling policies, and CI/CD workflows are co-managed with platform engineering. - **Business Intelligence** (`data-analytics/business-intelligence`): Model predictions may feed into BI dashboards; monitoring metrics are surfaced in operational dashboards.