反向传播

在深度学习中为什么要“反向”？

因为深度学习模型通常是一个由许多层组成的复合函数。比如：输出 = Layer3(Layer2(Layer1(输入)))。

根据微积分中的链式法则，要计算最终损失相对于最底层（Layer1）参数的梯度，我们需要从最后层开始，逐层反向相乘。

“反向”的路径，恰恰是利用了前向传播中计算并保存的中间结果，这使得计算变得非常高效。如果不用反向传播，而是为每个参数单独计算梯度，计算量将是天文数字。

链式法则

链式法则是反向传播的引擎。它指出，对于复合函数 y = f(g(x))，y 关于 x 的导数可以这样计算：
dy/dx = (dy/du) * (du/dx)，其中 u = g(x)。

在神经网络中，损失函数 L 是最终输出 a 的函数，而 a 又是上一层输出 z 的函数，z 又是权重 w 和上一层输出的函数... 如此反复。

示例

我们用一个极简的网络来演示整个过程。这个网络只有：

• 输入层：一个神经元 x（值为 1.0）

• 隐藏层：一个神经元（权重 w = 0.5，偏置 b = 0.0，激活函数为Sigmoid）

• 输出层：一个神经元 y_pred

• 目标值：y_true = 0.8

• 损失函数：均方误差 L = (y_true - y_pred)^2

网络结构：x → (z = w*x + b) → (a = σ(z)) → y_pred → L

步骤 1：前向传播

1. 计算加权输入 z：
   z = w * x + b = 0.5 * 1.0 + 0.0 = 0.5

2. 计算激活输出 a（即预测值 y_pred）：
   a = σ(z) = 1 / (1 + e^(-z)) = 1 / (1 + e^(-0.5)) ≈ 0.6225
   （Sigmoid函数将输入映射到0-1之间）

3. 计算损失 L：
   L = (y_true - y_pred)^2 = (0.8 - 0.6225)^2 ≈ (0.1775)^2 ≈ 0.0315

至此，前向传播结束，我们知道当前网络的预测值（0.6225）离真实值（0.8）还很远，损失是0.0315。

步骤 2：反向传播（计算梯度）

我们的目标是计算损失 L 关于权重 w 的梯度 ∂L/∂w，以便更新 w。

根据链式法则，w 影响了 z，z 影响了 a，a 影响了 L。所以：
∂L/∂w = (∂L/∂a) * (∂a/∂z) * (∂z/∂w)

现在我们一步步计算：

1. 计算 ∂L/∂a（损失对输出的梯度）：
   L = (y_true - a)^2
∂L/∂a = 2 * (y_true - a) * (-1) = -2 * (0.8 - 0.6225) ≈ -2 * 0.1775 = -0.355

2. 计算 ∂a/∂z（激活函数对输入的梯度）：
   a = σ(z) = 1 / (1 + e^(-z))
   Sigmoid函数的导数有一个很好的性质：σ‘(z) = σ(z) * (1 - σ(z)) = a * (1 - a)
∂a/∂z = a * (1 - a) = 0.6225 * (1 - 0.6225) ≈ 0.6225 * 0.3775 ≈ 0.235

3. 计算 ∂z/∂w（线性变换对权重的梯度）：
   z = w * x + b
∂z/∂w = x = 1.0

4. 组合起来，得到 ∂L/∂w：
   ∂L/∂w = (∂L/∂a) * (∂a/∂z) * (∂z/∂w) = (-0.355) * (0.235) * (1.0) ≈ -0.0834

这个结果 ∂L/∂w ≈ -0.0834 就是我们的梯度。它的负号意味着：增加 w 的值将会导致损失 L 减小。

步骤 3：参数更新（梯度下降）

现在我们使用这个梯度来更新权重 w。我们选择一个学习率 η（比如 η = 1，为了演示效果明显，实际中通常更小）。

更新公式：w_new = w_old - η * (∂L/∂w)
w_new = 0.5 - 1 * (-0.0834) = 0.5 + 0.0834 = 0.5834

一次迭代完成！

上面的例子可以用下面的这个结构来总结：

输入 -> 全连接层 -> sigmoid  ->均方差损失 -> loss值

优化器-Adam

什么是Adam?

中文名称为-自适应矩估计

简单来说，Adam 结合了另外两种优化器的核心思想：

1. 动量法：像重球滚下山坡一样，它不仅考虑当前的梯度方向，还积累之前的梯度，从而在相关方向上加速并减少震荡。

2. RMSProp：为每个参数自适应地调整学习率。对于不频繁更新的参数，给予更大的更新；对于频繁更新的参数，给予更小的更新。

Adam 将这两者结合起来，通过计算梯度的一阶矩（均值）和二阶矩（未中心化的方差）的指数移动平均值，来为每个参数提供自适应的学习率。

核心算法步骤

我们通常以向量形式描述，其中 θt 表示在时间步 t 时的模型参数，gt表示在 t 时刻的梯度。

1. 计算梯度：计算当前小批量数据的梯度 gt。

2. 更新一阶矩估计：计算梯度的指数移动平均值 mt​（动量）。

    

3. 更新二阶矩估计：计算梯度平方的指数移动平均值 vt（自适应学习率）。

    

