利用pytorch对加噪堆叠自编码器在MNIST数据集举行训练和验证 ...

实现背景：

最近在复现关于使用深度学习提取特征来举行聚类的论文，其中使用到了加噪堆叠自编码器，详细实现细节请参考论文：Improved Deep Embedded Clustering with Local Structure Preservation
其中加噪堆叠自编码器涉及两个过程：
预训练：预训练过程对原始数据加噪，贪心式地对每一层encoder和decoder举行训练，其中训练新的AE时冻结前面训练好的AE。详见：堆栈自编码器 Stacked AutoEncoder-CSDN博客
微调：在预训练完成之后使用所有AE和原始数据对整体模型举行微调。
我在网上找到了一个SAE的示范样例：python-pytorch 利用pytorch对堆叠自编码器举行训练和验证_pytoch把训练和验证写一起的代码-CSDN博客
但是这篇博客的数据集很小，如果应用到MNIST数据集时显存很容易溢出，因此我在原始的基础上举行了改进，直接上代码：
初始化数据集：

1. import torch
2. from torchvision import datasets, transforms
3. from torch.utils.data import Dataset, random_split
4. # 定义数据预处理
5. transform = transforms.Compose([
6.     transforms.ToTensor(),
7.     transforms.Normalize((0.1307,), (0.3081,))
8. ])
10. # 下载并加载 MNIST 训练数据集
11. original_dataset = datasets.MNIST(root='./data', train=True,
12.                                   download=False, transform=transform)
15. class NoLabelDataset(Dataset):
16.     def __init__(self, original_dataset):
17.         self.original_dataset = original_dataset
19.     def __getitem__(self, index):
20.         image, _ = self.original_dataset[index]
21.         return image
23.     def __len__(self):
24.         return len(self.original_dataset)
27. # 创建不包含标签的数据集
28. no_label_dataset = NoLabelDataset(original_dataset)
30. # 划分训练集和验证集
31. train_size = int(0.8 * len(no_label_dataset))
32. val_size = len(no_label_dataset) - train_size
33. train_dataset, val_dataset = random_split(no_label_dataset, [train_size, val_size])
35. # 创建数据加载器
36. train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=32, shuffle=True)
37. val_loader = torch.utils.data.DataLoader(val_dataset, batch_size=32, shuffle=False)
39. print(f"训练集样本数量: {len(train_dataset)}")
40. print(f"验证集样本数量: {len(val_dataset)}")   
41. device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

复制代码

 定义模型和训练函数：

1. import torch.nn as nn
3. class Autoencoder(nn.Module):
4.     def __init__(self, input_size, hidden_size):
5.         super(Autoencoder, self).__init__()
6.         self.encoder = nn.Sequential(
7.             nn.Conv2d(input_size, hidden_size, kernel_size=3, stride=1, padding=1),  # 输入通道1，输出通道16
8.             nn.ReLU())
9.         self.decoder = nn.Sequential(
10.             nn.ConvTranspose2d(hidden_size, input_size, kernel_size=3, stride=1, padding=1),
11.             nn.ReLU())
13.     def forward(self, x):
14.         x = self.encoder(x)
15.         x = self.decoder(x)
16.         return x
18. def train_ae(models, train_loader, val_loader, num_epochs, criterion, optimizer, noise_factor, finetune):
19.     device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
20.     for epoch in range(num_epochs):
21.         # Training
22.         models[-1].train()
23.         train_loss = 0
24.         for batch_data in train_loader:
25.             optimizer.zero_grad()
26.             if len(models) != 1:
27.                 batch_data = batch_data.to(device)
28.                 for model in models[:-1]:
29.                     with torch.no_grad():
30.                         batch_data = model.encoder(batch_data)
31.                 batch_data = batch_data.detach()
32.             if finetune == True:
33.                 batch_data = batch_data.to(device)
34.                 outputs = models[-1](batch_data)
35.                 loss = criterion(outputs, batch_data)
36.             else:
37.                 noisy_image = batch_data + noise_factor * torch.randn_like(batch_data)
38.                 noisy_image = torch.clamp(noisy_image, 0., 1.).to(device)
39.                 outputs = models[-1](noisy_image)
40.                 batch_data = batch_data.to(device)
41.                 loss = criterion(outputs, batch_data)
42.             loss.backward()
43.             optimizer.step()
44.             train_loss += loss.item()
45.         
46.         train_loss /= len(train_loader)
47.         print(f"Epoch {epoch+1}/{num_epochs}, Training Loss: {train_loss:.4f}")
49.         # Validation
50.         models[-1].eval()
51.         val_loss = 0
52.         with torch.no_grad():
53.             for batch_data in val_loader:
54.                 if len(models) != 1:
55.                     batch_data = batch_data.to(device)
56.                     for model in models[:-1]:
57.                         batch_data = model.encoder(batch_data)
58.                     batch_data = batch_data.detach()
59.                 if finetune == True:
60.                     batch_data = batch_data.to(device)
61.                     outputs = models[-1](batch_data)
62.                     loss = criterion(outputs, batch_data)
63.                 else:
64.                     noisy_image = batch_data + noise_factor * torch.randn_like(batch_data)
65.                     noisy_image = torch.clamp(noisy_image, 0., 1.).to(device)
66.                     outputs = models[-1](noisy_image)
67.                     batch_data = batch_data.to(device)
68.                     loss = criterion(outputs, batch_data)
69.                 val_loss += loss.item()
71.         val_loss /= len(val_loader)
72.         print(f"Epoch {epoch+1}/{num_epochs}, Validation Loss: {val_loss:.4f}")

