（附代码）实战 | 变分自编码器的原理与项目

import tensorflow as tffrom tensorflow.contrib import layers
## encoder 模块def fc_encoder(x, latent_dim, activation=None):    e = layers.fully_connected(x, 500, scope='fc-01')    e = layers.fully_connected(e, 200, scope='fc-02')    output = layers.fully_connected(e, 2 * latent_dim, activation_fn=activation,                                    scope='fc-final')
    return output
## decoder 模块def fc_decoder(z, observation_dim, activation=tf.sigmoid):    x = layers.fully_connected(z, 200, scope='fc-01')    x = layers.fully_connected(x, 500, scope='fc-02')    output = layers.fully_connected(x, observation_dim, activation_fn=activation,                                    scope='fc-final')    return output

   ## KL loss    def _kl_diagnormal_stdnormal(mu, log_var):        var = tf.exp(log_var)        kl = 0.5 * tf.reduce_sum(tf.square(mu) + var - 1. - log_var)        return kl      ## 基于高斯分布的重建loss    def gaussian_log_likelihood(targets, mean, std):        se = 0.5 * tf.reduce_sum(tf.square(targets - mean)) / (2*tf.square(std)) + tf.log(std)        return se           ## 基于伯努利分布的重建loss    def bernoulli_log_likelihood(targets, outputs, eps=1e-8):        log_like = -tf.reduce_sum(targets * tf.log(outputs + eps)                                  + (1. - targets) * tf.log((1. - outputs) + eps))        return log_like

!pip show tensorflow

!pip install tensorflow==2.3.0 -i https://pypi.tuna.tsinghua.edu.cn/simple

!pip install imageio

import tensorflow as tf
import osimport timeimport numpy as npimport globimport matplotlib.pyplot as pltimport PILimport imageio
from IPython import display
(train_images, _), (test_images, _) = tf.keras.datasets.mnist.load_data()
train_images = train_images.reshape(train_images.shape[0], 28, 28, 1).astype('float32')test_images = test_images.reshape(test_images.shape[0], 28, 28, 1).astype('float32')
# 标准化图片到区间 [0., 1.] 内train_images /= 255.test_images /= 255.
# 二值化train_images[train_images >= .5] = 1.train_images[train_images < .5] = 0.test_images[test_images >= .5] = 1.test_images[test_images < .5] = 0.
# 使用 tf.data 来将数据分批和打乱train_dataset = tf.data.Dataset.from_tensor_slices(train_images).shuffle(TRAIN_BUF).batch(BATCH_SIZE)test_dataset = tf.data.Dataset.from_tensor_slices(test_images).shuffle(TEST_BUF).batch(BATCH_SIZE)

