تخصيص ما يحدث في Model.fit

عرض على TensorFlow.org

تشغيل في Google Colab

عرض المصدر على جيثب

تحميل دفتر

مقدمة

عندما كنت تفعل التعلم تحت الإشراف، يمكنك استخدام fit() ويعمل كل شيء بسلاسة.

عندما تحتاج إلى كتابة حلقة التدريب الخاصة بك من الصفر، يمكنك استخدام GradientTape والسيطرة على كل التفاصيل الصغيرة.

ولكن ماذا لو كنت في حاجة الى خوارزمية التدريب المخصصة، لكنك لا تزال ترغب في الاستفادة من المزايا المريحة ل fit() ، مثل الاستدعاء، الذي بني في دعم التوزيع، أو خطوة الصمامات؟

والمبدأ الأساسي للKeras هو الكشف التدريجي من التعقيد. يجب أن تكون دائمًا قادرًا على الدخول في مهام سير العمل ذات المستوى الأدنى بطريقة تدريجية. يجب ألا تسقط من منحدر إذا كانت الوظيفة عالية المستوى لا تتطابق تمامًا مع حالة الاستخدام الخاصة بك. يجب أن تكون قادرًا على اكتساب المزيد من التحكم في التفاصيل الصغيرة مع الاحتفاظ بقدر متناسب من الراحة عالية المستوى.

عندما كنت في حاجة لتخصيص ما fit() لا ينبغي أن تتجاوز وظيفة التدريب خطوة من Model الطبقة. هذه هي وظيفة التي يتم استدعاؤها بواسطة fit() لكل دفعة من البيانات. سوف تكون بعد ذلك قادرة على دعوة fit() كالعادة - وسوف يتم تشغيل خوارزمية التعلم الخاصة بك.

لاحظ أن هذا النمط لا يمنعك من إنشاء نماذج باستخدام واجهة برمجة التطبيقات الوظيفية. يمكنك القيام بذلك سواء كنت بناء Sequential النماذج، نماذج API وظيفية، أو نماذج subclassed.

دعونا نرى كيف يعمل ذلك.

يثبت

يتطلب TensorFlow 2.2 أو أحدث.

import tensorflow as tf
from tensorflow import keras

أول مثال بسيط

لنبدأ بمثال بسيط:

• ونحن إنشاء فئة جديدة الفئات الفرعية keras.Model .
• نحن فقط تجاوز أسلوب train_step(self, data) .
• نعيد قاموسًا لتعيين أسماء المقاييس (بما في ذلك الخسارة) إلى قيمتها الحالية.

حجة إدخال data هو ما يتم تمريره لتناسب عن بيانات التدريب:

• إذا كنت تمر صفائف نمباي، من خلال الدعوة fit(x, y, ...) ، ثم data ستكون الصفوف (tuple) (x, y)
• إذا كنت تمر tf.data.Dataset ، من خلال الدعوة fit(dataset, ...) ، ثم data سوف يكون ما يحصل تجنى من dataset في كل دفعة.

في الجسم من train_step الطريقة، علينا أن ننفذ تحديث التدريب المنتظم، على غرار ما كنت بالفعل على دراية. والأهم من ذلك حساب فقدان عبر self.compiled_loss ، الذي يلتف ظيفة الخسارة (وفاق) (ق) التي تم تمريرها إلى compile() .

وبالمثل، فإننا ندعو self.compiled_metrics.update_state(y, y_pred) لتحديث الدولة من المقاييس التي صدرت في compile() ، ونحن الاستعلام نتائج self.metrics في نهاية لاسترداد قيمتها الحالية.

class CustomModel(keras.Model):
    def train_step(self, data):
        # Unpack the data. Its structure depends on your model and
        # on what you pass to `fit()`.
        x, y = data

        with tf.GradientTape() as tape:
            y_pred = self(x, training=True)  # Forward pass
            # Compute the loss value
            # (the loss function is configured in `compile()`)
            loss = self.compiled_loss(y, y_pred, regularization_losses=self.losses)

        # Compute gradients
        trainable_vars = self.trainable_variables
        gradients = tape.gradient(loss, trainable_vars)
        # Update weights
        self.optimizer.apply_gradients(zip(gradients, trainable_vars))
        # Update metrics (includes the metric that tracks the loss)
        self.compiled_metrics.update_state(y, y_pred)
        # Return a dict mapping metric names to current value
        return {m.name: m.result() for m in self.metrics}

لنجرب هذا:

import numpy as np

# Construct and compile an instance of CustomModel
inputs = keras.Input(shape=(32,))
outputs = keras.layers.Dense(1)(inputs)
model = CustomModel(inputs, outputs)
model.compile(optimizer="adam", loss="mse", metrics=["mae"])

