Набор данных Hengyuan Cloud_CIFAR-10 на практике: построение нейронной сети ResNet18

глубокое обучение
Набор данных Hengyuan Cloud_CIFAR-10 на практике: построение нейронной сети ResNet18

Источник статьи |Облачное сообщество Hengyuan

Исходный адрес |Боевой набор данных

Оригинальный автор | Матор


Честно говоря, я обожаю Матора, крупного парня в сообществе платформ! Нет, сегодня я здесь, чтобы поделиться с вами заметками к дипломной работе старшего брата. Поторопитесь и посмотрите, есть ли в следующем контенте нужные вам знания!

Текст начинается:

Если вы не знаете ResNet, вы можете сначала прочитать мой блогЧтение бумаги ResNet

Сначала реализуйте остаточный блок

import torch
from torch import nn
from torch.nn import functional as F

class ResBlk(nn.Module):
    def __init__(self, ch_in, ch_out, stride=1):
        super(ResBlk, self).__init__()
        self.conv1 = nn.Conv2d(ch_in, ch_out, kernel_size=3, stride=stride, padding=1)
        self.bn1 = nn.BatchNorm2d(ch_out)
        
        self.conv2 = nn.Conv2d(ch_out, ch_out, kernel_size=3, stride=1, padding=1)
        self.bn2 = nn.BatchNorm2d(ch_out)
        
        if ch_out == ch_in:
            self.extra = nn.Sequential()
        else:
            self.extra = nn.Sequential(
                
                # 1×1的卷积作用是修改输入x的channel
                # [b, ch_in, h, w] => [b, ch_out, h, w]
                nn.Conv2d(ch_in, ch_out, kernel_size=1, stride=stride),
                nn.BatchNorm2d(ch_out),
            )
        
    def forward(self, x):
        out = F.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))

        # short cut
        out = self.extra(x) + out
        out = F.relu(out)
        
        return out

В Block делается регуляризация, чтобы сделать процесс обучения более быстрым и стабильным. При этом следует учитывать, что если ch_in и ch_out двух элементов не совпадают, то при добавлении будет выдано сообщение об ошибке

пройти тест

blk = ResBlk(64, 128, stride=2)
tmp = torch.randn(2, 64, 32, 32)
out = blk(tmp)
print(out.shape)

Размер выходной формыtorch.Size([2, 128, 16, 16])

Вот объяснение, почему некоторые слои должны специально задавать шаг. Независимо от других слоев, для блока Residual канал увеличивается с 64 до 128. Если все шаги равны 1 и отступ также равен 1, то w и h картинки не изменятся, но канал увеличится. приведет к увеличению параметров всей сети. И это только один блок, не говоря уже о следующих FC и других блоках, так что шаг не может быть установлен равным 1, не позволяйте параметрам сети увеличиваться.

Затем мы собираем полный ResNet-18

class ResNet18(nn.Module):
    def __init__(self):
        super(ResNet18, self).__init__()
        
        self.conv1 = nn.Sequential(
            nn.Conv2d(3, 64, kernel_size=3, stride=3, padding=0),
            nn.BatchNorm2d(64),
        )
        # followed 4 blocks
        
        # [b, 64, h, w] => [b, 128, h, w]
        self.blk1 = ResBlk(64, 128, stride=2)
        # [b, 128, h, w] => [b, 256, h, w]
        self.blk2 = ResBlk(128, 256, stride=2)
        # [b, 256, h, w] => [b, 512, h, w]
        self.blk3 = ResBlk(256, 512, stride=2)
        # [b, 512, h, w] => [b, 512, h, w]
        self.blk4 = ResBlk(512, 512, stride=2)
        
        self.outlayer = nn.Linear(512*1*1, 10)
    
    def forward(self, x):
        x = F.relu(self.conv1(x))
        
        # 经过四个blk以后 [b, 64, h, w] => [b, 512, h, w]
        x = self.blk1(x)
        x = self.blk2(x)
        x = self.blk3(x)
        x = self.blk4(x)
        
        x = self.outlayer(x)
        
        return x

