إطلاق النموذج

النموذج الحبيس في notebook لا يساعد أحدًا. وكي يغيّر ما يفعله Cartwheel، يجب أن يعمل الـ pipeline البطل على بيانات عملاء جديدة، وفق جدول زمني أو عند الطلب، وأن تصل درجاته إلى الأشخاص الذين يرسلون عروض الاسترجاع. يغطي هذا الدرس الميل الأخير: حفظ النموذج، وحراسة مدخلاته، والاختيار بين مهمة دفعية وAPI، وإثبات أنه يعمل.

احفظ الـ pipeline، لا النموذج وحده

الشيء الذي يُطلق هو الـ pipeline المدرَّب كاملًا. فمتوسطات الـ scaler وانحرافاته المعيارية جزء من النموذج بقدر المعاملات؛ ودونها ستوحّد بيئة الإنتاج مقاييس العملاء الجدد بشكل مختلف، وستكون كل درجة منحرفة.

يغطي دليل حفظ النماذج في scikit-learn عدة صيغ (pickle وjoblib وskops وONNX)؛ وjoblib هو الأكثر شيوعًا حين تتولّى Python الحفظ والتحميل معًا. احفظ حزمة صغيرة فيها ما سيحتاجه أحدهم لاحقًا: الـ pipeline، وقائمة الـ features بالضبط، ووسم الإصدار، وإصدار المكتبة الذي دُرِّب به.

import os
import tempfile

import joblib
import pandas as pd
import sklearn
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

churn = pd.read_csv("churn.csv")
features = ["tenure_days", "orders", "total_spent", "avg_order_value",
            "days_since_last_order", "used_coupon", "support_tickets", "mobile_share"]
champion = make_pipeline(StandardScaler(), LogisticRegression()).fit(churn[features], churn["churned"])

bundle = {"model": champion, "features": features,
          "version": "churn-2025-09-30.1", "sklearn_version": sklearn.__version__}

with tempfile.TemporaryDirectory() as folder:
    path = os.path.join(folder, "churn_model.joblib")
    joblib.dump(bundle, path)
    print("saved", os.path.getsize(path), "bytes")

    loaded = joblib.load(path)
    same = (loaded["model"].predict_proba(churn[features]) == champion.predict_proba(churn[features])).all()
    print("identical predictions after reload:", same, "| trained with scikit-learn", loaded["sklearn_version"])

لاحظ أن النموذج النهائي مدرَّب على العملاء الـ 398 ذوي الـ labels كلهم. فمهمة بيانات الاختبار كانت قياس عملية النمذجة؛ وبعد أن قستها، تعطي إعادة التدريب على كل صف ذي label النموذجَ المُطلَق شيئًا أكثر يتعلّم منه. سجّل أنك فعلت ذلك، واحتفظ بالمقياس المنشور من تقييم بيانات الاختبار.

احرس المدخلات

سيقيّم النموذج الهراء بكل سرور. عميل لديه orders = 0، أو mobile_share يساوي 7، أو عمود مفقود، ينتج احتمالًا يبدو واثقًا كأي احتمال آخر. ضع طبقة رقيقة أمام predict_proba تتحقق من المُدخل وتفشل بوضوح:

# assumes `import pandas as pd` and a loaded `bundle` as above
def predict_churn(record: dict, bundle: dict) -> float:
    missing = [f for f in bundle["features"] if f not in record]
    if missing:
        raise ValueError(f"missing fields: {missing}")
    if not 0 <= record["mobile_share"] <= 1:
        raise ValueError("mobile_share must be between 0 and 1")
    row = pd.DataFrame([record])[bundle["features"]]
    return float(bundle["model"].predict_proba(row)[0, 1])

اختيار [bundle["features"]] يثبّت ترتيب الأعمدة، وهو ما يصرّ عليه scikit-learn كما بيّن القسم 1.

مهمة دفعية أم API؟

هناك طريقتان شائعتان لتشغيل النموذج:

