循环神经网络-RNN记忆能力实验 [HBU]

一、循环神经网络

二、循环神经网络的记忆能力实验

三、数据集构建

数据集的构建函数

加载数据并进行数据划分

构造Dataset类

四、模型构建

嵌入层

SRN层

五、模型训练

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

多组训练

损失曲线展示

六、模型评价

参考《神经网络与深度学习》中的公式(6.50)，改进SRN的循环单元，加入隐状态之间的残差连接，并重复数字求和实验。观察是否可以缓解长程依赖问题？

总结

参考原文章：aistudio.baidu.com/projectdetail/7214712

一、循环神经网络

循环神经网络（Recurrent Neural Network，RNN）是一类具有短期记忆能力的神经网络．在循环神经网络中，神经元不但可以接受其他神经元的信息，也可以接受自身的信息，形成具有环路的网络结构．和前馈神经网络相比，循环神经网络更加符合生物神经网络的结构．目前，循环神经网络已经被广泛应用在语音识别、语言模型以及自然语言生成等任务上．

本章内容基于邱锡鹏 -《神经网络与深度学习》第6章：循环神经网络的相关内容进行设计。在阅读本章之前，建议先了解如图6.1所示的关键知识点，以便更好地理解和掌握相应的理论和实践知识。

图6.1《神经网络与深度学习》关键知识点回顾

本章内容主要包含两部分：

模型解读：介绍经典循环神经网络原理，为了更好地理解长程依赖问题，我们设计一个简单的数字求和任务来验证简单循环网络的记忆能力。长程依赖问题具体可分为梯度爆炸和梯度消失两种情况。对于梯度爆炸，我们复现简单循环网络的梯度爆炸现象并尝试解决。对于梯度消失，一种有效的方式是改进模型，我们也动手实现一个长短期记忆网络，并观察是否可以缓解长程依赖问题。
案例实践：基于双向长短期记忆网络实现文本分类任务．并了解如何进行补齐序列数据，如何将文本数据转为向量表示，如何对补齐位置进行掩蔽等实践知识。

循环神经网络的参数可以通过梯度下降法来学习。和前馈神经网络类似，我们可以使用随时间反向传播（BackPropagation Through Time，BPTT）算法高效地手工计算梯度，也可以使用自动微分的方法，通过计算图自动计算梯度。

循环神经网络被认为是图灵完备的，一个完全连接的循环神经网络可以近似解决所有的可计算问题。然而，虽然理论上循环神经网络可以建立长时间间隔的状态之间的依赖关系，但是由于具体的实现方式和参数学习方式会导致梯度爆炸或梯度消失问题，实际上，通常循环神经网络只能学习到短期的依赖关系，很难建模这种长距离的依赖关系，称为长程依赖问题（Long-Term Dependencies Problem）。

二、循环神经网络的记忆能力实验

图6.2 展示了一个按时间展开的循环神经网络:

简单循环网络在参数学习时存在长程依赖问题，很难建模长时间间隔（Long Range）的状态之间的依赖关系。为了测试简单循环网络的记忆能力，本节构建一个数字求和任务进行实验。

数字求和任务的输入是一串数字，前两个位置的数字为0-9，其余数字随机生成（主要为0），预测目标是输入序列中前两个数字的加和。下图展示了长度为10的数字序列：

如果序列长度越长，准确率越高，则说明网络的记忆能力越好．因此，我们可以构建不同长度的数据集，通过验证简单循环网络在不同长度的数据集上的表现，从而测试简单循环网络的长程依赖能力。

三、数据集构建

我们首先构建不同长度的数字预测数据集DigitSum.

数据集的构建函数

由于在本任务中，输入序列的前两位数字为 0 − 9，其组合数是固定的，所以可以穷举所有的前两位数字组合，并在后面默认用0填充到固定长度. 但考虑到数据的多样性，这里对生成的数字序列中的零位置进行随机采样，并将其随机替换成0-9的数字以增加样本的数量．

