MLOps Implementation Patterns and Code Examples

This guide provides practical implementation patterns and code examples for building robust MLOps pipelines for AI/LLM applications.

🚀 CI/CD Pipeline Implementations

GitHub Actions for ML Model Deployment

name: ML Model CI/CD Pipeline
 
on:
  push:
    branches: [main]
    paths:
      - 'models/**'
      - 'src/**'
      - 'tests/**'
 
env:
  MODEL_REGISTRY: ghcr.io/${{ github.repository }}
  PYTHON_VERSION: '3.10'
 
jobs:
  test-and-validate:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Python
        uses: actions/setup-python@v4
        with:
          python-version: ${{ env.PYTHON_VERSION }}
      
      - name: Cache dependencies
        uses: actions/cache@v3
        with:
          path: ~/.cache/pip
          key: ${{ runner.os }}-pip-${{ hashFiles('**/requirements.txt') }}
      
      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install -r requirements-dev.txt
      
      - name: Run unit tests
        run: |
          pytest tests/unit -v --cov=src --cov-report=xml
      
      - name: Validate data schema
        run: |
          python scripts/validate_data_schema.py
      
      - name: Run model performance tests
        run: |
          python scripts/test_model_performance.py \
            --baseline-metrics metrics/baseline.json \
            --threshold 0.95
 
  build-and-push:
    needs: test-and-validate
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Set up Docker Buildx
        uses: docker/setup-buildx-action@v2
      
      - name: Log in to Container Registry
        uses: docker/login-action@v2
        with:
          registry: ghcr.io
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      
      - name: Build and push model serving image
        uses: docker/build-push-action@v4
        with:
          context: .
          push: true
          tags: |
            ${{ env.MODEL_REGISTRY }}/model-server:${{ github.sha }}
            ${{ env.MODEL_REGISTRY }}/model-server:latest
          cache-from: type=gha
          cache-to: type=gha,mode=max
 
  deploy-canary:
    needs: build-and-push
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      
      - name: Configure kubectl
        uses: azure/setup-kubectl@v3
        
      - name: Deploy canary version
        run: |
          kubectl apply -f - <<EOF
          apiVersion: v1
          kind: ConfigMap
          metadata:
            name: canary-rollout-config
          data:
            stages: |
              - duration: 10m
                setWeight: 5
              - duration: 30m
                setWeight: 20
              - duration: 1h
                setWeight: 50
              - duration: 2h
                setWeight: 80
          ---
          apiVersion: argoproj.io/v1alpha1
          kind: Rollout
          metadata:
            name: model-server
          spec:
            replicas: 10
            strategy:
              canary:
                steps:
                - setWeight: 5
                - pause: {duration: 10m}
                - setWeight: 20
                - pause: {duration: 30m}
                - analysis:
                    templates:
                    - templateName: model-performance
                    args:
                    - name: service-name
                      value: model-server
                - setWeight: 50
                - pause: {duration: 1h}
                - setWeight: 80
                - pause: {duration: 2h}
            selector:
              matchLabels:
                app: model-server
            template:
              metadata:
                labels:
                  app: model-server
              spec:
                containers:
                - name: model-server
                  image: ${{ env.MODEL_REGISTRY }}/model-server:${{ github.sha }}
                  ports:
                  - containerPort: 8080
                  resources:
                    requests:
                      memory: "4Gi"
                      cpu: "2"
                      nvidia.com/gpu: "1"
                    limits:
                      memory: "8Gi"
                      cpu: "4"
                      nvidia.com/gpu: "1"
          EOF

Jenkins Pipeline for MLOps

