How to Set Up AI System Monitoring¶

import time
from prometheus_client import Counter, Histogram, Gauge
import logging

# Define metrics
PREDICTION_COUNT = Counter(
    'ai_predictions_total',
    'Total number of predictions',
    ['model_name', 'model_version']
)

PREDICTION_LATENCY = Histogram(
    'ai_prediction_latency_seconds',
    'Prediction latency in seconds',
    ['model_name'],
    buckets=[0.01, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0]
)

PREDICTION_ERRORS = Counter(
    'ai_prediction_errors_total',
    'Total prediction errors',
    ['model_name', 'error_type']
)

MODEL_LOADED = Gauge(
    'ai_model_loaded',
    'Whether model is loaded and ready',
    ['model_name', 'model_version']
)

def predict_with_monitoring(model, input_data):
    """Wrapper that adds monitoring to predictions."""
    start_time = time.time()

    try:
        prediction = model.predict(input_data)

        # Record success metrics
        PREDICTION_COUNT.labels(
            model_name=model.name,
            model_version=model.version
        ).inc()

        latency = time.time() - start_time
        PREDICTION_LATENCY.labels(model_name=model.name).observe(latency)

        return prediction

    except Exception as e:
        PREDICTION_ERRORS.labels(
            model_name=model.name,
            error_type=type(e).__name__
        ).inc()
        logging.error(f"Prediction error: {e}")
        raise

import pandas as pd
from datetime import datetime, timedelta

class AccuracyMonitor:
    """Monitor model accuracy over time."""

    def __init__(self, model_name, alert_threshold=0.05):
        self.model_name = model_name
        self.baseline_accuracy = None
        self.alert_threshold = alert_threshold
        self.predictions = []

    def log_prediction(self, prediction_id, predicted, probability):
        """Log a prediction for later evaluation."""
        self.predictions.append({
            'prediction_id': prediction_id,
            'timestamp': datetime.now(),
            'predicted': predicted,
            'probability': probability,
            'actual': None
        })

    def record_actual(self, prediction_id, actual):
        """Record the actual outcome when known."""
        for pred in self.predictions:
            if pred['prediction_id'] == prediction_id:
                pred['actual'] = actual
                break

    def calculate_accuracy(self, days=7):
        """Calculate accuracy over recent period."""
        cutoff = datetime.now() - timedelta(days=days)

        recent = [p for p in self.predictions
                  if p['timestamp'] > cutoff and p['actual'] is not None]

        if not recent:
            return None

        correct = sum(1 for p in recent if p['predicted'] == p['actual'])
        accuracy = correct / len(recent)

        return accuracy

    def check_degradation(self):
        """Check if accuracy has degraded significantly."""
        current = self.calculate_accuracy(days=7)
        baseline = self.baseline_accuracy

        if current is None or baseline is None:
            return None

        degradation = baseline - current

        if degradation > self.alert_threshold:
            return {
                'alert': True,
                'message': f'Accuracy degraded by {degradation:.2%}',
                'baseline': baseline,
                'current': current
            }

        return {'alert': False, 'current': current}

from scipy import stats
import numpy as np

class DriftMonitor:
    """Monitor for data and concept drift."""

    def __init__(self, reference_data, feature_names):
        self.reference_data = reference_data
        self.feature_names = feature_names
        self.reference_stats = self._compute_stats(reference_data)

    def _compute_stats(self, data):
        """Compute distribution statistics."""
        stats_dict = {}
        for i, feature in enumerate(self.feature_names):
            if isinstance(data, np.ndarray):
                col = data[:, i]
            else:
                col = data[feature]

            stats_dict[feature] = {
                'mean': np.mean(col),
                'std': np.std(col),
                'min': np.min(col),
                'max': np.max(col),
                'distribution': col
            }
        return stats_dict

    def detect_drift(self, current_data, threshold=0.05):
        """Detect drift using statistical tests."""
        current_stats = self._compute_stats(current_data)
        drift_results = {}

        for feature in self.feature_names:
            ref_dist = self.reference_stats[feature]['distribution']
            cur_dist = current_stats[feature]['distribution']

            # Kolmogorov-Smirnov test
            ks_stat, p_value = stats.ks_2samp(ref_dist, cur_dist)

            drift_detected = p_value < threshold

            drift_results[feature] = {
                'drift_detected': drift_detected,
                'ks_statistic': ks_stat,
                'p_value': p_value,
                'mean_shift': (current_stats[feature]['mean'] -
                              self.reference_stats[feature]['mean'])
            }

        return drift_results

    def get_drift_summary(self, drift_results):
        """Summarize drift detection results."""
        drifted_features = [f for f, r in drift_results.items()
                          if r['drift_detected']]

        return {
            'total_features': len(self.feature_names),
            'drifted_features': len(drifted_features),
            'drift_percentage': len(drifted_features) / len(self.feature_names),
            'features_with_drift': drifted_features
        }

