模拟真实的线性模型做一个线性回归

假设有四种属性决定了一个数值,它们之间呈线性关系,即

y = ( w1*x1 + w2*x2 + w3*x3 + w4*x4 ) + b

给定真实的W和b,并以此生成一组随机的XY,利用线性回归从随机的数据中训练出接近真实值的W和b

def create_data(w, b, data_num):  # 用随机数替代真实数据
    x = torch.normal(0, 1, (data_num, len(w)))  # 500个人各自的四项数据
    y = torch.matmul(x, w) + b  # 矩阵相乘函数matmul;模拟数据间的线性关系
    noise = torch.normal(0, 0.01, y.shape) # 噪声最后要加到y上,因而它们的形状必须相同
    y += noise
    return x, y

用于生成随机数据的函数

torch.normal()的用法-CSDN博客

normal(mean, std, *, generator=None, out=None)

该函数返回从单独的正态分布中提取的随机数的张量,该正态分布的均值是mean,标准差是std。

torch.matmul(input, other, *, out=None) → Tensor

用于矩阵乘法张量乘法的函数,支持多种维度的张量运算。

  • input (Tensor) – 要相乘的第一个张量

  • other (Tensor) – 要相乘的第二个张量

利用真实的参数产生一组随机数据

num = 500
true_w = torch.tensor([8.1, 2, 2, 4])
true_b = torch.tensor(1.1)

X, Y = create_data(true_w, true_b, num)  # 初始数据
def data_provider(data, label, batchsize):    # 访问时提供一批数据的函数
    length = len(label)
    indices = list(range(length))
    #  训练时取数不能按顺序取,应随机取数
    random.shuffle(indices)

    for each in range(0, length, batchsize):
        get_indices = indices[each: each + batchsize]
        get_data = data[get_indices]
        get_label = label[get_indices]

        yield get_data, get_label  # 生成器函数,返回生成器对象,
                                    #  yield = 带存档点的return,
                                    #  函数反复执行时,第二次起,每次都从yield开始

从前面生成的随机数据中产出一批(batch)数据的函数

这里涉及到batch、epoch、iteration的概念

EpochBatch 和 Iteration

在深度学习训练中,EpochBatch 和 Iteration 是三个核心概念,它们共同决定了模型的训练方式与效率,但含义和作用各不相同。

Epoch 

一次 Epoch 表示整个训练集被模型完整地正向传播和反向传播一次,即所有样本都参与了一次训练。通常需要多个 Epoch 才能让模型逐渐收敛,但过多可能导致过拟合。

Batch 

Batch 是将训练集拆分成的小批次,每次训练只用一个批次的数据进行前向计算、损失计算和反向传播。这样做可以节省内存、提高计算效率,并引入一定的随机性,提升泛化能力。

Iteration 

一次 Iteration 指模型用一个 Batch 数据完成一次参数更新。

关系公式:Iteration 数 = 样本总数 / Batch Size × Epoch 数

例如:1000 个样本,Batch Size=100,1 个 Epoch 需要 10 次 Iteration。

注意:取数据时不能顺序取数,必须保证一批数据中的随机性

# 没有打乱的问题:
# 假设数据按类别排序:[猫,猫,猫,狗,狗,狗,车,车,车]
# 如果按顺序取batch_size=3:
# 第一批:猫,猫,猫  ← 全是猫,模型会过度拟合"猫"的特征
# 第二批:狗,狗,狗  ← 全是狗
# 这会导致训练不稳定

# 打乱后:
# 索引可能变为:[2,7,1,5,0,6,4,3,8]
# 第一批:车,车,猫  ← 混合类别,训练更稳定

这就是random.shuffle函数在做的事情,把按顺序生成的indices打乱,从而实现随机取数

for循环要做的就是每次间隔一个批次大小取被打乱的下标(indices,index的复数),于是就实现了每次取一个batch大小的随机数据用于模型的训练

data和label的概念

就像教小孩认动物:

  • data(数据) = 给小孩看的动物图片

  • label(标签) = 告诉小孩这是什么动物

  • 模型 = 小孩的大脑

  • 训练 = 教小孩认识动物的过程

所以,data = 特征/输入数据,label = 标签/目标数值

它们在回归任务和分类任务中具有不同的现实意义,但是在数学上它们是一样的

在这里,data和label指的就是输入的四种属性X(前面随机生成的),label就是每一组属性对应的某种目标数值

lr = 0.01  # 学习率

w_0 = torch.normal(0, 0.01, true_w.shape, requires_grad=True)
b_0 = torch.tensor(0.01, requires_grad=True)
print("初始的随机参数值w是:", w_0, "b是:", b_0)

epochs = 50
batchsize = 16