pipeline {
    agent any
    
    environment {
        MLFLOW_TRACKING_URI = credentials('mlflow-tracking-uri')
        MODEL_REGISTRY = 'your-registry.com/ml-models'
        SLACK_WEBHOOK = credentials('slack-webhook')
    }
    
    stages {
        stage('Data Validation') {
            steps {
                script {
                    sh '''
                        python -m data_validator \
                            --input-path s3://data-bucket/training-data/ \
                            --schema-path schemas/training_data_schema.json \
                            --output-report data_validation_report.html
                    '''
                    
                    def validation_passed = sh(
                        script: 'python -m data_validator --check-report data_validation_report.html',
                        returnStatus: true
                    ) == 0
                    
                    if (!validation_passed) {
                        error("Data validation failed. Check report for details.")
                    }
                }
            }
        }
        
        stage('Model Training') {
            steps {
                script {
                    sh '''
                        python train_model.py \
                            --data-path s3://data-bucket/training-data/ \
                            --model-type transformer \
                            --hyperparams config/hyperparameters.yaml \
                            --mlflow-experiment jenkins-training-${BUILD_NUMBER}
                    '''
                }
            }
        }
        
        stage('Model Evaluation') {
            parallel {
                stage('Performance Tests') {
                    steps {
                        sh '''
                            python evaluate_model.py \
                                --model-path runs/latest/model \
                                --test-data s3://data-bucket/test-data/ \
                                --metrics accuracy,f1,precision,recall \
                                --output-report evaluation_report.json
                        '''
                    }
                }
                
                stage('Bias Detection') {
                    steps {
                        sh '''
                            python detect_bias.py \
                                --model-path runs/latest/model \
                                --protected-attributes age,gender,race \
                                --fairness-metrics demographic_parity,equal_opportunity \
                                --output-report bias_report.json
                        '''
                    }
                }
                
                stage('Explainability Analysis') {
                    steps {
                        sh '''
                            python explain_model.py \
                                --model-path runs/latest/model \
                                --sample-size 1000 \
                                --methods shap,lime \
                                --output-report explainability_report.html
                        '''
                    }
                }
            }
        }
        
        stage('Model Registration') {
            when {
                expression {
                    def metrics = readJSON file: 'evaluation_report.json'
                    return metrics.accuracy > 0.95
                }
            }
            steps {
                script {
                    sh '''
                        python register_model.py \
                            --model-path runs/latest/model \
                            --model-name production-model \
                            --stage staging \
                            --tags "jenkins_build=${BUILD_NUMBER}"
                    '''
                }
            }
        }
        
        stage('Deploy to Staging') {
            steps {
                script {
                    sh '''
                        kubectl set image deployment/model-server \
                            model-server=${MODEL_REGISTRY}/model:${BUILD_NUMBER} \
                            -n staging
                    '''
                }
            }
        }
        
        stage('Shadow Testing') {
            steps {
                script {
                    sh '''
                        python shadow_test.py \
                            --staging-endpoint https://staging.api.company.com/predict \
                            --production-endpoint https://api.company.com/predict \
                            --duration 3600 \
                            --sample-rate 0.1 \
                            --output-report shadow_test_report.json
                    '''
                    
                    def shadow_results = readJSON file: 'shadow_test_report.json'
                    if (shadow_results.divergence_rate > 0.05) {
                        error("Shadow testing shows high divergence: ${shadow_results.divergence_rate}")
                    }
                }
            }
        }
        
        stage('Production Deployment') {
            input {
                message "Deploy to production?"
                ok "Deploy"
                parameters {
                    choice(
                        name: 'DEPLOYMENT_STRATEGY',
                        choices: ['canary', 'blue-green', 'rolling'],
                        description: 'Select deployment strategy'
                    )
                }
            }
            steps {
                script {
                    if (params.DEPLOYMENT_STRATEGY == 'canary') {
                        sh '''
                            python deploy_canary.py \
                                --model-version ${BUILD_NUMBER} \
                                --initial-traffic 5 \
                                --increment 10 \
                                --interval 600 \
                                --metrics-threshold 0.95
                        '''
                    } else if (params.DEPLOYMENT_STRATEGY == 'blue-green') {
                        sh '''
                            python deploy_blue_green.py \
                                --model-version ${BUILD_NUMBER} \
                                --health-check-endpoint /health \
                                --switch-timeout 300
                        '''
                    }
                }
            }
        }
    }
    
    post {
        always {
            archiveArtifacts artifacts: '**/*_report.*', fingerprint: true
            
            publishHTML([
                allowMissing: false,
                alwaysLinkToLastBuild: true,
                keepAll: true,
                reportDir: '.',
                reportFiles: 'evaluation_report.html,explainability_report.html',
                reportName: 'ML Model Reports'
            ])
        }
        
        success {
            slackSend(
                channel: '#ml-deployments',
                color: 'good',
                message: "Model deployment successful! Version: ${BUILD_NUMBER}"
            )
        }
        
        failure {
            slackSend(
                channel: '#ml-deployments',
                color: 'danger',
                message: "Model deployment failed! Version: ${BUILD_NUMBER}"
            )
        }
    }
}

🔄 Model Versioning Implementation

DVC Pipeline Configuration

# dvc.yaml
stages:
  prepare_data:
    cmd: python src/prepare_data.py
    deps:
      - src/prepare_data.py
      - data/raw
    params:
      - prepare.split_ratio
      - prepare.random_seed
    outs:
      - data/prepared
 
  train_model:
    cmd: python src/train_model.py
    deps:
      - src/train_model.py
      - data/prepared
    params:
      - train.epochs
      - train.batch_size
      - train.learning_rate
      - train.model_architecture
    outs:
      - models/model.pkl
      - models/tokenizer.pkl
    metrics:
      - metrics/train_metrics.json:
          cache: false
 
  evaluate_model:
    cmd: python src/evaluate_model.py
    deps:
      - src/evaluate_model.py
      - models/model.pkl
      - data/prepared/test.csv
    metrics:
      - metrics/eval_metrics.json:
          cache: false
    plots:
      - plots/confusion_matrix.png
      - plots/roc_curve.png

MLflow Model Registry Integration

import mlflow
import mlflow.pytorch
from mlflow.tracking import MlflowClient
import torch
import json
from datetime import datetime
 
