Understand MLOps fundamentals, the machine learning lifecycle, DevOps vs MLOps differences, and key principles for production machine learning systems.
MLOps (Machine Learning Operations) is the practice of applying DevOps principles to machine learning systems. As organizations move from experimental ML to production-scale deployments, they face unique challenges that traditional software engineering practices alone cannot solve.
While a data scientist might achieve 95% accuracy on a test set, taking that model to production reveals a whole new set of challenges: How do you version models? How do you detect when model performance degrades? How do you retrain automatically? How do you ensure consistent predictions across thousands of requests per second?
In this comprehensive lesson, you'll learn the fundamentals of MLOps, understand the complete ML lifecycle, explore the differences between DevOps and MLOps, and discover the principles that enable organizations to deploy and maintain ML systems at scale reliably.
MLOps is the discipline of deploying, monitoring, and maintaining machine learning models in production. It bridges the gap between ML development and operations, ensuring models deliver consistent value over time.
The ML Production Problem:
Consider a typical ML workflow without MLOps:
11. Data scientist trains model in Jupyter notebook
22. Model achieves great metrics on test data
33. Model is exported and "thrown over the wall" to engineering
44. Engineering struggles to deploy the model
55. Model works initially but degrades over time
66. No one notices until customers complain
77. Debugging is nearly impossible (which version? what data?)
88. Cycle repeats...MLOps Solution:
11. Automated, versioned training pipelines
22. Model registry with staging and production
33. Automated deployment with rollback capability
44. Continuous monitoring and alerting
55. Automated retraining when drift detected
66. Full reproducibility and traceabilityWhy MLOps Matters:
1# Traditional ML: One-time model training
2def train_model_once():
3 """Train once, hope for the best"""
4 data = load_training_data()
5 model = train(data)
6 save_model(model, "model.pkl")
7 # ... months later, model is stale
8
9# MLOps: Continuous ML lifecycle
10class MLOpsWorkflow:
11 """Automated, monitored, maintainable ML"""
12
13 def __init__(self):
14 self.model_registry = ModelRegistry()
15 self.monitoring = MonitoringService()
16 self.pipeline = TrainingPipeline()
17
18 def deploy_model(self, model_version):
19 """Deploy with safety checks"""
20 # Validate model performance
21 if not self.validate_model(model_version):
22 raise ValueError("Model fails validation")
23
24 # Deploy with canary rollout
25 self.canary_deploy(model_version)
26
27 # Start monitoring
28 self.monitoring.track_model(model_version)
29
30 def check_model_health(self):
31 """Continuous health monitoring"""
32 metrics = self.monitoring.get_metrics()
33
34 if metrics['accuracy'] < self.threshold:
35 self.alert("Model degradation detected")
36 self.trigger_retraining()Key MLOps Benefits:
While MLOps builds on DevOps principles, machine learning systems have unique characteristics that require specialized approaches.
Traditional Software vs ML Systems:
1Traditional Software:
2- Behavior defined by code
3- Deterministic outputs
4- Bugs are in the code
5- Testing is well-understood
6- Deploy code changes
7
8Machine Learning:
9- Behavior defined by code + data + model
10- Probabilistic outputs
11- "Bugs" can be in data, model, or code
12- Testing is complex (what is "correct"?)
13- Deploy code, data, AND model changesThe ML-Specific Challenges:
1# Challenge 1: Data Dependency
2class DataDependency:
3 """ML systems depend heavily on data quality"""
4
5 def validate_data(self, new_data):
6 checks = {
7 'schema_match': self.check_schema(new_data),
8 'distribution_drift': self.check_drift(new_data),
9 'missing_values': self.check_completeness(new_data),
10 'outliers': self.check_outliers(new_data),
11 'freshness': self.check_data_age(new_data)
12 }
13 return all(checks.values())
14
15# Challenge 2: Model Decay
16class ModelDecay:
17 """Models degrade over time as data distributions shift"""
18
19 def monitor_performance(self, model, production_data):
20 current_metrics = self.evaluate(model, production_data)
21 baseline_metrics = self.get_baseline_metrics(model.version)
22
23 # Detect decay
24 if current_metrics['accuracy'] < baseline_metrics['accuracy'] * 0.95:
25 return {'status': 'DEGRADED', 'action': 'RETRAIN'}
26 return {'status': 'HEALTHY'}
27
28# Challenge 3: Experiment Tracking
29class ExperimentTracking:
30 """Track hundreds of experiments systematically"""
31
32 def log_experiment(self, config, metrics, artifacts):
33 experiment = {
34 'id': uuid.uuid4(),
35 'timestamp': datetime.now(),
36 'hyperparameters': config,
37 'metrics': metrics,
38 'model_artifact': artifacts['model'],
39 'data_version': artifacts['data_version'],
40 'code_commit': git.get_commit_hash(),
41 'environment': self.capture_environment()
42 }
43 self.registry.save(experiment)DevOps vs MLOps Comparison:
| Aspect | DevOps | MLOps |
|--------|--------|-------|
| Artifact | Code | Code + Data + Model |
| Testing | Unit, Integration, E2E | + Data validation, Model validation |
| Versioning | Code (Git) | Code + Data + Model versions |
| CI/CD | Build, Test, Deploy code | + Train, Validate, Deploy models |
| Monitoring | Latency, Errors, Traffic | + Model accuracy, Data drift |
| Rollback | Revert code | Revert code + model + data |
The Three Levels of ML Technical Debt:
1# Level 1: Data Debt
2data_issues = [
3 "Unstable data sources",
4 "Undocumented data transformations",
5 "Data pipeline bugs",
6 "Missing data validation"
7]
8
9# Level 2: Model Debt
10model_issues = [
11 "Unversioned experiments",
12 "Lost hyperparameters",
13 "Irreproducible results",
14 "No model lineage"
15]
16
17# Level 3: Operational Debt
18ops_issues = [
19 "No monitoring",
20 "Manual deployment",
21 "No rollback strategy",
22 "Missing alerts"
23]Key Insight:
MLOps addresses all three levels by treating ML systems as first-class software that requires specialized tooling and practices.
The ML lifecycle is an iterative process that goes far beyond training a model. Understanding each phase is essential for building sustainable ML systems.
Complete ML Lifecycle:
1┌─────────────────────────────────────────────────────────────────┐
2│ ML LIFECYCLE │
3├─────────────────────────────────────────────────────────────────┤
4│ │
5│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
6│ │ Data │───▶│ Feature │───▶│ Model │───▶│ Model │ │
7│ │ Collect │ │ Engine │ │ Train │ │ Evaluate│ │
8│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
9│ │ │ │
10│ │ DEVELOPMENT │ │
11│ ▼ ▼ │
12│ ┌─────────────────────────────────────────────────────────┐ │
13│ │ MODEL REGISTRY │ │
14│ │ Staging ──────▶ Validation ──────▶ Production │ │
15│ └─────────────────────────────────────────────────────────┘ │
16│ │ │
17│ DEPLOYMENT │ │
18│ ▼ │
19│ ┌─────────┐ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
20│ │ Serve │───▶│ Monitor │───▶│ Detect │───▶│Retrain │ │
21│ │ Model │ │ Metrics │ │ Drift │ │ Trigger │ │
22│ └─────────┘ └─────────┘ └─────────┘ └─────────┘ │
23│ │ │ │
24│ │ OPERATIONS │ │
25│ └──────────────────────────────────────────────┘ │
26│ │
27└─────────────────────────────────────────────────────────────────┘Phase-by-Phase Implementation:
1from dataclasses import dataclass
2from typing import Dict, Any, List
3from datetime import datetime
4
5@dataclass
6class MLLifecyclePhase:
7 name: str
8 inputs: List[str]
9 outputs: List[str]
10 tools: List[str]
11
12# Phase 1: Data Collection & Preparation
13data_phase = MLLifecyclePhase(
14 name="Data Collection & Preparation",
15 inputs=["Raw data sources", "Business requirements"],
16 outputs=["Clean dataset", "Data version", "Data documentation"],
17 tools=["DVC", "Delta Lake", "Great Expectations"]
18)
19
20# Phase 2: Feature Engineering
21feature_phase = MLLifecyclePhase(
22 name="Feature Engineering",
23 inputs=["Clean dataset", "Domain knowledge"],
24 outputs=["Feature store", "Feature pipelines", "Feature documentation"],
25 tools=["Feast", "Tecton", "Feature Store"]
26)
27
28# Phase 3: Model Training
29training_phase = MLLifecyclePhase(
30 name="Model Training",
31 inputs=["Features", "Hyperparameters", "Training config"],
32 outputs=["Trained model", "Training metrics", "Experiment logs"],
33 tools=["MLflow", "Weights & Biases", "Kubeflow"]
34)
35
36# Phase 4: Model Evaluation
37evaluation_phase = MLLifecyclePhase(
38 name="Model Evaluation",
39 inputs=["Trained model", "Test data", "Evaluation criteria"],
40 outputs=["Evaluation report", "Model artifacts", "Deployment decision"],
41 tools=["MLflow", "Evidently", "Deepchecks"]
42)
43
44# Phase 5: Model Deployment
45deployment_phase = MLLifecyclePhase(
46 name="Model Deployment",
47 inputs=["Validated model", "Deployment config", "Infrastructure"],
48 outputs=["Deployed model", "API endpoints", "Deployment logs"],
49 tools=["MLflow", "Seldon", "BentoML", "TorchServe"]
50)
51
52# Phase 6: Monitoring & Maintenance
53monitoring_phase = MLLifecyclePhase(
54 name="Monitoring & Maintenance",
55 inputs=["Production model", "Production data", "Business metrics"],
56 outputs=["Performance metrics", "Alerts", "Retraining triggers"],
57 tools=["Prometheus", "Grafana", "Evidently", "Alibi Detect"]
58)
59
60class MLPipeline:
61 """Orchestrate the complete ML lifecycle"""
62
63 def __init__(self):
64 self.phases = [
65 data_phase,
66 feature_phase,
67 training_phase,
68 evaluation_phase,
69 deployment_phase,
70 monitoring_phase
71 ]
72 self.current_phase = 0
73
74 def execute_phase(self, phase: MLLifecyclePhase, inputs: Dict[str, Any]) -> Dict[str, Any]:
75 """Execute a lifecycle phase"""
76 print(f"Executing: {phase.name}")
77 print(f" Inputs: {phase.inputs}")
78 print(f" Tools: {phase.tools}")
79
80 # Execute phase logic
81 outputs = self._run_phase(phase, inputs)
82
83 print(f" Outputs: {phase.outputs}")
84 return outputs
85
86 def run_full_cycle(self, initial_data):
87 """Run complete ML lifecycle"""
88 results = {'initial_data': initial_data}
89
90 for phase in self.phases:
91 results = self.execute_phase(phase, results)
92
93 return resultsLifecycle Automation Levels:
1Level 0: Manual
2- Jupyter notebooks
3- Manual model export
4- Ad-hoc deployment
5- No monitoring
6
7Level 1: Automated Training
8- Automated data pipelines
9- Automated training scripts
10- Manual deployment
11- Basic monitoring
12
13Level 2: Automated Deployment
14- CI/CD for models
15- Automated testing
16- Automated deployment
17- Comprehensive monitoring
18
19Level 3: Full Automation
20- Automated retraining
21- Auto-scaling
22- Self-healing systems
23- Continuous optimizationKey Principle:
The ML lifecycle is continuous, not linear. Production feedback loops back to data collection and model training.
Organizations progress through MLOps maturity levels as they scale their ML capabilities. Understanding these levels helps plan the journey and set realistic expectations.
Google's MLOps Maturity Model:
1from enum import Enum
2from typing import List, Dict
3
4class MLOpsLevel(Enum):
5 LEVEL_0 = "Manual Process"
6 LEVEL_1 = "ML Pipeline Automation"
7 LEVEL_2 = "CI/CD Pipeline Automation"
8
9class MLOpsMaturityAssessment:
10 """Assess organizational MLOps maturity"""
11
12 def __init__(self):
13 self.capabilities = {
14 MLOpsLevel.LEVEL_0: {
15 'data_management': 'Manual data handling',
16 'training': 'Jupyter notebooks, manual execution',
17 'deployment': 'Manual, ad-hoc deployment',
18 'monitoring': 'None or manual checks',
19 'process': 'Disconnected, tribal knowledge',
20 'typical_org': 'Early ML adoption, POC stage'
21 },
22 MLOpsLevel.LEVEL_1: {
23 'data_management': 'Automated data pipelines',
24 'training': 'Automated training pipelines',
25 'deployment': 'Manual deployment of pipeline outputs',
26 'monitoring': 'Basic model monitoring',
27 'process': 'Repeatable experiments',
28 'typical_org': 'Growing ML team, multiple models'
29 },
30 MLOpsLevel.LEVEL_2: {
31 'data_management': 'Version-controlled, validated data',
32 'training': 'Automated retraining triggered by signals',
33 'deployment': 'Automated CI/CD for models',
34 'monitoring': 'Full observability, auto-alerts',
35 'process': 'Continuous training and deployment',
36 'typical_org': 'ML at scale, critical business use'
37 }
38 }
39
40 def assess(self, current_practices: Dict[str, str]) -> MLOpsLevel:
41 """Assess current maturity level"""
42 scores = {level: 0 for level in MLOpsLevel}
43
44 for capability, value in current_practices.items():
45 for level, expected in self.capabilities.items():
46 if capability in expected and value == expected[capability]:
47 scores[level] += 1
48
49 return max(scores, key=scores.get)
50
51 def get_roadmap(self, current_level: MLOpsLevel) -> List[str]:
52 """Get recommendations to advance"""
53 roadmaps = {
54 MLOpsLevel.LEVEL_0: [
55 "Implement version control for all code",
56 "Create automated data pipelines",
57 "Set up experiment tracking (MLflow)",
58 "Document model training procedures",
59 "Add basic model validation tests"
60 ],
61 MLOpsLevel.LEVEL_1: [
62 "Implement model registry",
63 "Create CI/CD pipelines for models",
64 "Add automated model validation",
65 "Set up model monitoring",
66 "Implement feature store"
67 ],
68 MLOpsLevel.LEVEL_2: [
69 "Implement automated retraining triggers",
70 "Add A/B testing infrastructure",
71 "Set up auto-scaling for inference",
72 "Implement canary deployments",
73 "Create feedback loops to training"
74 ]
75 }
76 return roadmaps.get(current_level, [])
77
78# Example assessment
79assessment = MLOpsMaturityAssessment()
80
81# Assess fictional organization
82org_practices = {
83 'data_management': 'Automated data pipelines',
84 'training': 'Automated training pipelines',
85 'deployment': 'Manual deployment of pipeline outputs',
86 'monitoring': 'Basic model monitoring'
87}
88
89current_level = assessment.assess(org_practices)
90print(f"Current MLOps Level: {current_level.value}")
91
92roadmap = assessment.get_roadmap(current_level)
93print(f"\nRoadmap to next level:")
94for step in roadmap:
95 print(f" - {step}")Level-by-Level Details:
Level 0: Manual Process
1Characteristics:
2├── Data scientists work in isolation
3├── Models trained in notebooks
4├── Manual handoff to engineering
5├── No standardization
6├── Results hard to reproduce
7└── High technical debt
8
9Typical Problems:
10- "It worked on my machine"
11- Lost experiments
12- No versioning
13- Deployment takes weeksLevel 1: ML Pipeline Automation
1Characteristics:
2├── Automated training pipelines
3├── Experiment tracking
4├── Model versioning
5├── Standardized environments
6├── Reproducible training
7└── Manual deployment
8
9Improvements over Level 0:
10- Reproducible experiments
11- Faster iteration
12- Better collaboration
13- Consistent environmentsLevel 2: CI/CD Pipeline Automation
1Characteristics:
2├── Automated CI/CD for models
3├── Continuous training
4├── Automated deployment
5├── Full monitoring
6├── Feedback loops
7└── Self-healing systems
8
9Improvements over Level 1:
10- Deploy in minutes, not days
11- Automated quality gates
12- Real-time monitoring
13- Automatic rollbackInvestment vs Return:
1Level 0 → Level 1:
2 Investment: Medium (3-6 months)
3 Return: High (reproducibility, faster experiments)
4
5Level 1 → Level 2:
6 Investment: High (6-12 months)
7 Return: Very High (scale, reliability, speed)Key Insight:
Most organizations should target Level 1 first. Level 2 is for organizations with critical ML workloads requiring high reliability and scale.
Successful MLOps implementations follow core principles and leverage specialized tools. Understanding these fundamentals provides a foundation for the rest of this course.
Core MLOps Principles:
1class MLOpsPrinciples:
2 """Foundational principles for production ML"""
3
4 principles = {
5 'reproducibility': {
6 'description': 'Any result can be recreated exactly',
7 'implementation': [
8 'Version control for code, data, and models',
9 'Deterministic training (set seeds)',
10 'Environment specification (Docker, conda)',
11 'Capture all hyperparameters'
12 ]
13 },
14 'automation': {
15 'description': 'Reduce manual steps and human error',
16 'implementation': [
17 'Automated data pipelines',
18 'Automated training and validation',
19 'CI/CD for model deployment',
20 'Automated monitoring and alerting'
21 ]
22 },
23 'continuous_improvement': {
24 'description': 'Models improve over time with new data',
25 'implementation': [
26 'Monitoring for drift and degradation',
27 'Automated retraining pipelines',
28 'A/B testing for model updates',
29 'Feedback loops from production'
30 ]
31 },
32 'collaboration': {
33 'description': 'Teams work together effectively',
34 'implementation': [
35 'Shared experiment tracking',
36 'Model registry with reviews',
37 'Documentation and lineage',
38 'Cross-functional workflows'
39 ]
40 },
41 'reliability': {
42 'description': 'Models perform consistently in production',
43 'implementation': [
44 'Comprehensive testing',
45 'Gradual rollouts',
46 'Automated rollback',
47 'SLA monitoring'
48 ]
49 }
50 }
51
52 @classmethod
53 def assess_implementation(cls, org_practices: dict) -> dict:
54 """Assess how well an organization follows MLOps principles"""
55 scores = {}
56 for principle, details in cls.principles.items():
57 implemented = sum(
58 1 for item in details['implementation']
59 if item in org_practices.get(principle, [])
60 )
61 scores[principle] = implemented / len(details['implementation'])
62 return scoresMLOps Tool Categories:
1mlops_tools = {
2 'experiment_tracking': {
3 'purpose': 'Track experiments, parameters, metrics',
4 'tools': ['MLflow', 'Weights & Biases', 'Neptune', 'Comet'],
5 'key_features': ['Metric logging', 'Artifact storage', 'Comparison views']
6 },
7 'model_registry': {
8 'purpose': 'Version and manage model lifecycle',
9 'tools': ['MLflow Registry', 'Sagemaker Registry', 'Vertex AI'],
10 'key_features': ['Model versioning', 'Stage transitions', 'Approval workflows']
11 },
12 'feature_store': {
13 'purpose': 'Manage and serve features consistently',
14 'tools': ['Feast', 'Tecton', 'Hopsworks'],
15 'key_features': ['Online/offline serving', 'Feature versioning', 'Time-travel']
16 },
17 'data_versioning': {
18 'purpose': 'Track data changes and enable reproducibility',
19 'tools': ['DVC', 'Delta Lake', 'LakeFS'],
20 'key_features': ['Data versioning', 'Pipeline tracking', 'Collaboration']
21 },
22 'model_serving': {
23 'purpose': 'Deploy models for inference',
24 'tools': ['TorchServe', 'TF Serving', 'Seldon', 'BentoML'],
25 'key_features': ['REST/gRPC APIs', 'Batching', 'Scaling']
26 },
27 'monitoring': {
28 'purpose': 'Track model performance in production',
29 'tools': ['Evidently', 'WhyLabs', 'Arize', 'Fiddler'],
30 'key_features': ['Drift detection', 'Performance tracking', 'Alerting']
31 },
32 'orchestration': {
33 'purpose': 'Coordinate ML workflows',
34 'tools': ['Airflow', 'Prefect', 'Kubeflow', 'Argo'],
35 'key_features': ['DAG execution', 'Scheduling', 'Retry logic']
36 }
37}
38
39# Print tool landscape
40for category, info in mlops_tools.items():
41 print(f"\n{category.upper().replace('_', ' ')}")
42 print(f"Purpose: {info['purpose']}")
43 print(f"Tools: {', '.join(info['tools'])}")Typical MLOps Architecture:
1┌────────────────────────────────────────────────────────────────────┐
2│ DATA LAYER │
3│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
4│ │ Data │ │ Feature │ │ Data │ │ Data │ │
5│ │ Lake │ │ Store │ │ Version │ │ Valid │ │
6│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
7├────────────────────────────────────────────────────────────────────┤
8│ DEVELOPMENT LAYER │
9│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
10│ │Experiment│ │ Model │ │ Training │ │ Model │ │
11│ │ Tracking │ │ Registry │ │ Pipeline │ │ Tests │ │
12│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
13├────────────────────────────────────────────────────────────────────┤
14│ DEPLOYMENT LAYER │
15│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
16│ │ CI/CD │ │Container │ │ Serving │ │ Load │ │
17│ │ Pipeline │ │ Registry │ │ Platform │ │ Balancer │ │
18│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
19├────────────────────────────────────────────────────────────────────┤
20│ OPERATIONS LAYER │
21│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌──────────┐ │
22│ │ Monitor │ │ Logging │ │ Alerting │ │ Feedback │ │
23│ │ System │ │ Service │ │ Service │ │ Loop │ │
24│ └──────────┘ └──────────┘ └──────────┘ └──────────┘ │
25└────────────────────────────────────────────────────────────────────┘Getting Started Checklist:
1mlops_checklist = {
2 'essential': [
3 '✅ Version control for all code (Git)',
4 '✅ Experiment tracking (MLflow)',
5 '✅ Model versioning',
6 '✅ Automated testing',
7 '✅ Basic monitoring'
8 ],
9 'important': [
10 '⬜ CI/CD pipeline',
11 '⬜ Model registry',
12 '⬜ Data versioning',
13 '⬜ Feature store',
14 '⬜ Drift detection'
15 ],
16 'advanced': [
17 '⬜ Automated retraining',
18 '⬜ A/B testing infrastructure',
19 '⬜ Canary deployments',
20 '⬜ Auto-scaling',
21 '⬜ Cost optimization'
22 ]
23}
24
25print("MLOps Implementation Checklist:")
26for level, items in mlops_checklist.items():
27 print(f"\n{level.upper()}:")
28 for item in items:
29 print(f" {item}")Key Insight:
Start with the essentials, then add capabilities as your ML systems mature. Don't try to implement everything at once.
Create a comprehensive framework to assess MLOps maturity and generate actionable recommendations for production ML systems.
1import json
2from datetime import datetime
3from typing import Dict, List, Optional
4from dataclasses import dataclass, field
5from enum import Enum
6
7print("=" * 70)
8print("MLOPS ASSESSMENT FRAMEWORK")
9print("=" * 70)
10
11# ============================================================================
12# DATA MODELS
13# ============================================================================
14
15class MaturityLevel(Enum):
16 """MLOps maturity levels"""
17 LEVEL_0 = "Manual Process"
18 LEVEL_1 = "ML Pipeline Automation"
19 LEVEL_2 = "CI/CD Pipeline Automation"
20 LEVEL_3 = "Full MLOps Automation"
21
22@dataclass
23class CapabilityAssessment:
24 """Assessment for a single capability"""
25 name: str
26 current_state: str
27 target_state: str
28 maturity_level: int # 0-3
29 priority: str # High, Medium, Low
30 effort: str # Days estimate
31 recommendations: List[str] = field(default_factory=list)
32
33@dataclass
34class MLOpsAssessment:
35 """Complete MLOps assessment"""
36 organization: str
37 assessment_date: str
38 assessor: str
39 overall_level: MaturityLevel
40 capabilities: List[CapabilityAssessment]
41 total_score: float
42 recommendations: List[str]
43
44# ============================================================================
45# ASSESSMENT FRAMEWORK
46# ============================================================================
47
48class MLOpsAssessor:
49 """Framework to assess and improve MLOps maturity"""
50
51 def __init__(self):
52 self.capability_definitions = self._define_capabilities()
53 self.maturity_criteria = self._define_maturity_criteria()
54
55 def _define_capabilities(self) -> Dict[str, Dict]:
56 """Define all MLOps capabilities to assess"""
57 return {
58 'version_control': {
59 'description': 'Code, data, and model versioning',
60 'levels': {
61 0: 'No version control',
62 1: 'Code in Git, models saved locally',
63 2: 'Code + models versioned, data tracked',
64 3: 'Full lineage tracking for code, data, models'
65 },
66 'category': 'Reproducibility'
67 },
68 'experiment_tracking': {
69 'description': 'Systematic experiment logging',
70 'levels': {
71 0: 'Notebooks with embedded results',
72 1: 'Manual logging to spreadsheets',
73 2: 'MLflow/W&B with parameters and metrics',
74 3: 'Full artifact tracking with comparisons'
75 },
76 'category': 'Reproducibility'
77 },
78 'data_management': {
79 'description': 'Data pipelines and quality',
80 'levels': {
81 0: 'Manual data collection',
82 1: 'Basic scripts for data loading',
83 2: 'Automated pipelines with validation',
84 3: 'Feature store with monitoring'
85 },
86 'category': 'Data'
87 },
88 'model_training': {
89 'description': 'Training process automation',
90 'levels': {
91 0: 'Jupyter notebooks',
92 1: 'Python scripts, manual execution',
93 2: 'Automated training pipelines',
94 3: 'Automated retraining on triggers'
95 },
96 'category': 'Training'
97 },
98 'model_validation': {
99 'description': 'Testing and validation',
100 'levels': {
101 0: 'Manual accuracy checks',
102 1: 'Basic test scripts',
103 2: 'Automated validation suite',
104 3: 'Shadow mode testing, A/B tests'
105 },
106 'category': 'Quality'
107 },
108 'deployment': {
109 'description': 'Model deployment process',
110 'levels': {
111 0: 'Manual deployment, ad-hoc',
112 1: 'Scripted deployment',
113 2: 'CI/CD with staging environments',
114 3: 'Canary/blue-green with auto-rollback'
115 },
116 'category': 'Deployment'
117 },
118 'serving': {
119 'description': 'Model serving infrastructure',
120 'levels': {
121 0: 'Flask app on single server',
122 1: 'Containerized API',
123 2: 'Kubernetes with auto-scaling',
124 3: 'Multi-region, GPU-optimized serving'
125 },
126 'category': 'Deployment'
127 },
128 'monitoring': {
129 'description': 'Production monitoring',
130 'levels': {
131 0: 'No monitoring',
132 1: 'Basic health checks',
133 2: 'Metrics dashboard with alerts',
134 3: 'Drift detection, auto-retraining triggers'
135 },
136 'category': 'Operations'
137 },
138 'collaboration': {
139 'description': 'Team workflows',
140 'levels': {
141 0: 'Individual work, email handoffs',
142 1: 'Shared notebooks, Slack updates',
143 2: 'Model registry with reviews',
144 3: 'Full ML platform with governance'
145 },
146 'category': 'Organization'
147 }
148 }
149
150 def _define_maturity_criteria(self) -> Dict[MaturityLevel, float]:
151 """Define score thresholds for maturity levels"""
152 return {
153 MaturityLevel.LEVEL_0: 0.0,
154 MaturityLevel.LEVEL_1: 0.4,
155 MaturityLevel.LEVEL_2: 0.6,
156 MaturityLevel.LEVEL_3: 0.8
157 }
158
159 def assess_capability(
160 self,
161 capability: str,
162 current_level: int,
163 target_level: int = 3
164 ) -> CapabilityAssessment:
165 """Assess a single capability"""
166 cap_def = self.capability_definitions[capability]
167
168 # Generate recommendations
169 recommendations = []
170 if current_level < target_level:
171 for level in range(current_level + 1, min(target_level + 1, 4)):
172 recommendations.append(
173 f"Upgrade to: {cap_def['levels'][level]}"
174 )
175
176 # Estimate effort
177 effort_map = {0: '2-4 weeks', 1: '1-2 weeks', 2: '3-5 days', 3: '1-2 days'}
178 effort = effort_map[3 - (target_level - current_level)]
179
180 # Set priority based on category
181 priority_map = {
182 'Reproducibility': 'High',
183 'Data': 'High',
184 'Training': 'Medium',
185 'Quality': 'High',
186 'Deployment': 'Medium',
187 'Operations': 'Medium',
188 'Organization': 'Low'
189 }
190
191 return CapabilityAssessment(
192 name=capability,
193 current_state=cap_def['levels'][current_level],
194 target_state=cap_def['levels'][target_level],
195 maturity_level=current_level,
196 priority=priority_map[cap_def['category']],
197 effort=effort,
198 recommendations=recommendations
199 )
200
201 def full_assessment(
202 self,
203 organization: str,
204 capability_scores: Dict[str, int],
205 assessor: str = "MLOps Assessor"
206 ) -> MLOpsAssessment:
207 """Perform full MLOps assessment"""
208
209 # Assess each capability
210 assessments = []
211 for capability, level in capability_scores.items():
212 if capability in self.capability_definitions:
213 assessments.append(
214 self.assess_capability(capability, level)
215 )
216
217 # Calculate overall score (0-1)
218 total_score = sum(a.maturity_level for a in assessments) / (len(assessments) * 3)
219
220 # Determine overall level
221 overall_level = MaturityLevel.LEVEL_0
222 for level, threshold in self.maturity_criteria.items():
223 if total_score >= threshold:
224 overall_level = level
225
226 # Generate top recommendations
227 recommendations = []
228 high_priority = [a for a in assessments if a.priority == 'High' and a.maturity_level < 2]
229 for assessment in sorted(high_priority, key=lambda x: x.maturity_level)[:3]:
230 if assessment.recommendations:
231 recommendations.append(
232 f"{assessment.name}: {assessment.recommendations[0]}"
233 )
234
235 return MLOpsAssessment(
236 organization=organization,
237 assessment_date=datetime.now().strftime("%Y-%m-%d"),
238 assessor=assessor,
239 overall_level=overall_level,
240 capabilities=assessments,
241 total_score=total_score,
242 recommendations=recommendations
243 )
244
245 def generate_report(self, assessment: MLOpsAssessment) -> str:
246 """Generate formatted assessment report"""
247 report = []
248 report.append("=" * 70)
249 report.append(f"MLOPS MATURITY ASSESSMENT REPORT")
250 report.append("=" * 70)
251 report.append(f"Organization: {assessment.organization}")
252 report.append(f"Date: {assessment.assessment_date}")
253 report.append(f"Assessor: {assessment.assessor}")
254 report.append("")
255 report.append("-" * 70)
256 report.append("EXECUTIVE SUMMARY")
257 report.append("-" * 70)
258 report.append(f"Overall Maturity Level: {assessment.overall_level.value}")
259 report.append(f"Maturity Score: {assessment.total_score:.1%}")
260 report.append("")
261
262 # Score visualization
263 filled = int(assessment.total_score * 20)
264 bar = "█" * filled + "░" * (20 - filled)
265 report.append(f"Progress: [{bar}] {assessment.total_score:.1%}")
266 report.append("")
267
268 report.append("-" * 70)
269 report.append("CAPABILITY SCORES")
270 report.append("-" * 70)
271
272 by_category = {}
273 for cap in assessment.capabilities:
274 cat = self.capability_definitions[cap.name]['category']
275 if cat not in by_category:
276 by_category[cat] = []
277 by_category[cat].append(cap)
278
279 for category, caps in by_category.items():
280 report.append(f"\n{category}:")
281 for cap in caps:
282 level_bar = "●" * cap.maturity_level + "○" * (3 - cap.maturity_level)
283 report.append(f" {cap.name:25} [{level_bar}] Level {cap.maturity_level}")
284
285 report.append("")
286 report.append("-" * 70)
287 report.append("TOP RECOMMENDATIONS")
288 report.append("-" * 70)
289 for i, rec in enumerate(assessment.recommendations, 1):
290 report.append(f"{i}. {rec}")
291
292 report.append("")
293 report.append("-" * 70)
294 report.append("DETAILED ROADMAP")
295 report.append("-" * 70)
296
297 gaps = [a for a in assessment.capabilities if a.maturity_level < 2]
298 for gap in sorted(gaps, key=lambda x: (x.priority == 'High', -x.maturity_level), reverse=True)[:5]:
299 report.append(f"\n{gap.name} (Priority: {gap.priority}, Effort: {gap.effort})")
300 report.append(f" Current: {gap.current_state}")
301 report.append(f" Target: {gap.target_state}")
302 for rec in gap.recommendations[:2]:
303 report.append(f" → {rec}")
304
305 report.append("")
306 report.append("=" * 70)
307
308 return "\n".join(report)
309
310# ============================================================================
311# RUN ASSESSMENT
312# ============================================================================
313
314print("\n1. Creating MLOps Assessor...")
315print("-" * 70)
316
317assessor = MLOpsAssessor()
318
319print("Capability definitions loaded:")
320for cap, details in assessor.capability_definitions.items():
321 print(f" - {cap}: {details['description']}")
322
323print("\n2. Running Sample Assessment...")
324print("-" * 70)
325
326# Sample organization scores (0-3 for each capability)
327sample_scores = {
328 'version_control': 2, # Code + models versioned
329 'experiment_tracking': 1, # Manual logging
330 'data_management': 1, # Basic scripts
331 'model_training': 2, # Automated pipelines
332 'model_validation': 1, # Basic tests
333 'deployment': 1, # Scripted deployment
334 'serving': 1, # Containerized
335 'monitoring': 0, # No monitoring
336 'collaboration': 1 # Shared notebooks
337}
338
339assessment = assessor.full_assessment(
340 organization="Acme ML Team",
341 capability_scores=sample_scores,
342 assessor="Senior MLOps Engineer"
343)
344
345print("\n3. Assessment Results:")
346print("-" * 70)
347
348report = assessor.generate_report(assessment)
349print(report)
350
351# ============================================================================
352# IMPROVEMENT TRACKING
353# ============================================================================
354
355print("\n4. Setting Improvement Goals...")
356print("-" * 70)
357
358class ImprovementTracker:
359 """Track MLOps improvement over time"""
360
361 def __init__(self, baseline: MLOpsAssessment):
362 self.baseline = baseline
363 self.checkpoints = []
364
365 def add_checkpoint(self, assessment: MLOpsAssessment, notes: str):
366 """Add progress checkpoint"""
367 improvement = assessment.total_score - self.baseline.total_score
368 self.checkpoints.append({
369 'date': assessment.assessment_date,
370 'score': assessment.total_score,
371 'improvement': improvement,
372 'level': assessment.overall_level.value,
373 'notes': notes
374 })
375
376 def get_progress_summary(self) -> str:
377 """Generate progress summary"""
378 if not self.checkpoints:
379 return "No checkpoints recorded yet"
380
381 latest = self.checkpoints[-1]
382 return f"""
383Progress Summary:
384 Baseline Score: {self.baseline.total_score:.1%}
385 Current Score: {latest['score']:.1%}
386 Improvement: {latest['improvement']:+.1%}
387 Current Level: {latest['level']}
388 Checkpoints: {len(self.checkpoints)}
389"""
390
391tracker = ImprovementTracker(assessment)
392
393# Simulate improvement after 3 months
394improved_scores = sample_scores.copy()
395improved_scores['monitoring'] = 2 # Added monitoring
396improved_scores['experiment_tracking'] = 2 # Added MLflow
397improved_scores['model_validation'] = 2 # Automated tests
398
399improved_assessment = assessor.full_assessment(
400 organization="Acme ML Team",
401 capability_scores=improved_scores,
402 assessor="Senior MLOps Engineer"
403)
404
405tracker.add_checkpoint(
406 improved_assessment,
407 "Added MLflow, monitoring, and automated validation"
408)
409
410print(tracker.get_progress_summary())
411
412print("\n" + "=" * 70)
413print("MLOPS ASSESSMENT FRAMEWORK COMPLETE")
414print("=" * 70)
415print("\nThis framework helps organizations:")
416print(" 1. Assess current MLOps maturity")
417print(" 2. Identify gaps and priorities")
418print(" 3. Create actionable roadmaps")
419print(" 4. Track improvement over time")
420print("\nNext lesson: Model Versioning with MLflow")This comprehensive MLOps assessment framework demonstrates production-ready patterns for evaluating and improving ML operations:
Framework Architecture:
1. Data Models (dataclasses):
MaturityLevel: Enum for 4 MLOps maturity levelsCapabilityAssessment: Detailed assessment for each capabilityMLOpsAssessment: Complete organizational assessment2. MLOpsAssessor Class:
3. Key Capabilities Assessed:
Assessment Process:
Output Includes:
Improvement Tracking:
Production Best Practices:
Excellent work understanding MLOps fundamentals! You now know what MLOps is, why it matters, and how organizations progress in maturity. In our next lesson, "Model Versioning with MLflow," you'll learn to use MLflow for tracking experiments, versioning models, and managing the model registry. This is the foundation for reproducibility in production ML systems.
Your progress is saved in this browser, on this device only.