إنتقل إلى المحتوى الرئيسي

الفصل 3: إصدار النماذج

المقدمة​

إصدار النماذج أمر حيوي لأنظمة تعلم الآلة في الإنتاج. فهو يمكّن من تتبع النماذج ومقارنتها والتراجع عنها مع الحفاظ على سلالة واضحة من البيانات إلى النشر.

لماذا إصدار النماذج؟​

تحدي إدارة النماذج​

فوائد إصدار النماذج​

الفائدةالوصف
إمكانية التكرارإعادة إنشاء أي إصدار من النموذج
السلالةتتبع البيانات → النموذج → التنبؤات
المقارنةمقارنة الأداء عبر الإصدارات
التراجعالتراجع بسرعة إلى الإصدارات السابقة
الامتثالسجل تدقيق للصناعات المنظمة

مفاهيم سجل النماذج​

المكونات الرئيسية​

مراحل دورة حياة النموذج​

المرحلةالوصف
لا شيءالحالة الأولية، مسجل حديثًا
مرحلة الاختباريتم اختبار/التحقق من النموذج
الإنتاجالنموذج يخدم حركة المرور الحية
مؤرشفالنموذج مهمل، يحتفظ به للرجوع إليه

سجل نماذج MLflow​

إعداد MLflow​

# تثبيت MLflow
pip install mlflow

# بدء خادم MLflow مع مخزن الخلفية
mlflow server \
    --backend-store-uri postgresql://user:pass@localhost/mlflow \
    --default-artifact-root s3://mlflow-artifacts/ \
    --host 0.0.0.0 \
    --port 5000

تسجيل النماذج​

# register_model.py
import mlflow
import mlflow.sklearn
from sklearn.ensemble import RandomForestClassifier

# تعيين URI التتبع
mlflow.set_tracking_uri("http://localhost:5000")

# تدريب وتسجيل النموذج
with mlflow.start_run() as run:
    # تدريب النموذج
    model = RandomForestClassifier(n_estimators=100)
    model.fit(X_train, y_train)
    
    # تسجيل المعاملات
    mlflow.log_param("n_estimators", 100)
    mlflow.log_param("max_depth", None)
    
    # تسجيل المقاييس
    accuracy = model.score(X_test, y_test)
    mlflow.log_metric("accuracy", accuracy)
    
    # تسجيل النموذج
    mlflow.sklearn.log_model(
        model,
        "model",
        registered_model_name="customer-churn-model"
    )
    
    print(f"Model registered with run_id: {run.info.run_id}")

إدارة إصدارات النماذج​

# model_management.py
from mlflow.tracking import MlflowClient

client = MlflowClient()

# الحصول على جميع إصدارات النموذج
versions = client.search_model_versions("name='customer-churn-model'")
for v in versions:
    print(f"Version {v.version}: {v.current_stage}")

# نقل النموذج إلى مرحلة الاختبار
client.transition_model_version_stage(
    name="customer-churn-model",
    version=2,
    stage="Staging"
)

# نقل النموذج إلى الإنتاج
client.transition_model_version_stage(
    name="customer-churn-model",
    version=2,
    stage="Production",
    archive_existing_versions=True  # أرشفة الإصدار الحالي في الإنتاج
)

# إضافة وصف لإصدار النموذج
client.update_model_version(
    name="customer-churn-model",
    version=2,
    description="تحسين هندسة الميزات، زيادة AUC بنسبة 5%"
)

# إضافة علامات
client.set_model_version_tag(
    name="customer-churn-model",
    version=2,
    key="validation_status",
    value="approved"
)

تحميل النماذج من السجل​

# load_model.py
import mlflow

# تحميل إصدار محدد
model_v1 = mlflow.sklearn.load_model(
    "models:/customer-churn-model/1"
)

# تحميل حسب المرحلة
production_model = mlflow.sklearn.load_model(
    "models:/customer-churn-model/Production"
)

# تحميل نموذج مرحلة الاختبار
staging_model = mlflow.sklearn.load_model(
    "models:/customer-churn-model/Staging"
)

# إجراء التنبؤات
predictions = production_model.predict(X_new)