class ModelVersionManager:
    def __init__(self, tracking_uri="http://mlflow-server:5000"):
        mlflow.set_tracking_uri(tracking_uri)
        self.client = MlflowClient()
    
    def register_model(self, model, model_name, metrics, tags=None):
        """Register a new model version with comprehensive metadata"""
        
        with mlflow.start_run() as run:
            # Log model
            mlflow.pytorch.log_model(
                model,
                "model",
                registered_model_name=model_name,
                signature=mlflow.models.infer_signature(
                    model_input=sample_input,
                    model_output=model(sample_input)
                )
            )
            
            # Log metrics
            for metric_name, metric_value in metrics.items():
                mlflow.log_metric(metric_name, metric_value)
            
            # Log parameters
            mlflow.log_params({
                "model_architecture": model.__class__.__name__,
                "num_parameters": sum(p.numel() for p in model.parameters()),
                "training_framework": "pytorch",
                "framework_version": torch.__version__
            })
            
            # Log tags
            if tags:
                mlflow.set_tags(tags)
            
            # Log additional artifacts
            mlflow.log_artifact("config/model_config.yaml")
            mlflow.log_artifact("requirements.txt")
            
            run_id = run.info.run_id
        
        # Transition model to staging
        model_version = self._get_latest_version(model_name)
        self.client.transition_model_version_stage(
            name=model_name,
            version=model_version,
            stage="Staging",
            archive_existing_versions=True
        )
        
        return run_id, model_version
    
    def promote_to_production(self, model_name, version=None, 
                            min_accuracy=0.95, max_latency_ms=100):
        """Promote model to production with validation checks"""
        
        if version is None:
            version = self._get_latest_version(model_name, stage="Staging")
        
        # Get model metrics
        model_version = self.client.get_model_version(model_name, version)
        run_id = model_version.run_id
        run = self.client.get_run(run_id)
        metrics = run.data.metrics
        
        # Validation checks
        if metrics.get("accuracy", 0) < min_accuracy:
            raise ValueError(f"Model accuracy {metrics['accuracy']} below threshold {min_accuracy}")
        
        if metrics.get("inference_latency_ms", float('inf')) > max_latency_ms:
            raise ValueError(f"Model latency {metrics['inference_latency_ms']}ms exceeds {max_latency_ms}ms")
        
        # Check for bias metrics
        bias_metrics = {k: v for k, v in metrics.items() if k.startswith("bias_")}
        if any(v > 0.1 for v in bias_metrics.values()):
            raise ValueError(f"Model shows significant bias: {bias_metrics}")
        
        # Transition to production
        self.client.transition_model_version_stage(
            name=model_name,
            version=version,
            stage="Production",
            archive_existing_versions=True
        )
        
        # Tag the production model
        self.client.set_model_version_tag(
            name=model_name,
            version=version,
            key="promoted_at",
            value=datetime.utcnow().isoformat()
        )
        
        return version
    
    def rollback_model(self, model_name, reason="Performance degradation"):
        """Rollback to previous production version"""
        
        # Get current and previous production versions
        current_prod = self._get_latest_version(model_name, stage="Production")
        all_versions = self.client.search_model_versions(f"name='{model_name}'")
        
        # Find previous production version
        prod_versions = [
            v for v in all_versions 
            if v.current_stage == "Archived" and 
            "Production" in v.tags.get("previous_stages", "")
        ]
        
        if not prod_versions:
            raise ValueError("No previous production version found for rollback")
        
        previous_version = sorted(prod_versions, key=lambda x: x.version)[-1].version
        
        # Transition current to archived
        self.client.transition_model_version_stage(
            name=model_name,
            version=current_prod,
            stage="Archived"
        )
        
        # Transition previous to production
        self.client.transition_model_version_stage(
            name=model_name,
            version=previous_version,
            stage="Production"
        )
        
        # Log rollback event
        self.client.set_model_version_tag(
            name=model_name,
            version=previous_version,
            key="rollback_reason",
            value=reason
        )
        
        return previous_version
    
    def _get_latest_version(self, model_name, stage=None):
        """Get latest version number for a model"""
        filter_string = f"name='{model_name}'"
        if stage:
            filter_string += f" AND current_stage='{stage}'"
        
        versions = self.client.search_model_versions(filter_string)
        if not versions:
            raise ValueError(f"No model versions found for {model_name}")
        
        return max(v.version for v in versions)

🚦 A/B Testing Framework

Feature Flag Based A/B Testing

import hashlib
import json
from typing import Dict, Any, Optional
from dataclasses import dataclass
from datetime import datetime
import redis
import numpy as np
 
@dataclass
class Experiment:
    name: str
    variants: Dict[str, float]  # variant_name -> traffic_percentage
    metrics: List[str]
    start_date: datetime
    end_date: Optional[datetime] = None
    
