PyTorch 实现 ResNet50 图像分类
本实验主要介绍了如何在昇腾上，使用 pytorch 对经典的 resnet50 小模型在公开的 CIFAR10 数据集进行分类训练的实战讲解。内容包括 resnet50 的网络架构 ，残差模块分析 ，训练代码分析等等
本实验的目录结构安排如下所示：
Resnet 系列网络结构
resnet50 网络搭建过程及代码详解
端到端训练 cifar 数据集实战
Resnet 系列网络结构
传统的卷积网络或者全连接网络在信息传递的时候或多或少会存在信息丢失，损耗等问题，同时还有导致梯度消失或者梯度爆炸，阻碍网络收敛，导致很深的网络无法训练。
此外，有部分神经网络在堆叠网络加深的过程中，出现了训练集准确率下降的现象.
基于上述两种问题，resnet 网络在 2015 年时被提出，其可以极快的加速神经网络的训练，模型的准确率也有比较大的提升。
ResNet 的主要思想是在网络中增加了直连通道，即 Highway Network 的思想。
此前的网络结构是性能输入做一个非线性变换，而 Highway Network 则允许保留之前网络层的一定比例的输出。
ResNet 的思想和 Highway Network 的思想也非常类似，允许原始输入信息直接传到后面的层中，如下图所示，resnet 网络主要用到了残差模块，主要分为两种结构，以 50 层作为一个区分边界，结构在原论文中定义如下：
resnet50 网络搭建过程及代码详解
从上述 resnet 系列结构图中可以看出，5 种不同层数的 resnet 网络的主要区别在于其基础卷积模块用到的卷积核大小不一致，后面将会重点介绍该这两个基础模块（BasicBlock 与 Bottleneck）的搭建过程。
由于网络的搭建过程中会使用到 torch 相关模块，包括 nn 与 optim，首先对这两个模块进行介绍。
nn 模块是专门为神经网络设计的模块化接口，构建于 autograd 之上，可以用来定义和运行神经网络。
实验会用到 nn 中的两大模块（functional 与 Module），functional 模块是具体网络层的实现，相比与 Module 更轻量。
Module 是一个类，是所有神经网络单元的基类，包含网络各层的定义及 forward 方法，是对 functional 中函数的功能扩展(添加了参数和信息管理等功能)，但是它的计算功能还是通过调用 functional 中的函数来实现的。
optim 实现了各种优化算法的库（例如：SGD 与 Adam），在使用 optimizer 时候需要构建一个 optimizer 对象，这个对象能够保持当前参数状态并基于计算得到的梯度进行参数更新。
## 导入torch相关模块import torchimport torch.nn as nnimport torch.optim as optimimport torch.nn.functional as F
复制代码
BasicBlock 模块介绍：该基础模块主要用于 50 层数以下的 resnet 网络之中，具体网络结构如下图所示：
该模块包含两个卷积层，其中 kernel_size=3 表示该层卷核是 3*3，每层卷积后皆有 bn 操作防止过拟合，卷积后 relu 操作。
该模块用一个 BasicBlock 类实现，其类中定义了两个函数'init'与'forward'，'init'用于初始化操作，'forward 用于网络的前向传播
注意：
模型中有 BN 层(Batch Normalization）和 Dropout，需要在训练时添加'model.train()'，在测试时添加'model.eval()'。对于 BN 层而言 ，在训练时候添加'model.train()'是保证使用的是每一批数据的均值和方差，而 'model.eval()' 则是保证用全部训练数据的均值和方差。针对 Dropout，'model.train()'则是随机取一部分网络连接来训练更新参数，而'model.eval()'是利用到了所有网络连接。这是因为 train 完样本后，模型会被要用来测试样本。在 model(test) 之前，需要加上'model.eval()'，model 中含有 BN 层和 Dropout，一旦 test 的 batch_size 过小，很容易就会被 BN 导致生成图片颜色失真极大。而'eval()'时，pytorch 会自动把 BN 和 DropOut 固定住，不会取平均，而是用训练好的值。如果不加'eval()'，在非训练的过程在一些网络层的值会发生变动，不会固定，神经网络每一次生成的结果也是不固定的，生成质量可能好也可能不好。
class BasicBlock(nn.Module): expansion = 1 def __init__(self, in_planes, planes, stride=1): super(BasicBlock, self).__init__() self.conv1 = nn.Conv2d( in_planes, planes, kernel_size=3, stride=stride, padding=1, bias=False) self.bn1 = nn.BatchNorm2d(planes) self.conv2 = nn.Conv2d(planes, planes, kernel_size=3, stride=1, padding=1, bias=False) self.bn2 = nn.BatchNorm2d(planes)
self.shortcut = nn.Sequential() if stride != 1 or in_planes != self.expansion*planes: self.shortcut = nn.Sequential( nn.Conv2d(in_planes, self.expansion*planes, kernel_size=1, stride=stride, bias=False), nn.BatchNorm2d(self.expansion*planes) )
def forward(self, x): out = F.relu(self.bn1(self.conv1(x))) out = self.bn2(self.conv2(out)) out += self.shortcut(x) out = F.relu(out) return out
复制代码
Bottleneck 模块介绍： 该基础模块主要用于 50 层数以上的 resnet 网络之中，其架构如下图所示：
该模块包含三卷积层，其中 kernel_size=3 表示该层卷核是 33, kernel_size=1 表示该层卷核是 11, 每层卷积后皆有 bn 操作防止过拟合，卷积后 relu 操作。
该模块用一个 Bottleneck 类实现，其类中定义了两个函数'init '与'forward'，'init '用于初始化操作，'forward 用于网络的前向传播'，在 forward 后跟有 shortcut 操作，这也是 resnet 网络的经典之处。
class Bottleneck(nn.Module): expansion = 4 def __init__(self, in_planes, planes, stride=1): super(Bottleneck, self).__init__() self.conv1 = nn.Conv2d(in_planes, planes, kernel_size=1, bias=False) self.bn1 = nn.BatchNorm2d(planes) self.conv2 = nn.Conv2d(planes, planes, kernel_size=3, stride=stride, padding=1, bias=False) self.bn2 = nn.BatchNorm2d(planes) self.conv3 = nn.Conv2d(planes, self.expansion * planes, kernel_size=1, bias=False) self.bn3 = nn.BatchNorm2d(self.expansion*planes)
self.shortcut = nn.Sequential() if stride != 1 or in_planes != self.expansion*planes: self.shortcut = nn.Sequential( nn.Conv2d(in_planes, self.expansion*planes, kernel_size=1, stride=stride, bias=False), nn.BatchNorm2d(self.expansion*planes) )
def forward(self, x): out = F.relu(self.bn1(self.conv1(x))) out = F.relu(self.bn2(self.conv2(out))) out = self.bn3(self.conv3(out)) out += self.shortcut(x) out = F.relu(out) return out
复制代码
resnet 系列网络定义及搭建过程： 主要通过一个 ResNet 类实现，通过指定不同的入参'block'与'num_blocks'值可以用来生成不同层数的 resnet 网络。
该类有三个函数，分别是'init'、'maker_layer'与'forward'。
其中'init'用来初始化网络层中的变量，'make_layer'用来构建不同基础模块（BasicBlock 与 Bottleneck），'forward'函数用来搭建前向的网络层。
class ResNet(nn.Module): def __init__(self, block, num_blocks, num_classes=10): super(ResNet, self).__init__() self.in_planes = 64
self.conv1 = nn.Conv2d(3, 64, kernel_size=3, stride=1, padding=1, bias=False) self.bn1 = nn.BatchNorm2d(64) self.layer1 = self._make_layer(block, 64, num_blocks[0], stride=1) self.layer2 = self._make_layer(block, 128, num_blocks[1], stride=2) self.layer3 = self._make_layer(block, 256, num_blocks[2], stride=2) self.layer4 = self._make_layer(block, 512, num_blocks[3], stride=2) self.linear = nn.Linear(512*block.expansion, num_classes)
def _make_layer(self, block, planes, num_blocks, stride): strides = [stride] + [1]*(num_blocks-1) layers = [] for stride in strides: layers.append(block(self.in_planes, planes, stride)) self.in_planes = planes * block.expansion return nn.Sequential(*layers)
def forward(self, x): out = F.relu(self.bn1(self.conv1(x))) out = self.layer1(out) out = self.layer2(out) out = self.layer3(out) out = self.layer4(out) out = F.avg_pool2d(out, 4) out = out.view(out.size(0), -1) out = self.linear(out) return out
复制代码
端到端训练 cifar 数据集实战
导入昇腾 npu 相关库 transfer_to_npu、该模块可以使能模型自动迁移至昇腾上。
import torch_npufrom torch_npu.contrib import transfer_to_npu
复制代码
torchvision 模块中集成了一些当今比较流行的数据集、模型架构和用于计算机视觉的常见图像转换功能，torchvision 模块中含有本次实验所需要的 CIFAR 数据集与 ResNet 网络系列。
import torchvisionimport torchvision.transforms as transforms
复制代码
定义本次实验需要用到的网络 resnet50，通过构造'ResNet'类，传入'Bottleneck'块，如果要定义不同层数的其他 resnet 网络也是类似的，指定 ResNet 的两个入参即可。
例如需要定义 ResNet18，则调用 ResNet(BasicBlock , [2, 2, 2, 2])即可，其中 2, 2, 2, 2 四个数与 Resnet 系列网络架构中后四层卷积基础模块的数值相对应。
def ResNet50(): return ResNet(Bottleneck, [3, 4, 6, 3])
复制代码
数据集预处理功能定义： 对图像数据集进行不同程度的变化，包括裁剪、翻转等方式增加数据的多样性，防止过拟合现象的出现，以增强模型的泛化能力。
调用了 torchvision 中的 transform 库中的 compose 方法，使用裁剪（RandomCrop）、翻转（RandomHorizontalFlip）等组合成 tensor 形式后并对 tensor 进行正则化（Normalize）。
transform_train = transforms.Compose([ transforms.RandomCrop(32, padding=4), transforms.RandomHorizontalFlip(), transforms.ToTensor(), transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010)),])transform_test = transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.4914, 0.4822, 0.4465), (0.2023, 0.1994, 0.2010)),])
复制代码
cifar 数据集加载： torchvision 中集成了一些通用的开源数据集，其中也包含 cifar，此处通过 torchvision 函数加载 cifar 数据集到工作目录上的指定路径，如果已经下载好了，会直接校验通过，不会二次进行下载。
trainset = torchvision.datasets.CIFAR10( root='./dataset/cifar-10-batches-py', train=True, download=True, transform=transform_train)trainloader = torch.utils.data.DataLoader( trainset, batch_size=128, shuffle=True)testset = torchvision.datasets.CIFAR10( root='./dataset/cifar-10-batches-py', train=False, download=True, transform=transform_test)testloader = torch.utils.data.DataLoader( testset, batch_size=100, shuffle=False)classes = ('plane', 'car', 'bird', 'cat', 'deer', 'dog', 'frog', 'horse', 'ship', 'truck')
复制代码
训练模块: 根据传入的迭代次数开始训练网络模型，这里需要在 model 开始前加入 net.train()，使用随机梯度下降算法是将梯度值初始化为 0（zero_grad()），计算梯度、通过梯度下降算法更新模型参数的值以及统计每次训练后的 loss 值（每隔 100 次打印一次）。
def train(epoch): net.train() train_loss = 0.0 epoch_loss = 0.0 for batch_idx, (inputs, targets) in enumerate(tqdm(trainloader, 0)): inputs, targets = inputs.to(device), targets.to(device) optimizer.zero_grad() outputs = net(inputs) loss = criterion(outputs, targets) loss.backward() optimizer.step() lr_scheduler.step()
train_loss += loss.item() epoch_loss += loss.item()
if batch_idx % 100 == 99: # 每100次迭代打印一次损失 #print(f'[Epoch {epoch + 1}, Iteration {batch_idx + 1}] loss: {train_loss / 100:.3f}') train_loss = 0.0 return epoch_loss / len(trainloader)
复制代码
测试模块: 每训练一轮将会对最新得到的训练模型效果进行测试，使用的是数据集准备时期划分得到的测试集。
def test(): net.eval() test_loss = 0 correct = 0 total = 0 with torch.no_grad(): for batch_idx, (inputs, targets) in enumerate(tqdm(testloader)): inputs, targets = inputs.to(device), targets.to(device) outputs = net(inputs) loss = criterion(outputs, targets)
test_loss += loss.item() _, predicted = outputs.max(1) total += targets.size(0) correct += predicted.eq(targets).sum().item() return 100 * correct / total
复制代码
主功能调用模块： 该模块用于开启模型在指定数据集（cifar）上训练，其中定义了硬件设备为昇腾 npu（device = 'npu'），定义了损失函数为交叉熵损失'CrossEntropyLoss()'，梯度下降优化算法为 SGD 并同时指定了学习率等参数。
训练与测试的次数为 60 次，这里用户可以根据需要自行选择设置更高或更低，每个 epoch 的测试准确率都会被打印出来，如果不需要将代码注释掉即可。
#定义模型训练在哪种类型的设备上跑device = 'npu'net = ResNet50()#将网络模型加载到指定设备上，这里device是昇腾的npunet = net.to(device)criterion = nn.CrossEntropyLoss()optimizer = optim.SGD(net.parameters(), lr=1.0, weight_decay=5e-4)lr_scheduler = torch.optim.lr_scheduler.OneCycleLR(optimizer,0.1,steps_per_epoch=len(trainloader), epochs=150,div_factor=25,final_div_factor=10000,pct_start=0.3)#开启模型训练与测试过程for epoch in range(60): epoch_loss = train(epoch) test_accuray = test() print(f'\nTest accuracy for ResNet50 at epoch {epoch + 1}: {test_accuray:.2f}%') print(f'Epoch loss for ResNet50 at epoch {epoch + 1}: {epoch_loss:.3f}')
复制代码
Reference
[1] He K, Zhang X, Ren S, et al. Deep residual learning for image recognition[C]//Proceedings of the IEEE conference on computer vision and pattern recognition. 2016: 770-778.
评论