一、PyTorch训练框架的两种实现方式

        1.数据预处理

        2.定义loss   

        3.定义优化器optimizer 

        4.前向传播,计算预测值,损失

        5.梯度清零 

        6.反向传播,计算梯度 

        7.更新参数 

方法一:使用PyTorch内置优化器(标准方式)

import torch
import torch.nn as nn

# 1. 数据预处理
# (假设数据X和标签y已准备好)

# 2. 定义损失函数
loss_fn = nn.CrossEntropyLoss()

# 3. 定义优化器
optimizer = torch.optim.SGD(model.parameters(), lr=learning_rate)
# 或 optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)

# 4. 前向传播,计算预测值和损失
y_hat = model(X)
loss = loss_fn(y_hat, y)

# 5. 梯度清零
optimizer.zero_grad()

# 6. 反向传播,计算梯度
loss.backward()

# 7. 更新参数
optimizer.step()

方法二:自定义实现(笔试手撕版)

import torch

# 1. 数据预处理(不变)

# 2. 自定义交叉熵损失函数
def cross_entropy(y_hat, y):
    """
    交叉熵损失函数实现
    公式: -∑ y_i log p_i,其中y为one-hot编码时,只有正确类别项为1
    简化实现: 直接使用正确类别的预测概率
    
    参数:
        y_hat: 预测概率分布 [batch_size, num_classes]
        y: 正确类别索引 [batch_size]
    
    返回:
        每个样本的损失 [batch_size]
    """
    return -torch.log(y_hat[range(len(y)), y])


# 3. 自定义优化器
def sgd(params, lr, batch_size):
    """
    小批量随机梯度下降
    
    参数:
        params: 需要更新的参数列表 [W, b, ...]
        lr: 学习率
        batch_size: 批量大小
    """
    with torch.no_grad():  # 不记录梯度更新操作
        for param in params:
            param -= lr * param.grad / batch_size  # 参数更新
            param.grad.zero_()  # 梯度清零

def updater(batch_size):
    """优化器调用函数"""
    sgd([W, b], lr, batch_size)

# 4. 前向传播,计算预测值和损失
y_hat = model(X)
loss = cross_entropy(y_hat, y)

# 5. 梯度清零(在自定义优化器中处理)

# 6. 反向传播计算梯度
loss.sum().backward()  # 将损失求和后反向传播

# 7. 更新参数
updater(X.shape[0])  # 传入批量大小

二、Softmax函数详解

Softmax是一个将神经网络最后一层的原始输出(logits)转换为归一化概率分布的函数,它通过指数运算放大分数差异,并确保所有类别概率之和为1,从而用于多分类任务。

将分数转换为概率分布:

\hat{y_j} =\tfrac{e^{x_i}}{\sum_{j=1}^{K} e^{x_j} },j = 1,2,3....,K

其中,K是类别总数,\hat{y_j}是预测的概率分布

  1. x 的数值分布越不均匀,则Softmax(x) 的两极化越明显
  2. 非负,和为1
  3. 导数简单:便于反向传播计算

问题:

  1. 数值上溢问题----减去最大值

当输入值非常大时,e^{x_{i}}可能超过计算机的浮点数表示范围(如float32),导致结果为inf或NaN

\hat{y_j} =\tfrac{e^{x_i-x_m}}{\Sigma e^{x_j-x_m} }

         2. 数值下溢问题---取对数

当输入值非常小时,e^{x_{i}}可能接近于0;在减法和规范化步骤之后,可能有些x_j-x_m具有较大的负值。由于精度受限,e^{x_j-x_m}将有接近零的值;

即为下溢(underflow)。 这些值可能会四舍五入为零,使\hat{y_j}为零, 并且使得log\hat{y_j}的值为-inf。 反向传播几步后,我们可能会发现自己面对一屏幕可怕的NaN结果。

为避免对e^{x_j-x_m}进行取对数,而可以直接使用x_j-x_m,因为log被抵消了。取对数

\hat{y_j} =log\tfrac{e^{x_i-x_m}}{\Sigma e^{x_j-x_m} } = x_i-x_m- log \Sigma e^{x_j-x_m}

