1. Dataset & DataLoader🍁
在
PyTorch
中,Dataset
和DataLoader
是用来处理数据的重要工具。它们的作用分别如下:
Dataset
: Dataset 用于存储数据样本及其对应的标签。在使用神经网络训练时,通常需要将原始数据集转换为 Dataset 对象,以便能够通过 DataLoader 进行批量读取数据,同时也可以方便地进行数据增强、数据预处理等操作。
DataLoader
: DataLoader 用于将 Dataset 封装成一个可迭代对象,以便轻松地访问数据集中的样本。通过设置 batch_size 参数,DataLoader 可以将数据集分成若干个批次,每个批次包含指定数量的样本。此外,DataLoader 还支持对数据进行 shuffle、多线程读取等操作,使得训练过程更加高效。
使用 Dataset 和 DataLoader 可以使得数据处理过程更加模块化和可维护,同时也可以提高训练效率。分别封装在torch.utils.data.Dataset
和torch.utils.data.DataLoader
。
class MyDataset(Dataset):def __init__(self):def __len__(self): def __getitem__(self):
这是一个定义了自定义数据集类 MyDataset 的模板代码,它继承了 PyTorch 中的
Dataset
类,其中包含了三个必要的函数:
__init__
:用于初始化数据集,可以在这个函数中读取数据、进行预处理等操作。
__len__
:用于返回数据集中样本的数量。
__getitem__
:用于根据给定的索引 index 返回对应的样本及其标签。在这个函数中,需要根据索引从数据集中读取相应的样本和标签,并进行相应的预处理和转换。
需要在这个模板代码中添加具体的代码实现,以实现自定义数据集的功能。
from torch.utils.data import DataLoadertrain_dataloader = DataLoader(training_data, batch_size=64, shuffle=True)
test_dataloader = DataLoader(test_data, batch_size=64, shuffle=True)
使用
DataLoaders
准备训练和测试数据。在训练模型时,我们通常希望以“小批量(minibatches)”方式传递样本,每个 epoch 重新洗牌数据以减少模型过拟合,DataLoader 是一个可迭代对象。
next(iter(train_dataloader))
iter(train_dataloader)
将 train_dataloader 转换为一个迭代器对象,可以通过next
函数逐一获取 DataLoader 中的数据。因此,next(iter(train_dataloader))
将返回一个包含一个 batch 数据的元组。
具体来说,next 函数会从 train_dataloader 中获取下一个 batch 的数据,并将其转换为一个元组 (batch_data, batch_labels),其中 batch_data 是一个张量(tensor),形状为 [batch_size, input_size],表示一个 batch 中所有样本的输入特征;batch_labels 也是一个张量,形状为 [batch_size, output_size],表示一个 batch 中所有样本的输出标签,下面再举个例子吧。
my_list = [1, 2, 3, 4, 5]
my_iterator = iter(my_list)
print(next(my_iterator)) # 输出 1
print(next(my_iterator)) # 输出 2
print(next(my_iterator)) # 输出 3
在上面的例子中,my_list 是一个列表对象,通过 iter() 函数将其转换为迭代器 my_iterator。然后通过 next() 函数依次获取 my_iterator 中的每一个元素。
DataLoader
在创建时可以指定多个参数来控制数据的加载方式,常用的参数如下:
dataset
:指定要加载的数据集。
batch_size
:指定每个 batch 中样本的数量。
shuffle
:指定是否在每个 epoch 开始时洗牌数据集。
sampler
:指定一个自定义的数据采样器,用于控制每个 batch 中的样本顺序。
batch_sampler
:指定一个自定义的 batch 采样器,用于控制 batch 的顺序和样本数量。
num_workers
:指定数据加载时的线程数,用于加速数据读取。
collate_fn
:指定一个自定义的函数,用于将一个 batch 中的多个样本拼接为一个张量(tensor)。
pin_memory
:指定是否将数据加载到 GPU 的显存中,以加速数据读取。
drop_last
:指定在数据集大小不是 batch_size 的倍数时,是否丢弃最后一个不足 batch_size 的 batch。
2. Build Model🌺
import torch
from torch import nndevice = "cuda" if torch.cuda.is_available() else "cpu"
print(f"Using {device} device")
我们通过继承
nn.Module
来定义神经网络,并在__init__
中初始化神经网络的层。每个nn.Module
子类在forward
方法中实现对输入数据的操作。
class NeuralNetwork(nn.Module):def __init__(self):super().__init__()self.flatten = nn.Flatten()self.linear_relu_stack = nn.Sequential(nn.Linear(28*28, 512),nn.ReLU(),nn.Linear(512, 512),nn.ReLU(),nn.Linear(512, 10),)def forward(self, x):x = self.flatten(x)logits = self.linear_relu_stack(x)return logits
这段代码定义了一个名为 NeuralNetwork 的神经网络类,它继承自
nn.Module
。
这个神经网络包含一个 Flatten 层和一个由3个线性层和2个 ReLU 激活函数组成的神经网络层。
__init__
方法:在Python
中,当一个类继承自另一个类时,它会继承该类的所有属性和方法。在PyTorch
中,当你定义一个自己的神经网络类时,你通常会继承nn.Module
这个基类,因为nn.Module
已经定义好了很多用于搭建神经网络的基本组件和方法。
当你定义自己的神经网络类时,你需要调用基类的构造函数来继承基类的属性和方法。super().__init__()
就是调用基类(nn.Module)的构造函数,并返回一个代表基类实例的对象,这样你的神经网络类就可以使用 nn.Module 的所有属性和方法了。
forward
方法:就是神经网络的前向传播过程。
model = NeuralNetwork().to(device)
这行代码创建了一个名为 model 的神经网络模型实例,使用了前面定义的 NeuralNetwork 类,并将其移动到了特定的设备(CPU 或 GPU)上。使用
to()
方法可以将模型移动到特定的设备上,从而利用 GPU 加速模型的训练和推理。如果设备是 GPU,则模型的所有参数和缓存都会复制到 GPU 上,如果设备是 CPU,则会复制到系统内存中。
3. Optimization🌹
import torch
from torch import nn
from torch.utils.data import DataLoader
from torchvision import datasets
from torchvision.transforms import ToTensortraining_data = datasets.FashionMNIST(root="data",train=True,download=True,transform=ToTensor()
)test_data = datasets.FashionMNIST(root="data",train=False,download=True,transform=ToTensor()
)train_dataloader = DataLoader(training_data, batch_size=64)
test_dataloader = DataLoader(test_data, batch_size=64)class NeuralNetwork(nn.Module):def __init__(self):super(NeuralNetwork, self).__init__()self.flatten = nn.Flatten()self.linear_relu_stack = nn.Sequential(nn.Linear(28*28, 512),nn.ReLU(),nn.Linear(512, 512),nn.ReLU(),nn.Linear(512, 10),)def forward(self, x):x = self.flatten(x)logits = self.linear_relu_stack(x)return logitsmodel = NeuralNetwork()
使用
FashionMNIST
数据集,和之前描述的Datasets & DataLoaders
和Build Model
。
learning_rate = 1e-3
batch_size = 64
epochs = 5
learning_rate
:在每个 batch/epoch 更新模型参数的量。较小的值会导致较慢的学习速度,而较大的值可能会在训练过程中产生不可预测的行为。
batch_size
:在更新参数之前,通过网络传播的数据样本数量。
epochs
:迭代数据集的次数。
def train_loop(dataloader, model, loss_fn, optimizer):size = len(dataloader.dataset)for batch, (X, y) in enumerate(dataloader):# Compute prediction and losspred = model(X)loss = loss_fn(pred, y)# Backpropagationoptimizer.zero_grad()loss.backward()optimizer.step()if batch % 100 == 0:loss, current = loss.item(), (batch + 1) * len(X)print(f"loss: {loss:>7f} [{current:>5d}/{size:>5d}]")
定义模型训练函数:
在for
循环中,我们使用enumerate
函数遍历 dataloader 中的每个批次(batch),并将批次索引(batch index)和包含输入数据和标签的元组解压缩为 X 和 y。
然后计算出当前批次中的预测(prediction)和损失(loss),以便我们可以通过优化器(optimizer)调整模型的参数以最小化损失。
其次的三行执行反向传播(backpropagation
)并使用优化器更新模型的参数。optimizer.zero_grad()
将优化器的梯度归零,否则梯度会出现累加现象。然后使用backward
函数计算损失相对于模型参数的梯度,最后使用step
函数将优化器的梯度更新应用到模型的参数上。
这个if
语句在每100个批次之后打印出当前的损失和训练样本数量,以便我们可以了解模型的训练进度。
def test_loop(dataloader, model, loss_fn):size = len(dataloader.dataset)num_batches = len(dataloader)test_loss, correct = 0, 0with torch.no_grad():for X, y in dataloader:pred = model(X)test_loss += loss_fn(pred, y).item()correct += (pred.argmax(1) == y).type(torch.float).sum().item()test_loss /= num_batchescorrect /= sizeprint(f"Test Error: \n Accuracy: {(100*correct):>0.1f}%, Avg loss: {test_loss:>8f} \n")
定义模型测试函数:
在前三行中,我们计算出数据集的大小和批次数量,并初始化测试损失(test loss)和正确分类数量(correct)。
这个with
语句在上下文中禁用梯度计算,因为测试阶段不需要计算梯度,以便我们可以仅使用模型的前向传递(forward pass)进行测试。在这个 for 循环中,我们遍历 dataloader 中的每个批次,使用模型计算出预测,计算当前批次的测试损失,并使用 argmax 函数找到每个样本的预测标签,然后将正确分类的数量累加到 correct 变量中。
计算出平均测试损失和正确分类的比例,并打印出测试结果。我们将测试损失除以批次数量来得到平均测试损失,并将正确分类的数量除以数据集大小来得到正确分类的比例。最后,我们打印出测试结果,其中包括正确分类的百分比和平均测试损失。
correct += (pred.argmax(1) == y).type(torch.float).sum().item()
这行代码有点抽象:
这行代码的作用是计算当前批次中正确分类的数量,它可以分为几个步骤来理解:
首先,pred.argmax(1)
用来计算模型预测的最大概率值对应的类别,其中1表示按行计算最大值,即计算每个样本最有可能属于哪个类别。
接下来,pred.argmax(1) == y
用于将预测类别与真实类别进行比较,生成一个大小为批次大小的布尔张量,表示哪些样本被正确分类了。
然后,(pred.argmax(1) == y).type(torch.float)
将布尔张量转换为浮点数张量,其中正确分类的样本对应的元素值为1,错误分类的样本对应的元素值为0。
最后,.sum().item()
用于将正确分类的样本的元素值求和,并将结果转换为 Python 数值类型。
loss_fn = nn.CrossEntropyLoss()
optimizer = torch.optim.SGD(model.parameters(), lr=learning_rate)for t in range(epochs):print(f"Epoch {t+1}\n-------------------------------")train_loop(train_dataloader, model, loss_fn, optimizer)test_loop(test_dataloader, model, loss_fn)
print("Done!")
定义了一个交叉熵损失函数和一个随机梯度下降(SGD)优化器。交叉熵损失通常用于多类别分类问题,而 SGD 优化器是一种基本的梯度下降算法,用于更新模型的参数,使其逐渐逼近最优值。
这里定义了一个循环,用于多次训练和测试模型。具体来说,循环会运行 epochs 次,其中每次循环代表一个“训练周期”(epoch),在每个训练周期中,代码会先调用 train_loop() 函数来训练模型,然后调用 test_loop() 函数来测试模型在测试集上的性能。
4. Save & Load Model🌸
# Additional information
# 记录模型的相关训练信息
EPOCH = 5
PATH = "model.pt"
LOSS = 0.4torch.save({'epoch': EPOCH,'model_state_dict': net.state_dict(),'optimizer_state_dict': optimizer.state_dict(),'loss': LOSS,}, PATH)
下面是模型的加载。
model = Net() # 自己定义的网络
optimizer = optim.SGD(net.parameters(), lr=0.001, momentum=0.9)checkpoint = torch.load(PATH)
model.load_state_dict(checkpoint['model_state_dict'])
optimizer.load_state_dict(checkpoint['optimizer_state_dict'])
epoch = checkpoint['epoch']
loss = checkpoint['loss']model.eval()
# - or -
model.train()