CycleGAN

2025年3月17日 | 阅读18分钟

图像到图像的翻译是创建现有图像的合成修改新版本的过程。例如，将夏日风景转换为冬日风景。通常需要大量匹配的实例集合来训练图像到图像的翻译模型。某些文件，例如已故画家的艺术品照片，可能非常昂贵、复杂，甚至不可能收集。一种名为 CycleGAN 的方法可在没有配对实例的情况下自动训练图像到图像的翻译模型。通过使用来自源域和目标域的照片集——这些域之间不必有任何连接——模型会自动进行训练。

CycleGAN 由两种类型的网络组成：判别器和生成器。在此示例中，判别器负责将图像分类为真实或伪造（对于 X 和 Y 两种类型的图像）。生成器负责为两种类型的图像生成逼真的伪造图像。

它因其无配对图像翻译能力而受到青睐，这使得它能够在不要求训练集中有匹配对的情况下学习跨不同图像域的映射。它提供了更大的灵活性和适应性，因为它以无监督的方式运行，并且可以从源域和目标域的图片集中学习，而无需明确的关联。循环一致性是指翻译后的图像在多次翻译后仍能保持其源的真实性，从而产生更逼真的结果。由于该方法大大减少了对配对数据集的依赖，因此 CycleGAN 在难以获取标注数据的场景中非常有用。

CycleGAN 的实现

导入库

import os
import torch
import torchvision
from torch.utils.data import DataLoader
from torchvision import transforms
import numpy as np
from torchvision.datasets import ImageFolder
from torch.utils.data import Dataset
import cv2

加载数据集

get_data_loader 函数返回可以快速加载数据并按预定批次加载数据的训练和测试 DataLoader。该函数具有以下参数：

image_dir：主图像目录的名称，其中包含所有训练和测试图像；-image_type：夏季或冬季，存储 X 和 Y 图像的文件夹名称
image_size：已下采样的方形图像尺寸；所有图像都将调整到此尺寸
batch_size：单个数据批次中的图像数量

测试数据旨在供我们未来的生成器使用，以便我们可以查看固定在测试数据上的一些生成样本。

data_dir='../input/gan-getting-started'
class GANdataGenerator(Dataset):
    def __init__(self,data_dir ,  data_files , image_size ):
        super(GANdataGenerator , self).__init__()
        self.data_dir = data_dir
        self.data = data_files
        self.transform = transforms.Compose([
            transforms.ToPILImage(),
            transforms.Resize(image_size),
            transforms.CenterCrop(image_size),
            transforms.ToTensor()
        ])
        
    def __len__(self):
        return len(self.data)
    
    def __getitem__(self, index):
        data_file = self.data[index]
        data_file = os.path.join(self.data_dir , data_file)
        img_bgr = cv2.imread(data_file)
        img_rgb = cv2.cvtColor(img_bgr , cv2.COLOR_BGR2RGB)
        img_tensor = self.transform(img_rgb)
        
        return img_tensor

正如你所见，此函数还负责确保我们的图像被转换为 Tensor 图像类型并具有正确的方形尺寸（128x128x3）。

注意：建议将这些设置保留为默认值。如果你尝试在另一组数据上运行此代码，更高的 image_size 和 batch_size 选项可能会产生更好的结果。在训练循环中调整 batch_size 之前，请务必构建完整的批次，因为这样做可能会在尝试存储样本数据时导致错误。

monet_jpg = os.listdir('../input/gan-getting-started/monet_jpg')
photo_jpg = os.listdir('../input/gan-getting-started/photo_jpg')

#load the Dataloader for X and Y image sets
X_type='photo_jpg'
Y_type='monet_jpg'
X_set = GANdataGenerator(data_dir='../input/gan-getting-started/photo_jpg' , data_files=photo_jpg , image_size=128 )
Y_set = GANdataGenerator(data_dir='../input/gan-getting-started/monet_jpg' , data_files=monet_jpg , image_size=128 )

dataloader_X = DataLoader(X_set , batch_size=16 , shuffle=True , num_workers=0)
dataloader_Y = DataLoader(Y_set , batch_size=16 , shuffle=True , num_workers=0)

x_data = next(iter(dataloader_X))
y_data = next(iter(dataloader_Y))

可视化数据

import matplotlib.pyplot as plt
print("X Data Visualization")
def imshow(img):
    npimg = img.numpy()
    plt.imshow(np.transpose(npimg, (1, 2, 0)))
fig = plt.figure(figsize=(12, 8))
imshow(torchvision.utils.make_grid(x_data))

输出

X 数据可视化

