AI Model Deployment: Production-Ready Machine Learning Pipeline 2024

This MLOps pipeline implementation shows automated training, validation, deployment, and monitoring for production AI systems.

import os
import json
import logging
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass
from abc import ABC, abstractmethod
import pandas as pd
import numpy as np
import mlflow
import mlflow.sklearn
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, precision_score, recall_score
import joblib
import requests
import time
from datetime import datetime, timedelta

# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@dataclass
class ModelMetrics:
    accuracy: float
    precision: float
    recall: float
    f1_score: float
    training_time: float
    inference_time: float

@dataclass
class ModelVersion:
    version: str
    model_path: str
    metrics: ModelMetrics
    created_at: datetime
    is_production: bool = False

class DataValidator:
    """Data validation and quality checks"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.validation_rules = config.get('validation_rules', {})
    
    def validate_data(self, data: pd.DataFrame) -> Dict:
        """Validate input data quality"""
        validation_results = {
            'is_valid': True,
            'errors': [],
            'warnings': [],
            'data_quality_score': 0.0
        }
        
        # Check for missing values
        missing_values = data.isnull().sum()
        if missing_values.any():
            validation_results['warnings'].append(f"Missing values detected: {missing_values.to_dict()}")
        
        # Check data types
        expected_types = self.validation_rules.get('expected_types', {})
        for column, expected_type in expected_types.items():
            if column in data.columns:
                actual_type = str(data[column].dtype)
                if expected_type not in actual_type:
                    validation_results['errors'].append(f"Column {column} has wrong type: {actual_type}")
                    validation_results['is_valid'] = False
        
        # Check data ranges
        range_rules = self.validation_rules.get('ranges', {})
        for column, (min_val, max_val) in range_rules.items():
            if column in data.columns:
                if data[column].min() < min_val or data[column].max() > max_val:
                    validation_results['warnings'].append(f"Column {column} out of expected range")
        
        # Calculate data quality score
        validation_results['data_quality_score'] = self._calculate_quality_score(validation_results)
        
        return validation_results
    
    def _calculate_quality_score(self, validation_results: Dict) -> float:
        """Calculate overall data quality score"""
        score = 1.0
        
        # Deduct for errors
        score -= len(validation_results['errors']) * 0.2
        
        # Deduct for warnings
        score -= len(validation_results['warnings']) * 0.1
        
        return max(0.0, score)

class ModelTrainer:
    """Model training and experimentation"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.mlflow_uri = config.get('mlflow_uri', 'http://localhost:5000')
        mlflow.set_tracking_uri(self.mlflow_uri)
    
    def train_model(self, data: pd.DataFrame, target_column: str, model_class, **kwargs) -> ModelVersion:
        """Train a new model version"""
        start_time = time.time()
        
        # Prepare data
        X = data.drop(columns=[target_column])
        y = data[target_column]
        
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=0.2, random_state=42
        )
        
        # Start MLflow run
        with mlflow.start_run() as run:
            # Train model
            model = model_class(**kwargs)
            model.fit(X_train, y_train)
            
            # Make predictions
            y_pred = model.predict(X_test)
            
            # Calculate metrics
            accuracy = accuracy_score(y_test, y_pred)
            precision = precision_score(y_test, y_pred, average='weighted')
            recall = recall_score(y_test, y_pred, average='weighted')
            f1 = 2 * (precision * recall) / (precision + recall)
            
            training_time = time.time() - start_time
            
            # Log metrics
            mlflow.log_metric("accuracy", accuracy)
            mlflow.log_metric("precision", precision)
            mlflow.log_metric("recall", recall)
            mlflow.log_metric("f1_score", f1)
            mlflow.log_metric("training_time", training_time)
            
            # Log model
            mlflow.sklearn.log_model(model, "model")
            
            # Create model version
            model_version = ModelVersion(
                version=run.info.run_id,
                model_path=f"runs:/{run.info.run_id}/model",
                metrics=ModelMetrics(
                    accuracy=accuracy,
                    precision=precision,
                    recall=recall,
                    f1_score=f1,
                    training_time=training_time,
                    inference_time=0.0  # Will be measured during deployment
                ),
                created_at=datetime.now()
            )
            
            logger.info(f"Model trained successfully: {model_version.version}")
            return model_version

