WGAN-GP 重写 Model.train_step

import os

os.environ["KERAS_BACKEND"] = "tensorflow"

import keras
import tensorflow as tf
from keras import layers

IMG_SHAPE = (28, 28, 1)
BATCH_SIZE = 512

# 噪声向量的大小
noise_dim = 128

fashion_mnist = keras.datasets.fashion_mnist
(train_images, train_labels), (test_images, test_labels) = fashion_mnist.load_data()
print(f"示例数量: {len(train_images)}")
print(f"数据集中图像的形状: {train_images.shape[1:]}")

# 将每个样本重塑为 (28, 28, 1) 并将像素值规范化到 [-1, 1] 范围
train_images = train_images.reshape(train_images.shape[0], *IMG_SHAPE).astype("float32")
train_images = (train_images - 127.5) / 127.5

def conv_block(
    x,
    filters,
    activation,
    kernel_size=(3, 3),
    strides=(1, 1),
    padding="same",
    use_bias=True,
    use_bn=False,
    use_dropout=False,
    drop_value=0.5,
):
    x = layers.Conv2D(
        filters, kernel_size, strides=strides, padding=padding, use_bias=use_bias
    )(x)
    if use_bn:
        x = layers.BatchNormalization()(x)
    x = activation(x)
    if use_dropout:
        x = layers.Dropout(drop_value)(x)
    return x


def get_discriminator_model():
    img_input = layers.Input(shape=IMG_SHAPE)
    # 对输入进行零填充，使输入图像大小为 (32, 32, 1)。
    x = layers.ZeroPadding2D((2, 2))(img_input)
    x = conv_block(
        x,
        64,
        kernel_size=(5, 5),
        strides=(2, 2),
        use_bn=False,
        use_bias=True,
        activation=layers.LeakyReLU(0.2),
        use_dropout=False,
        drop_value=0.3,
    )
    x = conv_block(
        x,
        128,
        kernel_size=(5, 5),
        strides=(2, 2),
        use_bn=False,
        activation=layers.LeakyReLU(0.2),
        use_bias=True,
        use_dropout=True,
        drop_value=0.3,
    )
    x = conv_block(
        x,
        256,
        kernel_size=(5, 5),
        strides=(2, 2),
        use_bn=False,
        activation=layers.LeakyReLU(0.2),
        use_bias=True,
        use_dropout=True,
        drop_value=0.3,
    )
    x = conv_block(
        x,
        512,
        kernel_size=(5, 5),
        strides=(2, 2),
        use_bn=False,
        activation=layers.LeakyReLU(0.2),
        use_bias=True,
        use_dropout=False,
        drop_value=0.3,
    )

    x = layers.Flatten()(x)
    x = layers.Dropout(0.2)(x)
    x = layers.Dense(1)(x)

    d_model = keras.models.Model(img_input, x, name="discriminator")
    return d_model


d_model = get_discriminator_model()
d_model.summary()

def upsample_block(
    x,
    filters,
    activation,
    kernel_size=(3, 3),
    strides=(1, 1),
    up_size=(2, 2),
    padding="same",
    use_bn=False,
    use_bias=True,
    use_dropout=False,
    drop_value=0.3,
):
    x = layers.UpSampling2D(up_size)(x)
    x = layers.Conv2D(
        filters, kernel_size, strides=strides, padding=padding, use_bias=use_bias
    )(x)

    if use_bn:
        x = layers.BatchNormalization()(x)

    if activation:
        x = activation(x)
    if use_dropout:
        x = layers.Dropout(drop_value)(x)
    return x


