Python深度学习基于Tensorflow（11）生成式深度学习

生成式模型是一种能够描述数据生成过程的统计模型。这种模型能够基于给定的输入数据，学习到数据的潜在分布，并能够生成与训练数据类似的新数据实例。在机器学习中，生成式模型通常用于概率推理任务，如图像生成、自然语言处理和音频合成等。

生成式模型：学习数据的联合概率分布 $P(X,Y)$，即同时建模输入 $X$ 和输出 $Y$ 的分布。它们能够生成新的数据实例，因为它们知道数据是如何生成的。常用于生成任务，如生成新的图像、文本或音乐，以及那些需要推断数据潜在分布的场景。
判别式模型：学习数据的条件概率分布 $P(Y\mid X)$，即给定输入 $X$ 时输出 $Y$ 的分布。它们主要关注于如何将输入映射到输出，而不关心数据是如何生成的。常用于分类和回归任务，如图像识别、垃圾邮件检测等，它们直接建模预测标签与输入之间的关系。

生成式模型有很多，如：高斯混合模型（Gaussian Mixture Model, GMM）、隐马尔可夫模型（Hidden Markov Model, HMM）、变分自编码器（Variational Autoencoder, VAE）和生成对抗网络（Generative Adversarial Network, GAN）；这里主要介绍变分自编码器（VAE）和生成对抗网络（GAN）；

变分自编码器

首先介绍一下自编码器，自编码器在前面自然语言处理和注意力机制章节都用过，本质是一个 Encoder过程 和一个 Decoder过程，自编码器是通过对输入 X 进行 Encoder 后得到一个低维的向量 Z ，然后根据这个向量 Z 进行 Decoder 得到目标 Y。在生成式模型中，我们可以把目标设定为输入 X ，通过对比输入和输出的误差，再利用神经网络去训练使得误差逐渐减小，从而达到非监督学习的目的。这样会有一个很严重的缺点：低维的向量 Z 的分布是未知的，我们不能够使用无条件随机生成的 Z 进行 Decoder 得到目标；

为解决这一问题，人们对潜在空间 Z（潜在变量对应的空间）增加一些约束，使 Z 满足正态分布，由此就出现了 VAE模型，VAE对编码器添加约束，就是强迫它产生服从单位正态分布的潜在变量。正是这种约束，把VAE和自编码器区分开来。

n维输入样本 X 通过 Encoder 输出两个m维向量：mu和log_var，这两个向量是潜在空间（假设满足正态分布）的两个参数（相当于均值和方差）。 利用如下公式，随机生成的 ε，以及mu和log_var去生成Z$$Z=\text{mu}+exp(\text{log-var})\cdot \epsilon$$
先从单位高斯中采样得到 ε ，然后乘以标准差 log_var 并加上平均值 mu ，这样可以确保梯度能确保梯度能传回编码器。这里严格意义来说使用 var 表示标准差不合适，最好使用 std ；

这就可保证潜在空间的连续性、良好的结构性。而这些特性使得潜在空间的每个方向都表示数据中有意义的变化方向。 以上这些步骤构成整个网络的正向传播过程，反向传播的损失函数从以下两个方面进行衡量：

1. 生成的新图像与原图像的相似度；
2. 隐含空间的条件分布与真实分布的相似度；

度量图像的相似度一般采用交叉熵，度量两个分布的相似度一般采用KL散度(Kullback-Leibler divergence)。这两个度量的和构成了整个模型的损失函数。

使用 MNIST 数据集实现 VAE 功能如下：

数据导入：

import tensorflow as tf
import numpy as np
import time
import matplotlib.pyplot as plt

