现在我们已经介绍了 RNN，在下一节中，我们将从头开始实现一个简单的 RNN 示例以提高我们的知识。

RNN 实施与训练

在上一节中，我们简要讨论了 RNN 是什么以及它们 可以 解决哪些问题。让我们通过一个非常简单的玩具示例来深入了解 RNN 的细节以及如何训练它：按序列计数。

在这个问题中，我们将教一个基本的 RNN 如何 计算 输入中 1 的数量，然后在序列末尾输出结果。这是我们在上一节中定义的多对一关系的示例。

我们将使用 Python（无 DL 库）和 NumPy 来实现这个示例。输入输出示例如下：

In: (0, 0, 0, 0, 1, 0, 1, 0, 1, 0) 
Out: 3

我们将使用的 RNN 如下图所示：

该网络将只有两个参数：输入权重U和递归权重W。输出权重V设置为 1 ，因此我们只需读出最后一个状态作为输出y。

在继续之前，让我们添加一些代码以便我们的示例可以执行。我们将导入并定义我们的训练和数据，和标签，。是二维的，因为第一维代表小批量中的样本。为简单起见，我们将使用带有单个样本的小批量： numpy xy x

import numpy as np

# The first dimension represents the mini-batch
x = np.array([[0, 0, 0, 0, 1, 0, 1, 0, 1, 0]])

y = np.array([3])

这个网络定义的递归关系是。请注意，这是一个线性模型，因为我们在这个公式中没有应用非线性函数。我们可以如下实现一个递归关系 ：

def step(s, x, U, W):
   return x * U + s * W

状态s t和权重W和U是单个标量值。一个很好的解决方案是只获得整个序列的输入 总和 。如果我们设置U=1，那么无论何时收到输入，我们都会得到它的完整值。如果我们设置W=1，那么我们将累积的值将永远不会衰减。因此，对于此示例，我们将获得所需的输出：3。

尽管如此，让我们使用这个简单的例子来网络化这个神经网络的训练和实现。正如我们将在本节的其余部分中看到的那样，这将很有趣。首先，让我们看看如何通过反向传播得到这个结果。

时间反向传播

时间反向传播是我们用来 训练 循环网络的典型算法（时间反向传播：它的作用和方法，http://axon.cs.byu.edu/~martinez/classes/678/Papers/Werbos_BPTT。 .pdf )。顾名思义，它基于我们在第 1 章“神经网络的基本要素”中讨论的反向传播算法。

常规反向传播和随时间反向传播的主要区别在于，循环网络是随时间展开一定数量的时间步长（如上图所示）。一旦展开完成，我们最终会得到一个与常规多层前馈网络非常相似的模型，也就是说，该网络的一个隐藏层代表时间的一个步骤。唯一的区别是每一层都有多个输入：前一个状态s t-1和当前输入x t。参数U和W在所有隐藏层之间共享。

前向传递沿序列展开 RNN，并为每个步骤构建一个状态堆栈。在下面的代码块中，我们可以看到前向传递的实现，它为每个循环步骤和批处理中的每个样本返回激活值s：

def forward(x, U, W):
    # Number of samples in the mini-batch
    number_of_samples = len(x)

    # Length of each sample
    sequence_length = len(x[0])

    # Initialize the state activation for each sample along the sequence
    s = np.zeros((number_of_samples, sequence_length + 1))

    # Update the states over the sequence
    for t in range(0, sequence_length):
        s[:, t + 1] = step(s[:, t], x[:, t], U, W)  # step function

    return s

现在我们有了前向步骤和损失函数，我们可以定义梯度如何向后传播。由于展开的 RNN 相当于一个常规的前馈网络，我们可以使用我们在第 1 章“神经网络的基本要素”中介绍的反向传播链式法则。

因为权重W 和U是跨层共享的，所以我们将累积每个循环步骤的误差导数，最后，我们将使用累积值更新权重。

