Bantuan melindungi Great Barrier Reef dengan TensorFlow pada Kaggle Bergabung Tantangan

Membuat Layer dan Model baru melalui subclassing

Lihat di TensorFlow.org Jalankan di Google Colab Lihat sumber di GitHub Unduh buku catatan

Mempersiapkan

import tensorflow as tf
from tensorflow import keras

The Layer kelas: kombinasi negara (bobot) dan beberapa perhitungan

Salah satu abstraksi sentral dalam Keras adalah Layer kelas. Sebuah lapisan merangkum baik keadaan ("bobot") lapisan dan transformasi dari input ke output ("panggilan", penerusan lapisan).

Berikut adalah lapisan yang terhubung secara padat. Memiliki negara: variabel w dan b .

class Linear(keras.layers.Layer):
    def __init__(self, units=32, input_dim=32):
        super(Linear, self).__init__()
        w_init = tf.random_normal_initializer()
        self.w = tf.Variable(
            initial_value=w_init(shape=(input_dim, units), dtype="float32"),
            trainable=True,
        )
        b_init = tf.zeros_initializer()
        self.b = tf.Variable(
            initial_value=b_init(shape=(units,), dtype="float32"), trainable=True
        )

    def call(self, inputs):
        return tf.matmul(inputs, self.w) + self.b

Anda akan menggunakan layer dengan memanggilnya pada beberapa input tensor, seperti fungsi Python.

x = tf.ones((2, 2))
linear_layer = Linear(4, 2)
y = linear_layer(x)
print(y)
tf.Tensor(
[[ 0.00962844 -0.01307489 -0.1452128   0.0538918 ]
 [ 0.00962844 -0.01307489 -0.1452128   0.0538918 ]], shape=(2, 4), dtype=float32)

Perhatikan bahwa bobot w dan b secara otomatis dilacak oleh lapisan atas yang ditetapkan sebagai atribut lapisan:

assert linear_layer.weights == [linear_layer.w, linear_layer.b]

Catatan Anda juga memiliki akses ke pintas cepat untuk menambah berat badan untuk lapisan: yang add_weight() metode:

class Linear(keras.layers.Layer):
    def __init__(self, units=32, input_dim=32):
        super(Linear, self).__init__()
        self.w = self.add_weight(
            shape=(input_dim, units), initializer="random_normal", trainable=True
        )
        self.b = self.add_weight(shape=(units,), initializer="zeros", trainable=True)

    def call(self, inputs):
        return tf.matmul(inputs, self.w) + self.b


x = tf.ones((2, 2))
linear_layer = Linear(4, 2)
y = linear_layer(x)
print(y)
tf.Tensor(
[[ 0.05790994  0.060931   -0.0402256  -0.09450993]
 [ 0.05790994  0.060931   -0.0402256  -0.09450993]], shape=(2, 4), dtype=float32)

Lapisan dapat memiliki bobot yang tidak dapat dilatih

Selain beban yang dapat dilatih, Anda juga dapat menambahkan beban yang tidak dapat dilatih ke lapisan. Bobot tersebut dimaksudkan untuk tidak diperhitungkan selama backpropagation, saat Anda melatih layer.

Berikut ini cara menambahkan dan menggunakan beban yang tidak dapat dilatih:

class ComputeSum(keras.layers.Layer):
    def __init__(self, input_dim):
        super(ComputeSum, self).__init__()
        self.total = tf.Variable(initial_value=tf.zeros((input_dim,)), trainable=False)

    def call(self, inputs):
        self.total.assign_add(tf.reduce_sum(inputs, axis=0))
        return self.total


x = tf.ones((2, 2))
my_sum = ComputeSum(2)
y = my_sum(x)
print(y.numpy())
y = my_sum(x)
print(y.numpy())
[2. 2.]
[4. 4.]

Itu bagian dari layer.weights , tapi itu akan dikategorikan sebagai badan non-dilatih:

print("weights:", len(my_sum.weights))
print("non-trainable weights:", len(my_sum.non_trainable_weights))

# It's not included in the trainable weights:
print("trainable_weights:", my_sum.trainable_weights)
weights: 1
non-trainable weights: 1
trainable_weights: []

Praktik terbaik: menunda pembuatan bobot hingga bentuk input diketahui

Kami Linear lapisan atas mengambil input_dim argumen yang digunakan untuk menghitung bentuk bobot w dan b di __init__() :