print("Y data Visualization")
fig = plt.figure(figsize=(12, 8))
imshow(torchvision.utils.make_grid(y_data))

输出

Y 数据可视化

import matplotlib.pyplot as plt
def viz_sample(samples_x , samples_y):
    #get random sample from the x and y
    #plot the sample
    plt.title("Real sample")
    plt.imshow(samples_x)
    plt.show()
    plt.title("Fake sample")
    plt.imshow(samples_y)
    plt.show()

缩放

由于我们知道 tanh 激活的生成器输出的像素值将在 -1 到 1 之间变化，因此我们需要进行一些预处理。因此，我们必须将训练图像重新缩放到此范围内。（目前，它们在 0 到 1 之间。）

img = x_data[0]

print("Min : ",img.min())
print("Max : ",img.max())

def rescale(x , feature_range=(-1,1)):
    min , max = feature_range
    x = x*(max - min) +min
    
    return x

输出

# scaled range
scaled_img = rescale(img)

print('Scaled min: ', scaled_img.min())
print('Scaled max: ', scaled_img.max())

输出

定义模型

CycleGAN 由两个生成器网络和两个判别器网络组成。

判别器

在此 CycleGAN 中，判别器 DX 和 DY 是卷积神经网络，它们分析图像并尝试确定它是真实的还是伪造的。在这里，输出接近 1 表示真实，输出接近 0 表示伪造。判别器具有以下架构：

将 256x256x3 大小的图像输入到该网络，并通过 5 个卷积层进行处理，这些卷积层将其下采样 2 倍。BatchNorm 和 ReLu 激活函数应用于前四个卷积层的输出，而最后一层充当分类层并产生单个值。

卷积辅助函数

你应该使用提供的 conv 函数，它会生成一个卷积层加上一个可选的 batch norm 层，来定义判别器。

import torch.nn as nn
import torch.nn.functional as F

#define the conv block for a discriminator
def conv_block(in_channels , out_channels , kernel_size=4 , strides=2 ,padding_type='reflect' , padding=1 , batch_norm=True):
    layers = []
    if(padding_type=='reflect'):
        pad = nn.ReflectionPad2d(padding)
        layers.append(pad)
        
    if(padding_type=='replicate'):
        pad = nn.ReplicationPad2d(padding)
        layers.append(pad)
        
        
    conv_layer = nn.Conv2d(in_channels=in_channels , out_channels=out_channels , 
                          kernel_size=kernel_size , stride=strides , padding=0 ,  bias=False)
    layers.append(conv_layer)
    if(batch_norm):
        bn = nn.InstanceNorm2d(out_channels)
        layers.append(bn)
        
    return nn.Sequential(*layers)

判别器架构

使用上述五层卷积网络设计，挑战在于完成 __init__ 函数。我们只需要指定一个类，然后实例化两个判别器，因为 DX 和 DY 共享相同的设计。

forward 函数决定了图像如何进入判别器；重要的是按顺序将图像通过每个卷积层，对除最后一层之外的所有层使用 ReLu 激活函数。

由于我们要使用平方误差损失进行训练，因此在这种情况下不应在输出中添加 sigmoid 激活函数。稍后你可以在笔记本中了解有关此损失函数的更多信息。

class Discriminator(nn.Module):
    
    def __init__(self , conv_dim , padding_mode):
        super(Discriminator , self).__init__()
        """
        Define the discriminator model to classify the images as real or fake
        in_channels =3
        out_channels=1
        """
        self.input_channels = 3
        self.output_channels = 1
        self.padding_mode = padding_mode
        # 256 - 256 -3 --> 128 - 128 -64
        self.conv_block1 = conv_block(self.input_channels , conv_dim  , padding_type=padding_mode)
        # 128 - 128 - 64 --> 64 - 64 - 128
        self.conv_block2 = conv_block(conv_dim , conv_dim*2 , padding_type=padding_mode)
        # 64 - 64 - 128 --> 32 - 32 - 256
        self.conv_block3 = conv_block(conv_dim*2 , conv_dim*4 , padding_type=padding_mode )
        # 32 - 32 - 256 --> 16 - 16 - 512
        self.conv_block4 = conv_block(conv_dim*4 , conv_dim*8 , padding_type=padding_mode)
        # 16 - 16 - 512 --> 8 - 8 - 1024
        self.conv_block5 = conv_block(conv_dim*8 , self.output_channels , kernel_size=1 ,
                                      strides=1 ,padding_type=None ,  padding=0 , batch_norm=False)
        self.leaky_relu = nn.LeakyReLU(negative_slope=0.2, inplace=True)
        
    def forward(self , x):
        x = self.leaky_relu(self.conv_block1(x))
        x = self.leaky_relu(self.conv_block2(x))
        x = self.leaky_relu(self.conv_block3(x))
        x = self.leaky_relu(self.conv_block4(x))
        
        out = self.conv_block5(x)
        
        return out

