简介：这份资源是CycleGAN的Keras实现代码包，面向具备一定深度学习基础、希望动手实践无监督图像到图像转换的开发者与研究者。它围绕生成对抗网络展开，重点解决无配对样本条件下的跨域图像转换问题，可应用于风格迁移、季节转换等场景。压缩包共7个文件，包含4个Python脚本与3个数据压缩包，整体约477.41MB；脚本分别承担模型构建、数据加载、预测推理与生成器骨干网络等职责，数据包则提供马与斑马、苹果与橘子、莫奈与照片等经典非配对图像集，便于直接开展训练与效果验证。目前已有548人学习下载。通过阅读代码，读者可以理解双生成器与双判别器的协作方式、循环一致性损失的实现细节，以及对抗损失与循环损失的联合优化流程，并借助预测脚本将训练好的模型用于新图像的风格转换，适合作为入门CycleGAN并复现实验的参考工程。1. 一份能跑通的 CycleGAN Keras 实现，到底解决了谁的燃眉之急手里有一批成对的图像数据，想训练一个风格迁移模型，结果发现配对标注成本高得离谱——这种场景做图像生成的人多半都遇到过。CycleGAN 的价值就在于绕开了成对样本的硬性要求：它用两组非配对图像，靠循环一致性损失把 A 域和 B 域之间的映射约束住，让马变斑马、照片变油画这类任务不再依赖一一对应的训练集。而这份 Keras 实现，把生成器、判别器、循环损失、身份损失、训练循环全部用 Keras 的层和模型 API 搭出来，没有藏着掖着的自定义算子，适合想真正读懂 CycleGAN 每一行在干什么、而不是调个库就完事的人。如果你已经会用 Keras 搭普通 CNN，但对 GAN 的训练流程还停留在“看论文知道大概”的阶段，这份代码就是拿来逐行拆解的好材料。2. CycleGAN 的核心机制与 Keras 实现选型：为什么不用 PyTorch 版本2.1 循环一致性损失到底在约束什么CycleGAN 要解决的核心问题是：没有成对数据，怎么保证 G(A)→B 生成的图像既像 B 域，又能保留 A 域的内容。答案就是两个方向的生成器加两个判别器，再叠加循环一致性损失。具体来说，设 G 负责 A→B，F 负责 B→A，那么 F(G(a)) 应该约等于 a，G(F(b)) 应该约等于 b。这个约束让生成器不能随便把输入映射到 B 域里一个无关的样本上，否则循环回来就对不上。在 Keras 里实现这个逻辑，关键是把 G、F、D_A、D_B 四个模型分别定义好，然后用tf.GradientTape或者train_on_batch手动控制训练步骤。常见做法是写一个CycleGANTrainer类，在train_step里依次计算：生成器总损失 = 对抗损失 + λ₁ × 循环一致性损失 + λ₂ × 身份损失判别器损失 = 两个域各自判别真伪的交叉熵之和其中 λ₁ 默认取 10，λ₂ 默认取 5 或 10，这两个系数直接决定生成图像是更偏向目标域风格还是更保留原始内容。调参时如果发现生成结果“风格化过头、内容全丢”，优先把 λ₁ 调大；如果风格迁移效果太弱，则适当降低 λ₁ 或提高 λ₂。2.2 Keras 实现相比 PyTorch 版本的取舍PyTorch 在学术界更流行，动态图写起来灵活，调试也直观。但这份 Keras 实现有它自己的适用场景：如果你的部署环境是 TensorFlow Serving 或者 TFLite，Keras 模型导出和量化链路更顺；如果你团队里大部分人用 Keras 做常规图像任务，引入 PyTorch 会增加维护成本。另外 Keras 的Model子类化写法虽然不如 PyTorch 的forward直观，但一旦把call方法写清楚，后续做模型剪枝、转 ONNX 都更方便。不过要提前说清楚边界：Keras 实现在自定义训练循环里对梯度裁剪、学习率调度这些细节的控制不如 PyTorch 灵活，遇到训练不稳定时排查手段会少一些。所以这份代码更适合作为学习模板和中小规模实验的起点，而不是直接拿去跑大规模生产训练。2.3 环境准备与依赖版本在动手之前，先把环境对齐。这份实现基于 TensorFlow 2.x，建议用 2.10 以上的版本，Python 3.8 到 3.10 之间比较稳。依赖不多，核心就是 tensorflow、numpy、Pillow 和 matplotlib（可视化用）。# 创建虚拟环境并安装依赖 python -m venv cyclegan_env source cyclegan_env/bin/activate # Windows 用 cyclegan_env\Scripts\activate pip install tensorflow==2.12.0 numpy pillow matplotlib这里固定 TensorFlow 版本是有原因的：2.12 之后 Keras 被拆成独立包，部分 API 路径变了，如果直接用最新版可能会遇到tf.keras和keras混用报错。如果你已经装了其他版本，先pip show tensorflow确认一下，避免后面调试半天发现是版本问题。提示：如果机器上有 GPU，装tensorflow-gpu或者确认tf.config.list_physical_devices('GPU')能返回设备列表。CycleGAN 训练对显存有一定要求，256×256 分辨率下 batch size 设 1 也可能占到 6GB 以上。3. 从零搭建生成器与判别器：Keras 层怎么堆才不翻车3.1 生成器结构：编码器-转换器-解码器三段式CycleGAN 的生成器本质是一个带残差连接的编解码网络。输入一张 256×256×3 的图像，先经过两层步长为 2 的卷积下采样到 64×64，再经过若干残差块做特征变换，最后用转置卷积上采样回原尺寸。Keras 里用Conv2D、BatchNormalization、Activation和Add就能拼出来。import tensorflow as tf from tensorflow.keras import layers, Model def build_generator(input_shape=(256, 256, 3), n_res_blocks=6): """构建 CycleGAN 生成器：下采样 -> 残差块 -> 上采样""" inputs = layers.Input(shape=input_shape) # 初始卷积 x = layers.Conv2D(64, 7, strides=1, padding='same')(inputs) x = layers.BatchNormalization()(x) x = layers.Activation('relu')(x) # 下采样两层 for filters in [128, 256]: x = layers.Conv2D(filters, 3, strides=2, padding='same')(x) x = layers.BatchNormalization()(x) x = layers.Activation('relu')(x) # 残差块 for _ in range(n_res_blocks): res = layers.Conv2D(256, 3, padding='same')(x) res = layers.BatchNormalization()(res) res = layers.Activation('relu')(res) res = layers.Conv2D(256, 3, padding='same')(res) res = layers.BatchNormalization()(res) x = layers.Add()([x, res]) # 上采样两层 for filters in [128, 64]: x = layers.Conv2DTranspose(filters, 3, strides=2, padding='same')(x) x = layers.BatchNormalization()(x) x = layers.Activation('relu')(x) # 输出层，tanh 把像素值压到 [-1, 1] outputs = layers.Conv2D(3, 7, padding='same', activation='tanh')(x) return Model(inputs, outputs, name='generator')这段代码里几个参数值得注意：n_res_blocks默认 6，对应 256×256 输入；如果降到 128×128，可以减到 3 或 4，否则参数量偏大容易过拟合。输出层用tanh而不是sigmoid，是因为训练时图像归一化到 [-1, 1] 区间，判别器输入也按这个范围处理，保持一致。残差块里没有加激活函数在Add之后，这是原论文的做法，实测影响不大但保持一致性更好。3.2 判别器结构：PatchGAN 为什么比全图判别更稳判别器用的是 PatchGAN，输出不是一个标量，而是一个 N×N 的 patch 矩阵，每个值代表原图一个感受野区域的真假。这样做的好处是参数量小、训练更稳定，而且能捕捉局部纹理细节。Keras 实现起来就是一个几层卷积堆到输出通道为 1 的Conv2D。def build_discriminator(input_shape=(256, 256, 3)): """PatchGAN 判别器，输出 patch 级别的真假评分""" inputs = layers.Input(shape=input_shape) x = layers.Conv2D(64, 4, strides=2, padding='same')(inputs) x = layers.LeakyReLU(0.2)(x) for filters in [128, 256, 512]: x = layers.Conv2D(filters, 4, strides=2, padding='same')(x) x = layers.BatchNormalization()(x) x = layers.LeakyReLU(0.2)(x) # 输出单通道 patch 评分 outputs = layers.Conv2D(1, 4, padding='same')(x) return Model(inputs, outputs, name='discriminator')判别器里用LeakyReLU(0.2)而不是普通 ReLU，是为了避免梯度稀疏。注意最后一层没有加 sigmoid，因为损失函数里用的是BinaryCrossentropy(from_logits=True)，这样数值更稳定。如果你习惯用sigmoid加普通交叉熵，训练初期容易遇到梯度消失，生成器更新不动。3.3 损失函数与训练循环的 Keras 写法把模型搭好之后，核心工作就是写训练循环。Keras 里可以用tf.GradientTape手动求梯度，也可以用train_on_batch，但后者对多模型交替训练的支持不够直观，所以常见做法是自定义train_step。class CycleGANTrainer(tf.keras.Model): def __init__(self, gen_ab, gen_ba, disc_a, disc_b, lambda_cycle=10.0, lambda_id=5.0): super().__init__() self.gen_ab = gen_ab self.gen_ba = gen_ba self.disc_a = disc_a self.disc_b = disc_b self.lambda_cycle = lambda_cycle self.lambda_id = lambda_id self.bce = tf.keras.losses.BinaryCrossentropy(from_logits=True) def train_step(self, data): real_a, real_b = data with tf.GradientTape(persistent=True) as tape: # 生成器前向 fake_b = self.gen_ab(real_a, training=True) fake_a = self.gen_ba(real_b, training=True) cycle_a = self.gen_ba(fake_b, training=True) cycle_b = self.gen_ab(fake_a, training=True) id_a = self.gen_ba(real_a, training=True) id_b = self.gen_ab(real_b, training=True) # 对抗损失 disc_real_a = self.disc_a(real_a, training=True) disc_fake_a = self.disc_a(fake_a, training=True) disc_real_b = self.disc_b(real_b, training=True) disc_fake_b = self.disc_b(fake_b, training=True) gen_ab_loss = self.bce(tf.ones_like(disc_fake_b), disc_fake_b) gen_ba_loss = self.bce(tf.ones_like(disc_fake_a), disc_fake_a) cycle_loss = tf.reduce_mean(tf.abs(real_a - cycle_a)) + \ tf.reduce_mean(tf.abs(real_b - cycle_b)) id_loss = tf.reduce_mean(tf.abs(real_a - id_a)) + \ tf.reduce_mean(tf.abs(real_b - id_b)) total_gen_loss = gen_ab_loss + gen_ba_loss + \ self.lambda_cycle * cycle_loss + self.lambda_id * id_loss disc_a_loss = self.bce(tf.ones_like(disc_real_a), disc_real_a) + \ self.bce(tf.zeros_like(disc_fake_a), disc_fake_a) disc_b_loss = self.bce(tf.ones_like(disc_real_b), disc_real_b) + \ self.bce(tf.zeros_like(disc_fake_b), disc_fake_b) # 分别更新生成器和判别器 gen_vars = self.gen_ab.trainable_variables + self.gen_ba.trainable_variables disc_vars = self.disc_a.trainable_variables + self.disc_b.trainable_variables self.optimizer.apply_gradients(zip(tape.gradient(total_gen_loss, gen_vars), gen_vars)) self.optimizer.apply_gradients(zip(tape.gradient(disc_a_loss, disc_vars), disc_vars)) self.optimizer.apply_gradients(zip(tape.gradient(disc_b_loss, disc_vars), disc_vars)) return {'gen_loss': total_gen_loss, 'disc_a_loss': disc_a_loss, 'disc_b_loss': disc_b_loss}这段代码里persistent=True是必须的，因为要对多个损失分别求梯度，普通 GradientTape 求完一次就释放了。lambda_cycle和lambda_id直接决定训练行为，前面已经说过调参方向。另外注意判别器更新了两次（D_A 和 D_B 各一次），但用的是同一个优化器，实际项目里也可以给判别器单独配一个优化器，学习率设成生成器的一半，训练会更稳。注意：如果显存不够，把train_step里的persistent=True改成分别用两个 tape 写，虽然代码长一点但能省显存。另外tf.abs算 L1 损失时注意张量形状要一致，循环回来的图像尺寸必须和原图完全对齐，否则会广播报错。4. 数据管道与训练调参：非配对图像怎么喂进去4.1 非配对数据集的加载与增强CycleGAN 的训练数据是两个独立的文件夹，比如trainA和trainB，里面各自放一堆图片，不需要文件名对应。用tf.data.Dataset分别读取两个目录，做随机裁剪、翻转、归一化，然后 zip 到一起。import os def load_image(path, size=256): img = tf.io.read_file(path) img = tf.image.decode_jpeg(img, channels=3) img = tf.image.resize(img, [size, size]) img = (tf.cast(img, tf.float32) / 127.5) - 1.0 # 归一化到 [-1, 1] return img def build_dataset(dir_a, dir_b, batch_size=1, size=256): files_a = [os.path.join(dir_a, f) for f in os.listdir(dir_a) if f.endswith(('.jpg', '.png'))] files_b = [os.path.join(dir_b, f) for f in os.listdir(dir_b) if f.endswith(('.jpg', '.png'))] ds_a = tf.data.Dataset.from_tensor_slices(files_a) ds_a = ds_a.map(lambda x: load_image(x, size), num_parallel_calls=tf.data.AUTOTUNE) ds_a = ds_a.cache().shuffle(len(files_a)).batch(batch_size).repeat() ds_b = tf.data.Dataset.from_tensor_slices(files_b) ds_b = ds_b.map(lambda x: load_image(x, size), num_parallel_calls=tf.data.AUTOTUNE) ds_b = ds_b.cache().shuffle(len(files_b)).batch(batch_size).repeat() return tf.data.Dataset.zip((ds_a, ds_b)).prefetch(tf.data.AUTOTUNE)这里batch_size默认设 1，是因为 CycleGAN 对 batch 内样本的独立性要求不高，而且显存占用大。如果显存充裕可以设 2 或 4，但要注意学习率也要相应调整。shuffle的 buffer 设成文件总数，保证每个 epoch 顺序不同。repeat()让数据集无限循环，训练步数由外部控制。4.2 学习率、优化器与训练轮次设置优化器用 Adam，生成器和判别器学习率都设 2e-4，beta_1 设 0.5。这个配置来自原论文，实测在多数风格迁移任务上都能收敛。训练轮次一般 100 到 200 个 epoch 起步，前 50 个 epoch 保持学习率不变，之后线性衰减到 0。# 优化器配置 gen_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5) disc_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5) # 学习率衰减回调（如果不用自定义 train_step 可以用这个） def lr_schedule(epoch, lr): if epoch < 50: return lr return lr * (1 - (epoch - 50) / 150)如果发现训练到 30 个 epoch 左右生成图像开始出现网格状伪影，多半是判别器太强了，把判别器学习率降到 1e-4 或者给生成器损失里的对抗部分加个权重系数。另一个常见问题是模式崩溃，生成器对所有输入都输出相似的图像，这时候可以适当提高lambda_cycle，让循环一致性约束更强。4.3 训练过程监控与模型保存训练过程中要定期保存生成器的输出图像，不然光看 loss 曲线很难判断效果。每 5 个 epoch 存一次模型权重，每 1 个 epoch 存一张生成样本拼图。def save_samples(gen_ab, gen_ba, real_a, real_b, epoch, output_dir='samples'): os.makedirs(output_dir, exist_ok=True) fake_b = gen_ab(real_a, training=False) fake_a = gen_ba(real_b, training=False) # 把 real_a, fake_b, real_b, fake_a 拼成一张图保存 # 具体拼接代码略，用 tf.concat 和 PIL 都可以 ... # 训练循环里每 5 个 epoch 保存一次权重 checkpoint_dir = './checkpoints' os.makedirs(checkpoint_dir, exist_ok=True) gen_ab.save_weights(os.path.join(checkpoint_dir, f'gen_ab_epoch{epoch}.h5'))保存权重用.h5格式兼容性好，后续要转 SavedModel 或者 TFLite 也方便。注意保存的是save_weights而不是save，因为自定义训练循环的模型用save可能会丢优化器状态，恢复训练时学习率调度会乱。5. 避坑与排查：训练 CycleGAN 时最常见的五个翻车现场5.1 生成图像全是噪点或纯色块现象：训练几十个 epoch 后，生成器输出要么是随机噪声，要么是一张纯灰或纯色的图，完全没有目标域的风格特征。原因：最常见的是判别器损失降得太快，生成器梯度消失。也可能是数据归一化不一致——生成器输出 tanh 在 [-1,1]，但判别器输入没做同样处理，导致判别器轻松区分真假，生成器学不到东西。解决：检查数据管道里load_image的归一化范围，确保生成器输出和判别器输入都在 [-1,1]。如果判别器 loss 已经降到 0.1 以下，把判别器学习率减半，或者给判别器加 label smoothing，把真实标签从 1.0 改成 0.9。5.2 循环一致性损失不下降现象：对抗损失在波动，但循环损失一直卡在较高值，生成的图像内容和输入完全对不上。原因：lambda_cycle设得太小，或者生成器容量不够，无法同时满足风格迁移和内容保留。另一个可能是两个生成器 G 和 F 的结构不对称，导致循环回来尺寸或通道对不上。解决：先把lambda_cycle从 10 提到 20 试试，观察循环损失是否下降。如果还不行，检查 G 和 F 的n_res_blocks是否一致，输入输出形状是否严格匹配。实在不行就减少残差块数量，降低任务难度。5.3 训练到一半 loss 突然爆炸现象：前几十个 epoch 正常，突然某一轮 loss 变成 NaN，之后模型完全失效。原因：梯度爆炸，通常发生在判别器或生成器某层权重更新过大时。Keras 默认不做梯度裁剪，遇到异常样本容易出问题。解决：在apply_gradients之前加梯度裁剪，把梯度范数限制在 1.0 以内。另外检查数据集里有没有损坏的图像文件，用tf.debugging.check_numerics定位是哪一步出的 NaN。# 梯度裁剪示例 gen_grads = tape.gradient(total_gen_loss, gen_vars) gen_grads = [tf.clip_by_norm(g, 1.0) for g in gen_grads] self.optimizer.apply_gradients(zip(gen_grads, gen_vars))5.4 显存溢出（OOM）现象：训练一开始就报ResourceExhaustedError，或者跑几个 step 后显存慢慢涨满。原因：persistent=True的 GradientTape 会保留中间变量，显存占用比普通 tape 高不少。另外如果train_step里没有用tf.function装饰，每次调用都会重新建图，显存泄漏。解决：给train_step加@tf.function装饰，让 TensorFlow 做图优化。如果还不行，把 batch size 降到 1，图像分辨率从 256 降到 128。实在不够就改用两个独立的 tape 分别求生成器和判别器梯度，牺牲一点代码简洁换显存。5.5 生成结果在训练集上很好，换新图就崩现象：用训练时保存的样本看效果不错，但拿一张训练集外的图进去，输出完全不可用。原因：过拟合。训练集太小或者训练轮次太多，生成器记住了训练样本而不是学到映射关系。另外如果训练集和测试集分布差异大，比如训练用的是夏天风景，测试用的是冬天，模型没见过这种域偏移。解决：增加训练数据量，或者做更强的数据增强（随机裁剪、颜色抖动、旋转）。如果数据量固定，减少训练轮次，在循环损失和对抗损失之间找平衡点。测试时如果域差异大，考虑用少量目标域样本做 fine-tune。6. 进阶技巧：用 Keras 回调做训练可视化与模型导出训练 CycleGAN 最难受的就是盯着 loss 曲线猜效果，所以我会在训练循环里挂一个自定义回调，每 N 个 step 就把当前生成结果拼成一张图存下来，顺便把模型转成 SavedModel 格式方便后续部署。class CycleGANMonitor(tf.keras.callbacks.Callback): def __init__(self, gen_ab, gen_ba, sample_a, sample_b, log_dir='logs', interval=5): super().__init__() self.gen_ab = gen_ab self.gen_ba = gen_ba self.sample_a = sample_a self.sample_b = sample_b self.log_dir = log_dir self.interval = interval os.makedirs(log_dir, exist_ok=True) def on_epoch_end(self, epoch, logs=None): if epoch % self.interval != 0: return fake_b = self.gen_ab(self.sample_a, training=False) fake_a = self.gen_ba(self.sample_b, training=False) # 拼接 real_a | fake_b | real_b | fake_a 并保存 grid = tf.concat([self.sample_a, fake_b, self.sample_b, fake_a], axis=0) grid = (grid + 1.0) * 127.5 # 反归一化到 [0, 255] grid = tf.cast(tf.clip_by_value(grid, 0, 255), tf.uint8) tf.io.write_file( os.path.join(self.log_dir, f'epoch_{epoch:04d}.png'), tf.image.encode_png(grid[0]) )这个回调的好处是把训练过程变成可视化的，不用等训练完再猜。interval设 5 表示每 5 个 epoch 存一张，太多会占磁盘，太少又看不出变化趋势。拼接时用axis=0是纵向堆叠，方便对比 A 域原图、B 域生成图、B 域原图、A 域生成图四者的关系。模型导出方面，训练结束后用tf.saved_model.save把生成器单独存下来，后续推理时直接加载：# 导出生成器为 SavedModel tf.saved_model.save(gen_ab, 'exported_gen_ab') # 推理时加载 loaded = tf.saved_model.load('exported_gen_ab') result = loaded(sample_input)如果目标是移动端部署，可以进一步用 TFLite Converter 量化成 int8 模型，体积能压到原来的四分之一左右，推理速度也快不少。但量化后生成质量会有轻微下降，做风格迁移这种对细节敏感的任务时，建议先用 float16 量化试试，不行再上 int8。从那以后我每次训练 GAN 类模型，都会在第一个 epoch 结束后先跑一遍可视化回调，确认生成器输出不是纯噪声再继续往下训。这个习惯帮我省过好几次通宵跑完才发现模型根本没学到的后悔药。希望帮到你。本文还有配套的精品资源，点击获取
简介：这份资源是CycleGAN的Keras实现代码包，面向具备一定深度学习基础、希望动手实践无监督图像到图像转换的开发者与研究者。它围绕生成对抗网络展开，重点解决无配对样本条件下的跨域图像转换问题，可应用于风格迁移、季节转换等场景。压缩包共7个文件，包含4个Python脚本与3个数据压缩包，整体约477.41MB；脚本分别承担模型构建、数据加载、预测推理与生成器骨干网络等职责，数据包则提供马与斑马、苹果与橘子、莫奈与照片等经典非配对图像集，便于直接开展训练与效果验证。目前已有548人学习下载。通过阅读代码，读者可以理解双生成器与双判别器的协作方式、循环一致性损失的实现细节，以及对抗损失与循环损失的联合优化流程，并借助预测脚本将训练好的模型用于新图像的风格转换，适合作为入门CycleGAN并复现实验的参考工程。1. 一份能跑通的 CycleGAN Keras 实现，到底解决了谁的燃眉之急手里有一批成对的图像数据，想训练一个风格迁移模型，结果发现配对标注成本高得离谱——这种场景做图像生成的人多半都遇到过。CycleGAN 的价值就在于绕开了成对样本的硬性要求：它用两组非配对图像，靠循环一致性损失把 A 域和 B 域之间的映射约束住，让马变斑马、照片变油画这类任务不再依赖一一对应的训练集。而这份 Keras 实现，把生成器、判别器、循环损失、身份损失、训练循环全部用 Keras 的层和模型 API 搭出来，没有藏着掖着的自定义算子，适合想真正读懂 CycleGAN 每一行在干什么、而不是调个库就完事的人。如果你已经会用 Keras 搭普通 CNN，但对 GAN 的训练流程还停留在“看论文知道大概”的阶段，这份代码就是拿来逐行拆解的好材料。2. CycleGAN 的核心机制与 Keras 实现选型：为什么不用 PyTorch 版本2.1 循环一致性损失到底在约束什么CycleGAN 要解决的核心问题是：没有成对数据，怎么保证 G(A)→B 生成的图像既像 B 域，又能保留 A 域的内容。答案就是两个方向的生成器加两个判别器，再叠加循环一致性损失。具体来说，设 G 负责 A→B，F 负责 B→A，那么 F(G(a)) 应该约等于 a，G(F(b)) 应该约等于 b。这个约束让生成器不能随便把输入映射到 B 域里一个无关的样本上，否则循环回来就对不上。在 Keras 里实现这个逻辑，关键是把 G、F、D_A、D_B 四个模型分别定义好，然后用tf.GradientTape或者train_on_batch手动控制训练步骤。常见做法是写一个CycleGANTrainer类，在train_step里依次计算：生成器总损失 = 对抗损失 + λ₁ × 循环一致性损失 + λ₂ × 身份损失判别器损失 = 两个域各自判别真伪的交叉熵之和其中 λ₁ 默认取 10，λ₂ 默认取 5 或 10，这两个系数直接决定生成图像是更偏向目标域风格还是更保留原始内容。调参时如果发现生成结果“风格化过头、内容全丢”，优先把 λ₁ 调大；如果风格迁移效果太弱，则适当降低 λ₁ 或提高 λ₂。2.2 Keras 实现相比 PyTorch 版本的取舍PyTorch 在学术界更流行，动态图写起来灵活，调试也直观。但这份 Keras 实现有它自己的适用场景：如果你的部署环境是 TensorFlow Serving 或者 TFLite，Keras 模型导出和量化链路更顺；如果你团队里大部分人用 Keras 做常规图像任务，引入 PyTorch 会增加维护成本。另外 Keras 的Model子类化写法虽然不如 PyTorch 的forward直观，但一旦把call方法写清楚，后续做模型剪枝、转 ONNX 都更方便。不过要提前说清楚边界：Keras 实现在自定义训练循环里对梯度裁剪、学习率调度这些细节的控制不如 PyTorch 灵活，遇到训练不稳定时排查手段会少一些。所以这份代码更适合作为学习模板和中小规模实验的起点，而不是直接拿去跑大规模生产训练。2.3 环境准备与依赖版本在动手之前，先把环境对齐。这份实现基于 TensorFlow 2.x，建议用 2.10 以上的版本，Python 3.8 到 3.10 之间比较稳。依赖不多，核心就是 tensorflow、numpy、Pillow 和 matplotlib（可视化用）。# 创建虚拟环境并安装依赖 python -m venv cyclegan_env source cyclegan_env/bin/activate # Windows 用 cyclegan_env\Scripts\activate pip install tensorflow==2.12.0 numpy pillow matplotlib这里固定 TensorFlow 版本是有原因的：2.12 之后 Keras 被拆成独立包，部分 API 路径变了，如果直接用最新版可能会遇到tf.keras和keras混用报错。如果你已经装了其他版本，先pip show tensorflow确认一下，避免后面调试半天发现是版本问题。提示：如果机器上有 GPU，装tensorflow-gpu或者确认tf.config.list_physical_devices('GPU')能返回设备列表。CycleGAN 训练对显存有一定要求，256×256 分辨率下 batch size 设 1 也可能占到 6GB 以上。3. 从零搭建生成器与判别器：Keras 层怎么堆才不翻车3.1 生成器结构：编码器-转换器-解码器三段式CycleGAN 的生成器本质是一个带残差连接的编解码网络。输入一张 256×256×3 的图像，先经过两层步长为 2 的卷积下采样到 64×64，再经过若干残差块做特征变换，最后用转置卷积上采样回原尺寸。Keras 里用Conv2D、BatchNormalization、Activation和Add就能拼出来。import tensorflow as tf from tensorflow.keras import layers, Model def build_generator(input_shape=(256, 256, 3), n_res_blocks=6): """构建 CycleGAN 生成器：下采样 -> 残差块 -> 上采样""" inputs = layers.Input(shape=input_shape) # 初始卷积 x = layers.Conv2D(64, 7, strides=1, padding='same')(inputs) x = layers.BatchNormalization()(x) x = layers.Activation('relu')(x) # 下采样两层 for filters in [128, 256]: x = layers.Conv2D(filters, 3, strides=2, padding='same')(x) x = layers.BatchNormalization()(x) x = layers.Activation('relu')(x) # 残差块 for _ in range(n_res_blocks): res = layers.Conv2D(256, 3, padding='same')(x) res = layers.BatchNormalization()(res) res = layers.Activation('relu')(res) res = layers.Conv2D(256, 3, padding='same')(res) res = layers.BatchNormalization()(res) x = layers.Add()([x, res]) # 上采样两层 for filters in [128, 64]: x = layers.Conv2DTranspose(filters, 3, strides=2, padding='same')(x) x = layers.BatchNormalization()(x) x = layers.Activation('relu')(x) # 输出层，tanh 把像素值压到 [-1, 1] outputs = layers.Conv2D(3, 7, padding='same', activation='tanh')(x) return Model(inputs, outputs, name='generator')这段代码里几个参数值得注意：n_res_blocks默认 6，对应 256×256 输入；如果降到 128×128，可以减到 3 或 4，否则参数量偏大容易过拟合。输出层用tanh而不是sigmoid，是因为训练时图像归一化到 [-1, 1] 区间，判别器输入也按这个范围处理，保持一致。残差块里没有加激活函数在Add之后，这是原论文的做法，实测影响不大但保持一致性更好。3.2 判别器结构：PatchGAN 为什么比全图判别更稳判别器用的是 PatchGAN，输出不是一个标量，而是一个 N×N 的 patch 矩阵，每个值代表原图一个感受野区域的真假。这样做的好处是参数量小、训练更稳定，而且能捕捉局部纹理细节。Keras 实现起来就是一个几层卷积堆到输出通道为 1 的Conv2D。def build_discriminator(input_shape=(256, 256, 3)): """PatchGAN 判别器，输出 patch 级别的真假评分""" inputs = layers.Input(shape=input_shape) x = layers.Conv2D(64, 4, strides=2, padding='same')(inputs) x = layers.LeakyReLU(0.2)(x) for filters in [128, 256, 512]: x = layers.Conv2D(filters, 4, strides=2, padding='same')(x) x = layers.BatchNormalization()(x) x = layers.LeakyReLU(0.2)(x) # 输出单通道 patch 评分 outputs = layers.Conv2D(1, 4, padding='same')(x) return Model(inputs, outputs, name='discriminator')判别器里用LeakyReLU(0.2)而不是普通 ReLU，是为了避免梯度稀疏。注意最后一层没有加 sigmoid，因为损失函数里用的是BinaryCrossentropy(from_logits=True)，这样数值更稳定。如果你习惯用sigmoid加普通交叉熵，训练初期容易遇到梯度消失，生成器更新不动。3.3 损失函数与训练循环的 Keras 写法把模型搭好之后，核心工作就是写训练循环。Keras 里可以用tf.GradientTape手动求梯度，也可以用train_on_batch，但后者对多模型交替训练的支持不够直观，所以常见做法是自定义train_step。class CycleGANTrainer(tf.keras.Model): def __init__(self, gen_ab, gen_ba, disc_a, disc_b, lambda_cycle=10.0, lambda_id=5.0): super().__init__() self.gen_ab = gen_ab self.gen_ba = gen_ba self.disc_a = disc_a self.disc_b = disc_b self.lambda_cycle = lambda_cycle self.lambda_id = lambda_id self.bce = tf.keras.losses.BinaryCrossentropy(from_logits=True) def train_step(self, data): real_a, real_b = data with tf.GradientTape(persistent=True) as tape: # 生成器前向 fake_b = self.gen_ab(real_a, training=True) fake_a = self.gen_ba(real_b, training=True) cycle_a = self.gen_ba(fake_b, training=True) cycle_b = self.gen_ab(fake_a, training=True) id_a = self.gen_ba(real_a, training=True) id_b = self.gen_ab(real_b, training=True) # 对抗损失 disc_real_a = self.disc_a(real_a, training=True) disc_fake_a = self.disc_a(fake_a, training=True) disc_real_b = self.disc_b(real_b, training=True) disc_fake_b = self.disc_b(fake_b, training=True) gen_ab_loss = self.bce(tf.ones_like(disc_fake_b), disc_fake_b) gen_ba_loss = self.bce(tf.ones_like(disc_fake_a), disc_fake_a) cycle_loss = tf.reduce_mean(tf.abs(real_a - cycle_a)) + \ tf.reduce_mean(tf.abs(real_b - cycle_b)) id_loss = tf.reduce_mean(tf.abs(real_a - id_a)) + \ tf.reduce_mean(tf.abs(real_b - id_b)) total_gen_loss = gen_ab_loss + gen_ba_loss + \ self.lambda_cycle * cycle_loss + self.lambda_id * id_loss disc_a_loss = self.bce(tf.ones_like(disc_real_a), disc_real_a) + \ self.bce(tf.zeros_like(disc_fake_a), disc_fake_a) disc_b_loss = self.bce(tf.ones_like(disc_real_b), disc_real_b) + \ self.bce(tf.zeros_like(disc_fake_b), disc_fake_b) # 分别更新生成器和判别器 gen_vars = self.gen_ab.trainable_variables + self.gen_ba.trainable_variables disc_vars = self.disc_a.trainable_variables + self.disc_b.trainable_variables self.optimizer.apply_gradients(zip(tape.gradient(total_gen_loss, gen_vars), gen_vars)) self.optimizer.apply_gradients(zip(tape.gradient(disc_a_loss, disc_vars), disc_vars)) self.optimizer.apply_gradients(zip(tape.gradient(disc_b_loss, disc_vars), disc_vars)) return {'gen_loss': total_gen_loss, 'disc_a_loss': disc_a_loss, 'disc_b_loss': disc_b_loss}这段代码里persistent=True是必须的，因为要对多个损失分别求梯度，普通 GradientTape 求完一次就释放了。lambda_cycle和lambda_id直接决定训练行为，前面已经说过调参方向。另外注意判别器更新了两次（D_A 和 D_B 各一次），但用的是同一个优化器，实际项目里也可以给判别器单独配一个优化器，学习率设成生成器的一半，训练会更稳。注意：如果显存不够，把train_step里的persistent=True改成分别用两个 tape 写，虽然代码长一点但能省显存。另外tf.abs算 L1 损失时注意张量形状要一致，循环回来的图像尺寸必须和原图完全对齐，否则会广播报错。4. 数据管道与训练调参：非配对图像怎么喂进去4.1 非配对数据集的加载与增强CycleGAN 的训练数据是两个独立的文件夹，比如trainA和trainB，里面各自放一堆图片，不需要文件名对应。用tf.data.Dataset分别读取两个目录，做随机裁剪、翻转、归一化，然后 zip 到一起。import os def load_image(path, size=256): img = tf.io.read_file(path) img = tf.image.decode_jpeg(img, channels=3) img = tf.image.resize(img, [size, size]) img = (tf.cast(img, tf.float32) / 127.5) - 1.0 # 归一化到 [-1, 1] return img def build_dataset(dir_a, dir_b, batch_size=1, size=256): files_a = [os.path.join(dir_a, f) for f in os.listdir(dir_a) if f.endswith(('.jpg', '.png'))] files_b = [os.path.join(dir_b, f) for f in os.listdir(dir_b) if f.endswith(('.jpg', '.png'))] ds_a = tf.data.Dataset.from_tensor_slices(files_a) ds_a = ds_a.map(lambda x: load_image(x, size), num_parallel_calls=tf.data.AUTOTUNE) ds_a = ds_a.cache().shuffle(len(files_a)).batch(batch_size).repeat() ds_b = tf.data.Dataset.from_tensor_slices(files_b) ds_b = ds_b.map(lambda x: load_image(x, size), num_parallel_calls=tf.data.AUTOTUNE) ds_b = ds_b.cache().shuffle(len(files_b)).batch(batch_size).repeat() return tf.data.Dataset.zip((ds_a, ds_b)).prefetch(tf.data.AUTOTUNE)这里batch_size默认设 1，是因为 CycleGAN 对 batch 内样本的独立性要求不高，而且显存占用大。如果显存充裕可以设 2 或 4，但要注意学习率也要相应调整。shuffle的 buffer 设成文件总数，保证每个 epoch 顺序不同。repeat()让数据集无限循环，训练步数由外部控制。4.2 学习率、优化器与训练轮次设置优化器用 Adam，生成器和判别器学习率都设 2e-4，beta_1 设 0.5。这个配置来自原论文，实测在多数风格迁移任务上都能收敛。训练轮次一般 100 到 200 个 epoch 起步，前 50 个 epoch 保持学习率不变，之后线性衰减到 0。# 优化器配置 gen_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5) disc_optimizer = tf.keras.optimizers.Adam(2e-4, beta_1=0.5) # 学习率衰减回调（如果不用自定义 train_step 可以用这个） def lr_schedule(epoch, lr): if epoch < 50: return lr return lr * (1 - (epoch - 50) / 150)如果发现训练到 30 个 epoch 左右生成图像开始出现网格状伪影，多半是判别器太强了，把判别器学习率降到 1e-4 或者给生成器损失里的对抗部分加个权重系数。另一个常见问题是模式崩溃，生成器对所有输入都输出相似的图像，这时候可以适当提高lambda_cycle，让循环一致性约束更强。4.3 训练过程监控与模型保存训练过程中要定期保存生成器的输出图像，不然光看 loss 曲线很难判断效果。每 5 个 epoch 存一次模型权重，每 1 个 epoch 存一张生成样本拼图。def save_samples(gen_ab, gen_ba, real_a, real_b, epoch, output_dir='samples'): os.makedirs(output_dir, exist_ok=True) fake_b = gen_ab(real_a, training=False) fake_a = gen_ba(real_b, training=False) # 把 real_a, fake_b, real_b, fake_a 拼成一张图保存 # 具体拼接代码略，用 tf.concat 和 PIL 都可以 ... # 训练循环里每 5 个 epoch 保存一次权重 checkpoint_dir = './checkpoints' os.makedirs(checkpoint_dir, exist_ok=True) gen_ab.save_weights(os.path.join(checkpoint_dir, f'gen_ab_epoch{epoch}.h5'))保存权重用.h5格式兼容性好，后续要转 SavedModel 或者 TFLite 也方便。注意保存的是save_weights而不是save，因为自定义训练循环的模型用save可能会丢优化器状态，恢复训练时学习率调度会乱。5. 避坑与排查：训练 CycleGAN 时最常见的五个翻车现场5.1 生成图像全是噪点或纯色块现象：训练几十个 epoch 后，生成器输出要么是随机噪声，要么是一张纯灰或纯色的图，完全没有目标域的风格特征。原因：最常见的是判别器损失降得太快，生成器梯度消失。也可能是数据归一化不一致——生成器输出 tanh 在 [-1,1]，但判别器输入没做同样处理，导致判别器轻松区分真假，生成器学不到东西。解决：检查数据管道里load_image的归一化范围，确保生成器输出和判别器输入都在 [-1,1]。如果判别器 loss 已经降到 0.1 以下，把判别器学习率减半，或者给判别器加 label smoothing，把真实标签从 1.0 改成 0.9。5.2 循环一致性损失不下降现象：对抗损失在波动，但循环损失一直卡在较高值，生成的图像内容和输入完全对不上。原因：lambda_cycle设得太小，或者生成器容量不够，无法同时满足风格迁移和内容保留。另一个可能是两个生成器 G 和 F 的结构不对称，导致循环回来尺寸或通道对不上。解决：先把lambda_cycle从 10 提到 20 试试，观察循环损失是否下降。如果还不行，检查 G 和 F 的n_res_blocks是否一致，输入输出形状是否严格匹配。实在不行就减少残差块数量，降低任务难度。5.3 训练到一半 loss 突然爆炸现象：前几十个 epoch 正常，突然某一轮 loss 变成 NaN，之后模型完全失效。原因：梯度爆炸，通常发生在判别器或生成器某层权重更新过大时。Keras 默认不做梯度裁剪，遇到异常样本容易出问题。解决：在apply_gradients之前加梯度裁剪，把梯度范数限制在 1.0 以内。另外检查数据集里有没有损坏的图像文件，用tf.debugging.check_numerics定位是哪一步出的 NaN。# 梯度裁剪示例 gen_grads = tape.gradient(total_gen_loss, gen_vars) gen_grads = [tf.clip_by_norm(g, 1.0) for g in gen_grads] self.optimizer.apply_gradients(zip(gen_grads, gen_vars))5.4 显存溢出（OOM）现象：训练一开始就报ResourceExhaustedError，或者跑几个 step 后显存慢慢涨满。原因：persistent=True的 GradientTape 会保留中间变量，显存占用比普通 tape 高不少。另外如果train_step里没有用tf.function装饰，每次调用都会重新建图，显存泄漏。解决：给train_step加@tf.function装饰，让 TensorFlow 做图优化。如果还不行，把 batch size 降到 1，图像分辨率从 256 降到 128。实在不够就改用两个独立的 tape 分别求生成器和判别器梯度，牺牲一点代码简洁换显存。5.5 生成结果在训练集上很好，换新图就崩现象：用训练时保存的样本看效果不错，但拿一张训练集外的图进去，输出完全不可用。原因：过拟合。训练集太小或者训练轮次太多，生成器记住了训练样本而不是学到映射关系。另外如果训练集和测试集分布差异大，比如训练用的是夏天风景，测试用的是冬天，模型没见过这种域偏移。解决：增加训练数据量，或者做更强的数据增强（随机裁剪、颜色抖动、旋转）。如果数据量固定，减少训练轮次，在循环损失和对抗损失之间找平衡点。测试时如果域差异大，考虑用少量目标域样本做 fine-tune。6. 进阶技巧：用 Keras 回调做训练可视化与模型导出训练 CycleGAN 最难受的就是盯着 loss 曲线猜效果，所以我会在训练循环里挂一个自定义回调，每 N 个 step 就把当前生成结果拼成一张图存下来，顺便把模型转成 SavedModel 格式方便后续部署。class CycleGANMonitor(tf.keras.callbacks.Callback): def __init__(self, gen_ab, gen_ba, sample_a, sample_b, log_dir='logs', interval=5): super().__init__() self.gen_ab = gen_ab self.gen_ba = gen_ba self.sample_a = sample_a self.sample_b = sample_b self.log_dir = log_dir self.interval = interval os.makedirs(log_dir, exist_ok=True) def on_epoch_end(self, epoch, logs=None): if epoch % self.interval != 0: return fake_b = self.gen_ab(self.sample_a, training=False) fake_a = self.gen_ba(self.sample_b, training=False) # 拼接 real_a | fake_b | real_b | fake_a 并保存 grid = tf.concat([self.sample_a, fake_b, self.sample_b, fake_a], axis=0) grid = (grid + 1.0) * 127.5 # 反归一化到 [0, 255] grid = tf.cast(tf.clip_by_value(grid, 0, 255), tf.uint8) tf.io.write_file( os.path.join(self.log_dir, f'epoch_{epoch:04d}.png'), tf.image.encode_png(grid[0]) )这个回调的好处是把训练过程变成可视化的，不用等训练完再猜。interval设 5 表示每 5 个 epoch 存一张，太多会占磁盘，太少又看不出变化趋势。拼接时用axis=0是纵向堆叠，方便对比 A 域原图、B 域生成图、B 域原图、A 域生成图四者的关系。模型导出方面，训练结束后用tf.saved_model.save把生成器单独存下来，后续推理时直接加载：# 导出生成器为 SavedModel tf.saved_model.save(gen_ab, 'exported_gen_ab') # 推理时加载 loaded = tf.saved_model.load('exported_gen_ab') result = loaded(sample_input)如果目标是移动端部署，可以进一步用 TFLite Converter 量化成 int8 模型，体积能压到原来的四分之一左右，推理速度也快不少。但量化后生成质量会有轻微下降，做风格迁移这种对细节敏感的任务时，建议先用 float16 量化试试，不行再上 int8。从那以后我每次训练 GAN 类模型，都会在第一个 epoch 结束后先跑一遍可视化回调，确认生成器输出不是纯噪声再继续往下训。这个习惯帮我省过好几次通宵跑完才发现模型根本没学到的后悔药。希望帮到你。本文还有配套的精品资源，点击获取