(x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()

# 在这里简单做个二值化处理
x_train, x_test = (x_train>0).astype(np.float32), (x_test>0).astype(np.float32)
train_x = tf.data.Dataset.from_tensor_slices(np.expand_dims(x_train, axis=-1)).batch(32)
test_x = tf.data.Dataset.from_tensor_slices(np.expand_dims(x_test, axis=-1)).batch(32)
1
2
3
4
5
6
7
8
9
10
11

VAE模型建立：

对于编码器，我们使用了两个卷积层加一个全连接层，而对于解码器，我们使用全连接层加3个反卷积层。 注意：训练VAE过程中要避免使用批归一化，因为这样会导致额外的随机性，从而加剧随机抽样的不稳定性。

class VAE(tf.keras.Model):
    def __init__(self, latent_dim):
        super(VAE, self).__init__()
        self.latent_dim = latent_dim
        self.inference_net = 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(),
            tf.keras.layers.Dense(self.latent_dim * 2)
        ])
        self.generative_net = tf.keras.Sequential([
            tf.keras.layers.InputLayer(input_shape=[self.latent_dim]),
            tf.keras.layers.Dense(units=7*7*32, activation='relu'),
            tf.keras.layers.Reshape(target_shape=[7,7,32]),
            tf.keras.layers.Conv2DTranspose(filters=64, kernel_size=3, strides=(2,2), activation='relu', padding='same'),
            tf.keras.layers.Conv2DTranspose(filters=32, kernel_size=3, strides=(2,2), activation='relu', padding='same'),
            tf.keras.layers.Conv2DTranspose(filters=1, kernel_size=3, strides=(1,1), activation='sigmoid', padding='same'),
        ])

    def reparameterize(self, mean, logvar):
        eps = tf.random.normal(shape=mean.shape)
        return eps * tf.exp(logvar * 0.5) + mean

    def encode(self, x):
        mu, logvar = tf.split(self.inference_net(x), num_or_size_splits=2, axis=1)
        return mu, logvar

    def decode(self, z):
        logits = self.generative_net(z)
        return logits

    def call(self, x):
        return self.decode(x)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

定义损失，损失有两个方面：

1. 生成的新图像与原图像的相似度；
2. 隐含空间的条件分布与真实分布的相似度；

由于数据是二值化数据，只有0-1，因此新图像与原图像的相似度可以使用二分类问题进行计算，激活函数是sigmoid，采用交叉熵计算如下：$${\mathcal{L}}_1=\sum y_{i}p(x_i)+(1-y_i)p(x_i)$$
由于隐含空间 Z 任一维度我们假设其是服从正态分布的，从现有模型的计算来看，无论是 y 相同还是不相同的情况下，每一张图其推理出的隐含空间的每一维度都是不同的均值和方差的正态分布，这里由于我们有目标 y ，我们可以利用 y 的方式对隐含空间 Z 的维度分布进行固定，这样我们就可以生成想要的 y 对应的X，这属于一个优化部分，这里暂时不利用；

在这里我们利用 KL散度 方法进行计算，有$$D_{KL}(p||q)=\int p(x)\log \frac{p(x)}{q(x)}dx$$
其可以度量两个分布相似程度，越小越相似；结合这里的问题我们有：$${\mathcal{L}}_2=D_{KL}(q(z|x)||p(z))=\int q(z|x)\log \frac{q(z|x)}{p(z)}dz=\int q(z|x)\log q(z|x)dz-\int q(z|x)\log p(z)dz=E_{z\sim q(z|x)}[\log q(z|x)]-E_{z\sim q(z|x)}[\log p(z)]$$
由于直接计算期望是非常困难的，我们可以使用蒙特卡洛抽样的方法，那么最终的结果为：$${\mathcal{L}}_2=\frac{1}{n}\sum _{i=1}^n[\log q(z_i|x)-\log p(z_i)]$$
这里 $z_i$ 可以在 $q(z|x)$ 中用抽样获得，也就是 上述代码中 reparameterize 函数后的结果；这里由于假设 $p(z)$ 服从0-1正态分布，$q(z|x)$ 服从 mu-logvar 的正态分布，得到损失计算如下：

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

@tf.function
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, 0.0)
    logpz_x = log_normal_pdf(z, mean, logvar)
    return tf.reduce_mean(logpx_z+logpz_x-logpz)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

定义超参数，优化器，模型并开始训练：

@tf.function
def compute_apply_gradients(model, x, optimizer):
    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))
    return loss

# 定义超参数
epochs = 10
latent_dim = 50