lr = learning rate 学习率,是梯度下降法优化参数时的步长

w0和b0是随机产生的模型初值

设定epoch和batchsize

for epoch in range(epochs):
    data_loss = 0

    for batch_x, batch_y in data_provider(X, Y, batchsize):
        pred_y = func(batch_x, w_0, b_0)
        loss = Loss(pred_y, batch_y)
        loss.backward()
        sgd([w_0, b_0], lr)
        data_loss += loss

    print("epoch " + str(epoch+1) + "的损失为:", data_loss)

这里就是模型训练的过程了,epoch决定了训练次数为50次

涉及到的几个自定义函数:

def func(x, w, b):
    pred_y = torch.matmul(x, w) + b
    return pred_y


def Loss(pred_y, y):  # 差值损失函数
    return torch.sum(abs(pred_y - y))/len(y)


def sgd(paras, lr):  # 随机梯度下降  Gradient Descent
    with torch.no_grad():  # 属于这一句代码的部分不计算梯度
        for para in paras:
            para -= para.grad * lr  # lr = learning rate 学习率
            para.grad.zero_()

func() 是用于计算 预测值 ŷ 的,决定了所使用的模型是线性模型

Loss() 是用于计算模型损失的损失函数,用的是很简单的作差,即令 目标数值的预测值 直接减去 真实值 从而得到模型的误差

sgd() 是随机梯度下降函数,sgd = Stochastic Gradient Descent       Stochastic = 随机的

(修饰变量的时候用 random。修饰过程的时候用 stochastic。除这一区别外,两者是同义词。
A variable is random. A process is stochastic. Apart from this difference, the two words are synonyms. -- Riccardo Alestra)

要理解sgd这个函数,首先要知道pytorch自动微分和计算图

计算图

【深度学习】Pytorch教程(十三):PyTorch数据结构:5、张量的梯度计算:变量(Variable)、自动微分、计算图及其可视化-腾讯云开发者社区-腾讯云

计算图是一种用来表示数学运算过程的图形化结构,它将数学计算表达为节点和边的关系,提供了一种直观的方式来理解和推导复杂的数学运算过程。

在深度学习中,计算图帮助我们理解模型的训练过程,直观地把握损失函数对模型参数的影响,同时为反向传播算法提供了理论基础。

现代深度学习框架如 PyTorch 和 TensorFlow 都是基于计算图的理论基础构建出来的。

计算图包括两种节点:计算节点(Compute Node)和数据节点(Data Node)。

  • 数据节点:表示输入数据、参数或中间变量,在计算图中通常用圆形结点表示。数据节点始终是叶节点,它们没有任何输入,仅表示数据。
  • 计算节点:表示数学运算过程,它将输入的数据节点进行数学运算后输出结果。在计算图中通常用方形结点表示。计算节点可以有多个输入和一个输出。反向传播算法中的梯度计算正是通过计算节点来实现的。

一个完整的计算图可以分为正向传播和反向传播两个阶段:

  • 正向传播(Forward Propagation):输入数据经过计算节点逐层传播,最终得到输出结果
  • 反向传播(Backward Propagation):首先根据损失函数计算输出结果与真实标签之间的误差,然后利用链式法则,逐个计算每个计算节点对应的输入的梯度,最终得到参数的梯度信息

对于上面的那张图,函数h1~h6体现了正向传播过程,图中各个偏导数则体现了反向传播过程。

通过链式求导法则就可以得到输出(loss损失函数)对于参数(w、b等)的偏导数,从而得到其梯度

基于上述理论,pytorch的自动微分机制就好理解了

Pytorch的自动微分机制

自动微分基础 — PyTorch Tutorials 2.5.0+cu124 documentation

Autograd 的强大之处在于它在运行时动态地 跟踪你的计算这意味着如果你的模型有决策分支或长度在运行时才知道的循环,计算仍然会被正确跟踪,你会得到正确的梯度来驱动学习。

结合你的模型是用 Python 构建的事实,这比 依赖于对更加严格结构化的模型 进行静态分析 来计算梯度的框架 提供了更大的灵活性。

也就是说,在运用pytorch中的张量进行计算时,pytorch会自动构造一个计算图,记录你对某个张量所做的每次操作

# 1. 创建叶子节点(输入)
x = torch.tensor([2.0], requires_grad=True)  # 叶子节点1
w = torch.tensor([3.0], requires_grad=True)  # 叶子节点2
b = torch.tensor([1.0], requires_grad=True)  # 叶子节点3

# 2. 执行计算(构建计算图)
y_pred = x * w + b  # 前向传播,构建计算图

