Agent skill

mlops-advanced

Implement advanced MLOps practices for production ML systems. Use for: building CI/CD pipelines for ML models, implementing continuous training and monitoring, managing model registries and versioning, deploying with blue-green and canary strategies, monitoring model drift and performance degradation, orchestrating ML workflows with Kubeflow/Airflow, implementing feature stores, ensuring model governance and compliance, and scaling ML infrastructure with Kubernetes.

View SKILL.md on GitHub Repository

Stars 163

Forks 31

Install this agent skill to your Project

npx add-skill https://github.com/majiayu000/claude-skill-registry/tree/main/skills/other/other/mlops-advanced

SKILL.md

MLOps Advanced

Implement production-grade MLOps practices for scalable, reliable machine learning systems.

Overview

MLOps (Machine Learning Operations) extends DevOps principles to machine learning, automating and streamlining the entire ML lifecycle from development to deployment and monitoring. Advanced MLOps addresses unique ML challenges including data dependencies, model versioning, continuous training, performance monitoring, and infrastructure scaling. It encompasses CI/CD for ML, automated retraining, drift detection, and governance frameworks.

MLOps Maturity Levels

Level 0 - Manual Process:

Manual data preparation and training
Notebook-driven development
Manual deployment
No CI/CD or monitoring

Level 1 - ML Pipeline Automation:

Automated training pipelines
Continuous training (CT)
Model and data validation
Automated deployment of prediction service

Level 2 - CI/CD Pipeline Automation:

Full CI/CD system for ML
Automated testing (data, model, infrastructure)
Rapid experimentation and deployment
Comprehensive monitoring and logging

CI/CD for Machine Learning

Continuous Integration Components

yaml

# .github/workflows/ml-ci.yml
name: ML CI Pipeline

on: [push, pull_request]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v2
      
      - name: Data Validation
        run: |
          python scripts/validate_data.py
          python scripts/check_data_drift.py
      
      - name: Feature Validation
        run: python scripts/validate_features.py
      
      - name: Model Training
        run: python scripts/train_model.py
      
      - name: Model Evaluation
        run: |
          python scripts/evaluate_model.py
          python scripts/check_fairness.py
      
      - name: Unit Tests
        run: pytest tests/
      
      - name: Integration Tests
        run: pytest tests/integration/

Continuous Deployment

python

# deploy_model.py
import mlflow
from kubernetes import client, config

def deploy_model(model_uri, deployment_name, namespace='production'):
    # Load model from registry
    model = mlflow.pyfunc.load_model(model_uri)
    
    # Create deployment configuration
    deployment = create_k8s_deployment(
        name=deployment_name,
        image=f'ml-model-server:{model.metadata.run_id}',
        replicas=3,
        resources={'cpu': '2', 'memory': '4Gi'}
    )
    
    # Deploy to Kubernetes
    config.load_kube_config()
    apps_v1 = client.AppsV1Api()
    apps_v1.create_namespaced_deployment(namespace, deployment)
    
    # Create service
    service = create_k8s_service(deployment_name)
    core_v1 = client.CoreV1Api()
    core_v1.create_namespaced_service(namespace, service)

Model Registry and Versioning

MLflow Model Registry

python

import mlflow
from mlflow.tracking import MlflowClient

# Register model
mlflow.set_tracking_uri("http://mlflow-server:5000")
client = MlflowClient()

# Log and register model
with mlflow.start_run():
    mlflow.log_params(params)
    mlflow.log_metrics(metrics)
    mlflow.sklearn.log_model(model, "model")
    
    # Register to model registry
    model_uri = f"runs:/{mlflow.active_run().info.run_id}/model"
    mlflow.register_model(model_uri, "my_model")

# Transition model stage
client.transition_model_version_stage(
    name="my_model",
    version=1,
    stage="Production"
)

# Load production model
model = mlflow.pyfunc.load_model("models:/my_model/Production")

Model Versioning Strategy

python

class ModelVersion:
    def __init__(self, model_name, version, metadata):
        self.model_name = model_name
        self.version = version
        self.metadata = {
            'training_date': metadata['date'],
            'dataset_version': metadata['data_version'],
            'metrics': metadata['metrics'],
            'hyperparameters': metadata['params'],
            'framework': metadata['framework'],
            'dependencies': metadata['dependencies']
        }
    
    def compare_with(self, other_version):
        return {
            'accuracy_diff': self.metadata['metrics']['accuracy'] - 
                           other_version.metadata['metrics']['accuracy'],
            'latency_diff': self.metadata['metrics']['latency'] - 
                          other_version.metadata['metrics']['latency']
        }

Deployment Strategies

Blue-Green Deployment

python