class Linear(keras.layers.Layer):
    def __init__(self, units=32, input_dim=32):
        super(Linear, self).__init__()
        self.w = self.add_weight(
            shape=(input_dim, units), initializer="random_normal", trainable=True
        )
        self.b = self.add_weight(shape=(units,), initializer="zeros", trainable=True)

    def call(self, inputs):
        return tf.matmul(inputs, self.w) + self.b

Dalam banyak kasus, Anda mungkin tidak mengetahui sebelumnya ukuran input Anda, dan Anda ingin dengan malas membuat bobot saat nilai tersebut diketahui, beberapa saat setelah membuat instance layer.

Dalam API Keras, kami sarankan menciptakan lapisan bobot dalam build(self, inputs_shape) metode lapisan Anda. Seperti ini:

class Linear(keras.layers.Layer):
    def __init__(self, units=32):
        super(Linear, self).__init__()
        self.units = units

    def build(self, input_shape):
        self.w = self.add_weight(
            shape=(input_shape[-1], self.units),
            initializer="random_normal",
            trainable=True,
        )
        self.b = self.add_weight(
            shape=(self.units,), initializer="random_normal", trainable=True
        )

    def call(self, inputs):
        return tf.matmul(inputs, self.w) + self.b

The __call__() metode lapisan Anda secara otomatis akan menjalankan membangun pertama kali disebut. Anda sekarang memiliki lapisan yang malas dan dengan demikian lebih mudah digunakan:

# At instantiation, we don't know on what inputs this is going to get called
linear_layer = Linear(32)

# The layer's weights are created dynamically the first time the layer is called
y = linear_layer(x)

Menerapkan build() secara terpisah seperti yang ditunjukkan di atas baik memisahkan menciptakan bobot hanya sekali menggunakan bobot di setiap panggilan. Namun, untuk beberapa lapisan kustom tingkat lanjut, memisahkan pembuatan dan penghitungan status menjadi tidak praktis. Pelaksana lapisan diperbolehkan untuk menunda pembuatan berat untuk pertama __call__() , tapi perlu berhati-hati bahwa panggilan kemudian menggunakan bobot yang sama. Selain itu, karena __call__() kemungkinan akan dieksekusi untuk pertama kalinya di dalam tf.function , setiap penciptaan variabel yang berlangsung di __call__() harus dibungkus dalam tf.init_scope .

Lapisan dapat disusun secara rekursif

Jika Anda menetapkan instance Layer sebagai atribut Layer lain, lapisan luar akan mulai melacak bobot yang dibuat oleh lapisan dalam.

Sebaiknya buat sublayer seperti di __init__() metode dan meninggalkannya untuk pertama __call__() untuk memicu membangun bobot mereka.

class MLPBlock(keras.layers.Layer):
    def __init__(self):
        super(MLPBlock, self).__init__()
        self.linear_1 = Linear(32)
        self.linear_2 = Linear(32)
        self.linear_3 = Linear(1)

    def call(self, inputs):
        x = self.linear_1(inputs)
        x = tf.nn.relu(x)
        x = self.linear_2(x)
        x = tf.nn.relu(x)
        return self.linear_3(x)


mlp = MLPBlock()
y = mlp(tf.ones(shape=(3, 64)))  # The first call to the `mlp` will create the weights
print("weights:", len(mlp.weights))
print("trainable weights:", len(mlp.trainable_weights))
weights: 6
trainable weights: 6

The add_loss() metode

Ketika menulis call() metode lapisan, Anda dapat membuat tensor kerugian yang Anda akan ingin menggunakan kemudian, ketika menulis lingkaran pelatihan Anda. Ini bisa dilakukan dengan memanggil self.add_loss(value) :

# A layer that creates an activity regularization loss
class ActivityRegularizationLayer(keras.layers.Layer):
    def __init__(self, rate=1e-2):
        super(ActivityRegularizationLayer, self).__init__()
        self.rate = rate

    def call(self, inputs):
        self.add_loss(self.rate * tf.reduce_sum(inputs))
        return inputs

Kerugian ini (termasuk yang dibuat oleh lapisan dalam) dapat diambil melalui layer.losses . Properti ini adalah ulang pada awal setiap __call__() ke lapisan tingkat atas, sehingga layer.losses selalu mengandung nilai-nilai kerugian yang diciptakan selama maju lulus terakhir.

