机器学习中的生成模型

2025年2月28日 | 阅读 12 分钟

它是机器学习的一个子领域，模型的目的是生成与原始训练数据相似的新数据。与判别模型（传统上设计为从输入到输出的映射，例如，识别图像是否包含狗或猫）相反，生成模型试图逼近数据本身，即底层分布。这样，它们就有可能生成最有可能来自与训练样本相同分布的新样本。近年来，生成模型因其在图像合成、自然语言处理和药物发现等领域的成功应用而备受关注。本文将深入探讨生成建模的哲学基础，并重点介绍一些常见的生成模型类型及其在实践中的应用。

本质上，生成建模只是学习数据的联合概率分布 P(X)，其中 X 代表特征，或者更确切地说，是数据点。概率分布捕捉了数据中不同值和模式发生的可能性。一旦学习了这种分布，生成模型就可以通过从 P(X) 中抽取来用于生成新样本。这与判别模型形成对比，判别模型侧重于对条件概率 P(Y|X) 进行建模，其中 X 是输入数据，Y 是目标标签。虽然判别模型在分类任务上表现出色，但它们并不试图理解数据本身的结构。生成模型由于其更丰富的数据拟合能力而表现更好，因为它们主要关注数据点实际是如何生成的。适用于数据增强、合成和无监督学习等应用。

生成模型类型

有几种类型的生成模型，每种模型都有不同的学习和生成数据技术，以下是一些例子：

01. 生成对抗网络（GANs）

自 2014 年由 Ian Goodfellow 发明以来，GANs 彻底改变了机器学习领域，特别是生成建模领域。作为一类机器学习模型，GANs 主要用于生成新的、合成的数据，这些数据具有真实世界数据的特征。它在图像合成、视频生成、文本到图像翻译甚至科学研究等各种领域都有创新应用。

本质上，生成对抗网络是两个神经网络的交互：生成器和判别器，它们相互影响，但在目标上截然相反。这些网络进行一场零和博弈，其中生成器试图创建尽可能与真实数据相似的合成数据，而判别器则试图区分真实数据和生成数据（“假数据”）。

生成器网络：生成器通常接收从某种分布（例如高斯分布或均匀分布）采样的噪声，并输出看起来像现实世界数据的各种数据。生成器会尝试生成图像，尤其是在有明确定义目标的场景下。反过来，生成器网络会进行微调，使其能够“欺骗”判别器，让判别器接受生成的作为真实数据。通过找到原始数据中出现的模式和特征，它在生成相似输出方面变得更加出色。
判别器网络：判别器的作用是作为二元分类器，接受真实数据和假数据作为输入，然后尝试将每个实例分类为真实数据或生成器生成的数据。判别器基本上是一个评论家，它向生成器提供有关生成数据质量的反馈。如果判别器能够很好地、清晰地确定生成数据是真实的还是假的，那么生成器仍然需要一些训练。如果判别器无法区分它们，则意味着生成器正在改进。
对抗过程：GANs 从对抗性训练过程中学习。对抗性训练过程是指在训练过程中同时学习生成器和判别器。生成器试图通过欺骗判别器来输入更好的数据，而判别器则试图提高其检测假数据的能力。

现在，这里将展示 GAN 架构的代码。

 
# Architecture of Generator
class Generator(nn.Module):
    def __init__(self, input_dim, output_dim):
        super(Generator, self).__init__()
        self.model = nn.Sequential(
            nn.Linear(input_dim, 256),
            nn.ReLU(True),
            nn.Linear(256, 512),
            nn.ReLU(True),
            nn.Linear(512, 1024),
            nn.ReLU(True),
            nn.Linear(1024, output_dim),
            nn.Tanh()
        )

    def forward(self, x):
        return self.model(x)