class FairnessMonitor:
    """Monitor fairness metrics in production."""

    def __init__(self, protected_attributes, alert_threshold=0.8):
        self.protected_attributes = protected_attributes
        self.alert_threshold = alert_threshold  # 80% rule
        self.predictions_log = []

    def log_prediction(self, prediction, protected_values):
        """Log prediction with protected attribute values."""
        entry = {
            'timestamp': datetime.now(),
            'prediction': prediction,
            **{attr: val for attr, val in
               zip(self.protected_attributes, protected_values)}
        }
        self.predictions_log.append(entry)

    def calculate_disparities(self, days=7):
        """Calculate fairness disparities over recent period."""
        cutoff = datetime.now() - timedelta(days=days)
        recent = [p for p in self.predictions_log if p['timestamp'] > cutoff]

        if len(recent) < 100:
            return None  # Not enough data

        df = pd.DataFrame(recent)
        results = {}

        for attr in self.protected_attributes:
            groups = df.groupby(attr)['prediction'].mean()

            if len(groups) < 2:
                continue

            min_rate = groups.min()
            max_rate = groups.max()
            disparity = min_rate / max_rate if max_rate > 0 else 1.0

            results[attr] = {
                'disparity_ratio': disparity,
                'alert': disparity < self.alert_threshold,
                'group_rates': groups.to_dict()
            }

        return results

    def generate_fairness_report(self):
        """Generate fairness monitoring report."""
        disparities = self.calculate_disparities()

        if disparities is None:
            return "Insufficient data for fairness analysis"

        report = ["Fairness Monitoring Report", "=" * 40]

        for attr, results in disparities.items():
            status = "ALERT" if results['alert'] else "OK"
            report.append(f"\n{attr}: {status}")
            report.append(f"  Disparity ratio: {results['disparity_ratio']:.3f}")
            report.append("  Group rates:")
            for group, rate in results['group_rates'].items():
                report.append(f"    {group}: {rate:.3f}")

        return "\n".join(report)

# alerting_config.py

ALERT_RULES = [
    {
        'name': 'model_error_rate_critical',
        'metric': 'ai_prediction_errors_total / ai_predictions_total',
        'threshold': 0.10,
        'operator': '>',
        'severity': 'critical',
        'window': '5m',
        'message': 'Model error rate exceeds 10%'
    },
    {
        'name': 'accuracy_degradation',
        'metric': 'baseline_accuracy - current_accuracy',
        'threshold': 0.05,
        'operator': '>',
        'severity': 'high',
        'window': '24h',
        'message': 'Model accuracy has dropped by more than 5%'
    },
    {
        'name': 'fairness_violation',
        'metric': 'min(group_rates) / max(group_rates)',
        'threshold': 0.80,
        'operator': '<',
        'severity': 'high',
        'window': '7d',
        'message': 'Fairness disparity exceeds threshold'
    },
    {
        'name': 'latency_high',
        'metric': 'ai_prediction_latency_seconds_p99',
        'threshold': 2.0,
        'operator': '>',
        'severity': 'high',
        'window': '10m',
        'message': 'P99 latency exceeds 2 seconds'
    },
    {
        'name': 'data_drift_detected',
        'metric': 'drift_features_percentage',
        'threshold': 0.20,
        'operator': '>',
        'severity': 'medium',
        'window': '24h',
        'message': 'Data drift detected in >20% of features'
    }
]