复制代码

 模型训练以及微调：

1. batch_size = 16
2. noise_factor = 0.4
5. ae1 = Autoencoder(input_size=1, hidden_size=16).to(device)
6. optimizer = torch.optim.Adam(ae1.parameters(), lr=0.001)
7. criterion = nn.MSELoss()
8. train_ae([ae1], train_loader, val_loader, 10, criterion, optimizer, noise_factor, finetune = False)
11. ae2 = Autoencoder(input_size=16, hidden_size=64).to(device)
12. optimizer = torch.optim.Adam(ae2.parameters(), lr=0.001)
13. train_ae([ae1, ae2], train_loader, val_loader, 10, criterion, optimizer, noise_factor, finetune = False)
16. ae3 = Autoencoder(input_size=64, hidden_size=128).to(device)
17. optimizer = torch.optim.Adam(ae3.parameters(), lr=0.001)
18. train_ae([ae1, ae2, ae3], train_loader, val_loader, 10, criterion, optimizer, noise_factor, finetune = False)
20. class StackedAutoencoder(nn.Module):
21.     def __init__(self, ae1, ae2, ae3):
22.         super(StackedAutoencoder, self).__init__()
23.         self.encoder = nn.Sequential(ae1.encoder, ae2.encoder, ae3.encoder)
24.         self.decoder = nn.Sequential(ae3.decoder, ae2.decoder, ae1.decoder)
26.     def forward(self, x):
27.         x = self.encoder(x)
28.         x = self.decoder(x)
29.         return x
31. sae = StackedAutoencoder(ae1, ae2, ae3)
33. optimizer = torch.optim.Adam(ae1.parameters(), lr=0.001)
34. criterion = nn.MSELoss()
35. train_ae([sae], train_loader, val_loader, 10, criterion, optimizer, noise_factor, finetune = True)

复制代码

 结果可视化：

1. import matplotlib.pyplot as plt
2. import numpy as np
3. dataiter = iter(val_loader)
4. image = next(dataiter)[1]
5. print(image.shape)
6. image = image.to(device)
9. # 通过自编码器模型进行前向传播
10. with torch.no_grad():
11.     output = sae(image)
13. noise_factor = 0.4
14. noisy_image = image + noise_factor * torch.randn_like(image)
15. noisy_image = torch.clamp(noisy_image, 0., 1.).cpu().numpy()
17. # 将张量转换为 numpy 数组以便可视化
18. image = image.cpu().numpy()
19. output = output.cpu().numpy()
21. # 定义一个函数来显示图片
22. def imshow(img):
23.     img = img * 0.3081 + 0.1307  # 反归一化
24.     npimg = img.squeeze()  # 去除单维度
25.     plt.imshow(npimg, cmap='gray')
27. # 可视化输入和输出图片
28. plt.figure(figsize=(10, 5))
30. # 显示输入图像
31. plt.subplot(1, 3, 1)
32. imshow(torch.from_numpy(image))
33. plt.title('Input Image')
34. plt.axis('off')
36. # 显示加噪图像
37. plt.subplot(1, 3, 2)
38. imshow(torch.from_numpy(noisy_image))
39. plt.title('Noisy Image')
40. plt.axis('off')
42. # 显示输出图像
43. plt.subplot(1, 3, 3)
44. imshow(torch.from_numpy(output))
45. plt.title('Output Image')
46. plt.axis('off')
48. plt.savefig("预训练图.svg", dpi=300,format="svg")

复制代码

 下面附上训练好的可视化图：

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。