生成器

生成器 G_XtoY 和 G_YtoX（有时称为 F）由一个编码器（一个将图像压缩为较小特征表示的卷积网络）和一个解码器（一个将该表示转换为修改后图像的转置卷积网络）组成。从 Y 到 X 以及从 X 到 Y 的这些生成器的构造如下：

当该网络接收到 256x256x3 的图像时，它会将其压缩为特征表示，并通过三个卷积层，然后进入一组残差块。它会通过多个此类残差块——通常是六个或更多——然后通过三个转置卷积层，也称为反卷积层，它们将残差块的输出上采样以生成新图像！

除了最后一个转置卷积层应用 tanh 激活函数到输出之外，请注意，大多数卷积层和转置卷积层在其输出上应用了 BatchNorm 和 ReLu 函数。此外，卷积层和批归一化层构成了残差块；我们稍后将更详细地讨论这些。

残差块类

为了定义生成器，我们必须构建一个 ResidualBlock 类。此类将使我们能够连接生成器的编码器和解码器部分。也许你正在想，Resnet 块具体是什么？它可能看起来与图像分类系统 ResNet50 相似，如下所示。

通过使用残差块，我们可以学习所谓的残差函数，当它们应用于层输入时，这是解决此问题的一种方法。

残差函数

典型的深度学习模型由许多带有激活函数的层组成，其任务是学习从输入 (x) 到输出 (y) 的映射 M。

通过定义残差函数，我们可以避免学习从 x 到 y 的直接映射。

这会检查原始输入 x 和应用于 x 的映射之间的差异。通常，F(x) 由一个归一化层、两个卷积层和一个中间的 ReLu 组成。这些卷积层的输入和输出的数量应相等。然后，映射可以表示为输入 x 和残差函数的函数。通过加法步骤，在输入 (x) 和输出 (y) 之间形成一个几乎圆形的连接。

定义 ResidualBlock 类

我们将构建残差函数，这是一组层，将它们应用于输入 x，然后将它们添加到相同的输入，以定义 ResidualBlock 类。这与任何其他神经网络一样，使用相同的 __init__ 函数和 forward 函数加法步骤来定义。

在这种情况下，残差块应定义如下：

两个输入和输出尺寸相同的卷积层
卷积层输出进行批归一化。
应用于第一个卷积层输出的 ReLu 函数

接下来，在 forward 函数中将输入 x 添加到此残差块。你可以使用上面提到的辅助 conv 方法来创建此块。

#define the discriminator network with encoder - resblock - decoder
class ResidualBlock(nn.Module):
    """
    apply two conv layers with Batchnormalization and relu activation on the first block
    """
    def __init__(self , conv_dim , padding_mode):
        super(ResidualBlock , self).__init__()
        self.conv_dim = conv_dim
        self.conv_block1 = conv_block(self.conv_dim , self.conv_dim , kernel_size=3 ,
                                      strides=1 ,padding_type=padding_mode , padding=1 , batch_norm=True)
        
        self.conv_block2 = conv_block(self.conv_dim , self.conv_dim , kernel_size=3 ,
                                      strides=1 ,padding_type=padding_mode, padding=1 , batch_norm=True)
        
        self.dropout = nn.Dropout(p=0.3)
        self.relu = nn.ReLU(inplace=True)
        
    def forward(self , x):
        res_1 = self.dropout(self.relu(self.conv_block1(x)))
        res_2 = self.conv_block2(res_1)
        out  = self.relu(res_2 + x)
        
        return out

转置卷积辅助函数

然后，我们使用 ResidualBlock 类、上面的 conv 方法和下面的 deconv 辅助函数来定义生成器。这些将生成一个转置卷积层以及一个可选的 batchnorm 层。

def deconv_block(in_channels , out_channels , kernel_size=4 , strides=2 , padding=1 , batch_norm=True):
    layers = []        
    deconv_layer = nn.ConvTranspose2d(in_channels=in_channels , out_channels=out_channels ,
                                      kernel_size=kernel_size , stride=strides , padding=padding , bias=False)
    layers.append(deconv_layer)
    if(batch_norm):
        bn = nn.InstanceNorm2d(out_channels)
        layers.append(bn)
        
    return nn.Sequential(*layers)

生成器架构

