实战1. 利用Pytorch解决 CIFAR 数据集中的图像分类为 10 类的问题

你的任务是建立一个用于 CIFAR 图像分类的神经网络，并实现分类质量 > 0.5。

注意：因为我们实战1里只讨论最简单的神经网络构建，所以准确率达到0.5以上就符合我们的目标，后面会不断学习新的模型进行优化

CIFAR的数据集如下图所示：

我们大概所需要的功能包如下：

import numpy as np

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

import torchvision
from torchvision import datasets, transforms

from matplotlib import pyplot as plt
from IPython.display import clear_output

下面我们开始构建模型

加载数据

加载数据的代码已经完全实现，不需要做任何改变。

CIFAR10 是一个分为 10 个类的彩色图像数据集。图片中有汽车、飞机和动物的图像。

train_data = datasets.CIFAR10(root="./cifar10_data", train=True, download=True, transform=transforms.ToTensor())
test_data = datasets.CIFAR10(root="./cifar10_data", train=False, download=True, transform=transforms.ToTensor())

# 将训练样本分为train和val
# 我们将把 80% 的图片纳入训练样本
train_size = int(len(train_data) * 0.8)
# 进行验证 - 剩余 20%
val_size = len(train_data) - train_size

train_data, val_data = torch.utils.data.random_split(train_data, [train_size, val_size])

# 启动数据加载器
train_loader = torch.utils.data.DataLoader(train_data, batch_size=64, shuffle=True)
val_loader = torch.utils.data.DataLoader(val_data, batch_size=64, shuffle=False)
test_loader = torch.utils.data.DataLoader(test_data, batch_size=64, shuffle=False)

让我们看一些图片来了解我们正在处理的事情。要绘制训练集的十张图像，请运行以下单元：

# 此函数绘制多幅图像
def show_images(images, labels):
    f, axes= plt.subplots(1, 10, figsize=(30,5))

    for i, axis in enumerate(axes):
        # 将图像从张量转换为numpy
        img = images[i].numpy()
        # 将图像转换为尺寸（长度、宽度、颜色通道）
        img = np.transpose(img, (1, 2, 0))

        axes[i].imshow(img)
        axes[i].set_title(labels[i].numpy())

    plt.show()

# 从训练数据加载器中获取一批图像
for batch in train_loader:
    # 一批图片和一批图片答案
    images, labels = batch
    break

# 调用函数绘制图像
show_images(images, labels)

每张图片上方都写有该图片所属类别的编号。

答案编号与类别对应表：

让我们看看图片的尺寸：

images.shape

输出：torch.Size([64, 3, 32, 32])

这里64是批量大小，3是颜色通道数（因为图片是彩色的，是RGB）32和32是图片的宽度和高度。

事实证明，每张图片由 32 * 32 * 3 = 3072 个值表示。网络第一层应该有3072个神经元。

建立模型

你的任务是建立一个模型，然后训练它。请不要构建过于复杂的网络，不要使其深度超过四层（更少是可能的）。你的主要任务是训练模型并在测试样本上获得良好的质量。

您可以使用专栏里的课程3中的模型代码作为基础。
不要忘记，第一层的神经元数量应该不同。

你可以尝试什么来提高模型的质量：

• 添加更多隐藏层；
• 分层制造更多神经元；
• 添加批量规范层。

批处理层是 BatchNorm1d（在下一个单元中导入）。该层应用在全连接层之后。例子：

def __init__(self):
        ...
        self.fc = nn.Linear(500, 100)
        self.bn = BatchNorm1d(100)
        ...

def forward(self, x):
        ...
        x = F.relu(self.fc(x))
        x = self.bn(x)
        ...

尝试在每个网络层之后插入一个 BatchNorm 层（最后一个网络层除外）。

为了成功通过任务，实现一个具有三层的网络和它们之间的 BatchNorm 就足够了。

这里我构建3072->128->64->10的网络：

# 导入 BatchNorm
from torch.nn.modules.batchnorm import BatchNorm1d

# 实现模型

class Model(nn.Module):
    def __init__(self):
      super(Model, self).__init__()
      # CIFAR - 10 图像尺寸为 3x32x32，展平后为 3072
      self.fc1 = nn.Linear(3 * 32 * 32, 128)
      self.bn1 = BatchNorm1d(128)
      self.fc2 = nn.Linear(128, 64)
      self.bn2 = BatchNorm1d(64)
      self.fc3 = nn.Linear(64, 10)

    def forward(self, x):
      # 展平输入
      x = x.view(-1, 3 * 32 * 32)
      x = F.relu(self.bn1(self.fc1(x)))
      x = F.relu(self.bn2(self.fc2(x)))
      x = self.fc3(x)
      return x