# 保持随机向量恒定以进行生成，并画图可视化训练过程
random_vector_for_generation = tf.random.normal(shape=[16, latent_dim])
def generate_and_save_images(model, epoch, test_input):
    predictions = model(test_input)
    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')

    plt.savefig('image_at_epoch_{:04d}.png'.format(epoch))
    plt.show()

# 定义优化器以及模型
optimizer = tf.keras.optimizers.Adam(1e-4)
model = VAE(latent_dim)

for epoch in range(1, epochs + 1):
    start_time = time.time()
    for data in train_x:
        loss = compute_apply_gradients(model, data, optimizer)
    end_time = time.time()
    # 可视化训练过程
    generate_and_save_images(model, epoch, random_vector_for_generation)
    print('Epoch: {}, Test set ELBO: {}, time elapse for current epoch {}'.format(epoch, loss, end_time - start_time))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

得到训练过程如下：

Epoch: 1, Test set ELBO: 543.1875, time elapse for current epoch 7.332747459411621
Epoch: 2, Test set ELBO: 541.3853149414062, time elapse for current epoch 6.801848649978638
Epoch: 3, Test set ELBO: 533.6055297851562, time elapse for current epoch 7.25009298324585
Epoch: 4, Test set ELBO: 530.90087890625, time elapse for current epoch 6.932183742523193
Epoch: 5, Test set ELBO: 527.5391845703125, time elapse for current epoch 6.950530529022217
Epoch: 6, Test set ELBO: 527.8551025390625, time elapse for current epoch 6.936391830444336
Epoch: 7, Test set ELBO: 526.4480590820312, time elapse for current epoch 6.905040502548218
Epoch: 8, Test set ELBO: 526.1814575195312, time elapse for current epoch 6.9060218334198
Epoch: 9, Test set ELBO: 525.7576293945312, time elapse for current epoch 6.993067264556885
Epoch: 10, Test set ELBO: 524.8110961914062, time elapse for current epoch 6.8317344188690186
1
2
3
4
5
6
7
8
9
10

可视化图像如下：

生成对抗网络

可以想象一个名画伪造者想伪造一幅达芬奇的画作，开始时，伪造者技术不精，但他，将自己的一些赝品和达芬奇的作品混在一起，请一个艺术商人对每一幅画进行真实性评估，并向伪造者反馈，告诉他哪些看起来像真迹、哪些看起来不像真迹。 伪造者根据这些反馈，改进自己的赝品。随着时间的推移，伪造者技能越来越高，艺术商人也变得越来越擅长找出赝品。最后，他们手上就拥有了一些非常逼真的赝品。 这就是GAN的基本原理。这里有两个角色，一个是伪造者，另一个是技术鉴赏者。他们训练的目的都是打败对方。

因此，GAN从网络的角度来看，它由两部分组成。

1. 生成器网络：它一个潜在空间的随机向量作为输入，并将其解码为一张合成图像；
2. 判别器网络：以一张图像（真实的或合成的均可）作为输入，并预测该图像来自训练集还是来自生成器网络；

首先观察判别器，这里以 D 表示，很明显判别器执行的是一个二分类任务，其任务目标是判断输入的图像是真实的还是伪造的；