4. 偏差校正：由于 mt​ 和 vt​ 初始化为 0，在训练初期它们会偏向于 0。因此需要进行偏差校正，使其更接近无偏估计。

       

         

     5.更新参数：使用校正后的矩估计来更新模型参数。

            

参数详解

1. 学习率 - lr（公式中的 α）

• 作用：这是整个优化过程中最重要的超参数。它决定了参数沿着梯度反方向更新的步长。

• 影响：

  • 过大：可能导致模型在最优解附近来回震荡，甚至发散，无法收敛。

  • 过小：导致收敛速度过慢，训练时间大大延长，也可能陷入局部最优点。

• 默认值与经验：Adam 对学习率不那么敏感，因为它有自适应的机制。通常默认值 1e-3 是一个不错的起点。在实践中，常用的学习率是 3e-4, 1e-3, 5e-4 等。

2. 一阶矩衰减率 - beta1（公式中的 β1）

• 作用：控制一阶矩（动量） mtmt​ 的指数衰减率。它决定了过去梯度对当前方向的影响程度。

• 物理意义：可以理解为“摩擦系数”或“惯性”。

  • beta1 越大（如 0.99），历史梯度的权重越大，优化过程惯性越大，能更平滑地穿过狭窄的峡谷或逃离局部最小值，但可能对变化反应迟钝。

  • beta1 越小，则更依赖于当前批次的梯度，震荡可能更大。

• 默认值与经验：通常设置为一个接近 1 的值，默认值 0.9 在绝大多数情况下都工作得很好。

3. 二阶矩衰减率 - beta2（公式中的 β2​）

• 作用：控制二阶矩（梯度平方） vtvt​ 的指数衰减率。它决定了历史梯度平方对当前自适应学习率的影响程度。

• 物理意义：它使得对于频繁出现大梯度的参数，其有效学习率会变小；对于出现小梯度或稀疏梯度的参数，其有效学习率会变大。这有助于处理稀疏梯度问题。

• 影响：

  • beta2 越大，历史梯度平方的窗口越长，自适应学习率变化越平滑。

  • beta2 越小，自适应学习率只关注近期梯度，变化更剧烈。

• 默认值与经验：通常设置为一个非常接近 1 的值，默认值 0.999 是经过大量实验验证的有效值。

4. epsilon - eps

• 作用：一个非常小的常数，添加到分母中，主要目的是为了防止除以零的数值不稳定问题。

• 影响：

  • 理论上，它的值不重要，只要是一个很小的数即可。

  • 但在实践中，它的选择有时会对结果产生微妙影响，尤其是在训练的初期阶段，当 还很小时。

  • 设置过大，会过度平滑更新，影响收敛。

• 默认值与经验：框架不同，默认值略有差异。PyTorch 中默认是 1e-8，TensorFlow (Keras) 中默认是 1e-7。通常不需要修改。

5. 权重衰减 - weight_decay

• 注意：这是一个在 PyTorch 的 torch.optim.Adam 等实现中常见的参数，但它不是原始 Adam 论文的一部分。

• 作用：实现 L2 正则化，通过在损失函数中添加一个惩罚项 来防止模型过拟合。它通过在梯度上直接加上 λθ 来实现。

• 与原始 Adam 的关系：在 PyTorch 中，Adam 的 weight_decay 参数实现的是 AdamW 优化器中的方式，这是一种更正确的将权重衰减与 Adam 结合的方法，而不是简单地在损失函数里加 L2 正则化。

• 默认值与经验：默认值为 0（即不使用）。如果模型出现过拟合，可以尝试设置一个小的值，如 1e-4 或 1e-5。

补充：

解析Adam与SGD两大核心优化器的区别以及如何选择

核心区别

特性维度	SGD（随机梯度下降）	Adam（自适应矩估计）
核心哲学	简单、精确、可控。相信“慢工出细活”，通过精细调整稳步逼近最优解。	高效、自适应、稳健。像智能导航，动态调整每一步的速度和方向，力求快速收敛。
更新机制	沿当前批次梯度的负方向，以固定学习率更新：θ_new = θ_old - η * g_t	综合历史梯度（动量）和梯度平方（自适应学习率），为每个参数单独计算更新步长。
关键组件	学习率(η)，可附加动量(β)。	学习率(α)、一阶矩衰减(β₁)、二阶矩衰减(β₂)、极小值(ε)。
收敛速度	通常较慢，尤其在高维、稀疏或病态曲面的问题上。	初期收敛极快，能快速显著降低损失，是其主要优势之一。
泛化性能	经精心调优（如学习率衰减） 后，常能达到更优的最终测试精度。	收敛速度快，但有时在测试集上的最终精度可能略逊于调优后的SGD。
调参复杂度	高。需仔细调整初始学习率、衰减策略和动量。	低。对默认超参数(如α=0.001, β₁=0.9, β₂=0.999)不敏感，通常开箱即用。
内存占用	低（仅需存储梯度）。	较高（需为每个参数存储一阶、二阶矩估计）。

简单比喻：SGD像一位严谨的登山者，每一步都深思熟虑；Adam则像装备了智能助力系统和地形雷达的全地形车，起步迅猛，能适应复杂路面。

使用场景以及选择策略

1. 何时优先使用 Adam？

Adam因其稳健性和速度，是绝大多数情况的首选和默认基准，尤其适合以下场景：

• 深度学习入门与实践：项目初期、原型验证阶段，希望快速看到模型初步效果。