# Architecture of  Discriminator
class Discriminator(nn.Module):
    def __init__(self, input_dim):
        super(Discriminator, self).__init__()
        self.model = nn.Sequential(
            nn.Linear(input_dim, 1024),
            nn.LeakyReLU(0.2, inplace=True),
            nn.Linear(1024, 512),
            nn.LeakyReLU(0.2, inplace=True),
            nn.Linear(512, 256),
            nn.LeakyReLU(0.2, inplace=True),
            nn.Linear(256, 1),
            nn.Sigmoid()
        )

    def forward(self, x):
        return self.model(x)

# Hyperparameters
latent_dim = 100  # Size of the noise vector
image_size = 28 * 28  # MNIST image size (28x28 pixels)
batch_size = 64
epochs = 50
learning_rate = 0.0002

# Data loader (MNIST dataset)
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize([0.5], [0.5])  # Normalize the images to [-1, 1] for Tanh activation
])

train_dataset = datasets.MNIST(root='./data', train=True, transform=transform, download=True)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

# Initialize generator and discriminator
generator = Generator(input_dim=latent_dim, output_dim=image_size)
discriminator = Discriminator(input_dim=image_size)

# Loss and optimizer
criterion = nn.BCELoss()
optimizer_g = optim.Adam(generator.parameters(), lr=learning_rate)
optimizer_d = optim.Adam(discriminator.parameters(), lr=learning_rate)

# Training the GAN
for epoch in range(epochs):
    for batch_idx, (real_images, _) in enumerate(train_loader):
        batch_size = real_images.size(0)
        real_images = real_images.view(batch_size, -1)  # Flatten images

        # Train Discriminator
        real_labels = torch.ones(batch_size, 1)  # Label for real images
        fake_labels = torch.zeros(batch_size, 1)  # Label for fake images

        outputs = discriminator(real_images)
        d_loss_real = criterion(outputs, real_labels)

        z = torch.randn(batch_size, latent_dim)  # Generate noise
        fake_images = generator(z)
        outputs = discriminator(fake_images.detach())  # Detach to avoid training the generator
        d_loss_fake = criterion(outputs, fake_labels)

        d_loss = d_loss_real + d_loss_fake
        optimizer_d.zero_grad()
        d_loss.backward()
        optimizer_d.step()

        # Train Generator
        outputs = discriminator(fake_images)
        g_loss = criterion(outputs, real_labels)  # Generator wants to fool the discriminator

        optimizer_g.zero_grad()
        g_loss.backward()
        optimizer_g.step()   

02. VAE

变分自编码器是其中一种生成模型，它们在机器学习领域取得了一些伟大的成就，显著加速了无监督学习和数据生成。本质上，VAE 代表了一种结构化的方法，用于对输入数据的潜在空间表示进行概率学习。VAE 由 Kingma 和 Welling 于 2013 年提出，因其能够对复杂的数据分布进行建模并与贝叶斯推理密切相关而广受欢迎。

可以将变分自编码器视为自编码器，但它是概率性的。常规自编码器将数据压缩到较低维度的空间，称为瓶颈或潜在空间，然后从这种压缩表示中重建数据。VAE 不同，它不那么注重学习从输入到潜在空间的确定性映射，而是对潜在空间上的概率分布进行建模，从而允许生成新的数据点。

现在，这里将展示 VAE 架构的代码。

 
# Encoder network
class Encoder(nn.Module):
    def __init__(self, input_dim, hidden_dim, latent_dim):
        super(Encoder, self).__init__()
        self.fc1 = nn.Linear(input_dim, hidden_dim)
        self.fc_mu = nn.Linear(hidden_dim, latent_dim)  # Mean of latent variable
        self.fc_logvar = nn.Linear(hidden_dim, latent_dim)  # Log variance of latent variable

    def forward(self, x):
        h = torch.relu(self.fc1(x))
        mu = self.fc_mu(h)
        logvar = self.fc_logvar(h)
        return mu, logvar

# Decoder network
class Decoder(nn.Module):
    def __init__(self, latent_dim, hidden_dim, output_dim):
        super(Decoder, self).__init__()
        self.fc1 = nn.Linear(latent_dim, hidden_dim)
        self.fc2 = nn.Linear(hidden_dim, output_dim)

    def forward(self, z):
        h = torch.relu(self.fc1(z))
        return torch.sigmoid(self.fc2(h))

