Ai Ops

AI Ops

Overview

Deploying a machine learning model is the easy part. Keeping it correct, fast, fair, and cost-effective in production is where most AI projects fail. AI Ops (MLOps) bridges the gap between data science experimentation and production engineering — covering model deployment, monitoring, retraining, and governance.

The AI Ops contract: A model in production is software that happens to be probabilistic. It needs all the same operational rigor as any other service — plus additional checks for correctness drift, data distribution shifts, and fairness degradation.

When to Use

• Deploying a model to production for inference
• Setting up model monitoring and drift detection
• Designing retraining pipelines and strategies
• Building feature stores and serving infrastructure
• Implementing A/B testing for model variants
• Ensuring model governance, explainability, and compliance
• Optimizing inference latency and throughput

NOT for:

• Model training and experimentation (this is data science work)
• Building the initial model prototype
• Pure research without production intent

The AI Ops Process

Step 1: Define Model Success Criteria

Before deploying, define what "working" means:

MODEL SUCCESS CONTRACT:
├── Accuracy: [metric and threshold, e.g. AUC-ROC > 0.85]
├── Latency: [p50  1s (batch)           → Async queue + worker pool (batch inference)

Example: Containerized inference service:

# FastAPI inference service with health checks and metrics
from fastapi import FastAPI, HTTPException
from prometheus_client import Counter, Histogram
import joblib

app = FastAPI()
model = joblib.load('/models/v1.pkl')

predictions_total = Counter('model_predictions_total', 'Total predictions')
latency_histogram = Histogram('model_latency_seconds', 'Prediction latency')

@app.get('/health')
def health():
    return {
        'status': 'healthy',
        'model_loaded': model is not None,
        'model_version': '1.0.0'
    }

@app.post('/predict')
@latency_histogram.time()
def predict(features: dict):
    try:
        result = model.predict([features])
        predictions_total.inc()
        return {
            'prediction': result[0],
            'model_version': '1.0.0',
            'confidence': getattr(model, 'predict_proba', lambda x: [None])([features])[0].tolist()
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

Step 3: Implement Model Monitoring

Monitor three types of drift:

DRIFT TYPES:
1. DATA DRIFT — Input distribution changes
   └── Example: A fraud model trained on US transactions sees EU transactions
   
2. CONCEPT DRIFT — The relationship between input and output changes
   └── Example: A recommendation model trained before a major UI redesign
   
3. LABEL DRIFT — Ground truth distribution changes
   └── Example: A medical diagnosis model sees a new disease variant

Monitoring implementation:

from scipy.stats import ks_2samp
import numpy as np

def detect_drift(reference_data, current_data, threshold=0.05):
    """
    Kolmogorov-Smirnov test for data drift.
    Returns True if drift is detected.
    """
    drift_detected = False
    report = {}
    
    for feature in reference_data.columns:
        statistic, p_value = ks_2samp(
            reference_data[feature],
            current_data[feature]
        )
        
        if p_value  1 day, >$1000) → Triggered retraining only, heavy validation

Retraining pipeline pattern:

# Scheduled retraining with validation gates
def retrain_pipeline():
    # 1. Gather new training data
    new_data = fetch_data_since(last_training_date)
    
    # 2. Validate data quality
    if not validate_training_data(new_data):
        raise DataQualityError("Training data failed quality checks")
    
    # 3. Train new model
    new_model = train(new_data)
    
    # 4. Validate on holdout set
    holdout_metrics = evaluate(new_model, holdout_data)
    
    # 5. Compare to production model
    production_metrics = load_production_metrics()
    
    if holdout_metrics['accuracy'] < production_metrics['accuracy'] * 0.95:
        raise ModelQualityError("New model performs worse than production")
    
    # 6. Shadow deployment (serve predictions, don't use them)
    deploy_shadow(new_model, traffic_split=0.1)
    
    # 7. After validation period, promote
    if shadow_metrics_acceptable():
        promote_to_production(new_model)
        archive_old_model()

Step 5: Ensure Model Governance

GOVERNANCE CHECKLIST:
├── Versioning: Every model has a version, training data hash, and config snapshot
├── Lineage: Can trace any prediction back to training data and model version
├── Explainability: Can explain why a specific prediction was made
├── Fairness: Regular audits for demographic bias
├── Security: Model artifacts are access-controlled, no secrets in training data
├── Compliance: GDPR right-to-explanation, CCPA opt-out mechanisms
└── Rollback: Can revert to previous model version in < 5 minutes

Model card pattern:

# Model Card: [Model Name] v[Version]

## Overview
- Purpose: [What does this model do?]
- Architecture: [e.g., XGBoost, Transformer, CNN]
- Training date: [Date]
- Training data: [Dataset name, size, time range]

## Performance
| Metric | Value | Evaluation Dataset |
|--------|-------|-------------------|
| Accuracy | 0.92 | Holdout (2024-01-01 to 2024-03-01) |
| F1 Score | 0.89 | Holdout |
| Latency (p99) | 45ms | Production shadow |

## Limitations
- [Known failure modes]
- [Data coverage gaps]
- [Temporal limitations]

## Fairness
| Group | Accuracy | FPR | FNR |
|-------|----------|-----|-----|
| Group A | 0.93 | 0.05 | 0.07 |
| Group B | 0.91 | 0.06 | 0.08 |

## Changelog
- v1.0.0: Initial deployment
- v1.1.0: Retrained with Q2 data, +2% accuracy

Feature Store Pattern

# Feature store: Centralized, versioned feature computation
class FeatureStore:
    def __init__(self):
        self.online_store = RedisCluster()   # < 10ms lookups
        self.offline_store = Snowflake()     # Historical data
    
    def get_online_features(self, entity_id, feature_names):
        """Real-time feature serving for inference"""
        return self.online_store.hmget(f"entity:{entity_id}", feature_names)
    
    def get_offline_features(self, entity_ids, feature_names, time_range):
        """Batch feature serving for training"""
        return self.offline_store.query(f"""
            SELECT {', '.join(feature_names)}
            FROM feature_store
            WHERE entity_id IN ({', '.join(entity_ids)})
            AND timestamp BETWEEN '{time_range.start}' AND '{time_range.end}'
        """)
    
    def materialize_features(self, feature_definition):
        """Compute and store features from raw data"""
        # Run on schedule, backfill on demand
        computed = feature_definition.compute()
        self.online_store.load(computed)
        self.offline_store.load(computed)

Common Rationalizations

| Rationalization | Reality | |---|---| | "The model is accurate, that's enough" | Accuracy is necessary but not sufficient. Latency, fairness, explainability, and drift monitoring are equally critical in production. | | "We'll monitor after deployment" | Monitoring must be in place BEFORE deployment. You can't retroactively catch the first failure. | | "Retraining is the data scientist's job" | Retraining is an operational pipeline. It needs CI/CD, testing, and rollback — all engineering concerns. | | "Our model is too simple to need this" | Even a linear regression can drift, become unfair, or get exploited. Simplicity doesn't exempt you from operational rigor. | | "We'll just use the latest model" | "Latest" is not a deployment strategy. Every model needs validation, shadow deployment, and a rollback plan. |

Red Flags

• Model deployed without monitoring
• No holdout evaluation before production promotion
• Retraining is manual and ad-hoc
• Can't explain why a specific prediction was made
• No model versioning or lineage tracking
• Training data is not access-controlled
• Predictions are served synchronously without timeout handling
• No A/B testing framework for model variants

Verification

Before an AI system goes to production:

• [ ] Model success criteria are defined and measurable
• [ ] Serving architecture meets latency and throughput requirements
• [ ] Monitoring detects data drift, concept drift, and performance degradation
• [ ] Retraining pipeline is automated and validated
• [ ] Model card is complete and reviewed
• [ ] Shadow deployment validated the model against production traffic
• [ ] Rollback procedure is tested and documented
• [ ] Explainability mechanism works for all predictions
• [ ] Fairness audit passed for relevant demographic groups
• [ ] Security review completed (no data leakage, proper access controls)

See Also

• For data pipeline design, see data-engineering
• For observability setup, see observability-and-instrumentation
• For testing ML systems, see test-driven-development
• For production deployment, see shipping-and-launch

Source & license

This open-source skill is cataloged on AgentStack and links to its original source — we do not rehost the code.

• Author: borhen68
• Source: borhen68/SkillEngine
• License: MIT

Install and usage instructions live in the source repository linked above.