首先，我们需要得到输出相对于损失函数 ( ∂J/∂s )的梯度s t 。一旦我们拥有它，我们将通过我们在向前步骤中构建的活动堆栈向后传播它。这种向后传递将活动从堆栈中弹出，以在每个时间步累积它们的误差导数。通过网络传播这个梯度的递归关系可以写成如下（链式法则）：

这里，J 是损失函数。

权重U和W的梯度累积如下：

以下是反向传递的实现：

1.和的梯度分别在和中累积： U W gU gW

def backward(x, s, y, W):
    sequence_length = len(x[0])

    # The network output is just the last activation of sequence
    s_t = s[:, -1]

    # Compute the gradient of the output w.r.t. MSE loss function 
      at final state
    gS = 2 * (s_t - y)

    # Set the gradient accumulations to 0
    gU, gW = 0, 0

    # Accumulate gradients backwards
    for k in range(sequence_length, 0, -1):
        # Compute the parameter gradients and accumulate the
          results
        gU += np.sum(gS * x[:, k - 1])
        gW += np.sum(gS * s[:, k - 1])

        # Compute the gradient at the output of the previous layer
        gS = gS * W

    return gU, gW

2.我们现在可以尝试使用梯度下降来 优化 我们的网络。我们在函数的帮助下计算gradients（使用均方误差）backward并使用它们来更新weights值：

def train(x, y, epochs, learning_rate=0.0005):
    """Train the network"""

    # Set initial parameters
    weights = (-2, 0) # (U, W)

    # Accumulate the losses and their respective weights
    losses = list()
    gradients_u = list()
    gradients_w = list()

    # Perform iterative gradient descent
    for i in range(epochs):
        # Perform forward and backward pass to get the gradients
        s = forward(x, weights[0], weights[1])

        # Compute the loss
        loss = (y[0] - s[-1, -1]) ** 2

        # Store the loss and weights values for later display
        losses.append(loss)

        gradients = backward(x, s, y, weights[1])
        gradients_u.append(gradients[0])
        gradients_w.append(gradients[1])

        # Update each parameter `p` by p = p - (gradient *
          learning_rate).
        # `gp` is the gradient of parameter `p`
        weights = tuple((p - gp * learning_rate) for p, gp in
        zip(weights, gradients))

    print(weights)

    return np.array(losses), np.array(gradients_u),
    np.array(gradients_w)

3.接下来，我们将实现相关的plot_training函数，它显示loss函数和每个权重在 epoch 上的梯度：

