Pytorch教程——卷积神经网络识别Fashion-MNIST
针对使用卷积神经网络进行图像分类的问题，下面会使用Pytorch搭建一个类似LeNet-5的网络结构，用于Fashion-MNIST数据集的图像分类。针对该问题的分析可以分为数据准备、模型建立以及使用训练集进行训练与使用测试集测试模型的效果。针对卷积网络的建立，将会分别建立常用的卷积神经网络与基于空洞卷积的卷积神经网络。首先导入所需要的库及相关模块。
import pandas as pd
from sklearn.metrics import accuracy_score,confusion_matrix,classification_report
import matplotlib.pyplot as plt
import seaborn as sns
import copy
import numpy as np
import time
import torch
import torch.nn as nn
from torch.optim import Adam
import torch.utils.data as Data
from torchvision import transforms
from torchvision.datasets import FashionMNIST
1、图像数据准备
首先准备FashionMNIST数据集，该数据集可以直接使用torchvision库中datasets模块的FashionMNIST()的API函数读取，如果指定的工作文件夹中没有当前的数据，可以从网络上自动下载该数据集，数据的准备程序如下所示：
# 使用 FashionMNIST 数据，准备训练数据集
train_data = FashionMNIST(
root = '...data/FashionMNIST', #数据集路径
train = True, #只使用训练数据集
transform = transforms.ToTensor(),
download = False
)
# 定义一个数据加载器
train_loader = Data.DataLoader(
dataset = train_data, # 使用的数据集
batch_size = 64, # 批量处理样本大小
shuffle = False, # 每次迭代前不打乱数据
num_workers = 2, # 使用两个进程
)
# 计算train_loader有多少个batch
print("train_loader的batch数量为:",len(train_loader))
Out[1]：train_loader的batch数量为: 938
上面的程序导入了训练数据集，然后使用Data.DataLoader()函数将其定义为数据加载器，每个batch中会包含64个样本，通过len()函数可以计算数据加载器中包含的batch数量，输出显示train_loader中包含938个batch。需要注意的是参数shuffle = False，表示加载器中每个batch使用的样本都是固定的，这样有利于在训练模型时根据迭代的次数将其切分为训练集和验证集。
为了观察数据集中每个图像的内容，可以获取一个batch的图像，然后将其可视化，以观察数据。获取数据并可视化的程序如下所示：
# 获得一个batch的数据
for step, (b_x, b_y) in enumerate(train_loader):
if step > 0:
break
# 可视化一个 batch 的图像
batch_x = b_x.squeeze().numpy()
batch_y = b_y.numpy()
class_label = train_data.classes
class_label[0] = 'T-shirt'
plt.figure(figsize = (12, 5))
for ii in np.arange(len(batch_y)):
plt.subplot(4, 16, ii + 1)
plt.imshow(batch_x[ii, :, :], cmap = plt.cm.gray)
plt.title(class_label[batch_y[ii]], size = 9)
plt.axis('off')
plt.subplots_adjust(wspace = 0.05)
在上面的程序中，使用for循环获取一个batch的数据b_x和b_y，并使用XX.numpy()方法将张量数据XX转化为Numpy数组的形式，通过train_data.classes可以获取10类数据的标签，利用for循环和plt.subplot()、plt.imshow()等绘图函数，可以将64张图像进行可视化，得到的可视化图像如图1所示：
在对训练集进行处理后，下面对测试数据集进行处理。导入测试数据集后，将所有的样本处理为一个整体，看作一个batch用于测试，可使用如下程序：
# 对测试集进行处理
test_data = FashionMNIST(
root = '...data/FashionMNIST', #数据的路径
train = False, #不使用训练数据集
download = True, #数据未下载，所以这里要下载
)
# 为数据添加一个通道维度，并且取值范围归一化，即数据范围缩放到0 ~ 1之间
test_data_x = test_data.data.type(torch.FloatTensor) / 255.0
test_data_x = torch.unsqueeze(test_data_x, dim = 1)
test_data_y = test_data.targets # 测试集标签
print("test_data_x.shape:",test_data_x.shape)
print("test_data_y.shape:",test_data_y.shape)
Out[2]：test_data_x.shape: torch.Size([10000, 1, 28, 28])
test_data_y.shape: torch.Size([10000])
上面的程序同样使用FashionMNIST()函数导入测试数据集，并且将其处理为一个整体，从输出结果可发现测试集有10000张28 X 28的图像。
2、卷积神经网络的搭建
在数据准备完毕后，可以搭建一个卷积神经网络，并且使用训练数据对网络进行训练，使用测试集验证所搭建的网络的识别精度。针对搭建的卷积神经网络，可以使用如图2所示的网络结构。
图2搭建的卷积神经网络有2个卷积层，分别包含16个和32个3 X 3卷积核，并且卷积后使用ReLU激活函数进行激活，两个池化层均为平均值池化，而两个全连接层分别有256和128个神经元，最后的分类器则包含了10个神经元。
针对上述网络结构，可以使用下面的程序代码对卷积神经网络进行定义。
#卷积神经网络的搭建
class MyConvNet(nn.Module):
def __init__(self):
super(MyConvNet,self).__init__()
# 定义第一层卷积
self.conv1 = nn.Sequential(
nn.Conv2d(
in_channels = 1, # 输入的feature map
out_channels = 16, # 输出的feature map
kernel_size = 3, # 卷积核大小
stride = 1, # 卷积核步长为1
padding = 1, # 边缘填充1
),
nn.ReLU(), # 激活函数
nn.AvgPool2d(
kernel_size = 2, # 平均值池化，2*2
stride = 2, # 池化步长2
), #池化后：（16*28*28）->（16*14*14）
)
#定义第二个卷积层
self.conv2 = nn.Sequential(
nn.Conv2d(16, 32, 3, 1, 0), #卷积操作（16*14*14）->（32*12*12）
nn.ReLU(),
nn.AvgPool2d(2, 2), #最大值池化（32*12*12）->（32*6*6）
)
self.classifier = nn.Sequential(
nn.Linear(32 * 6 * 6, 256),
nn.ReLU(),
nn.Linear(256, 128),
nn.ReLU(),
nn.Linear(128, 10),
)
def forward(self,x):
# 定义前向传播路径
x = self.conv1(x)
x = self.conv2(x)
x = x.view(x.size(0), -1) # 展平多维的卷积图层
output = self.classifier(x)
return output
#输出网络结构
myconvnet = MyConvNet()
myconvnet
上面程序中的类MyConvNet()通过nn.Sequential()、nn.Conv2d()、nn.ReLU()、nn.AvgPool2d()、nn.Linear()等层，定义了一个拥有两个卷积层和三个全连接层的卷积神经网络分类器，并且在forward()函数中定义了数据在网络中的前向传播过程，然后使用myconvnet = MyConvNet()得到可用于学习的网络myconvnet，其网络结构输出如下所示：
Out[3]：MyConvNet(
(conv1): Sequential(
(0): Conv2d(1, 16, kernel_size=(3, 3), stride=(1, 1), padding=(1, 1))
(1): ReLU()
(2): AvgPool2d(kernel_size=2, stride=2, padding=0)
)
(conv2): Sequential(
(0): Conv2d(16, 32, kernel_size=(3, 3), stride=(1, 1))
(1): ReLU()
(2): AvgPool2d(kernel_size=2, stride=2, padding=0)
)
(classifier): Sequential(
(0): Linear(in_features=1152, out_features=256, bias=True)
(1): ReLU()
(2): Linear(in_features=256, out_features=128, bias=True)
(3): ReLU()
(4): Linear(in_features=128, out_features=10, bias=True)
)
)
3、卷积神经网络训练与预测
为了训练定义好的网络结构myconvnet。还需要定义一个train_model()函数，该函数可以用训练数据集来训练myconvnet。训练数据整体包含60000张图像，938个batch，可以使用80%的batch用于模型的训练，20%的batch用于模型的验证，所以在定义train_model()函数时，应该包含模型的训练和验证两个过程。train_model()函数的程序如下所示：
# 定义网络的训练过程函数
def train_model(model, traindataloader, train_rate, criterion, optimizer, num_epochs = 25):
'''
model：网络模型；traindataloader：训练数据集，会切分为训练集和验证集；
train_rate：训练集batchsize百分比；criterion：损失函数；optimizer：优化方法；num_epochs：训练的轮数
'''
# 计算训练使用的 batch 数量
batch_num = len(traindataloader)
train_batch_num = round(batch_num * train_rate)
# 复制模型参数
best_model_wts = copy.deepcopy(model.state_dict())
best_acc = 0.0
train_loss_all = []
train_acc_all = []
val_loss_all =[]
val_acc_all = []
since = time.time()
# 训练框架
for epoch in range(num_epochs):
print('Epoch {}/{}'.format(epoch, num_epochs - 1))
print('-' * 10)
#每个epoch有两个训练阶段
train_loss = 0.0
train_corrects = 0
train_num = 0
val_loss = 0.0
val_corrects = 0
val_num = 0
for step, (b_x, b_y) in enumerate(traindataloader):
if step < train_batch_num:
model.train() # 设置模型为训练模式
output = model(b_x)
pre_lab = torch.argmax(output, 1)
loss = criterion(output, b_y) # 计算误差损失
optimizer.zero_grad() # 清空过往梯度
loss.backward() # 误差反向传播
optimizer.step() # 根据误差更新参数
train_loss += loss.item() * b_x.size(0)
train_corrects += torch.sum(pre_lab == b_y.data)
train_num += b_x.size(0)
else:
model.eval() # 设置为验证模式
output = model(b_x)
pre_lab = torch.argmax(output, 1)
loss = criterion(output, b_y)
val_loss += loss.item() * b_x.size(0)
val_corrects += torch.sum(pre_lab == b_y.data)
val_num += b_x.size(0)
# ======================小循环结束========================
# 计算一个epoch在训练集和验证集上的损失和精度
train_loss_all.append(train_loss / train_num)
train_acc_all.append(train_corrects.double().item() / train_num)
val_loss_all.append(val_loss / val_num)
val_acc_all.append(val_corrects.double().item() / val_num)
print('{} Train Loss: {:.4f} Train Acc: {:.4f}'.format(epoch, train_loss_all[-1], train_acc_all[-1]))
print('{} Val Loss: {:.4f} Val Acc: {:.4f}'.format(epoch, val_loss_all[-1], val_acc_all[-1]))
# 拷贝模型最高精度下的参数
if val_acc_all[-1] > best_acc:
best_acc = val_acc_all[-1]
best_model_wts = copy.deepcopy(model.state_dict())
time_use = time.time() - since
print('Train and Val complete in {:.0f}m {:.0f}s'.format(time_use // 60, time_use % 60))
# ===========================大循环结束===========================
# 使用最好模型的参数
model.load_state_dict(best_model_wts)
train_process = pd.DataFrame(
data = {"epoch": range(num_epochs),
"train_loss_all": train_loss_all,
"val_loss_all": val_loss_all,
"train_acc_all": train_acc_all,
"val_acc_all": val_acc_all})
return model, train_process
在上面的train_model()函数通过train_batch_num确定用于训练的batch数量，并且在每轮的迭代中，如果step<train_batch_num，则进入训练模式，否则进入验证模式。在模型的训练和验证过程中，分别输出当前的损失函数大小和对应的识别精度，并将它们保存在列表汇总中，最后组成数据表格train_process输出。为了保存模型最高精度下的训练参数，使用copy.deepcopy()函数将模型最优的参数保存在best_model_wts中，最终将所有的训练结果使用 model.load_state_dict(best_model_wts)将最优的参数赋值给最终的模型。
下面使用train_model()函数，对指定的模型和优化器进行训练。
# 对模型进行训练
optimizer = torch.optim.Adam(myconvnet.parameters(), lr = 0.0003) # Adam优化器
criterion = nn.CrossEntropyLoss() # 损失函数
myconvnet, train_process = train_model(myconvnet, train_loader, 0.8, criterion, optimizer, num_epochs = 25)
myconvnet分类器使用了Adam优化器，损失函数为交叉熵损失函数。train_model()将训练集train_loader的80%用于训练，20%用于测试，共训练25轮，训练过程中的输出如下所示：
Out[4]： Epoch 0/24
----------
0 Train Loss: 0.7745 Train Acc: 0.7146
0 Val Loss: 0.5506 Val Acc: 0.7888
Train and Val complete in 0m 12s
Epoch 1/24
......
24 Train Loss: 0.1780 Train Acc: 0.9336
24 Val Loss: 0.2592 Val Acc: 0.9128
Train and Val complete in 4m 56s
在模型训练结束后使用折线图将模型训练过程中的精度和损失函数进行可视化，得到的图像如图3所示：
# 可视化训练过程
plt.figure(figsize = (12, 4))
plt.subplot(1, 2, 1)
plt.plot(train_process.epoch, train_process.train_loss_all, 'ro-', label = 'Train loss')
plt.plot(train_process.epoch, train_process.val_loss_all, 'bs-', label = 'Val loss')
plt.legend()
plt.xlabel('epoch')
plt.ylabel('Loss')
plt.subplot(1, 2, 2)
plt.plot(train_process.epoch, train_process.train_acc_all, 'ro-', label = 'Train acc')
plt.plot(train_process.epoch, train_process.val_acc_all, 'bs-', label = 'Val acc')
plt.xlabel('epoch')
plt.ylabel('Acc')
plt.legend()
plt.show()
从图3可以发现模型在训练过程中，损失函数在训练集上迅速减小，在验证集上先减小然后逐渐收敛到一个很小的区间，说明模型已经稳定。在训练集上的精度一直在增大，而在验证集上的精度收敛到一个小区间内。
为了得到计算模型的泛化能力，使用输出的模型在测试集上进行预测。
# 测试集预测，并可视化预测效果
myconvnet.eval()
output = myconvnet(test_data_x)
pre_lab = torch.argmax(output, 1)
acc = accuracy_score(test_data_y, pre_lab)
print(test_data_y)
print(pre_lab)
print("测试集上的预测精度为:", acc)
Out[5]：tensor([9, 2, 1, ..., 8, 1, 5])
tensor([9, 2, 1, ..., 8, 1, 5])
测试集上的预测精度为: 0.9073
从输出结果可以发现，模型在测试集上的预测精度为90.73%。针对测试样本的预测结果，使用混淆矩阵表示，并将其可视化，观察其在每类数据上的预测情况。
# 计算测试集上的混淆矩阵并可视化
conf_mat = confusion_matrix(test_data_y, pre_lab)
df_cm = pd.DataFrame(conf_mat, index = class_label, columns = class_label)
heatmap = sns.heatmap(df_cm, annot = True, fmt = 'd', cmap = 'YlGnBu')
heatmap.yaxis.set_ticklabels(heatmap.yaxis.get_ticklabels(), rotation = 0, ha = 'right')
heatmap.xaxis.set_ticklabels(heatmap.xaxis.get_ticklabels(), rotation = 45, ha = 'right')
plt.ylabel('True label')
plt.xlabel('Predicted label')
plt.show()
上述代码输出如图4所示的混淆矩阵。
从图4中可以发现，最容易预测发生错误的是T-shirt和Shirt，相互预测的样本量超过了100个。
4、空洞卷积神经网络的搭建
在Pytorch库中使用nn.Conv2d()函数，通过调节参数dilation的取值，进行不同大小卷积核的空洞卷积运算。针对搭建的空洞卷积神经网络，可以使用图5所示的网络结构。
图5 搭建的卷积神经网路含有两个空洞卷积层，两个池化层以及两个全连接层，并且分类器包含10个神经元。该网络结构除了卷积方式的差异外，与图2所示的卷积结构完全相同。
下面搭建图5所表示的空洞卷积，程序如下：
class MyConvdilaNet(nn.Module):
def __init__(self):
super(MyConvdilaNet, self).__init__()
# 定义第一层卷积
self.conv1 = nn.Sequential(
nn.Conv2d(
in_channels = 1, # 输入图像通道数
out_channels = 16, # 输出特征数(卷积核个数)
kernel_size = 3, # 卷积核大小
stride = 1, # 卷积核步长1
padding = 1, # 边缘填充1
dilation = 2,
),
nn.ReLU(), # 激活函数
nn.AvgPool2d(
kernel_size = 2, # 平均值池化，2*2
stride = 2, # 池化步长2
),
)
self.conv2 = nn.Sequential(
nn.Conv2d(16, 32, 3, 1, 0, dilation = 2),
nn.ReLU(),
nn.AvgPool2d(2, 2),
)
self.classifier = nn.Sequential(
nn.Linear(32 * 4 * 4, 256),
nn.ReLU(),
nn.Linear(256, 128),
nn.ReLU(),
nn.Linear(128, 10),
)
def forward(self, x):
# 定义前向传播路径
x = self.conv1(x)
x = self.conv2(x)
x = x.view(x.size(0), -1) # 展平多维的卷积图层
output = self.classifier(x)
return output
#输出网络结构
myconvdilation = MyConvdilaNet()
上面的程序在类MyConvdilaNet()中，同样通过nn.Sequential()，nn.Conv2d()，nn.ReLU()，nn.AvgPool2d()，nn.Linear()等层定义了一个拥有两个空洞卷积层和三个全连接层的卷积神经网络的分类器，其中在空洞卷积中使用参数dilation = 2来实现，最后在forward()函数中定义了数据在网络中的前向传播过程，并使用myconvdilation = MyConvdilaNet()得到可用于学习的网络myconvnet。
5、空洞卷积神经网络训练与预测
下面使用定义好的train_model()函数对网络myconvdilation进行训练，程序代码如下：
# 对模型进行训练
optimizer = torch.optim.Adam(myconvdilanet.parameters(), lr = 0.0003)
criterion = nn.CrossEntropyLoss()
myconvdilanet, train_process = train_model(myconvdilanet, train_loader, 0.8, criterion, optimizer, num_epochs = 25)
Out[6]：Epoch 0/24
----------
0 Train Loss: 0.8922 Train Acc: 0.6718
0 Val Loss: 0.6322 Val Acc: 0.7498
Train and Val complete in 1m 2s
...
24 Train Loss: 0.2531 Train Acc: 0.9062
24 Val Loss: 0.2907 Val Acc: 0.8942
Train and Val complete in 24m 58s
使用折线图将模型训练过程中的精度和损失函数进行可视化，程序代码如下：
# 可视化训练过程
plt.figure(figsize = (12, 4))
plt.subplot(1, 2, 1)
plt.plot(train_process.epoch, train_process.train_loss_all, 'ro-', label = 'Train loss')
plt.plot(train_process.epoch, train_process.val_loss_all, 'bs-', label = 'Val loss')
plt.legend()
plt.xlabel('epoch')
plt.ylabel('Loss')
plt.subplot(1, 2, 2)
plt.plot(train_process.epoch, train_process.train_acc_all, 'ro-', label = 'Train acc')
plt.plot(train_process.epoch, train_process.val_acc_all, 'bs-', label = 'Val acc')
plt.legend()
plt.xlabel('epoch')
plt.ylabel('acc')
plt.show()
从模型训练过程中可以看出，损失函数在训练集上迅速缩小，在验证集上先减小然后逐渐收敛到一个很小的区间，说明模型已经稳定。在训练集上的精度在一直增大，而在验证集上的精度收敛到一个小区间内。
下面使用输出的模型在测试集上进行预测，以计算模型的泛化能力，可使用如下所示的程序：
# 对测试集进行预测，并可视化预测效果
myconvdilanet.eval()
output = myconvdilanet(test_data_x)
pre_lab = torch.argmax(output, 1)
acc = accuracy_score(test_data_y, pre_lab)
print(''测试集上的预测精度为：'', acc)
Out[7]: 测试集上的预测精度为：0.8841
从输出中可以发现模型在测试集上的预测精度为88.41%，识别精度略低于卷积神经网路，针对测试样本的预测结果，同样可以使用混淆矩阵表示，并将其可视化，程序如下：
# 计算混淆矩阵并可视化
conf_mat = confusion_matrix(test_data_y, pre_lab)
df_cm = pd.DataFrame(conf_mat, index = label, columns = label)
heatmap = sns.heatmap(df_cm, annot = True, fmt = 'd', cmap = 'YlGnBu')
heatmap.yaxis.set_ticklabels(heatmap.yaxis.get_ticklabels(), rotation = 0, ha = 'right')
heatmap.xaxis.set_ticklabels(heatmap.xaxis.get_ticklabels(), rotation = 45, ha = 'right')
plt.ylabel('True label')
plt.xlabel('Predicted label')
plt.show()
如图7所示：
