mlops-best-practices

MLOps Best Practices

Reproducibility

Essential Elements

1. Version Everything:

• Code (Git)
• Data (DVC, hash checksums)
• Models (model registry with versioning)
• Environment (requirements.txt, Docker)
• Hyperparameters (config files, MLflow)

2. Set Random Seeds:

import random
import numpy as np
import torch

def set_seed(seed=42):
    random.seed(seed)
    np.random.seed(seed)
    torch.manual_seed(seed)
    if torch.cuda.is_available():
        torch.cuda.manual_seed_all(seed)

3. Document Dependencies:

# requirements.txt
transformers==4.35.0
torch==2.1.0
pandas==2.1.3
scikit-learn==1.3.2

---

Experiment Tracking

Using MLflow

import mlflow

def train_with_tracking(model, train_data, config):
    """Train model with experiment tracking"""

    with mlflow.start_run():
        # Log hyperparameters
        mlflow.log_params(config)

        # Train model
        model.fit(train_data)

        # Evaluate
        metrics = evaluate(model, val_data)

        # Log metrics
        mlflow.log_metrics(metrics)

        # Log model
        mlflow.sklearn.log_model(model, "model")

        # Log artifacts (plots, configs)
        mlflow.log_artifact("confusion_matrix.png")

Experiment Organization

experiments/
├── exp_001_baseline/
│   ├── config.yaml
│   ├── results.json
│   └── model.pkl
├── exp_002_xlm_r/
│   ├── config.yaml
│   ├── results.json
│   └── model/
└── exp_003_ensemble/
    ├── config.yaml
    ├── results.json
    └── models/

---

Model Versioning

Model Registry Pattern

class ModelRegistry:
    """Simple model registry"""

    def register_model(self, model, version, metrics, metadata):
        """Register new model version"""
        model_info = {
            'version': version,
            'metrics': metrics,
            'metadata': metadata,
            'timestamp': datetime.now().isoformat(),
            'status': 'staging'  # staging, production, archived
        }

        # Save model
        model_path = f'models/v{version}/'
        os.makedirs(model_path, exist_ok=True)
        torch.save(model.state_dict(), f'{model_path}/model.pt')

        # Save metadata
        with open(f'{model_path}/metadata.json', 'w') as f:
            json.dump(model_info, f, indent=2)

        return model_path

    def promote_to_production(self, version):
        """Promote model version to production"""
        # Update status
        metadata = self.load_metadata(version)
        metadata['status'] = 'production'
        metadata['production_timestamp'] = datetime.now().isoformat()

        # Save updated metadata
        self.save_metadata(version, metadata)

        # Update production symlink
        os.symlink(f'models/v{version}', 'models/production', exist_ok=True)