Pytorch学习笔记（五）——多层感知机的实现

本文预备知识：Pytorch学习笔记（四）——torchvision工具箱

• 一、FashionMNIST数据集
• 二、torch.nn
• • 2.1 nn.Module
  • 2.2 nn.Sequential
  • 2.3 查看神经网络的参数
• 三、Loss Function
• 四、torch.optim
• 五、训练与测试
• 附录：完整代码

数据集官网：https://github.com/zalandoresearch/fashion-mnist.

FashionMNIST数据集是一个用于图像分类的数据集（服装类型），总共有十类，对应关系如下：

其中训练集有 60000 60000 60000 个样本，测试集有 10000 10000 10000 个样本，每个样本是一个 28 × 28 28times 28 28×28 的灰度图（即图像模式为 L，单通道），且样本的标签都是整型数字。

设每个 batch 的大小为 64 64 64，我们通过如下操作来下载数据集并将其装载到 DataLoader 中：

import torch
from torchvision.transforms import ToTensor
from torch.utils.data import DataLoader

train_data = datasets.FashionMNIST(root='./data', train=True, transform=ToTensor(), download=True)
test_data = datasets.FashionMNIST(root='./data', train=False, transform=ToTensor(), download=True)
train_loader = DataLoader(train_data, batch_size=64)
test_loader = DataLoader(test_data, batch_size=64)

nn.Module 是所有和神经网络有关的模块的基类。自定义的神经网络必须继承该类，并重写 forward 方法（用于计算正向传播）。

模块中还可以包含其他模块，并允许它们嵌套成一个树状结构。

我们通常会按照如下模板来自定义神经网络：

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        # 写出神经网络的架构
        
    def forward(self, inputs):
        # 计算正向传播

需要注意的是，当我们调用 Net 的实例时，forward 方法会被自动执行，如下：

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        
    def forward(self, inputs):
        return inputs + 1


net = Net()
print(net(1))
# 2

我们可以把 nn.Sequential 理解为一个流水线，里面装有各种模块。当输入传入流水线后，它会经过各个模块最终得到一个输出。

例如对于单隐层的神经网络而言，假设输入层有 100 100 100 个神经元，隐层有 50 50 50 个神经元，输出层有 10 10 10 个神经元，则相应的 nn.Sequential 可写为：

model = nn.Sequential(
    nn.Linear(100, 50),
    nn.ReLU(),
    nn.Linear(50, 10),
)

nn.Linear() 会在合理的范围内进行初始化。若要计算正向传播，则只需将输入传入 model 即可：

inputs = torch.rand(100)
print(model(inputs))
# tensor([ 0.1086, -0.1213,  0.0516, -0.1837, -0.0326, -0.1893,  0.1336,  0.0476,
#          0.1902,  0.1514], grad_fn=)

于是我们自定义的神经网络类可以写为：

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.model = nn.Sequential(
            nn.Linear(100, 50),
            nn.ReLU(),
            nn.Linear(50, 10),
        )

    def forward(self, inputs):
        return self.model(inputs)

现在我们来看一下，对于FashionMNIST数据集，我们该如何定制神经网络。注意到每张图片都已经转化成了一个 28 × 28 28times 28 28×28 的张量，我们需要先将其展平成长度为 784 784 784 的向量才能输入到神经网络。为此只需要使用 nn.Flatten()。

假设有两个隐层，神经元个数分别为 512 512 512 和 256 256 256，则自定义的神经网络如下：

class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.model = nn.Sequential(
            nn.Flatten(),
            nn.Linear(784, 512),
            nn.ReLU(),
            nn.Linear(512, 256),
            nn.ReLU(),
            nn.Linear(256, 10),
        )

    def forward(self, inputs):
        return self.model(inputs)

实例化后，我们可以随时调用 parameters() 或 named_parameters() 方法来查看神经网络的参数。

对于上述的神经网络而言，它一共有六个参数。分别是第 1 1 1 个模块的权重和阈值（从 0 0 0 开始编号）、第 3 3 3 个模块的权重和阈值、第 5 5 5 个模块的权重和阈值。

为简洁起见，这里只打印参数相应的名称和形状：

net = Net()
for name, param in net.named_parameters():
    print(name, param.shape)
# model.1.weight torch.Size([512, 784])
# model.1.bias torch.Size([512])
# model.3.weight torch.Size([256, 512])
# model.3.bias torch.Size([256])
# model.5.weight torch.Size([10, 256])
# model.5.bias torch.Size([10])

而 net.parameters() 中只含参数，不含名称：

net = Net()
for param in net.parameters():
    print(param.shape)
# torch.Size([512, 784])
# torch.Size([512])
# torch.Size([256, 512])
# torch.Size([256])
# torch.Size([10, 256])
# torch.Size([10])