我们可以通过设置k的数值来指定一条样本随机生成的数字序列数量.当生成某个指定长度的数据集时，会同时生成训练集、验证集和测试集。当k=3时，生成训练集。当k=1时，生成验证集和测试集. 流程及代码实现如下：

#NNDL 实验
#NNDL 数据集的构建函数
import random
import numpy as np

# 固定随机种子
random.seed(0)
np.random.seed(0)


def generate_data(length, k, save_path):
    if length < 3:
        raise ValueError("The length of data should be greater than 2.")
    if k == 0:
        raise ValueError("k should be greater than 0.")
    # 生成100条长度为length的数字序列，除前两个字符外，序列其余数字暂用0填充
    base_examples = []
    for n1 in range(0, 10):
        for n2 in range(0, 10):
            seq = [n1, n2] + [0] * (length - 2)
            label = n1 + n2
            base_examples.append((seq, label))

    examples = []
    # 数据增强：对base_examples中的每条数据，默认生成k条数据，放入examples
    for base_example in base_examples:
        for _ in range(k):
            # 随机生成替换的元素位置和元素
            idx = np.random.randint(2, length)
            val = np.random.randint(0, 10)
            # 对序列中的对应零元素进行替换
            seq = base_example[0].copy()
            label = base_example[1]
            seq[idx] = val
            examples.append((seq, label))

    # 保存增强后的数据
    with open(save_path, "w", encoding="utf-8") as f:
        for example in examples:
            # 将数据转为字符串类型，方便保存
            seq = [str(e) for e in example[0]]
            label = str(example[1])
            line = " ".join(seq) + "\t" + label + "\n"
            f.write(line)

    print(f"generate data to: {save_path}.")


# 定义生成的数字序列长度
lengths = [5, 10, 15, 20, 25, 30, 35]
for length in lengths:
    # 生成长度为length的训练数据
    save_path = f"./datasets/{length}/train.txt"
    k = 3
    generate_data(length, k, save_path)
    # 生成长度为length的验证数据
    save_path = f"./datasets/{length}/dev.txt"
    k = 1
    generate_data(length, k, save_path)
    # 生成长度为length的测试数据
    save_path = f"./datasets/{length}/test.txt"
    k = 1
    generate_data(length, k, save_path)

结果：

注意在自己的python代码项目中添加datasets文件夹，dataseta下再设置5 10 15 20 25 30 35的文件夹，之后再设置train.txt dev.txt,test.txt 设置好相对路径。生成完数字序列后，可以查看一下每个txt文件夹中的内容：

> datasets-> 5 10 15 20 25 30 35:

> 如5 -> dev.txt ，test.txt 和 train.txt (其余的文件夹也一样) 其实不设置也行，跑一遍代码之后就会自动生成txt文件：

> 查看15->text. txt文件中的信息，可以看到生成了长度为15个数字序列，每行数字序列后还会带有1个数字标签：

加载数据并进行数据划分

import os
# 加载数据
def load_data(data_path):
    # 加载训练集
    train_examples = []
    train_path = os.path.join(data_path, "train.txt")
    with open(train_path, "r", encoding="utf-8") as f:
        for line in f.readlines():
            # 解析一行数据，将其处理为数字序列seq和标签label
            items = line.strip().split("\t")
            seq = [int(i) for i in items[0].split(" ")]
            label = int(items[1])
            train_examples.append((seq, label))

    # 加载验证集
    dev_examples = []
    dev_path = os.path.join(data_path, "dev.txt")
    with open(dev_path, "r", encoding="utf-8") as f:
        for line in f.readlines():
            # 解析一行数据，将其处理为数字序列seq和标签label
            items = line.strip().split("\t")
            seq = [int(i) for i in items[0].split(" ")]
            label = int(items[1])
            dev_examples.append((seq, label))

    # 加载测试集
    test_examples = []
    test_path = os.path.join(data_path, "test.txt")
    with open(test_path, "r", encoding="utf-8") as f:
        for line in f.readlines():
            # 解析一行数据，将其处理为数字序列seq和标签label
            items = line.strip().split("\t")
            seq = [int(i) for i in items[0].split(" ")]
            label = int(items[1])
            test_examples.append((seq, label))

    return train_examples, dev_examples, test_examples

