نشر Canary واختبار A/B — أساسيات MLOps: خطوط أنابيب التعلم الآلي والتحكم في الإصدارات والبنية التحتية للإنتاج — Nerd Level Tech

نشر نماذج ML بأمان يتطلب إطلاق تدريجي وتحقق قائم على البيانات. نشر Canary واختبار A/B يُقللان المخاطر بينما يُعظمان التعلم.

استراتيجيات النشر

الاستراتيجية	المخاطرة	سرعة التراجع	حالة الاستخدام
Big Bang	عالية	بطيء	التطوير فقط
Blue-Green	متوسطة	سريع	تبديل سريع
Canary	منخفضة	سريع	إنتاج ML
اختبار A/B	منخفضة	سريع	مقارنة النماذج

نشر Canary

المفهوم

انشر النموذج الجديد لنسبة صغيرة من حركة المرور، راقب المقاييس، ثم زد تدريجياً.

تدفق حركة المرور:
┌─────────────────────────────────────────────┐
│                                             │
│  المستخدمون (100%)                           │
│      │                                      │
│      ▼                                      │
│  ┌─────────┐                                │
│  │  موزع   │                                │
│  │ الحمل   │                                │
│  └────┬────┘                                │
│       │                                     │
│   ┌───┴───┐                                 │
│   │       │                                 │
│   ▼       ▼                                 │
│ ┌───┐   ┌───┐                               │
│ │95%│   │ 5%│  ◄── Canary                   │
│ │v1 │   │v2 │                               │
│ └───┘   └───┘                               │
└─────────────────────────────────────────────┘

تنفيذ Kubernetes

# model-v1-deployment.yaml (مستقر)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: fraud-detector-stable
  labels:
    app: fraud-detector
    version: v1
spec:
  replicas: 9  # 90% من حركة المرور
  selector:
    matchLabels:
      app: fraud-detector
      version: v1
  template:
    metadata:
      labels:
        app: fraud-detector
        version: v1
    spec:
      containers:
      - name: model
        image: myregistry/fraud-detector:v1
        ports:
        - containerPort: 3000
---
# model-v2-deployment.yaml (canary)
apiVersion: apps/v1
kind: Deployment
metadata:
  name: fraud-detector-canary
  labels:
    app: fraud-detector
    version: v2
spec:
  replicas: 1  # 10% من حركة المرور
  selector:
    matchLabels:
      app: fraud-detector
      version: v2
  template:
    metadata:
      labels:
        app: fraud-detector
        version: v2
    spec:
      containers:
      - name: model
        image: myregistry/fraud-detector:v2
        ports:
        - containerPort: 3000
---
# service.yaml (يوجه لكليهما)
apiVersion: v1
kind: Service
metadata:
  name: fraud-detector
spec:
  selector:
    app: fraud-detector  # يطابق v1 و v2
  ports:
  - port: 80
    targetPort: 3000

تقسيم حركة المرور مع Istio

# virtual-service.yaml
apiVersion: networking.istio.io/v1beta1
kind: VirtualService
metadata:
  name: fraud-detector
spec:
  hosts:
  - fraud-detector
  http:
  - route:
    - destination:
        host: fraud-detector
        subset: stable
      weight: 95
    - destination:
        host: fraud-detector
        subset: canary
      weight: 5
---
# destination-rule.yaml
apiVersion: networking.istio.io/v1beta1
kind: DestinationRule
metadata:
  name: fraud-detector
spec:
  host: fraud-detector
  subsets:
  - name: stable
    labels:
      version: v1
  - name: canary
    labels:
      version: v2

سكربت الإطلاق التدريجي

# canary_rollout.py
import subprocess
import time
from typing import Callable