def log_softmax_stable(x):
    """
    数值稳定的Log-Softmax实现
    用于直接计算对数概率,避免计算log(softmax(x))的数值问题
    
    参数:
        x: 输入张量
        
    返回:
        对数概率
    """
    m = np.max(x, axis=-1, keepdims=True)
    exp_x = np.exp(x - m)
    sum_exp = np.sum(exp_x, axis=-1, keepdims=True)
    log_sum_exp = np.log(sum_exp)
    
    return x - m - log_sum_exp

三.从0开始实现softmax对fashion_mnist进行分类完整代码:

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

from IPython import display
from d2l import torch as d2l

#------------------------1.数据预处理-----------------------#

def get_dataloader_workers():  #@save
    """windows使用0个进程来读取数据"""
    return 4

def load_data_fashion_mnist(batch_size, resize=None):  #@save
    """下载Fashion-MNIST数据集,然后将其加载到内存中"""
    # transforms.ToTensor() 做了三件事:
    # 1. 将PIL图像或numpy数组转换为PyTorch Tensor
    # 2. 将像素值从 [0, 255] 缩放到 [0.0, 1.0]
    # 3. 调整维度顺序:(H, W) → (C, H, W)
    trans = [transforms.ToTensor()]
    if resize:
        trans.insert(0, transforms.Resize(resize))

    trans = transforms.Compose(trans)
    mnist_train = torchvision.datasets.FashionMNIST(
        root="data", train=True, transform=trans, download=True)
    mnist_test = torchvision.datasets.FashionMNIST(
        root="data", train=False, transform=trans, download=True)
    return (data.DataLoader(mnist_train, batch_size, shuffle=True,
                            num_workers=get_dataloader_workers()),
            data.DataLoader(mnist_test, batch_size, shuffle=False,
                            num_workers=get_dataloader_workers()))

#----------------------------类工具--------------------------#

class Accumulator:  #@save
    """累加器,在我们的实例中用于存储训练损失总和、训练准确度总和、样本数,n=3
    """
    def __init__(self,n):
        self.data = [0.0]*n
    
    def add(self,*args):
        self.data = [a + float(b) for a,b in zip(self.data,args)]
    
    def reset(self):
        self.data = [0.0]*len(self.data)

    def __getitem__(self,idx):
        return self.data[idx]

class Animator:  #@save
    """在动画中绘制数据"""
    def __init__(self, xlabel=None, ylabel=None, legend=None, xlim=None,
                 ylim=None, xscale='linear', yscale='linear',
                 fmts=('-', 'm--', 'g-.', 'r:'), nrows=1, ncols=1,
                 figsize=(3.5, 2.5)):
        # 增量地绘制多条线
        if legend is None:
            legend = []
        d2l.use_svg_display()
        self.fig, self.axes = d2l.plt.subplots(nrows, ncols, figsize=figsize)
        if nrows * ncols == 1:
            self.axes = [self.axes, ]
        # 使用lambda函数捕获参数
        self.config_axes = lambda: d2l.set_axes(
            self.axes[0], xlabel, ylabel, xlim, ylim, xscale, yscale, legend)
        self.X, self.Y, self.fmts = None, None, fmts

    def add(self, x, y):
        # 向图表中添加多个数据点
        if not hasattr(y, "__len__"):
            y = [y]
        n = len(y)
        if not hasattr(x, "__len__"):
            x = [x] * n
        if not self.X:
            self.X = [[] for _ in range(n)]
        if not self.Y:
            self.Y = [[] for _ in range(n)]
        for i, (a, b) in enumerate(zip(x, y)):
            if a is not None and b is not None:
                self.X[i].append(a)
                self.Y[i].append(b)
        self.axes[0].cla()
        for x, y, fmt in zip(self.X, self.Y, self.fmts):
            self.axes[0].plot(x, y, fmt)
        self.config_axes()
        display.display(self.fig)
        display.clear_output(wait=True)

#-------------------2.初始化模型-----------------------#

num_inputs = 28*28
num_outputs = 10

#初始化权重矩阵,偏置 0
W = torch.normal(0,0.01,size=(num_inputs,num_outputs),requires_grad=True)
b = torch.zeros(num_outputs,requires_grad=True)

def model(X):
    y_hat = torch.matmul((X.reshape(-1,W.shape[0])),W)+b
    return softmax(y_hat)

def softmax(X,dim=-1):
    max_x = torch.max(X,dim=dim,keepdim=True).values
    x_exp = torch.exp(X-max_x)#torch.Size([2, 5])
    sum_x = (x_exp).sum(dim=dim,keepdim=True)#torch.Size([2, 1])
    return x_exp/sum_x # 这里应用了广播机制,(从后往前)sum_x维度被扩展到x_exp