# 设定加载的数据集的长度
length = 5
# 该长度的数据集的存放目录
data_path = f"./datasets/{length}"
# 加载该数据集
train_examples, dev_examples, test_examples = load_data(data_path)
print("dev example:", dev_examples[:2])
print("训练集数量：", len(train_examples))
print("验证集数量：", len(dev_examples))
print("测试集数量：", len(test_examples))

运行结果：

构造Dataset类

为了方便使用梯度下降法进行优化，构造了DigitSum数据集的Dataset类，函数__getitem__负责根据索引读取数据，并将数据转换为张量。代码及流程实现如下：

from torch.utils.data import Dataset,DataLoader
import torch
class DigitSumDataset(Dataset):
    def __init__(self, data):
        self.data = data

    def __getitem__(self, idx):
        example = self.data[idx]
        seq = torch.tensor(example[0], dtype=torch.int64)
        label = torch.tensor(example[1], dtype=torch.int64)
        return seq, label

    def __len__(self):
        return len(self.data)

四、模型构建

使用SRN模型进行数字加和任务的模型结构为如下图所示：

整个模型由以下几个部分组成：
（1） 嵌入层：将输入的数字序列进行向量化，即将每个数字映射为向量；
（2） SRN 层：接收向量序列，更新循环单元，将最后时刻的隐状态作为整个序列的表示；
（3） 输出层：一个线性层，输出分类的结果.

嵌入层

本任务输入的样本是数字序列，为了更好地表示数字，我们需要将数字映射为一个嵌入（Embedding）向量。嵌入向量中的每个维度均能用来刻画该数字本身的某种特性。由于向量能够表达该数字更多的信息，利用向量进行数字求和任务，可以使得模型具有更强的拟合能力。

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

或者也可以将每个数字表示为10维的one-hot向量，使用矩阵运算得到嵌入表示：