def get_generator_model():
    noise = layers.Input(shape=(noise_dim,))
    x = layers.Dense(4 * 4 * 256, use_bias=False)(noise)
    x = layers.BatchNormalization()(x)
    x = layers.LeakyReLU(0.2)(x)

    x = layers.Reshape((4, 4, 256))(x)
    x = upsample_block(
        x,
        128,
        layers.LeakyReLU(0.2),
        strides=(1, 1),
        use_bias=False,
        use_bn=True,
        padding="same",
        use_dropout=False,
    )
    x = upsample_block(
        x,
        64,
        layers.LeakyReLU(0.2),
        strides=(1, 1),
        use_bias=False,
        use_bn=True,
        padding="same",
        use_dropout=False,
    )
    x = upsample_block(
        x, 1, layers.Activation("tanh"), strides=(1, 1), use_bias=False, use_bn=True
    )
    # At this point, we have an output which has the same shape as the input, (32, 32, 1).
    # We will use a Cropping2D layer to make it (28, 28, 1).
    x = layers.Cropping2D((2, 2))(x)

    g_model = keras.models.Model(noise, x, name="generator")
    return g_model


g_model = get_generator_model()
g_model.summary()

class WGAN(keras.Model):
    def __init__(
        self,
        discriminator,
        generator,
        latent_dim,
        discriminator_extra_steps=3,
        gp_weight=10.0,
    ):
        super().__init__()
        self.discriminator = discriminator
        self.generator = generator
        self.latent_dim = latent_dim
        self.d_steps = discriminator_extra_steps
        self.gp_weight = gp_weight

    def compile(self, d_optimizer, g_optimizer, d_loss_fn, g_loss_fn):
        super().compile()
        self.d_optimizer = d_optimizer
        self.g_optimizer = g_optimizer
        self.d_loss_fn = d_loss_fn
        self.g_loss_fn = g_loss_fn

    def gradient_penalty(self, batch_size, real_images, fake_images):
        """计算梯度惩罚。

        此损失是在插值图像上计算的
        并添加到鉴别器损失中。
        """
        # 获取插值图像
        alpha = tf.random.uniform([batch_size, 1, 1, 1], 0.0, 1.0)
        diff = fake_images - real_images
        interpolated = real_images + alpha * diff

        with tf.GradientTape() as gp_tape:
            gp_tape.watch(interpolated)
            # 1. 获取此插值图像的鉴别器输出。
            pred = self.discriminator(interpolated, training=True)

        # 2. 计算该插值图像的梯度。
        grads = gp_tape.gradient(pred, [interpolated])[0]
        # 3. 计算梯度的范数。
        norm = tf.sqrt(tf.reduce_sum(tf.square(grads), axis=[1, 2, 3]))
        gp = tf.reduce_mean((norm - 1.0) ** 2)
        return gp

    def train_step(self, real_images):
        if isinstance(real_images, tuple):
            real_images = real_images[0]

        # 获取批量大小
        batch_size = tf.shape(real_images)[0]

        # 对于每个批次，我们将按照原始论文中列出的步骤执行：
        # 1. 训练生成器并获取生成器损失
        # 2. 训练鉴别器并获取鉴别器损失
        # 3. 计算梯度惩罚
        # 4. 用一个常数权重因子乘以这个梯度惩罚
        # 5. 将梯度惩罚添加到鉴别器损失中
        # 6. 将生成器和鉴别器损失作为损失字典返回

        # 首先训练鉴别器。原始论文建议将
        # 鉴别器训练 `x` 次步骤（通常为 5），
        # 相对于生成器的一次步骤。在这里，我们将训练
        # 3 次额外步骤，相对于 5 次，以减少训练时间。
        for i in range(self.d_steps):
            # 获取潜在向量
            random_latent_vectors = tf.random.normal(
                shape=(batch_size, self.latent_dim)
            )
            with tf.GradientTape() as tape:
                # 从潜在向量生成假图像
                fake_images = self.generator(random_latent_vectors, training=True)
                # 获取假图像的 logits
                fake_logits = self.discriminator(fake_images, training=True)
                # 获取真实图像的 logits
                real_logits = self.discriminator(real_images, training=True)

                # 使用假图像和真实图像的 logits 计算鉴别器损失
                d_cost = self.d_loss_fn(real_img=real_logits, fake_img=fake_logits)
                # 计算梯度惩罚
                gp = self.gradient_penalty(batch_size, real_images, fake_images)
                # 将梯度惩罚添加到原始鉴别器损失
                d_loss = d_cost + gp * self.gp_weight

            # 获取相对于鉴别器损失的梯度
            d_gradient = tape.gradient(d_loss, self.discriminator.trainable_variables)
            # 使用鉴别器优化器更新鉴别器的权重
            self.d_optimizer.apply_gradients(
                zip(d_gradient, self.discriminator.trainable_variables)
            )

        # 训练生成器
        # 获取潜在向量
        random_latent_vectors = tf.random.normal(shape=(batch_size, self.latent_dim))
        with tf.GradientTape() as tape:
            # 使用生成器生成假图像
            generated_images = self.generator(random_latent_vectors, training=True)
            # 获取假图像的鉴别器 logits
            gen_img_logits = self.discriminator(generated_images, training=True)
            # 计算生成器损失
            g_loss = self.g_loss_fn(gen_img_logits)

        # 获取相对于生成器损失的梯度
        gen_gradient = tape.gradient(g_loss, self.generator.trainable_variables)
        # 使用生成器优化器更新生成器的权重
        self.g_optimizer.apply_gradients(
            zip(gen_gradient, self.generator.trainable_variables)
        )
        return {"d_loss": d_loss, "g_loss": g_loss}

