HBU深度学习实验15-循环神经网络（2）

LSTM的记忆能力实验

飞桨AI Studio星河社区-人工智能学习与实训社区 (baidu.com)

长短期记忆网络（Long Short-Term Memory Network，LSTM）是一种可以有效缓解长程依赖问题的循环神经网络．LSTM 的特点是引入了一个新的内部状态（Internal State） $c\in\mathbb{R}^D$ 和门控机制（Gating Mechanism）．不同时刻的内部状态以近似线性的方式进行传递，从而缓解梯度消失或梯度爆炸问题．同时门控机制进行信息筛选，可以有效地增加记忆能力．例如，输入门可以让网络忽略无关紧要的输入信息，遗忘门可以使得网络保留有用的历史信息．在上一节的数字求和任务中，如果模型能够记住前两个非零数字，同时忽略掉一些不重要的干扰信息，那么即时序列很长，模型也有效地进行预测.

LSTM 模型在第 t 步时，循环单元的内部结构如图所示．

提醒：为了和代码的实现保存一致性，这里使用形状为 (样本数量 × 序列长度 × 特征维度) 的张量来表示一组样本.

假设一组输入序列为 $X\in\mathbb{R}^{B\times L\times M}$ ，，其中B为批大小，L为序列长度，M为输入特征维度，LSTM从从左到右依次扫描序列，并通过循环单元计算更新每一时刻的状态内部状态 $C_{t}\in\mathbb{R}^{B\times D}$ ，和输出状态 $H_{t}\in\mathbb{R}^{B\times D}$ 。

具体计算分为三步：

（1）计算三个“门”

在时刻t，LSTM的循环单元将当前时刻的输入， $\boldsymbol{X}_t\in\mathbb{R}^{B\times M}$ ，与上一时刻的输出状态 $H_{t-1}\in\mathbb{R}^{B\times D}$ ，计算一组输入门 $I_{t}$ 、遗忘门 $F_{t}$ 和输出门 $O_{t}$ ，其计算公式为

$\begin{gathered} I_t=\sigma(\boldsymbol{X}_t\boldsymbol{W}_i+\boldsymbol{H}_{t-1}\boldsymbol{U}_i+\boldsymbol{b}_i)\in\mathbb{R}^{B\times D} \\ \boldsymbol{F}_t=\sigma(\boldsymbol{X}_t\boldsymbol{W}_f+\boldsymbol{H}_{t-1}\boldsymbol{U}_f+\boldsymbol{b}_f)\in\mathbb{R}^{B\times D}, \\ O_t=\sigma(\boldsymbol{X}_t\boldsymbol{W}_o+\boldsymbol{H}_{t-1}\boldsymbol{U}_o+\boldsymbol{b}_o)\in\mathbb{R}^{B\times D}, \end{gathered}$

其中 $W_*\in\mathbb{R}^{M\times D},U_*\in\mathbb{R}^{D\times D},\boldsymbol{b}_*\in\mathbb{R}^D$ 为可学习的参数，σ表示Logistic函数，将“门”的取值控制在(0,1)区间。这里的“门”都是B个样本组成的矩阵，每一行为一个样本的“门”向量。

（2）计算内部状态

首先计算候选内部状态：

$\tilde{C}_t=\tanh(\boldsymbol{X}_t\boldsymbol{W}_c+\boldsymbol{H}_{t-1}\boldsymbol{U}_c+\boldsymbol{b}_c)\in\mathbb{R}^{B\times D}$

其中 $\boldsymbol{W}_c\in\mathbb{R}^{M\times D},\boldsymbol{U}_c\in\mathbb{R}^{D\times D},\boldsymbol{b}_c\in\mathbb{R}^D$ 为可学习的参数。

使用遗忘门和输入门，计算时刻t的内部状态：

$C_t=F_t\odot C_{t-1}+I_t\odot\tilde{C}_t$

其中⊙为逐元素积。

（3）计算输出状态

当前LSTM单元状态（候选状态）的计算公式为: LSTM单元状态向量Ct和Ht的计算公式为

$C_t=F_t\odot C_{t-1}+I_t\odot\tilde{C}_t,H_t=O_t\odot\tanh(C_t).$

LSTM循环单元结构的输入是t−1时刻内部状态向量 ${C}_{t-1}\in\mathbb{R}^{B\times D}$ ，和隐状态向量 ${H}_{t-1}\in\mathbb{R}^{B\times D}$ ，输出是当前时刻t的状态向量 ${C}_{t}\in\mathbb{R}^{B\times D}$ ，通过LSTM循环单元，整个网络可以建立较长距离的时序依赖关系。

通过学习这些门的设置，LSTM可以选择性地忽略或者强化当前的记忆或是输入信息，帮助网络更好地学习长句子的语义信息。

模型构建

在本实验中，我们将使用上个实验中定义Model_RNN4SeqClass模型，并构建 LSTM 算子．只需要实例化 LSTM 算，并传入Model_RNN4SeqClass模型，就可以用 LSTM 进行数字求和实验

LSTM层

LSTM层的代码与SRN层结构相似，只是在SRN层的基础上增加了内部状态、输入门、遗忘门和输出门的定义和计算。这里LSTM层的输出也依然为序列的最后一个位置的隐状态向量。代码实现如下：