def blue_green_deployment(new_model_uri, service_name):
    # Deploy green (new) version
    deploy_model(new_model_uri, f"{service_name}-green")
    
    # Run smoke tests
    if not run_smoke_tests(f"{service_name}-green"):
        rollback(f"{service_name}-green")
        return False
    
    # Switch traffic from blue to green
    update_service_selector(service_name, version='green')
    
    # Monitor for issues
    if monitor_deployment(service_name, duration=300):
        # Success - remove blue deployment
        delete_deployment(f"{service_name}-blue")
        return True
    else:
        # Rollback to blue
        update_service_selector(service_name, version='blue')
        delete_deployment(f"{service_name}-green")
        return False

Canary Deployment

python

def canary_deployment(new_model_uri, service_name, canary_percentage=10):
    # Deploy canary version
    deploy_model(new_model_uri, f"{service_name}-canary", replicas=1)
    
    # Route small percentage of traffic to canary
    update_traffic_split(service_name, {
        'stable': 100 - canary_percentage,
        'canary': canary_percentage
    })
    
    # Monitor canary metrics
    canary_metrics = monitor_canary(f"{service_name}-canary", duration=600)
    stable_metrics = monitor_canary(f"{service_name}-stable", duration=600)
    
    if canary_metrics['error_rate'] <= stable_metrics['error_rate'] * 1.1:
        # Gradually increase canary traffic
        for percentage in [25, 50, 75, 100]:
            update_traffic_split(service_name, {
                'stable': 100 - percentage,
                'canary': percentage
            })
            time.sleep(300)
        
        # Replace stable with canary
        delete_deployment(f"{service_name}-stable")
        rename_deployment(f"{service_name}-canary", f"{service_name}-stable")
        return True
    else:
        # Rollback
        delete_deployment(f"{service_name}-canary")
        return False

Continuous Monitoring

Model Performance Monitoring

python

import prometheus_client as prom

# Define metrics
prediction_latency = prom.Histogram('model_prediction_latency_seconds', 
                                    'Model prediction latency')
prediction_counter = prom.Counter('model_predictions_total', 
                                 'Total predictions', ['model_version'])
error_counter = prom.Counter('model_errors_total', 
                            'Total errors', ['error_type'])

class MonitoredModel:
    def __init__(self, model, version):
        self.model = model
        self.version = version
    
    @prediction_latency.time()
    def predict(self, features):
        try:
            prediction = self.model.predict(features)
            prediction_counter.labels(model_version=self.version).inc()
            return prediction
        except Exception as e:
            error_counter.labels(error_type=type(e).__name__).inc()
            raise

Data Drift Detection

python

from scipy.stats import ks_2samp
import numpy as np

class DriftDetector:
    def __init__(self, reference_data, threshold=0.05):
        self.reference_data = reference_data
        self.threshold = threshold
    
    def detect_drift(self, current_data):
        drift_detected = {}
        
        for feature in self.reference_data.columns:
            # Kolmogorov-Smirnov test
            statistic, p_value = ks_2samp(
                self.reference_data[feature],
                current_data[feature]
            )
            
            drift_detected[feature] = {
                'drift': p_value < self.threshold,
                'p_value': p_value,
                'statistic': statistic
            }
        
        return drift_detected
    
    def calculate_psi(self, reference, current, bins=10):
        # Population Stability Index
        ref_hist, bin_edges = np.histogram(reference, bins=bins)
        cur_hist, _ = np.histogram(current, bins=bin_edges)
        
        ref_pct = ref_hist / len(reference)
        cur_pct = cur_hist / len(current)
        
        psi = np.sum((cur_pct - ref_pct) * np.log(cur_pct / (ref_pct + 1e-10)))
        return psi

Model Drift Detection

python

class ModelDriftDetector:
    def __init__(self, baseline_metrics):
        self.baseline_metrics = baseline_metrics
    
    def detect_performance_drift(self, current_metrics):
        drift_alerts = []
        
        for metric, baseline_value in self.baseline_metrics.items():
            current_value = current_metrics[metric]
            drift_percentage = abs(current_value - baseline_value) / baseline_value * 100
            
            if drift_percentage > 10:  # 10% threshold
                drift_alerts.append({
                    'metric': metric,
                    'baseline': baseline_value,
                    'current': current_value,
                    'drift_percentage': drift_percentage
                })
        
        return drift_alerts

Feature Store

python

from feast import FeatureStore, Entity, FeatureView, Field
from feast.types import Float32, Int64
from datetime import timedelta

# Define entity
user = Entity(name="user", join_keys=["user_id"])

# Define feature view
user_features = FeatureView(
    name="user_features",
    entities=[user],
    ttl=timedelta(days=1),
    schema=[
        Field(name="age", dtype=Int64),
        Field(name="avg_purchase", dtype=Float32),
        Field(name="total_purchases", dtype=Int64)
    ],
    source=parquet_source
)

# Initialize feature store
store = FeatureStore(repo_path=".")

# Get online features
features = store.get_online_features(
    features=["user_features:age", "user_features:avg_purchase"],
    entity_rows=[{"user_id": 123}, {"user_id": 456}]
).to_dict()

