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 torch
import torch.nn as nn
import torch.optim as optim
import 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_npu
from torch_npu.contrib import transfer_to_npu

torchvision模块中集成了一些当今比较流行的数据集、模型架构和用于计算机视觉的常见图像转换功能，torchvision模块中含有本次实验所需要的CIFAR数据集与ResNet网络系列。

import torchvision
import 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是昇腾的npu
net = 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.