使用CPU/GPU操作：

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
device

输出：device(type=‘cpu’)

这里我用的是cpu

# 声明模型并将其传输到CPU/GPU
model = Model().to(device)

下面的单元格包含用于检查您的模型的测试。如果单元格没有返回错误，则模型运行正常。

assert model is not None, '模型变量为空。那么你的模型在哪里？'

try:
    x = images.reshape(-1, 3072).to(device)
    y = labels

    # 给定输入计算输出，两者都是变量
    y_predicted = model(x)
except Exception as e:
    print('该模型有问题')
    raise e


assert y_predicted.shape[-1] == 10, '模型的最后一层的神经元数量错误'

模型训练

现在我们需要训练模型。让我们使用课程3的用于评估模型质量的代码（评估函数）已经实现。

from sklearn.metrics import accuracy_score

def evaluate(model, dataloader, loss_fn):

    y_pred_list = []
    y_true_list = []
    losses = []

    # 我们浏览数据加载器批次
    for i, batch in enumerate(tqdm(dataloader)):

        # 这就是我们得到当前批次的方法
        X_batch, y_batch = batch

        # 关闭任何梯度的计算
        with torch.no_grad():

            # 我们收到该批次的网络响应
            logits = model(X_batch.to(device))

            # 我们计算批次上的损失函数值
            loss = loss_fn(logits, y_batch.to(device))
            loss = loss.item()

            # 将当前批次的损失保存到数组中
            losses.append(loss)

            # 对于我们理解的批次中的每个元素，
            # 网络将它分配到 0 到 9 中的哪个类别
            y_pred = torch.argmax(logits, dim=1)

        # 我们将当前批次的正确答案保存到数组中
        # 以及对当前批次的网络响应
        y_pred_list.extend(y_pred.cpu().numpy())
        y_true_list.extend(y_batch.numpy())

    # 我们计算网络响应和正确答案之间的准确率
    accuracy = accuracy_score(y_pred_list, y_true_list)

    return accuracy, np.mean(losses)

def train(model, loss_fn, optimizer, n_epoch=6):

    model.train(True)

    data = {
        'acc_train': [],
        'loss_train': [],
        'acc_val': [],
        'loss_val': []
    }

    # 网络训练周期
    for epoch in tqdm(range(n_epoch)):

        for i, batch in enumerate(tqdm(train_loader)):

            # 这就是我们获得当前一批图片和对它们的回应的方式
            X_batch, y_batch = batch

            # 前向传递（接收一批图像的网络响应）
            logits = model(X_batch.to(device))

            # 根据网络给出的答案和批次的正确答案计算损失
            loss = loss_fn(logits, y_batch.to(device))

            optimizer.zero_grad() # 我们重置优化器梯度的值
            loss.backward() # 反向传播（梯度计算）
            optimizer.step() # 更新网络权重

        # 时代终结，模型验证
        print('On epoch end', epoch)

        acc_train_epoch, loss_train_epoch = evaluate(model, train_loader, loss_fn)
        print('Train acc:', acc_train_epoch, 'Train loss:', loss_train_epoch)

        acc_val_epoch, loss_val_epoch = evaluate(model, val_loader, loss_fn)
        print('Val acc:', acc_val_epoch, 'Val loss:', loss_val_epoch)


        data['acc_train'].append(acc_train_epoch)
        data['loss_train'].append(loss_train_epoch)
        data['acc_val'].append(acc_val_epoch)
        data['loss_val'].append(loss_val_epoch)

    return model, data

# 声明模型并将其传输到 CPU/GPU
model = Model().to(device)

# 损失函数
loss_fn = torch.nn.CrossEntropyLoss()

# 优化器
learning_rate = 1e-3
optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

model, data = train(model, loss_fn, optimizer, n_epoch=10)

测试评估

让我们在测试样本上评估一下模型的质量：

test_acc, test_loss = evaluate(model, test_loader, loss_fn)
test_acc

输出：0.5214

满足我们的要求，我们进行模型保存，之后就可以直接调用了：

x = torch.randn((64, 32*32*3))
torch.jit.save(torch.jit.trace(model.cpu(), (x)), "model.pth")