net.parameters() 和 net.named_parameters() 均是生成器，进而是迭代器。

在得到神经网络的输出后，我们还需要计算损失函数以便后续的参数优化。对于分类问题，常用的损失函数为交叉熵损失。调用 nn.CrossEntropyLoss() 后它会自动对神经网络的原始输出（即 logits，又称 scores）应用 SoftMax 变换并根据 target 计算相应的交叉熵损失。

先不考虑 batch，只考虑单个样本。此时 logits 应是长度为 C C C 的张量，其中 C C C 代表类别个数。而 target 可以有两种形式，第一种是类别的下标，取值必须在 { 0 , 1 , 2 , ⋯   , C − 1 } {0,1,2,cdots,C-1} {0,1,2,⋯,C−1} 内。接下来举个例子方便理解：

import torch
from torch.nn import functional as F

""" 代码片段一 """
torch.manual_seed(42)
logits = torch.randn(3, requires_grad=True)
# tensor([0.3367, 0.1288, 0.2345], requires_grad=True)
target = torch.tensor(1)
F.cross_entropy(logits, target)
# tensor(1.2067, grad_fn=)

""" 代码片段二 """
def cross_entropy(logits, target):
    logits = logits.softmax(dim=0)
    return -torch.log(logits[target])


torch.manual_seed(42)
logits = torch.randn(3, requires_grad=True)
target = torch.tensor(1)
cross_entropy(logits, target)
# tensor(1.2067, grad_fn=)

以上两段代码完全等价。

考虑 batch 时，此时 logits 是形状为 (B, C) 的张量，其中 B B B 是每个 batch 中样本的个数，target 是形状为 (B, ) 的张量。

target 的第二种形式是每个样本属于每个类别的概率，此时形状应与 logits 相同：

logits = torch.randn(2, 3, requires_grad=True)
# tensor([[ 0.9652,  1.0090, -0.0337],
#         [-1.0090, -1.2315, -1.0470]], requires_grad=True)
target = torch.randn(2, 3).softmax(dim=1)
# tensor([[0.1144, 0.4511, 0.4345],
#         [0.0283, 0.1125, 0.8592]])
output = F.cross_entropy(logits, target)
# tensor(1.1846, grad_fn=)

上述代码说明每个 batch 的大小是 2 2 2，且是三分类问题。

一般我们使用 nn.CrossEntropyLoss()，假设 batch 大小为 64 64 64，数据集共有十类：

loss_fn = nn.CrossEntropyLoss()
logits = torch.randn(64, 10, requires_grad=True)
target = torch.randint(10, (64, ))
loss = loss_fn(logits, target)
loss
# tensor(2.6828, grad_fn=)

在得到损失函数的值后，我们一般使用 torch.optim 来构造一个优化器，它能够利用计算好的梯度对网络中的各个参数进行更新。

torch.optim.Optimizer 是 Pytorch 中所有优化器的基类，我们可以基于该类构造自己的优化器，但本文并不会讲解这一点，而是聚焦于如何使用现有的优化器。

神经网络大多使用随机梯度下降算法来优化参数，我们可以使用 torch.optim 中的 SGD 优化器，格式如下：

from torch.optim import SGD

optimizer = SGD(net.parameters(), lr=0.001)  # 学习率可根据自己的需要进行设置

在使用 loss.backward() 计算梯度后，我们就可以使用 optimizer.step() 来更新参数。但需要注意的是，optimizer 默认累积梯度，因此在 backward() 之前我们需要调用 optimizer.zero_grad() 清零梯度，因此完整的一个反向传播过程为：

optimizer.zero_grad()
loss.backward()
optimizer.step()

因为每个 epoch 中有多次训练和测试，所以我们可以事先写好训练和测试的代码，这样可以使最终的代码更加简洁。

def train_loop(dataloader, net, loss_fn, optimizer):
    # 数据集大小
    size = len(dataloader.dataset)
    for batch_idx, (X, y) in enumerate(dataloader):
        # 正向传播
        loss = loss_fn(net(X), y)

        # 反向传播
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        # 每隔200个batch输出损失
        if batch_idx % 200 == 0:
            loss, current = loss.item(), batch_idx * len(X)
            print(f"loss: {loss:>7f}  [{current:>5d}/{size:>5d}]")

def test_loop(dataloader, net, loss_fn):
    # 初始化
    size = len(dataloader.dataset)
    num_batches = len(dataloader)
    test_loss, correct = 0, 0
    
    # 计算所有batch上的损失和分类正确的个数
    with torch.no_grad():
        for X, y in dataloader:
            pred = net(X)
            test_loss += loss_fn(pred, y).item()
            correct += (pred.argmax(dim=1) == y).sum().item()
    
    # 计算分类准确率和平均损失
    test_loss /= num_batches
    correct /= size
    print(f"Test Error: n Accuracy: {(100*correct):>0.1f}%, Avg loss: {test_loss:>8f} n")