class GANMonitor(keras.callbacks.Callback):
    def __init__(self, num_img=6, latent_dim=128):
        self.num_img = num_img
        self.latent_dim = latent_dim

    def on_epoch_end(self, epoch, logs=None):
        random_latent_vectors = tf.random.normal(shape=(self.num_img, self.latent_dim))
        generated_images = self.model.generator(random_latent_vectors)
        generated_images = (generated_images * 127.5) + 127.5

        for i in range(self.num_img):
            img = generated_images[i].numpy()
            img = keras.utils.array_to_img(img)
            img.save("generated_img_{i}_{epoch}.png".format(i=i, epoch=epoch))

# 为两个网络实例化优化器
# （建议使用 learning_rate=0.0002，beta_1=0.5）
generator_optimizer = keras.optimizers.Adam(
    learning_rate=0.0002, beta_1=0.5, beta_2=0.9
)
discriminator_optimizer = keras.optimizers.Adam(
    learning_rate=0.0002, beta_1=0.5, beta_2=0.9
)


# 定义判别器的损失函数，
# 应该是 (假图像损失 - 真实图像损失)。
# 我们稍后将为这个损失函数添加梯度惩罚。
def discriminator_loss(real_img, fake_img):
    real_loss = tf.reduce_mean(real_img)
    fake_loss = tf.reduce_mean(fake_img)
    return fake_loss - real_loss


# 定义生成器的损失函数。
def generator_loss(fake_img):
    return -tf.reduce_mean(fake_img)


# 设置训练的纪元数量。
epochs = 20

# 实例化客户 `GANMonitor` Keras 回调。
cbk = GANMonitor(num_img=3, latent_dim=noise_dim)

# 获取 wgan 模型
wgan = WGAN(
    discriminator=d_model,
    generator=g_model,
    latent_dim=noise_dim,
    discriminator_extra_steps=3,
)

# 编译 wgan 模型
wgan.compile(
    d_optimizer=discriminator_optimizer,
    g_optimizer=generator_optimizer,
    g_loss_fn=generator_loss,
    d_loss_fn=discriminator_loss,
)

# 开始训练
wgan.fit(train_images, batch_size=BATCH_SIZE, epochs=epochs, callbacks=[cbk])

from IPython.display import Image, display

display(Image("generated_img_0_19.png"))
display(Image("generated_img_1_19.png"))
display(Image("generated_img_2_19.png"))