使用提供的三层编码器卷积网络、一系列残差块（n_res_blocks 表示数量）和一个三层解码器转置卷积网络来创建 __init__ 函数。
接下来，完成 forward 函数以指定生成器如何进行。请记住，最后一层有一个 tanh 激活函数。

由于 GXtoY 和 GYtoX 的架构相同，因此我们只需要编写一个类，然后实例化两个生成器。

class CycleGenerator(nn.Module):
    def __init__(self , conv_dim , n_resblocks , padding_mode):
        super(CycleGenerator , self).__init__()
        self.conv_dim = conv_dim
        self.n_resblocks = n_resblocks
        self.in_channels = 3
        self.out_channels = 3
        self.encoder = nn.Sequential(
            conv_block(self.in_channels , conv_dim , padding_type=padding_mode) ,
            nn.ReLU(True),
            conv_block(conv_dim , conv_dim*2 , padding_type=padding_mode), 
            nn.ReLU(True),
            conv_block(conv_dim*2 , conv_dim*4 , padding_type=padding_mode),
            nn.ReLU(True)
        )
        layers = []
        for i_res in range(self.n_resblocks):
            layers.append(ResidualBlock(conv_dim*4 , padding_mode=padding_mode))
        self.res_block = nn.Sequential(*layers)
        
        self.decoder =nn.Sequential(
            deconv_block(conv_dim*4 , conv_dim*2 ),
            nn.ReLU(True),
            deconv_block(conv_dim*2, conv_dim ),
            nn.ReLU(True),
            deconv_block(conv_dim , self.out_channels , batch_norm=False),
            nn.Tanh()
        )
    def forward(self, x):
        #inference throght the encoder
        x = self.encoder(x)
        x = self.res_block(x)
        x = self.decoder(x)
        
        return x

完成网络

我们可以指定构建完整 CycleGAN 所需的生成器和判别器，使用您已经建立的类。提供的设置对于训练应该很有效。

首先，创建两个判别器：一个用于验证 X 样本图像的真实性，另一个用于验证 Y 样本图像的真实性。然后是生成器。创建两个实例：一个用于将一幅画转换为逼真的图像，另一个用于将一张图像转换为一幅画。

def weights_init_normal(m):
    classname = m.__class__.__name__
    if(classname.find('Conv') != -1):
        torch.nn.init.normal(m.weight.data , 0.0 , 0.02)
    elif(classname.find('Batchnorm2d')!= -1):
        torch.nn.init.normal(m.weight.data , 0.0 , 0.02)
        torch.nn.init.constant(m.bias.data , 0.0)

def create_model(g_conv_dim=64 , d_conv_dim=64 , n_resblocks = 9 , padding_mode='reflect'):
    G_X2Y = CycleGenerator(g_conv_dim ,n_resblocks , padding_mode)
    G_Y2X = CycleGenerator(g_conv_dim ,n_resblocks , padding_mode)
    
    D_X = Discriminator(d_conv_dim  ,padding_mode)
    D_Y = Discriminator(d_conv_dim , padding_mode)
    
    G_X2Y.apply(weights_init_normal)
    G_Y2X.apply(weights_init_normal)
    D_X.apply(weights_init_normal)
    D_Y.apply(weights_init_normal)
    
    device = 'cuda:0' if torch.cuda.is_available() else 'CPU'
    #move the models to gpu if available
    G_X2Y.to(device)
    G_Y2X.to(device)
    D_X.to(device)
    D_Y.to(device)
    
    return G_X2Y , G_Y2X , D_X , D_Y


G_X2Y , G_Y2X , D_X , D_Y = create_model(padding_mode='replicate')

输出

生成器和判别器的损失

CycleGAN 中包含的两个映射函数 G: X→Y 和 F: Y→X 连接着对抗判别器 DY 和 DX。（a）DY 促使 G 将 X 转换为与域 Y 的输出相同的输出，而 DX 和 F 则做同样的事情。
我们向映射添加了两个循环一致性损失，以进一步规范它们。这些损失表示，如果我们从一个域翻译到另一个域然后再返回，我们应该回到起点。有两种类型的循环一致性损失：（b）前向循环和（c）后向循环。

最小二乘 GAN

如前所述，常规 GAN 使用 sigmoid 交叉熵损失函数将判别器视为分类器。然而，在学习阶段，此损失函数可能会导致梯度消失问题。我们将为判别器使用最小二乘损失函数来解决此问题。此结构也称为 LSGAN，即最小二乘 GAN。

判别器损失

判别器损失定义为判别器的输出（图像）与目标值之间的均方误差，目标值可以是 0 或 1，具体取决于判别器应该将图像分类为真实还是伪造。例如，使用均方误差，我们可以通过检查 DX 在识别真实图像 x 时的接近程度来训练它。

