import tensorflow as tf
from tensorflow.keras import layers
import numpy as np
# Load MNIST data
(x_train, _), (x_test, _) = tf.keras.datasets.mnist.load_data()
x_train = x_train.astype('float32') / 255.
x_test = x_test.astype('float32') / 255.
x_train = np.reshape(x_train, (-1, 28, 28, 1))
x_test = np.reshape(x_test, (-1, 28, 28, 1))
latent_dim = 2
# Encoder
inputs = tf.keras.Input(shape=(28, 28, 1))
x = layers.Conv2D(32, 3, activation='relu', strides=2, padding='same')(inputs)
x = layers.BatchNormalization()(x)
x = layers.Conv2D(64, 3, activation='relu', strides=2, padding='same')(x)
x = layers.BatchNormalization()(x)
x = layers.Flatten()(x)
x = layers.Dense(16, activation='relu')(x)
z_mean = layers.Dense(latent_dim)(x)
z_log_var = layers.Dense(latent_dim)(x)
# Sampling layer
class Sampling(layers.Layer):
def call(self, inputs):
z_mean, z_log_var = inputs
batch = tf.shape(z_mean)[0]
dim = tf.shape(z_mean)[1]
epsilon = tf.random.normal(shape=(batch, dim))
return z_mean + tf.exp(0.5 * z_log_var) * epsilon
z = Sampling()([z_mean, z_log_var])
# Decoder
latent_inputs = tf.keras.Input(shape=(latent_dim,))
x = layers.Dense(7 * 7 * 64, activation='relu')(latent_inputs)
x = layers.Reshape((7, 7, 64))(x)
x = layers.Conv2DTranspose(64, 3, strides=2, activation='relu', padding='same')(x)
x = layers.BatchNormalization()(x)
x = layers.Conv2DTranspose(32, 3, strides=2, activation='relu', padding='same')(x)
x = layers.BatchNormalization()(x)
outputs = layers.Conv2DTranspose(1, 3, activation='sigmoid', padding='same')(x)
def vae_loss(x, x_decoded_mean, z_mean, z_log_var):
reconstruction_loss = tf.reduce_mean(tf.keras.losses.binary_crossentropy(x, x_decoded_mean)) * 28 * 28
kl_loss = -0.5 * tf.reduce_mean(1 + z_log_var - tf.square(z_mean) - tf.exp(z_log_var))
return reconstruction_loss + kl_loss
encoder = tf.keras.Model(inputs, [z_mean, z_log_var, z], name='encoder')
decoder = tf.keras.Model(latent_inputs, outputs, name='decoder')
# VAE model
outputs = decoder(encoder(inputs)[2])
vae = tf.keras.Model(inputs, outputs, name='vae')
optimizer = tf.keras.optimizers.Adam(learning_rate=0.0005)
# Custom train step
@tf.function
def train_step(x):
with tf.GradientTape() as tape:
z_mean, z_log_var, z = encoder(x)
x_decoded = decoder(z)
loss = vae_loss(x, x_decoded, z_mean, z_log_var)
gradients = tape.gradient(loss, vae.trainable_variables)
optimizer.apply_gradients(zip(gradients, vae.trainable_variables))
return loss
# Training loop
batch_size = 128
epochs = 30
train_dataset = tf.data.Dataset.from_tensor_slices(x_train).shuffle(60000).batch(batch_size)
for epoch in range(epochs):
total_loss = 0
for step, x_batch in enumerate(train_dataset):
loss = train_step(x_batch)
total_loss += loss
print(f'Epoch {epoch + 1}, Loss: {total_loss / (step + 1):.4f}')
# Evaluate on test data
z_mean, z_log_var, z = encoder.predict(x_test, batch_size=batch_size)
x_decoded = decoder.predict(z, batch_size=batch_size)
reconstruction_loss = np.mean(tf.keras.losses.binary_crossentropy(x_test.reshape(-1, 28*28), x_decoded.reshape(-1, 28*28)), axis=1) * 28 * 28
average_loss = np.mean(reconstruction_loss - (-0.5 * np.sum(1 + z_log_var - np.square(z_mean) - np.exp(z_log_var), axis=1)))
# Calculate reconstruction accuracy as percentage of images with average pixel difference < 0.1
pixel_diff = np.mean(np.abs(x_test - x_decoded), axis=(1,2,3))
reconstruction_accuracy = np.mean(pixel_diff < 0.1) * 100
print(f'Validation loss: {average_loss:.2f}, Validation reconstruction accuracy: {reconstruction_accuracy:.2f}%')