class OuterLayer(keras.layers.Layer):
    def __init__(self):
        super(OuterLayer, self).__init__()
        self.activity_reg = ActivityRegularizationLayer(1e-2)

    def call(self, inputs):
        return self.activity_reg(inputs)


layer = OuterLayer()
assert len(layer.losses) == 0  # No losses yet since the layer has never been called

_ = layer(tf.zeros(1, 1))
assert len(layer.losses) == 1  # We created one loss value

# `layer.losses` gets reset at the start of each __call__
_ = layer(tf.zeros(1, 1))
assert len(layer.losses) == 1  # This is the loss created during the call above

Selain itu, loss properti juga mengandung kerugian regularisasi dibuat untuk bobot dari setiap lapisan dalam:

class OuterLayerWithKernelRegularizer(keras.layers.Layer):
    def __init__(self):
        super(OuterLayerWithKernelRegularizer, self).__init__()
        self.dense = keras.layers.Dense(
            32, kernel_regularizer=tf.keras.regularizers.l2(1e-3)
        )

    def call(self, inputs):
        return self.dense(inputs)


layer = OuterLayerWithKernelRegularizer()
_ = layer(tf.zeros((1, 1)))

# This is `1e-3 * sum(layer.dense.kernel ** 2)`,
# created by the `kernel_regularizer` above.
print(layer.losses)
[<tf.Tensor: shape=(), dtype=float32, numpy=0.0024520475>]

Kerugian ini dimaksudkan untuk diperhitungkan saat menulis loop pelatihan, seperti ini:

# Instantiate an optimizer.
optimizer = tf.keras.optimizers.SGD(learning_rate=1e-3)
loss_fn = keras.losses.SparseCategoricalCrossentropy(from_logits=True)

# Iterate over the batches of a dataset.
for x_batch_train, y_batch_train in train_dataset:
  with tf.GradientTape() as tape:
    logits = layer(x_batch_train)  # Logits for this minibatch
    # Loss value for this minibatch
    loss_value = loss_fn(y_batch_train, logits)
    # Add extra losses created during this forward pass:
    loss_value += sum(model.losses)

  grads = tape.gradient(loss_value, model.trainable_weights)
  optimizer.apply_gradients(zip(grads, model.trainable_weights))

Untuk panduan rinci tentang menulis loop pelatihan, lihat panduan untuk menulis sebuah loop pelatihan dari awal .

Kerugian ini juga bekerja secara lancar dengan fit() (mereka bisa secara otomatis dijumlahkan dan ditambahkan ke kerugian utama, jika ada):

import numpy as np

inputs = keras.Input(shape=(3,))
outputs = ActivityRegularizationLayer()(inputs)
model = keras.Model(inputs, outputs)

# If there is a loss passed in `compile`, the regularization
# losses get added to it
model.compile(optimizer="adam", loss="mse")
model.fit(np.random.random((2, 3)), np.random.random((2, 3)))

# It's also possible not to pass any loss in `compile`,
# since the model already has a loss to minimize, via the `add_loss`
# call during the forward pass!
model.compile(optimizer="adam")
model.fit(np.random.random((2, 3)), np.random.random((2, 3)))
1/1 [==============================] - 0s 209ms/step - loss: 0.1948
1/1 [==============================] - 0s 49ms/step - loss: 0.0298
<keras.callbacks.History at 0x7fce9052d290>

The add_metric() metode

Demikian pula untuk add_loss() , lapisan juga memiliki add_metric() metode untuk melacak rata-rata bergerak kuantitas selama pelatihan.

Pertimbangkan lapisan berikut: lapisan "titik akhir logistik". Dibutuhkan sebagai masukan prediksi & target, itu menghitung kerugian yang melacak melalui add_loss() , dan menghitung sebuah skalar akurasi, yang melacak melalui add_metric() .

class LogisticEndpoint(keras.layers.Layer):
    def __init__(self, name=None):
        super(LogisticEndpoint, self).__init__(name=name)
        self.loss_fn = keras.losses.BinaryCrossentropy(from_logits=True)
        self.accuracy_fn = keras.metrics.BinaryAccuracy()

    def call(self, targets, logits, sample_weights=None):
        # Compute the training-time loss value and add it
        # to the layer using `self.add_loss()`.
        loss = self.loss_fn(targets, logits, sample_weights)
        self.add_loss(loss)

        # Log accuracy as a metric and add it
        # to the layer using `self.add_metric()`.
        acc = self.accuracy_fn(targets, logits, sample_weights)
        self.add_metric(acc, name="accuracy")

        # Return the inference-time prediction tensor (for `.predict()`).
        return tf.nn.softmax(logits)

