基于PaddlePaddle实现ResNet在Cifar10上的

当网络层数走向更深时，出现了网络退化问题，即增加网络层数之后，训练误差往往不降反升，网络性能快速下降。2015年何恺明推出的ResNet，通过增加一个identity mapping（恒等映射），将原始所需要学的函数H(x)转换成F(x)+x，成功训练152层深的神经网络，在ILSVRC2015比赛中获得了冠军，top-5错误率为3.57%，同时参数量却比VGGNet低很多。

引用自:Deep Residual Learning for Image Recognition

引用地址：Deep Residual Learning for Image Recognition

残差块结构：

残差块结构

代码实现：

import paddle 
import paddle.nn.functional as F # 组网相关的函数，如conv2d, relu...
import numpy as np
from paddle.nn.layer.common import Dropout 
from paddle.vision.transforms import Compose, Resize, Transpose, Normalize, ToTensor
from paddle.vision.datasets import Cifar10

# 构建ResNet网络
# Sequential：顺序容器，子Layer将按构造函数参数的顺序添加到此容器中，传递给构造函数的参数可以Layers或可迭代的name Layer元组
from paddle.nn import Sequential, Conv2D, ReLU, MaxPool2D, Linear, Dropout, Flatten, BatchNorm2D, AvgPool2D

#构建模型
class Residual(paddle.nn.Layer):
    def __init__(self, in_channel, out_channel, use_conv1x1=False, stride=1):
        super().__init__()
        self.conv1 = Conv2D(in_channel, out_channel, kernel_size=3, padding=1, stride=stride)
        self.conv2 = Conv2D(out_channel, out_channel, kernel_size=3, padding=1)
        if use_conv1x1: #使用1x1卷积核
            self.conv3 = Conv2D(in_channel, out_channel, kernel_size=1, stride=stride)
        else:
            self.conv3 = None
        self.batchNorm1 = BatchNorm2D(out_channel)
        self.batchNorm2 = BatchNorm2D(out_channel)

    def forward(self, x):
        y = F.relu(self.batchNorm1(self.conv1(x)))
        y = self.batchNorm2(self.conv2(y))
        if self.conv3:
            x = self.conv3(x)
        out = F.relu(y+x) #核心代码
        return out

残差网络ResNet50代码实现：

def ResNetBlock(in_channel, out_channel, num_layers, is_first=False):
    if is_first:
        assert in_channel == out_channel
    block_list = []
    for i in range(num_layers):
        if i == 0 and not is_first:
            block_list.append(Residual(in_channel, out_channel, use_conv1x1=True, stride=2))
        else:
            block_list.append(Residual(out_channel, out_channel))
    resNetBlock = Sequential(*block_list) #用*号可以把list列表展开为元素
    return resNetBlock

class ResNet50(paddle.nn.Layer):
    def __init__(self, num_classes=10):
        super().__init__()
        self.b1 = Sequential(
                    Conv2D(3, 64, kernel_size=7, stride=2, padding=3),
                    BatchNorm2D(64), 
                    ReLU(),
                    MaxPool2D(kernel_size=3, stride=2, padding=1))
        self.b2 = ResNetBlock(64, 64, 3, is_first=True)
        self.b3 = ResNetBlock(64, 128, 4)
        self.b4 = ResNetBlock(128, 256, 6)
        self.b5 = ResNetBlock(256, 512, 3)
        self.AvgPool = AvgPool2D(2)
        self.flatten = Flatten()
        self.Linear = Linear(512, num_classes)
        
    def forward(self, x):
        x = self.b1(x)
        x = self.b2(x)
        x = self.b3(x)
        x = self.b4(x)
        x = self.b5(x)
        x = self.AvgPool(x)
        x = self.flatten(x)
        x = self.Linear(x)
        return x
        
resnet = ResNet50(num_classes=10)
model = paddle.Model(resnet)
from paddle.static import InputSpec
input = InputSpec([None, 3, 96, 96], 'float32', 'image')
label = InputSpec([None, 1], 'int64', 'label')
model = paddle.Model(resnet, input, label)
model.summary()

训练代码如下所示：

resnet = ResNet50(num_classes=10)
model = paddle.Model(resnet)
from paddle.static import InputSpec
input = InputSpec([None, 3, 96, 96], 'float32', 'image')
label = InputSpec([None, 1], 'int64', 'label')
model = paddle.Model(resnet, input, label)
model.summary()

# Compose: 以列表的方式组合数据集预处理功能
# Resize: 调整图像大小
# Transpose: 调整通道顺序, eg, HWC(img) -> CHW(NN)
# Normalize: 对图像数据归一化
# ToTensor: 将 PIL.Image 或 numpy.ndarray 转换成 paddle.Tensor
# cifar10 手动计算均值和标准差：mean = [125.31, 122.95, 113.86] 和 std = [62.99, 62.08, 66.7] link:https://www.jianshu.com/p/a3f3ffc3cac1

t = Compose([Resize(size=96), 
             Normalize(mean=[125.31, 122.95, 113.86], std=[62.99, 62.08, 66.7], data_format='HWC'), 
             Transpose(order=(2,0,1)), 
             ToTensor(data_format='HWC')])

train_dataset = Cifar10(mode='train', transform=t, backend='cv2') 
test_dataset  = Cifar10(mode='test', transform=t, backend='cv2')
BATCH_SIZE = 256
train_loader = paddle.io.DataLoader(train_dataset, shuffle=True, batch_size=BATCH_SIZE)
test_loader = paddle.io.DataLoader(test_dataset, batch_size=BATCH_SIZE)
# 为模型训练做准备，设置优化器，损失函数和精度计算方式
learning_rate = 0.001
loss_fn = paddle.nn.CrossEntropyLoss()
opt = paddle.optimizer.Adam(learning_rate=learning_rate, parameters=model.parameters())
model.prepare(optimizer=opt, loss=loss_fn, metrics=paddle.metric.Accuracy())

# 启动模型训练，指定训练数据集，设置训练轮次，设置每次数据集计算的批次大小，设置日志格式
model.fit(train_loader, test_loader, batch_size=256, epochs=20, eval_freq= 5, verbose=1)
model.evaluate(test_loader, verbose=1)

训练结果：测试数据集上的精度在：80%左右

Epoch 20/20
step 196/196 [==============================] - loss: 0.0529 - acc: 0.9840 - 318ms/step
Eval begin...
The loss value printed in the log is the current batch, and the metric is the average value of previous step.
step 40/40 [==============================] - loss: 0.1676 - acc: 0.7816 - 198ms/step

在PaddlePaddle中ResNet网络还有一种实现方式，即直接用PaddlePaddle自带的ResNet类，范例代码如下所示：

from paddle.vision.models import ResNet   
from paddle.vision.models.resnet import BottleneckBlock     
# resnet = ResNet50(num_classes=10)
resnet = ResNet(BottleneckBlock, 50, num_classes=10)
model = paddle.Model(resnet)
from paddle.static import InputSpec
input = InputSpec([None, 3, 96, 96], 'float32', 'image')
label = InputSpec([None, 1], 'int64', 'label')
model = paddle.Model(resnet, input, label)
model.summary()

运行结果：

Epoch 20/20
step 196/196 [==============================] - loss: 0.0661 - acc: 0.9743 - 467ms/step
Eval begin...
The loss value printed in the log is the current batch, and the metric is the average value of previous step.
step 40/40 [==============================] - loss: 0.7514 - acc: 0.7846 - 235ms/step