class ABTestingFramework:
    def __init__(self, redis_client: redis.Redis):
        self.redis = redis_client
        self.experiments = {}
        
    def create_experiment(self, experiment: Experiment):
        """Create a new A/B test experiment"""
        
        # Validate traffic allocation
        total_traffic = sum(experiment.variants.values())
        if not np.isclose(total_traffic, 1.0):
            raise ValueError(f"Traffic allocation must sum to 1.0, got {total_traffic}")
        
        # Store experiment configuration
        exp_data = {
            "name": experiment.name,
            "variants": experiment.variants,
            "metrics": experiment.metrics,
            "start_date": experiment.start_date.isoformat(),
            "end_date": experiment.end_date.isoformat() if experiment.end_date else None,
            "created_at": datetime.utcnow().isoformat()
        }
        
        self.redis.hset(
            "experiments",
            experiment.name,
            json.dumps(exp_data)
        )
        
        # Initialize metric counters
        for variant in experiment.variants:
            for metric in experiment.metrics:
                key = f"metrics:{experiment.name}:{variant}:{metric}"
                self.redis.set(key, 0)
    
    def get_variant(self, experiment_name: str, user_id: str) -> str:
        """Deterministically assign user to variant"""
        
        # Load experiment
        exp_data = self.redis.hget("experiments", experiment_name)
        if not exp_data:
            raise ValueError(f"Experiment {experiment_name} not found")
        
        experiment = json.loads(exp_data)
        
        # Check if experiment is active
        now = datetime.utcnow()
        start_date = datetime.fromisoformat(experiment["start_date"])
        end_date = datetime.fromisoformat(experiment["end_date"]) if experiment["end_date"] else None
        
        if now < start_date or (end_date and now > end_date):
            return "control"  # Default to control if experiment not active
        
        # Deterministic assignment based on user_id
        hash_value = int(hashlib.md5(f"{experiment_name}:{user_id}".encode()).hexdigest(), 16)
        normalized_hash = (hash_value % 10000) / 10000.0
        
        # Assign to variant based on traffic allocation
        cumulative_traffic = 0.0
        for variant, traffic_pct in experiment["variants"].items():
            cumulative_traffic += traffic_pct
            if normalized_hash < cumulative_traffic:
                return variant
        
        return list(experiment["variants"].keys())[-1]  # Fallback
    
    def track_metric(self, experiment_name: str, variant: str, metric: str, value: float = 1.0):
        """Track metric for experiment variant"""
        
        # Increment counter
        counter_key = f"metrics:{experiment_name}:{variant}:{metric}:count"
        sum_key = f"metrics:{experiment_name}:{variant}:{metric}:sum"
        
        self.redis.incr(counter_key, 1)
        self.redis.incrbyfloat(sum_key, value)
        
        # Track in time series for monitoring
        timestamp = int(datetime.utcnow().timestamp())
        ts_key = f"timeseries:{experiment_name}:{variant}:{metric}"
        self.redis.zadd(ts_key, {f"{timestamp}:{value}": timestamp})
    
    def get_results(self, experiment_name: str) -> Dict[str, Any]:
        """Get experiment results with statistical significance"""
        
        exp_data = json.loads(self.redis.hget("experiments", experiment_name))
        results = {}
        
        for variant in exp_data["variants"]:
            variant_results = {}
            
            for metric in exp_data["metrics"]:
                count_key = f"metrics:{experiment_name}:{variant}:{metric}:count"
                sum_key = f"metrics:{experiment_name}:{variant}:{metric}:sum"
                
                count = int(self.redis.get(count_key) or 0)
                total = float(self.redis.get(sum_key) or 0)
                
                variant_results[metric] = {
                    "count": count,
                    "total": total,
                    "average": total / count if count > 0 else 0
                }
            
            results[variant] = variant_results
        
        # Calculate statistical significance
        if len(exp_data["variants"]) == 2 and "control" in exp_data["variants"]:
            treatment_variant = [v for v in exp_data["variants"] if v != "control"][0]
            
            for metric in exp_data["metrics"]:
                control_data = results["control"][metric]
                treatment_data = results[treatment_variant][metric]
                
                # Simple two-proportion z-test
                if control_data["count"] > 30 and treatment_data["count"] > 30:
                    p1 = control_data["average"]
                    p2 = treatment_data["average"]
                    n1 = control_data["count"]
                    n2 = treatment_data["count"]
                    
                    pooled_p = (p1 * n1 + p2 * n2) / (n1 + n2)
                    se = np.sqrt(pooled_p * (1 - pooled_p) * (1/n1 + 1/n2))
                    
                    if se > 0:
                        z_score = (p2 - p1) / se
                        p_value = 2 * (1 - stats.norm.cdf(abs(z_score)))
                        
                        results[treatment_variant][metric]["lift"] = (p2 - p1) / p1 if p1 > 0 else 0
                        results[treatment_variant][metric]["p_value"] = p_value
                        results[treatment_variant][metric]["significant"] = p_value < 0.05
        
        return results

Model A/B Testing Service

from flask import Flask, request, jsonify
import torch
import logging
from prometheus_client import Counter, Histogram, generate_latest
import time
 
app = Flask(__name__)
 
# Metrics
prediction_counter = Counter(
    'model_predictions_total',
    'Total number of predictions',
    ['model_version', 'variant']
)
 
prediction_latency = Histogram(
    'model_prediction_duration_seconds',
    'Prediction latency',
    ['model_version', 'variant']
)
 
