مراقبة أداء النموذج

بعد انحراف البيانات، يجب مراقبة تنبؤات النموذج الفعلية وأدائه. هذا يكتشف التدهور حتى عندما تبدو بيانات المدخلات مستقرة.

ما يجب مراقبته

الفئة	المقاييس	لماذا
زمن الاستجابة	p50، p95، p99 وقت الاستجابة	تجربة المستخدم، SLA
الإنتاجية	طلبات/ثانية	تخطيط السعة
الأخطاء	معدل 5xx، معدل timeout	الموثوقية
التنبؤات	التوزيع، درجات الثقة	سلوك النموذج
الأعمال	التحويل، تأثير الإيرادات	القيمة الفعلية

مقاييس Prometheus لـ ML

مقاييس مخصصة

# metrics.py
from prometheus_client import Counter, Histogram, Gauge, start_http_server
import time

# مقاييس الطلبات
prediction_requests = Counter(
    'ml_prediction_requests_total',
    'إجمالي طلبات التنبؤ',
    ['model_name', 'model_version']
)

prediction_latency = Histogram(
    'ml_prediction_latency_seconds',
    'زمن استجابة التنبؤ',
    ['model_name'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5]
)

prediction_errors = Counter(
    'ml_prediction_errors_total',
    'إجمالي أخطاء التنبؤ',
    ['model_name', 'error_type']
)

# مقاييس مخرجات النموذج
prediction_confidence = Histogram(
    'ml_prediction_confidence',
    'درجات ثقة النموذج',
    ['model_name', 'predicted_class'],
    buckets=[0.5, 0.6, 0.7, 0.8, 0.9, 0.95, 0.99]
)

prediction_class_distribution = Counter(
    'ml_prediction_class_total',
    'التنبؤات حسب الفئة',
    ['model_name', 'predicted_class']
)

# إحصائيات الميزات
feature_value = Gauge(
    'ml_feature_value',
    'إحصائيات الميزات',
    ['feature_name', 'statistic']  # mean، std، min، max
)

خدمة تنبؤ مُراقبة

# service.py
import time
from contextlib import contextmanager

class MonitoredPredictor:
    def __init__(self, model, model_name: str, model_version: str):
        self.model = model
        self.model_name = model_name
        self.model_version = model_version

    @contextmanager
    def track_latency(self):
        start = time.perf_counter()
        try:
            yield
        finally:
            duration = time.perf_counter() - start
            prediction_latency.labels(model_name=self.model_name).observe(duration)

    def predict(self, features: dict) -> dict:
        prediction_requests.labels(
            model_name=self.model_name,
            model_version=self.model_version
        ).inc()

        try:
            with self.track_latency():
                prediction = self.model.predict([features])
                confidence = self.model.predict_proba([features])

            # تتبع المخرجات
            predicted_class = str(prediction[0])
            prediction_class_distribution.labels(
                model_name=self.model_name,
                predicted_class=predicted_class
            ).inc()

            max_confidence = float(max(confidence[0]))
            prediction_confidence.labels(
                model_name=self.model_name,
                predicted_class=predicted_class
            ).observe(max_confidence)

            return {
                "prediction": predicted_class,
                "confidence": max_confidence
            }

        except Exception as e:
            prediction_errors.labels(
                model_name=self.model_name,
                error_type=type(e).__name__
            ).inc()
            raise

# ابدأ خادم المقاييس
start_http_server(8000)

لوحات Grafana

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

# معدل الطلبات حسب النموذج
sum(rate(ml_prediction_requests_total[5m])) by (model_name)

# زمن استجابة P99
histogram_quantile(0.99,
  sum(rate(ml_prediction_latency_seconds_bucket[5m])) by (le, model_name)
)

# معدل الأخطاء
sum(rate(ml_prediction_errors_total[5m])) by (model_name)
/
sum(rate(ml_prediction_requests_total[5m])) by (model_name)

# توزيع التنبؤات (لاكتشاف الانحراف)
sum(rate(ml_prediction_class_total[1h])) by (predicted_class)

# تنبؤات منخفضة الثقة (مشاكل محتملة)
histogram_quantile(0.10,
  sum(rate(ml_prediction_confidence_bucket[1h])) by (le)
)

JSON لوحة التحكم

{
  "panels": [
    {
      "title": "طلبات التنبؤ/ثانية",
      "type": "graph",
      "targets": [{
        "expr": "sum(rate(ml_prediction_requests_total[5m])) by (model_name)"
      }]
    },
    {
      "title": "زمن استجابة P99",
      "type": "graph",
      "targets": [{
        "expr": "histogram_quantile(0.99, sum(rate(ml_prediction_latency_seconds_bucket[5m])) by (le))"
      }]
    },
    {
      "title": "معدل الأخطاء",
      "type": "stat",
      "targets": [{
        "expr": "sum(rate(ml_prediction_errors_total[5m])) / sum(rate(ml_prediction_requests_total[5m]))"
      }]
    }
  ]
}

قواعد التنبيه

قواعد تنبيه Prometheus