import torch
import torch.nn as nn
import torch.nn.functional as F


class LSTM(nn.Module):
    def __init__(self, input_size, hidden_size, Wi_attr=None, Wf_attr=None, Wo_attr=None, Wc_attr=None,
                 Ui_attr=None, Uf_attr=None, Uo_attr=None, Uc_attr=None, bi_attr=None, bf_attr=None,
                 bo_attr=None, bc_attr=None):
        super(LSTM, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size

        # 初始化模型参数
        if Wi_attr is None:
            Wi = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wi = torch.tensor(Wi_attr, dtype=torch.float32)
        self.W_i = torch.nn.Parameter(Wi)

        if Wf_attr is None:
            Wf = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wf = torch.tensor(Wf_attr, dtype=torch.float32)
        self.W_f = torch.nn.Parameter(Wf)

        if Wo_attr is None:
            Wo = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wo = torch.tensor(Wo_attr, dtype=torch.float32)
        self.W_o = torch.nn.Parameter(Wo)

        if Wc_attr is None:
            Wc = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wc = torch.tensor(Wc_attr, dtype=torch.float32)
        self.W_c = torch.nn.Parameter(Wc)

        if Ui_attr is None:
            Ui = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Ui = torch.tensor(Ui_attr, dtype=torch.float32)
        self.U_i = torch.nn.Parameter(Ui)

        if Uf_attr is None:
            Uf = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uf = torch.tensor(Uf_attr, dtype=torch.float32)
        self.U_f = torch.nn.Parameter(Uf)

        if Uo_attr is None:
            Uo = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uo = torch.tensor(Uo_attr, dtype=torch.float32)
        self.U_o = torch.nn.Parameter(Uo)

        if Uc_attr is None:
            Uc = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uc = torch.tensor(Uc_attr, dtype=torch.float32)
        self.U_c = torch.nn.Parameter(Uc)

        if bi_attr is None:
            bi = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bi = torch.tensor(bi_attr, dtype=torch.float32)
        self.b_i = torch.nn.Parameter(bi)

        if bf_attr is None:
            bf = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bf = torch.tensor(bf_attr, dtype=torch.float32)
        self.b_f = torch.nn.Parameter(bf)

        if bo_attr is None:
            bo = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bo = torch.tensor(bo_attr, dtype=torch.float32)
        self.b_o = torch.nn.Parameter(bo)

        if bc_attr is None:
            bc = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bc = torch.tensor(bc_attr, dtype=torch.float32)
        self.b_c = torch.nn.Parameter(bc)

    # 初始化状态向量和隐状态向量
    def init_state(self, batch_size):
        hidden_state = torch.zeros(size=[batch_size, self.hidden_size], dtype=torch.float32)
        cell_state = torch.zeros(size=[batch_size, self.hidden_size], dtype=torch.float32)
        return hidden_state, cell_state

    # 定义前向计算
    def forward(self, inputs, states=None):
        # inputs: 输入数据，其shape为batch_size x seq_len x input_size
        batch_size, seq_len, input_size = inputs.shape

        # 初始化起始的单元状态和隐状态向量，其shape为batch_size x hidden_size
        if states is None:
            states = self.init_state(batch_size)
        hidden_state, cell_state = states

        # 执行LSTM计算，包括：输入门、遗忘门和输出门、候选内部状态、内部状态和隐状态向量
        for step in range(seq_len):
            # 获取当前时刻的输入数据step_input: 其shape为batch_size x input_size
            step_input = inputs[:, step, :]
            # 计算输入门, 遗忘门和输出门, 其shape为：batch_size x hidden_size
            I_gate = F.sigmoid(torch.matmul(step_input, self.W_i) + torch.matmul(hidden_state, self.U_i) + self.b_i)
            F_gate = F.sigmoid(torch.matmul(step_input, self.W_f) + torch.matmul(hidden_state, self.U_f) + self.b_f)
            O_gate = F.sigmoid(torch.matmul(step_input, self.W_o) + torch.matmul(hidden_state, self.U_o) + self.b_o)
            # 计算候选状态向量, 其shape为：batch_size x hidden_size
            C_tilde = F.tanh(torch.matmul(step_input, self.W_c) + torch.matmul(hidden_state, self.U_c) + self.b_c)
            # 计算单元状态向量, 其shape为：batch_size x hidden_size
            cell_state = F_gate * cell_state + I_gate * C_tilde
            # 计算隐状态向量，其shape为：batch_size x hidden_size
            hidden_state = O_gate * F.tanh(cell_state)

        return hidden_state


Wi_attr = [[0.1, 0.2], [0.1, 0.2]]
Wf_attr = [[0.1, 0.2], [0.1, 0.2]]
Wo_attr = [[0.1, 0.2], [0.1, 0.2]]
Wc_attr = [[0.1, 0.2], [0.1, 0.2]]
Ui_attr = [[0.0, 0.1], [0.1, 0.0]]
Uf_attr = [[0.0, 0.1], [0.1, 0.0]]
Uo_attr = [[0.0, 0.1], [0.1, 0.0]]
Uc_attr = [[0.0, 0.1], [0.1, 0.0]]
bi_attr = [[0.1, 0.1]]
bf_attr = [[0.1, 0.1]]
bo_attr = [[0.1, 0.1]]
bc_attr = [[0.1, 0.1]]

lstm = LSTM(2, 2, Wi_attr=Wi_attr, Wf_attr=Wf_attr, Wo_attr=Wo_attr, Wc_attr=Wc_attr,
            Ui_attr=Ui_attr, Uf_attr=Uf_attr, Uo_attr=Uo_attr, Uc_attr=Uc_attr,
            bi_attr=bi_attr, bf_attr=bf_attr, bo_attr=bo_attr, bc_attr=bc_attr)

inputs = torch.as_tensor([[[1, 0]]], dtype=torch.float32)
hidden_state = lstm(inputs)
print(hidden_state)

Pytorch框架已经内置了LSTM的API nn.LSTM

# 这里创建一个随机数组作为测试数据，数据shape为batch_size x seq_len x input_size
batch_size, seq_len, input_size = 8, 20, 32
inputs = torch.randn(size=[batch_size, seq_len, input_size])

# 设置模型的hidden_size
hidden_size = 32
torch_lstm = nn.LSTM(input_size, hidden_size)
self_lstm = LSTM(input_size, hidden_size)

self_hidden_state = self_lstm(inputs)
torch_outputs, (torch_hidden_state, torch_cell_state) = torch_lstm(inputs)

print("self_lstm hidden_state: ", self_hidden_state.shape)
print("torch_lstm outpus:", torch_outputs.shape)
print("torch_lstm hidden_state:", torch_hidden_state.shape)
print("torch_lstm cell_state:", torch_cell_state.shape)

在进行实验时，首先定义输入数据inputs，然后将该数据分别传入Pytorch内置的LSTM与自己实现的LSTM模型中，最后通过对比两者的隐状态输出向量。代码实现如下：

import torch
torch.seed()
 
# 这里创建一个随机数组作为测试数据，数据shape为batch_size x seq_len x input_size
batch_size, seq_len, input_size, hidden_size = 2, 5, 10, 10
inputs = torch.randn([batch_size, seq_len, input_size])
 
# 设置模型的hidden_size
torch_lstm = nn.LSTM(input_size, hidden_size, bias=True)
# 获取torch_lstm中的参数，并设置相应的paramAttr,用于初始化lstm
print(torch_lstm.weight_ih_l0.T.shape)
chunked_W = torch.split(torch_lstm.weight_ih_l0.T, split_size_or_sections=hidden_size, dim=-1)
# 因为手写LSTM算子与输入相关的权重矩阵大小为input_size * hidden_size
# torch.nn.LSTM 在之前我们提到过，矩阵大小为 [4 * hidden_size, input_size] 所以如果需要将模型的参数取出来
# 首先需要转置 [input_size, 4 * hidden_size] 并且调用分割函数，对列按照每hidden_size大小分割即可
chunked_U = torch.split(torch_lstm.weight_hh_l0.T, split_size_or_sections=hidden_size, dim=-1)
chunked_b = torch.split(torch_lstm.bias_hh_l0, split_size_or_sections=hidden_size)
# torch_lstm.bias_hh_l0 因为大小为tensor[4 * hidden_size] 属于一维所以不用转置
Wi_attr = chunked_W[0].detach().numpy()
Wf_attr = chunked_W[1].detach().numpy()
Wc_attr = chunked_W[2].detach().numpy()
Wo_attr = chunked_W[3].detach().numpy()
Ui_attr = chunked_U[0].detach().numpy()
Uf_attr = chunked_U[1].detach().numpy()
Uc_attr = chunked_U[2].detach().numpy()
Uo_attr = chunked_U[3].detach().numpy()
bi_attr = chunked_b[0].detach().numpy()
bf_attr = chunked_b[1].detach().numpy()
bc_attr = chunked_b[2].detach().numpy()
bo_attr = chunked_b[3].detach().numpy()
self_lstm = LSTM(input_size, hidden_size, Wi_attr=Wi_attr, Wf_attr=Wf_attr, Wo_attr=Wo_attr, Wc_attr=Wc_attr,
                 Ui_attr=Ui_attr, Uf_attr=Uf_attr, Uo_attr=Uo_attr, Uc_attr=Uc_attr,
                 bi_attr=bi_attr, bf_attr=bf_attr, bo_attr=bo_attr, bc_attr=bc_attr)
 
# 进行前向计算，获取隐状态向量，并打印展示
self_hidden_state = self_lstm(inputs)
torch_outputs, (torch_hidden_state, _) = torch_lstm(inputs)
print("torch SRN:\n", torch_hidden_state.detach().numpy().squeeze(0))
print("self SRN:\n", self_hidden_state.detach().numpy())

可以看到，两者的输出基本是一致的。另外，还可以进行对比两者在运算速度方面的差异。代码实现如下：

import time
 
# 这里创建一个随机数组作为测试数据，数据shape为batch_size x seq_len x input_size
batch_size, seq_len, input_size = 8, 20, 32
inputs = torch.randn([batch_size, seq_len, input_size])
 
# 设置模型的hidden_size
hidden_size = 32
self_lstm = LSTM(input_size, hidden_size)
torch_lstm = nn.LSTM(input_size, hidden_size)
 
# 计算自己实现的SRN运算速度
model_time = 0
for i in range(100):
    strat_time = time.time()
    hidden_state = self_lstm(inputs)
    # 预热10次运算，不计入最终速度统计
    if i < 10:
        continue
    end_time = time.time()
    model_time += (end_time - strat_time)
avg_model_time = model_time / 90
print('self_lstm speed:', avg_model_time, 's')
 
# 计算torch内置的SRN运算速度
model_time = 0
for i in range(100):
    strat_time = time.time()
    outputs, (hidden_state, cell_state) = torch_lstm(inputs)
    # 预热10次运算，不计入最终速度统计
    if i < 10:
        continue
    end_time = time.time()
    model_time += (end_time - strat_time)
avg_model_time = model_time / 90
print('torch_lstm speed:', avg_model_time, 's')

可以看到，由于Pytorch框架的LSTM底层采用了C++实现并进行优化，Pytorch框架内置的LSTM运行效率远远高于自己实现的LSTM。

模型训练

训练指定长度的数字预测模型

本节将基于RunnerV3类进行训练，首先定义模型训练的超参数，并保证和简单循环网络的超参数一致. 然后定义一个train函数，其可以通过指定长度的数据集，并进行训练. 在train函数中，首先加载长度为length的数据，然后实例化各项组件并创建对应的Runner，然后训练该Runner。同时在本节将使用之前定义的准确度（Accuracy）作为评估指标，代码实现如下：

# 训练轮次
num_epochs = 500
# 学习率
lr = 0.001
# 输入数字的类别数
num_digits = 10
# 将数字映射为向量的维度
input_size = 32
# 隐状态向量的维度
hidden_size = 32
# 预测数字的类别数
num_classes = 19
# 批大小
batch_size = 8
# 模型保存目录
save_dir = "./checkpoints"
 
# 可以设置不同的length进行不同长度数据的预测实验
def train(length):
    print(f"\n====> Training LSTM with data of length {length}.")
    np.random.seed(0)
    random.seed(0)
 
    # 加载长度为length的数据
    data_path = f"./datasets/{length}"
    train_examples, dev_examples, test_examples = load_data(data_path)
    train_set, dev_set, test_set = DigitSumDataset(train_examples), DigitSumDataset(dev_examples), DigitSumDataset(test_examples)
    train_loader = DataLoader(train_set, batch_size=batch_size)
    dev_loader = DataLoader(dev_set, batch_size=batch_size)
    test_loader = DataLoader(test_set, batch_size=batch_size)
    # 实例化模型
    base_model = LSTM(input_size, hidden_size)
    model = Model_RNN4SeqClass(base_model, num_digits, input_size, hidden_size, num_classes)
    # 指定优化器
    optimizer = torch.optim.Adam(lr=lr, params=model.parameters())
    # 定义评价指标
    metric = Accuracy()
    # 定义损失函数
    loss_fn = torch.nn.CrossEntropyLoss()
    # 基于以上组件，实例化Runner
    runner = RunnerV3(model, optimizer, loss_fn, metric)
 
    # 进行模型训练
    model_save_path = os.path.join(save_dir, f"best_lstm_model_{length}.pdparams")
    runner.train(train_loader, dev_loader, num_epochs=num_epochs, eval_steps=100, log_steps=100, save_path=model_save_path)
 
    return runner

多组训练

接下来，分别进行数据长度为10, 15, 20, 25, 30, 35的数字预测模型训练实验，训练后的runner保存至runners字典中。

lstm_runners = {}
 
lengths = [10, 15, 20, 25, 30, 35]
for length in lengths:
    runner = train(length)
    lstm_runners[length] = runner

损失曲线展示

分别画出基于LSTM的各个长度的数字预测模型训练过程中，在训练集和验证集上的损失曲线，代码实现如下：

# 画出训练过程中的损失图
for length in lengths:
    runner = lstm_runners[length]
    fig_name = f"./images/6.11_{length}.pdf"
    plot_training_loss(runner, fig_name, sample_step=100)

下图展示了LSTM模型在不同长度数据集上进行训练后的损失变化，同SRN模型一样，随着序列长度的增加，训练集上的损失逐渐不稳定，验证集上的损失整体趋向于变大，这说明当序列长度增加时，保持长期依赖的能力同样在逐渐变弱. 同图6.5相比，LSTM模型在序列长度增加时，收敛情况比SRN模型更好。

模型评价

在测试集上进行模型评价

使用测试数据对在训练过程中保存的最好模型进行评价，观察模型在测试集上的准确率. 同时获取模型在训练过程中在验证集上最好的准确率，实现代码如下:

lstm_dev_scores = []
lstm_test_scores = []
for length in lengths:
    print(f"Evaluate LSTM with data length {length}.")
    runner = lstm_runners[length]
    # 加载训练过程中效果最好的模型
    model_path = os.path.join(save_dir, f"best_lstm_model_{length}.pdparams")
    runner.load_model(model_path)
 
    # 加载长度为length的数据
    data_path = f"./datasets/{length}"
    train_examples, dev_examples, test_examples = load_data(data_path)
    test_set = DigitSumDataset(test_examples)
    test_loader = DataLoader(test_set, batch_size=batch_size)
 
    # 使用测试集评价模型，获取测试集上的预测准确率
    score, _ = runner.evaluate(test_loader)
    lstm_test_scores.append(score)
    lstm_dev_scores.append(max(runner.dev_scores))
 
for length, dev_score, test_score in zip(lengths, lstm_dev_scores, lstm_test_scores):
    print(f"[LSTM] length:{length}, dev_score: {dev_score}, test_score: {test_score: .5f}")

模型在不同长度的数据集上的准确率变化图

接下来，将SRN和LSTM在不同长度的验证集和测试集数据上的准确率绘制成图片，以方面观察。

plt.plot(lengths, lstm_dev_scores, '-o', color='#e8609b',  label="LSTM Dev Accuracy")
plt.plot(lengths, lstm_test_scores,'-o', color='#000000', label="LSTM Test Accuracy")
 
#绘制坐标轴和图例
plt.ylabel("accuracy", fontsize='large')
plt.xlabel("sequence length", fontsize='large')
plt.legend(loc='lower left', fontsize='x-large')
 
fig_name = "./images/6.12.pdf"
plt.savefig(fig_name)
plt.show()

下图展示了LSTM模型与SRN模型在不同长度数据集上的准确度对比。随着数据集长度的增加，LSTM模型在验证集和测试集上的准确率整体也趋向于降低；同时LSTM模型的准确率显著高于SRN模型，表明LSTM模型保持长期依赖的能力要优于SRN模型.

LSTM模型门状态和单元状态的变化

LSTM模型通过门控机制控制信息的单元状态的更新，这里可以观察当LSTM在处理一条数字序列的时候，相应门和单元状态是如何变化的。首先需要对以上LSTM模型实现代码中，定义相应列表进行存储这些门和单元状态在每个时刻的向量。

class LSTM(nn.Module):
    def __init__(self, input_size, hidden_size, Wi_attr=None, Wf_attr=None, Wo_attr=None, Wc_attr=None,
                 Ui_attr=None, Uf_attr=None, Uo_attr=None, Uc_attr=None, bi_attr=None, bf_attr=None,
                 bo_attr=None, bc_attr=None):
        super(LSTM, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size
 
        # 初始化模型参数
        if Wi_attr == None:
            Wi = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wi = torch.tensor(Wi_attr, dtype=torch.float32)
        self.W_i = torch.nn.Parameter(Wi)
 
        if Wf_attr == None:
            Wf = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wf = torch.tensor(Wf_attr, dtype=torch.float32)
        self.W_f = torch.nn.Parameter(Wf)
 
        if Wo_attr == None:
            Wo = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wo = torch.tensor(Wo_attr, dtype=torch.float32)
        self.W_o = torch.nn.Parameter(Wo)
 
        if Wc_attr == None:
            Wc = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wc = torch.tensor(Wc_attr, dtype=torch.float32)
        self.W_c = torch.nn.Parameter(Wc)
 
        if Ui_attr == None:
            Ui = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Ui = torch.tensor(Ui_attr, dtype=torch.float32)
        self.U_i = torch.nn.Parameter(Ui)
        if Uf_attr == None:
            Uf = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uf = torch.tensor(Uf_attr, dtype=torch.float32)
        self.U_f = torch.nn.Parameter(Uf)
 
        if Uo_attr == None:
            Uo = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uo = torch.tensor(Uo_attr, dtype=torch.float32)
        self.U_o = torch.nn.Parameter(Uo)
 
        if Uc_attr == None:
            Uc = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uc = torch.tensor(Uc_attr, dtype=torch.float32)
        self.U_c = torch.nn.Parameter(Uc)
 
        if bi_attr == None:
            bi = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bi = torch.tensor(bi_attr, dtype=torch.float32)
        self.b_i = torch.nn.Parameter(bi)
        if bf_attr == None:
            bf = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bf = torch.tensor(bf_attr, dtype=torch.float32)
        self.b_f = torch.nn.Parameter(bf)
        if bo_attr == None:
            bo = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bo = torch.tensor(bo_attr, dtype=torch.float32)
        self.b_o = torch.nn.Parameter(bo)
        if bc_attr == None:
            bc = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bc = torch.tensor(bc_attr, dtype=torch.float32)
        self.b_c = torch.nn.Parameter(bc)
 
    # 初始化状态向量和隐状态向量
    def init_state(self, batch_size):
        hidden_state = torch.zeros(size=[batsize, self.hidden_size], dtype=torch.float32)
        cell_state = torch.zeros(size=[batch_ch_size, self.hidden_size], dtype=torch.float32)
        return hidden_state, cell_state
 
    # 定义前向计算
    def forward(self, inputs, states=None):
        # inputs: 输入数据，其shape为batch_size x seq_len x input_size
        batch_size, seq_len, input_size = inputs.shape
 
        # 初始化起始的单元状态和隐状态向量，其shape为batch_size x hidden_size
        if states is None:
            states = self.init_state(batch_size)
        hidden_state, cell_state = states
 
        # 定义相应的门状态和单元状态向量列表
        self.Is = []
        self.Fs = []
        self.Os = []
        self.Cs = []
        # 初始化状态向量和隐状态向量
        cell_state = torch.zeros(size=[batch_size, self.hidden_size], dtype=torch.float32)
        hidden_state = torch.zeros(size=[batch_size, self.hidden_size], dtype=torch.float32)
 
        # 执行LSTM计算，包括：隐藏门、输入门、遗忘门、候选状态向量、状态向量和隐状态向量
        for step in range(seq_len):
            input_step = inputs[:, step, :]
            I_gate = F.sigmoid(torch.matmul(input_step, self.W_i) + torch.matmul(hidden_state, self.U_i) + self.b_i)
            F_gate = F.sigmoid(torch.matmul(input_step, self.W_f) + torch.matmul(hidden_state, self.U_f) + self.b_f)
            O_gate = F.sigmoid(torch.matmul(input_step, self.W_o) + torch.matmul(hidden_state, self.U_o) + self.b_o)
            C_tilde = F.tanh(torch.matmul(input_step, self.W_c) + torch.matmul(hidden_state, self.U_c) + self.b_c)
            cell_state = F_gate * cell_state + I_gate * C_tilde
            hidden_state = O_gate * F.tanh(cell_state)
            # 存储门状态向量和单元状态向量
            self.Is.append(I_gate.detach().numpy().copy())
            self.Fs.append(F_gate.detach().numpy().copy())
            self.Os.append(O_gate.detach().numpy().copy())
            self.Cs.append(cell_state.detach().numpy().copy())
        return hidden_state

接下来，需要使用新的LSTM模型，重新实例化一个runner，本节使用序列长度为10的模型进行此项实验，因此需要加载序列长度为10的模型。

# 实例化模型
base_model = LSTM(input_size, hidden_size)
model = Model_RNN4SeqClass(base_model, num_digits, input_size, hidden_size, num_classes)
# 指定优化器
optimizer = torch.optim.Adam(lr=lr, params=model.parameters())
# 定义评价指标
metric = Accuracy()
# 定义损失函数
loss_fn = torch.nn.CrossEntropyLoss()
# 基于以上组件，重新实例化Runner
runner = RunnerV3(model, optimizer, loss_fn, metric)
 
length = 10
# 加载训练过程中效果最好的模型
model_path = os.path.join(save_dir, f"best_lstm_model_{length}.pdparams")
runner.load_model(model_path)

接下来，给定一条数字序列，并使用数字预测模型进行数字预测，这样便会将相应的门状态和单元状态向量保存至模型中. 然后分别从模型中取出这些向量，并将这些向量进行绘制展示。代码实现如下：

import seaborn as sns
import matplotlib.pyplot as plt
def plot_tensor(inputs, tensor,  save_path, vmin=0, vmax=1):
    tensor = np.stack(tensor, axis=0)
    tensor = np.squeeze(tensor, 1).T
 
    plt.figure(figsize=(16,6))
    # vmin, vmax定义了色彩图的上下界
    ax = sns.heatmap(tensor, vmin=vmin, vmax=vmax)
    ax.set_xticklabels(inputs)
    ax.figure.savefig(save_path)
 
 
# 定义模型输入
inputs = [6, 7, 0, 0, 1, 0, 0, 0, 0, 0]
X = torch.as_tensor(inputs.copy())
X = X.unsqueeze(0)
# 进行模型预测，并获取相应的预测结果
logits = runner.predict(X)
predict_label = torch.argmax(logits, dim=-1)
print(f"predict result: {predict_label.numpy()[0]}")
 
# 输入门
Is = runner.model.rnn_model.Is
plot_tensor(inputs, Is, save_path="./images/6.13_I.pdf")
# 遗忘门
Fs = runner.model.rnn_model.Fs
plot_tensor(inputs, Fs, save_path="./images/6.13_F.pdf")
# 输出门
Os = runner.model.rnn_model.Os
plot_tensor(inputs, Os, save_path="./images/6.13_O.pdf")
# 单元状态
Cs = runner.model.rnn_model.Cs
plot_tensor(inputs, Cs, save_path="./images/6.13_C.pdf", vmin=-5, vmax=5)

当LSTM处理序列数据[6, 7, 0, 0, 1, 0, 0, 0, 0, 0]的过程中单元状态和门数值的变化图，其中横坐标为输入数字，纵坐标为相应门或单元状态向量的维度，颜色的深浅代表数值的大小。可以看到，当输入门遇到不同位置的数字0时，保持了相对一致的数值大小，表明对于0元素保持相同的门控过滤机制，避免输入信息的变化给当前模型带来困扰；当遗忘门遇到数字1后，遗忘门数值在一些维度上变小，表明对某些信息进行了遗忘；随着序列的输入，输出门和单元状态在某些维度上数值变小，在某些维度上数值变大，表明输出门在根据信息的重要性选择信息进行输出，同时单元状态也在保持着对文本预测重要的一些信息.

双向LSTM

import torch
import torch.nn as nn
import torch.nn.functional as F


class BiLSTM(nn.Module):
    def __init__(self, input_size, hidden_size, Wi_attr=None, Wf_attr=None, Wo_attr=None, Wc_attr=None,
                 Ui_attr=None, Uf_attr=None, Uo_attr=None, Uc_attr=None, bi_attr=None, bf_attr=None,
                 bo_attr=None, bc_attr=None):
        super(BiLSTM, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size

        # 初始化正向LSTM的模型参数
        if Wi_attr is None:
            Wi_forward = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wi_forward = torch.tensor(Wi_attr, dtype=torch.float32)
        self.W_i_forward = torch.nn.Parameter(Wi_forward)

        if Wf_attr is None:
            Wf_forward = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wf_forward = torch.tensor(Wf_attr, dtype=torch.float32)
        self.W_f_forward = torch.nn.Parameter(Wf_forward)

        if Wo_attr is None:
            Wo_forward = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wo_forward = torch.tensor(Wo_attr, dtype=torch.float32)
        self.W_o_forward = torch.nn.Parameter(Wo_forward)

        if Wc_attr is None:
            Wc_forward = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wc_forward = torch.tensor(Wc_attr, dtype=torch.float32)
        self.W_c_forward = torch.nn.Parameter(Wc_forward)

        if Ui_attr is None:
            Ui_forward = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Ui_forward = torch.tensor(Ui_attr, dtype=torch.float32)
        self.U_i_forward = torch.nn.Parameter(Ui_forward)

        if Uf_attr is None:
            Uf_forward = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uf_forward = torch.tensor(Uf_attr, dtype=torch.float32)
        self.U_f_forward = torch.nn.Parameter(Uf_forward)

        if Uo_attr is None:
            Uo_forward = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uo_forward = torch.tensor(Uo_attr, dtype=torch.float32)
        self.U_o_forward = torch.nn.Parameter(Uo_forward)

        if Uc_attr is None:
            Uc_forward = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uc_forward = torch.tensor(Uc_attr, dtype=torch.float32)
        self.U_c_forward = torch.nn.Parameter(Uc_forward)

        if bi_attr is None:
            bi_forward = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bi_forward = torch.tensor(bi_attr, dtype=torch.float32)
        self.b_i_forward = torch.nn.Parameter(bi_forward)

        if bf_attr is None:
            bf_forward = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bf_forward = torch.tensor(bf_attr, dtype=torch.float32)
        self.b_f_forward = torch.nn.Parameter(bf_forward)

        if bo_attr is None:
            bo_forward = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bo_forward = torch.tensor(bo_attr, dtype=torch.float32)
        self.b_o_forward = torch.nn.Parameter(bo_forward)

        if bc_attr is None:
            bc_forward = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bc_forward = torch.tensor(bc_attr, dtype=torch.float32)
        self.b_c_forward = torch.nn.Parameter(bc_forward)

        # 初始化反向LSTM的模型参数，初始化方式与正向一致，但参数独立
        if Wi_attr is None:
            Wi_backward = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wi_backward = torch.tensor(Wi_attr, dtype=torch.float32)
        self.W_i_backward = torch.nn.Parameter(Wi_backward)

        if Wf_attr is None:
            Wf_backward = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wf_backward = torch.tensor(Wf_attr, dtype=torch.float32)
        self.W_f_backward = torch.nn.Parameter(Wf_backward)

        if Wo_attr is None:
            Wo_backward = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wo_backward = torch.tensor(Wo_attr, dtype=torch.float32)
        self.W_o_backward = torch.nn.Parameter(Wo_backward)

        if Wc_attr is None:
            Wc_backward = torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            Wc_backward = torch.tensor(Wc_attr, dtype=torch.float32)
        self.W_c_backward = torch.nn.Parameter(Wc_backward)

        if Ui_attr is None:
            Ui_backward = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Ui_backward = torch.tensor(Ui_attr, dtype=torch.float32)
        self.U_i_backward = torch.nn.Parameter(Ui_backward)

        if Uf_attr is None:
            Uf_backward = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uf_backward = torch.tensor(Uf_attr, dtype=torch.float32)
        self.U_f_backward = torch.nn.Parameter(Uf_backward)

        if Uo_attr is None:
            Uo_backward = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uo_backward = torch.tensor(Uo_attr, dtype=torch.float32)
        self.U_o_backward = torch.nn.Parameter(Uo_backward)

        if Uc_attr is None:
            Uc_backward = torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            Uc_backward = torch.tensor(Uc_attr, dtype=torch.float32)
        self.U_c_backward = torch.nn.Parameter(Uc_backward)

        if bi_attr is None:
            bi_backward = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bi_backward = torch.tensor(bi_attr, dtype=torch.float32)
        self.b_i_backward = torch.nn.Parameter(bi_backward)

        if bf_attr is None:
            bf_backward = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bf_backward = torch.tensor(bf_attr, dtype=torch.float32)
        self.b_f_backward = torch.nn.Parameter(bf_backward)

        if bo_attr is None:
            bo_backward = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bo_backward = torch.tensor(bo_attr, dtype=torch.float32)
        self.b_o_backward = torch.nn.Parameter(bo_backward)

        if bc_attr is None:
            bc_backward = torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            bc_backward = torch.tensor(bc_attr, dtype=torch.float32)
        self.b_c_backward = torch.nn.Parameter(bc_backward)

    def init_state(self, batch_size):
        # 分别初始化正向和反向的起始单元状态和隐状态向量，形状均为batch_size x hidden_size
        hidden_state_forward = torch.zeros(size=[batch_size, self.hidden_size], dtype=torch.float32)
        cell_state_forward = torch.zeros(size=[batch_size, self.hidden_size], dtype=torch.float32)

        hidden_state_backward = torch.zeros(size=[batch_size, self.hidden_size], dtype=torch.float32)
        cell_state_backward = torch.zeros(size=[batch_size, self.hidden_size], dtype=torch.float32)

        return (hidden_state_forward, cell_state_forward), (hidden_state_backward, cell_state_backward)

    def forward(self, inputs, states=None):
        batch_size, seq_len, input_size = inputs.shape

        # 初始化起始状态，如果未传入则调用init_state方法获取
        if states is None:
            states = self.init_state(batch_size)
        (hidden_state_forward, cell_state_forward), (hidden_state_backward, cell_state_backward) = states

        # 正向LSTM计算
        forward_hidden_states = []
        for step in range(seq_len):
            step_input = inputs[:, step, :]
            I_gate_forward = F.sigmoid(torch.matmul(step_input, self.W_i_forward) +
                                       torch.matmul(hidden_state_forward, self.U_i_forward) +
                                       self.b_i_forward)
            F_gate_forward = F.sigmoid(torch.matmul(step_input, self.W_f_forward) +
                                       torch.matmul(hidden_state_forward, self.U_f_forward) +
                                       self.b_f_forward)
            O_gate_forward = F.sigmoid(torch.matmul(step_input, self.W_o_forward) +
                                       torch.matmul(hidden_state_forward, self.U_o_forward) +
                                       self.b_o_forward)
            C_tilde_forward = F.tanh(torch.matmul(step_input, self.W_c_forward) +
                                     torch.matmul(hidden_state_forward, self.U_c_forward) +
                                     self.b_c_forward)
            cell_state_forward = F_gate_forward * cell_state_forward + I_gate_forward * C_tilde_forward
            hidden_state_forward = O_gate_forward * F.tanh(cell_state_forward)
            forward_hidden_states.append(hidden_state_forward)

        # 将正向隐状态序列转换为合适的张量形式，形状为batch_size x seq_len x hidden_size
        forward_hidden_states = torch.stack(forward_hidden_states, dim=1)

        # 反转输入序列，用于反向LSTM计算
        reversed_inputs = torch.flip(inputs, dims=[1])

        # 反向LSTM计算
        backward_hidden_states = []
        for step in range(seq_len):
            step_input = reversed_inputs[:, step, :]
            I_gate_backward = F.sigmoid(torch.matmul(step_input, self.W_i_backward) +
                                        torch.matmul(hidden_state_backward, self.U_i_backward) +
                                        self.b_i_backward)
            F_gate_backward = F.sigmoid(torch.matmul(step_input, self.W_f_backward) +
                                        torch.matmul(hidden_state_backward, self.U_f_backward) +
                                        self.b_f_backward)
            O_gate_backward = F.sigmoid(torch.matmul(step_input, self.W_o_backward) +
                                        torch.matmul(hidden_state_backward, self.U_o_backward) +
                                        self.b_o_backward)
            C_tilde_backward = F.tanh(torch.matmul(step_input, self.W_c_backward) +
                                      torch.matmul(hidden_state_backward, self.U_c_backward) +
                                      self.b_c_backward)
            cell_state_backward = F_gate_backward * cell_state_backward + I_gate_backward * C_tilde_backward
            hidden_state_backward = O_gate_backward * F.tanh(cell_state_backward)
            backward_hidden_states.append(hidden_state_backward)

        # 将反向隐状态序列转换为合适的张量形式，并反转回原来的顺序，形状为batch_size x seq_len x hidden_size
        backward_hidden_states = torch.stack(backward_hidden_states, dim=1)
        backward_hidden_states = torch.flip(backward_hidden_states, dims=[1])

        # 常见的做法是将正向和反向的隐状态在最后一个维度上进行拼接，得到最终双向LSTM的输出隐状态
        final_hidden_states = torch.cat([forward_hidden_states, backward_hidden_states], dim=2)

        return final_hidden_states

# 以下是测试代码部分

# 设置输入维度、隐藏层维度、批次大小以及序列长度等参数
input_size = 10
hidden_size = 20
batch_size = 3
seq_len = 5

# 创建一个随机的输入张量，模拟输入数据，形状为batch_size x seq_len x input_size
input_data = torch.randn(batch_size, seq_len, input_size)

# 实例化BiLSTM模型
bi_lstm_model = BiLSTM(input_size, hidden_size)

# 调用模型的前向传播方法，传入输入数据，获取输出结果
output = bi_lstm_model(input_data)

# 打印输出结果的形状，检查是否符合预期（batch_size x seq_len x 2 * hidden_size）
print("Output shape:", output.shape)

总结

试着自己画画LSTM结构原理图（用visio画的，也可以用ppt）

LSTM通过引入“门”结构和“细胞状态”来解决这个问题。门结构可以控制信息的流入和流出，细胞状态则可以长期存储状态。这使得LSTM能够在长序列中更好地捕捉依赖关系。

LSTM的反向传播的数学推导很繁琐，因为涉及到的变量很多，但是LSTM确实是可以在一定程度上解决梯度消失和梯度爆炸的问题。我简单说一下，RNN的连乘主要是W的连乘，而W是一样的，因此就是一个指数函数（在梯度中出现指数函数并不是一件友好的事情）；相反，LSTM的连乘是 $C_{t}$ 对 $C_{t-1}$ 的偏导的不断累乘，如果前后的记忆差别不大，那偏导的值就是1，那就是多个1相乘。当然，也可能出现某一些偏导的值很大，但是一定不会很多（换句话说，一句话的前后没有逻辑，那完全没有训练的必要）。