Passen Sie an, was in Model.fit passiert

Auf TensorFlow.org ansehen In Google Colab ausführen Quelle auf GitHub anzeigen Notizbuch herunterladen

Einführung

Wenn Sie überwachtes Lernen tun, können Sie fit() und alles funktioniert reibungslos.

Wenn Sie Ihre eigenen Trainingsschleife von Grund auf neu zu schreiben, können Sie die Verwendung GradientTape und die Kontrolle über jedes Detail nehmen.

Aber was , wenn Sie einen benutzerdefinierten Trainingsalgorithmus benötigen, aber Sie wollen immer noch profitieren von den praktischen Funktionen von fit() , wie Rückrufe, Einbau-Vertriebsunterstützung oder Schritt Verschmelzen?

Ein Kernprinzip der Keras ist progressiv Offenlegung von Komplexität. Sie sollten immer in der Lage sein, schrittweise in untergeordnete Workflows einzusteigen. Sie sollten nicht von einer Klippe fallen, wenn die High-Level-Funktionalität nicht genau Ihrem Anwendungsfall entspricht. Sie sollten in der Lage sein, mehr Kontrolle über die kleinen Details zu erlangen und gleichzeitig ein entsprechendes Maß an Komfort auf hohem Niveau zu erhalten.

Wenn Sie anpassen müssen , was fit() sind, sollten Sie den Trainingsschritt Funktion der außer Kraft setzen Model Dies ist die Funktion , die durch genannt wird fit() für jede Charge von Daten. Sie werden dann auf Aufruf der Lage sein , fit() wie gewohnt - und es wird Ihren eigenen Lernalgorithmus ausgeführt werden.

Beachten Sie, dass dieses Muster Sie nicht daran hindert, Modelle mit der funktionalen API zu erstellen. Sie können dies tun , ob Sie bauen Sequential Modelle, Functional API Modelle oder subclassed Modelle.

Mal sehen, wie das funktioniert.

Aufstellen

Erfordert TensorFlow 2.2 oder höher.

import tensorflow as tf
from tensorflow import keras

Ein erstes einfaches Beispiel

Beginnen wir mit einem einfachen Beispiel:

  • Wir schaffen eine neue Klasse , die Unterklassen keras.Model .
  • Wir überschreiben Sie einfach die Methode train_step(self, data) .
  • Wir geben ein Wörterbuch zurück, das Metriknamen (einschließlich des Verlusts) auf ihren aktuellen Wert zuordnet.

Das Eingabeargument data sind , was übergeben wird als Trainingsdaten zu passen:

  • Wenn Sie Numpy Arrays passieren, durch den Aufruf fit(x, y, ...) , dann data wird das Tupel sein (x, y)
  • Wenn Sie einen Pass tf.data.Dataset , durch den Aufruf fit(dataset, ...) - data dataset fit(dataset, ...) , dann data sein werden , was durch nachgegeben wird dataset bei jedem Charge.

Im Körper des train_step Methode implementieren wir ein regelmäßiges Training Update, ähnlich dem , was Sie bereits kennen. Wichtig ist , berechnen wir den Verlust über self.compiled_loss , die der Verlust (n) Funktion (en) umhüllt, die übergeben wurden compile() .

Ebenso rufen wir self.compiled_metrics.update_state(y, y_pred) den Zustand der Metriken zu aktualisieren , die in übergeben wurden compile() , und wir Abfrageergebnisse aus self.metrics am Ende ihren aktuellen Wert abzurufen.

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}

Probieren wir das aus:

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>

Untere Ebene gehen

Natürlich könnten Sie einfach überspringen eine Verlustfunktion beim Übergang der compile() , und stattdessen alles manuell in tun train_step . Ebenso für Metriken.

Hier ist eine niedrigere Ebene verwendet beispielsweise , dass nur die compile() das Optimierungsprogramm zu konfigurieren:

  • Wir beginnen mit der Erstellung von Metric Instanzen unseren Verlust und ein MAE - Score zu verfolgen.
  • Wir implementieren eine benutzerdefinierte train_step() , die den Zustand dieser Metriken aktualisiert (durch den Aufruf update_state() auf sie), dann fragen sie (via result() ) ihre aktuellen Mittelwert zurückzukehren, von der Fortschrittsbalken angezeigt werden und zu sein an jeden Rückruf übergeben.
  • Beachten Sie, dass wir nennen müssten reset_states() zwischen jeder Epoche auf unsere Metriken! Ansonsten ruft result() würde einen Durchschnitt seit Beginn der Ausbildung zurückkehren, während wir in der Regel der Arbeit mit pro-Epoche - Durchschnitt. Zum Glück kann der Rahmen , die für uns tun: nur Liste jeder Metrik , die Sie wollen in der zurücksetzen metrics Eigenschaft des Modells. Das Modell wird rufen reset_states() für jedes Objekt aufgeführt hier am Anfang jeder fit() Epoche oder zu Beginn eines Anrufs 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>

Unterstützung von sample_weight & class_weight

Sie haben vielleicht bemerkt, dass in unserem ersten einfachen Beispiel die Stichprobengewichtung nicht erwähnt wurde. Wenn Sie die unterstützen wollen fit() Argumente sample_weight und class_weight , würden Sie einfach folgendes tun:

  • Auspacken sample_weight aus dem data
  • Direkt compiled_loss & compiled_metrics (natürlich könnte man es auch nur manuell anwenden , wenn Sie verlassen Sie sich nicht auf der compile() für Verluste und Messwerte)
  • Das ist es. Das ist die Liste.
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>

Bereitstellung eines eigenen Bewertungsschritts

Was passiert , wenn Sie das gleiche für Anrufe tun möchten model.evaluate() ? Dann würden Sie außer Kraft setzen test_step auf genau die gleiche Art und Weise. So sieht es aus:

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]

Zum Abschluss: ein durchgängiges GAN-Beispiel

Lassen Sie uns ein End-to-End-Beispiel durchgehen, das alles nutzt, was Sie gerade gelernt haben.

Betrachten wir:

  • Ein Generatornetzwerk, das 28x28x1-Bilder erzeugen soll.
  • Ein Diskriminatornetzwerk, das 28x28x1-Bilder in zwei Klassen einteilen soll ("Fake" und "Real").
  • Ein Optimierer für jeden.
  • Eine Verlustfunktion zum Trainieren des Diskriminators.
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",
)

Hier ist eine Feature-complete GAN - Klasse und überschreibt der compile() seine eigene Signatur zu verwenden, und die Umsetzung des gesamten GAN - Algorithmus in 17 Zeilen in 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}

Lassen Sie es uns testen:

# 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>

Die Ideen hinter Deep Learning sind einfach, also warum sollte ihre Implementierung schmerzhaft sein?