PyTorch詳解經典網絡ResNet實現流程

更新時間：2022年05月06日 16:47:03 作者：峽谷的小魚

ResNet全稱residual neural network,主要是解決過深的網絡帶來的梯度彌散,梯度爆炸,網絡退化(即網絡層數越深時,在數據集上表現的性能卻越差)的問題

簡述

GoogleNet 和 VGG 等網絡證明了，更深度的網絡可以抽象出表達能力更強的特征，進而獲得更強的分類能力。在深度網絡中，隨之網絡深度的增加，每層輸出的特征圖分辨率主要是高和寬越來越小，而深度逐漸增加。

深度的增加理論上能夠提升網絡的表達能力，但是對于優(yōu)化來說就會產生梯度消失的問題。在深度網絡中，反向傳播時，梯度從輸出端向數據端逐層傳播，傳播過程中，梯度的累乘使得近數據段接近0值，使得網絡的訓練失效。

為了解決梯度消失問題，可以在網絡中加入BatchNorm，激活函數換成ReLU，一定程度緩解了梯度消失問題。

深度增加的另一個問題就是網絡的退化（Degradation of deep network）問題。即，在現有網絡的基礎上，增加網絡的深度，理論上，只有訓練到最佳情況，新網絡的性能應該不會低于淺層的網絡。因為，只要將新增加的層學習成恒等映射（identity mapping）就可以。換句話說，淺網絡的解空間是深的網絡的解空間的子集。但是由于Degradation問題，更深的網絡并不一定好于淺層網絡。

Residual模塊的想法就是認為的讓網絡實現這種恒等映射。如圖，殘差結構在兩層卷積的基礎上，并行添加了一個分支，將輸入直接加到最后的ReLU激活函數之前，如果兩層卷積改變大量輸入的分辨率和通道數，為了能夠相加，可以在添加的分支上使用1x1卷積來匹配尺寸。

殘差結構

ResNet網絡有兩種殘差塊，一種是兩個3x3卷積，一種是1x1，3x3，1x1三個卷積網絡串聯成殘差模塊。

PyTorch 實現：

class Residual_1(nn.Module):
    r""" 
    18-layer, 34-layer 殘差塊
    1. 使用了類似VGG的3×3卷積層設計；
    2. 首先使用兩個相同輸出通道數的3×3卷積層，后接一個批量規(guī)范化和ReLU激活函數；
    3. 加入跨過卷積層的通路，加到最后的ReLU激活函數前；
    4. 如果要匹配卷積后的輸出的尺寸和通道數，可以在加入的跨通路上使用1×1卷積；
    """
    def __init__(self, input_channels, num_channels, use_1x1conv=False, strides=1):
        r"""
        parameters:
            input_channels: 輸入的通道上數
            num_channels: 輸出的通道數
            use_1x1conv: 是否需要使用1x1卷積控制尺寸
            stride: 第一個卷積的步長
        """
        super().__init__()
        # 3×3卷積，strides控制分辨率是否縮小
        self.conv1 = nn.Conv2d(input_channels, 
                               num_channels,
                               kernel_size=3, 
                               padding=1, 
                               stride=strides)
        # 3×3卷積，不改變分辨率
        self.conv2 = nn.Conv2d(num_channels,
                               num_channels, 
                               kernel_size=3, 
                               padding=1)
        # 使用 1x1 卷積變換輸入的分辨率和通道
        if use_1x1conv:
            self.conv3 = nn.Conv2d(input_channels, 
                                   num_channels, 
                                   kernel_size=1, 
                                   stride=strides)
        else:
            self.conv3 = None
        # 批量規(guī)范化層
        self.bn1 = nn.BatchNorm2d(num_channels)
        self.bn2 = nn.BatchNorm2d(num_channels)
    def forward(self, X):
        Y = F.relu(self.bn1(self.conv1(X)))
        Y = self.bn2(self.conv2(Y))
        if self.conv3:
            X = self.conv3(X)
        # print(X.shape)
        Y += X
        return F.relu(Y)

class Residual_2(nn.Module):
    r""" 
    50-layer, 101-layer, 152-layer 殘差塊
    1. 首先使用1x1卷積，ReLU激活函數；
    2. 然后用3×3卷積層，在接一個批量規(guī)范化，ReLU激活函數；
    3. 再接1x1卷積層；
    4. 加入跨過卷積層的通路，加到最后的ReLU激活函數前；
    5. 如果要匹配卷積后的輸出的尺寸和通道數，可以在加入的跨通路上使用1×1卷積；
    """
    def __init__(self, input_channels, num_channels, use_1x1conv=False, strides=1):
        r"""
        parameters:
            input_channels: 輸入的通道上數
            num_channels: 輸出的通道數
            use_1x1conv: 是否需要使用1x1卷積控制尺寸
            stride: 第一個卷積的步長
        """
        super().__init__()
        # 1×1卷積，strides控制分辨率是否縮小
        self.conv1 = nn.Conv2d(input_channels, 
                               num_channels,
                               kernel_size=1, 
                               padding=1, 
                               stride=strides)
        # 3×3卷積，不改變分辨率
        self.conv2 = nn.Conv2d(num_channels,
                               num_channels, 
                               kernel_size=3, 
                               padding=1)
        # 1×1卷積，strides控制分辨率是否縮小
        self.conv3 = nn.Conv2d(input_channels, 
                               num_channels,
                               kernel_size=1, 
                               padding=1)
        # 使用 1x1 卷積變換輸入的分辨率和通道
        if use_1x1conv:
            self.conv3 = nn.Conv2d(input_channels, 
                                   num_channels, 
                                   kernel_size=1, 
                                   stride=strides)
        else:
            self.conv3 = None
        # 批量規(guī)范化層
        self.bn1 = nn.BatchNorm2d(num_channels)
        self.bn2 = nn.BatchNorm2d(num_channels)
    def forward(self, X):
        Y = F.relu(self.bn1(self.conv1(X)))
        Y = F.relu(self.bn2(self.conv2(Y)))
        Y = self.conv3(Y)
        if self.conv3:
            X = self.conv3(X)
        # print(X.shape)
        Y += X
        return F.relu(Y)

