第 6 课：模型搭建

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

在上一章中我们学习了 如何读取数据。这章来学习如何搭建神经网络。

深度学习通常通过深度神经网络实现，这些网络由多个层组成，我们通常称之为模型（Model）
在pytorch中整个模型是一个Module，各网络层、模块也可以称之为 Module

Module是所有神经网络的基类，所有的模型都必须继承于 torch.nn.Module，
一个Module里可以嵌套另外一个Module

（一）搭建基本的神经网络

在使用 Module 类时，有两个步骤：

1. 定义网络层（__init__ 方法）：在该方法中定义模型的各个层（如卷积层、线性层等）。
2. 前向传播（forward 方法）：在该方法中定义数据通过网络时的前向传播过程。

import torch
import torch.nn as nn
import torch.optim as optim

class MyModel(nn.Module):
    def __init__(self):
        super(MyModel, self).__init__()    # 调用父类 `nn.Module` 的初始化方法。
        # 定义网络层
        self.conv1 = nn.Conv2d(1, 32, kernel_size=3, stride=1, padding=1)  # 卷积层：将输入图像从 1 通道转换为 32 通道
        self.conv2 = nn.Conv2d(32, 64, kernel_size=3, stride=1, padding=1) # 卷积层：从 32 通道转换为 64 通道
        self.fc1 = nn.Linear(64 * 28 * 28, 128)                            # 全连接层：假设输入图像大小是 28x28，将展平后的特征图转换为 128 维
        self.fc2 = nn.Linear(128, 10)                                      # 全连接层：假设分类任务有 10 类，将 128 维的特征转换为输出的 10 维（适合 10 类分类问题）。

    def forward(self, x):
        # 定义前向传播
        x = self.conv1(x)  # 数据先通过卷积层 `conv1`。
        x = torch.relu(x)  # ReLU 激活函数
        x = self.conv2(x)
        x = torch.relu(x)
        x = x.view(-1, 64 * 28 * 28)  # 卷积层输出一个 4D 张量，`view` 将其展平为 1D 张量，以便输入到全连接层。
        x = self.fc1(x)    # 数据通过第一个全连接层。
        x = torch.relu(x)
        x = self.fc2(x)    # 数据通过第二个全连接层，最终输出。
        return x  

我们使用经典的 MNIST手写数字数据集 来做一个示范

第一步：加载数据集

MNIST 数据集（28x28 像素的灰度手写数字图像，共有 10 类，分别对应数字 0-9）。

# 数据预处理
transform = transforms.Compose([
    transforms.ToTensor(),  # 将像素值从 `0-255` 映射到 `0-1` 的浮点数
    transforms.Normalize((0.5,), (0.5,))])   # 归一化 将输入值从 `[0, 1]` 映射到 `[-1, 1]
# 下载并加载 MNIST 数据集
train_dataset = datasets.MNIST(root='./data', train=True, download=True, transform=transform)
test_dataset = datasets.MNIST(root='./data', train=False, download=True, transform=transform)

train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True) # 在每个 epoch 开始时随机打乱数据，提高模型的泛化能力
test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False)  # 测试数据无需打乱，保持顺序加载。

可以用以下代码查看数据加载结果：

# 获取一个批次的数据
images, labels = next(iter(train_loader))
# iter()：将 train_loader 转换为一个迭代器。
# next()：从迭代器中获取下一个批次的数据。每个批次的数据包括图像和标签（images 和 labels）

print(f"Batch size: {images.shape}")  # 输出: torch.Size([64, 1, 28, 28])
print(f"Labels: {labels[:10]}")       # 打印前 10 个标签

# 可视化一个样本
import matplotlib.pyplot as plt
plt.figure(figsize=(2,2))
plt.imshow(images[0].squeeze(), cmap="gray")
plt.title(f"Label: {labels[0].item()}")
plt.show()

Batch size: torch.Size([64, 1, 28, 28]) Labels: tensor([3, 2, 4, 6, 7, 6, 3, 8, 1, 3])

第二步：搭建模型

我们构建一个简单的全连接神经网络（Fully Connected Neural Network）：

1. 输入层：接收 28x28 的图像（展平成 784 个输入）。
2. 隐藏层：有 128 个神经元，激活函数为 ReLU。
3. 输出层：10 个输出（对应 10 个类别）。

class SimpleNN(nn.Module):
    def __init__(self):
        super(SimpleNN, self).__init__()
        self.fc1 = nn.Linear(28 * 28, 128)  # 输入层到隐藏层
        self.fc2 = nn.Linear(128, 10)      # 隐藏层到输出层
        self.relu = nn.ReLU()              # 激活函数

    def forward(self, x):
        x = x.view(-1, 28 * 28)  # 展平成一维张量
        x = self.relu(self.fc1(x))
        x = self.fc2(x)
        return x

# 创建模型实例
model = SimpleNN()
print(model)

SimpleNN(
  (fc1): Linear(in_features=784, out_features=128, bias=True)
  (fc2): Linear(in_features=128, out_features=10, bias=True)
  (relu): ReLU()
)

第三步：定义损失函数和优化器

• 损失函数：我们使用交叉熵损失（CrossEntropyLoss）处理分类任务。
• 优化器：随机梯度下降（SGD）优化模型参数，学习率设置为 0.01。

import torch.optim as optim
criterion = nn.CrossEntropyLoss()  # 定义损失函数
optimizer = optim.SGD(model.parameters(), lr=0.01)  # 定义优化器
# model.parameters() 返回模型中所有可以训练的参数（即网络的权重和偏置）。
# 优化器将根据这些参数的梯度来更新权重，以最小化损失函数。