class ModelABService:
    def __init__(self, ab_framework: ABTestingFramework):
        self.ab_framework = ab_framework
        self.models = {}
        self.experiment_name = "model_v2_test"
        
    def load_models(self):
        """Load multiple model versions for A/B testing"""
        
        # Load control model (current production)
        self.models["control"] = torch.load("models/production/model_v1.pt")
        self.models["control"].eval()
        
        # Load treatment model (new version)
        self.models["treatment"] = torch.load("models/staging/model_v2.pt")
        self.models["treatment"].eval()
        
    @app.route('/predict', methods=['POST'])
    def predict(self):
        """Handle prediction request with A/B testing"""
        
        start_time = time.time()
        
        try:
            # Extract user ID and input
            data = request.json
            user_id = data.get('user_id', 'anonymous')
            input_data = torch.tensor(data['input'])
            
            # Get variant for user
            variant = self.ab_framework.get_variant(self.experiment_name, user_id)
            
            # Select model based on variant
            model = self.models[variant]
            model_version = "v1" if variant == "control" else "v2"
            
            # Make prediction
            with torch.no_grad():
                prediction = model(input_data)
                result = prediction.numpy().tolist()
            
            # Track metrics
            latency = time.time() - start_time
            prediction_counter.labels(
                model_version=model_version,
                variant=variant
            ).inc()
            
            prediction_latency.labels(
                model_version=model_version,
                variant=variant
            ).observe(latency)
            
            # Track business metrics
            if data.get('track_conversion'):
                self.ab_framework.track_metric(
                    self.experiment_name,
                    variant,
                    'conversion',
                    1.0 if data.get('converted') else 0.0
                )
            
            return jsonify({
                'prediction': result,
                'model_version': model_version,
                'variant': variant,
                'latency_ms': latency * 1000
            })
            
        except Exception as e:
            logging.error(f"Prediction error: {str(e)}")
            return jsonify({'error': str(e)}), 500
    
    @app.route('/metrics')
    def metrics(self):
        """Expose Prometheus metrics"""
        return generate_latest()

📊 Monitoring and Observability Setup

Comprehensive LLM Monitoring

import time
import json
import logging
from typing import Dict, Any, Optional, List
from datetime import datetime
from dataclasses import dataclass, asdict
import openai
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
import tiktoken
 
# Setup OpenTelemetry
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)
 
# Add OTLP exporter
otlp_exporter = OTLPSpanExporter(endpoint="http://localhost:4317")
span_processor = BatchSpanProcessor(otlp_exporter)
trace.get_tracer_provider().add_span_processor(span_processor)
 
@dataclass
class LLMMetrics:
    request_id: str
    model: str
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int
    latency_ms: float
    temperature: float
    user_id: str
    timestamp: datetime
    prompt_hash: str
    response_hash: str
    estimated_cost: float
    
class LLMObservability:
    def __init__(self, datadog_api_key: Optional[str] = None):
        self.encoding = tiktoken.get_encoding("cl100k_base")
        self.datadog_api_key = datadog_api_key
        self.safety_checks = SafetyChecker()
        
    def trace_llm_call(self, func):
        """Decorator to trace LLM API calls"""
        
        def wrapper(*args, **kwargs):
            with tracer.start_as_current_span("llm_call") as span:
                # Extract parameters
                prompt = kwargs.get('prompt', args[0] if args else '')
                model = kwargs.get('model', 'gpt-4')
                temperature = kwargs.get('temperature', 0.7)
                user_id = kwargs.get('user_id', 'anonymous')
                
                # Set span attributes
                span.set_attribute("llm.model", model)
                span.set_attribute("llm.temperature", temperature)
                span.set_attribute("llm.prompt_length", len(prompt))
                span.set_attribute("user.id", user_id)
                
                # Count tokens
                prompt_tokens = len(self.encoding.encode(prompt))
                span.set_attribute("llm.prompt_tokens", prompt_tokens)
                
                # Safety checks
                safety_result = self.safety_checks.check_prompt(prompt)
                if not safety_result.is_safe:
                    span.set_attribute("llm.safety_violation", True)
                    span.set_attribute("llm.safety_reason", safety_result.reason)
                    raise ValueError(f"Unsafe prompt detected: {safety_result.reason}")
                
                # Execute LLM call
                start_time = time.time()
                try:
                    result = func(*args, **kwargs)
                    latency_ms = (time.time() - start_time) * 1000
                    
                    # Extract response
                    if isinstance(result, dict):
                        response_text = result.get('choices', [{}])[0].get('text', '')
                        completion_tokens = result.get('usage', {}).get('completion_tokens', 0)
                    else:
                        response_text = str(result)
                        completion_tokens = len(self.encoding.encode(response_text))
                    
                    # Set response attributes
                    span.set_attribute("llm.completion_tokens", completion_tokens)
                    span.set_attribute("llm.total_tokens", prompt_tokens + completion_tokens)
                    span.set_attribute("llm.latency_ms", latency_ms)
                    span.set_attribute("llm.response_length", len(response_text))
                    
                    # Check for hallucinations
                    hallucination_score = self.safety_checks.check_hallucination(
                        prompt, response_text
                    )
                    span.set_attribute("llm.hallucination_score", hallucination_score)
                    
                    # Log metrics
                    metrics = LLMMetrics(
                        request_id=span.get_span_context().span_id,
                        model=model,
                        prompt_tokens=prompt_tokens,
                        completion_tokens=completion_tokens,
                        total_tokens=prompt_tokens + completion_tokens,
                        latency_ms=latency_ms,
                        temperature=temperature,
                        user_id=user_id,
                        timestamp=datetime.utcnow(),
                        prompt_hash=hashlib.md5(prompt.encode()).hexdigest(),
                        response_hash=hashlib.md5(response_text.encode()).hexdigest(),
                        estimated_cost=self._calculate_cost(
                            model, prompt_tokens, completion_tokens
                        )
                    )
                    
                    self._send_metrics(metrics)
                    
                    return result
                    
                except Exception as e:
                    span.record_exception(e)
                    span.set_status(trace.Status(trace.StatusCode.ERROR))
                    raise
                    
        return wrapper
    
    def _calculate_cost(self, model: str, prompt_tokens: int, completion_tokens: int) -> float:
        """Calculate estimated cost based on model and token usage"""
        
        # Pricing as of 2024 (example rates)
        pricing = {
            "gpt-4": {"prompt": 0.03, "completion": 0.06},
            "gpt-4-turbo": {"prompt": 0.01, "completion": 0.03},
            "gpt-3.5-turbo": {"prompt": 0.0005, "completion": 0.0015},
            "claude-3-opus": {"prompt": 0.015, "completion": 0.075},
            "claude-3-sonnet": {"prompt": 0.003, "completion": 0.015}
        }
        
        rates = pricing.get(model, {"prompt": 0.01, "completion": 0.01})
        
        prompt_cost = (prompt_tokens / 1000) * rates["prompt"]
        completion_cost = (completion_tokens / 1000) * rates["completion"]
        
        return prompt_cost + completion_cost
    
    def _send_metrics(self, metrics: LLMMetrics):
        """Send metrics to monitoring backend"""
        
        # Log locally
        logging.info(f"LLM Metrics: {json.dumps(asdict(metrics), default=str)}")
        
        # Send to Datadog if configured
        if self.datadog_api_key:
            # Implementation for Datadog metrics API
            pass
        
        # Store in time-series database
        self._store_metrics_influxdb(metrics)
    
    def _store_metrics_influxdb(self, metrics: LLMMetrics):
        """Store metrics in InfluxDB for analysis"""
        
        from influxdb_client import InfluxDBClient, Point
        
        client = InfluxDBClient(
            url="http://localhost:8086",
            token="your-token",
            org="your-org"
        )
        
        write_api = client.write_api()
        
        point = Point("llm_metrics") \
            .tag("model", metrics.model) \
            .tag("user_id", metrics.user_id) \
            .field("prompt_tokens", metrics.prompt_tokens) \
            .field("completion_tokens", metrics.completion_tokens) \
            .field("total_tokens", metrics.total_tokens) \
            .field("latency_ms", metrics.latency_ms) \
            .field("temperature", metrics.temperature) \
            .field("estimated_cost", metrics.estimated_cost) \
            .time(metrics.timestamp)
        
        write_api.write(bucket="ml_metrics", record=point)
 