Metrik dilacak dengan cara ini dapat diakses melalui layer.metrics :

layer = LogisticEndpoint()

targets = tf.ones((2, 2))
logits = tf.ones((2, 2))
y = layer(targets, logits)

print("layer.metrics:", layer.metrics)
print("current accuracy value:", float(layer.metrics[0].result()))
layer.metrics: [<keras.metrics.BinaryAccuracy object at 0x7fce90578490>]
current accuracy value: 1.0

Sama seperti untuk add_loss() , metrik ini dilacak oleh fit() :

inputs = keras.Input(shape=(3,), name="inputs")
targets = keras.Input(shape=(10,), name="targets")
logits = keras.layers.Dense(10)(inputs)
predictions = LogisticEndpoint(name="predictions")(logits, targets)

model = keras.Model(inputs=[inputs, targets], outputs=predictions)
model.compile(optimizer="adam")

data = {
    "inputs": np.random.random((3, 3)),
    "targets": np.random.random((3, 10)),
}
model.fit(data)
1/1 [==============================] - 0s 274ms/step - loss: 0.9291 - binary_accuracy: 0.0000e+00
<keras.callbacks.History at 0x7fce90448c50>

Anda dapat secara opsional mengaktifkan serialisasi pada lapisan Anda

Jika Anda membutuhkan lapisan kustom Anda untuk menjadi Serializable sebagai bagian dari model yang Fungsional , anda dapat menerapkan get_config() metode:

class Linear(keras.layers.Layer):
    def __init__(self, units=32):
        super(Linear, self).__init__()
        self.units = units

    def build(self, input_shape):
        self.w = self.add_weight(
            shape=(input_shape[-1], self.units),
            initializer="random_normal",
            trainable=True,
        )
        self.b = self.add_weight(
            shape=(self.units,), initializer="random_normal", trainable=True
        )

    def call(self, inputs):
        return tf.matmul(inputs, self.w) + self.b

    def get_config(self):
        return {"units": self.units}


# Now you can recreate the layer from its config:
layer = Linear(64)
config = layer.get_config()
print(config)
new_layer = Linear.from_config(config)
{'units': 64}

Perhatikan bahwa __init__() metode dasar Layer kelas mengambil beberapa argumen kata kunci, dalam khususnya name dan dtype . Ini praktik yang baik untuk melewati argumen ini untuk kelas induk di __init__() dan untuk memasukkan mereka dalam konfigurasi lapisan:

class Linear(keras.layers.Layer):
    def __init__(self, units=32, **kwargs):
        super(Linear, self).__init__(**kwargs)
        self.units = units

    def build(self, input_shape):
        self.w = self.add_weight(
            shape=(input_shape[-1], self.units),
            initializer="random_normal",
            trainable=True,
        )
        self.b = self.add_weight(
            shape=(self.units,), initializer="random_normal", trainable=True
        )

    def call(self, inputs):
        return tf.matmul(inputs, self.w) + self.b

    def get_config(self):
        config = super(Linear, self).get_config()
        config.update({"units": self.units})
        return config


layer = Linear(64)
config = layer.get_config()
print(config)
new_layer = Linear.from_config(config)
{'name': 'linear_8', 'trainable': True, 'dtype': 'float32', 'units': 64}

Jika Anda membutuhkan lebih banyak fleksibilitas ketika deserializing lapisan dari config, Anda juga dapat menimpa from_config() metode kelas. Ini adalah implementasi dasar from_config() :

def from_config(cls, config):
  return cls(**config)

Untuk mempelajari lebih lanjut tentang serialisasi dan menyimpan, melihat lengkap panduan untuk menyelamatkan dan serialisasi model .

Keistimewaan training argumen dalam call() metode

Beberapa lapisan, khususnya BatchNormalization lapisan dan Dropout lapisan, memiliki perilaku yang berbeda selama pelatihan dan inferensi. Untuk lapisan seperti itu, praktek standar untuk mengekspos training (boolean) argumen dalam call() metode.