def update_traffic_weight(canary_weight: int):
    """حدّث وزن VirtualService في Istio."""
    stable_weight = 100 - canary_weight

    patch = f'''
    spec:
      http:
      - route:
        - destination:
            host: fraud-detector
            subset: stable
          weight: {stable_weight}
        - destination:
            host: fraud-detector
            subset: canary
          weight: {canary_weight}
    '''

    subprocess.run([
        "kubectl", "patch", "virtualservice", "fraud-detector",
        "--type=merge", "-p", patch
    ])

def canary_rollout(
    check_metrics: Callable[[], bool],
    weights: list[int] = [5, 10, 25, 50, 100],
    wait_minutes: int = 10
):
    """زد حركة مرور canary تدريجياً إذا نجحت المقاييس."""
    for weight in weights:
        print(f"تعيين وزن canary إلى {weight}%")
        update_traffic_weight(weight)

        print(f"انتظار {wait_minutes} دقائق...")
        time.sleep(wait_minutes * 60)

        if not check_metrics():
            print("فشل فحص المقاييس! التراجع...")
            update_traffic_weight(0)
            return False

        print(f"نجحت المقاييس عند {weight}%")

    print("اكتمل إطلاق canary!")
    return True

# الاستخدام
def check_model_metrics() -> bool:
    """تحقق إذا كان نموذج canary يفي بـ SLOs."""
    # استعلم Prometheus/Grafana
    latency_ok = get_p99_latency("canary") < 100  # مللي ثانية
    error_rate_ok = get_error_rate("canary") < 0.01  # 1%
    accuracy_ok = get_accuracy("canary") > 0.90

    return latency_ok and error_rate_ok and accuracy_ok

canary_rollout(check_model_metrics)

اختبار A/B للنماذج

المفهوم

قسّم حركة المرور بين النماذج لقياس تأثير الأعمال، ليس فقط المقاييس التقنية.

Canary	اختبار A/B
تحقق من الاستقرار	قس تأثير الأعمال
إطلاق سريع	أهمية إحصائية
مقاييس تقنية	مقاييس أعمال
مؤقت	يمكن أن يستمر لأسابيع

التنفيذ مع Feature Flags

# ab_test_router.py
import hashlib
from enum import Enum

class ModelVariant(Enum):
    CONTROL = "v1"
    TREATMENT = "v2"

def get_variant(user_id: str, experiment: str, treatment_pct: int = 50) -> ModelVariant:
    """تعيين حتمي بناءً على user_id."""
    hash_input = f"{user_id}:{experiment}"
    hash_value = int(hashlib.md5(hash_input.encode()).hexdigest(), 16)
    bucket = hash_value % 100

    if bucket < treatment_pct:
        return ModelVariant.TREATMENT
    return ModelVariant.CONTROL

# الاستخدام في الخدمة
class FraudDetectorABTest:
    def __init__(self):
        self.model_v1 = load_model("fraud_detector:v1")
        self.model_v2 = load_model("fraud_detector:v2")

    def predict(self, user_id: str, features: dict) -> dict:
        variant = get_variant(user_id, "fraud_model_v2_test")

        if variant == ModelVariant.TREATMENT:
            model = self.model_v2
        else:
            model = self.model_v1

        prediction = model.predict(features)

        # سجّل للتحليل
        log_prediction(
            user_id=user_id,
            variant=variant.value,
            prediction=prediction,
            timestamp=time.time()
        )

        return {
            "prediction": prediction,
            "variant": variant.value
        }

التحليل الإحصائي

# ab_analysis.py
import scipy.stats as stats
import pandas as pd

def analyze_ab_test(
    control_conversions: int,
    control_total: int,
    treatment_conversions: int,
    treatment_total: int,
    confidence: float = 0.95
) -> dict:
    """حلل نتائج اختبار A/B."""
    control_rate = control_conversions / control_total
    treatment_rate = treatment_conversions / treatment_total

    # اختبار z لنسبتين
    pooled = (control_conversions + treatment_conversions) / (control_total + treatment_total)
    se = (pooled * (1 - pooled) * (1/control_total + 1/treatment_total)) ** 0.5
    z_stat = (treatment_rate - control_rate) / se
    p_value = 2 * (1 - stats.norm.cdf(abs(z_stat)))

    # فاصل الثقة
    z_critical = stats.norm.ppf((1 + confidence) / 2)
    diff = treatment_rate - control_rate
    margin = z_critical * se

    return {
        "control_rate": control_rate,
        "treatment_rate": treatment_rate,
        "lift": (treatment_rate - control_rate) / control_rate * 100,
        "p_value": p_value,
        "significant": p_value < (1 - confidence),
        "confidence_interval": (diff - margin, diff + margin)
    }