• 处理稀疏或噪声较大的数据：如自然语言处理、推荐系统。其自适应学习率对不频繁特征更友好。

• 模型复杂、超参数预算有限：不想或没有足够资源进行精细的超参数搜索。

• 训练深度 Transformer、GAN 等现代架构：这些模型通常默认使用Adam/AdamW（Adam的权重衰减修正版）。

2. 何时应考虑（带动量的）SGD？

当追求极致性能，并有充足调优资源时，SGD可能带来惊喜：

• 追求：在图像分类（如ImageNet）、语音识别等经典任务上，经充分调优的SGD+Momentum常被报道能达到比Adam 稍高的最终精度（可能高出1-2个百分点）。

• 理论分析需求：由于更新规则简单，SGD的理论性质（如收敛性证明）更清晰，便于学术分析。

• 极端注重模型轻量化：在内存极度受限的边缘设备上，SGD的内存优势会被放大。

#coding:utf8

import torch
import torch.nn as nn
import numpy as np
import copy

"""
基于pytorch的网络编写
手动实现梯度计算和反向传播
加入激活函数
"""

class TorchModel(nn.Module):
    def __init__(self, hidden_size):
        super(TorchModel, self).__init__()
        self.layer = nn.Linear(hidden_size, hidden_size, bias=False) #w = hidden_size * hidden_size  wx+b -> wx
        self.activation = torch.sigmoid
        self.loss = nn.functional.mse_loss  #loss采用均方差损失

    #当输入真实标签，返回loss值；无真实标签，返回预测值
    def forward(self, x, y=None):
        y_pred = self.layer(x)
        y_pred = self.activation(y_pred)
        if y is not None:
            return self.loss(y_pred, y)
        else:
            return y_pred


#自定义模型，接受一个参数矩阵作为入参
class DiyModel:
    def __init__(self, weight):
        self.weight = weight

    def forward(self, x, y=None):
        x = np.dot(x, self.weight.T)
        y_pred = self.diy_sigmoid(x)
        if y is not None:
            return self.diy_mse_loss(y_pred, y)
        else:
            return y_pred

    #sigmoid
    def diy_sigmoid(self, x):
        return 1 / (1 + np.exp(-x))

    #手动实现mse，均方差loss
    def diy_mse_loss(self, y_pred, y_true):
        return np.sum(np.square(y_pred - y_true)) / len(y_pred)

    #手动实现梯度计算
    def calculate_grad(self, y_pred, y_true, x):
        #前向过程
        # wx = np.dot(self.weight, x)
        # sigmoid_wx = self.diy_sigmoid(wx)
        # loss = self.diy_mse_loss(sigmoid_wx, y_true)
        #反向过程
        # 均方差函数 (y_pred - y_true) ^ 2 / n 的导数 = 2 * (y_pred - y_true) / n , 结果为2维向量
        grad_mse = 2/len(x) * (y_pred - y_true)
        # sigmoid函数 y = 1/(1+e^(-x)) 的导数 = y * (1 - y), 结果为2维向量
        grad_sigmoid = y_pred * (1 - y_pred)
        # wx矩阵运算，见ppt拆解, wx = [w11*x0 + w21*x1, w12*x0 + w22*x1]
        #导数链式相乘
        grad_w11 = grad_mse[0] * grad_sigmoid[0] * x[0]
        grad_w12 = grad_mse[1] * grad_sigmoid[1] * x[0]
        grad_w21 = grad_mse[0] * grad_sigmoid[0] * x[1]
        grad_w22 = grad_mse[1] * grad_sigmoid[1] * x[1]
        grad = np.array([[grad_w11, grad_w12],
                         [grad_w21, grad_w22]])
        #由于pytorch存储做了转置，输出时也做转置处理
        return grad.T

#梯度更新
def diy_sgd(grad, weight, learning_rate):
    return weight - learning_rate * grad

#adam梯度更新
def diy_adam(grad, weight):
    #参数应当放在外面，此处为保持后方代码整洁简单实现一步
    alpha = 1e-3  #学习率
    beta1 = 0.9   #超参数
    beta2 = 0.999 #超参数
    eps = 1e-8    #超参数
    t = 0         #初始化
    mt = 0        #初始化
    vt = 0        #初始化
    #开始计算
    t = t + 1
    gt = grad
    mt = beta1 * mt + (1 - beta1) * gt
    vt = beta2 * vt + (1 - beta2) * gt ** 2
    mth = mt / (1 - beta1 ** t)
    vth = vt / (1 - beta2 ** t)
    weight = weight - (alpha * mth/ (np.sqrt(vth) + eps))
    return weight

x = np.array([-0.5, 0.1])  #输入
y = np.array([0.1, 0.2])  #预期输出

#torch实验
torch_model = TorchModel(2)
torch_model_w = torch_model.state_dict()["layer.weight"]
print(torch_model_w, "初始化权重")
numpy_model_w = copy.deepcopy(torch_model_w.numpy())
#numpy array -> torch tensor, unsqueeze的目的是增加一个batchsize维度
torch_x = torch.from_numpy(x).float().unsqueeze(0) 
torch_y = torch.from_numpy(y).float().unsqueeze(0)
#torch的前向计算过程，得到loss
torch_loss = torch_model(torch_x, torch_y)
print("torch模型计算loss：", torch_loss)
# #手动实现loss计算
diy_model = DiyModel(numpy_model_w)
diy_loss = diy_model.forward(x, y)
print("diy模型计算loss：", diy_loss)