class CVAE(tf.keras.Model):  """Convolutional variational autoencoder."""
  def __init__(self, latent_dim):    super(CVAE, self).__init__()    self.latent_dim = latent_dim    self.encoder = tf.keras.Sequential(        [            tf.keras.layers.InputLayer(input_shape=(28, 28, 1)),            tf.keras.layers.Conv2D(                filters=32, kernel_size=3, strides=(2, 2), activation='relu'),            tf.keras.layers.Conv2D(                filters=64, kernel_size=3, strides=(2, 2), activation='relu'),            tf.keras.layers.Flatten(),            # No activation            tf.keras.layers.Dense(latent_dim + latent_dim),        ]    )
    self.decoder = tf.keras.Sequential(        [            tf.keras.layers.InputLayer(input_shape=(latent_dim,)),            tf.keras.layers.Dense(units=7*7*32, activation=tf.nn.relu),            tf.keras.layers.Reshape(target_shape=(7, 7, 32)),            tf.keras.layers.Conv2DTranspose(                filters=64, kernel_size=3, strides=2, padding='same',                activation='relu'),            tf.keras.layers.Conv2DTranspose(                filters=32, kernel_size=3, strides=2, padding='same',                activation='relu'),            # No activation            tf.keras.layers.Conv2DTranspose(                filters=1, kernel_size=3, strides=1, padding='same'),        ]    )
  @tf.function  def sample(self, eps=None):    if eps is None:      eps = tf.random.normal(shape=(100, self.latent_dim))    return self.decode(eps, apply_sigmoid=True)
  def encode(self, x):    mean, logvar = tf.split(self.encoder(x), num_or_size_splits=2, axis=1)    return mean, logvar
  def reparameterize(self, mean, logvar):    eps = tf.random.normal(shape=mean.shape)    return eps * tf.exp(logvar * .5) + mean
  def decode(self, z, apply_sigmoid=False):    logits = self.decoder(z)    if apply_sigmoid:      probs = tf.sigmoid(logits)      return probs    return logits

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

def log_normal_pdf(sample, mean, logvar, raxis=1):  log2pi = tf.math.log(2. * np.pi)  return tf.reduce_sum(      -.5 * ((sample - mean) ** 2. * tf.exp(-logvar) + logvar + log2pi),      axis=raxis)

def compute_loss(model, x):  mean, logvar = model.encode(x)  z = model.reparameterize(mean, logvar)  x_logit = model.decode(z)  cross_ent = tf.nn.sigmoid_cross_entropy_with_logits(logits=x_logit, labels=x)  logpx_z = -tf.reduce_sum(cross_ent, axis=[1, 2, 3])  logpz = log_normal_pdf(z, 0., 0.)  logqz_x = log_normal_pdf(z, mean, logvar)  return -tf.reduce_mean(logpx_z + logpz - logqz_x)

@tf.functiondef train_step(model, x, optimizer):  """Executes one training step and returns the loss.
  This function computes the loss and gradients, and uses the latter to  update the model's parameters.  """  with tf.GradientTape() as tape:    loss = compute_loss(model, x)  gradients = tape.gradient(loss, model.trainable_variables)  optimizer.apply_gradients(zip(gradients, model.trainable_variables))

epochs = 100# set the dimensionality of the latent space to a plane for visualization laterlatent_dim = 2num_examples_to_generate = 16
# keeping the random vector constant for generation (prediction) so# it will be easier to see the improvement.random_vector_for_generation = tf.random.normal(    shape=[num_examples_to_generate, latent_dim])model = CVAE(latent_dim)
def generate_and_save_images(model, epoch, test_sample):  mean, logvar = model.encode(test_sample)  z = model.reparameterize(mean, logvar)  predictions = model.sample(z)  fig = plt.figure(figsize=(4, 4))
  for i in range(predictions.shape[0]):    plt.subplot(4, 4, i + 1)    plt.imshow(predictions[i, :, :, 0], cmap='gray')    plt.axis('off')
  # tight_layout minimizes the overlap between 2 sub-plots  plt.savefig('image_at_epoch_{:04d}.png'.format(epoch))  plt.show()
# Pick a sample of the test set for generating output imagesassert batch_size >= num_examples_to_generatefor test_batch in test_dataset.take(1):  test_sample = test_batch[0:num_examples_to_generate, :, :, :]generate_and_save_images(model, 0, test_sample)
for epoch in range(1, epochs + 1):  start_time = time.time()  for train_x in train_dataset:    train_step(model, train_x, optimizer)  end_time = time.time()
  loss = tf.keras.metrics.Mean()  for test_x in test_dataset:    loss(compute_loss(model, test_x))  elbo = -loss.result()  display.clear_output(wait=False)  print('Epoch: {}, Test set ELBO: {}, time elapse for current epoch: {}'        .format(epoch, elbo, end_time - start_time))  generate_and_save_images(model, epoch, test_sample)

anim_file = 'cvae.gif'
with imageio.get_writer(anim_file, mode='I') as writer:  filenames = glob.glob('image*.png')  filenames = sorted(filenames)  for filename in filenames:    image = imageio.imread(filename)    writer.append_data(image)  image = imageio.imread(filename)  writer.append_data(image)

import tensorflow_docs.vis.embed as embedembed.embed_file(anim_file)