class SafetyChecker:
    """Check for safety issues in LLM interactions"""
    
    def check_prompt(self, prompt: str) -> SafetyResult:
        """Check prompt for safety issues"""
        
        # Check for injection attempts
        injection_patterns = [
            "ignore previous instructions",
            "disregard all prior commands",
            "new system prompt:",
            "你是" # Multi-language injection attempts
        ]
        
        for pattern in injection_patterns:
            if pattern.lower() in prompt.lower():
                return SafetyResult(
                    is_safe=False,
                    reason=f"Potential injection attempt: {pattern}"
                )
        
        # Check for PII
        if self._contains_pii(prompt):
            return SafetyResult(
                is_safe=False,
                reason="Prompt contains potential PII"
            )
        
        return SafetyResult(is_safe=True)
    
    def check_hallucination(self, prompt: str, response: str) -> float:
        """Calculate hallucination score (0-1, higher = more likely hallucination)"""
        
        # Simple heuristic - in production, use specialized models
        factual_keywords = ["according to", "research shows", "studies indicate"]
        confidence_phrases = ["I think", "probably", "might be", "I'm not sure"]
        
        score = 0.0
        
        # Check for unfounded factual claims
        for keyword in factual_keywords:
            if keyword in response and keyword not in prompt:
                score += 0.2
        
        # Check for low confidence indicators
        for phrase in confidence_phrases:
            if phrase in response:
                score -= 0.1
        
        return max(0.0, min(1.0, score))
    
    def _contains_pii(self, text: str) -> bool:
        """Check for potential PII in text"""
        
        import re
        
        # Simple patterns - enhance with more sophisticated detection
        patterns = {
            'ssn': r'\b\d{3}-\d{2}-\d{4}\b',
            'credit_card': r'\b\d{4}[\s-]?\d{4}[\s-]?\d{4}[\s-]?\d{4}\b',
            'email': r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b',
            'phone': r'\b\d{3}[-.]?\d{3}[-.]?\d{4}\b'
        }
        
        for pattern_name, pattern in patterns.items():
            if re.search(pattern, text):
                return True
        
        return False

Grafana Dashboard Configuration