设定
D ( x ) = { 1 x   i s   t r u e 0 x   i s   f a l s e

$$\begin{align} D(x) = \begin{cases} 1 \quad x \,is \,true \\ 0 \quad x \, is \, false \end{cases} \end{align}$$ D(x)={1xistrue0xisfalse​​​
其损失容易得到$${\mathcal{L}}_D=-\sum D_i\log p(x_i)+(1-D_i)\log (1-p(x_i))$$
接着我们观察生成器，这里以 G 表示，利用噪声随机生成图像，$x_i=G(noise)$，再利用判别器判别，让其判别错误；相当于变向扩大判别器的损失，得到损失如下：$${\mathcal{L}}_G=\sum D_i\log p(x_i)+(1-D_i)\log (1-p(x_i))$$
这里可以发现 GAN 的优化过程不像通常的求损失函数的最小值，而是保持生成和判别两股力量的动态平衡，因此，其训练要比一般的神经网络难很多；

定义数据如下：

import tensorflow as tf
import matplotlib.pyplot as plt
import numpy as np

(x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()

# 在这里简单做个二值化处理
x_train, x_test = (x_train>0).astype(np.float32), (x_test>0).astype(np.float32)
train_x = tf.data.Dataset.from_tensor_slices(np.expand_dims(x_train, axis=-1)).batch(32)
test_x = tf.data.Dataset.from_tensor_slices(np.expand_dims(x_test, axis=-1)).batch(32)
1
2
3
4
5
6
7
8
9
10

定义模型如下：

class GAN(tf.keras.Model):
    def __init__(self, latent_dim):
        super(GAN, self).__init__()
        self.latent_dim = latent_dim
        self.discriminator = tf.keras.models.Sequential([
            tf.keras.layers.InputLayer(input_shape=[28, 28, 1]),
            tf.keras.layers.Conv2D(32, 3, strides=(2, 2), padding='same'),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Conv2D(64, 3, strides=(2, 2), padding='same'),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Flatten(),
            tf.keras.layers.Dense(1)
        ])
        self.generator = tf.keras.models.Sequential([
            tf.keras.layers.InputLayer(input_shape=[latent_dim]),
            tf.keras.layers.Dense(7*7*64),
            tf.keras.layers.BatchNormalization(),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Reshape(target_shape=[7,7,64]),
            tf.keras.layers.Conv2DTranspose(32, 5, strides=(2, 2), padding='same'),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Conv2DTranspose(16, 5, strides=(2, 2), padding='same'),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Conv2DTranspose(1, 5, strides=(1, 1), padding='same', activation='tanh'),
        ])
        
        # 定义交叉熵损失
        self.cross_entropy = tf.keras.losses.BinaryCrossentropy(from_logits=True)

        # 定义损失tracker
        self.gen_loss_tracker = tf.keras.metrics.Mean(name="generator_loss")
        self.disc_loss_tracker = tf.keras.metrics.Mean(name="discriminator_loss")

    def compile(self, d_optimizer, g_optimizer):
        super(GAN, self).compile()
        self.d_optimizer = d_optimizer
        self.g_optimizer = g_optimizer
        
    # 判别器损失
    def compute_discriminator_loss(self, real_output, fake_output):
        real_loss = self.cross_entropy(tf.ones_like(real_output), real_output)
        fake_loss = self.cross_entropy(tf.zeros_like(fake_output), fake_output)
        total_loss = real_loss + fake_loss
        return total_loss

    # 生成器损失
    def compute_generator_loss(self, fake_output):
        return self.cross_entropy(tf.ones_like(fake_output), fake_output)

    def train_step(self, data):
        x = data
        
        # 训练判别器
        with tf.GradientTape() as tape:
            x_noise = tf.random.normal(shape=[tf.shape(x)[0], self.latent_dim])
            x_ = self.generator(x_noise, training=False)
            real_output = self.discriminator(x, training=True)
            fake_output = self.discriminator(x_, training=True)
            discriminator_loss = self.compute_discriminator_loss(real_output, fake_output)
            
        grads = tape.gradient(discriminator_loss, self.discriminator.trainable_weights)
        self.d_optimizer.apply_gradients(zip(grads, self.discriminator.trainable_weights))

        # 训练生成器
        with tf.GradientTape() as tape:
            x_noise = tf.random.normal(shape=[tf.shape(x)[0], self.latent_dim])
            x_ = self.generator(x_noise, training=True)
            fake_output = self.discriminator(x_, training=False)
            generator_loss = self.compute_generator_loss(fake_output)

        grads = tape.gradient(generator_loss, self.generator.trainable_weights)
        self.g_optimizer.apply_gradients(zip(grads, self.generator.trainable_weights))

        # 保存损失
        self.disc_loss_tracker.update_state(discriminator_loss)
        self.gen_loss_tracker.update_state(generator_loss)
        
        return {
            "g_loss": self.gen_loss_tracker.result(),
            "d_loss": self.disc_loss_tracker.result(),
        }

    def call(self, x):
        return self.generator(x)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84

初始化模型以及损失：

latent_dim = 100
epoch = 100

model = GAN(latent_dim)
model.compile(
    d_optimizer=tf.keras.optimizers.Adam(learning_rate=1e-4),
    g_optimizer=tf.keras.optimizers.Adam(learning_rate=1e-4),
)
model.fit(train_x, epoch=epoch)
1
2
3
4
5
6
7
8
9

可视化图像如下：

VAE 和 GAN 的异同

VAE 适合学习具有良好结构的潜在空间，潜在空间有比较好的连续性，其中存在一些有特定意义的方向。VAE 能够捕捉图像的结构变化（倾斜角度，圈的位置，形状变化，表情变化等）。它具有显式的分布，能够容易地可视化图像的分布。

GAN 生成的潜在空间可能没有良好结构，但 GAN 生成的图像一般比 VAE 生成的图像更清晰；

1. 生成样本的方式：

GAN：GAN采用对抗训练的方式生成样本。生成器网络通过学习从潜在空间到数据空间的映射关系，生成与真实样本相似的样本。判别器网络则尝试区分生成的样本和真实样本。生成器和判别器通过对抗的方式相互学习，使生成器生成更真实的样本，同时使判别器更准确地区分真假样本。

 VAE：VAE通过变分推断生成样本。编码器网络将输入样本映射到潜在空间中的潜在变量表示，解码器网络则将潜在变量映射回样本空间生成样本。VAE的训练过程通过最大化观测数据的边际概率来实现，同时最小化潜在空间中的潜在变量与先验分布之间的差异。

2. 训练方法的差异：

GAN：GAN的训练过程是通过生成器和判别器之间的对抗学习来进行的。生成器的目标是尽量生成逼真的样本，而判别器的目标是尽量准确地区分真假样本。这种对抗学习的过程使得生成器和判别器能够相互促进，从而提高生成器的生成能力。
VAE：VAE的训练过程是通过最大化观测数据的边际概率来实现的。它使用了一种重参数化技巧，将潜在变量的采样过程转化为可微分的操作，从而可以使用梯度下降方法进行优化。


3. 生成样本的特点：

GAN：GAN生成的样本通常非常逼真，并且具有良好的样本多样性和细节表达能力。它可以生成各种逼真的图像、音频和文本等。
VAE：VAE生成的样本相对较为平滑，更加控制和可预测。它可以用于生成具有连续变化的样本，如图像的插值和变形。

综上所述，GAN和VAE在生成模型领域具有不同的特点和应用。GAN通过对抗学习生成逼真多样的样本，而VAE通过变分推断生成可控制和可预测的样本。它们在训练方式、生成样本的特点以及应用场景上都有所区别，因此在具体任务中，选择适合的模型取决于需求和目标。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

条件式生成对抗网络

为了让 GAN 生成我们想要的目标而不是随机的，我们可以在判别器中将 $x$ 和 $y$ 放在一起进行输入判别；如图所示：

首先观察判别器，这里以 D 表示，很明显判别器同样执行的是一个二分类任务，其任务目标是判断输入的图像在给定 $y$ 的情况下是真实的还是伪造的；

设定
D ( x ∣ y ) = { 1 x   i s   t r u e 0 x   i s   f a l s e

$$\begin{align} D(x|y) = \begin{cases} 1 \quad x \,is \,true \\ 0 \quad x \, is \, false \end{cases} \end{align}$$ D(x∣y)={1xistrue0xisfalse​​​
其损失容易和 GAN 一样，有：$${\mathcal{L}}_D=-\sum D_i\log p(x_i)+(1-D_i)\log (1-p(x_i))$$
接着我们观察生成器，这里以 G 表示，如图：

同样的，我们需要给一个限制 $y$ ，在限制下利用噪声随机生成图像，$x_i=G(noise|y_i)$，再次结合 $y_i$ 利用判别器判别，让其判别错误；相当于变向扩大判别器的损失，得到损失如下：$${\mathcal{L}}_G=\sum D_i\log p(x_i)+(1-D_i)\log (1-p(x_i))$$
如何将标签放入数据进行训练？ （条件生成对抗网络——cGAN原理与代码 - 知乎 (zhihu.com)）

大家可能会疑惑，这个有必要讲么？标签数据直接放在输入中就可以了，多么简单的事情！其实不然，在深度学习当中，不同的预测任务可能有着不同的标签，且标签的形态可能因任务而改变（例如，分类任务的标签往往是一串序列，而分割任务的标签则是一张图像）；同样，不同的预测任务可能有着不同的特征，特征的形态与架构的形态息息相关，当架构确认之后，其他形态的特征将无法输入架构（比如，我们无法将序列输入到CNN中，也无法将图像输入到DNN中）。因此，想要将真实标签作为特征的一部分输入到生成器和判别器来训练，我们必须要将真实标签转换为当前架构可以接受的形态，更具体地来说，转变为与输入架构的特征完全一致的形态。通常来说，常见的有如下4类情况：

特征为二维表，标签为序列的数据集

对于这种数据，只需要将标签作为新的特征加入到原始特征中的最后一列即可，相当于原始特征有n个，新的特征为n+1个。

特征标签都为图像的数据集

对于这种数据，也比较简单，只需要在通道的维度，对特征与标签进行合并即可。

特征为二维表，标签为序列的数据集，但架构是卷积层

由于模型架构是卷积层，因此没办法将二维的东西作为输入，因此，需要将特征及标签映射为卷积可接受的四维输入。

特征为图像，标签为序列的数据集

可以将每个标签映射为与特征一样的维度，然后在通道维度上进行合并，先看结论，后面会细讲。

这里同样适用 MNIST 数据集，过程如下：

定义数据如下：

import tensorflow as tf
import matplotlib.pyplot as plt
import numpy as np

(x_train, y_train), (x_test, y_test) = tf.keras.datasets.mnist.load_data()

# 在这里简单做个二值化处理
x_train, x_test = (x_train>0).astype(np.float32), (x_test>0).astype(np.float32)
y_train, y_test = tf.keras.utils.to_categorical(y_train, num_classes=10), tf.keras.utils.to_categorical(y_test, num_classes=10)
train_x = tf.data.Dataset.from_tensor_slices(np.expand_dims(x_train, axis=-1)).batch(32)
test_x = tf.data.Dataset.from_tensor_slices(np.expand_dims(x_test, axis=-1)).batch(32)
1
2
3
4
5
6
7
8
9
10
11

定义模型如下：

class CGAN(tf.keras.Model):
    def __init__(self, latent_dim):
        super(CGAN, self).__init__()
        self.latent_dim = latent_dim
        self.discriminator = tf.keras.models.Sequential([
            tf.keras.layers.InputLayer(input_shape=[28, 28, 1 + 10]),
            tf.keras.layers.Conv2D(32, 3, strides=(2, 2), padding='same'),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Conv2D(64, 3, strides=(2, 2), padding='same'),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Flatten(),
            tf.keras.layers.Dense(1)
        ])
        self.generator = tf.keras.models.Sequential([
            tf.keras.layers.InputLayer(input_shape=[latent_dim + 10]),
            tf.keras.layers.Dense(7*7*64),
            tf.keras.layers.BatchNormalization(),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Reshape(target_shape=[7,7,64]),
            tf.keras.layers.Conv2DTranspose(32, 4, strides=(2, 2), padding='same'),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Conv2DTranspose(16, 4, strides=(2, 2), padding='same'),
            tf.keras.layers.LeakyReLU(),
            tf.keras.layers.Conv2DTranspose(1, 3, strides=(1, 1), padding='same', activation='tanh'),
        ])
        
        # 定义交叉熵损失
        self.cross_entropy = tf.keras.losses.BinaryCrossentropy(from_logits=True)

        # 定义损失tracker
        self.gen_loss_tracker = tf.keras.metrics.Mean(name="generator_loss")
        self.disc_loss_tracker = tf.keras.metrics.Mean(name="discriminator_loss")

    def compile(self, d_optimizer, g_optimizer):
        super(CGAN, self).compile()
        self.d_optimizer = d_optimizer
        self.g_optimizer = g_optimizer
        
    # 判别器损失
    def compute_discriminator_loss(self, real_output, fake_output):
        real_loss = self.cross_entropy(tf.ones_like(real_output), real_output)
        fake_loss = self.cross_entropy(tf.zeros_like(fake_output), fake_output)
        total_loss = real_loss + fake_loss
        return total_loss

    # 生成器损失
    def compute_generator_loss(self, fake_output):
        return self.cross_entropy(tf.ones_like(fake_output), fake_output)

    def train_step(self, data):
    
        x, y = data

        bath_size = tf.shape(x)[0]
        width = tf.shape(x)[1]
        height = tf.shape(x)[2]
        
        x = tf.expand_dims(x, axis=-1)
        y_ = tf.cast(y, dtype=x.dtype)
        y_d = tf.reshape(tf.repeat(y_, [width*height]), [bath_size, width, height, -1])
        
        # 训练判别器
        with tf.GradientTape() as tape:
            x_noise = tf.random.normal(shape=[bath_size, self.latent_dim])
            x_noise = tf.concat([x_noise, y_], axis=-1)
            x_ = self.generator(x_noise, training=False)
            real_output = self.discriminator(tf.concat([x, y_d], axis=-1), training=True)
            fake_output = self.discriminator(tf.concat([x_, y_d], axis=-1), training=True)
            discriminator_loss = self.compute_discriminator_loss(real_output, fake_output)
            
        grads = tape.gradient(discriminator_loss, self.discriminator.trainable_weights)
        self.d_optimizer.apply_gradients(zip(grads, self.discriminator.trainable_weights))

        # 训练生成器
        with tf.GradientTape() as tape:
            x_noise = tf.random.normal(shape=[bath_size, self.latent_dim])
            x_noise = tf.concat([x_noise, y_], axis=-1)
            x_ = self.generator(x_noise, training=True)
            fake_output = self.discriminator(tf.concat([x_, y_d], axis=-1), training=True)
            generator_loss = self.compute_generator_loss(fake_output)

        grads = tape.gradient(generator_loss, self.generator.trainable_weights)
        self.g_optimizer.apply_gradients(zip(grads, self.generator.trainable_weights))

        # 保存损失
        self.disc_loss_tracker.update_state(discriminator_loss)
        self.gen_loss_tracker.update_state(generator_loss)
        
        return {
            "g_loss": self.gen_loss_tracker.result(),
            "d_loss": self.disc_loss_tracker.result(),
        }

    def call(self, x):
        return self.generator(x)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

初始化模型以及损失：

latent_dim = 100
epoch = 100

model = CGAN(latent_dim)
model.compile(
    d_optimizer=tf.keras.optimizers.Adam(),
    g_optimizer=tf.keras.optimizers.Adam(),
)
model.fit(train_x, epoch=epoch)
1
2
3
4
5
6
7
8
9

可视化图像如下：

提升 GAN 训练效果的一些技巧

1. 批量加载和批规范化，有利于提升训练过程中博弈的稳定性；
2. 使用 tanh 激活函数作为生成器的最后一层，将图像数据规范在 -1 和 1 之间，一般不使用 sigmoid；
3. 选用 Leaky-ReLU 作为生成器和判别器的激活函数，有利于改善梯度的稀疏性，稀疏的梯度会妨碍 GAN 的训练；
4. 使用卷积层时，考虑卷积核的大小能被步幅整除，否则，可能导致生成的图像中存在棋盘状伪影；

参考文献

一文搞懂所有 VAE 模型（4个AE+12个VAE原理汇总） - 知乎 (zhihu.com)

一文理解变分自编码器（VAE） - 知乎 (zhihu.com)

证据下界（ELBO）的概念解析 | 西山晴雪的知识笔记 (xishansnow.github.io)

变分推断中的ELBO(证据下界)-CSDN博客

条件生成对抗网络——cGAN原理与代码 - 知乎 (zhihu.com)