AI Model Performance Optimization in Production

import torch
import torch.quantization as quantization

# Post-training quantization example
def quantize_model(model, data_loader):
    # Set model to evaluation mode
    model.eval()
    
    # Fuse modules for better performance
    model_fused = torch.quantization.fuse_modules(
        model, [['conv', 'bn', 'relu']]
    )
    
    # Prepare model for quantization
    model_prepared = torch.quantization.prepare(
        model_fused, 
        {'': torch.quantization.default_qconfig}
    )
    
    # Calibrate with representative data
    with torch.no_grad():
        for data, _ in data_loader:
            model_prepared(data)
    
    # Convert to quantized model
    model_quantized = torch.quantization.convert(model_prepared)
    
    return model_quantized

# Quantization-aware training
def setup_qat_model(model):
    model.train()
    model.qconfig = torch.quantization.get_default_qat_qconfig('fbgemm')
    model_prepared = torch.quantization.prepare_qat(model)
    return model_prepared

import torch.nn.utils.prune as prune

def apply_structured_pruning(model, amount=0.3):
    """Apply structured pruning to remove entire channels"""
    for name, module in model.named_modules():
        if isinstance(module, torch.nn.Conv2d):
            prune.ln_structured(
                module, 
                name='weight', 
                amount=amount, 
                n=2, 
                dim=0
            )
    return model

def apply_unstructured_pruning(model, amount=0.2):
    """Apply unstructured pruning to remove individual weights"""
    parameters_to_prune = []
    for name, module in model.named_modules():
        if isinstance(module, (torch.nn.Linear, torch.nn.Conv2d)):
            parameters_to_prune.append((module, 'weight'))
    
    prune.global_unstructured(
        parameters_to_prune,
        pruning_method=prune.L1Unstructured,
        amount=amount,
    )
    return model

# Multi-stage Dockerfile for AI model deployment
FROM python:3.9-slim as builder

# Install build dependencies
RUN apt-get update && apt-get install -y \
    build-essential \
    && rm -rf /var/lib/apt/lists/*

# Create virtual environment
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# Install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Production stage
FROM python:3.9-slim

# Copy virtual environment from builder
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# Create non-root user
RUN useradd --create-home --shell /bin/bash app
USER app
WORKDIR /home/app

# Copy application code
COPY --chown=app:app model/ ./model/
COPY --chown=app:app src/ ./src/
COPY --chown=app:app app.py .

# Health check
HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000
CMD ["python", "app.py"]

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-model-deployment
  labels:
    app: ai-model
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ai-model
  template:
    metadata:
      labels:
        app: ai-model
    spec:
      containers:
      - name: ai-model
        image: your-registry/ai-model:latest
        ports:
        - containerPort: 8000
        resources:
          requests:
            memory: "2Gi"
            cpu: "1000m"
          limits:
            memory: "4Gi"
            cpu: "2000m"
        env:
        - name: MODEL_PATH
          value: "/app/model"
        livenessProbe:
          httpGet:
            path: /health
            port: 8000
          initialDelaySeconds: 30
          periodSeconds: 10
        readinessProbe:
          httpGet:
            path: /ready
            port: 8000
          initialDelaySeconds: 5
          periodSeconds: 5
---
apiVersion: v1
kind: Service
metadata:
  name: ai-model-service
spec:
  selector:
    app: ai-model
  ports:
  - protocol: TCP
    port: 80
    targetPort: 8000
  type: LoadBalancer

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

# Define metrics
REQUEST_COUNT = Counter('model_requests_total', 'Total model requests')
REQUEST_LATENCY = Histogram('model_request_duration_seconds', 'Request latency')
GPU_UTILIZATION = Gauge('gpu_utilization_percent', 'GPU utilization percentage')
MEMORY_USAGE = Gauge('memory_usage_bytes', 'Memory usage in bytes')

class ModelMonitor:
    def __init__(self):
        self.start_time = time.time()
    
    def track_request(self, func):
        """Decorator to track request metrics"""
        def wrapper(*args, **kwargs):
            start_time = time.time()
            REQUEST_COUNT.inc()
            
            try:
                result = func(*args, **kwargs)
                return result
            finally:
                REQUEST_LATENCY.observe(time.time() - start_time)
        return wrapper
    
    def update_system_metrics(self):
        """Update system resource metrics"""
        # CPU and memory
        cpu_percent = psutil.cpu_percent()
        memory = psutil.virtual_memory()
        MEMORY_USAGE.set(memory.used)
        
        # GPU metrics (if available)
        try:
            import GPUtil
            gpus = GPUtil.getGPUs()
            if gpus:
                GPU_UTILIZATION.set(gpus[0].load * 100)
        except ImportError:
            pass

# Usage example
monitor = ModelMonitor()

@monitor.track_request
def predict(input_data):
    # Your model prediction logic
    return model.predict(input_data)

# Horizontal Pod Autoscaler
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: ai-model-hpa
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: ai-model-deployment
  minReplicas: 2
  maxReplicas: 10
  metrics:
  - type: Resource
    resource:
      name: cpu
      target:
        type: Utilization
        averageUtilization: 70
  - type: Resource
    resource:
      name: memory
      target:
        type: Utilization
        averageUtilization: 80

import asyncio
import aiohttp
from typing import List
import random

class ModelLoadBalancer:
    def __init__(self, model_endpoints: List[str]):
        self.endpoints = model_endpoints
        self.health_status = {endpoint: True for endpoint in model_endpoints}
    
    async def health_check(self):
        """Check health of all endpoints"""
        async with aiohttp.ClientSession() as session:
            for endpoint in self.endpoints:
                try:
                    async with session.get(f"{endpoint}/health", timeout=5) as response:
                        self.health_status[endpoint] = response.status == 200
                except:
                    self.health_status[endpoint] = False
    
    def get_healthy_endpoints(self) -> List[str]:
        """Return list of healthy endpoints"""
        return [ep for ep, healthy in self.health_status.items() if healthy]
    
    async def predict(self, input_data):
        """Route prediction request to healthy endpoint"""
        healthy_endpoints = self.get_healthy_endpoints()
        
        if not healthy_endpoints:
            raise Exception("No healthy endpoints available")
        
        # Round-robin or random selection
        endpoint = random.choice(healthy_endpoints)
        
        async with aiohttp.ClientSession() as session:
            async with session.post(
                f"{endpoint}/predict", 
                json=input_data,
                timeout=30
            ) as response:
                return await response.json()