# prometheus-rules.yaml
groups:
- name: ml-model-alerts
  rules:
  # زمن استجابة عالي
  - alert: MLModelHighLatency
    expr: |
      histogram_quantile(0.99,
        sum(rate(ml_prediction_latency_seconds_bucket[5m])) by (le, model_name)
      ) > 0.5
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "النموذج {{ $labels.model_name }} زمن استجابة p99 > 500ms"

  # معدل أخطاء عالي
  - alert: MLModelHighErrorRate
    expr: |
      sum(rate(ml_prediction_errors_total[5m])) by (model_name)
      /
      sum(rate(ml_prediction_requests_total[5m])) by (model_name)
      > 0.01
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "النموذج {{ $labels.model_name }} معدل أخطاء > 1%"

  # انحراف التنبؤ (تغير عدم توازن الفئات)
  - alert: MLModelPredictionSkew
    expr: |
      max(
        sum(rate(ml_prediction_class_total[1h])) by (predicted_class)
      ) / sum(rate(ml_prediction_class_total[1h]))
      > 0.95
    for: 1h
    labels:
      severity: warning
    annotations:
      summary: "تنبؤات النموذج منحرفة بشدة لفئة واحدة"

  # تزايد تنبؤات منخفضة الثقة
  - alert: MLModelLowConfidence
    expr: |
      histogram_quantile(0.25,
        sum(rate(ml_prediction_confidence_bucket[1h])) by (le)
      ) < 0.6
    for: 1h
    labels:
      severity: warning
    annotations:
      summary: "25% من التنبؤات لها ثقة < 60%"

  # لا تنبؤات (النموذج متوقف)
  - alert: MLModelDown
    expr: |
      sum(rate(ml_prediction_requests_total[5m])) by (model_name) == 0
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "النموذج {{ $labels.model_name }} لا يستلم طلبات"

مراقبة الحقيقة الأرضية

جمع التسميات المتأخرة

# ground_truth_monitor.py
from datetime import datetime, timedelta
import pandas as pd

class GroundTruthMonitor:
    def __init__(self, prediction_store, label_store):
        self.prediction_store = prediction_store
        self.label_store = label_store

    def calculate_accuracy(self, start: datetime, end: datetime) -> dict:
        """احسب الدقة للتنبؤات مع الحقيقة الأرضية."""
        predictions = self.prediction_store.query(start, end)
        labels = self.label_store.query(start, end)

        # ادمج على prediction_id
        joined = predictions.merge(labels, on="prediction_id", how="inner")

        if len(joined) == 0:
            return {"accuracy": None, "sample_size": 0}

        correct = (joined["predicted"] == joined["actual"]).sum()
        total = len(joined)

        return {
            "accuracy": correct / total,
            "sample_size": total,
            "precision": self._precision(joined),
            "recall": self._recall(joined),
            "f1": self._f1(joined)
        }

    def monitor_rolling_accuracy(self, window_days: int = 7):
        """راقب الدقة على نافذة متدحرجة."""
        end = datetime.now()
        start = end - timedelta(days=window_days)

        metrics = self.calculate_accuracy(start, end)

        # صدّر لـ Prometheus
        accuracy_gauge.set(metrics["accuracy"])
        sample_size_gauge.set(metrics["sample_size"])

        # أنبه إذا انخفضت الدقة
        if metrics["accuracy"] < 0.85:
            send_alert(
                f"دقة النموذج انخفضت إلى {metrics['accuracy']:.2%} "
                f"(عينة: {metrics['sample_size']})"
            )

        return metrics

مراقبة SLA

تعريفات SLO

# slo.yaml
slos:
  fraud_detector:
    availability:
      target: 99.9
      window: 30d
    latency:
      p99_target_ms: 100
      p50_target_ms: 20
    accuracy:
      target: 0.95
      window: 7d

تنبيهات معدل حرق SLO

# معدل حرق عالي = استخدام ميزانية الخطأ بسرعة
- alert: MLModelErrorBudgetBurn
  expr: |
    (
      sum(rate(ml_prediction_errors_total[1h]))
      /
      sum(rate(ml_prediction_requests_total[1h]))
    ) > (1 - 0.999) * 14.4
  for: 5m
  labels:
    severity: critical
  annotations:
    summary: "حرق ميزانية الخطأ 14.4x أسرع من المستدام"

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

الممارسة	التنفيذ
راقب التنبؤات، ليس فقط المدخلات	تتبع توزيع المخرجات
عيّن خطوط أساس من الإنتاج	ليس بيانات التدريب
أنبه على الاتجاهات، ليس الضوضاء	استخدم نوافذ متدحرجة
ضمّن مقاييس الأعمال	الإيرادات، التحويل
أتمت المعالجة	تراجع، محفزات إعادة التدريب

الرؤية الرئيسية: المراقبة الجيدة تجيب "هل نموذجي يساعد المستخدمين؟" ليس فقط "هل نموذجي يعمل؟"

التالي، سنستكشف حوكمة ML ومتطلبات الامتثال. :::