# # #设定优化器
learning_rate = 0.1
# optimizer = torch.optim.SGD(torch_model.parameters(), lr=learning_rate)
optimizer = torch.optim.Adam(torch_model.parameters())
# optimizer.zero_grad()
# #
# # #pytorch的反向传播操作
torch_loss.backward()
print(torch_model.layer.weight.grad, "torch 计算梯度")  #查看某层权重的梯度

# # #手动实现反向传播
grad = diy_model.calculate_grad(diy_model.forward(x), y, x)
print(grad, "diy 计算梯度")
# #
# #torch梯度更新
optimizer.step()
# # #查看更新后权重
update_torch_model_w = torch_model.state_dict()["layer.weight"]
print(update_torch_model_w, "torch更新后权重")
# #
# # #手动梯度更新
# diy_update_w = diy_sgd(grad, numpy_model_w, learning_rate)
diy_update_w = diy_adam(grad, numpy_model_w)
print(diy_update_w, "diy更新权重")

神经网络文本处理

总体步骤

step1 字符数值化

step2 矩阵转化为向量

step3 向量到数值

step4 数值归一化

Embedding层

它是一个查找和/或学习的过程，目标是将高维，稀疏、离散符号转化为低维、密集、连续的向量。

最直接的表现形式就是查找表或者字典。

第一步：建立词汇表

首先，我们需要一个词汇表。例如，我们有一个包含10，000个单词的词汇表。我们给每个单词分配一个唯一的整数索引。

• “king” -> 1

• “queen” -> 2

• “man” -> 3

• “woman” -> 4

• ...

• “apple” -> 9999

第二步：定义 Embedding 矩阵
我们创建一个可学习的矩阵，我们称之为 Embedding 矩阵 E。这个矩阵的大小是 [词汇表大小, 嵌入维度]。

• 词汇表大小：10，000

• 嵌入维度：我们设定的一个超参数，比如 300。这意味着每个单词将被表示成一个长度为300的向量。

• 所以，矩阵 E 的形状是 [10000, 300]。这个矩阵的每一行就对应一个单词的向量表示。

第三步：转换过程（查找）
当模型遇到一个单词，比如 “queen”，其索引是2。

这个过程在技术上称为“索引查找”。 输入是一个整数，输出是一个向量。

1. 模型将单词 “queen” 转换为其索引 2。

2. 然后，它去 Embedding 矩阵 E 中查找第2行。

3. 取出这一行的300个数字，这个长度为300的向量就是 “queen” 的 Embedding。

代码示例

#coding:utf8
import torch
import torch.nn as nn

'''
embedding层的处理
'''

num_embeddings = 8  #通常对于nlp任务，此参数为字符集字符总数
embedding_dim = 5   #每个字符向量化后的向量维度
embedding_layer = nn.Embedding(num_embeddings, embedding_dim, padding_idx=0)
print("随机初始化权重")
print(embedding_layer.weight)
print("################")

#构造字符表
vocab = {
    "[pad]" : 0,
    "你" : 1,
    "好" : 2,
    "中" : 3,
    "国" : 4,
    "欢" : 5,
    "迎" : 6,
    "[unk]":7
}


#中国欢迎你 -> 中 国 欢 迎 你 -> [3,4,5,6,1] -> embedding_layer([3,4,5,6,1]) -> 5*5 矩阵
#你好中国 -> 你 好 中 国 -> [1,2,3,4] -> embedding_layer([1,2,3,4]) -> 4*5 矩阵



# 为了让不同长度的训练样本能够放在同一个batch中，需要将所有样本补齐或截断到相同长度
# 限制最大长度为5
# padding 补齐
# [1,2,3,0,0]
# [1,2,3,4,0]
# [1,2,3,4,5]
# 截断
# [1,2,3,4,5,6,7] -> [1,2,3,4,5]

def str_to_sequence(string, vocab):
    seq = [vocab.get(s, vocab["[unk]"]) for s in string][:5]
    if len(seq) < 5:
        seq += [vocab["[pad]"]] * (5 - len(seq))
    return seq

string1 = "abcd"
string2 = "ddcc"
string3 = "feda"

sequence1 = str_to_sequence(string1, vocab)
sequence2 = str_to_sequence(string2, vocab)
sequence3 = str_to_sequence(string3, vocab)

print(sequence1)
print(sequence2)
print(sequence3)
print("################")
x = torch.LongTensor([sequence1, sequence2, sequence3])
embedding_out = embedding_layer(x)
print(embedding_out)

池化层

平均池化

输入特征图 (4x4):
[[ 1, 3, 2, 4 ],
 [ 8, 6, 7, 5 ],
 [ 4, 2, 1, 3 ],
 [ 9, 7, 5, 6 ]]

第一步 放置池化窗口

池化层有一个预定义的窗口大小（例如2x2）和步长。它从输入特征图的左上角开始，不重叠地（当步长=窗口大小时）滑动这个窗口。

• 第一个窗口覆盖 [ [1,3], [8,6] ]

• 第二个窗口覆盖 [ [2,4], [7,5] ]

• 第三个窗口覆盖 [ [4,2], [9,7] ]