out_x = D_X(x)
real_err = torch.mean((out_x-1)**2)

生成器损失

生成器损失计算过程将包含与判别器损失计算过程相似的阶段；这些过程包括创建看起来属于 X 图像集但实际上基于 Y 真实图像的伪造图像，反之亦然。这次，你的生成器试图让判别器将这些伪造图像识别为真实图像，因此你将通过检查判别器对这些伪造图像的应用来计算这些伪造图像的“真实损失”。

循环一致性损失

除了对抗损失之外，生成器损失项还将包含循环一致性损失。此损失是用于评估重建图像质量与原始图像质量的指标。

假设你有一个生成的伪造图像 x_hat 和一张真实图像 y。应用 G_XtoY(x_hat) = y_hat 将得到一个重建的 y_hat。然后，你可以验证这个重建的 y_hat 与原始图像 y 是否匹配。为此，我们建议计算原始图像和重建图像之间的 L1 损失——绝对差值。为了强调此损失的重要性，你还可以选择将其乘以权重值 lambda_weight。

生成器损失总额将由生成器损失以及前向和后向循环中的一致性损失的总和决定。

criterion_Idt = nn.L1Loss()
#define loss functions which are helpful
def real_mse_loss(D_out):
    # How close is the produced output from being "real"?
    return torch.mean((D_out -1)**2)

def fake_mse_loss(D_out):
    # How close is the produced output from being "false"?
    return torch.mean(D_out**2)
    

def cycle_consistency_loss(real_im, reconstructed_im, lambda_weight):
    # calculate reconstruction loss 
    # return weighted loss
    reconstruct_loss = torch.mean(torch.abs(real_im-reconstructed_im))
    return reconstruct_loss*lambda_weight

def identity_loss(lambda_idt , idt_B ,real_B , idt_A , real_A , lambda_weight):
    
    if lambda_idt > 0:
        # G_A should be identity if real_B is fed: ||G_A(B) - B||
        loss_idt_A = criterion_Idt(idt_B, real_B) * lambda_weight * lambda_idt
        # G_B should be identity if real_A is fed: ||G_B(A) - A||
        loss_idt_B = criterion_Idt(idt_A, real_A) * lambda_weight * lambda_idt
    else:
        self.loss_idt_A = 0
        self.loss_idt_B = 0
        
    return loss_idt_A + loss_idt_B

定义优化器

#define the optimizers
import torch.optim as optim
from torch.optim.lr_scheduler import MultiStepLR

# hyperparams for Adam optimizers
lr=0.0002
beta1=0.5
beta2= 0.999

g_params = list(G_X2Y.parameters()) + list(G_Y2X.parameters())  # Get generator parameters

# Create optimizers for the generators and discriminators
g_optimizer   = optim.Adam(g_params, lr, [beta1, beta2])
d_x_optimizer = optim.Adam(D_X.parameters(), lr, [beta1, beta2])
d_y_optimizer = optim.Adam(D_Y.parameters(), lr, [beta1, beta2])

lambda1 = 3500
lambda2 = 8500

#define lr scheduler
#g_lr_scheduler = MultiStepLR(g_optimizer, milestones=[lambda1,lambda2], gamma=0.1)
#dx_lr_scheduler =MultiStepLR(d_x_optimizer, milestones=[lambda1,lambda2], gamma=0.1)
#dy_lr_scheduler = MultiStepLR(d_y_optimizer, milestones=[lambda1,lambda2], gamma=0.1)

训练

CycleGAN 在看到 X 和 Y 集合的一个真实图像批次后，通过执行以下操作进行训练：

判别器训练

使用真实图像确定判别器 DX 损失。
以域 Y 中的真实图像为基础，创建模仿域 X 中图像的伪造图像。
计算 DX 的伪造损失。
计算总损失，并执行 DX 优化和反向传播。
重复步骤 1-4，但这次使用 DY 并交换你的域！

生成器训练

以域 Y 中的真实图像为基础，创建模仿域 X 中图像的伪造图像。
通过计算 DX 对伪造 X 的响应来确定生成器损失。
使用步骤 1 中创建的伪造 X 图像，创建重建的 Y 图像。
通过比较重建图像和真实 Y 图像来确定循环一致性的下降。
重复步骤 1-4，但交换域
计算所有重建和生成器损失，然后执行反向传播 + 优化。