# VAE model combining the Encoder and Decoder
class VAE(nn.Module):
    def __init__(self, input_dim, hidden_dim, latent_dim):
        super(VAE, self).__init__()
        self.encoder = Encoder(input_dim, hidden_dim, latent_dim)
        self.decoder = Decoder(latent_dim, hidden_dim, input_dim)

    def reparameterize(self, mu, logvar):
        std = torch.exp(0.5 * logvar)  # Standard deviation
        eps = torch.randn_like(std)  # Random noise
        return mu + eps * std  # Sampling with reparameterization trick

    def forward(self, x):
        mu, logvar = self.encoder(x)
        z = self.reparameterize(mu, logvar)
        recon_x = self.decoder(z)
        return recon_x, mu, logvar

# Loss function: Sum of loss in reconstruction and KL divergence
def loss_function(recon_x, x, mu, logvar):
    recon_loss = nn.functional.binary_cross_entropy(recon_x, x, reduction='sum')
    kl_divergence = -0.5 * torch.sum(1 + logvar - mu.pow(2) - logvar.exp())
    return recon_loss + kl_divergence

# Hyperparameters
input_dim = 28 * 28  # MNIST images are 28x28 pixels
hidden_dim = 400
latent_dim = 20
batch_size = 64
epochs = 10
learning_rate = 1e-3

# Data loading
transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize([0.5], [0.5])
])
train_dataset = datasets.MNIST(root='./data', train=True, transform=transform, download=True)
train_loader = DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

# Initialize the model, optimizer
vae = VAE(input_dim, hidden_dim, latent_dim)
optimizer = optim.Adam(vae.parameters(), lr=learning_rate)

# Training the VAE
for epoch in range(epochs):
    vae.train()
    train_loss = 0
    for batch_idx, (data, _) in enumerate(train_loader):
        data = data.view(-1, input_dim)  # Flatten the images
        optimizer.zero_grad()
        recon_batch, mu, logvar = vae(data)
        loss = loss_function(recon_batch, data, mu, logvar)
        loss.backward()
        train_loss += loss.item()
        optimizer.step()   

03. 自回归模型

基于过去数据预测未来值被称为自回归（AR）模型，它是统计学和机器学习中的一类重要模型。这类自回归模型的基础是时间序列或顺序数据中的未来观测值依赖于过去观测值。这种模型已广泛应用于金融、经济、自然语言处理和语音识别等许多领域。自回归模型假设当前的输出或数据点是先前数据点的某种线性组合，并添加了噪声。也就是说，模型根据其先前的值“回归”当前值。

现在，这里将展示自回归模型的架构代码。

 
# Define the AutoRegressive model using RNN
class ARModel(nn.Module):
    def __init__(self, input_size, hidden_size, output_size, num_layers=1):
        super(ARModel, self).__init__()
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        
        # RNN layer
        self.rnn = nn.RNN(input_size, hidden_size, num_layers, batch_first=True)
        
        # Output layer
        self.fc = nn.Linear(hidden_size, output_size)
    
    def forward(self, x, hidden):
        out, hidden = self.rnn(x, hidden)
        out = self.fc(out[:, -1, :])  # Output the last time step only
        return out, hidden
    
    def init_hidden(self, batch_size):
        return torch.zeros(self.num_layers, batch_size, self.hidden_size)

# Hyperparameters
input_size = 1
hidden_size = 50
output_size = 1
num_layers = 1
learning_rate = 0.01
num_epochs = 100

# Create synthetic data (sinusoidal time series for example)
time_steps = np.linspace(0, np.pi * 2, 100, dtype=np.float32)
data = np.sin(time_steps)

# Convert data to supervised learning problem for autoregressive task
def create_dataset(data, sequence_length):
    sequences = []
    labels = []
    for i in range(len(data) - sequence_length):
        sequences.append(data[i:i + sequence_length])
        labels.append(data[i + sequence_length])
    return np.array(sequences), np.array(labels)