#--------------------3.工具函数-------------------------#

def cross_entropy(y_hat,y):
    return -torch.log(y_hat[range(len(y_hat)),y])


def accuracy (y_hat,y):
    """
    计算预测正确的样本数量
    
    Args:
        predictions: 预测值张量/数组
        targets: 真实标签张量/数组
    
    Returns:
        正确预测的样本数量(整数)
    """
    if len(y_hat.shape) > 1 and y_hat.shape[1] > 1:#多样本且多类别的时候
        y_hat = torch.argmax(y_hat,dim=1) #dim=0按列比较 dim=1每行最大值的索引
    cmp = y_hat == y
    correct_sum = torch.sum(cmp)
    correct_sum.item()
    return correct_sum

def evaluate_accuracy(model,data_iter):
    """计算模型在训练集/验证集/测试集的表现
    Args:
        model (_type_): _description_
        data_iter (_type_): _description_
    """
    if isinstance(model,torch.nn.Module):
        model.eval()
    metric = Accumulator(2)#记录每个epoch的正确预测数,预测总数
    with torch.no_grad():
        for X,y in data_iter:
            metric.add(accuracy(model(X),y),y.numel())
    return metric[0] / metric[1]

#----------------4.训练函数--------------------#

def train_epoch(model,train_iter,lr):
    """训练模型一个迭代周期

    """

    if isinstance(model,torch.nn.Module):#其实这里没有继承
        model.train()
    #记录每个批次损失,训练准确度,样本数(其实这里也就是batch_size)
    metric = Accumulator(3)
    
    #loss =.....                    #1.定义损失函数(一般会有,只是这里前面已经定义了返回
                                    #loss的函数,后面直接调用了)
    # optimizer = torch.optim.SGD(model.parameters(),lr = lr)
    # 因为这里model没有继承torch.nn.Module 所以不能调用parameters()
    
    optimizer = torch.optim.SGD([W, b], lr=lr) 
                                    #2.定义优化器  

    for X, y in train_iter:
        y_hat = model(X)        
        l = cross_entropy(y_hat,y)  #3.计算loss
        
        optimizer.zero_grad()       #4.梯度清零
        l.mean().backward()         #5.反向传播计算梯度
        optimizer.step()            #6.更新参数
        
        metric.add(float(l.sum()),accuracy(y_hat,y),y.numel())
    return metric[0] / metric[2], metric[1] / metric[2]

def train(model,train_iter,test_iter,num_epochs,lr):
    """训练函数

    Args:
        model (_type_): _description_
        train_iter (_type_): _description_
        test_iter (_type_): _description_
        loss (_type_): _description_
        num_epochs (_type_): _description_
        optimizer (_type_): _description_
    """
    animator = Animator(xlabel='epoch', xlim=[1, num_epochs], ylim=[0.3, 0.9],
                        legend=['train loss', 'train acc', 'test acc'])

    
    for epoch in range(num_epochs):
        train_metrics = train_epoch(model,train_iter,lr)
        test_acc = evaluate_accuracy(model,test_iter)
        train_loss, train_acc = train_metrics
        print(f'Epoch {epoch+1}: '
              f'Train Loss={train_loss:.4f}, '
              f'Train Acc={train_acc:.4f}, '
              f'Test Acc={test_acc:.4f}')
        animator.add(epoch + 1, train_metrics + (test_acc,))

#------------------设参数,调用训练函数---------------#


num_epochs = 15
batch_size = 512
istrain = True
lr = 0.1
  
train_iter, test_iter = d2l.load_data_fashion_mnist(batch_size)

for X,y in train_iter:
    print(X.shape,y.shape)
    break

if istrain:
   train(model, train_iter, test_iter, num_epochs, lr)



def predict(model,test_iter,n=5):
    for X,y in test_iter:
        break
    trues = d2l.get_fashion_mnist_labels(y)
    preds = d2l.get_fashion_mnist_labels(model(X).argmax(axis=1))
    titles = [true +'\n'+pred for true, pred in zip(trues,preds)]
    d2l.show_images(
        X[0:n].reshape((n, 28, 28)), 1, n, titles=titles[0:n]
    )
predict(model,test_iter)    


参考李沐动手学习深度学习教程

更多推荐