• 第四个窗口覆盖 [ [1,3], [5,6] ]

第二步 应用池化函数

平均池化：取窗口内所有值的平均值。

• 窗口1 (1+3+8+6)/4 = 4.5

• 窗口2 (2+4+7+5)/4 = 4.5

• 窗口3 (4+2+9+7)/4 = 5.5

• 窗口4 (1+3+5+6)/4 = 3.75

第三步 生成输出特征图

[[ 4.5, 4.5 ],
 [ 5.5, 3.75]]

以上也可以概括为

1. 滑动窗口：使用一个固定大小和步长的窗口在输入特征图上滑动。

2. 汇总计算：对每个窗口内的数值应用一个汇总函数（最常用的是取最大值或取平均值）。

3. 输出降维：将每个窗口的计算结果组合成一个新的、空间尺寸更小的输出特征图。

池化代码

#coding:utf8
import torch
import torch.nn as nn

'''
pooling层的处理
'''

#pooling操作默认对于输入张量的最后一维进行
#入参5，代表把五维池化为一维
layer = nn.AvgPool1d(4)  
#随机生成一个维度为3x4x5的张量
#可以想象成3条,文本长度为4,向量长度为5的样本
x = torch.rand([3, 4, 5])
print(x)
print(x.shape)
x = x.transpose(1,2)
print(x.shape, "交换后")
#经过pooling层
y = layer(x)
print(y)
print(y.shape)
#squeeze方法去掉值为1的维度
y = y.squeeze()
print(y)
print(y.shape)

RNN 循环神经网络

主要思想是：将整个序列划分成多个时间步，将每一个时间步的信息依次输入模型，同时将模型输出的结果传给下一步

公式如下：

公式中每个字母的含义：

1.

• 含义：在时间步 t 的隐藏状态向量

• 解释：

  • 这是RNN的"记忆"或"状态"，包含了从序列开始到当前时间步 tt 的所有相关信息

  • 它有两个重要作用：

    • 作为当前时间步的特征表示，用于计算输出

    • 作为"记忆"传递给下一个时间步 t+1

• 维度：通常是 (hidden_size, 1)

2.

• 含义：在时间步 t−1 的隐藏状态向量

• 解释：

  • 这是前一个时间步的隐藏状态，包含了之前所有时间步的信息

  • 正是这个连接使得RNN具有"循环"特性，能够处理序列数据

• 维度：(hidden_size, 1)

3.

• 含义：在时间步 tt 的输入向量

• 解释：

  • 这是网络在当前时刻接收到的外部输入数据

  • 例子：

    • 在文本处理中：可能是一个词的词向量

    • 在时间序列预测中：可能是t时刻的传感器读数

    • 在语音识别中：可能是一个音频帧的特征

• 维度：(input_size, 1)

4.W

• 含义：隐藏状态到隐藏状态的权重矩阵

• 解释：

  • 这个矩阵控制过去的记忆 对当前新状态  的影响程度

  • 它是RNN中最关键的参数，决定了网络如何保留和更新记忆

  • 通过训练这个矩阵，RNN学习序列中的模式和长期依赖关系

• 维度：(hidden_size, hidden_size)

5.U

• 含义：输入到隐藏状态的权重矩阵

• 解释：

  • 这个矩阵控制当前输入 如何影响新的隐藏状态

  • 它负责将外部输入信息整合到网络的记忆中

• 维度：(hidden_size, input_size)

6.b

• 含义：偏置向量

• 解释：

  • 为隐藏状态的计算增加一个固定的偏移量

  • 让模型有更强的拟合能力，即使所有输入都为0时也能产生非零输出

• 维度：(hidden_size, 1)

7.tanh

• 含义：双曲正切激活函数

• 解释：

  • 作用：引入非线性变换，使网络能够学习复杂模式

  • 特性：输出范围在 (-1, 1) 之间，将数据压缩到有界范围内

  • 为什么用tanh：

    • 有助于稳定梯度流动（相比sigmoid）

    • 以0为中心，通常训练效果更好

    • 但RNN仍然面临梯度消失/爆炸问题

计算过程的直观理解

这个公式描述了一个信息融合的过程：

1. 处理过去信息：W→ 将过去的记忆进行变换

2. 处理当前输入： → 将新的输入信息进行变换

3. 融合信息： → 将过去和现在的信息结合，加上偏置

4. 非线性激活：tanh⁡(⋯ )→ 产生新的、规范化的隐藏状态

比喻理解

• 你对前面章节的理解（记忆）

• ：当前正在读的这一页内容（新输入）

• W：你如何根据已有理解来解读新内容（记忆的重要性）

• U：你如何直接理解当前页面的文字（新信息的重要性）

• ：读完当前页后，你对整本书到此为止的新理解

RNN的缺点（梯度消失和梯度爆炸）：

1. 矩阵中有非常小的值，经过矩阵相乘N次之后，梯度值快速的以指数形式收缩，较远的时刻梯度变为0
2. 如果矩阵的值非常大，就会出现梯度爆炸

pytorch的RNN和手动实现RNN

#coding:utf8

import torch
import torch.nn as nn
import numpy as np


"""
手动实现简单的神经网络
使用pytorch实现RNN
手动实现RNN
对比
"""