sequence_length = 10
train_sequences, train_labels = create_dataset(data, sequence_length)
train_sequences = torch.tensor(train_sequences).unsqueeze(-1)  # Add feature dimension
train_labels = torch.tensor(train_labels)

# Optimizer
# Loss Function
# Model Initialisation
model = ARModel(input_size=input_size, hidden_size=hidden_size, output_size=output_size, num_layers=num_layers)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# Training the model
for epoch in range(num_epochs):
    model.train()
    hidden = model.init_hidden(batch_size=len(train_sequences))
    
    # Forward pass
    output, hidden = model(train_sequences, hidden)
    loss = criterion(output.squeeze(), train_labels)
    
    # Backward pass and optimization
    optimizer.zero_grad()
    loss.backward()
    optimizer.step()
    
    if (epoch+1) % 10 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.4f}')

# Testing the model with future predictions
model.eval()
test_inputs = train_sequences[-1].unsqueeze(0)  # Start with the last sequence in the training set
hidden = model.init_hidden(batch_size=1)
predicted_values = []

for _ in range(50):  # Predict the next 50 time steps
    with torch.no_grad():
        prediction, hidden = model(test_inputs, hidden)
        predicted_values.append(prediction.item())
        
        # Append the prediction to the test inputs to create an autoregressive loop
        test_inputs = torch.cat((test_inputs[:, 1:, :], prediction.unsqueeze(0).unsqueeze(-1)), dim=1)   

04. 流模型

流模型是一类非常有趣的深度生成模型，用于学习和生成复杂数据分布。与其他生成模型（如生成对抗网络或变分自编码器）不同，流模型可以提供数据的精确似然，并允许数据和潜在空间之间的双向映射。它们既是生成性的，因为它们允许生成新的数据集，又是可逆的，因为它们允许从潜在表示中重建数据。

流模型是生成模型，它们学习将简单的概率分布（例如高斯分布）转换为更复杂的分布：图像、音频或文本。这通过一系列称为归一化流的可逆变换来实现。它们是模型可以执行的从数据空间到潜在空间的映射。