# Just use `fit` as usual
x = np.random.random((1000, 32))
y = np.random.random((1000, 1))
model.fit(x, y, epochs=3)

Epoch 1/3
32/32 [==============================] - 1s 2ms/step - loss: 0.9909 - mae: 0.8601
Epoch 2/3
32/32 [==============================] - 0s 2ms/step - loss: 0.4363 - mae: 0.5345
Epoch 3/3
32/32 [==============================] - 0s 2ms/step - loss: 0.2906 - mae: 0.4311
<keras.callbacks.History at 0x7f5ad1ca1090>

الذهاب إلى المستوى الأدنى

وبطبيعة الحال، هل يمكن أن مجرد تخطي تمرير وظيفة الخسارة في compile() ، وبدلا من قيام كل شيء يدويا في train_step . وبالمثل بالنسبة للمقاييس.

وإليك مثال على مستوى أقل من ذلك، يستخدم فقط compile() لتكوين محسن:

• نبدأ من خلال خلق Metric الحالات لتتبع دينا الخسارة والنتيجة MAE.
• علينا أن ننفذ مخصص train_step() يقوم بتحديث حالة هذه المقاييس (عن طريق الاتصال update_state() عليهم)، ثم الاستعلام منهم (عبر result() ) للعودة متوسط قيمة الحالية، ليتم عرضه بواسطة شريط التقدم وأن يكون تمرير إلى أي رد.
• ملاحظة أننا بحاجة إلى استدعاء reset_states() على مقاييس لدينا بين كل عصر! داعيا إلا result() سيعود في المتوسط منذ بداية التدريب، في حين أننا عادة ما العمل مع المتوسطات لكل عصر. لحسن الحظ، لا يمكن للإطار أن نفعل ذلك بالنسبة لنا: مجرد إدراج أي متري تريد إعادة تعيين في metrics ملكا للنموذج. فإن نموذج الاتصال reset_states() على أي كائن المذكورة هنا في بداية كل fit() العصر أو في بداية دعوة لل evaluate() .

loss_tracker = keras.metrics.Mean(name="loss")
mae_metric = keras.metrics.MeanAbsoluteError(name="mae")


class CustomModel(keras.Model):
    def train_step(self, data):
        x, y = data

        with tf.GradientTape() as tape:
            y_pred = self(x, training=True)  # Forward pass
            # Compute our own loss
            loss = keras.losses.mean_squared_error(y, y_pred)

        # Compute gradients
        trainable_vars = self.trainable_variables
        gradients = tape.gradient(loss, trainable_vars)

        # Update weights
        self.optimizer.apply_gradients(zip(gradients, trainable_vars))

        # Compute our own metrics
        loss_tracker.update_state(loss)
        mae_metric.update_state(y, y_pred)
        return {"loss": loss_tracker.result(), "mae": mae_metric.result()}

    @property
    def metrics(self):
        # We list our `Metric` objects here so that `reset_states()` can be
        # called automatically at the start of each epoch
        # or at the start of `evaluate()`.
        # If you don't implement this property, you have to call
        # `reset_states()` yourself at the time of your choosing.
        return [loss_tracker, mae_metric]


# Construct an instance of CustomModel
inputs = keras.Input(shape=(32,))
outputs = keras.layers.Dense(1)(inputs)
model = CustomModel(inputs, outputs)

# We don't passs a loss or metrics here.
model.compile(optimizer="adam")

# Just use `fit` as usual -- you can use callbacks, etc.
x = np.random.random((1000, 32))
y = np.random.random((1000, 1))
model.fit(x, y, epochs=5)

Epoch 1/5
32/32 [==============================] - 0s 1ms/step - loss: 1.5969 - mae: 1.1523
Epoch 2/5
32/32 [==============================] - 0s 1ms/step - loss: 0.7352 - mae: 0.7310
Epoch 3/5
32/32 [==============================] - 0s 1ms/step - loss: 0.3830 - mae: 0.4999
Epoch 4/5
32/32 [==============================] - 0s 1ms/step - loss: 0.2809 - mae: 0.4215
Epoch 5/5
32/32 [==============================] - 0s 1ms/step - loss: 0.2590 - mae: 0.4058
<keras.callbacks.History at 0x7f5ad1b62c50>

دعم sample_weight و class_weight

ربما لاحظت أن أول مثال أساسي لدينا لم يذكر أي وزن للعينة. إذا كنت ترغب في دعم fit() الحجج sample_weight و class_weight ، وكنت ببساطة القيام بما يلي:

