الوحدة 8 — استدعاء النموذج من التطبيق

حتّى الآن، دالّة score_simule تعرض قواعد تعسّفيّة. الوقت الآن لربط النموذج الحقيقيّ. سنعرض طريقتَين متكاملتَين: تحميل النموذج محلّيًّا داخل عمليّة Streamlit، أو استدعاء API بعيدة (كما تعالجه الدورة 40). كلتاهما مشروعتان، ولكلٍّ منهما استعمال أنسب. وسنُغلّف الاستدعاء بمعالجة أخطاء ومهلات تجعل التطبيق يصمد أمام أعطال حقيقيّة.

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

النموذج المحلّيّ يُحمَّل من القرص إلى الذاكرة عند بدء التطبيق. بعده، كلّ تسجيل استدعاء دالّة .predict_proba بلا شبكة، بلا زمن استجابة، بلا تكلفة إضافيّة.

import streamlit as st
import joblib
import pandas as pd

@st.cache_resource
def charger_modele_local():
    modele = joblib.load("modele_churn.joblib")
    return modele

modele = charger_modele_local()

def scorer(df: pd.DataFrame) -> pd.Series:
    X = df[["age", "anciennete_mois", "mensuel_usd", "offre_code"]]
    probas = modele.predict_proba(X)[:, 1]
    return pd.Series(probas, index=df.index, name="score_risque")

كما رأينا في الوحدة 5، @st.cache_resource يحمّل النموذج مرّة واحدة لكلّ الجلسات. أوّل مستخدم يفتح التطبيق ينتظر ثانيتَين إلى خمس (حسب حجم النموذج)، ثمّ الجميع يستفيد.

متى تختار المحلّيّ: نموذج معتدل الحجم (تحت 500 ميغابايت)، تطبيق داخليّ بعدد مستخدمين معلوم (تحت مئة)، تحديث النموذج غير متواتر (مرّة كلّ أسبوع أو شهر). المميّزات: زمن استجابة قريب من الصفر، بلا تكلفة API، بلا نقطة فشل خارجيّة.

متى لا تختاره: النموذج ضخم (يزيد عن غيغابايت)، عدّة تطبيقات تستعمل نفس النموذج (تكرار للذاكرة)، تحديثات يوميّة (كلّ تحديث يستوجب إعادة نشر التطبيق).

API بعيدة: تصميم متيقّظ

الشكل الأنقى: نموذج مُنشَر على خدمة خاصّة (FastAPI في الدورة 40، أو منصّة مُدارة). التطبيق يرسل طلبًا HTTP ويستقبل الدرجة:

import requests

URL_API = "https://api.entreprise.com/v1/churn/score"

def scorer_via_api(client: dict) -> float:
    reponse = requests.post(
        URL_API,
        json=client,
        headers={"Authorization": f"Bearer {st.secrets['api_token']}"},
        timeout=5.0,
    )
    reponse.raise_for_status()
    return reponse.json()["score_risque"]

النقاط الحاسمة هنا: timeout=5.0 إلزاميّة. بلا مهلة، طلب HTTP قد ينتظر إلى الأبد إذا تعطّلت الشبكة، فيتجمّد التطبيق. مفتاح API في st.secrets، لا في الشيفرة (سنراه في الوحدة 10). raise_for_status() يرفع استثناء على أخطاء HTTP (4xx, 5xx).

معالجة الأخطاء الشاملة

استدعاء API قد يفشل بأربعة أشكال: مهلة (Timeout)، خطأ اتّصال (ConnectionError)، خطأ HTTP (4xx, 5xx)، خطأ في محتوى الجواب (JSON غير متوقّع). كلّ واحد يستحقّ معالجة صريحة:

import requests
import streamlit as st

def scorer_avec_erreurs(client: dict) -> float | None:
    try:
        reponse = requests.post(
            URL_API,
            json=client,
            headers={"Authorization": f"Bearer {st.secrets['api_token']}"},
            timeout=5.0,
        )
    except requests.Timeout:
        st.error("مهلة استدعاء النموذج (5 ثوانٍ). حاول لاحقًا.")
        return None
    except requests.ConnectionError:
        st.error("تعذّر الاتّصال بخدمة النموذج. تحقّق من الشبكة.")
        return None

    if reponse.status_code == 401:
        st.error("مفتاح API منتهي الصلاحيّة. تواصل مع المسؤول.")
        return None
    if reponse.status_code == 429:
        st.warning("تجاوز حدّ الاستدعاءات. حاول بعد دقيقة.")
        return None
    if not reponse.ok:
        st.error(f"خطأ خدمة النموذج: {reponse.status_code}")
        return None

    try:
        data = reponse.json()
        return float(data["score_risque"])
    except (ValueError, KeyError) as e:
        st.error(f"جواب غير متوقّع من النموذج: {e}")
        return None

كلّ فرع يُعطي رسالة للمستخدم يفهمها ويعرف ماذا يفعل. الفرق العمليّ في الإنتاج بين تطبيق يعرض 500 Internal Server Error وتطبيق يعرض «مهلة استدعاء النموذج، حاول لاحقًا» فرق كبير على ثقة الفريق.

إعادة المحاولة مع تراجع أسّيّ

للأخطاء العابرة (429, 503, timeout)، إعادة محاولة تلقائيّة تصلح كثيرًا من الحالات:

import time

def scorer_avec_reprise(client: dict, tentatives_max: int = 3) -> float | None:
    for tentative in range(tentatives_max):
        try:
            reponse = requests.post(URL_API, json=client, timeout=5.0)
            if reponse.status_code in (429, 503):
                temps_attente = 2 ** tentative
                time.sleep(temps_attente)
                continue
            reponse.raise_for_status()
            return float(reponse.json()["score_risque"])
        except (requests.Timeout, requests.ConnectionError):
            if tentative == tentatives_max - 1:
                st.error("فشل الاستدعاء بعد ثلاث محاولات.")
                return None
            time.sleep(2 ** tentative)
    return None