Dengan mengekspos argumen ini di call() , Anda mengaktifkan built-in loop pelatihan dan evaluasi (misalnya fit() ) yang benar menggunakan lapisan dalam pelatihan dan inferensi.

class CustomDropout(keras.layers.Layer):
    def __init__(self, rate, **kwargs):
        super(CustomDropout, self).__init__(**kwargs)
        self.rate = rate

    def call(self, inputs, training=None):
        if training:
            return tf.nn.dropout(inputs, rate=self.rate)
        return inputs

Keistimewaan mask argumen dalam call() metode

Argumen istimewa lain yang didukung oleh call() adalah mask argumen.

Anda akan menemukannya di semua lapisan Keras RNN. Mask adalah tensor boolean (satu nilai boolean per langkah waktu dalam input) yang digunakan untuk melewati langkah waktu input tertentu saat memproses data deret waktu.

Keras otomatis akan melewati benar mask argumen untuk __call__() untuk lapisan yang mendukungnya, ketika topeng yang dihasilkan oleh lapisan sebelumnya. Topeng-menghasilkan lapisan adalah Embedding lapisan dikonfigurasi dengan mask_zero=True , dan Masking lapisan.

Untuk mempelajari lebih lanjut tentang masking dan bagaimana menulis masking-enabled lapisan, silahkan periksa panduan "memahami padding dan masking" .

The Model kelas

Secara umum, Anda akan menggunakan Layer kelas untuk mendefinisikan blok perhitungan batin, dan akan menggunakan Model kelas untuk menentukan model luar - objek yang akan melatih.

Misalnya, dalam model ResNet50, Anda akan memiliki beberapa blok ResNet subclassing Layer , dan satu Model meliputi seluruh jaringan ResNet50.

The Model kelas memiliki API yang sama seperti Layer , dengan perbedaan berikut:

  • Ini mengekspos built-in loop pelatihan, evaluasi, dan prediksi ( model.fit() , model.evaluate() , model.predict() ).
  • Ini memperlihatkan daftar lapisan dalamnya, melalui model.layers properti.
  • Ini memperlihatkan tabungan dan API serialisasi ( save() , save_weights() ...)

Secara efektif, Layer kelas berkorespondensi dengan apa yang kita sebut dalam literatur sebagai "lapisan" (seperti dalam "lapisan lilit" atau "lapisan berulang") atau sebagai "blok" (seperti dalam "ResNet blok" atau "blok Inception" ).

Sementara itu, Model kelas berkorespondensi dengan apa yang disebut dalam literatur sebagai "model" (seperti dalam "model pembelajaran mendalam") atau sebagai "jaringan" (seperti dalam "jaringan saraf dalam").

Jadi jika Anda bertanya-tanya, "saya harus menggunakan Layer kelas atau Model kelas?", Tanyakan pada diri sendiri: Saya akan perlu untuk panggilan fit() di atasnya? Aku akan perlu memanggil save() di atasnya? Jika demikian, pergi dengan Model . Jika tidak (baik karena kelas Anda hanya blok dalam sistem yang lebih besar, atau karena Anda menulis pelatihan & kode sendiri tabungan), penggunaan Layer .

Misalnya, kita bisa mengambil kami contoh mini-resnet di atas, dan menggunakannya untuk membangun Model yang kami bisa berlatih dengan fit() , dan bahwa kita bisa simpan dengan save_weights() :

class ResNet(tf.keras.Model):

    def __init__(self, num_classes=1000):
        super(ResNet, self).__init__()
        self.block_1 = ResNetBlock()
        self.block_2 = ResNetBlock()
        self.global_pool = layers.GlobalAveragePooling2D()
        self.classifier = Dense(num_classes)

    def call(self, inputs):
        x = self.block_1(inputs)
        x = self.block_2(x)
        x = self.global_pool(x)
        return self.classifier(x)


resnet = ResNet()
dataset = ...
resnet.fit(dataset, epochs=10)
resnet.save(filepath)

Menyatukan semuanya: contoh ujung ke ujung

Inilah yang telah Anda pelajari sejauh ini:

  • Sebuah Layer merangkum keadaan (dibuat di __init__() atau build() ) dan beberapa perhitungan (didefinisikan dalam call() ).
  • Lapisan dapat disarangkan secara rekursif untuk membuat blok komputasi baru yang lebih besar.
  • Lapisan dapat membuat dan kerugian track (biasanya kerugian regularisasi) serta metrik, melalui add_loss() dan add_metric()
  • Wadah luar, hal yang Anda inginkan untuk melatih, adalah Model . Sebuah Model adalah seperti Layer , tapi dengan menambahkan pelatihan dan serialisasi utilitas.