البيانات الوصفية للنموذج والسلالة​

بطاقة نموذج شاملة​

# model_card.py
import json
from dataclasses import dataclass, asdict
from typing import List, Dict, Optional
from datetime import datetime

@dataclass
class ModelCard:
    """توثيق النموذج المنظم وفقًا للمعايير الصناعية."""
    
    # معلومات أساسية
    model_name: str
    version: str
    description: str
    owner: str
    created_at: str
    
    # معلومات التدريب
    training_data: Dict[str, str]
    features: List[str]
    target: str
    algorithm: str
    hyperparameters: Dict[str, any]
    
    # الأداء
    metrics: Dict[str, float]
    evaluation_data: str
    
    # الاستخدام المقصود
    intended_use: str
    out_of_scope_use: str
    
    # القيود والتحيزات
    limitations: List[str]
    ethical_considerations: List[str]
    
    # السلالة
    parent_model: Optional[str] = None
    training_run_id: Optional[str] = None
    data_version: Optional[str] = None
    code_version: Optional[str] = None

# إنشاء بطاقة النموذج
model_card = ModelCard(
    model_name="customer-churn-model",
    version="2.1.0",
    description="يتنبأ باحتمالية مغادرة العميل خلال 30 يومًا",
    owner="ml-team@company.com",
    created_at=datetime.now().isoformat(),
    
    training_data={
        "source": "s3://data-lake/customer-features/",
        "date_range": "2023-01-01 to 2024-01-01",
        "records": "1,500,000"
    },
    features=[
        "account_age_days",
        "total_purchases",
        "days_since_last_login",
        "support_tickets_90d",
        "nps_score"
    ],
    target="churned_within_30_days",
    algorithm="GradientBoostingClassifier",
    hyperparameters={
        "n_estimators": 200,
        "max_depth": 7,
        "learning_rate": 0.1,
        "min_samples_split": 10
    },
    
    metrics={
        "auc_roc": 0.89,
        "precision": 0.82,
        "recall": 0.75,
        "f1_score": 0.78
    },
    evaluation_data="s3://data-lake/test-set-2024-01/",
    
    intended_use="تحديد العملاء المعرضين للخطر لحملات الاحتفاظ",
    out_of_scope_use="تقييم الائتمان، قرارات التوظيف",
    
    limitations=[
        "قد يتدهور الأداء للعملاء الذين تقل أعمارهم عن 30 يومًا",
        "لم يتم التحقق من صحة النموذج للعملاء من فئة المؤسسات",
        "يتطلب حد أدنى من 6 أشهر من تاريخ المعاملات"
    ],
    ethical_considerations=[
        "لا ينبغي استخدام النموذج لرفض الخدمة",
        "يجب مراجعة التنبؤات من قبل الإنسان قبل اتخاذ الإجراءات"
    ],
    
    parent_model="customer-churn-model/1.0.0",
    training_run_id="abc123def456",
    data_version="data-v2.1.0",
    code_version="git-sha-789xyz"
)

# حفظ بطاقة النموذج
with open("model_card.json", "w") as f:
    json.dump(asdict(model_card), f, indent=2)

تسجيل سلالة النموذج​

# lineage_tracking.py
import mlflow
import git
import hashlib

def get_code_version():
    """الحصول على تجزئة الالتزام الحالية في git."""
    repo = git.Repo(search_parent_directories=True)
    return repo.head.object.hexsha

def get_data_hash(data_path: str) -> str:
    """حساب تجزئة ملف البيانات."""
    with open(data_path, 'rb') as f:
        return hashlib.md5(f.read()).hexdigest()