def plot_training(losses, gradients_u, gradients_w):
    import matplotlib.pyplot as plt

    # remove nan and inf values
    losses = losses[~np.isnan(losses)][:-1]
    gradients_u = gradients_u[~np.isnan(gradients_u)][:-1]
    gradients_w = gradients_w[~np.isnan(gradients_w)][:-1]

    # plot the weights U and W
    fig, ax1 = plt.subplots(figsize=(5, 3.4))

    ax1.set_ylim(-3, 20)
    ax1.set_xlabel('epochs')
    ax1.plot(gradients_u, label='grad U', color='blue',
    linestyle=':')
    ax1.plot(gradients_w, label='grad W', color='red', linestyle='--
    ')
    ax1.legend(loc='upper left')

    # instantiate a second axis that shares the same x-axis
    # plot the loss on the second axis
    ax2 = ax1.twinx()

    # uncomment to plot exploding gradients
    ax2.set_ylim(-3, 10)
    ax2.plot(losses, label='Loss', color='green')
    ax2.tick_params(axis='y', labelcolor='green')
    ax2.legend(loc='upper right')

    fig.tight_layout()

    plt.show()

4.最后，我们可以 运行 这段代码：

losses, gradients_u, gradients_w = train(x, y, epochs=150)
plot_training(losses, gradients_u, gradients_w)

前面的代码生成下 图：

现在我们已经了解了反向传播，让我们讨论熟悉的梯度消失和爆炸问题如何影响它。

消失和爆炸梯度

但是，前面的示例 有 一个问题。让我们以更长的序列运行训练过程：

x = np.array([[0, 0, 0, 0, 1, 0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1, 0, 1, 0, 0, 0, 0, 0, 1, 0, 1, 0, 1, 0]])

y = np.array([12])

losses, gradients_u, gradients_w = train(x, y, epochs=150)
plot_training(losses, gradients_u, gradients_w)

输出如下：

Sum of ones RNN from scratch
chapter07-rnn/simple_rnn.py:5: RuntimeWarning: overflow encountered in multiply
  return x * U + s * W
chapter07-rnn/simple_rnn.py:40: RuntimeWarning: invalid value encountered in multiply
  gU += np.sum(gS * x[:, k - 1])
chapter07-rnn/simple_rnn.py:41: RuntimeWarning: invalid value encountered in multiply
  gW += np.sum(gS * s[:, k - 1])
(nan, nan)

这些警告的原因是最终参数U和W以非数字( NaN )结尾。为了正确显示梯度，我们需要将函数中梯度轴的比例从更改为，以及将损失轴的比例从更改为。 plot_trainingax1.set_ylim(-3, 20)ax1.set_ylim(-3, 600)ax2.set_ylim(-3, 10)ax2.set_ylim(-3, 200)

现在，程序将生成新的损失和梯度的下图：

在初始时期，梯度缓慢增加，类似于它们在较短序列中增加的方式。然而，当它们到达 epoch 23（确切的 epoch 并不重要）时，梯度变得如此之大，以至于超出了float变量的范围并变为 NaN（如图中的跳跃所示）。这个问题被称为梯度爆炸。我们可以在常规前馈神经网络中偶然发现爆炸梯度，但在 RNN 中尤其明显。为了理解为什么，让我们回顾一下我们在通过时间的反向传播部分定义的两个连续序列步骤的循环梯度传播链规则：

根据序列的长度，展开的 RNN 可以比常规网络更深。同时， RNN的权重W在所有步骤中共享。因此，我们可以推广这个公式来计算序列的两个非连续步骤之间的梯度。因为W是共享的，所以方程形成一个几何级数：

        ​​​​​​​   

在我们简单的线性 RNN 中，如果|W| ，梯度呈指数增长。> 1（爆炸梯度），其中W是单个标量权重，例如，W=1.5 上的 50 个时间步是W 50 ≈ 637621500。如果|W|则梯度呈指数收缩 <1（梯度消失），例如，W=0.6上的 10 个时间步是W 20 = 0.00097 。如果权重参数W是矩阵而不是标量，则此爆炸或消失梯度与W的最大特征值 ( ρ ) 有关 （也称为光谱半径）。ρ < 1足以使梯度消失，而ρ > 1则它们必须爆炸。

我们在第 1 章“神经网络的基本要素”中首次提到的梯度消失问题在 RNN 中还有另一个更微妙的影响。梯度随着步数呈指数衰减，直到在早期状态下变得非常小。实际上，它们被最近时间步的更大梯度所掩盖，并且网络保留这些早期状态历史的能力消失了。这个问题更难检测，因为训练仍然有效并且网络将产生有效的输出（与爆炸梯度不同）。它只是无法学习长期依赖关系。

现在，我们熟悉了 RNN 的一些问题。这些知识对我们很有帮助，因为在下一节中，我们将讨论如何在特殊类型的 RNN 的帮助下解决这些问题。

引入长短期记忆

Hochreiter 和 Schmidhuber 广泛研究了梯度消失和爆炸的问题，并提出了一种称为长短期记忆（LSTM，https: //www.bioinf.jku.at/publications/older/2604.pdf ）的解决方案。由于特制的记忆单元，LSTM 可以处理长期依赖关系。事实上，它们工作得非常好，以至于目前在各种问题上训练 RNN 的大部分成就都归功于 LSTM 的使用。在本节中，我们将探讨此存储单元的工作原理以及它如何解决梯度消失问题。

LSTM 的关键思想是单元状态c t（除了隐藏的 RNN 状态h t），其中信息只能显式写入或删除，以便在没有外部干扰的情况下保持状态不变。细胞状态只能通过特定的门来修改，这是让信息通过的一种方式。这些门由 sigmoid 函数和元素乘法组成。因为 sigmoid 只输出 0 和 1 之间的值，所以乘法只能减少通过门的值。典型的 LSTM 由三个门组成：遗忘门、输入门和输出门。单元状态、输入和输出都是向量，因此 LSTM 可以在每个时间步保存不同信息块的组合。

下面是一个 LSTM 单元的示意图：

import math
import typing

import torch

class LSTMCell(torch.nn.Module):

    def __init__(self, input_size: int, hidden_size: int):
        """
        :param input_size: input vector size
        :param hidden_size: cell state vector size
        """

        super(LSTMCell, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size

        # combine all gates in a single matrix multiplication
        self.x_fc = torch.nn.Linear(input_size, 4 * hidden_size)
        self.h_fc = torch.nn.Linear(hidden_size, 4 * hidden_size)

        self.reset_parameters()

def reset_parameters(self):
    """Xavier initialization """
    size = math.sqrt(3.0 / self.hidden_size)
    for weight in self.parameters():
        weight.data.uniform_(-size, size)

def forward(self,
            x_t: torch.Tensor,
            hidden: typing.Tuple[torch.Tensor, torch.Tensor] =      (None, None)) \
        -> typing.Tuple[torch.Tensor, torch.Tensor]:
    h_t_1, c_t_1 = hidden # t_1 is equivalent to t-1

    # in case of more than 2-dimensional input
    # flatten the tensor (similar to numpy.reshape)
    x_t = x_t.view(-1, x_t.size(1))
    h_t_1 = h_t_1.view(-1, h_t_1.size(1))
    c_t_1 = c_t_1.view(-1, c_t_1.size(1))

gates = self.x_fc(x_t) + self.h_fc(h_t_1)

i_t, f_t, candidate_c_t, o_t = gates.chunk(4, 1)

i_t, f_t, candidate_c_t, o_t = \
    i_t.sigmoid(), f_t.sigmoid(), candidate_c_t.tanh(), o_t.sigmoid()

c_t = torch.mul(f_t, c_t_1) + torch.mul(i_t, candidate_c_t)

h_t = torch.mul(o_t, torch.tanh(c_t))
return h_t, c_t

class LSTMModel(torch.nn.Module):
    def __init__(self, input_dim, hidden_size, output_dim):
        super(LSTMModel, self).__init__()
        self.hidden_size = hidden_size

        # Our own LSTM implementation
        self.lstm = LSTMCell(input_dim, hidden_size)

        # Fully connected output layer
        self.fc = torch.nn.Linear(hidden_size, output_dim)

    def forward(self, x):
        # Start with empty network output and cell state to initialize the sequence
        c_t = torch.zeros((x.size(0), self.hidden_size)).to(x.device)
        h_t = torch.zeros((x.size(0), self.hidden_size)).to(x.device)

        # Iterate over all sequence elements across all sequences of the mini-batch
        for seq in range(x.size(1)):
            h_t, c_t = self.lstm(x[:, seq, :], (h_t, c_t))

        # Final output layer
        return self.fc(h_t)

train = generate_dataset(SEQUENCE_LENGTH, TRAINING_SAMPLES)
train_loader = torch.utils.data.DataLoader(train, batch_size=BATCH_SIZE, shuffle=True)

test = generate_dataset(SEQUENCE_LENGTH, TEST_SAMPLES)
test_loader = torch.utils.data.DataLoader(test, batch_size=BATCH_SIZE, shuffle=True)

model = LSTMModel(input_size=1, hidden_size=HIDDEN_UNITS, output_size=1)

loss_function = torch.nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())

for epoch in range(EPOCHS):
    print('Epoch {}/{}'.format(epoch + 1, EPOCHS))

    train_model(model, loss_function, optimizer, train_loader)
    test_model(model, loss_function, test_loader)

import math
import torch

class GRUCell(torch.nn.Module):

    def __init__(self, input_size: int, hidden_size: int):
        """
        :param input_size: 输入向量大小
        :param hidden_​​size: 单元状态向量大小
        """

        super(GRUCell, self).__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size

        # x 重置门 r 
        self.x_r_fc = torch.nn.Linear(input_size, hidden_size)

        # x 更新门 z 
        self.x_z_fc = torch.nn.Linear(input_size, hidden_size)

        # x 到候选状态 h'(t)
        self.x_h_fc = torch.nn.Linear(input_size, hidden_size)

        # 网络输出/状态 h(t-1) 重置门 r
        self.h_r_fc = torch.nn.Linear(hidden_size, hidden_size)

        # 网络输出/状态 h(t-1) 更新门 z 
        self.h_z_fc = torch.nn.Linear(hidden_size, hidden_size)

        # 网络状态 h(t- 1) 通过复位门 r 向候选状态 h(t) 
        self.hr_h_fc = torch.nn.Linear(hidden_size, hidden_size)

def forward(self,
            x_t: torch.Tensor,
            h_t_1: torch.Tensor = None) \
        -> torch.Tensor:

    # 计算更新门向量
    z_t = torch.sigmoid(self.x_z_fc(x_t) + self.h_z_fc(h_t_1))

    # 计算复位门向量
    r_t = torch.sigmoid(self.x_r_fc(x_t) + self.h_r_fc(h_t_1))

candidate_h_t = torch.tanh(self.x_h_fc(x_t) + self.hr_h_fc(torch.mul(r_t, h_t_1)))

h_t = torch.mul(z_t, h_t_1) + torch.mul(1 - z_t, candidate_h_t)

import torch
import torchtext

TEXT = torchtext.data.Field(
    tokenize='spacy',  # 使用 SpaCy 分词器
    lower=True,  # 将所有字母转换为小写
    include_lengths=True,  # 包括电影评论的长度
)

LABEL = torchtext.data.LabelField（dtype =torch.float）

train, test = torchtext.datasets.IMDB.splits(TEXT, LABEL)

TEXT.build_vocab(train, vectors=torchtext.vocab.GloVe(name='6B', dim=100))
LABEL.build_vocab(train)

train_iter, test_iter = torchtext.data.BucketIterator.splits(
    (train, test), sort_within_batch=True, batch_size=64, device=device)

class LSTMModel(torch.nn.Module):
    def __init__(self, vocab_size, embedding_size, hidden_size, output_size, pad_idx):
        super().__init__()

        # 嵌入字段
        self.embedding=torch.nn.Embedding(num_embeddings=vocab_size,
        embedding_dim=embedding_size,padding_idx=pad_idx)

        # LSTM cell
        self.rnn = torch.nn.LSTM(input_size=embedding_size,
        hidden_size=hidden_size)

        # 全连接输出
        self.fc = torch.nn.Linear(hidden_size, output_size)

    def forward(self, text_sequence, text_lengths):
        # 提取嵌入向量
        embeddings = self.embedding(text_sequence)

        # 将序列填充为相等长度
        packed_sequence =torch.nn.utils.rnn.pack_padded_sequence
        (embeddings, text_lengths)

        packed_output, (hidden, cell) = self.rnn(packed_sequence)

        return self.fc(hidden)


model = LSTMModel(vocab_size=len(TEXT.vocab),
                  embedding_size=EMBEDDING_SIZE,
                  hidden_size=HIDDEN_SIZE,
                  output_size=1,
                  pad_idx=TEXT.vocab.stoi[TEXT.pad_token])

model.embedding.weight.data.copy_(TEXT.vocab.vectors)

model.embedding.weight.data[TEXT.vocab.stoi[TEXT.unk_token]] = torch.zeros(EMBEDDING_SIZE) 
model.embedding.weight.data[TEXT.vocab.stoi[TEXT.pad_token]] = torch.zeros( EMBEDDING_SIZE)

optimizer = torch.optim.Adam(model.parameters())
loss_function = torch.nn.BCEWithLogitsLoss().to(device)

model = model.to(device)

for epoch in range(5):
    print(f"Epoch {epoch + 1}/5")
    train_model(model, loss_function, optimizer, train_iter)
    test_model(model, loss_function, test_iter)