在所有的类和函数都定义好后，我们就可以开始训练/测试神经网络了：

learning_rate = 0.001
num_epochs = 5

net = Net()
loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(net.parameters(), lr=learning_rate)

for epoch in range(num_epochs):
    print(f"Epoch {epoch+1}n-------------------------------")
    train_loop(train_loader, net, loss_fn, optimizer)
    test_loop(test_loader, net, loss_fn)
print("Done!")

训练了5个epoch的结果如下：

Epoch 1
-------------------------------
loss: 2.305738  [    0/60000]
loss: 2.273370  [12800/60000]
loss: 2.245479  [25600/60000]
loss: 2.233475  [38400/60000]
loss: 2.205799  [51200/60000]
Test Error: 
 Accuracy: 46.1%, Avg loss: 2.170497 

Epoch 2
-------------------------------
loss: 2.179702  [    0/60000]
loss: 2.114199  [12800/60000]
loss: 2.074621  [25600/60000]
loss: 2.058244  [38400/60000]
loss: 1.992608  [51200/60000]
Test Error: 
 Accuracy: 57.3%, Avg loss: 1.923667 

Epoch 3
-------------------------------
loss: 1.950942  [    0/60000]
loss: 1.804799  [12800/60000]
loss: 1.734545  [25600/60000]
loss: 1.719184  [38400/60000]
loss: 1.638393  [51200/60000]
Test Error: 
 Accuracy: 59.3%, Avg loss: 1.555051 

Epoch 4
-------------------------------
loss: 1.618746  [    0/60000]
loss: 1.427796  [12800/60000]
loss: 1.377516  [25600/60000]
loss: 1.395492  [38400/60000]
loss: 1.354157  [51200/60000]
Test Error: 
 Accuracy: 61.5%, Avg loss: 1.283596 

Epoch 5
-------------------------------
loss: 1.366808  [    0/60000]
loss: 1.176103  [12800/60000]
loss: 1.147784  [25600/60000]
loss: 1.201756  [38400/60000]
loss: 1.180527  [51200/60000]
Test Error: 
 Accuracy: 63.2%, Avg loss: 1.120887 

Done!

import torch
from torchvision.transforms import ToTensor
from torchvision import datasets
from torch.utils.data import DataLoader
from torch import nn


class Net(nn.Module):
    def __init__(self):
        super().__init__()
        self.model = nn.Sequential(
            nn.Flatten(),
            nn.Linear(784, 512),
            nn.ReLU(),
            nn.Linear(512, 256),
            nn.ReLU(),
            nn.Linear(256, 10),
        )

    def forward(self, inputs):
        return self.model(inputs)


def train_loop(dataloader, net, loss_fn, optimizer):
    size = len(dataloader.dataset)
    for batch_idx, (X, y) in enumerate(dataloader):
        loss = loss_fn(net(X), y)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        if batch_idx % 200 == 0:
            loss, current = loss.item(), batch_idx * len(X)
            print(f"loss: {loss:>7f}  [{current:>5d}/{size:>5d}]")


def test_loop(dataloader, net, loss_fn):
    size = len(dataloader.dataset)
    num_batches = len(dataloader)
    test_loss, correct = 0, 0
    with torch.no_grad():
        for X, y in dataloader:
            pred = net(X)
            test_loss += loss_fn(pred, y).item()
            correct += (pred.argmax(dim=1) == y).sum().item()
    test_loss /= num_batches
    correct /= size
    print(f"Test Error: n Accuracy: {(100*correct):>0.1f}%, Avg loss: {test_loss:>8f} n")


if __name__ == '__main__':
    train_data = datasets.FashionMNIST(root='./data', train=True, transform=ToTensor(), download=True)
    test_data = datasets.FashionMNIST(root='./data', train=False, transform=ToTensor(), download=True)
    train_loader = DataLoader(train_data, batch_size=64)
    test_loader = DataLoader(test_data, batch_size=64)

    learning_rate = 0.001
    num_epochs = 5

    net = Net()
    loss_fn = nn.CrossEntropyLoss()
    optimizer = torch.optim.SGD(net.parameters(), lr=learning_rate)

    for epoch in range(num_epochs):
        print(f"Epoch {epoch+1}n-------------------------------")
        train_loop(train_loader, net, loss_fn, optimizer)
        test_loop(test_loader, net, loss_fn)
    print("Done!")