class TorchRNN(nn.Module):
    def __init__(self, input_size, hidden_size):
        super(TorchRNN, self).__init__()
        self.layer = nn.RNN(input_size, hidden_size, bias=False, batch_first=True)

    def forward(self, x):
        return self.layer(x)

#自定义RNN模型
class DiyModel:
    def __init__(self, w_ih, w_hh, hidden_size):
        self.w_ih = w_ih
        self.w_hh = w_hh
        self.hidden_size = hidden_size

    def forward(self, x):
        ht = np.zeros((self.hidden_size))
        output = []
        for xt in x:
            ux = np.dot(self.w_ih, xt)
            wh = np.dot(self.w_hh, ht)
            ht_next = np.tanh(ux + wh)
            output.append(ht_next)
            ht = ht_next
        return np.array(output), ht


x = np.array([[1, 2, 3],
              [3, 4, 5],
              [5, 6, 7]])  #网络输入

#torch实验
hidden_size = 4
torch_model = TorchRNN(3, hidden_size)

# print(torch_model.state_dict())
w_ih = torch_model.state_dict()["layer.weight_ih_l0"]
w_hh = torch_model.state_dict()["layer.weight_hh_l0"]
print(w_ih, w_ih.shape)
print(w_hh, w_hh.shape)
#
torch_x = torch.FloatTensor([x])
output, h = torch_model.forward(torch_x)
print(h)
print(output.detach().numpy(), "torch模型预测结果")
print(h.detach().numpy(), "torch模型预测隐含层结果")
print("---------------")
diy_model = DiyModel(w_ih, w_hh, hidden_size)
output, h = diy_model.forward(x)
print(output, "diy模型预测结果")
print(h, "diy模型预测隐含层结果")

下面是一个demo使用RNN来实现多分类问题

# coding:utf8

import torch
import torch.nn as nn
import numpy as np
import random
import json

"""

基于pytorch的网络编写
实现一个网络完成一个简单nlp任务
判断文本中五个汉字的位置类别

"""


class TorchModel(nn.Module):
    def __init__(self, vector_dim, sentence_length, vocab):
        super(TorchModel, self).__init__()
        self.embedding = nn.Embedding(len(vocab), vector_dim, padding_idx=0)  # embedding层
        self.rnn = nn.RNN(vector_dim, 128, batch_first=True, nonlinearity='tanh')
        self.classify = nn.Linear(128, 5)  # 线性层
        self.loss = nn.CrossEntropyLoss()

    # 当输入真实标签，返回loss值；无真实标签，返回预测值
    def forward(self, x, y=None):
        x = self.embedding(x)  # (batch_size, sen_len) -> (batch_size, sen_len, vector_dim)
        rnn_out, rnn_hidden = self.rnn(x)
        last_hidden = rnn_hidden.squeeze()
        x = self.classify(last_hidden)
        if y is not None:
            return self.loss(x, y)  # 预测值和真实值计算损失
        else:
            return torch.softmax(x, dim=1)


def build_vocab():
    chars = "你我他她好是非对错来去上下左右前后中间旁边内外大小多少长短高低快慢远近新旧老幼男女父母子师生朋友同学事领导客户板员工"  # 字符集
    vocab = {"pad": 0}
    for index, char in enumerate(chars):
        vocab[char] = index + 1  # 每个字对应一个序号
    vocab['unk'] = len(vocab)
    return vocab


# 随机生成一个样本
# 从所有字中选取sentence_length个字
def build_sample(vocab, sentence_length):
    x = [random.choice(list(vocab.keys())) for _ in range(sentence_length)]

    # 定义五个类别对应的目标汉字
    target_chars = ["我", "你", "他", "是", "好"]
    y = 0
    for i, char in enumerate(target_chars):
        if char in x:
            position = x.index(char)
            y = position
            break
    x = [vocab.get(word, vocab['unk']) for word in x]  # 将字转换成序号，为了做embedding
    return x, y


# 建立数据集
# 输入需要的样本数量。需要多少生成多少
def build_dataset(sample_length, vocab, sentence_length):
    dataset_x = []
    dataset_y = []
    for i in range(sample_length):
        x, y = build_sample(vocab, sentence_length)
        dataset_x.append(x)
        dataset_y.append(y)
    return torch.LongTensor(dataset_x), torch.LongTensor(dataset_y)


# 建立模型
def build_model(vocab, char_dim, sentence_length):
    model = TorchModel(char_dim, sentence_length, vocab)
    return model


# 测试代码
# 用来测试每轮模型的准确率
def evaluate(model, vocab, sample_length):
    model.eval()
    x, y = build_dataset(200, vocab, sample_length)  # 建立200个用于测试的样本
    class_count = [0] * 6
    for label in y:
        class_count[label] += 1
    print("给类样本数量：", class_count)
    correct, wrong = 0, 0
    with torch.no_grad():
        y_pred = model(x)  # 模型预测
        predict_classes = torch.argmax(y_pred, dim=1)
        for y_p, y_t in zip(predict_classes, y):  # 与真实标签进行对比
            if int(y_p) == int(y_t):
                correct += 1
            else:
                wrong += 1
    print("正确预测个数：%d, 正确率：%f" % (correct, correct / (correct + wrong)))
    return correct / (correct + wrong)