• التقييم الدفعي (batch scoring): مهمة مجدولة (ليلية أو أسبوعية) تحمّل النموذج، وتقيّم كل عميل نشط من مستودع البيانات، وتكتب الاحتمالات في جدول. وتقرأ أدوات التسويق الجدول. بسيطة، ورخيصة، وسهلة المراقبة، ومناسبة للانقطاع، حيث لا يكلّف القرار المتأخر يومًا شيئًا.
• الـ API المباشر (online): خدمة ويب تقيّم عميلًا واحدًا لكل طلب، في أجزاء من الثانية. وتلزم حين يحدث القرار أثناء جلسة المستخدم، مثل إظهار عرض استبقاء في صفحة الإلغاء.

وإذا احتجت فعلًا إلى API، فخدمة FastAPI صغيرة خيار شائع في Python. هذا المخطّط لا يعمل في المتصفّح، لكنه مكتمل بما يكفي لتشغيله محليًا بـ uvicorn app:app:

import joblib
import pandas as pd
from fastapi import FastAPI
from pydantic import BaseModel, Field

# joblib uses pickle: load only model files from your own trusted, access-controlled store
bundle = joblib.load("churn_model.joblib")
app = FastAPI()

class Customer(BaseModel):
    tenure_days: int = Field(ge=0)
    orders: int = Field(ge=1)
    total_spent: float = Field(ge=0)
    avg_order_value: float = Field(ge=0)
    days_since_last_order: int = Field(ge=0)
    used_coupon: int = Field(ge=0, le=1)
    support_tickets: int = Field(ge=0)
    mobile_share: float = Field(ge=0, le=1)

@app.post("/predict")
def predict(customer: Customer):
    row = pd.DataFrame([customer.model_dump()])[bundle["features"]]
    prob = float(bundle["model"].predict_proba(row)[0, 1])
    return {"churn_probability": round(prob, 3), "model_version": bundle["version"]}

يتولّى نموذج Pydantic التحقق من المدخلات بشكل تصريحي: الطلب الذي فيه orders: 0 أو حقل مفقود يحصل على خطأ 422 قبل أن يراه النموذج أصلًا. وإعادة إصدار النموذج مع كل إجابة تتيح تتبّع أي تنبؤ إلى النموذج المحدّد الذي أنتجه. سجّل كل طلب واستجابة؛ فهذه السجلّات هي ما تعمل عليه فحوص الانجراف من الدرس السابق.

هل نجح؟ أجرِ تجربة

الـ ROC AUC الجيد يعني أن النموذج يرتّب العملاء جيدًا. ولا يعني أن عرض الاسترجاع يغيّر رأي أحد. ولقياس ذلك، احجز شريحة عشوائية من العملاء المُعلَّمين (لنقل 20%) لا يتلقّون العرض. وبعد 90 يومًا، قارن الانقطاع بين المجموعة التي تواصلت معها والمجموعة المحجوزة. الفرق هو أثر العرض على العملاء الذين يختارهم النموذج، ومضروبًا في قيمة العميل يصبح الرقم الذي تهتم به الشركة فعلًا. ودون المجموعة المحجوزة، قد يكون أي انخفاض في الانقطاع بسبب الموسم، أو تعثّر منافس، أو الحظ.

أين أنت الآن

بدأت بسؤال وجدول، وصار لديك الآن إجراء يصمد: صُغ المسألة، وقسّم قبل أن تنظر، وتفوّق على نموذج وهمي للفئة الأكثر شيوعًا وعلى قاعدة، واختر المقاييس بحسب كلفة كل خطأ، وقارن النماذج بالتحقق المتقاطع داخل pipelines خالية من التسرّب، وتحقّق ممن يخدمهم النموذج بشكل أسوأ، وأطلق شيئًا تستطيع مراقبته واختباره. وتبيّن أن البطل هو أبسط نموذج جادّ، وأنت تعرف بالضبط لماذا. عادة السؤال “مقارنةً بماذا؟” هي أثمن ما كان لدى هذه الدورة لتعلّمه، وهي تنطبق دون تغيير على النموذج القادم الذي ستبنيه، أيًّا كان.