التسجيل وتصحيح أنظمة ML

تصحيح ML الإنتاجي يتطلب تقنيات متخصصة. يختبر المحاورون قدرتك على تشخيص المشاكل التي تمتد عبر البيانات والنماذج والبنية التحتية.

متطلبات التسجيل الخاصة بـ ML

نوع السجل	ما يجب التقاطه	لماذا يهم
سجلات الطلب	ميزات المدخلات، الطوابع الزمنية	إعادة إنتاج المشاكل
سجلات التوقع	المخرجات، الثقة، زمن الاستجابة	تصحيح التوقعات الخاطئة
سجلات النموذج	الإصدار، وقت التحميل، الأخطاء	تتبع دورة حياة النموذج
سجلات البيانات	المخطط، معدلات null، النطاقات	اكتشاف مشاكل البيانات

التسجيل المنظم لـ ML

import structlog
import json
from datetime import datetime

# تكوين التسجيل المنظم
structlog.configure(
    processors=[
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.JSONRenderer()
    ]
)

logger = structlog.get_logger()

def log_prediction(request_id, model_name, features, prediction, latency_ms):
    logger.info(
        "prediction_completed",
        request_id=request_id,
        model_name=model_name,
        model_version=get_model_version(model_name),
        feature_count=len(features),
        # عينة الميزات للتصحيح (ليس الكل - الخصوصية + الحجم)
        feature_sample={k: features[k] for k in list(features.keys())[:5]},
        prediction=prediction,
        confidence=prediction.get("confidence"),
        latency_ms=latency_ms,
        timestamp=datetime.utcnow().isoformat()
    )

المخرجات:

{
  "event": "prediction_completed",
  "request_id": "abc-123",
  "model_name": "fraud_detector",
  "model_version": "v2.3.1",
  "feature_count": 45,
  "feature_sample": {"amount": 150.0, "merchant_id": "M123"},
  "prediction": {"label": "fraud", "score": 0.87},
  "confidence": 0.87,
  "latency_ms": 23,
  "timestamp": "2026-01-04T10:30:00Z"
}

سؤال المقابلة: تصحيح التوقعات الخاطئة

السؤال: "يبلغ المستخدمون أن نموذج الاحتيال يُعلم المعاملات المشروعة. كيف تصحح؟"

نهج التصحيح المنظم:

def debug_false_positives(flagged_transactions):
    """
    تصحيح خطوة بخطوة للتوقعات الخاطئة
    """

    # الخطوة 1: إعادة إنتاج التوقع
    debug_steps = {}

    debug_steps["1_replay_prediction"] = """
    - جلب ميزات المدخلات الدقيقة من سجلات التوقع
    - تشغيل نفس إصدار النموذج مع تلك الميزات
    - التحقق من أننا نحصل على نفس التوقع (التكرارية)
    """

    debug_steps["2_analyze_features"] = """
    - مقارنة المعاملات المُعلمة ببيانات التدريب
    - تحديد الميزات الشاذة (المبلغ، الوقت، الموقع)
    - التحقق من القيم المفقودة أو غير المتوقعة
    """

    debug_steps["3_check_feature_pipeline"] = """
    - التحقق من أن مخزن الميزات يرجع القيم الصحيحة
    - التحقق من الميزات المخزنة مؤقتاً القديمة
    - البحث عن أخطاء تحويل الميزات
    """

    debug_steps["4_analyze_model_behavior"] = """
    - الحصول على أهمية الميزات لهذه التوقعات
    - استخدام SHAP/LIME للتفسيرات المحلية
    - تحديد الميزات التي دفعت القرار
    """

    debug_steps["5_check_for_drift"] = """
    - مقارنة توزيع بيانات الإنتاج بالتدريب
    - التحقق مما إذا كان هذا يمثل شريحة مستخدمين جديدة
    - تقييم ما إذا كان النموذج قد تدرب على حالات مماثلة
    """

    return debug_steps

التتبع لأنظمة ML

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

# إعداد التتبع
trace.set_tracer_provider(TracerProvider())
tracer = trace.get_tracer(__name__)

jaeger_exporter = JaegerExporter(
    agent_host_name="jaeger",
    agent_port=6831,
)
trace.get_tracer_provider().add_span_processor(
    BatchSpanProcessor(jaeger_exporter)
)

@tracer.start_as_current_span("predict")
def predict(request):
    span = trace.get_current_span()

    # تتبع جلب الميزات
    with tracer.start_as_current_span("fetch_features"):
        features = fetch_from_feature_store(request.user_id)
        span.set_attribute("feature_count", len(features))

    # تتبع المعالجة المسبقة
    with tracer.start_as_current_span("preprocess"):
        processed = preprocess(features)
        span.set_attribute("preprocessing_time_ms", timer.elapsed())

    # تتبع الاستدلال
    with tracer.start_as_current_span("model_inference"):
        prediction = model.predict(processed)
        span.set_attribute("model_version", model.version)
        span.set_attribute("prediction", str(prediction))

    return prediction

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

# خط أنابيب السجل الموصى به
log_architecture:
  application_logs:
    format: "JSON منظم"
    destination: "Loki أو Elasticsearch"
    retention: "30 يوم"

  prediction_logs:
    format: "JSON مع ربط request_id"
    destination: "مخزن سجل ML مخصص (BigQuery)"
    retention: "90 يوم (لتصحيح النموذج)"

  metrics:
    format: "صيغة عرض Prometheus"
    destination: "Prometheus + Thanos (طويل المدى)"
    retention: "سنة واحدة"

  traces:
    format: "OpenTelemetry"
    destination: "Jaeger أو Tempo"
    retention: "7 أيام"

سيناريوهات التصحيح الشائعة

العرض	السبب المحتمل	التحقيق
انخفاض مفاجئ في الدقة	تغيير خط أنابيب البيانات	تحقق من مخطط المنبع
انخفاض تدريجي في الدقة	انحراف المفهوم	قارن التوزيعات
ارتفاعات زمن الاستجابة العالية	ضغط الذاكرة	تحقق من سجلات ذاكرة GPU
أخطاء عشوائية	تبعية متقطعة	تحقق من فترات التتبع
توقعات خاطئة	خطأ في الميزات	قارن الميزات المسجلة مقابل المتوقعة

نصيحة احترافية: "نحتفظ بسجلات التوقع لمدة 90 يوماً حتى نتمكن من إعادة تشغيل أي توقع عندما يبلغ المستخدمون عن مشاكل. request_id هو مفتاح الربط عبر جميع الأنظمة."

الوحدة التالية تغطي أسئلة مقابلات CI/CD لأنظمة ML. :::