• فك sample_weight من data حجة
• تمريرها إلى compiled_loss و compiled_metrics (بالطبع، يمكنك أيضا أن مجرد تطبيقه يدويا إذا كنت لا تعتمد على compile() لخسائر والمقاييس)
• هذا هو. هذه القائمة.

class CustomModel(keras.Model):
    def train_step(self, data):
        # Unpack the data. Its structure depends on your model and
        # on what you pass to `fit()`.
        if len(data) == 3:
            x, y, sample_weight = data
        else:
            sample_weight = None
            x, y = data

        with tf.GradientTape() as tape:
            y_pred = self(x, training=True)  # Forward pass
            # Compute the loss value.
            # The loss function is configured in `compile()`.
            loss = self.compiled_loss(
                y,
                y_pred,
                sample_weight=sample_weight,
                regularization_losses=self.losses,
            )

        # Compute gradients
        trainable_vars = self.trainable_variables
        gradients = tape.gradient(loss, trainable_vars)

        # Update weights
        self.optimizer.apply_gradients(zip(gradients, trainable_vars))

        # Update the metrics.
        # Metrics are configured in `compile()`.
        self.compiled_metrics.update_state(y, y_pred, sample_weight=sample_weight)

        # Return a dict mapping metric names to current value.
        # Note that it will include the loss (tracked in self.metrics).
        return {m.name: m.result() for m in self.metrics}


# Construct and compile an instance of CustomModel
inputs = keras.Input(shape=(32,))
outputs = keras.layers.Dense(1)(inputs)
model = CustomModel(inputs, outputs)
model.compile(optimizer="adam", loss="mse", metrics=["mae"])

# You can now use sample_weight argument
x = np.random.random((1000, 32))
y = np.random.random((1000, 1))
sw = np.random.random((1000, 1))
model.fit(x, y, sample_weight=sw, epochs=3)

Epoch 1/3
32/32 [==============================] - 0s 2ms/step - loss: 0.1365 - mae: 0.4196
Epoch 2/3
32/32 [==============================] - 0s 2ms/step - loss: 0.1285 - mae: 0.4068
Epoch 3/3
32/32 [==============================] - 0s 2ms/step - loss: 0.1212 - mae: 0.3971
<keras.callbacks.History at 0x7f5ad1ba64d0>

تقديم خطوة التقييم الخاصة بك

ماذا لو كنت تريد أن تفعل الشيء نفسه بالنسبة إلى المكالمات model.evaluate() ؟ ثم هل تجاوز test_step في بنفس الطريقة تماما. هذا ما يبدو عليه:

class CustomModel(keras.Model):
    def test_step(self, data):
        # Unpack the data
        x, y = data
        # Compute predictions
        y_pred = self(x, training=False)
        # Updates the metrics tracking the loss
        self.compiled_loss(y, y_pred, regularization_losses=self.losses)
        # Update the metrics.
        self.compiled_metrics.update_state(y, y_pred)
        # Return a dict mapping metric names to current value.
        # Note that it will include the loss (tracked in self.metrics).
        return {m.name: m.result() for m in self.metrics}


# Construct an instance of CustomModel
inputs = keras.Input(shape=(32,))
outputs = keras.layers.Dense(1)(inputs)
model = CustomModel(inputs, outputs)
model.compile(loss="mse", metrics=["mae"])

# Evaluate with our custom test_step
x = np.random.random((1000, 32))
y = np.random.random((1000, 1))
model.evaluate(x, y)

32/32 [==============================] - 0s 1ms/step - loss: 2.7584 - mae: 1.5920
[2.758362054824829, 1.59201979637146]

الخلاصة: مثال على GAN من طرف إلى طرف

دعنا نتعرف على مثال شامل يستفيد من كل ما تعلمته للتو.

لنأخذ بعين الاعتبار:

• تهدف شبكة المولدات إلى إنشاء صور بحجم 28 × 28 × 1.
• شبكة أداة تمييز تهدف إلى تصنيف الصور مقاس 28 × 28 × 1 إلى فئتين ("وهمية" و "حقيقية").
• مُحسِّن واحد لكل منها.
• وظيفة الخسارة لتدريب المميّز.

from tensorflow.keras import layers

# Create the discriminator
discriminator = keras.Sequential(
    [
        keras.Input(shape=(28, 28, 1)),
        layers.Conv2D(64, (3, 3), strides=(2, 2), padding="same"),
        layers.LeakyReLU(alpha=0.2),
        layers.Conv2D(128, (3, 3), strides=(2, 2), padding="same"),
        layers.LeakyReLU(alpha=0.2),
        layers.GlobalMaxPooling2D(),
        layers.Dense(1),
    ],
    name="discriminator",
)