ResNet有不同的網絡層數，比較常用的是50-layer，101-layer，152-layer。他們都是由上述的殘差模塊堆疊在一起實現的。

以18-layer為例，層數是指：首先，conv_1 的一層7x7卷積，然后conv_2~conv_5四個模塊，每個模塊兩個殘差塊，每個殘差塊有兩層的3x3卷積組成，共4×2×2=16層，最后是一層分類層(fc)，加總一起共1+16+1=18層。

18-layer 實現

首先定義由殘差結構組成的模塊：

# ResNet模塊
def resnet_block(input_channels, num_channels, num_residuals, first_block=False):
    r"""殘差塊組成的模塊"""
    blk = []
    for i in range(num_residuals):
        if i == 0 and not first_block:
            blk.append(Residual_1(input_channels, 
                                num_channels, 
                                use_1x1conv=True, 
                                strides=2))
        else:
            blk.append(Residual_1(num_channels, num_channels))
    return blk

定義18-layer的最開始的層：

# ResNet的前兩層：
#    1. 輸出通道數64， 步幅為2的7x7卷積層
#    2. 步幅為2的3x3最大匯聚層
conv_1 = nn.Sequential(nn.Conv2d(1, 64, kernel_size=7, stride=2, padding=3),
                   nn.BatchNorm2d(64), 
                   nn.ReLU(), 
                   nn.MaxPool2d(kernel_size=3, stride=2, padding=1))

定義殘差組模塊：

# ResNet模塊
conv_2 = nn.Sequential(*resnet_block(64, 64, 2, first_block=True))
conv_3 = nn.Sequential(*resnet_block(64, 128, 2))
conv_4 = nn.Sequential(*resnet_block(128, 256, 2))
conv_5 = nn.Sequential(*resnet_block(256, 512, 2))

ResNet 18-layer模型：

net = nn.Sequential(conv_1, conv_2, conv_3, conv_4, conv_5, 
                    nn.AdaptiveAvgPool2d((1, 1)), 
                    nn.Flatten(), 
                    nn.Linear(512, 10))
# 觀察模型各層的輸出尺寸
X = torch.rand(size=(1, 1, 224, 224))
for layer in net:
    X = layer(X)
    print(layer.__class__.__name__,'output shape:\t', X.shape)

輸出：

Sequential output shape:   torch.Size([1, 64, 56, 56])
Sequential output shape:   torch.Size([1, 64, 56, 56])
Sequential output shape:   torch.Size([1, 128, 28, 28])
Sequential output shape:   torch.Size([1, 256, 14, 14])
Sequential output shape:   torch.Size([1, 512, 7, 7])
AdaptiveAvgPool2d output shape:   torch.Size([1, 512, 1, 1])
Flatten output shape:   torch.Size([1, 512])
Linear output shape:   torch.Size([1, 10])

在數據集訓練

def load_datasets_Cifar10(batch_size, resize=None):
    trans = [transforms.ToTensor()]
    if resize:
        transform = trans.insert(0, transforms.Resize(resize))
    trans = transforms.Compose(trans)
    train_data = torchvision.datasets.CIFAR10(root="../data", train=True, transform=trans, download=True)
    test_data = torchvision.datasets.CIFAR10(root="../data", train=False, transform=trans, download=True)
    print("Cifar10 下載完成...")
    return (torch.utils.data.DataLoader(train_data, batch_size, shuffle=True),
            torch.utils.data.DataLoader(test_data, batch_size, shuffle=False))
def load_datasets_FashionMNIST(batch_size, resize=None):
    trans = [transforms.ToTensor()]
    if resize:
        transform = trans.insert(0, transforms.Resize(resize))
    trans = transforms.Compose(trans)
    train_data = torchvision.datasets.FashionMNIST(root="../data", train=True, transform=trans, download=True)
    test_data = torchvision.datasets.FashionMNIST(root="../data", train=False, transform=trans, download=True)
    print("FashionMNIST 下載完成...")
    return (torch.utils.data.DataLoader(train_data, batch_size, shuffle=True),
            torch.utils.data.DataLoader(test_data, batch_size, shuffle=False))
def load_datasets(dataset, batch_size, resize):
    if dataset == "Cifar10":
        return load_datasets_Cifar10(batch_size, resize=resize)
    else:
        return load_datasets_FashionMNIST(batch_size, resize=resize)
train_iter, test_iter = load_datasets("", 128, 224) # Cifar10