{
  "dashboard": {
    "title": "LLM Operations Dashboard",
    "panels": [
      {
        "title": "Request Rate by Model",
        "type": "graph",
        "targets": [
          {
            "expr": "rate(llm_requests_total[5m])",
            "legendFormat": "{{model}}"
          }
        ]
      },
      {
        "title": "Token Usage and Cost",
        "type": "graph",
        "targets": [
          {
            "expr": "sum(rate(llm_tokens_total[5m])) by (model)",
            "legendFormat": "{{model}} tokens/sec"
          },
          {
            "expr": "sum(rate(llm_cost_dollars[5m])) by (model) * 3600",
            "legendFormat": "{{model}} $/hour"
          }
        ]
      },
      {
        "title": "Latency Percentiles",
        "type": "graph",
        "targets": [
          {
            "expr": "histogram_quantile(0.99, rate(llm_request_duration_seconds_bucket[5m]))",
            "legendFormat": "p99"
          },
          {
            "expr": "histogram_quantile(0.95, rate(llm_request_duration_seconds_bucket[5m]))",
            "legendFormat": "p95"
          },
          {
            "expr": "histogram_quantile(0.50, rate(llm_request_duration_seconds_bucket[5m]))",
            "legendFormat": "p50"
          }
        ]
      },
      {
        "title": "Safety Violations",
        "type": "stat",
        "targets": [
          {
            "expr": "sum(rate(llm_safety_violations_total[1h]))"
          }
        ]
      },
      {
        "title": "Hallucination Score Distribution",
        "type": "heatmap",
        "targets": [
          {
            "expr": "llm_hallucination_score_bucket"
          }
        ]
      },
      {
        "title": "Model Performance Comparison",
        "type": "table",
        "targets": [
          {
            "expr": "avg_over_time(model_accuracy[1h])",
            "format": "table"
          }
        ]
      }
    ]
  }
}

🏗️ Infrastructure as Code Examples

Terraform Configuration for ML Platform

# versions.tf
terraform {
  required_version = ">= 1.0"
  
  required_providers {
    aws = {
      source  = "hashicorp/aws"
      version = "~> 5.0"
    }
    kubernetes = {
      source  = "hashicorp/kubernetes"
      version = "~> 2.23"
    }
    helm = {
      source  = "hashicorp/helm"
      version = "~> 2.11"
    }
  }
}
 
# modules/ml-platform/main.tf
module "ml_platform" {
  source = "./modules/ml-platform"
  
  environment = var.environment
  region      = var.aws_region
  
  # EKS Configuration
  eks_config = {
    cluster_name    = "${var.project_name}-ml-cluster"
    cluster_version = "1.28"
    
    node_groups = {
      cpu_nodes = {
        instance_types = ["m5.2xlarge"]
        min_size       = 2
        max_size       = 10
        desired_size   = 3
      }
      
      gpu_nodes = {
        instance_types = ["p3.2xlarge"]
        min_size       = 0
        max_size       = 5
        desired_size   = 1
        
        taints = [{
          key    = "nvidia.com/gpu"
          value  = "true"
          effect = "NO_SCHEDULE"
        }]
        
        labels = {
          "workload-type" = "gpu"
        }
      }
    }
  }
  
  # S3 Buckets for ML Artifacts
  ml_buckets = {
    data = {
      name = "${var.project_name}-ml-data"
      versioning = true
      lifecycle_rules = [{
        id      = "archive-old-data"
        enabled = true
        
        transition = [{
          days          = 90
          storage_class = "GLACIER"
        }]
      }]
    }
    
    models = {
      name = "${var.project_name}-ml-models"
      versioning = true
      replication = {
        region = "us-east-1"
        bucket = "${var.project_name}-ml-models-replica"
      }
    }
    
    experiments = {
      name = "${var.project_name}-ml-experiments"
      versioning = true
    }
  }
  
  # RDS for MLflow Backend
  mlflow_db = {
    engine         = "postgres"
    engine_version = "15.4"
    instance_class = "db.r6g.large"
    storage_size   = 100
    
    backup_retention_period = 30
    backup_window          = "03:00-04:00"
    maintenance_window     = "sun:04:00-sun:05:00"
    
    high_availability = true
  }
  
  # Redis for Feature Store Cache
  redis_config = {
    node_type = "cache.r6g.large"
    num_cache_nodes = 3
    
    automatic_failover_enabled = true
    multi_az_enabled          = true
    
    snapshot_retention_limit = 7
    snapshot_window         = "03:00-05:00"
  }
}
 
# modules/ml-platform/monitoring.tf
resource "helm_release" "prometheus_operator" {
  name       = "prometheus-operator"
  repository = "https://prometheus-community.github.io/helm-charts"
  chart      = "kube-prometheus-stack"
  namespace  = "monitoring"
  
  values = [
    yamlencode({
      prometheus = {
        prometheusSpec = {
          retention = "30d"
          
          storageSpec = {
            volumeClaimTemplate = {
              spec = {
                accessModes = ["ReadWriteOnce"]
                resources = {
                  requests = {
                    storage = "100Gi"
                  }
                }
              }
            }
          }
          
          additionalScrapeConfigs = [
            {
              job_name = "ml-models"
              kubernetes_sd_configs = [{
                role = "pod"
                selectors = [{
                  role = "pod"
                  label = "app=model-server"
                }]
              }]
            }
          ]
        }
      }
      
      grafana = {
        adminPassword = random_password.grafana_admin.result
        
        dashboardProviders = {
          dashboardproviders.yaml = {
            apiVersion = 1
            providers = [{
              name      = "ml-dashboards"
              folder    = "ML Operations"
              type      = "file"
              disableDeletion = false
              editable  = true
              options = {
                path = "/var/lib/grafana/dashboards/ml-dashboards"
              }
            }]
          }
        }
        
        dashboards = {
          ml-dashboards = {
            "model-performance" = file("${path.module}/dashboards/model-performance.json")
            "training-metrics"  = file("${path.module}/dashboards/training-metrics.json")
            "inference-metrics" = file("${path.module}/dashboards/inference-metrics.json")
          }
        }
      }
    })
  ]
}
 
