الوحدة 3 — تحميل النموذج عند بدء التشغيل

في الوحدتَين السابقتَين بنينا مسارًا يستقبل طلبات ويتحقّق من مدخلاتها، لكنّه لا يستدعي أيّ نموذج بعد. حان الوقت لتحميل نموذج تسرّب المشتركين. الخيار الذي يُتّخذ هنا يُقرّر أداء الخدمة كاملًا: هل نُحمّل النموذج مرّة واحدة عند البدء ونُبقيه في الذاكرة، أم في كلّ طلب؟ الإجابة الوحيدة الصحيحة في الإنتاج هي الأولى، وهذه الوحدة تشرح كيفيّة ذلك ولماذا الخطأ فيها يُكلّف كثيرًا.

الخطأ المُغري: التحميل في كلّ طلب

الشيفرة التالية تعمل، ولذلك خطيرة:

from fastapi import FastAPI
import joblib

app = FastAPI()


@app.post("/predict")
def predire(req: dict) -> dict:
    modele = joblib.load("churn_model.joblib")  # في كلّ طلب !
    proba = modele.predict_proba([list(req.values())])[0][1]
    return {"proba": float(proba)}

المشكلة: joblib.load يقرأ القرص، يفكّ التسلسل (deserialize)، ويعيد بناء الشجرة الكاملة للنموذج في الذاكرة. لنموذج LightGBM بألف شجرة، هذا يستغرق 200 إلى 800 ميلّي ثانية. طلب التنبّؤ نفسه بضع ميلّي ثوانٍ. النتيجة: زمن استجابة يزيد ×100 بلا سبب.

الأسوأ: تحت حمل عالٍ (مئة طلب في الثانية)، عمليّة تحميل متزامنة تُشبع نوى الـCPU والقرص، فتنهار الخدمة بلا سبب ظاهر.

الحلّ: lifespan

FastAPI منذ إصداره 0.93 يوفّر واجهة lifespan (مبنيّة على مدير سياق async) تُشغّل شيفرة قبل استقبال أوّل طلب وأخرى بعد إغلاق الخدمة. نُحمّل النموذج مرّة واحدة ونحتفظ به في app.state:

from contextlib import asynccontextmanager
from fastapi import FastAPI
import joblib


@asynccontextmanager
async def lifespan(app: FastAPI):
    print("[startup] chargement du modele...")
    app.state.model = joblib.load("churn_model.joblib")
    app.state.model_version = "churn-lgbm-v1.3.0"
    print("[startup] modele charge :", app.state.model_version)
    yield
    print("[shutdown] liberation du modele")
    app.state.model = None


app = FastAPI(lifespan=lifespan)


@app.post("/predict")
def predire(req: dict) -> dict:
    modele = app.state.model
    proba = modele.predict_proba([list(req.values())])[0][1]
    return {"proba": float(proba), "model_version": app.state.model_version}

الآن التحميل يحدث مرّة واحدة قبل استقبال الطلب الأوّل. كلّ عامل Uvicorn لديه نسخة من النموذج في ذاكرته (نتحدّث عن هذا في الوحدة 9)، لكن داخل عامل واحد لا يُقرأ الملفّ إلّا مرّة.

Uvicorn قبل 0.93 أو نسخ FastAPI القديمة تستعمل @app.on_event("startup") و@app.on_event("shutdown"). هذان مُبطَلان الآن؛ استعمل lifespan.

كشف إصدار النموذج

فرق الإنتاج تحتاج معرفة أيّ نموذج يخدم الطلبات في هذه اللحظة، لأربعة أسباب: التشخيص عند خطأ، تحديد لحظة تدهور الأداء، مطابقة سجلّ MLflow، وإعلام العميل بالتغييرات. نُصرّح بمسار مخصّص:

class InfoModele(BaseModel):
    model_version: str
    loaded_at: str
    input_features: list[str]


@app.get("/model/info", response_model=InfoModele)
def info_modele() -> InfoModele:
    return InfoModele(
        model_version=app.state.model_version,
        loaded_at=app.state.loaded_at,
        input_features=app.state.feature_names,
    )