# Get historical features for training
training_data = store.get_historical_features(
    entity_df=entity_df,
    features=["user_features:age", "user_features:avg_purchase"]
).to_df()

ML Pipeline Orchestration

Kubeflow Pipeline

python

import kfp
from kfp import dsl

@dsl.component
def preprocess_data(input_path: str, output_path: str):
    # Data preprocessing logic
    pass

@dsl.component
def train_model(data_path: str, model_path: str, params: dict):
    # Model training logic
    pass

@dsl.component
def evaluate_model(model_path: str, test_data_path: str) -> dict:
    # Model evaluation logic
    pass

@dsl.pipeline(name='ML Training Pipeline')
def ml_pipeline(input_data: str, model_output: str):
    preprocess_task = preprocess_data(input_path=input_data, output_path='/data/processed')
    
    train_task = train_model(
        data_path=preprocess_task.outputs['output_path'],
        model_path=model_output,
        params={'learning_rate': 0.001, 'epochs': 10}
    )
    
    evaluate_task = evaluate_model(
        model_path=train_task.outputs['model_path'],
        test_data_path='/data/test'
    )
    
    with dsl.Condition(evaluate_task.outputs['accuracy'] > 0.9):
        deploy_model(model_path=train_task.outputs['model_path'])

# Compile and run
kfp.compiler.Compiler().compile(ml_pipeline, 'pipeline.yaml')

Model Governance

python

class ModelGovernance:
    def __init__(self):
        self.approval_required = True
        self.compliance_checks = []
    
    def register_compliance_check(self, check_func):
        self.compliance_checks.append(check_func)
    
    def validate_model(self, model, metadata):
        results = {
            'fairness': self.check_fairness(model, metadata),
            'explainability': self.check_explainability(model),
            'performance': self.check_performance(metadata),
            'compliance': all(check(model, metadata) for check in self.compliance_checks)
        }
        
        return all(results.values()), results
    
    def check_fairness(self, model, metadata):
        # Check for bias across protected attributes
        fairness_metrics = metadata.get('fairness_metrics', {})
        return all(metric < 0.1 for metric in fairness_metrics.values())
    
    def check_explainability(self, model):
        # Ensure model has explanation capabilities
        return hasattr(model, 'explain') or hasattr(model, 'feature_importances_')

Best Practices

Pipeline Design:

Modularize components for reusability
Implement idempotent operations
Use containerization for reproducibility
Version all artifacts (data, code, models)

Monitoring:

Track both technical and business metrics
Set up automated alerts for drift
Monitor model latency and throughput
Log predictions for debugging

Deployment:

Use gradual rollout strategies
Implement automated rollback
Maintain model lineage
Document deployment procedures

Governance:

Establish approval workflows
Implement audit logging
Ensure regulatory compliance
Document model decisions

Tools and Platforms

Experiment Tracking: MLflow, Weights & Biases, Neptune.ai Model Registry: MLflow, Vertex AI, SageMaker Pipeline Orchestration: Kubeflow, Airflow, Prefect, Metaflow Feature Stores: Feast, Tecton, Hopsworks Monitoring: Prometheus, Grafana, Evidently, WhyLabs Deployment: Kubernetes, Docker, TensorFlow Serving, TorchServe Cloud Platforms: AWS SageMaker, Google Vertex AI, Azure ML

Learning Path

Foundations: CI/CD basics, containerization, version control
ML Pipelines: Experiment tracking, model registry, orchestration
Deployment: Kubernetes, serving infrastructure, scaling
Monitoring: Drift detection, performance tracking, alerting
Governance: Compliance, auditing, model documentation

See references/ for pipeline templates, monitoring dashboards, and deployment configurations.

Maintainer

majiayu000 Core maintainer

Source details

Full Name: majiayu000/claude-skill-registry
Branch: main
Path in repo: skills/other/other/mlops-advanced
License: MIT License

Featured Tools

Join Our Newsletter

Maintain .agent state files. Use at session start, after meaningful steps, and before concluding: read/update constitution/memory/focus/issues/baseline consistently.

163 31

Explore

Didn't find tool you were looking for?

Search AI Tools

Install this agent skill to your Project

SKILL.md

MLOps Advanced

Overview

MLOps Maturity Levels

CI/CD for Machine Learning

Continuous Integration Components

Continuous Deployment

Model Registry and Versioning

MLflow Model Registry

Model Versioning Strategy

Deployment Strategies

Blue-Green Deployment

Canary Deployment

Continuous Monitoring

Model Performance Monitoring

Data Drift Detection

Model Drift Detection

Feature Store

ML Pipeline Orchestration

Kubeflow Pipeline

Model Governance

Best Practices

Tools and Platforms

Learning Path

Recommended Agent Skills

agent-ops-spec

agent-ops-state

agent-ops-spec

agent-ops-testing

agent-ops-testing

agent-ops-state