第四步：训练模型

epochs = 5  # 训练轮次
# 一个 epoch 表示网络在整个训练数据集上完成一次前向传播和反向传播。

for epoch in range(epochs):
    model.train()  # 设置模型为训练模式， 启用模型中与训练相关的功能（如 Dropout 和 BatchNorm）
    running_loss = 0.0  #初始化损失值
     
    for images, labels in train_loader:  
        #train_loader 会将数据集拆分成小批次（batch），每个批次大小为 64
        # images 是一个形状为 [batch_size, channels, height, width] 的张量。

        
        optimizer.zero_grad()  # 清空梯度
        
        outputs = model(images) # 前向传播
        # outputs 是模型的预测结果，通常形状为 [batch_size, num_classes]，即每个样本的预测分数。
        
        loss = criterion(outputs, labels) # 计算损失
        
        # 反向传播
        loss.backward()
        
        # 更新参数
        optimizer.step()
        
        running_loss += loss.item()   #累计该轮训练过程中的总损失，以便在结束时输出平均损失
    
    print(f"Epoch {epoch+1}/{epochs}, Loss: {running_loss/len(train_loader):.4f}")

Epoch 1/5, Loss: 0.7274
Epoch 2/5, Loss: 0.3662
Epoch 3/5, Loss: 0.3214
Epoch 4/5, Loss: 0.2960
Epoch 5/5, Loss: 0.2760

第五步：测试

# 测试模型
model.eval()  # 设置模型为评估模式
correct = 0
total = 0

with torch.no_grad():  # 测试时不需要计算梯度
    for images, labels in test_loader:
        # print(labels)
        outputs = model(images)
        _, predicted = torch.max(outputs, 1)  # 获取最大概率的类别
        total += labels.size(0)
        correct += (predicted == labels).sum().item()

accuracy = correct / total
print(f"Test Accuracy: {accuracy * 100:.2f}%")

Test Accuracy: 92.39%

（二）在自定义的数据集上搭建神经网络

第一步：加载数据

from Covid_DataSet import COVID19Dataset

root_dir = "./data/cov_19_demo"  # path to datasets——covid-19-demo
img_dir = os.path.join(root_dir, "imgs")
path_txt_train = os.path.join(root_dir, "labels", "train.txt")
path_txt_valid = os.path.join(root_dir, "labels", "valid.txt")
transforms_func = transforms.Compose([
    transforms.Resize((8, 8)),
    transforms.ToTensor(),
])
train_data = COVID19Dataset(root_dir=img_dir, txt_path=path_txt_train, transform=transforms_func)
valid_data = COVID19Dataset(root_dir=img_dir, txt_path=path_txt_valid, transform=transforms_func)
train_loader = DataLoader(dataset=train_data, batch_size=2)
valid_loader = DataLoader(dataset=valid_data, batch_size=2)

第二步：定义模型

class TinnyCNN(nn.Module):
    def __init__(self, cls_num=2):
        super(TinnyCNN, self).__init__()
        self.convolution_layer = nn.Conv2d(1, 1, kernel_size=(3, 3))
        self.fc = nn.Linear(36, cls_num)
    
    def forward(self, x):
        x = self.convolution_layer(x)
        x = x.view(x.size(0), -1)
        out = self.fc(x)
        return out

model = TinnyCNN(2)

第三步：优化模块

设定损失函数与优化器，用于在训练过程中对网络参数进行更新

loss_f = nn.CrossEntropyLoss()
optimizer = optim.SGD(model.parameters(), lr=0.1, momentum=0.9, weight_decay=5e-4)
scheduler = optim.lr_scheduler.StepLR(optimizer, gamma=0.1, step_size=50)

第四步：迭代模块

循环迭代地进行模型训练，数据一轮又一轮的喂给模型，不断优化模型，直到我们让它停止训练

for epoch in range(100):
    # 训练集训练
    model.train()
    for data, labels in train_loader:
        # forward & backward
        outputs = model(data)
        optimizer.zero_grad()

        # loss 计算
        loss = loss_f(outputs, labels)
        loss.backward()
        optimizer.step()

        # 计算分类准确率
        _, predicted = torch.max(outputs.data, 1)
        correct_num = (predicted == labels).sum()
        acc = correct_num / labels.shape[0]
        print("Epoch:{} Train Loss:{:.2f} Acc:{:.0%}".format(epoch, loss, acc))
        # print(predicted, labels)

    # 验证集验证
    model.eval()
    for data, label in valid_loader:
        # forward
        outputs = model(data)

        # loss 计算
        loss = loss_f(outputs, labels)

        # 计算分类准确率
        _, predicted = torch.max(outputs.data, 1)
        correct_num = (predicted == labels).sum()
        acc_valid = correct_num / labels.shape[0]
        print("Epoch:{} Valid Loss:{:.2f} Acc:{:.0%}".format(epoch, loss, acc_valid))

    # 添加停止条件
    if acc_valid == 1:
        break

    # 学习率调整
    scheduler.step()

Epoch:0 Train Loss:0.69 Acc:50%
Epoch:0 Valid Loss:0.70 Acc:50%
Epoch:1 Train Loss:0.69 Acc:50%
Epoch:1 Valid Loss:0.70 Acc:50%
...
Epoch:97 Train Loss:0.00 Acc:100%
Epoch:97 Valid Loss:8.29 Acc:50%
Epoch:98 Train Loss:0.00 Acc:100%
Epoch:98 Valid Loss:8.29 Acc:50%
Epoch:99 Train Loss:0.00 Acc:100%
Epoch:99 Valid Loss:8.29 Acc:50%