Mari kita gabungkan semua hal ini menjadi contoh ujung ke ujung: kita akan menerapkan Variational AutoEncoder (VAE). Kami akan melatihnya pada angka MNIST.

Vae kami akan menjadi subclass dari Model , dibangun sebagai komposisi bersarang lapisan yang subclass Layer . Ini akan menampilkan kerugian regularisasi (KL divergensi).

from tensorflow.keras import layers


class Sampling(layers.Layer):
    """Uses (z_mean, z_log_var) to sample z, the vector encoding a digit."""

    def call(self, inputs):
        z_mean, z_log_var = inputs
        batch = tf.shape(z_mean)[0]
        dim = tf.shape(z_mean)[1]
        epsilon = tf.keras.backend.random_normal(shape=(batch, dim))
        return z_mean + tf.exp(0.5 * z_log_var) * epsilon


class Encoder(layers.Layer):
    """Maps MNIST digits to a triplet (z_mean, z_log_var, z)."""

    def __init__(self, latent_dim=32, intermediate_dim=64, name="encoder", **kwargs):
        super(Encoder, self).__init__(name=name, **kwargs)
        self.dense_proj = layers.Dense(intermediate_dim, activation="relu")
        self.dense_mean = layers.Dense(latent_dim)
        self.dense_log_var = layers.Dense(latent_dim)
        self.sampling = Sampling()

    def call(self, inputs):
        x = self.dense_proj(inputs)
        z_mean = self.dense_mean(x)
        z_log_var = self.dense_log_var(x)
        z = self.sampling((z_mean, z_log_var))
        return z_mean, z_log_var, z


class Decoder(layers.Layer):
    """Converts z, the encoded digit vector, back into a readable digit."""

    def __init__(self, original_dim, intermediate_dim=64, name="decoder", **kwargs):
        super(Decoder, self).__init__(name=name, **kwargs)
        self.dense_proj = layers.Dense(intermediate_dim, activation="relu")
        self.dense_output = layers.Dense(original_dim, activation="sigmoid")

    def call(self, inputs):
        x = self.dense_proj(inputs)
        return self.dense_output(x)


class VariationalAutoEncoder(keras.Model):
    """Combines the encoder and decoder into an end-to-end model for training."""

    def __init__(
        self,
        original_dim,
        intermediate_dim=64,
        latent_dim=32,
        name="autoencoder",
        **kwargs
    ):
        super(VariationalAutoEncoder, self).__init__(name=name, **kwargs)
        self.original_dim = original_dim
        self.encoder = Encoder(latent_dim=latent_dim, intermediate_dim=intermediate_dim)
        self.decoder = Decoder(original_dim, intermediate_dim=intermediate_dim)

    def call(self, inputs):
        z_mean, z_log_var, z = self.encoder(inputs)
        reconstructed = self.decoder(z)
        # Add KL divergence regularization loss.
        kl_loss = -0.5 * tf.reduce_mean(
            z_log_var - tf.square(z_mean) - tf.exp(z_log_var) + 1
        )
        self.add_loss(kl_loss)
        return reconstructed

Mari kita menulis loop pelatihan sederhana di MNIST:

original_dim = 784
vae = VariationalAutoEncoder(original_dim, 64, 32)

optimizer = tf.keras.optimizers.Adam(learning_rate=1e-3)
mse_loss_fn = tf.keras.losses.MeanSquaredError()

loss_metric = tf.keras.metrics.Mean()

(x_train, _), _ = tf.keras.datasets.mnist.load_data()
x_train = x_train.reshape(60000, 784).astype("float32") / 255

train_dataset = tf.data.Dataset.from_tensor_slices(x_train)
train_dataset = train_dataset.shuffle(buffer_size=1024).batch(64)

epochs = 2