# Create the generator
latent_dim = 128
generator = keras.Sequential(
    [
        keras.Input(shape=(latent_dim,)),
        # We want to generate 128 coefficients to reshape into a 7x7x128 map
        layers.Dense(7 * 7 * 128),
        layers.LeakyReLU(alpha=0.2),
        layers.Reshape((7, 7, 128)),
        layers.Conv2DTranspose(128, (4, 4), strides=(2, 2), padding="same"),
        layers.LeakyReLU(alpha=0.2),
        layers.Conv2DTranspose(128, (4, 4), strides=(2, 2), padding="same"),
        layers.LeakyReLU(alpha=0.2),
        layers.Conv2D(1, (7, 7), padding="same", activation="sigmoid"),
    ],
    name="generator",
)

وهنا فئة GAN ميزة كاملة، تجاوز compile() لاستخدام التوقيع الخاص به، وتنفيذ خوارزمية GAN بأكملها في 17 خطوط في train_step :

class GAN(keras.Model):
    def __init__(self, discriminator, generator, latent_dim):
        super(GAN, self).__init__()
        self.discriminator = discriminator
        self.generator = generator
        self.latent_dim = latent_dim

    def compile(self, d_optimizer, g_optimizer, loss_fn):
        super(GAN, self).compile()
        self.d_optimizer = d_optimizer
        self.g_optimizer = g_optimizer
        self.loss_fn = loss_fn

    def train_step(self, real_images):
        if isinstance(real_images, tuple):
            real_images = real_images[0]
        # Sample random points in the latent space
        batch_size = tf.shape(real_images)[0]
        random_latent_vectors = tf.random.normal(shape=(batch_size, self.latent_dim))

        # Decode them to fake images
        generated_images = self.generator(random_latent_vectors)

        # Combine them with real images
        combined_images = tf.concat([generated_images, real_images], axis=0)

        # Assemble labels discriminating real from fake images
        labels = tf.concat(
            [tf.ones((batch_size, 1)), tf.zeros((batch_size, 1))], axis=0
        )
        # Add random noise to the labels - important trick!
        labels += 0.05 * tf.random.uniform(tf.shape(labels))

        # Train the discriminator
        with tf.GradientTape() as tape:
            predictions = self.discriminator(combined_images)
            d_loss = self.loss_fn(labels, predictions)
        grads = tape.gradient(d_loss, self.discriminator.trainable_weights)
        self.d_optimizer.apply_gradients(
            zip(grads, self.discriminator.trainable_weights)
        )

        # Sample random points in the latent space
        random_latent_vectors = tf.random.normal(shape=(batch_size, self.latent_dim))

        # Assemble labels that say "all real images"
        misleading_labels = tf.zeros((batch_size, 1))

        # Train the generator (note that we should *not* update the weights
        # of the discriminator)!
        with tf.GradientTape() as tape:
            predictions = self.discriminator(self.generator(random_latent_vectors))
            g_loss = self.loss_fn(misleading_labels, predictions)
        grads = tape.gradient(g_loss, self.generator.trainable_weights)
        self.g_optimizer.apply_gradients(zip(grads, self.generator.trainable_weights))
        return {"d_loss": d_loss, "g_loss": g_loss}

دعونا نختبرها:

# Prepare the dataset. We use both the training & test MNIST digits.
batch_size = 64
(x_train, _), (x_test, _) = keras.datasets.mnist.load_data()
all_digits = np.concatenate([x_train, x_test])
all_digits = all_digits.astype("float32") / 255.0
all_digits = np.reshape(all_digits, (-1, 28, 28, 1))
dataset = tf.data.Dataset.from_tensor_slices(all_digits)
dataset = dataset.shuffle(buffer_size=1024).batch(batch_size)

gan = GAN(discriminator=discriminator, generator=generator, latent_dim=latent_dim)
gan.compile(
    d_optimizer=keras.optimizers.Adam(learning_rate=0.0003),
    g_optimizer=keras.optimizers.Adam(learning_rate=0.0003),
    loss_fn=keras.losses.BinaryCrossentropy(from_logits=True),
)

# To limit the execution time, we only train on 100 batches. You can train on
# the entire dataset. You will need about 20 epochs to get nice results.
gan.fit(dataset.take(100), epochs=1)

100/100 [==============================] - 3s 11ms/step - d_loss: 0.4031 - g_loss: 0.9305
<keras.callbacks.History at 0x7f5ad1b37c50>

إن الأفكار الكامنة وراء التعلم العميق بسيطة ، فلماذا يجب أن يكون تنفيذها مؤلمًا؟