Loading
Loading
In this project you'll build a complete integrated MLOps pipeline covering all stages from data to production.
Data (DVC) ā Training (MLflow) ā Quality Gate ā Staging ā Production ā Monitoring (Evidently)
After completing this pipeline, you can safely deploy a new model every day instead of once a month.
import random
import json
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from datetime import datetime
# āāā Data āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
@dataclass
class Dataset:
name: str
version: str
rows: int
features: int
def validate(self) -> Dict:
issues = []
if self.rows < 5000:
issues.append(f"Ų¹ŲÆŲÆ Ų§ŁŲµŁŁŁ ŁŁŁŁ Ų¬ŲÆŲ§Ł ({self.rows})")
null_rate = random.uniform(0, 0.05)
if null_rate > 0.02:
rate_str = f"{null_rate:.1%}"
issues.append(f"ŁŲ³ŲØŲ© Ų§ŁŁŁŁ
Ų§ŁŁŲ§Ų±ŲŗŲ© Ų¹Ų§ŁŁŲ© ({rate_str})")
return {"valid": len(issues) == 0, "issues": issues,
"null_rate": null_rate, "rows": self.rows}
# āāā Model āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
@dataclass
class ModelVersion:
name: str
version: str
accuracy: float
f1: float
latency_ms: float
stage: str = "Staging"
def meets_gate(self, min_acc=0.90, max_lat=100.0) -> bool:
return self.accuracy >= min_acc and self.latency_ms <= max_lat
# āāā Pipeline Components āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
class DataPipeline:
def ingest(self, name: str, rows: int, features: int) -> Optional[Dataset]:
ds = Dataset(name, f"v{datetime.now().strftime('%Y%m%d')}", rows, features)
result = ds.validate()
status = "ā
" if result["valid"] else "ā"
print(f" {status} Data: {name} ({rows:,} rows)")
for issue in result["issues"]:
print(f" ā ļø {issue}")
return ds if result["valid"] else None
class TrainingPipeline:
def __init__(self):
self.runs: List[Dict] = []
self._counter = 0
def train(self, dataset: Dataset, cfg: Dict) -> ModelVersion:
random.seed(self._counter * 7 + 42)
self._counter += 1
acc = min(0.98, 0.83 + cfg.get("epochs", 10) * 0.006 + random.uniform(-0.03, 0.03))
f1 = acc - random.uniform(0.01, 0.04)
lat = max(15, 130 - cfg.get("epochs", 10) * 3 + random.uniform(-5, 5))
ver = f"1.{self._counter}.0"
run = {"run_id": f"run_{self._counter:03d}", "params": cfg,
"metrics": {"accuracy": round(acc, 4), "f1": round(f1, 4)}}
self.runs.append(run)
acc_str = f"{acc:.1%}"
lat_str = f"{lat:.0f}ms"
model_name = cfg.get('model', 'model')
print(f" ā
{run['run_id']} ({model_name}): acc={acc_str}, lat={lat_str}")
return ModelVersion(dataset.name, ver, round(acc, 4), round(f1, 4), round(lat, 1))
class QualityGate:
def evaluate(self, mv: ModelVersion, current: Optional[ModelVersion] = None) -> bool:
passed = mv.meets_gate()
icon = "ā
" if passed else "ā"
acc_s = f"{mv.accuracy:.1%}"
lat_s = f"{mv.latency_ms:.0f}ms"
print(f" {icon} Gate v{mv.version}: acc={acc_s}, lat={lat_s}")
if current:
improved = mv.accuracy > current.accuracy
diff_s = f"{(mv.accuracy - current.accuracy):+.2%}"
comp_icon = "š" if improved else "š"
print(f" {comp_icon} Ł
ŁŲ§Ų±ŁŲ© ŲØŲ§ŁŲŲ§ŁŁ: {diff_s}")
return passed
class DeploymentManager:
def __init__(self):
self.envs: Dict[str, ModelVersion] = {}
def deploy(self, mv: ModelVersion, env: str):
mv.stage = "Production" if env == "production" else "Staging"
self.envs[env] = mv
acc_s = f"{mv.accuracy:.1%}"
print(f" ā
Deployed v{mv.version} ā {env} (acc={acc_s})")
def status(self):
print(f"\n š Ų§ŁŁŁ
Ų§Ų°Ų¬ Ų§ŁŁ
ŁŲ“ŁŲ±Ų©:")
for env, mv in self.envs.items():
acc_s = f"{mv.accuracy:.1%}"
lat_s = f"{mv.latency_ms:.0f}ms"
print(f" [{env:<12}] v{mv.version} ā acc={acc_s}, lat={lat_s}")
# āāā ŲŖŲ“ŲŗŁŁ Ų®Ų· MLOps Ų§ŁŁŲ§Ł
Ł āāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāāā
print("š Ų®Ų· MLOps Ų§ŁŁŲ§Ł
Ł")
print("=" * 58)
data_pipe = DataPipeline()
train_pipe = TrainingPipeline()
gate = QualityGate()
deployer = DeploymentManager()
current_mv: Optional[ModelVersion] = None
# 1. Ų§ŁŲØŁŲ§ŁŲ§ŲŖ
print(f"\n{'ā'*58}")
print("1ļøā£ DataPipeline ā Ų§Ų³ŲŖŁŲ¹Ų§ŲØ ŁŲŖŲŁŁ:")
ds = data_pipe.ingest("customer-reviews-v4", rows=80000, features=15)
if not ds:
print(" š Ų®Ų· Ų§ŁŲ£ŁŲ§ŲØŁŲØ Ł
ŲŖŁŁŁ ā ŲØŁŲ§ŁŲ§ŲŖ ŲŗŁŲ± ŲµŲ§ŁŲŲ©")
exit()
# 2. Ų§ŁŲŖŲÆŲ±ŁŲØ
print(f"\n{'ā'*58}")
print("2ļøā£ TrainingPipeline ā 3 ŲŖŲ¬Ų§Ų±ŲØ:")
configs = [
{"learning_rate": 0.001, "epochs": 10, "model": "DistilBERT"},
{"learning_rate": 0.0005, "epochs": 20, "model": "BERT-base"},
{"learning_rate": 0.0001, "epochs": 25, "model": "RoBERTa"},
]
versions = [train_pipe.train(ds, cfg) for cfg in configs]
best_mv = max(versions, key=lambda mv: mv.accuracy)
# 3. ŲØŁŲ§ŲØŲ© Ų§ŁŲ¬ŁŲÆŲ©
print(f"\n{'ā'*58}")
print("3ļøā£ QualityGate (min_acc=90%, max_lat=100ms):")
if not gate.evaluate(best_mv, current_mv):
print(" š Ų§ŁŁŲ“Ų± Ł
ŁŁŁŁ ā Ų§ŁŁŁ
ŁŲ°Ų¬ ŁŲ§ ŁŲ³ŲŖŁŁŁ Ų§ŁŁ
Ų¹Ų§ŁŁŲ±")
else:
# 4. Ų§ŁŁŲ“Ų±
print(f"\n{'ā'*58}")
print("4ļøā£ Deployment ā Blue-Green:")
deployer.deploy(best_mv, "staging")
print(" ā³ Smoke tests Ų¹ŁŁ Staging...")
deployer.deploy(best_mv, "production")
current_mv = best_mv
# 5. Ų§ŁŲŖŁŲ±ŁŲ± Ų§ŁŁŁŲ§Ų¦Ł
print(f"\n{'ā'*58}")
print("5ļøā£ ŲŖŁŲ±ŁŲ± Ų§ŁŲŲ§ŁŲ© Ų§ŁŁŁŲ§Ų¦ŁŲ©:")
deployer.status()
print(f"\n š ŲŖŲ¬Ų§Ų±ŲØ Ų§ŁŲŖŲÆŲ±ŁŲØ: {len(train_pipe.runs)}")
all_acc = [r['metrics']['accuracy'] for r in train_pipe.runs]
best_acc_str = f"{max(all_acc):.4f}"
avg_acc_str = f"{sum(all_acc)/len(all_acc):.4f}"
print(f" š Ų£ŁŲ¶Ł ŲÆŁŲ© : {best_acc_str}")
print(f" š Ł
ŲŖŁŲ³Ų· ŲÆŁŲ© : {avg_acc_str}")
print(f"\nā
Ų®Ų· MLOps Ų§ŁŁŲ§Ł
Ł Ų§ŁŲŖŁ
Ł!")
print(f"š Ł
ŲØŲ±ŁŁ! Ų£ŲŖŁ
Ł
ŲŖ ŲÆŁŲ±Ų© MLOps")