 TensorFlow.orgで表示 |  GoogleColabで実行 |  GitHubでソースを表示 |  ノートブックをダウンロード |
セットアップ
import tensorflow as tf
from tensorflow import keras
Layerクラス：状態（重み）といくつかの計算の組み合わせ
Kerasの中心的な抽象化の1つは、 Layerクラスです。レイヤーは、状態（レイヤーの「重み」）と入力から出力への変換（「呼び出し」、レイヤーのフォワードパス）の両方をカプセル化します。
これが密に接続されたレイヤーです。状態があります：変数wと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
Python関数のように、いくつかのテンソル入力でレイヤーを呼び出すことでレイヤーを使用します。
x = tf.ones((2, 2))
linear_layer = Linear(4, 2)
y = linear_layer(x)
print(y)
tf.Tensor( [[-0.03354992 0.00427916 -0.15999871 -0.00302821] [-0.03354992 0.00427916 -0.15999871 -0.00302821]], shape=(2, 4), dtype=float32)
重みwとbは、レイヤー属性として設定されると、レイヤーによって自動的に追跡されることに注意してください。
assert linear_layer.weights == [linear_layer.w, linear_layer.b]
レイヤーにウェイトを追加するためのより迅速なショートカットにもアクセスできることに注意してください： add_weight()メソッド：
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.04145671 0.06744058 0.05072812 0.0403844 ] [0.04145671 0.06744058 0.05072812 0.0403844 ]], shape=(2, 4), dtype=float32)
層は訓練不可能な重みを持つことができます
トレーニング可能なウェイトに加えて、トレーニング不可能なウェイトをレイヤーに追加することもできます。このような重みは、レイヤーをトレーニングしているときのバックプロパゲーション中に考慮されないようになっています。
訓練不可能なウェイトを追加して使用する方法は次のとおりです。
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.]
これはlayer.weightsの一部ですが、トレーニング不可能なウェイトとして分類されます。
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: []
ベストプラクティス：入力の形状がわかるまでウェイトの作成を延期する
上記のLinearレイヤーは、 input_dim __init__()で重みwとb形状を計算するために使用されたinput_dim引数を取りました。
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
多くの場合、入力のサイズを事前に知らない可能性があり、レイヤーをインスタンス化した後しばらくして、その値がわかったときにウェイトを遅延的に作成する必要があります。
Keras APIでは、レイヤーのbuild(self, inputs_shape)メソッドでレイヤーウェイトを作成することをお勧めします。このような：
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
レイヤーの__call__()メソッドは、最初に呼び出されたときにビルドを自動的に実行します。これで、怠惰で使いやすいレイヤーができました。
# 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)
レイヤーは再帰的に構成可能です
レイヤーインスタンスを別のレイヤーの属性として割り当てると、外側のレイヤーが内側のレイヤーの重みの追跡を開始します。
このようなサブレイヤーは、 __init__()メソッドで作成することをお勧めします（サブレイヤーには通常buildメソッドがあるため、外側のレイヤーがビルドされるときにビルドされます）。
# Let's assume we are reusing the Linear class
# with a `build` method that we defined above.
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
add_loss()メソッド
レイヤーのcall()メソッドを作成するときに、後でトレーニングループを作成するときに使用する損失テンソルを作成できます。これは、 self.add_loss(value)呼び出すことで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
これらの損失（内部レイヤーによって作成されたものを含む）は、 layer.lossesを介して取得できます。このプロパティは、最上位レイヤーへのすべての__call__()の開始時にリセットされるため、 layer.lossesには、最後のフォワードパス中に作成された損失値が常に含まれます。
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
さらに、 lossプロパティには、内層の重みに対して作成された正規化損失も含まれます。
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.0021053506>]
これらの損失は、次のようなトレーニングループを作成するときに考慮に入れることを目的としています。
# 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))
トレーニングループの作成に関する詳細なガイドについては、トレーニングループを最初から作成するためのガイドを参照してください。
これらの損失は、 fit()とシームレスに連携しますfit()自動的に合計され、メインの損失がある場合は追加されます）。
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`, thee 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 122ms/step - loss: 0.1508 1/1 [==============================] - 0s 49ms/step - loss: 0.0365 <tensorflow.python.keras.callbacks.History at 0x7f91d04d9128>
add_metric()メソッド
add_loss()と同様に、レイヤーには、トレーニング中の数量の移動平均を追跡するためのadd_metric()メソッドもあります。
次のレイヤーについて考えてみます。「ロジスティックエンドポイント」レイヤー。それはそれを介して追跡損失計算、入力予測＆標的として取るadd_loss() 、そしてそれは介して追跡精度スカラー演算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)
この方法で追跡されたメトリックには、 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: [<tensorflow.python.keras.metrics.BinaryAccuracy object at 0x7f91d06ef0b8>] current accuracy value: 1.0
add_loss()場合と同様に、これらのメトリックは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 254ms/step - loss: 0.9207 - binary_accuracy: 0.0000e+00 <tensorflow.python.keras.callbacks.History at 0x7f91d04b70f0>
オプションで、レイヤーでシリアル化を有効にすることができます
カスタムレイヤーを機能モデルの一部としてシリアル化できるようにする必要がある場合は、オプションでget_config()メソッドを実装できます。
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}
基本Layerクラスの__init__()メソッドは、いくつかのキーワード引数、特にnameとdtype 。これらの引数を__init__()親クラスに渡し、レイヤー構成に含めることをお__init__()します。
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}
レイヤーを構成から逆シリアル化するときにさらに柔軟性が必要な場合は、 from_config()クラスメソッドをオーバーライドすることもできます。これはfrom_config()基本実装です：
def from_config(cls, config):
return cls(**config)
シリアル化と保存の詳細については、モデルの保存とシリアル化の完全ガイドを参照してください。
call()メソッドの特権training引数
一部のレイヤー、特にBatchNormalizationレイヤーとDropoutレイヤーは、トレーニングと推論中に異なる動作をします。このようなレイヤーの場合、 call()メソッドでtraining （ブール）引数を公開するのが標準的な方法です。
この引数をcall()で公開することにより、組み込みのトレーニングループと評価ループ（例： fit() ）がトレーニングと推論でレイヤーを正しく使用できるようにします。
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
call()メソッドの特権mask引数
call()サポートされている他の特権引数は、 mask引数です。
すべてのKerasRNNレイヤーにあります。マスクは、時系列データを処理するときに特定の入力タイムステップをスキップするために使用されるブールテンソル（入力のタイムステップごとに1つのブール値）です。
Kerasは、前のレイヤーによってマスクが生成されると、それをサポートするレイヤーの正しいmask引数を__call__()に自動的に渡します。マスク生成レイヤーは、 mask_zero=Trueで構成されたEmbeddingレイヤーと、 Maskingレイヤーです。
マスキングとマスキング対応レイヤーの作成方法の詳細については、ガイド「パディングとマスキングについて」をご覧ください。
Modelクラス
一般に、 Layerクラスを使用して内部計算ブロックを定義し、 Modelクラスを使用して外部モデル（トレーニングするオブジェクト）を定義します。
たとえば、ResNet50モデルでは、 Layerサブクラス化する複数のResNetブロックと、ResNet50ネットワーク全体を網羅する単一のModelあります。
ModelクラスのAPIはLayerと同じですが、次の違いがあります。
- 組み込みのトレーニング、評価、および予測ループ（
model.fit()、model.evaluate()、model.predict()）をmodel.predict()ます。 -
model.layersプロパティを介して、内部レイヤーのリストを公開します。 - 保存およびシリアル化APIを公開します（
save()、save_weights()...）
事実上、 Layerクラスは、文献で「レイヤー」（「畳み込みレイヤー」または「リカレントレイヤー」など）または「ブロック」（「ResNetブロック」または「インセプションブロック」など）と呼ばれるものに対応します。 ）。
一方、 Modelクラスは、文献で「モデル」（「深層学習モデル」のように）または「ネットワーク」（「深層ニューラルネットワーク」のように）と呼ばれるものに対応します。
したがって、「 LayerクラスとModelクラスのどちらを使用する必要がありますか？」と疑問に思っている場合は、 fit()を呼び出す必要がありますか？その上でsave()を呼び出す必要がありますか？もしそうなら、 Modelで行きます。そうでない場合（クラスがより大きなシステムの単なるブロックであるため、またはトレーニングを作成してコードを自分で保存しているため）、 Layer使用します。
たとえば、上記のミニresnetの例を使用して、 fit()でsave_weights()でき、 save_weights()保存できるModelを作成できます。
class ResNet(tf.keras.Model):
def __init__(self):
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)
すべてをまとめる：エンドツーエンドの例
これまでに学んだことは次のとおりです。
-
Layerは、状態（__init__()またはbuild()）といくつかの計算（call()定義）をカプセル化しcall()。 - レイヤーを再帰的にネストして、新しい、より大きな計算ブロックを作成できます。
- レイヤーは、
add_loss()およびadd_metric()介して、メトリックだけでなく損失（通常は正規化損失）を作成および追跡できます。 - トレーニングしたい外側のコンテナは
Modelです。ModelはLayerと同じですが、トレーニングとシリアル化のユーティリティが追加されています。
これらすべてをエンドツーエンドの例にまとめましょう。VariationalAutoEncoder（VAE）を実装します。 MNISTディジットでトレーニングします。
私たちのVAEはModelサブクラスであり、 Layerをサブクラス化するLayerネストされた構成として構築されます。正則化損失（KL発散）が特徴です。
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
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.3442 step 100: mean loss = 0.1247 step 200: mean loss = 0.0988 step 300: mean loss = 0.0888 step 400: mean loss = 0.0841 step 500: mean loss = 0.0807 step 600: mean loss = 0.0786 step 700: mean loss = 0.0770 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.0739 step 200: mean loss = 0.0734 step 300: mean loss = 0.0730 step 400: mean loss = 0.0726 step 500: mean loss = 0.0722 step 600: mean loss = 0.0720 step 700: mean loss = 0.0717 step 800: mean loss = 0.0714 step 900: mean loss = 0.0712
VAEはModelサブクラス化しているため、組み込みのトレーニングループを備えていることに注意してください。したがって、次のようにトレーニングすることもできます。
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.0944 Epoch 2/2 938/938 [==============================] - 2s 3ms/step - loss: 0.0676 <tensorflow.python.keras.callbacks.History at 0x7f91d0263588>
オブジェクト指向開発を超えて：Functional API
この例は、オブジェクト指向の開発が多すぎましたか？ FunctionalAPIを使用してモデルを構築することもできます。重要なのは、いずれかのスタイルを選択しても、他のスタイルで記述されたコンポーネントを活用することを妨げるものではないということです。いつでも組み合わせることができます。
たとえば、以下のFunctional APIの例では、上記の例で定義したものと同じSamplingレイヤーを再利用しています。
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.0961 Epoch 2/3 938/938 [==============================] - 2s 3ms/step - loss: 0.0677 Epoch 3/3 938/938 [==============================] - 2s 3ms/step - loss: 0.0676 <tensorflow.python.keras.callbacks.History at 0x7f91d00b6358>
詳細については、 FunctionalAPIガイドを必ずお読みください。