# modules/ml-platform/mlflow.tf
resource "helm_release" "mlflow" {
  name       = "mlflow"
  repository = "https://community-charts.github.io/helm-charts"
  chart      = "mlflow"
  namespace  = "ml-platform"
  
  values = [
    yamlencode({
      backend = {
        store_uri = "postgresql://${aws_db_instance.mlflow.username}:${random_password.mlflow_db.result}@${aws_db_instance.mlflow.endpoint}/${aws_db_instance.mlflow.name}"
        
        default_artifact_root = "s3://${aws_s3_bucket.ml_experiments.id}/mlflow"
      }
      
      service = {
        type = "LoadBalancer"
        annotations = {
          "service.beta.kubernetes.io/aws-load-balancer-type" = "nlb"
        }
      }
      
      ingress = {
        enabled = true
        className = "nginx"
        
        hosts = [{
          host = "mlflow.${var.domain_name}"
          paths = [{
            path = "/"
            pathType = "Prefix"
          }]
        }]
        
        tls = [{
          secretName = "mlflow-tls"
          hosts = ["mlflow.${var.domain_name}"]
        }]
      }
    })
  ]
  
  depends_on = [
    aws_db_instance.mlflow,
    aws_s3_bucket.ml_experiments
  ]
}

🚀 Production Deployment Patterns

Kubernetes Deployment with Argo Rollouts

apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: llm-model-server
  namespace: production
spec:
  replicas: 10
  strategy:
    blueGreen:
      activeService: llm-model-active
      previewService: llm-model-preview
      autoPromotionEnabled: false
      scaleDownDelaySeconds: 30
      prePromotionAnalysis:
        templates:
        - templateName: model-performance-analysis
        args:
        - name: service-name
          value: llm-model-preview
      postPromotionAnalysis:
        templates:
        - templateName: model-performance-analysis
        args:
        - name: service-name
          value: llm-model-active
  selector:
    matchLabels:
      app: llm-model-server
  template:
    metadata:
      labels:
        app: llm-model-server
    spec:
      containers:
      - name: model-server
        image: your-registry/llm-model-server:v2.0.0
        ports:
        - containerPort: 8080
        env:
        - name: MODEL_PATH
          value: "/models/llama-2-7b"
        - name: MAX_BATCH_SIZE
          value: "32"
        - name: GPU_MEMORY_FRACTION
          value: "0.9"
        resources:
          requests:
            memory: "16Gi"
            cpu: "4"
            nvidia.com/gpu: "1"
          limits:
            memory: "32Gi"
            cpu: "8"
            nvidia.com/gpu: "1"
        readinessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 60
          periodSeconds: 10
        livenessProbe:
          httpGet:
            path: /health
            port: 8080
          initialDelaySeconds: 120
          periodSeconds: 30
        volumeMounts:
        - name: model-cache
          mountPath: /models
        - name: shm
          mountPath: /dev/shm
      volumes:
      - name: model-cache
        persistentVolumeClaim:
          claimName: model-cache-pvc
      - name: shm
        emptyDir:
          medium: Memory
          sizeLimit: 16Gi
---
apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: model-performance-analysis
spec:
  args:
  - name: service-name
  metrics:
  - name: accuracy
    interval: 5m
    successCondition: result >= 0.95
    failureLimit: 3
    provider:
      prometheus:
        address: http://prometheus-server.monitoring:9090
        query: |
          avg(
            rate(model_predictions_correct_total{service="{{args.service-name}}"}[5m]) /
            rate(model_predictions_total{service="{{args.service-name}}"}[5m])
          )
  - name: latency
    interval: 5m
    successCondition: result < 200
    failureLimit: 3
    provider:
      prometheus:
        address: http://prometheus-server.monitoring:9090
        query: |
          histogram_quantile(0.99,
            sum(rate(http_request_duration_seconds_bucket{service="{{args.service-name}}"}[5m]))
            by (le)
          ) * 1000
  - name: error-rate
    interval: 5m
    successCondition: result < 0.01
    failureLimit: 3
    provider:
      prometheus:
        address: http://prometheus-server.monitoring:9090
        query: |
          sum(rate(http_requests_total{service="{{args.service-name}}",status=~"5.."}[5m])) /
          sum(rate(http_requests_total{service="{{args.service-name}}"}[5m]))

This comprehensive guide provides practical implementation patterns and code examples for building robust MLOps pipelines. The examples demonstrate real-world scenarios including CI/CD pipelines, model versioning, A/B testing frameworks, monitoring setups, and infrastructure as code configurations that can be adapted to specific organizational needs.