# train the network
def training_loop(dataloader_X, dataloader_Y, test_dataloader_X , test_dataloader_Y, 
                  n_epochs=10000):
    
    print_every=50
    checkpoint_every=1000
    # keep track of losses over time
    losses = []

    test_iter_X = iter(dataloader_X)
    test_iter_Y = iter(dataloader_Y)

    # Get some fixed data from domains X and Y for sampling. These are images that are held
    # constant throughout training, that allows us to inspect the model's performance.
    fixed_X = test_iter_X.next()[0]
    fixed_Y = test_iter_Y.next()[0]
    fixed_X = rescale(fixed_X) # Make sure to scale to a range -1 to 1
    fixed_Y = rescale(fixed_Y)

    # batches per epoch
    iter_X = iter(dataloader_X)
    iter_Y = iter(dataloader_Y)
    batches_per_epoch = min(len(iter_X), len(iter_Y))

    for epoch in range(1, n_epochs+1):
        
        
        # Reset iterators for each epoch
        if epoch % batches_per_epoch == 0:
            iter_X = iter(dataloader_X)
            iter_Y = iter(dataloader_Y)

        images_X = iter_X.next()
        images_X = rescale(images_X) # make sure to scale to a range -1 to 1

        images_Y = iter_Y.next()
        images_Y = rescale(images_Y)
        
        # move images to GPU if available (otherwise stay on CPU)
        device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
        images_X = images_X.to(device)
        images_Y = images_Y.to(device)


        # ============================================
        #            TRAIN THE DISCRIMINATORS
        # ============================================

        ##   First: D_X, real and fake loss components   ##

        # 1. Compute the discriminator losses on real images in the X domain
        Dx_real_out = D_X(images_X)
        Dx_real_loss = real_mse_loss(Dx_real_out)
        
        # 2. Generate fake images that look like domain X based on real images in domain Y
        Gx_fake = G_Y2X(images_Y)
        
        # 3. Compute the fake loss for D_X on fake X
        Dx_fake_out= D_X(Gx_fake)
        Dx_fake_loss = fake_mse_loss(Dx_fake_out)
        
        # 4. Compute the total loss and perform backdrop
        d_x_loss = Dx_real_loss + Dx_fake_loss
        #reste the optimizer
        d_x_optimizer.zero_grad()
        #backprop the loss
        d_x_loss.backward()
        #optimze the descrimintaer x
        d_x_optimizer.step()

        
        ##   Second: D_Y, real and fake loss components   ##
        Dy_real_out = D_Y(images_Y)
        Dy_real_loss = real_mse_loss(Dy_real_out)
        
        #generate fake images that look like the domain y on the real images in domain x
        Gy_fake = G_X2Y(images_X)
        
        #compute the decrementer fake loss
        Dy_fake_out = D_Y(Gy_fake)
        Dy_fake_loss = fake_mse_loss(Dy_fake_out)
        
        d_y_loss = Dy_fake_loss + Dy_real_loss
        #reset the optimizer
        d_y_optimizer.zero_grad()
        #backprop the loss
        d_y_loss.backward()
        #optimize the model
        d_y_optimizer.step()
        


        # =========================================
        #            TRAIN THE GENERATORS
        # =========================================

        ##    First: generate fake X images and reconstructed Y images    ##

        # 1. Generate fake images that look like domain X based on real images in domain Y
        Gx_fake = G_Y2X(images_Y)

        # 2. Compute the generator loss based on domain X
        Dx_fake_out = D_X(Gx_fake)
        Gx_fake_loss = real_mse_loss(Dx_fake_out)

        # 3. Create a reconstructed y
        Gy_fake_re =  G_X2Y(Gx_fake) 
        
        # 4. Compute the cycle consistency loss (the reconstruction loss)
        Gy_reconstruct_loss = cycle_consistency_loss(images_Y , Gy_fake_re , lambda_weight=10)

        ##    Second: generate fake Y images and reconstructed X images    ##
        Gy_fake = G_X2Y(images_X)
        
        #compute the generator fake loss on the discriminator
        Dy_fake_out = D_Y(Gy_fake)
        Gy_fake_loss = real_mse_loss(Dy_fake_out)
        
        #regenerate the x images from fake y domain images
        Gx_fake_re = G_Y2X(Gy_fake)
        
        #compute the cycle consistent loss on the real x and reconstructed x images
        Gx_reconstuct_loss = cycle_consistency_loss(images_X , Gx_fake_re , lambda_weight=10)
        
        #compute the identity loss
        idt_B = G_X2Y(images_Y)
        idt_A = G_Y2X(images_X)
        id_loss = identity_loss(0.1 , idt_B , images_Y ,
                                idt_A , images_X , lambda_weight=8)

        # 5. Add up all generator and reconstructed losses and perform backdrop
        g_total_loss = Gx_fake_loss + Gy_fake_loss + Gy_reconstruct_loss + Gx_reconstuct_loss + id_loss
        
        #optimize the generator model
        g_optimizer.zero_grad()
        #back prop the loss
        g_total_loss.backward()
        #optimize the generator model
        g_optimizer.step()

        
        # Print the log info
        if epoch % print_every == 0:
            # append real and fake discriminator losses and the generator loss
            losses.append((d_x_loss.item(), d_y_loss.item(), g_total_loss.item()))
            print('Epoch [{:5d}/{:5d}] | d_X_loss: {:6.4f} | d_Y_loss: {:6.4f} | g_total_loss: {:6.4f}'.format(
                    epoch, n_epochs, d_x_loss.item(), d_y_loss.item(), g_total_loss.item()))

            
        sample_every=1000
        # Save the generated samples
        if epoch % sample_every == 0:
            G_Y2X.eval() # Set generators to the eval mode for sample generation
            G_X2Y.eval()
            save_samples(epoch, fixed_Y, fixed_X, G_Y2X, G_X2Y, batch_size=16)
            G_Y2X.train()
            G_X2Y.train()
        
        #g_lr_scheduler.step()
        #dx_lr_scheduler.step()
        #dy_lr_scheduler.step()
        # Uncomment these lines, if you want to save your model