现在，这里将展示流模型的架构代码。

 
# Define coupling layer (RealNVP transformation)
class RealNVPCouplingLayer(nn.Module):
    def __init__(self, in_features, hidden_features):
        super(RealNVPCouplingLayer, self).__init__()
        self.scale_net = nn.Sequential(
            nn.Linear(in_features // 2, hidden_features),
            nn.ReLU(),
            nn.Linear(hidden_features, hidden_features),
            nn.ReLU(),
            nn.Linear(hidden_features, in_features // 2),
            nn.Tanh()
        )
        self.translate_net = nn.Sequential(
            nn.Linear(in_features // 2, hidden_features),
            nn.ReLU(),
            nn.Linear(hidden_features, hidden_features),
            nn.ReLU(),
            nn.Linear(hidden_features, in_features // 2)
        )

    def forward(self, x, reverse=False):
        x1, x2 = x.chunk(2, dim=1)
        scale = self.scale_net(x1)
        translate = self.translate_net(x1)
        
        if reverse:
            x2 = (x2 - translate) * torch.exp(-scale)
        else:
            x2 = x2 * torch.exp(scale) + translate
        
        return torch.cat([x1, x2], dim=1), scale.sum(dim=1)

# Define RealNVP model (stack of coupling layers)
class RealNVP(nn.Module):
    def __init__(self, in_features, hidden_features, num_coupling_layers):
        super(RealNVP, self).__init__()
        self.coupling_layers = nn.ModuleList([
            RealNVPCouplingLayer(in_features, hidden_features)
            for _ in range(num_coupling_layers)
        ])
        self.base_dist = MultivariateNormal(torch.zeros(in_features), torch.eye(in_features))

    def forward(self, x, reverse=False):
        log_det_jacobian = 0
        for i, layer in enumerate(self.coupling_layers):
            x, log_scale = layer(x, reverse=reverse)
            log_det_jacobian += log_scale
        return x, log_det_jacobian

    def sample(self, num_samples):
        z = self.base_dist.sample((num_samples,))
        x, _ = self.forward(z, reverse=True)
        return x

    def log_prob(self, x):
        z, log_det_jacobian = self.forward(x, reverse=False)
        return self.base_dist.log_prob(z) + log_det_jacobian

# Training the RealNVP model
def train(model, optimizer, data, epochs=100):
    for epoch in range(epochs):
        optimizer.zero_grad()
        log_prob = model.log_prob(data).mean()
        loss = -log_prob
        loss.backward()
        optimizer.step()
        if (epoch + 1) % 10 == 0:
            print(f'Epoch [{epoch+1}/{epochs}], Loss: {loss.item()}')

# Example usage
if __name__ == '__main__':
    in_features = 2  # Dimension of input data
    hidden_features = 256  # Number of hidden units
    num_coupling_layers = 5  # Number of coupling layers

    # Create RealNVP model
    model = RealNVP(in_features, hidden_features, num_coupling_layers)
    optimizer = optim.Adam(model.parameters(), lr=1e-3)

    # Generate some random 2D data for training
    data = torch.randn(1000, 2)

    # Train the model
    train(model, optimizer, data, epochs=100)   

05. Transformer 模型

Transformer 模型彻底改变了自然语言处理、计算机视觉等领域的机器学习，包括顺序数据的建模。Transformer 由 Vaswani 等人在 2017 年的论文“Attention Is All You Need”中提出，它已成为 BERT、GPT 和 T5 等最先进模型的基础，在文本生成、翻译和图像处理等大多数任务中都提供卓越的性能。

现在，这里将展示 Transformer 模型的架构代码。

 
# Scaled Dot-Product Attention
class ScaledDotProductAttention(nn.Module):
    def __init__(self, d_k):
        super(ScaledDotProductAttention, self).__init__()
        self.d_k = d_k

    def forward(self, Q, K, V, mask=None):
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.d_k)
        if mask is not None:
            scores = scores.masked_fill(mask == 0, -1e9)
        attention = F.softmax(scores, dim=-1)
        output = torch.matmul(attention, V)
        return output, attention

# Multi-Head Attention
class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, num_heads):
        super(MultiHeadAttention, self).__init__()
        self.d_model = d_model
        self.num_heads = num_heads
        self.d_k = d_model // num_heads

        self.W_q = nn.Linear(d_model, d_model)
        self.W_k = nn.Linear(d_model, d_model)
        self.W_v = nn.Linear(d_model, d_model)
        self.fc = nn.Linear(d_model, d_model)

    def forward(self, Q, K, V, mask=None):
        batch_size = Q.size(0)
        
        Q = self.W_q(Q).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        K = self.W_k(K).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)
        V = self.W_v(V).view(batch_size, -1, self.num_heads, self.d_k).transpose(1, 2)

        output, attention = ScaledDotProductAttention(self.d_k)(Q, K, V, mask)
        output = output.transpose(1, 2).contiguous().view(batch_size, -1, self.d_model)
        output = self.fc(output)

        return output, attention

# Position-wise Feed Forward Network
class FeedForward(nn.Module):
    def __init__(self, d_model, d_ff):
        super(FeedForward, self).__init__()
        self.fc1 = nn.Linear(d_model, d_ff)
        self.fc2 = nn.Linear(d_ff, d_model)

    def forward(self, x):
        return self.fc2(F.relu(self.fc1(x)))

# Positional Encoding
class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super(PositionalEncoding, self).__init__()
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len).unsqueeze(1)
        term_div = torch.exp(torch.arange(0, d_model, 2) * -(math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(position * term_div)
        pe[:, 1::2] = torch.cos(position * term_div)
        pe = pe.unsqueeze(0)
        self.register_buffer('pe', pe)

    def forward(self, x):
        x = x + self.pe[:, :x.size(1)]
        return x

# Transformer Encoder Layer
class TransformerEncoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff):
        super(TransformerEncoderLayer, self).__init__()
        self.self_attn = MultiHeadAttention(d_model, num_heads)
        self.feed_forward = FeedForward(d_model, d_ff)
        self.layernorm1 = nn.LayerNorm(d_model)
        self.layernorm2 = nn.LayerNorm(d_model)

    def forward(self, x, mask=None):
        output_attn, _ = self.self_attn(x, x, x, mask)
        x = self.layernorm1(x + output_attn)
        ff_output = self.feed_forward(x)
        x = self.layernorm2(x + ff_output)
        return x

# Transformer Decoder Layer
class TransformerDecoderLayer(nn.Module):
    def __init__(self, d_model, num_heads, d_ff):
        super(TransformerDecoderLayer, self).__init__()
        self.self_attn = MultiHeadAttention(d_model, num_heads)
        self.cross_attn = MultiHeadAttention(d_model, num_heads)
        self.feed_forward = FeedForward(d_model, d_ff)
        self.layernorm1 = nn.LayerNorm(d_model)
        self.layernorm2 = nn.LayerNorm(d_model)
        self.layernorm3 = nn.LayerNorm(d_model)

    def forward(self, x, enc_output, src_mask=None, tgt_mask=None):
        output_attn, _ = self.self_attn(x, x, x, tgt_mask)
        x = self.layernorm1(x + output_attn)

        cross_output_attn, _ = self.cross_attn(x, enc_output, enc_output, src_mask)
        x = self.layernorm2(x + cross_output_attn)

        ff_output = self.feed_forward(x)
        x = self.layernorm3(x + ff_output)

        return x

# Transformer Model
class Transformer(nn.Module):
    def __init__(self, src_vocab_size, tgt_vocab_size, d_model=512, num_heads=8, num_encoder_layers=6, num_decoder_layers=6, d_ff=2048, max_len=100):
        super(Transformer, self).__init__()
        self.src_embedding = nn.Embedding(src_vocab_size, d_model)
        self.tgt_embedding = nn.Embedding(tgt_vocab_size, d_model)
        self.positional_encoding = PositionalEncoding(d_model, max_len)

        self.encoder_layers = nn.ModuleList([TransformerEncoderLayer(d_model, num_heads, d_ff) for _ in range(num_encoder_layers)])
        self.decoder_layers = nn.ModuleList([TransformerDecoderLayer(d_model, num_heads, d_ff) for _ in range(num_decoder_layers)])

        self.fc_out = nn.Linear(d_model, tgt_vocab_size)

    def forward(self, src, tgt, src_mask=None, tgt_mask=None):
        src = self.src_embedding(src) * math.sqrt(self.d_model)
        tgt = self.tgt_embedding(tgt) * math.sqrt(self.d_model)

        src = self.positional_encoding(src)
        tgt = self.positional_encoding(tgt)

        for encoder_layer in self.encoder_layers:
            src = encoder_layer(src, src_mask)

        for decoder_layer in self.decoder_layers:
            tgt = decoder_layer(tgt, src, src_mask, tgt_mask)

        output = self.fc_out(tgt)
        return output   

下一主题机器学习中的归一化

机器学习中的生成模型

生成模型类型

01. 生成对抗网络（GANs）

02. VAE

03. 自回归模型

04. 流模型

05. Transformer 模型

联系信息

关注我们

教程

面试题

在线编译器

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

机器学习

监督式学习

分类

杂项

相关教程

面试题

机器学习中的生成模型

生成模型类型

01. 生成对抗网络（GANs）

02. VAE

03. 自回归模型

04. 流模型

05. Transformer 模型

相关帖子

连续值预测

Kubernetes 资源模型 (KRM) 以及如何使用 YAML

机器学习中的互信息

机器学习必备数学知识

时间序列中的季节性

什么是序数数据？

乘数法

什么是 1 维卷积神经网络

NTLK 语料库

UCI 机器学习仓库

订阅 Tpoint Tech

联系信息

关注我们

教程

面试题

在线编译器