# Iterate over epochs.
for epoch in range(epochs):
    print("Start of epoch %d" % (epoch,))

    # Iterate over the batches of the dataset.
    for step, x_batch_train in enumerate(train_dataset):
        with tf.GradientTape() as tape:
            reconstructed = vae(x_batch_train)
            # Compute reconstruction loss
            loss = mse_loss_fn(x_batch_train, reconstructed)
            loss += sum(vae.losses)  # Add KLD regularization loss

        grads = tape.gradient(loss, vae.trainable_weights)
        optimizer.apply_gradients(zip(grads, vae.trainable_weights))

        loss_metric(loss)

        if step % 100 == 0:
            print("step %d: mean loss = %.4f" % (step, loss_metric.result()))
Start of epoch 0
step 0: mean loss = 0.3184
step 100: mean loss = 0.1252
step 200: mean loss = 0.0989
step 300: mean loss = 0.0890
step 400: mean loss = 0.0841
step 500: mean loss = 0.0807
step 600: mean loss = 0.0787
step 700: mean loss = 0.0771
step 800: mean loss = 0.0759
step 900: mean loss = 0.0749
Start of epoch 1
step 0: mean loss = 0.0746
step 100: mean loss = 0.0740
step 200: mean loss = 0.0735
step 300: mean loss = 0.0730
step 400: mean loss = 0.0727
step 500: mean loss = 0.0723
step 600: mean loss = 0.0720
step 700: mean loss = 0.0717
step 800: mean loss = 0.0715
step 900: mean loss = 0.0712

Perhatikan bahwa sejak Vae adalah subclassing Model , itu fitur built-in loop pelatihan. Jadi Anda juga bisa melatihnya seperti ini:

vae = VariationalAutoEncoder(784, 64, 32)

optimizer = tf.keras.optimizers.Adam(learning_rate=1e-3)

vae.compile(optimizer, loss=tf.keras.losses.MeanSquaredError())
vae.fit(x_train, x_train, epochs=2, batch_size=64)
Epoch 1/2
938/938 [==============================] - 3s 3ms/step - loss: 0.0745
Epoch 2/2
938/938 [==============================] - 3s 3ms/step - loss: 0.0676
<keras.callbacks.History at 0x7fce90282750>

Di luar pengembangan berorientasi objek: API Fungsional

Apakah contoh ini terlalu banyak pengembangan berorientasi objek untuk Anda? Anda juga dapat membangun model menggunakan API Fungsional . Yang penting, memilih satu gaya atau lainnya tidak mencegah Anda memanfaatkan komponen yang ditulis dalam gaya lain: Anda selalu dapat memadupadankan.

Misalnya, API contoh Fungsional bawah digunakan ulang sama Sampling lapisan kami didefinisikan dalam contoh di atas:

original_dim = 784
intermediate_dim = 64
latent_dim = 32

# Define encoder model.
original_inputs = tf.keras.Input(shape=(original_dim,), name="encoder_input")
x = layers.Dense(intermediate_dim, activation="relu")(original_inputs)
z_mean = layers.Dense(latent_dim, name="z_mean")(x)
z_log_var = layers.Dense(latent_dim, name="z_log_var")(x)
z = Sampling()((z_mean, z_log_var))
encoder = tf.keras.Model(inputs=original_inputs, outputs=z, name="encoder")

# Define decoder model.
latent_inputs = tf.keras.Input(shape=(latent_dim,), name="z_sampling")
x = layers.Dense(intermediate_dim, activation="relu")(latent_inputs)
outputs = layers.Dense(original_dim, activation="sigmoid")(x)
decoder = tf.keras.Model(inputs=latent_inputs, outputs=outputs, name="decoder")

# Define VAE model.
outputs = decoder(z)
vae = tf.keras.Model(inputs=original_inputs, outputs=outputs, name="vae")

# Add KL divergence regularization loss.
kl_loss = -0.5 * tf.reduce_mean(z_log_var - tf.square(z_mean) - tf.exp(z_log_var) + 1)
vae.add_loss(kl_loss)

# Train.
optimizer = tf.keras.optimizers.Adam(learning_rate=1e-3)
vae.compile(optimizer, loss=tf.keras.losses.MeanSquaredError())
vae.fit(x_train, x_train, epochs=3, batch_size=64)
Epoch 1/3
938/938 [==============================] - 3s 3ms/step - loss: 0.0748
Epoch 2/3
938/938 [==============================] - 3s 3ms/step - loss: 0.0676
Epoch 3/3
938/938 [==============================] - 3s 3ms/step - loss: 0.0676
<keras.callbacks.History at 0x7fce90233cd0>

Untuk informasi lebih lanjut, pastikan untuk membaca panduan API Fungsional .