def main():
    # 配置参数
    epoch_num = 30  # 训练轮数
    batch_size = 64  # 每次训练样本个数
    train_sample = 5000  # 每轮训练总共训练的样本总数
    char_dim = 128  # 每个字的维度
    sentence_length = 5  # 样本文本长度
    learning_rate = 0.001  # 学习率
    # 建立字表
    vocab = build_vocab()
    # 建立模型
    model = build_model(vocab, char_dim, sentence_length)
    # 选择优化器
    optim = torch.optim.Adam(model.parameters(), lr=learning_rate)
    log = []
    # 训练过程
    for epoch in range(epoch_num):
        model.train()
        watch_loss = []
        for batch in range(int(train_sample / batch_size)):
            x, y = build_dataset(batch_size, vocab, sentence_length)  # 构造一组训练样本
            optim.zero_grad()  # 梯度归零
            loss = model(x, y)  # 计算loss
            loss.backward()  # 计算梯度
            optim.step()  # 更新权重

            watch_loss.append(loss.item())
        print("=========\n第%d轮平均loss:%f" % (epoch + 1, np.mean(watch_loss)))
        acc = evaluate(model, vocab, sentence_length)  # 测试本轮模型结果
        log.append([acc, np.mean(watch_loss)])

    # 保存模型
    torch.save(model.state_dict(), "model.pth")
    # 保存词表
    writer = open("vocab.json", "w", encoding="utf8")
    writer.write(json.dumps(vocab, ensure_ascii=False, indent=2))
    writer.close()
    return


# 使用训练好的模型做预测
def predict(model_path, vocab_path, input_strings):
    char_dim = 128  # 每个字的维度
    sentence_length = 5  # 样本文本长度
    vocab = json.load(open(vocab_path, "r", encoding="utf8"))  # 加载字符表
    model = build_model(vocab, char_dim, sentence_length)  # 建立模型
    model.load_state_dict(torch.load(model_path))  # 加载训练好的权重
    # 类别说明
    class_names = {
        0: "目标字在第1个位置",
        1: "目标字在第2个位置",
        2: "目标字在第3个位置",
        3: "目标字在第4个位置",
        4: "目标字在第5个位置"
    }
    x = []
    for input_string in input_strings:
        seq = [vocab.get(s, vocab['unk']) for s in input_string[:sentence_length]]
        if len(seq) < sentence_length:
            seq += [0] * (sentence_length - len(seq))
        x.append(seq)
    model.eval()  # 测试模式
    with torch.no_grad():  # 不计算梯度
        result = model.forward(torch.LongTensor(x))  # 模型预测
    for i, input_string in enumerate(input_strings):
        pred_class = torch.argmax(result[i]).item()
        prob = result[i][pred_class].item()
        print("输入：%s, 预测类别：%d(%s), 概率值：%f" % (input_string, pred_class, class_names[pred_class], prob))  # 打印结果


if __name__ == "__main__":
    main()
    test_strings = ["我今天很好", "是你对的", "他们来了", "这是好书", "你好朋友"]
    predict("model.pth", "vocab.json", test_strings)

# coding:utf8

import torch
import torch.nn as nn
import numpy as np
import random
import json

"""

基于pytorch的网络编写
实现一个网络完成一个简单nlp任务
判断目标字在样本中的位置

"""


class TorchModel(nn.Module):
    def __init__(self, vector_dim, sentence_length, vocab):
        super(TorchModel, self).__init__()
        self.sentence_length = sentence_length
        self.embedding = nn.Embedding(len(vocab), vector_dim, padding_idx=0)  # embedding层
        self.rnn = nn.RNN(vector_dim, 128, batch_first=True, nonlinearity='tanh')
        self.classify = nn.Linear(128, 5 * (sentence_length + 1))  # 线性层
        self.loss = nn.CrossEntropyLoss()

    # 当输入真实标签，返回loss值；无真实标签，返回预测值
    def forward(self, x, y=None):
        x = self.embedding(x)  # (batch_size, sen_len) -> (batch_size, sen_len, vector_dim)
        rnn_out, rnn_hidden = self.rnn(x)
        last_hidden = rnn_hidden.squeeze()
        x = self.classify(last_hidden)
        x = x.view(-1, 5, self.sentence_length + 1)
        if y is not None:
            total_loss = 0
            for i in range(5):
                total_loss += self.loss(x[:, i, :], y[:,i])
            return total_loss  # 预测值和真实值计算损失
        else:
            return torch.softmax(x, dim=2)


def build_vocab():
    chars = "你我他她好是非对错来去上下左右前后中间旁边内外大小多少长短高低快慢远近新旧老幼男女父母子师生朋友同学事领导客户板员工"  # 字符集
    vocab = {"pad": 0}
    for index, char in enumerate(chars):
        vocab[char] = index + 1  # 每个字对应一个序号
    vocab['unk'] = len(vocab)
    return vocab


# 随机生成一个样本
# 从所有字中选取sentence_length个字
'''
def build_sample(vocab, sentence_length):
    x = [random.choice(list(vocab.keys())) for _ in range(sentence_length)]

    # 定义五个类别对应的目标汉字
    target_chars = ["我", "你", "他", "是", "好"]
    y = 0
    for i, char in enumerate(target_chars):
        if char in x:
            position = x.index(char)
            y = position
            break
    x = [vocab.get(word, vocab['unk']) for word in x]  # 将字转换成序号，为了做embedding
    return x, y
'''
def build_sample(vocab, sentence_length):
    x = [random.choice(list(vocab.keys())) for _ in range(sentence_length)]

    # 定义五个类别对应的目标汉字
    target_chars = ["我", "你", "他", "是", "好"]
    y = []
    for i, char in enumerate(target_chars):
        if char in x:
            position = x.index(char)
            y.append(position)
        else:
            y.append(sentence_length)
    x = [vocab.get(word, vocab['unk']) for word in x]  # 将字转换成序号，为了做embedding
    return x, y