class ModelValidator:
    """Model validation and testing"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.min_accuracy = config.get('min_accuracy', 0.8)
        self.max_inference_time = config.get('max_inference_time', 0.1)
    
    def validate_model(self, model_version: ModelVersion, test_data: pd.DataFrame) -> Dict:
        """Validate model performance and requirements"""
        validation_results = {
            'is_valid': True,
            'errors': [],
            'warnings': [],
            'validation_score': 0.0
        }
        
        # Check accuracy requirements
        if model_version.metrics.accuracy < self.min_accuracy:
            validation_results['errors'].append(
                f"Accuracy {model_version.metrics.accuracy:.3f} below minimum {self.min_accuracy}"
            )
            validation_results['is_valid'] = False
        
        # Check inference time
        if model_version.metrics.inference_time > self.max_inference_time:
            validation_results['warnings'].append(
                f"Inference time {model_version.metrics.inference_time:.3f}s exceeds maximum {self.max_inference_time}s"
            )
        
        # Check for data drift
        drift_score = self._check_data_drift(test_data)
        if drift_score > 0.1:  # 10% drift threshold
            validation_results['warnings'].append(f"Data drift detected: {drift_score:.3f}")
        
        # Calculate validation score
        validation_results['validation_score'] = self._calculate_validation_score(
            model_version.metrics, validation_results
        )
        
        return validation_results
    
    def _check_data_drift(self, data: pd.DataFrame) -> float:
        """Check for data drift (simplified implementation)"""
        # In production, you'd compare with training data statistics
        # This is a placeholder implementation
        return 0.05  # 5% drift
    
    def _calculate_validation_score(self, metrics: ModelMetrics, validation_results: Dict) -> float:
        """Calculate overall validation score"""
        score = metrics.accuracy
        
        # Deduct for errors
        score -= len(validation_results['errors']) * 0.2
        
        # Deduct for warnings
        score -= len(validation_results['warnings']) * 0.1
        
        return max(0.0, score)

class ModelDeployer:
    """Model deployment and serving"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.deployment_endpoint = config.get('deployment_endpoint')
        self.model_registry = {}
    
    def deploy_model(self, model_version: ModelVersion) -> Dict:
        """Deploy model to production"""
        try:
            # Load model from MLflow
            model = mlflow.sklearn.load_model(model_version.model_path)
            
            # Save model locally for serving
            model_path = f"models/{model_version.version}.joblib"
            os.makedirs(os.path.dirname(model_path), exist_ok=True)
            joblib.dump(model, model_path)
            
            # Update model registry
            self.model_registry[model_version.version] = {
                'model_path': model_path,
                'deployed_at': datetime.now(),
                'is_active': True
            }
            
            # Deactivate previous production model
            self._deactivate_previous_models(model_version.version)
            
            # Test deployment
            test_result = self._test_deployment(model_version.version)
            
            deployment_result = {
                'success': True,
                'model_version': model_version.version,
                'endpoint': f"{self.deployment_endpoint}/predict/{model_version.version}",
                'test_result': test_result
            }
            
            logger.info(f"Model deployed successfully: {model_version.version}")
            return deployment_result
            
        except Exception as e:
            logger.error(f"Model deployment failed: {str(e)}")
            return {
                'success': False,
                'error': str(e)
            }
    
    def _deactivate_previous_models(self, current_version: str):
        """Deactivate previous production models"""
        for version, info in self.model_registry.items():
            if version != current_version:
                info['is_active'] = False
    
    def _test_deployment(self, model_version: str) -> Dict:
        """Test deployed model"""
        try:
            # Create test data
            test_data = np.random.rand(1, 10)  # Simplified test data
            
            # Make prediction request
            response = requests.post(
                f"{self.deployment_endpoint}/predict/{model_version}",
                json={'data': test_data.tolist()},
                timeout=10
            )
            
            if response.status_code == 200:
                return {
                    'success': True,
                    'response_time': response.elapsed.total_seconds(),
                    'prediction': response.json()
                }
            else:
                return {
                    'success': False,
                    'error': f"HTTP {response.status_code}"
                }
                
        except Exception as e:
            return {
                'success': False,
                'error': str(e)
            }