سنُثريه في الوحدة 8 بمسارَي liveness/readiness.

الفشل عند البدء بدل الفشل عند أوّل طلب

قاعدة ذهبيّة: إذا كان النموذج مفقودًا أو تالفًا، الخدمة يجب أن ترفض البدء، لا أن تبدأ وتفشل عند أوّل طلب. مثال:

@asynccontextmanager
async def lifespan(app: FastAPI):
    chemin = os.getenv("MODEL_PATH", "churn_model.joblib")
    if not os.path.exists(chemin):
        raise RuntimeError(f"modele introuvable : {chemin}")
    try:
        app.state.model = joblib.load(chemin)
    except Exception as e:
        raise RuntimeError(f"chargement du modele echoue : {e}") from e

    # sanity check : le modele repond a un exemple minimal
    _ = app.state.model.predict_proba([[24, 79.5, 0, 0, 1, 0]])
    yield

النتيجة: إن نُشرت صورة Docker بنموذج تالف، Uvicorn يخرج فورًا بخطأ مقروء. مُنسّق الحاويات (Kubernetes، ECS) يكتشف الفشل ولا يُوجّه إليه أيّ حركة. لو انتظرنا الطلب الأوّل، لكان مستخدم حقيقيّ من رأى الخطأ. مبدأ fail fast غير قابل للتفاوض في MLOps.

تحميل من MLflow Registry

في الوحدة 5 من دورة MLOps بنينا سجلّ نماذج MLflow. تحميل مباشر منه بديل أنيق للملفّ المحلّيّ:

import mlflow.pyfunc


@asynccontextmanager
async def lifespan(app: FastAPI):
    uri = f"models:/churn/{os.getenv('MODEL_STAGE', 'Production')}"
    app.state.model = mlflow.pyfunc.load_model(uri)
    app.state.model_version = mlflow.tracking.MlflowClient() \
        .get_latest_versions("churn", stages=["Production"])[0].version
    yield

المزيّة: تغيير النموذج في السجلّ يتطلّب فقط إعادة تشغيل الخدمة، بلا إعادة بناء صورة. العيب: تبعيّة شبكيّة عند البدء؛ يجب معالجة عدم توفّر MLflow بمنطق إعادة محاولة.

مشاركة النموذج بين العمّال

عمليًّا كلّ عامل Uvicorn عمليّة مستقلّة، فذاكرتها منفصلة. أربعة عمّال يعني أربع نسخ من النموذج في الذاكرة. لنموذج LightGBM بمئة ميغا هذا 400 ميغا. مقبول عادةً. لنماذج ضخمة (transformers بغيغابايتات)، نضغط عدد العمّال أو نستعمل خدمة توفير مخصّصة (TorchServe، Triton). نُوسّع هذا في الوحدة 10.

اختبار محلّيّ سريع

ملفّ اختبار خفيف يُثبت أنّ التحميل يعمل:

from fastapi.testclient import TestClient
from app import app  # importe l'application FastAPI

client = TestClient(app)


def test_info_modele():
    with client:  # declenche lifespan
        r = client.get("/model/info")
        assert r.status_code == 200
        assert r.json()["model_version"].startswith("churn-lgbm-")

with client: مهمّ: هو الذي يشغّل lifespan (startup ثمّ shutdown). بدونه، النموذج لا يُحمَّل والاختبار يفشل بأخطاء مربكة.

الخلاصة

• تحميل النموذج في كلّ طلب خطأ يُضاعف زمن الاستجابة عشرات المرّات ويُنهار الخدمة تحت حمل.
• lifespan هو الآليّة الرسميّة الحديثة لتحميل الموارد مرّة واحدة عند بدء التطبيق.
• الفشل عند البدء (fail fast) يُكشَف فورًا من مُنسّق الحاويات، بخلاف الفشل عند أوّل طلب الذي يراه المستخدم.
• كلّ عامل Uvicorn يحمل نسخة من النموذج في ذاكرته المستقلّة؛ هذا يُحدّد سقف عدد العمّال المعقول.