пройти тест

x = torch.randn(2, 3, 32, 32)
model = ResNet18()
out = model(x)
print("ResNet:", out.shape)

Результат - ошибка, сообщение об ошибке выглядит следующим образом

size mismatch, m1: [2048 x 2], m2: [512 x 10] at /pytorch/aten/src/TH/generic/THTensorMath.cpp:961

Проблема в том, что мы, наконец, определяем входной размер линейного слоя, который не соответствует выходному размеру предыдущего блока.После запуска последнего блока ResNet18 напечатайте форму текущего x, и результатtorch.Size([2, 512, 2, 2])

Есть много решений.Вы можете изменить ввод линейного слоя, чтобы он соответствовал, или вы можете выполнить некоторые операции после последнего слоя блока, чтобы он соответствовал 512.

Модифицированный код дается первым, а объяснение дается

class ResNet18(nn.Module):
    def __init__(self):
        super(ResNet18, self).__init__()
        
        self.conv1 = nn.Sequential(
            nn.Conv2d(3, 64, kernel_size=3, stride=3, padding=0),
            nn.BatchNorm2d(64),
        )
        # followed 4 blocks
        
        # [b, 64, h, w] => [b, 128, h, w]
        self.blk1 = ResBlk(64, 128, stride=2)
        # [b, 128, h, w] => [b, 256, h, w]
        self.blk2 = ResBlk(128, 256, stride=2)
        # [b, 256, h, w] => [b, 512, h, w]
        self.blk3 = ResBlk(256, 512, stride=2)
        # [b, 512, h, w] => [b, 512, h, w]
        self.blk4 = ResBlk(512, 512, stride=2)
        
        self.outlayer = nn.Linear(512*1*1, 10)
    
    def forward(self, x):
        x = F.relu(self.conv1(x))
        
        # 经过四个blk以后 [b, 64, h, w] => [b, 512, h, w]
        x = self.blk1(x)
        x = self.blk2(x)
        x = self.blk3(x)
        x = self.blk4(x)
        
        # print("after conv:", x.shape) # [b, 512, 2, 2]
        
        # [b, 512, h, w] => [b, 512, 1, 1]
        x = F.adaptive_avg_pool2d(x, [1, 1])
        
        x = x.view(x.size(0), -1) # [b, 512, 1, 1] => [b, 512*1*1]
        x = self.outlayer(x)
        
        return x

Здесь я использую второй способ.После окончания последнего блока подключается слой адаптивного пула.Функция этого пула-выводить тензор с шириной и высотой 1 независимо от ширины и высоты входных данных., остальные размеры остаются неизменными. Затем выполните операцию изменения формы, чтобы[batchsize, 512, 1, 1]reshapeстать[batchsize, 512*1*1]Размер тензора, чтобы он был выровнен со следующим линейным слоем, входной размер линейного слоя 512, а выходной 10. Следовательно, окончательная выходная форма всей сети будет[batchsize, 10]

Наконец, мы скопировали код обучения LeNet5 ранее и поместили код внутрь.model=LeNet5()изменить наmodel=ResNet18()Вот и все. Полный код выглядит следующим образом

import torch
from torch import nn, optim
import torch.nn.functional as F
from torch.utils.data import DataLoader
from torchvision import datasets, transforms


batch_size=32
cifar_train = datasets.CIFAR10(root='cifar', train=True, transform=transforms.Compose([
    transforms.Resize([32, 32]),
    transforms.ToTensor(),
]), download=True)

cifar_train = DataLoader(cifar_train, batch_size=batch_size, shuffle=True)

cifar_test = datasets.CIFAR10(root='cifar', train=False, transform=transforms.Compose([
    transforms.Resize([32, 32]),
    transforms.ToTensor(),
]), download=True)
    
cifar_test = DataLoader(cifar_test, batch_size=batch_size, shuffle=True)      