# مثال استخدام
results = analyze_ab_test(
    control_conversions=450,
    control_total=10000,
    treatment_conversions=520,
    treatment_total=10000
)

print(f"Control: {results['control_rate']:.2%}")
print(f"Treatment: {results['treatment_rate']:.2%}")
print(f"الرفع: {results['lift']:.1f}%")
print(f"P-value: {results['p_value']:.4f}")
print(f"ذو دلالة: {results['significant']}")

مراقبة نشر Canary

المقاييس الرئيسية للتتبع

# prometheus-rules.yaml
groups:
- name: canary-alerts
  rules:
  - alert: CanaryHighLatency
    expr: |
      histogram_quantile(0.99,
        sum(rate(http_request_duration_seconds_bucket{version="canary"}[5m]))
        by (le)
      ) > 0.5
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "زمن استجابة canary p99 فوق 500ms"

  - alert: CanaryHighErrorRate
    expr: |
      sum(rate(http_requests_total{version="canary",status=~"5.."}[5m]))
      /
      sum(rate(http_requests_total{version="canary"}[5m]))
      > 0.05
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "معدل خطأ canary فوق 5%"

استعلام لوحة Grafana

# قارن زمن الاستجابة بين الإصدارات
histogram_quantile(0.99,
  sum by (version, le) (
    rate(http_request_duration_seconds_bucket[5m])
  )
)

# قارن معدلات الأخطاء
sum by (version) (
  rate(http_requests_total{status=~"5.."}[5m])
) / sum by (version) (
  rate(http_requests_total[5m])
)

التراجع الآلي

Argo Rollouts

# rollout.yaml
apiVersion: argoproj.io/v1alpha1
kind: Rollout
metadata:
  name: fraud-detector
spec:
  replicas: 10
  selector:
    matchLabels:
      app: fraud-detector
  template:
    metadata:
      labels:
        app: fraud-detector
    spec:
      containers:
      - name: model
        image: myregistry/fraud-detector:v2
  strategy:
    canary:
      steps:
      - setWeight: 5
      - pause: {duration: 10m}
      - setWeight: 20
      - pause: {duration: 10m}
      - setWeight: 50
      - pause: {duration: 10m}
      - setWeight: 100
      analysis:
        templates:
        - templateName: success-rate
        startingStep: 1
---
apiVersion: argoproj.io/v1alpha1
kind: AnalysisTemplate
metadata:
  name: success-rate
spec:
  metrics:
  - name: success-rate
    interval: 1m
    successCondition: result[0] >= 0.95
    provider:
      prometheus:
        address: http://prometheus:9090
        query: |
          sum(rate(http_requests_total{status!~"5.."}[5m]))
          /
          sum(rate(http_requests_total[5m]))

أفضل الممارسات

الممارسة	لماذا
ابدأ صغيراً	1-5% canary أولي
راقب مقاييس الأعمال	ليس فقط زمن الاستجابة/الأخطاء
أتمت التراجع	تفاعل أسرع من البشر
استخدم توجيه حتمي	تجربة مستخدم متسقة
شغّل وقتاً كافياً	أهمية إحصائية

الرؤية الرئيسية: نشر النموذج الآمن تكراري—انشر صغيراً، قس التأثير، أتمت القرارات، ودائماً لديك خطة تراجع.

الوحدة التالية: مراقبة ML وعمليات الإنتاج. :::