#         checkpoint_every=1000
#         # Save the model parameters
        #if epoch % checkpoint_every == 0:
            #checkpoint(epoch, G_XtoY, G_YtoX, D_X, D_Y)

    return losses

辅助函数

# Helper functions for saving sample data and models

# import data-loading libraries
import os
import pdb
import pickle
import argparse

import warnings
warnings.filterwarnings("ignore")

# import torch
import torch


# numpy & scipy imports
import numpy as np
import scipy
import scipy.misc


def checkpoint(iteration, G_XtoY, G_YtoX, D_X, D_Y, checkpoint_dir='checkpoints_cyclegan'):
    """Saves the parameters of both generators G_YtoX, G_XtoY, and discriminators D_X, D_Y.
        """
    G_XtoY_path = os.path.join(checkpoint_dir, 'G_XtoY.pkl')
    G_YtoX_path = os.path.join(checkpoint_dir, 'G_YtoX.pkl')
    D_X_path = os.path.join(checkpoint_dir, 'D_X.pkl')
    D_Y_path = os.path.join(checkpoint_dir, 'D_Y.pkl')
    torch.save(G_XtoY.state_dict(), G_XtoY_path)
    torch.save(G_YtoX.state_dict(), G_YtoX_path)
    torch.save(D_X.state_dict(), D_X_path)
    torch.save(D_Y.state_dict(), D_Y_path)


def merge_images(sources, targets, batch_size=16):
    """Creates a grid consisting of pairs of columns, where the first column in
        each pair contains images source images and the second column in each pair
        contains images generated by the CycleGAN from the corresponding images in
        the first column.
        """
    _, _, h, w = sources.shape
    row = int(np.sqrt(batch_size))
    merged = np.zeros([3, row*h, row*w*2])
    for idx, (s, t) in enumerate(zip(sources, targets)):
        i = idx // row
        j = idx % row
        merged[:, i*h:(i+1)*h, (j*2)*h:(j*2+1)*h] = s
        merged[:, i*h:(i+1)*h, (j*2+1)*h:(j*2+2)*h] = t
    merged = merged.transpose(1, 2, 0)
    return merged
    

def to_data(x):
    """Converts variable to numpy."""
    if torch.cuda.is_available():
        x = x.cpu()
    x = x.squeeze(0)
    x = x.data.numpy()
    x = np.transpose(x,(1,2,0))
    x = ((x +1)*255 / (2)).astype(np.uint8) # rescale to 0-255
    return x

def save_samples(iteration, fixed_Y, fixed_X, G_YtoX, G_XtoY, batch_size=16, sample_dir='samples_cyclegan'):
    """Saves samples from both generators X->Y and Y->X.
        """
    # move input data to correct device
    device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")

    fake_X = G_YtoX(fixed_Y.unsqueeze(0).to(device))
    fake_Y = G_XtoY(fixed_X.unsqueeze(0).to(device))
    
    X, fake_X = to_data(fixed_X), to_data(fake_X)
    Y, fake_Y = to_data(fixed_Y), to_data(fake_Y)
    
    #merged = merge_images(X, fake_Y, batch_size)
    #path = os.path.join(sample_dir, 'sample-{:06d}-X-Y.png'.format(iteration))
    #scipy.misc.imsave(path, merged)
    #print('Saved {}'.format(path))
    viz_sample(X,fake_Y)
    #viz_sample(Y,fake_X)
    #return fake_Y