class ResBlk(nn.Module):
    def __init__(self, ch_in, ch_out, stride=1):
        super(ResBlk, self).__init__()
        self.conv1 = nn.Conv2d(ch_in, ch_out, kernel_size=3, stride=stride, padding=1)
        self.bn1 = nn.BatchNorm2d(ch_out)
        
        self.conv2 = nn.Conv2d(ch_out, ch_out, kernel_size=3, stride=1, padding=1)
        self.bn2 = nn.BatchNorm2d(ch_out)
        
        if ch_out == ch_in:
            self.extra = nn.Sequential()
        else:
            self.extra = nn.Sequential(
                
                # 1×1的卷积作用是修改输入x的channel
                # [b, ch_in, h, w] => [b, ch_out, h, w]
                nn.Conv2d(ch_in, ch_out, kernel_size=1, stride=stride),
                nn.BatchNorm2d(ch_out),
            )
        
    def forward(self, x):
        out = F.relu(self.bn1(self.conv1(x)))
        out = self.bn2(self.conv2(out))

        # short cut
        out = self.extra(x) + out
        out = F.relu(out)
        
        return out
        
class ResNet18(nn.Module):
    def __init__(self):
        super(ResNet18, self).__init__()
        
        self.conv1 = nn.Sequential(
            nn.Conv2d(3, 64, kernel_size=3, stride=3, padding=0),
            nn.BatchNorm2d(64),
        )
        # followed 4 blocks
        
        # [b, 64, h, w] => [b, 128, h, w]
        self.blk1 = ResBlk(64, 128, stride=2)
        # [b, 128, h, w] => [b, 256, h, w]
        self.blk2 = ResBlk(128, 256, stride=2)
        # [b, 256, h, w] => [b, 512, h, w]
        self.blk3 = ResBlk(256, 512, stride=2)
        # [b, 512, h, w] => [b, 512, h, w]
        self.blk4 = ResBlk(512, 512, stride=2)
        
        self.outlayer = nn.Linear(512*1*1, 10)
    
    def forward(self, x):
        x = F.relu(self.conv1(x))
        
        # 经过四个blk以后 [b, 64, h, w] => [b, 512, h, w]
        x = self.blk1(x)
        x = self.blk2(x)
        x = self.blk3(x)
        x = self.blk4(x)
        
        # print("after conv:", x.shape) # [b, 512, 2, 2]
        
        # [b, 512, h, w] => [b, 512, 1, 1]
        x = F.adaptive_avg_pool2d(x, [1, 1])
        
        x = x.view(x.size(0), -1) # [b, 512, 1, 1] => [b, 512*1*1]
        x = self.outlayer(x)
        
        return x

def main():

    ##########  train  ##########
    #device = torch.device('cuda')
    #model = ResNet18().to(device)
    criteon = nn.CrossEntropyLoss()
    model = ResNet18()
    optimizer = optim.Adam(model.parameters(), 1e-3)
    for epoch in range(1000):
        model.train()
        for batchidx, (x, label) in enumerate(cifar_train):
            #x, label = x.to(device), label.to(device)
            logits = model(x)
            # logits: [b, 10]
            # label:  [b]
            loss = criteon(logits, label)
            
            # backward
            optimizer.zero_grad()
            loss.backward()
            optimizer.step()
        
        print('train:', epoch, loss.item())
        
        ########## test  ##########
        model.eval()
        with torch.no_grad():
            total_correct = 0
            total_num = 0
            for x, label in cifar_test:
                # x, label = x.to(device), label.to(device)

                # [b]
                logits = model(x)
                # [b]
                pred = logits.argmax(dim=1)
                # [b] vs [b]
                total_correct += torch.eq(pred, label).float().sum().item()
                total_num += x.size(0)
            acc = total_correct / total_num
            print('test:', epoch, acc)

if __name__ == '__main__':
    main()

По сравнению с LeNet, ResNet очень быстро повышает точность, но из-за увеличения количества слоев неизбежно увеличивается время работы, без GPU прогон эпохи занимает около 15 минут. Читатели также могут изменить структуру сети на этой основе и использовать некоторые приемы, такие как нормализация изображения в начале и т. д.