Callbacks و Early Stopping — التعلم العميق

Callbacks والإيقاف المبكر

كم epoch تدرب؟ قليل ← Underfitting، كثير ← Overfitting ووقت ضائع. Callbacks دوال يستدعيها Keras في لحظات محددة (نهاية كل epoch مثلاً) لمراقبة التدريب والتحكم فيه.

EarlyStopping: توقف عندما يتوقف التحسن

from tensorflow import keras

early_stop = keras.callbacks.EarlyStopping(
    monitor="val_loss",          # راقب خسارة التحقق لا التدريب
    patience=5,                  # انتظر 5 epochs بلا تحسن قبل التوقف
    restore_best_weights=True,   # ارجع لأفضل أوزان، لا آخرها
)

مع EarlyStopping يمكنك كتابة epochs=100 بأمان: سيتوقف حين يلزم. ودون restore_best_weights ستحتفظ بأوزان آخر epoch (الأسوأ بـ patience خطوات).

ModelCheckpoint: احفظ الأفضل

from tensorflow import keras

checkpoint = keras.callbacks.ModelCheckpoint("best_model.keras", monitor="val_loss", save_best_only=True)

في Keras 3 يجب أن ينتهي المسار بـ .keras (أو .weights.h5 مع save_weights_only=True). يحميك أيضاً من انقطاع التدريب الطويل.

ReduceLROnPlateau

from tensorflow import keras

reduce_lr = keras.callbacks.ReduceLROnPlateau(monitor="val_loss", factor=0.5, patience=3, min_lr=1e-6)

عندما يتوقف التحسن، تقليل معدل التعلم غالباً يسمح بخطوات أدق نحو حل أفضل. اجعل patience هنا أقل من patience الإيقاف المبكر ليجرب التخفيض قبل التوقف.

TensorBoard

keras.callbacks.TensorBoard(log_dir="logs") يسجّل المنحنيات، وتشغّل tensorboard --logdir logs لترى لوحة تفاعلية تقارن التجارب.

Callback مخصص

from tensorflow import keras

class StopAtAccuracy(keras.callbacks.Callback):
    def __init__(self, target=0.98):
        super().__init__()
        self.target = target

    def on_epoch_end(self, epoch, logs=None):
        if logs and logs.get("val_accuracy", 0) >= self.target:
            print(f"وصلنا {self.target:.0%} في epoch {epoch + 1}")
            self.model.stop_training = True

نقاط الاستدعاء المتاحة: on_train_begin/end، on_epoch_begin/end، on_batch_begin/end.

الفكرة نفسها بلا Keras

best, patience, wait = float("inf"), 3, 0
val_losses = [0.9, 0.7, 0.62, 0.60, 0.61, 0.63, 0.59, 0.64, 0.66, 0.70]
for epoch, v in enumerate(val_losses, 1):
    if v < best:
        best, wait = v, 0
        print(f"epoch {epoch}: تحسن ← {v} (نحفظ النموذج)")
    else:
        wait += 1
        if wait >= patience:
            print(f"epoch {epoch}: لا تحسن منذ {patience} epochs ← توقف. الأفضل {best}")
            break

لاحظ: التحسن في epoch 7 أعاد العداد للصفر — لهذا نعطي patience أكبر من 1.

تمرين

في المثال الكامل غيّر patience إلى 1 ثم 10. متى توقف التدريب في كل حالة؟ وهل اختلفت دقة الاختبار؟

import tensorflow as tf
from tensorflow import keras

# ===== EarlyStopping =====
early_stop = keras.callbacks.EarlyStopping(
    monitor='val_loss',     # ما نراقبه
    patience=5,             # عدد epochs بدون تحسن
    restore_best_weights=True,  # استرجاع الأفضل
    verbose=1
)

# ===== ModelCheckpoint =====
checkpoint = keras.callbacks.ModelCheckpoint(
    'best_model.keras',     # مسار الحفظ
    monitor='val_accuracy',
    save_best_only=True,    # أفضل نموذج فقط
    verbose=1
)

# ===== ReduceLROnPlateau =====
reduce_lr = keras.callbacks.ReduceLROnPlateau(
    monitor='val_loss',
    factor=0.5,             # LR الجديد = القديم × 0.5
    patience=3,
    min_lr=1e-7,
    verbose=1
)

# ===== TensorBoard =====
tensorboard = keras.callbacks.TensorBoard(
    log_dir='./logs',
    histogram_freq=1
)

# ===== Custom Callback =====
class PrintProgress(keras.callbacks.Callback):
    def on_epoch_end(self, epoch, logs=None):
        if (epoch + 1) % 5 == 0:
            print(f"\n=== Epoch {epoch+1} ===")
            print(f"Loss: {logs['loss']:.4f}")
            print(f"Val Loss: {logs['val_loss']:.4f}")

# ===== استخدام Callbacks =====
(X_train, y_train), (X_test, y_test) = keras.datasets.mnist.load_data()
X_train = X_train.reshape(-1, 784) / 255.0
X_test = X_test.reshape(-1, 784) / 255.0

model = keras.Sequential([
    keras.layers.Dense(128, activation='relu', input_shape=(784,)),
    keras.layers.Dense(64, activation='relu'),
    keras.layers.Dense(10, activation='softmax')
])

model.compile(optimizer='adam', loss='sparse_categorical_crossentropy', metrics=['accuracy'])

history = model.fit(
    X_train, y_train,
    epochs=100,  # كثير عمداً
    validation_split=0.2,
    callbacks=[early_stop, checkpoint, reduce_lr, PrintProgress()],
    verbose=1
)

print(f"\nتوقف عند epoch: {len(history.history['loss'])}")