class ModelMonitor:
    """Model monitoring and observability"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.monitoring_endpoint = config.get('monitoring_endpoint')
        self.alert_thresholds = config.get('alert_thresholds', {})
    
    def monitor_model(self, model_version: str, predictions: List, actuals: List = None) -> Dict:
        """Monitor model performance in production"""
        monitoring_results = {
            'timestamp': datetime.now(),
            'model_version': model_version,
            'metrics': {},
            'alerts': []
        }
        
        # Calculate performance metrics
        if actuals:
            accuracy = accuracy_score(actuals, predictions)
            monitoring_results['metrics']['accuracy'] = accuracy
            
            # Check for performance degradation
            if accuracy < self.alert_thresholds.get('min_accuracy', 0.8):
                monitoring_results['alerts'].append({
                    'type': 'performance_degradation',
                    'message': f"Accuracy dropped to {accuracy:.3f}",
                    'severity': 'high'
                })
        
        # Check for prediction drift
        prediction_stats = self._calculate_prediction_stats(predictions)
        monitoring_results['metrics']['prediction_stats'] = prediction_stats
        
        # Check for data drift
        data_drift_score = self._check_production_data_drift()
        monitoring_results['metrics']['data_drift_score'] = data_drift_score
        
        if data_drift_score > self.alert_thresholds.get('max_drift', 0.1):
            monitoring_results['alerts'].append({
                'type': 'data_drift',
                'message': f"Data drift detected: {data_drift_score:.3f}",
                'severity': 'medium'
            })
        
        # Send alerts if any
        if monitoring_results['alerts']:
            self._send_alerts(monitoring_results['alerts'])
        
        return monitoring_results
    
    def _calculate_prediction_stats(self, predictions: List) -> Dict:
        """Calculate prediction statistics"""
        predictions_array = np.array(predictions)
        return {
            'mean': float(np.mean(predictions_array)),
            'std': float(np.std(predictions_array)),
            'min': float(np.min(predictions_array)),
            'max': float(np.max(predictions_array))
        }
    
    def _check_production_data_drift(self) -> float:
        """Check for data drift in production (simplified)"""
        # In production, you'd compare with training data
        return 0.05  # 5% drift
    
    def _send_alerts(self, alerts: List[Dict]):
        """Send monitoring alerts"""
        for alert in alerts:
            logger.warning(f"ALERT: {alert['message']}")

class MLOpsPipeline:
    """Complete MLOps pipeline orchestration"""
    
    def __init__(self, config: Dict):
        self.config = config
        self.data_validator = DataValidator(config)
        self.model_trainer = ModelTrainer(config)
        self.model_validator = ModelValidator(config)
        self.model_deployer = ModelDeployer(config)
        self.model_monitor = ModelMonitor(config)
    
    def run_pipeline(self, data: pd.DataFrame, target_column: str, model_class, **kwargs) -> Dict:
        """Run complete MLOps pipeline"""
        pipeline_results = {
            'success': False,
            'model_version': None,
            'deployment_result': None,
            'errors': []
        }
        
        try:
            # Step 1: Data validation
            logger.info("Step 1: Validating data")
            validation_result = self.data_validator.validate_data(data)
            if not validation_result['is_valid']:
                pipeline_results['errors'].extend(validation_result['errors'])
                return pipeline_results
            
            # Step 2: Model training
            logger.info("Step 2: Training model")
            model_version = self.model_trainer.train_model(data, target_column, model_class, **kwargs)
            
            # Step 3: Model validation
            logger.info("Step 3: Validating model")
            model_validation = self.model_validator.validate_model(model_version, data)
            if not model_validation['is_valid']:
                pipeline_results['errors'].extend(model_validation['errors'])
                return pipeline_results
            
            # Step 4: Model deployment
            logger.info("Step 4: Deploying model")
            deployment_result = self.model_deployer.deploy_model(model_version)
            if not deployment_result['success']:
                pipeline_results['errors'].append(deployment_result['error'])
                return pipeline_results
            
            # Step 5: Start monitoring
            logger.info("Step 5: Starting model monitoring")
            # In production, you'd start continuous monitoring here
            
            pipeline_results.update({
                'success': True,
                'model_version': model_version.version,
                'deployment_result': deployment_result
            })
            
            logger.info(f"MLOps pipeline completed successfully: {model_version.version}")
            
        except Exception as e:
            logger.error(f"MLOps pipeline failed: {str(e)}")
            pipeline_results['errors'].append(str(e))
        
        return pipeline_results

# Example usage
if __name__ == "__main__":
    # Configuration
    config = {
        'mlflow_uri': 'http://localhost:5000',
        'deployment_endpoint': 'http://localhost:8000',
        'monitoring_endpoint': 'http://localhost:8001',
        'min_accuracy': 0.8,
        'max_inference_time': 0.1,
        'alert_thresholds': {
            'min_accuracy': 0.8,
            'max_drift': 0.1
        }
    }
    
    # Create pipeline
    pipeline = MLOpsPipeline(config)
    
    # Example: Run pipeline with sample data
    # This would be replaced with actual data loading
    sample_data = pd.DataFrame({
        'feature_1': np.random.randn(1000),
        'feature_2': np.random.randn(1000),
        'target': np.random.randint(0, 2, 1000)
    })
    
    # Run pipeline
    result = pipeline.run_pipeline(sample_data, 'target', 'sklearn.ensemble.RandomForestClassifier')
    print(f"Pipeline result: {result}")