قاعدة: لا تُعيد المحاولة على أخطاء 4xx (باستثناء 429). 400 Bad Request يعني أنّ طلبك خاطئ؛ إعادة المحاولة لن تغيّر شيئًا. 500 Server Error يستحقّ محاولة أو اثنتَين. 429 Too Many Requests يستوجب انتظارًا.

مؤشّر التحميل

المستخدم يجب أن يرى أنّ شيئًا يحدث. st.spinner قبل الاستدعاء:

with st.spinner("جارٍ استدعاء النموذج..."):
    score = scorer_avec_reprise(client)

if score is None:
    st.stop()

st.metric("درجة الخطر", f"{score:.2f}")

للاستدعاءات الأطول (بضع ثوانٍ)، st.status أفضل لعرض خطوات:

with st.status("تسجيل العميل...", expanded=True) as status:
    st.write("تحقّق من الحقول...")
    valider(client)
    st.write("إرسال إلى النموذج...")
    score = scorer_avec_reprise(client)
    st.write(f"استُلمت درجة: {score:.2f}")
    status.update(label="اكتمل التسجيل", state="complete")

st.status يبدأ موسّعًا ويُطوى مع رسالة ناجحة. إذا حدث خطأ، status.update(state="error") يعرضه بلون أحمر.

تصميم قابل للاستبدال

الاختيار بين النموذج المحلّيّ وAPI ليس نهائيًّا. تصميم جيّد يسمح بالتنقّل بلا إعادة كتابة كلّ التطبيق:

from abc import ABC, abstractmethod

class Scoreur(ABC):
    @abstractmethod
    def scorer(self, df: pd.DataFrame) -> pd.Series:
        pass

class ScoreurLocal(Scoreur):
    def __init__(self, chemin_modele: str):
        self.modele = joblib.load(chemin_modele)
    def scorer(self, df: pd.DataFrame) -> pd.Series:
        probas = self.modele.predict_proba(df)[:, 1]
        return pd.Series(probas, index=df.index)

class ScoreurAPI(Scoreur):
    def __init__(self, url: str, token: str):
        self.url = url
        self.token = token
    def scorer(self, df: pd.DataFrame) -> pd.Series:
        payload = df.to_dict(orient="records")
        r = requests.post(self.url, json={"clients": payload},
                          headers={"Authorization": f"Bearer {self.token}"}, timeout=30)
        r.raise_for_status()
        return pd.Series(r.json()["scores"], index=df.index)

@st.cache_resource
def obtenir_scoreur() -> Scoreur:
    if st.secrets.get("api_token"):
        return ScoreurAPI(st.secrets["api_url"], st.secrets["api_token"])
    return ScoreurLocal("modele_churn.joblib")

هذا نمط قياسيّ: واجهة مجرَّدة، تحقيقان، اختيار من التهيئة. في التطوير: نموذج محلّيّ. في الإنتاج: API. لا تُعدَّل بقيّة الشيفرة.

اختبار API في التطوير

لا تختبر API الإنتاج من التطوير. إمّا اختبر بخدمة تجريبيّة، أو استعمل ScoreurLocal كنسخة احتياطيّة في التطوير. سبب: كلّ استدعاء يظهر في سجلّات ولوحات الإنتاج، وقد يخدع لوحات القياس أو يستنفد ميزانيّة الاستدعاءات. القاعدة: الفصل بين البيئات فرض، لا اختيار.

قياس زمن الاستجابة

في تطبيق الإنتاج، قياس زمن الاستدعاء يكشف تدهورًا مبكّرًا:

import time
import streamlit as st

def scorer_avec_metrique(client):
    debut = time.time()
    score = scorer_avec_reprise(client)
    duree = time.time() - debut
    st.session_state.setdefault("dernieres_durees", []).append(duree)
    if len(st.session_state.dernieres_durees) > 20:
        st.session_state.dernieres_durees.pop(0)
    return score

# في مكان ما من الصفحة
if "dernieres_durees" in st.session_state and st.session_state.dernieres_durees:
    moy = sum(st.session_state.dernieres_durees) / len(st.session_state.dernieres_durees)
    st.sidebar.metric("متوسّط زمن الاستدعاء (ث)", f"{moy:.2f}")

هذا مثال بسيط. في الإنتاج الفعليّ، الأرصاد تُرسل إلى منصّة قياس (Datadog, Prometheus)، لا session_state.

الخلاصة

• نموذج محلّيّ (@st.cache_resource + joblib): مناسب للنماذج المعتدلة والتطبيقات الداخليّة؛ زمن استجابة قريب من الصفر.
• API بعيدة (requests + timeout + معالجة أخطاء): مناسبة للنماذج الكبيرة والتحديثات المتواترة؛ تستوجب معالجة مهلات وأخطاء صريحة.
• معالجة الأخطاء بأربعة أشكال: Timeout، ConnectionError، أخطاء HTTP، JSON غير متوقّع. لكلّ رسالة واضحة للمستخدم.
• إعادة المحاولة مع تراجع أسّيّ للأخطاء العابرة (429, 503, timeout)، لا لأخطاء 4xx الأخرى.
• تصميم بواجهة مجرَّدة (Scoreur) يسمح بالتنقّل بين المحلّيّ وAPI بلا إعادة كتابة.
• st.spinner للانتظار القصير، st.status مع خطوات للانتظار الأطول.

الوحدة التالية: السمة والمظهر وسهولة الاستخدام لإلباس التطبيق ألوان الشركة.