$X=S^{'}E$

其中是 $S^{'}\epsilon \mathbb{R}^{B*L*10}$ 是序列S对应的one-hot表示。

基于索引方式的嵌入层的实现如下：

class Embedding(nn.Module):
    def __init__(self, num_embeddings, embedding_dim):
        super(Embedding, self).__init__()
        self.W = nn.init.xavier_uniform_(torch.empty(num_embeddings, embedding_dim),gain=1.0)

    def forward(self, inputs):
        # 根据索引获取对应词向量
        embs = self.W[inputs]
        return embs


emb_layer = Embedding(10, 5)
inputs = torch.tensor([0, 1, 2, 3])
emb_layer(inputs)

SRN层

简单循环网络的代码实现如下:

import torch
import torch.nn as nn
import torch.nn.functional as F
torch.manual_seed(0)

# SRN模型
class SRN(nn.Module):
    def __init__(self, input_size,  hidden_size, W_attr=None, U_attr=None, b_attr=None):
        super(SRN, self).__init__()
        # 嵌入向量的维度
        self.input_size = input_size
        # 隐状态的维度
        self.hidden_size = hidden_size
        # 定义模型参数W，其shape为 input_size x hidden_size
        if W_attr==None:
            W=torch.zeros(size=[input_size, hidden_size], dtype=torch.float32)
        else:
            W=torch.tensor(W_attr,dtype=torch.float32)
        self.W = torch.nn.Parameter(W)
        # 定义模型参数U，其shape为hidden_size x hidden_size
        if U_attr==None:
            U=torch.zeros(size=[hidden_size, hidden_size], dtype=torch.float32)
        else:
            U=torch.tensor(U_attr,dtype=torch.float32)
        self.U = torch.nn.Parameter(U)
        # 定义模型参数b，其shape为 1 x hidden_size
        if b_attr==None:
            b=torch.zeros(size=[1, hidden_size], dtype=torch.float32)
        else:
            b=torch.tensor(b_attr,dtype=torch.float32)
        self.b = torch.nn.Parameter(b)

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

    # 定义前向计算
    def forward(self, inputs, hidden_state=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 hidden_state is None:
            hidden_state = self.init_state(batch_size)

        # 循环执行RNN计算
        for step in range(seq_len):
            # 获取当前时刻的输入数据step_input, 其shape为 batch_size x input_size
            step_input = inputs[:, step, :]
            # 获取当前时刻的隐状态向量hidden_state, 其shape为 batch_size x hidden_size
            hidden_state = F.tanh(torch.matmul(step_input, self.W) + torch.matmul(hidden_state, self.U) + self.b)
        return hidden_state



## 初始化参数并运行
U_attr = [[0.0, 0.1], [0.1,0.0]]
b_attr = [[0.1, 0.1]]
W_attr=[[0.1, 0.2], [0.1,0.2]]

srn = SRN(2, 2, W_attr=W_attr, U_attr=U_attr, b_attr=b_attr)

inputs = torch.tensor([[[1, 0],[0, 2]]], dtype=torch.float32)
hidden_state = srn(inputs)
print("hidden_state", hidden_state)

运行结果：

PyTorch框架内置了SRN的API torch.nn.RNN：


## 初始化参数并运行
U_attr = [[0.0, 0.1], [0.1,0.0]]
b_attr = [[0.1, 0.1]]
W_attr=[[0.1, 0.2], [0.1,0.2]]

srn = SRN(2, 2, W_attr=W_attr, U_attr=U_attr, b_attr=b_attr)

inputs = torch.tensor([[[1, 0],[0, 2]]], dtype=torch.float32)
hidden_state = srn(inputs)
print("hidden_state", hidden_state)

# 这里创建一个随机数组作为测试数据，数据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
torch_srn = nn.RNN(input_size, hidden_size)
self_srn = SRN(input_size, hidden_size)

self_hidden_state = self_srn(inputs)
torch_outputs, torch_hidden_state = torch_srn(inputs)

print("self_srn hidden_state: ", self_hidden_state.shape)
print("torch_srn outpus:", torch_outputs.shape)
print("torch_srn hidden_state:", torch_hidden_state.shape)

运行结果：

可以看到，自己实现的SRN由于没有考虑多层因素，因此没有层次这个维度，因此其输出shape为[8, 32]。同时由于在以上代码使用PyTorch内置API实例化SRN时，默认定义的是1层的单向SRN，因此其shape为[1, 20, 32]，同时隐状态向量为[1,20, 32].

将自己实现的SRN和PyTorch框架内置的RNN返回的结果进行打印展示：

#这里创建一个随机数组作为测试数据，数据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_srn = nn.RNN(input_size, hidden_size, bias=False)

# 获取torch_srn中的参数，并设置相应的paramAttr,用于初始化SRN
W_attr = torch_srn.weight_ih_l0.T
U_attr = torch_srn.weight_hh_l0.T
self_srn = SRN(input_size, hidden_size, W_attr=W_attr, U_attr=U_attr)

# 进行前向计算，获取隐状态向量，并打印展示
self_hidden_state = self_srn(inputs)
torch_outputs, torch_hidden_state = torch_srn(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, hidden_size = 2, 5, 10, 10
inputs = torch.randn([batch_size, seq_len, input_size])

# 实例化模型
self_srn = SRN(input_size, hidden_size)
torch_srn = nn.RNN(input_size, hidden_size)

# 计算自己实现的SRN运算速度
model_time = 0
for i in range(100):
    strat_time = time.time()
    out = self_srn(inputs)
    if i < 10:
        continue
    end_time = time.time()
    model_time += (end_time - strat_time)
avg_model_time = model_time / 90
print('self_srn speed:', avg_model_time, 's')

# 计算torch内置的SRN运算速度
model_time = 0
for i in range(100):
    strat_time = time.time()
    out = torch_srn(inputs)
    # 预热10次运算，不计入最终速度统计
    if i < 10:
        continue
    end_time = time.time()
    model_time += (end_time - strat_time)
avg_model_time = model_time / 90
print('torch_srn speed:', avg_model_time, 's')

结果：

可以看到自定义算子self_srn的速度相对于torch自带算子torch_srn的速度要慢一些。PyTorch框架实现的SRN的运行效率显著高于自己实现的SRN效率。

线性层

线性层直接使用torch.nn.Linear算子。在定义了每一层的算子之后，我们定义一个数字求和模型Model_RNN4SeqClass，该模型会将嵌入层、SRN层和线性层进行组合，以实现数字求和的功能：

# 基于RNN实现数字预测的模型
class Model_RNN4SeqClass(nn.Module):
    def __init__(self, model, num_digits, input_size, hidden_size, num_classes):
        super(Model_RNN4SeqClass, self).__init__()
        # 传入实例化的RNN层，例如SRN
        self.rnn_model = model
        # 词典大小
        self.num_digits = num_digits
        # 嵌入向量的维度
        self.input_size = input_size
        # 定义Embedding层
        self.embedding = Embedding(num_digits, input_size)
        # 定义线性层
        self.linear = nn.Linear(hidden_size, num_classes)

    def forward(self, inputs):
        # 将数字序列映射为相应向量
        inputs_emb = self.embedding(inputs)
        # 调用RNN模型
        hidden_state = self.rnn_model(inputs_emb)
        # 使用最后一个时刻的状态进行数字预测
        logits = self.linear(hidden_state)
        return logits

# 实例化一个input_size为4， hidden_size为5的SRN
srn = SRN(4, 5)
# 基于srn实例化一个数字预测模型实例
model = Model_RNN4SeqClass(srn, 10, 4, 5, 19)
# 生成一个shape为 2 x 3 的批次数据
inputs = torch.tensor([[1, 2, 3], [2, 3, 4]])
# 进行模型前向预测
logits = model(inputs)
print(logits)

运行结果：

五、模型训练

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

基于RunnerV3类进行训练，只需要指定length便可以加载相应的数据。设置超参数，使用Adam优化器，学习率为 0.001，实例化模型，使用第4.5.4节定义的Accuracy计算准确率。使用Runner进行训练，训练回合数设为500。代码实现如下：

import os
import random
import torch
import numpy as np

# 训练轮次
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 SRN with data of length {length}.")
    # 加载长度为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 = SRN(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 = nn.CrossEntropyLoss()

    # 基于以上组件，实例化Runner
    runner = RunnerV3(model, optimizer, loss_fn, metric)

    # 进行模型训练
    model_save_path = os.path.join(save_dir, f"best_srn_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
srn_runners = {}

注：RunnerV3类与Accuracy类如下：

RunnerV3：

class RunnerV3(object):
    def __init__(self, model, optimizer, loss_fn, metric, **kwargs):
        self.model = model
        self.optimizer = optimizer
        self.loss_fn = loss_fn
        self.metric = metric  # 只用于计算评价指标

        # 记录训练过程中的评价指标变化情况
        self.dev_scores = []

        # 记录训练过程中的损失函数变化情况
        self.train_epoch_losses = []  # 一个epoch记录一次loss
        self.train_step_losses = []  # 一个step记录一次loss
        self.dev_losses = []

        # 记录全局最优指标
        self.best_score = 0

    def train(self, train_loader, dev_loader=None, **kwargs):
        # 将模型切换为训练模式
        self.model.train()

        # 传入训练轮数，如果没有传入值则默认为0
        num_epochs = kwargs.get("num_epochs", 0)
        # 传入log打印频率，如果没有传入值则默认为100
        log_steps = kwargs.get("log_steps", 100)
        # 评价频率
        eval_steps = kwargs.get("eval_steps", 0)

        # 传入模型保存路径，如果没有传入值则默认为"best_model.pdparams"
        save_path = kwargs.get("save_path", "best_model.pdparams")

        custom_print_log = kwargs.get("custom_print_log", None)

        # 训练总的步数
        num_training_steps = num_epochs * len(train_loader)

        if eval_steps:
            if self.metric is None:
                raise RuntimeError('Error: Metric can not be None!')
            if dev_loader is None:
                raise RuntimeError('Error: dev_loader can not be None!')

        # 运行的step数目
        global_step = 0

        # 进行num_epochs轮训练
        for epoch in range(num_epochs):
            # 用于统计训练集的损失
            total_loss = 0
            for step, data in enumerate(train_loader):
                X, y = data
                # 获取模型预测
                logits = self.model(X)
                loss = self.loss_fn(logits, y.long())  # 默认求mean
                total_loss += loss

                # 训练过程中，每个step的loss进行保存
                self.train_step_losses.append((global_step, loss.item()))

                if log_steps and global_step % log_steps == 0:
                    print(
                        f"[Train] epoch: {epoch}/{num_epochs}, step: {global_step}/{num_training_steps}, loss: {loss.item():.5f}")

                # 梯度反向传播，计算每个参数的梯度值
                loss.backward()

                if custom_print_log:
                    custom_print_log(self)

                # 小批量梯度下降进行参数更新
                self.optimizer.step()
                # 梯度归零
                self.optimizer.zero_grad()

                # 判断是否需要评价
                if eval_steps > 0 and global_step > 0 and \
                        (global_step % eval_steps == 0 or global_step == (num_training_steps - 1)):

                    dev_score, dev_loss = self.evaluate(dev_loader, global_step=global_step)
                    print(f"[Evaluate]  dev score: {dev_score:.5f}, dev loss: {dev_loss:.5f}")

                    # 将模型切换为训练模式
                    self.model.train()

                    # 如果当前指标为最优指标，保存该模型
                    if dev_score > self.best_score:
                        self.save_model(save_path)
                        print(
                            f"[Evaluate] best accuracy performence has been updated: {self.best_score:.5f} --> {dev_score:.5f}")
                        self.best_score = dev_score

                global_step += 1

            # 当前epoch 训练loss累计值
            trn_loss = (total_loss / len(train_loader)).item()
            # epoch粒度的训练loss保存
            self.train_epoch_losses.append(trn_loss)

        print("[Train] Training done!")

    # 模型评估阶段，使用'torch.no_grad()'控制不计算和存储梯度
    @torch.no_grad()
    def evaluate(self, dev_loader, **kwargs):
        assert self.metric is not None

        # 将模型设置为评估模式
        self.model.eval()

        global_step = kwargs.get("global_step", -1)

        # 用于统计训练集的损失
        total_loss = 0

        # 重置评价
        self.metric.reset()

        # 遍历验证集每个批次
        for batch_id, data in enumerate(dev_loader):
            X, y = data

            # 计算模型输出
            logits = self.model(X)

            # 计算损失函数
            loss = self.loss_fn(logits, y.long()).item()
            # 累积损失
            total_loss += loss

            # 累积评价
            self.metric.update(logits, y)

        dev_loss = (total_loss / len(dev_loader))
        dev_score = self.metric.accumulate()

        # 记录验证集loss
        if global_step != -1:
            self.dev_losses.append((global_step, dev_loss))
            self.dev_scores.append(dev_score)

        return dev_score, dev_loss

    # 模型评估阶段，使用'torch.no_grad()'控制不计算和存储梯度
    @torch.no_grad()
    def predict(self, x, **kwargs):
        # 将模型设置为评估模式
        self.model.eval()
        # 运行模型前向计算，得到预测值
        logits = self.model(x)
        return logits

    def save_model(self, save_path):
        torch.save(self.model.state_dict(), save_path)

    def load_model(self, model_path):
        state_dict = torch.load(model_path)
        self.model.load_state_dict(state_dict)

Accuracy：

class Accuracy():
    def __init__(self, is_logist=True):
        # 用于统计正确的样本个数
        self.num_correct = 0
        # 用于统计样本的总数
        self.num_count = 0

        self.is_logist = is_logist

    def update(self, outputs, labels):

        # 判断是二分类任务还是多分类任务，shape[1]=1时为二分类任务，shape[1]>1时为多分类任务
        if outputs.shape[1] == 1:  # 二分类
            outputs = torch.squeeze(outputs, dim=-1)
            if self.is_logist:
                # logist判断是否大于0
                preds = torch.tensor((outputs >= 0), dtype=torch.float32)
            else:
                # 如果不是logist，判断每个概率值是否大于0.5，当大于0.5时，类别为1，否则类别为0
                preds = torch.tensor((outputs >= 0.5), dtype=torch.float32)
        else:
            # 多分类时，使用'torch.argmax'计算最大元素索引作为类别
            preds = torch.argmax(outputs, dim=1)

        # 获取本批数据中预测正确的样本个数
        labels = torch.squeeze(labels, dim=-1)
        batch_correct = torch.sum(torch.tensor(preds == labels, dtype=torch.float32)).cpu().numpy()
        batch_count = len(labels)

        # 更新num_correct 和 num_count
        self.num_correct += batch_correct
        self.num_count += batch_count

    def accumulate(self):
        # 使用累计的数据，计算总的指标
        if self.num_count == 0:
            return 0
        return self.num_correct / self.num_count

    def reset(self):
        # 重置正确的数目和总数
        self.num_correct = 0
        self.num_count = 0

    def name(self):
        return "Accuracy"

多组训练

srn_runners = {}

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

(记得在python文件夹处新建一个空的文件夹，命名为chackpoints)

运行结果：

损失曲线展示


import matplotlib.pyplot as plt
def plot_training_loss(runner, fig_name, sample_step):
    plt.figure()
    train_items = runner.train_step_losses[::sample_step]
    train_steps = [x[0] for x in train_items]
    train_losses = [x[1] for x in train_items]
    plt.plot(train_steps, train_losses, color='#e4007f', label="Train loss")

    dev_steps = [x[0] for x in runner.dev_losses]
    dev_losses = [x[1] for x in runner.dev_losses]
    plt.plot(dev_steps, dev_losses, color='#f19ec2', linestyle='--', label="Dev loss")

    # 绘制坐标轴和图例
    plt.ylabel("loss", fontsize='large')
    plt.xlabel("step", fontsize='large')
    plt.legend(loc='upper right', fontsize='x-large')

    plt.savefig(fig_name)
    plt.show()

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

结果：

print出的非可视化结果就不一一截图展示了，不如直接对比可视化结果图形，这样更加直观。

可视化结果：

k=10

k=15

k=20

k=25

k=30

k=35

上面展示了在6个数据集上的损失变化情况，数据集的长度分别为10、15、20、25、30和35. 从输出结果看，随着数据序列长度的增加，虽然训练集损失逐渐逼近于0，但是验证集损失整体趋向越来越大，这表明当序列变长时，SRN模型保持序列长期依赖能力在逐渐变弱，越来越无法学习到有用的知识。

六、模型评价

在模型评价时，加载不同长度的效果最好的模型，然后使用测试集对该模型进行评价，观察模型在测试集上预测的准确度. 同时记录一下不同长度模型在训练过程中，在验证集上最好的效果。代码实现如下。


srn_dev_scores = []
srn_test_scores = []
for length in lengths:
    print(f"Evaluate SRN with data length {length}.")
    runner = srn_runners[length]
    # 加载训练过程中效果最好的模型
    model_path = os.path.join(save_dir, f"best_srn_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)
    srn_test_scores.append(score)
    srn_dev_scores.append(max(runner.dev_scores))

for length, dev_score, test_score in zip(lengths, srn_dev_scores, srn_test_scores):
    print(f"[SRN] length:{length}, dev_score: {dev_score}, test_score: {test_score: .5f}")

运行结果：

接下来，将SRN在不同长度的验证集和测试集数据上的表现，绘制成图片进行观察：

import matplotlib.pyplot as plt

plt.plot(lengths, srn_dev_scores, '-o', color='#e4007f',  label="Dev Accuracy")
plt.plot(lengths, srn_test_scores,'-o', color='#f19ec2', label="Test Accuracy")

#绘制坐标轴和图例
plt.ylabel("accuracy", fontsize='large')
plt.xlabel("sequence length", fontsize='large')
plt.legend(loc='upper right', fontsize='x-large')

fig_name = "./images/6.7.pdf"
plt.savefig(fig_name)
plt.show()

下图展示了SRN模型在不同长度数据训练出来的最好模型在验证集和测试集上的表现。可以看到，随着序列长度的增加，验证集和测试集的准确度整体趋势是降低的，这同样说明SRN模型保持长期依赖的能力在不断降低，运行结果为：

参考《神经网络与深度学习》中的公式(6.50)，改进SRN的循环单元，加入隐状态之间的残差连接，并重复数字求和实验。观察是否可以缓解长程依赖问题？

造成简单循环网络较难建模长程依赖问题的原因有两个：梯度爆炸和梯度消失。一般来讲，循环网络的梯度爆炸问题比较容易解决，一般通过权重衰减或梯度截断可以较好地来避免；对于梯度消失问题，更加有效的方式是改变模型，比如通过长短期记忆网络LSTM来进行缓解。
这道题老师作为NNDL每周作业已经留过了，可以参考我写的这篇博客(当然只是复现了一下书上的推导内容，惭愧地讲，我真的没有弄清楚为什么会得到这样的公式)。

NNDL 作业10 BPTT [HBU]-CSDN博客

根据(6.50)公式可知，只需要将原来的hidden_state加上上一个时刻的hidden_state即可:

只需要修改SRN代码中的一行,修改后的代码为：

 hidden_state =hidden_state + F.tanh(torch.matmul(step_input, self.W) + torch.matmul(hidden_state, self.U) + self.b)

重新运行一下，运行结果为：

可以看到，修改后准确率大幅上涨，有效缓解了长程依赖问题，但这种解决办法仍有两个缺点：

为了解决这两个问题，引入门控机制来进一步改进模型，也就是广为人知的LSTM。

总结

1.实验的初期，正在构建数据集--不同长度的数字预测数据集DigitSum，就遇到了错误：

报错信息显示FileNotFoundError,文件路径错误。这里我们使用的是相对路径，而不是绝对路径，我只想着如何构建序列了，没有考虑到路径的问题。只要添加设置好路径，创建相应的文件，就可以正常运行了。

参考文章：python出现Errno 2] No such file or directory错误解决方法_[errno 2] no such file or directory_木心的博客-CSDN博客

2.解释一下长程依赖问题，我找到了一篇博客，博主比喻的很恰当：
RNN系列之四长期依赖问题-CSDN博客

最后的was与were如何选择是和前面的单复数有关系的，但对于简单的RNN来说，两个词相隔比较远，如何判断是单数还是复数就很关键。长期依赖的根本问题是，经过许多阶段传播后的梯度倾向于消失（大部分情况）或爆炸（很少，但对优化过程影响很大）。

在这次实验中，跑了几遍程序，亲身的感受到了长程依赖问题的存在：数据长度短的情况下，模型训练的性能好一些，但是随着数据集长度增加，模型的性能变差了，准确度下降了。

当添加上隐状态之间的残差链接之后，模型的训练效果好了很多，准确率相较于不添加隐状态之间的残差连接时的模型高了很多。

3.通过这次实验，跟着教材手动敲写了一遍SRN算子的构建，对SRN的结构有了更加清晰的了解。而且这次的实验相较于上一个实验 --- CIFAR10图像分类任务要轻松的多，主要还是体现在训练模型的时长减少了很多：每训练一次模型，付出的时间代价很少(大概跑十分钟左右就可以运行出结果)，有利于调整参数。上一个实验一跑就是五个小时起步，调起参来非常不方便...