def train_with_lineage(X_train, y_train, X_test, y_test, data_path: str):
    """تدريب النموذج مع تتبع السلالة الكاملة."""
    
    with mlflow.start_run() as run:
        # تتبع إصدار الكود
        code_version = get_code_version()
        mlflow.set_tag("code_version", code_version)
        mlflow.set_tag("git_repo", "https://github.com/company/ml-project")
        
        # تتبع إصدار البيانات
        data_hash = get_data_hash(data_path)
        mlflow.set_tag("data_hash", data_hash)
        mlflow.set_tag("data_path", data_path)
        
        # تتبع البيئة
        mlflow.set_tag("python_version", "3.10.0")
        mlflow.set_tag("sklearn_version", "1.3.0")
        
        # تتبع تكوين التدريب
        mlflow.log_param("training_samples", len(X_train))
        mlflow.log_param("test_samples", len(X_test))
        mlflow.log_param("features", list(X_train.columns))
        
        # تدريب النموذج
        model = GradientBoostingClassifier(
            n_estimators=200,
            max_depth=7,
            learning_rate=0.1
        )
        model.fit(X_train, y_train)
        
        # تسجيل المقاييس
        train_accuracy = model.score(X_train, y_train)
        test_accuracy = model.score(X_test, y_test)
        mlflow.log_metric("train_accuracy", train_accuracy)
        mlflow.log_metric("test_accuracy", test_accuracy)
        
        # تسجيل النموذج مع التوقيع
        from mlflow.models.signature import infer_signature
        signature = infer_signature(X_train, model.predict(X_train))
        
        mlflow.sklearn.log_model(
            model,
            "model",
            signature=signature,
            registered_model_name="customer-churn-model"
        )
        
        return run.info.run_id

استراتيجيات إصدار النماذج​

الإصدار الدلالي للنماذج​

MAJOR.MINOR.PATCH

أمثلة:
- 1.0.0 → 1.0.1  # تصحيح: إصلاح خطأ، بدون إعادة تدريب
- 1.0.0 → 1.1.0  # ثانوي: إعادة تدريب، نفس الهيكل
- 1.0.0 → 2.0.0  # رئيسي: هيكل جديد أو تغييرات جذرية

زيادة الإصدار التلقائية​

# version_manager.py
import re
from enum import Enum
from mlflow.tracking import MlflowClient

class VersionBump(Enum):
    MAJOR = "major"
    MINOR = "minor"
    PATCH = "patch"

def get_next_version(current_version: str, bump: VersionBump) -> str:
    """حساب الإصدار الدلالي التالي."""
    match = re.match(r"(\d+)\.(\d+)\.(\d+)", current_version)
    if not match:
        return "1.0.0"
    
    major, minor, patch = map(int, match.groups())
    
    if bump == VersionBump.MAJOR:
        return f"{major + 1}.0.0"
    elif bump == VersionBump.MINOR:
        return f"{major}.{minor + 1}.0"
    else:
        return f"{major}.{minor}.{patch + 1}"

def register_new_version(
    model_name: str,
    run_id: str,
    bump: VersionBump,
    description: str
):
    """تسجيل النموذج مع الإصدار المناسب."""
    client = MlflowClient()
    
    # الحصول على أحدث إصدار
    versions = client.search_model_versions(f"name='{model_name}'")
    if versions:
        latest_version = max(v.version for v in versions)
        latest_tags = client.get_model_version(model_name, latest_version).tags
        current_semantic = latest_tags.get("semantic_version", "0.0.0")
    else:
        current_semantic = "0.0.0"
    
    # حساب الإصدار الدلالي الجديد
    new_semantic = get_next_version(current_semantic, bump)
    
    # تسجيل النموذج
    result = client.create_model_version(
        name=model_name,
        source=f"runs:/{run_id}/model",
        run_id=run_id,
        description=description
    )
    
    # تعيين علامة الإصدار الدلالي
    client.set_model_version_tag(
        name=model_name,
        version=result.version,
        key="semantic_version",
        value=new_semantic
    )
    
    print(f"Registered {model_name} v{result.version} (semantic: {new_semantic})")
    return result

سير عمل الموافقة على النموذج​

خط أنابيب التحقق التلقائي​

# model_validation.py
from typing import Dict, List, Tuple
import mlflow
from mlflow.tracking import MlflowClient