"""
计算图构建:
叶子:  x(2)  w(3)  b(1)
         \   /      |
          乘法      |
           v       |
         x*w(6)   |
            \    /
             加法
              v
           输出(7)
"""
loss = (y_pred - 5.0) ** 2  # 假设目标值是5.0

通过前向传播构造起计算图后,就可以通过.backward()进行反向传播,计算loss关于每个参数的偏导数

# 反向传播过程
loss.backward()  # 从loss开始反向计算所有梯度

"""
反向传播路径:
loss → (y_pred-5)² → y_pred → x*w+b → [x,w,b]

链式法则计算:
∂loss/∂x = ∂loss/∂y_pred * ∂y_pred/∂x
         = 2*(y_pred-5) * w
         = 2*(7-5)*3 = 12

PyTorch自动完成:
x.grad = 12
w.grad = 8  # 2*(7-5)*2
b.grad = 4  # 2*(7-5)*1
"""

print(f"梯度: x={x.grad}, w={w.grad}, b={b.grad}")

计算出的偏导数会保存在张量的.grad属性中,如loss对w的偏导数就会保存在w.grad中

于是就知道了为什么要加一句 with torch.no_grad() 了。如果不在更新参数时禁用梯度计算,那么针对参数更新所做的减法操作也会被记录到计算图中,导致计算图无限延长,这显然不是我们想要的。禁用时会停止记录操作,防止计算图增大。

中间的para -= para.grad * lr就是梯度下降

para.grad.zero_()是清空para中记录的梯度值。原因在于每次进行反向传播所计算的梯度值并不是直接被新值替换,而是会进行累加,累加是pytorch的特性,估计在其他更加深入的应用中会用到,但是这里不能用,所以必须每次计算后都清零一次,防止其累加。

完整代码:

import torch
import matplotlib.pyplot as plt
import random


def create_data(w, b, data_num):  # 用随机数替代真实数据
    x = torch.normal(0, 1, (data_num, len(w)))  # 500个人各自的四项数据
    y = torch.matmul(x, w) + b  # 矩阵相乘函数matmul;模拟数据间的线性关系
    noise = torch.normal(0, 0.01, y.shape) # 噪声最后要加到y上,因而它们的形状必须相同
    y += noise
    return x, y


def data_provider(data, label, batchsize):    # 访问时提供一批数据的函数
    length = len(label)
    indices = list(range(length))
    #  训练时取数不能按顺序取,应随机取数
    random.shuffle(indices)

    for each in range(0, length, batchsize):
        get_indices = indices[each: each + batchsize]
        get_data = data[get_indices]
        get_label = label[get_indices]

        yield get_data, get_label  # 生成器函数,返回生成器对象,
                                    #  yield = 带存档点的return,
                                    #  函数反复执行时,第二次起,每次都从yield开始


def func(x, w, b):
    pred_y = torch.matmul(x, w) + b
    return pred_y


def Loss(pred_y, y):  # 差值损失函数
    return torch.sum(abs(pred_y - y))/len(y)


def sgd(paras, lr):  # 随机梯度下降  Gradient Descent
    with torch.no_grad():  # 属于这一句代码的部分不计算梯度
        for para in paras:
            para -= para.grad * lr  # lr = learning rate 学习率
            para.grad.zero_()


num = 500
true_w = torch.tensor([8.1, 2, 2, 4])
true_b = torch.tensor(1.1)

X, Y = create_data(true_w, true_b, num)  # 初始数据

lr = 0.01  # 学习率

w_0 = torch.normal(0, 0.01, true_w.shape, requires_grad=True)
b_0 = torch.tensor(0.01, requires_grad=True)
print("初始的随机参数值w是:", w_0, "b是:", b_0)

min_ = 300
epochs = 50
w_final, b_final = w_0, b_0
epoch_best = 0

batchsize = 16

for epoch in range(epochs):
    data_loss = 0

    for batch_x, batch_y in data_provider(X, Y, batchsize):
        pred_y = func(batch_x, w_0, b_0)
        loss = Loss(pred_y, batch_y)
        loss.backward()
        sgd([w_0, b_0], lr)
        data_loss += loss

    print("epoch " + str(epoch+1) + "的损失为:", data_loss)
    if data_loss < min_:
        epoch_best = epoch
        min_ = data_loss
        w_final = w_0
        b_final = b_0

print("真实的参数值w是:", true_w, "b是:", true_b)
print("训练得到的参数值w是:", w_final, "b是:", b_final)
print("损失最低epoch是:", epoch_best+1)

idxs = 4
for idx in range(idxs):
    plt.plot(X[:, idx].detach().numpy(), X[:, idx]*w_0[idx].detach().numpy() + b_0.detach().numpy())
    plt.scatter(X[:, idx], Y, 1)
    plt.show()

训练结果

更多推荐