训练和损失模式

找到理想的超参数，使得判别器和生成器不会互相压倒，这需要大量的反复试验。我建议阅读这篇 DCGAN 研究以及原始的 CycleGAN 论文，看看他们是如何做的。查看现有论文以了解早期研究中哪些有效通常是个好主意。之后，你将有一个坚实的基础来测试你自己的实验。

判别器损失

请记住，我们正在尝试创建一个能够生成高质量“伪造”图像的模型，因此当你绘制生成器和判别器损失时，你应该注意到总有一些判别器损失。因此，总会有一些损失，因为完美的判别器将无法区分真实图像和伪造图像。此外，你应该注意到 DX 和 DY 的损失水平大致相同。如果不是这样，这表明你的训练中偏向于某一种判别器，你可能需要检查你的模型或数据中的偏差。

生成器损失

由于生成器损失同时考虑了生成器损失和加权的重建误差，因此它应该比判别器损失高得多。由于最初生成的图像往往离好的伪造相去甚远，你应该注意到在训练初期损失会大幅下降。随着训练的进行，判别器和生成器都会进步，因此通常在一段时间后会趋于平稳。如果你注意到损失随着时间的推移波动很大，可以尝试调整循环一致性损失的权重，使其稍多或稍少，或者降低学习率。

n_epochs = 10000 # Keep this small when testing if a model first works, then increase it to >=1000

losses = training_loop(dataloader_X, dataloader_Y, dataloader_X, dataloader_Y, n_epochs=n_epochs)

输出

转换可视化

test_iter_X = iter(dataloader_X)
test_iter_Y = iter(dataloader_Y)

# Get some fixed data from domains X and Y for sampling. These are images that are held
# constant throughout training, that allows us to inspect the model's performance.
fixed_X = test_iter_X.next()
fixed_Y = test_iter_Y.next()
fixed_X = rescale(fixed_X) # Make sure to scale to a range -1 to 1
fixed_Y = rescale(fixed_Y)
for fix_x , fix_y in zip(fixed_X , fixed_Y):
    save_samples(0, fix_y, fix_x, G_Y2X, G_X2Y, batch_size=1)

输出

转换翻译后，我们可以看到伪造图像得到了改进。

模型似乎在每个 epoch 中都显示出判别器（d_X_loss, d_Y_loss）和生成器（g_total_loss）的损失下降，这可能表明性能尚可。

注意：但是，仅凭这些损失无法精确确定模型的有效性。

下一个主题DNN 机器学习

CycleGAN

CycleGAN 的实现

导入库

加载数据集

可视化数据

缩放

定义模型

判别器

卷积辅助函数

判别器架构

生成器

残差块类

残差函数

定义 ResidualBlock 类

转置卷积辅助函数

生成器架构

完成网络

生成器和判别器的损失

最小二乘 GAN

判别器损失

生成器损失

循环一致性损失

定义优化器

训练

判别器训练

生成器训练

辅助函数

训练和损失模式

判别器损失

生成器损失

转换可视化

注意：但是，仅凭这些损失无法精确确定模型的有效性。

联系信息

关注我们

教程

面试题

在线编译器

Python

Java

.Net Framework

AI, ML and Data Science

Cloud Technology

B.Tech and MCA

Web Technology

PHP

Software Testing

Technical Interview

Java Interview

Python

Web Interview

Database Interview

B.Tech / MCA

Important Interview

Software Testing Interview

Company Interviews

Online Compilers

Multiple Choice Questions

机器学习

监督式学习

分类

杂项

相关教程

面试题

CycleGAN

CycleGAN 的实现

导入库

加载数据集

可视化数据

缩放

定义模型

判别器

卷积辅助函数

判别器架构

生成器

残差块类

残差函数

定义 ResidualBlock 类

转置卷积辅助函数

生成器架构

完成网络

生成器和判别器的损失

最小二乘 GAN

判别器损失

生成器损失

循环一致性损失

定义优化器

训练

判别器训练

生成器训练

辅助函数

训练和损失模式

判别器损失

生成器损失

转换可视化

注意：但是，仅凭这些损失无法精确确定模型的有效性。

相关帖子

StyleGAN

图像分割的平均交并比 (mIoU)

为什么每次在机器学习中得到的结果都不同

局部加权线性回归

深度学习中缩放数据、模型大小和性能

机器学习中的梯度下降

CNB 算法

归纳学习与转导学习的区别

机器学习中的 Epoch

时间序列 - 指数平滑

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器