class ModelValidator:
    """التحقق من النماذج قبل النشر في الإنتاج."""
    
    def __init__(self, model_name: str, version: int):
        self.client = MlflowClient()
        self.model_name = model_name
        self.version = version
        self.model_uri = f"models:/{model_name}/{version}"
        
    def validate_performance(
        self, 
        X_test, 
        y_test,
        thresholds: Dict[str, float]
    ) -> Tuple[bool, Dict[str, float]]:
        """التحقق مما إذا كان النموذج يفي بعتبات الأداء."""
        model = mlflow.sklearn.load_model(self.model_uri)
        
        from sklearn.metrics import accuracy_score, roc_auc_score, f1_score
        
        predictions = model.predict(X_test)
        probabilities = model.predict_proba(X_test)[:, 1]
        
        metrics = {
            "accuracy": accuracy_score(y_test, predictions),
            "auc_roc": roc_auc_score(y_test, probabilities),
            "f1_score": f1_score(y_test, predictions)
        }
        
        # التحقق من العتبات
        passed = all(
            metrics.get(metric, 0) >= threshold
            for metric, threshold in thresholds.items()
        )
        
        return passed, metrics
    
    def validate_latency(
        self, 
        X_sample,
        max_latency_ms: float = 100
    ) -> Tuple[bool, float]:
        """التحقق مما إذا كان النموذج يفي بمتطلبات التأخير."""
        import time
        
        model = mlflow.sklearn.load_model(self.model_uri)
        
        # تسخين
        _ = model.predict(X_sample[:1])
        
        # قياس التأخير
        times = []
        for _ in range(100):
            start = time.time()
            _ = model.predict(X_sample[:1])
            times.append((time.time() - start) * 1000)
        
        p99_latency = sorted(times)[98]
        passed = p99_latency <= max_latency_ms
        
        return passed, p99_latency
    
    def validate_consistency(
        self,
        X_sample,
        production_model_uri: str
    ) -> Tuple[bool, float]:
        """التحقق من تطابق التنبؤات مع نموذج الإنتاج."""
        new_model = mlflow.sklearn.load_model(self.model_uri)
        prod_model = mlflow.sklearn.load_model(production_model_uri)
        
        new_predictions = new_model.predict(X_sample)
        prod_predictions = prod_model.predict(X_sample)
        
        agreement = (new_predictions == prod_predictions).mean()
        passed = agreement >= 0.95  # على الأقل 95% توافق
        
        return passed, agreement
    
    def run_all_validations(
        self,
        X_test,
        y_test,
        thresholds: Dict[str, float]
    ) -> Dict[str, any]:
        """تشغيل جميع عمليات التحقق وإرجاع النتائج."""
        results = {
            "model_name": self.model_name,
            "version": self.version,
            "validations": {}
        }
        
        # التحقق من الأداء
        perf_passed, perf_metrics = self.validate_performance(
            X_test, y_test, thresholds
        )
        results["validations"]["performance"] = {
            "passed": perf_passed,
            "metrics": perf_metrics
        }
        
        # التحقق من التأخير
        latency_passed, p99_latency = self.validate_latency(X_test[:100])
        results["validations"]["latency"] = {
            "passed": latency_passed,
            "p99_ms": p99_latency
        }
        
        # النتيجة العامة
        results["approved"] = all(
            v["passed"] for v in results["validations"].values()
        )
        
        # تحديث علامات النموذج
        status = "approved" if results["approved"] else "rejected"
        self.client.set_model_version_tag(
            name=self.model_name,
            version=self.version,
            key="validation_status",
            value=status
        )
        
        return results

النقاط الرئيسية​

ملخص
  1. استخدام سجل النماذج لإدارة النماذج المركزية
  2. تتبع السلالة الكاملة - البيانات، الكود، المعاملات، البيئة
  3. تنفيذ الإصدار الدلالي لتقدم الإصدارات الواضح
  4. إنشاء بطاقات النماذج للتوثيق والامتثال
  5. أتمتة التحقق قبل النشر في الإنتاج
  6. الحفاظ على سجلات التدقيق للصناعات المنظمة

تمرين عملي​

  1. إعداد خادم تتبع MLflow
  2. تدريب وتسجيل إصدارات متعددة من النماذج
  3. إنشاء بطاقة نموذج لأفضل نموذج لديك
  4. تنفيذ خط أنابيب التحقق من النموذج
  5. ممارسة نقل النماذج بين المراحل

التالي: اختبار الوحدة 2