# 建立数据集
# 输入需要的样本数量。需要多少生成多少
def build_dataset(sample_length, vocab, sentence_length):
    dataset_x = []
    dataset_y = []
    for i in range(sample_length):
        x, y = build_sample(vocab, sentence_length)
        dataset_x.append(x)
        dataset_y.append(y)
    return torch.LongTensor(dataset_x), torch.LongTensor(dataset_y)


# 建立模型
def build_model(vocab, char_dim, sentence_length):
    model = TorchModel(char_dim, sentence_length, vocab)
    return model


# 测试代码
# 用来测试每轮模型的准确率
def evaluate(model, vocab, sample_length):
    model.eval()
    x, y = build_dataset(200, vocab, sample_length)  # 建立200个用于测试的样本

    with torch.no_grad():
        y_pred = model(x)  # 模型预测
        for target_idx in range(5):
            target_correct = 0
            predict_classes = torch.argmax(y_pred[:, target_idx, :], dim=1)
            for y_p, y_t in zip(predict_classes, y[:,target_idx]):  # 与真实标签进行对比
                if int(y_p) == int(y_t):
                    target_correct += 1
            print(f"目标{target_correct}准确率：{target_correct/len(y):.4f}")
        all_correct = 0
        for i in range(len(y)):
            correct_count = 0
            for target_idx in range(5):
                pred = torch.argmax(y_pred[i,target_idx, :])
                if pred == y[i, target_idx]:
                    correct_count += 1
            if correct_count == 5:
                all_correct += 1
        accuracy = all_correct / len(y)
        print(f"全对的准确率：{accuracy:.4f}")
    return accuracy


def main():
    # 配置参数
    epoch_num = 30  # 训练轮数
    batch_size = 64  # 每次训练样本个数
    train_sample = 5000  # 每轮训练总共训练的样本总数
    char_dim = 128  # 每个字的维度
    sentence_length = 5  # 样本文本长度
    learning_rate = 0.001  # 学习率
    # 建立字表
    vocab = build_vocab()
    # 建立模型
    model = build_model(vocab, char_dim, sentence_length)
    # 选择优化器
    optim = torch.optim.Adam(model.parameters(), lr=learning_rate)
    log = []
    # 训练过程
    for epoch in range(epoch_num):
        model.train()
        watch_loss = []
        for batch in range(int(train_sample / batch_size)):
            x, y = build_dataset(batch_size, vocab, sentence_length)  # 构造一组训练样本
            optim.zero_grad()  # 梯度归零
            loss = model(x, y)  # 计算loss
            loss.backward()  # 计算梯度
            optim.step()  # 更新权重

            watch_loss.append(loss.item())
        print("=========\n第%d轮平均loss:%f" % (epoch + 1, np.mean(watch_loss)))
        acc = evaluate(model, vocab, sentence_length)  # 测试本轮模型结果
        log.append([acc, np.mean(watch_loss)])

    # 保存模型
    torch.save(model.state_dict(), "model.pth")
    # 保存词表
    writer = open("vocab.json", "w", encoding="utf8")
    writer.write(json.dumps(vocab, ensure_ascii=False, indent=2))
    writer.close()
    return


# 使用训练好的模型做预测
def predict(model_path, vocab_path, input_strings):
    char_dim = 128  # 每个字的维度
    sentence_length = 5  # 样本文本长度
    vocab = json.load(open(vocab_path, "r", encoding="utf8"))  # 加载字符表
    model = build_model(vocab, char_dim, sentence_length)  # 建立模型
    model.load_state_dict(torch.load(model_path))  # 加载训练好的权重
    # 类别说明
    class_names = {
        0: "第1个位置", 1: "第2个位置", 2: "第3个位置", 
        3: "第4个位置", 4: "第5个位置", 5: "不存在"
    }
    target_names = ["我", "你", "他", "是", "好"]
    x = []
    for input_string in input_strings:
        seq = [vocab.get(s, vocab['unk']) for s in input_string[:sentence_length]]
        if len(seq) < sentence_length:
            seq += [0] * (sentence_length - len(seq))
        x.append(seq)
    model.eval()  # 测试模式
    with torch.no_grad():  # 不计算梯度
        result = model.forward(torch.LongTensor(x))  # 模型预测
    for i, input_string in enumerate(input_strings):
        print(f"输入:{input_string}")
        for target_idx in range(5):
            pred_class = torch.argmax(result[i,target_idx,:]).item()
            prob = result[i,target_idx, pred_class].item()
            print(f"  '{target_names[target_idx]}': {class_names[pred_class]}, 概率：{prob:.4f}")
        print() 


if __name__ == "__main__":
    #main()
    test_strings = ["我我今天很好", "是你对的", "他们来了", "这是好书", "哈哈你好吗"]
    predict("model.pth", "vocab.json", test_strings)