Deep Learning(七) 循环神经网络入门

序列模型

---

序列

序列通常被描述为被排成一列的元素对象，是一种最常见的数据结构形式之一。序列模型(Sequence Model)是一类专门用于处理和预测序列数据的模型。这类模型在自然语言处理、音频处理、时间序列分析等领域有着广泛的应用，举例如下

• 人类语言的语音识别
• 股市的每日行情预测
• 大语言模型的文本生成
• 视频、游戏的画面帧生成

我们以离散时间序列为例，来讨论如何处理和预测序列数据。令时间 $t=1,2,\cdots$ 观察到的随机变量为 $x_t$，那么观察 $T$ 次后，我们就得到了 $T$ 个不一定独立的随机变量。这些随机变量的联合分布律可以表示为

$$(x_1,x_2,\cdots,x_T)\sim P(\boldsymbol{x})$$

根据条件概率的乘法公式

$$P(x_ix_j)=P(x_i)P(x_j|x_i)=P(x_j)P(x_i|x_j)$$

因此，可以对 $P(\boldsymbol{x})$ 进行概率展开

$$\begin{aligned}P(\boldsymbol{x})&=P(x_1)P(x_2|x_1)P(x_3|x_1x_2)\cdots P(x_T|x_1x_2\cdots x_{T-1})\\&=\prod_{t=1}^T P(x_t|x_{1}\cdots x_{t-1})\end{aligned}$$

因此，我们需要讨论如何求这 $T$ 个条件概率。这实际上可以看做一个建模问题，我们需要基于不同的假设给出不同的求法。下面是两个常用的模型假设方案。

马尔科夫假设

对于一个长序列数据，我们是不一定要考虑每一个前置变量的。举例而言，我们要预测每日的天气情况，目前已经预测了325天。如果我们把前面所有325天的天气情况都考虑进来，去预测第326天的天气，这显然是不必要的，因为很久之前的天气和现在的天气已经关系不大了。

一般的，假设当前数据只跟过去 $\tau$ 个数据点相关，那么有

$$P(x_T|x_1\cdots x_{T-1})=P(x_T|x_{T-\tau}\cdots x_{T-1})$$

该假设称为马尔科夫假设，此时的预测结果变为

$$P(\boldsymbol{x})=\prod_{t=1}^T P(x_t|x_{t-\tau}\cdots x_{t-1})$$

若假设数据点满足线性关系

$$x_t=c+\phi_1x_{t-1}+\phi_2x_{t-2}+\cdots+\phi_{\tau}x_{t-\tau}+\varepsilon_t$$

称之为 $\tau$ 阶自回归模型(Autoregressive, AR)，其中 $c$ 为偏置项，$\varepsilon_t\sim N(0,\sigma^2)$ 为白噪声项，$\phi_i$ 为参数。

当 $\tau=1$ 时，称之为一阶马尔科夫模型。

潜变量假设

另一种假设方案是潜变量假设，其核心思想可以抽象如下：事物的发展是由一个不可见的内部隐藏状态 $h$ 驱动的。我们观测到的数据，只是这个隐藏状态的外在表现 $x_i$。

假设当前的数据 $\boldsymbol{x}$ 对应潜变量 $h$，现在想要预测下一步的 $\boldsymbol{x}^*$，我们可以这样做

1. 考虑建模问题：通过 $\boldsymbol{x}$ 和其潜变量 $h$ 推导出下一时刻的潜变量 $h^*$。因为我们在假设中提到，潜变量 $h$ 才是整个序列发展的关键因素
2. 基于建模结果，通过预测的潜变量 $h^*$ 和已有 $\boldsymbol{x}$，预测外在表现 $\boldsymbol{x}^*$

这个过程中的潜变量 $h$ 一般需要通过人为建模找出。比如在股市行情预测中，我们观察到的是股价的时间序列 $x_t$。由于股市行和市场心理是紧密相关的，这是一个抽象的、不可预测的潜变量 $h_t$。因此，我们可以通过数学建模的方法整理出 $h_t=g(\boldsymbol{x}_{t-1},h_{t-1})$ ，再根据 $h_t$ 预测下一步的 $\boldsymbol{x}_t$。

这种假设称为潜变量序列模型。

代码实现

我们尝试把马尔科夫假设模型运用到MLP上，我们使用正弦函数和白噪声生成长度为1000的序列。

%matplotlib inline
import torch
from torch import nn
from d2l import torch as d2l

T = 1000  # 总共产生1000个点
time = torch.arange(1, T + 1, dtype=torch.float32)
x = torch.sin(0.01 * time) + torch.normal(0, 0.2, (T,))

想要使用MLP模型，主要难点是把一维的时间序列，变成传统机器学习/深度学习模型所需要的二维矩阵数据（特征矩阵 $\boldsymbol{X}$ 和标签向量 $\boldsymbol{y}$）。根据马尔科夫假设，令 $\tau=4$，那么就把 $\boldsymbol{x}_t=(x_{t-\tau},\cdots,x_{t-1})$ 作为特征 features，$y_t=x_t$ 作为标签 labels。我们采用前600个特征-标签对作为训练数据

tau = 4
# 初始化特征矩阵，一共有996对
features = torch.zeros((T - tau, tau))
for i in range(tau):
    # 按列进行赋值，因为一共有 T-tau 对标签
    features[:, i] = x[i: T - tau + i]
# 标签向量
labels = x[tau:].reshape((-1, 1))

batch_size, n_train = 16, 600
# 只有前n_train个样本用于训练
train_iter = d2l.load_array((features[:n_train], labels[:n_train]),
                            batch_size, is_train=True)

基于MLP部分，搭建一个简单的多层感知机并执行训练

# 初始化网络权重的函数
def init_weights(m):
    if type(m) == nn.Linear:
        nn.init.xavier_uniform_(m.weight)

# 一个简单的多层感知机
def get_net():
    net = nn.Sequential(nn.Linear(4, 10),
                        nn.ReLU(),
                        nn.Linear(10, 1))
    net.apply(init_weights)
    return net

# 平方损失
loss = nn.MSELoss(reduction='none')

def train(net, train_iter, loss, epochs, lr):
    trainer = torch.optim.Adam(net.parameters(), lr)
    for epoch in range(epochs):
        for X, y in train_iter:
            trainer.zero_grad()
            l = loss(net(X), y)
            l.sum().backward()
            trainer.step()
        print(f'epoch {epoch + 1}, '
              f'loss: {d2l.evaluate_loss(net, train_iter, loss):f}')

net = get_net()
train(net, train_iter, loss, 5, 0.01)

我们根据前 600 对训练数据，考察模型的序列预测能力。对于直到 $x_t$ 的观测序列，其在时间 $t+k$ 处的预测输出 $\hat{x}_{t+k}$ 称为 $k$ 步预测。对于单步预测（$k=1$）而言，我们的预测都是基于已有的真实数据 X 做出的下一步预测，而对于 $k>1$ 的情况，我们需要借助自己模型的预测值去预测。

max_steps = 64

features = torch.zeros((T - tau - max_steps + 1, tau + max_steps))

for i in range(tau):
    features[:, i] = x[i: i + T - tau - max_steps + 1]

for i in range(tau, tau + max_steps):
    features[:, i] = net(features[:, i - tau:i]).reshape(-1)

steps = (1, 4, 16, 64)
d2l.plot([time[tau + i - 1: T - max_steps + i] for i in steps],
         [features[:, (tau + i - 1)].detach().numpy() for i in steps], 'time', 'x',
         legend=[f'{i}-step preds' for i in steps], xlim=[5, 1000],
         figsize=(6, 3))

下图给出了不同步长预测的结果，可以发现 $k$ 增大的同时，预测能力变差。

我们提到的 600 对训练数据只是为了训练模型参数，不要和 $k$ 步预测的使用参数搞混。

文本预处理

---

文本序列

对于序列数据处理问题，文本处理是最常见例子之一。一篇文章可以被简单地看作一串单词序列，甚至是一串字符序列。本节中，我们将解析文本的常见预处理步骤，这些步骤通常包括以下四点

1. 将文本作为字符串加载到内存中
2. 将字符串拆分为词元(Token)
3. 建立一个词表，将拆分的词元映射到数字索引
4. 将文本转换为数字索引序列，方便模型操作

我们从 H. G. Well 的科幻文章 The Time Machine 为例，尝试进行文本预处理。这个文章一共包括约三万个单词，我们忽略其中的标点和字母的大写，把文章读取为若干个文本行 lines

import collections
import re
from d2l import torch as d2l

d2l.DATA_HUB['time_machine'] = (d2l.DATA_URL + 'timemachine.txt',
                                '090b5e7e70c295757f55df93cb0a180b9691891a')

# 正则表达式读取
def read_time_machine(): 
    with open(d2l.download('time_machine'), 'r') as f:
        lines = f.readlines()
    return [re.sub('[^A-Za-z]+', ' ', line).strip().lower() for line in lines]

lines = read_time_machine()

Token 化

为了实现 Token 化，我们将文本行列表 lines 作为输入，其每个元素是一个文本序列（如一条文本行），而每个文本序列又被处理为一个词元列表。词元(Token) 是人工智能模型处理文本信息的最小单位，类型为字符串，我们可以粗略理解为一个单词或语言单位。

例如，英文文本序列 I am a boy. 的 Token 列表可以表示为 ['I', 'am', 'a', 'boy', '.'] ；中文文本序列 姬你太美！ 的 Token 列表可以表示为 ['姬', '你', '太', '美', '!']。这里只是举个例子，真实的 Token 的划分遵循一套特殊的算法，后续文章会介绍。

# token 化处理
def tokenize(lines, token='word'):
    if token == 'word':
        return [line.split() for line in lines]
    elif token == 'char':
        return [list(line) for line in lines]
    else:
        print('错误：未知词元类型：' + token)

# 按单词划分
tokens = tokenize(lines)

词表

词元的类型是字符串，而模型需要的输入是数字，因此词元类型不方便模型使用。幸运的是，我们可以借助 Python 等工具创建一个词表(Vocabulary)，将字符串类型的词元映射为从 $0$ 开始的数字索引中。

首先，我们需要将 Token 化后所有的词元进行整合与统计，得到一个语料库(Corpus)，其记录了每个出现词元的频数。我们一般按照词元的频率高低分配对应的数字索引，而频率高的词一般优先分配，频率低的词靠后分配。特别地，对于一些很少出现的词元，我们一般将其剔除，原因有二

• 可以降低复杂性，减少内存占用
• 防止过拟合

当模型的训练文本数据较少时，面对某个新的未知输入词元，我们可能无法匹配一个语料库中的已有索引，此时我们规定一个未知词元 <unk>，将语料库中不存在或者被删除的词元映射为 <unk>。

与此同时，我们可以增加一些特殊功能词元。

• <pad>：填充词元，作为占位无意义词
• <bos>：序列开始词元，作为文本启动词
• <eos>：序列结束词元，作为文本休止词

# 文本词表
class Vocab:
    def __init__(self, tokens=None, min_freq=0, reserved_tokens=None):
        if tokens is None:
            tokens = []
        if reserved_tokens is None:
            reserved_tokens = []
        # 按出现频率排序
        counter = count_corpus(tokens)
        self._token_freqs = sorted(counter.items(), key=lambda x: x[1],
                                   reverse=True)
        # 未知词元的索引为0
        self.idx_to_token = ['<unk>'] + reserved_tokens
        self.token_to_idx = {token: idx
                             for idx, token in enumerate(self.idx_to_token)}
        for token, freq in self._token_freqs:
            if freq < min_freq:
                break
            if token not in self.token_to_idx:
                self.idx_to_token.append(token)
                self.token_to_idx[token] = len(self.idx_to_token) - 1

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

    def __getitem__(self, tokens):
        if not isinstance(tokens, (list, tuple)):
            return self.token_to_idx.get(tokens, self.unk)
        return [self.__getitem__(token) for token in tokens]

    def to_tokens(self, indices):
        if not isinstance(indices, (list, tuple)):
            return self.idx_to_token[indices]
        return [self.idx_to_token[index] for index in indices]

    @property
    def unk(self):  # 未知词元的索引为0
        return 0

    @property
    def token_freqs(self):
        return self._token_freqs

# 统计词元频率
def count_corpus(tokens):
    # 这里的tokens是1D列表或2D列表
    if len(tokens) == 0 or isinstance(tokens[0], list):
        # 将词元列表展平成一个列表
        tokens = [token for line in tokens for token in line]
    return collections.Counter(tokens)

我们尝试获取 The Time Machine 的语料库和词库大小。

def load_corpus_time_machine(max_tokens=-1):  #@save
    lines = read_time_machine()
    # 这里采用字符 Token 化
    tokens = tokenize(lines, 'char')
    vocab = Vocab(tokens)
    # 所以将所有文本行展平到一个列表中
    corpus = [vocab[token] for line in tokens for token in line]
    if max_tokens > 0:
        corpus = corpus[:max_tokens]
    return corpus, vocab

corpus, vocab = load_corpus_time_machine()
print(len(corpus), len(vocab))

这里，代码会自动下载文本文件 timemachine.txt 在如下路径。

Root\
  ├── code\                   
  │    └── TextPre.ipynb
  └── data\
       └── timemachine.txt

语言模型

---

模型目标

对于某个文本序列，其词元分别为 $x_1,x_2,\cdots,x_T$，那么 $x_t(1\leqslant t\leqslant T)$ 可以视作文本序列在时间 $t$ 处的值。因此，给定这样一个文本序列，语言模型(Language Model)要做的就是估计序列的联合概率

$$P(x_1,x_2,\cdots,x_T)=\prod_{t=1}^T P(x_t|x_{t-\tau}\cdots x_{t-1})$$

根据前文有关时间序列的讨论，为了训练语言模型，我们需要计算单词的概率，以及给定前面几个单词后出现某个单词的条件概率。这些概率本质上就是语言模型的参数。假设 $n(x_i)$ 表示词元 $x_i$ 出现的频数，$n(x_i,x_j,\cdots)$ 表示连续词元对 $x_i,x_j,\cdots$ 出现的频数，那么有估计概率

$$\hat{P}(x_2|x_1)=\frac{n(x_1,x_2)}{n(x_1)}$$

N元语法

当序列很长时，由于文本量不够大，通常会导致某个连续多词元对的出现频数 $n(x_i,x_j,\cdots)\leqslant 1$，因此上述估计通常是不准确的。此时可以考虑马尔科夫假设。对于最简单的情况，如果每个词元之间相互独立，那么就有一元语法模型(Unigram)

$$P(x_1,x_2,\cdots,x_T)=P(x_1)P(x_2)\cdots P(x_T)$$

这通常是不成立的，因为文本序列是一种高度依赖前后文的序列。进一步的，假设每个词元之和前一个词元有关，这就是一个二元语法模型(Bigram)。模型可以表示为

$$P(x_1,x_2,\cdots,x_{T})=P(x_1)\prod_{t=2}^T P(x_t|x_{t-1})$$

根据大数定律，当训练数据足够多时有

$$P(x_t|x_{t-1})\approx \frac{n(x_{t-1},x_t)}{n(x_{t-1})}$$

如果假设每个词元和前两个词元有关，便得到一个三元语法模型(Trigram)

$$P(x_1,x_2,\cdots,x_{T})=P(x_1,x_2)\prod_{t=3}^T P(x_t|x_{t-1}x_{t-2})$$

一般的，假设某个词元只和前 $n-1$ 个词元有关，那么该预测模型为 $n$ 元语法模型。但是，当 $n$ 变大时，连续词元组合仍然有可能很少甚至不出现，这就导致模型再次失效。一种常见的策略是执行拉普拉斯平滑，对每个计数加上一个小正数（通常为1）来避免零概率问题出现。

齐普夫定律困境

我们根据前文的 The Time Machine 词表，画出双对数坐标下的词频图。

首先考虑单个词元（一元语法），第 $i$ 个最常用词元的频率 $n_i$ 近似满足

$$n_i\propto \frac{1}{i^{\alpha}}$$

这一统计规律通常记为齐普夫定律(Zipf’s Law)，其表明在自然语言文本中，一个词出现的频率（的幂次）与它在频率表里的排名近似成反比。我们可以打印出 The Time Machine 词表中前十个频数最高的词元如下

[('the', 2261),
 ('i', 1267),
 ('and', 1245),
 ('of', 1155),
 ('a', 816),
 ('to', 695),
 ('was', 552),
 ('in', 541),
 ('that', 443),
 ('my', 440)]

这些词通常称为高频停用词，他们在文本含义上通常简简单单，但是出现的频率是很高的；相反的，对于更多具有直观含义的词语，他们出现的次数可能只有一两次，但是这样的词有很多。这在常规坐标下的频数分布图上体现为长尾分布。

从上图中还可以看出，除了一元语法词，多元单词序列似乎也遵循齐普夫定律。由于 $n$ 元词元组通常很少出现，这使得拉普拉斯平滑也非常不适合语言建模。因此我们需要使用基于深度学习的模型来解决这一问题。

读取长序列

由于序列数据具有连续性，当序列变得太长而不能被模型一次性全部处理时，我们可能希望拆分这样的序列方便模型读取。下面，我们将描述如何借助随机采样(Random Sampling)和顺序分区(Sequential Partitioning)策略实现长序列的读取。

• 在随机采样中，每个样本都是在原始的长序列上任意捕获的子序列。 在迭代过程中，来自两个相邻的、随机的、小批量中的子序列不一定在原始序列上相邻。对于语言建模，目标是基于到目前为止我们看到的词元来预测下一个词元，因此标签是移位了一个词元的原始序列。
• 在顺序分区中，我们保证两个相邻的小批量中的子序列在原始序列上也是相邻的。这种策略在基于小批量的迭代过程中保留了拆分的子序列的顺序，因此称为顺序分区。

循环神经网络

---

隐状态

前面我们提到，$n$ 元语法模型通常不是一个理想的语言模型，因此我们回过头来看第一节中提出的潜变量假设。通常，我们可以基于当前输入 $\boldsymbol{x}_{t}$ 和先前隐状态 $h_{t-1}$ 来计算时间步 $t$ 处的任何时间的隐状态(Hidden State)

$$h_t=f(\boldsymbol{x}_t,h_{t-1})$$

循环神经网络(Recurrent Neural Network，RNN)是具有隐状态的神经网络。在介绍循环神经网络模型之前，我们首先回顾一下多层感知机模型。对于一个单隐藏层的MLP而言，其模型可以表示为

$$\boldsymbol{H}=\sigma(\boldsymbol{XW}_{xh}+\boldsymbol{b}_h)$$

$$\boldsymbol{O}=\boldsymbol{HW}_{hq}+\boldsymbol{b}_q$$

其中参数矩阵 $\boldsymbol{W}_{xh},\boldsymbol{b}_h$ 表示其将输入 $\boldsymbol{X}$ 转化为隐藏层变量 $\boldsymbol{H}$，$\sigma(\cdot)$ 是激活函数，$\boldsymbol{W}_{hq},\boldsymbol{b}_q$ 表示其将隐藏层变量 $\boldsymbol{H}$ 转化为输出层 $\boldsymbol{O}$。对于一个分类问题，我们可以把 $\mathrm{softmax}(\boldsymbol{O})$ 作为输出类别的概率分布。

如果我们要引入隐状态，即对于时间步 $t$ 而言有小批量输入 $\boldsymbol{X}_t\in\mathbb{R}^{m\times n}$，其每一行对应该序列在时间步 $t$ 的一个样本。令该时间步的隐变量为 $\boldsymbol{H}_t\in\mathbb{R}^{m\times h}$，并引入新的权重参数 $\boldsymbol{W}_{hh}\in\mathbb{R}^{h\times h}$ 来描述前一个时间步的 $\boldsymbol{H}_{t-1}$ 对当前时间步的贡献。具体而言有

$$\boldsymbol{H}_t=\sigma(\boldsymbol{X}_t\boldsymbol{W}_{xh}+\boldsymbol{H}_{t-1}\boldsymbol{W}_{hh}+\boldsymbol{b}_h)$$

从上述关系式中看出，隐藏变量 $\boldsymbol{H}_{t}$ 等捕获并保留了序列直到其当前时间步的历史信息，就如当前时间步下神经网络的状态或记忆，因此这样的隐藏变量被称为隐状态(Hidden State)。对于每一个时间步 $t$ 而言，其输出可以表示为

$$\boldsymbol{O}_t=\boldsymbol{H}_t\boldsymbol{W}_{hq}+\boldsymbol{b}_q$$

对于序列模型而言，每一个时间步 $t=1,2,\cdots$ 都应该有对应的输出。值得一提的是，即使在不同的时间步，循环神经网络也总是使用这些模型参数。 因此，循环神经网络的参数开销不会随着时间步的增加而增加。

注意到，在隐状态的循环神经网络中，我们将当前输入和上一步的隐藏变量用加法链接，即 $\boldsymbol{X}_t\boldsymbol{W}_{xh}+\boldsymbol{H}_{t-1}\boldsymbol{W}_{hh}$。事实上，这可以看做输入和隐藏变量的拼接矩阵运算，即有分块矩阵形式

$$\boldsymbol{X}_t\boldsymbol{W}_{xh}+\boldsymbol{H}_{t-1}\boldsymbol{W}_{hh}=\begin{bmatrix}\boldsymbol{X}_t&\boldsymbol{H}_{t-1}\end{bmatrix}\begin{bmatrix}\boldsymbol{W}_{xh}\\\boldsymbol{W}_{hh}\end{bmatrix}$$

这从某种程度上，更直观的展示了循环神经网络和隐状态的融合使用。

字符级语言建模

Bengio 等人首先提出了使用神经网络进行语言建模的概念。在一个循环神经网络模型中，设小批量大小为1，批量中的文本序列为一个单词。为了简化后续部分的训练，我们考虑使用字符级语言模型(Character-level Language Model)，将文本词元化为字符而不是单词。下图演示了如何通过基于字符级语言建模的循环神经网络， 使用当前的和先前的字符预测下一个字符。

实践中，我们使用批量大小为 $m>1$，每个词元由一个 $n$ 维向量表示，因此输入 $\boldsymbol{X}\in\mathbb{R}^{m\times n}$。

困惑度

最后，让我们讨论如何度量语言模型的质量，这将在后续部分中用于评估基于循环神经网络的模型。信息论知识表明，一个长度为 $T$ 序列的交叉熵可以表示为

$$H=-\frac{1}{T}\sum_{t=1}^T \log P(x_t|x_1\cdots x_{t-1})$$

定义其困惑度(Perplexity)为

$$PP=2^{H}$$

困惑度的最好的理解是“下一个词元的实际选择数的调和平均数”，其等价形式为

$$PP=P(x_1,x_2,\cdots,x_T)^{-\frac{1}{T}}$$

• 如果困惑度较低，说明语言模型对序列的预测较为准确
• 如果困惑度较高，说明模型在预测下一个词时存在较大的不确定性，性能较差

RNN代码实现

---

模型定义

在本节，我们将从零开始搭建一个循环神经网络。以 The Time Machine 文本序列为训练集，引入必要的库。

由于目前 d2l 包的新版本删去了可以直接读取的函数 d2l.load_data_time_machine，所以建议直接把下面一大段等价作用的代码复制粘贴到文件的开头。其中涉及的大部分代码内容在文本预处理小节中已经提到过，有需要请往前重新翻阅。

请确保你的同文件夹目录中有 timemachine.txt 文件。

%matplotlib inline
import math
import torch
import random
import re
from torch import nn
from torch.nn import functional as F
from d2l import torch as d2l
from torch.utils import data

def read_time_machine(filepath='timemachine.txt'):
    with open(filepath, 'r', encoding='utf-8') as f:
        lines = f.read().split('\n')
    return [re.sub('[^A-Za-z]+', ' ', line).strip().lower() for line in lines]

def tokenize(lines, token='char'):
    if token == 'char':
        return [list(line) for line in lines]
    elif token == 'word':
        return [line.split() for line in lines]
    else:
        raise ValueError(f'Unknown token type: {token}')

class Vocab:
    def __init__(self, tokens=None, min_freq=0, reserved_tokens=None):
        if tokens is None: tokens = []
        if reserved_tokens is None: reserved_tokens = []
        if tokens and isinstance(tokens[0], list):
            tokens = [t for line in tokens for t in line]
        counter = {}
        for t in tokens:
            counter[t] = counter.get(t, 0) + 1
        self._token_freqs = sorted(counter.items(), key=lambda x: x[1], reverse=True)
        self.idx_to_token = ['<unk>'] + reserved_tokens
        self.token_to_idx = {t: i for i, t in enumerate(self.idx_to_token)}
        for token, freq in self._token_freqs:
            if freq < min_freq:
                break
            if token not in self.token_to_idx:
                self.idx_to_token.append(token)
                self.token_to_idx[token] = len(self.idx_to_token) - 1

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

    def __getitem__(self, tokens):
        if not isinstance(tokens, (list, tuple)):
            return self.token_to_idx.get(tokens, 0)
        return [self.__getitem__(t) for t in tokens]

    def to_tokens(self, indices):
        if not isinstance(indices, (list, tuple)):
            return self.idx_to_token[indices]
        return [self.idx_to_token[i] for i in indices]

def seq_data_iter_random(corpus, batch_size, num_steps):
    corpus = corpus[random.randint(0, num_steps - 1):]
    num_seqs = (len(corpus) - 1) // num_steps
    initial_indices = list(range(0, num_seqs * num_steps, num_steps))
    random.shuffle(initial_indices)
    for i in range(0, len(initial_indices) - batch_size + 1, batch_size):
        batch_idx = initial_indices[i: i + batch_size]
        X = torch.tensor([corpus[j: j + num_steps] for j in batch_idx])
        Y = torch.tensor([corpus[j + 1: j + 1 + num_steps] for j in batch_idx])
        yield X, Y

def seq_data_iter_sequential(corpus, batch_size, num_steps):
    offset = random.randint(0, num_steps)
    num_tokens = ((len(corpus) - offset - 1) // batch_size) * batch_size
    Xs = torch.tensor(corpus[offset: offset + num_tokens])
    Ys = torch.tensor(corpus[offset + 1: offset + 1 + num_tokens])
    Xs, Ys = Xs.reshape(batch_size, -1), Ys.reshape(batch_size, -1)
    num_batches = Xs.shape[1] // num_steps
    for i in range(0, num_steps * num_batches, num_steps):
        X = Xs[:, i: i + num_steps]
        Y = Ys[:, i: i + num_steps]
        yield X, Y

class SeqDataLoader:
    def __init__(self, batch_size, num_steps, use_random_iter, max_tokens):
        if use_random_iter:
            self.data_iter_fn = seq_data_iter_random
        else:
            self.data_iter_fn = seq_data_iter_sequential
        self.corpus, self.vocab = self._load_corpus(max_tokens)
        self.batch_size = batch_size
        self.num_steps = num_steps

    def _load_corpus(self, max_tokens):
        lines = read_time_machine()
        tokens = tokenize(lines, token='char')
        vocab = Vocab(tokens)
        corpus = [vocab[t] for line in tokens for t in line]
        if max_tokens > 0:
            corpus = corpus[:max_tokens]
        return corpus, vocab

    def __iter__(self):
        return self.data_iter_fn(self.corpus, self.batch_size, self.num_steps)
    
def load_data_time_machine(batch_size, num_steps,
                           use_random_iter=False, max_tokens=10000):

    data_iter = SeqDataLoader(batch_size, num_steps, use_random_iter, max_tokens)
    return data_iter, data_iter.vocab

# 读取数据集
lines = read_time_machine()
batch_size, num_steps = 32, 35
# 如果想使用随机采样，在后面加上 use_random_iter=True 
train_iter, vocab = load_data_time_machine(batch_size, num_steps)

为了表示词元，我们使用独热编码的形式。考虑到一个小批量 X 的形状是一个二维张量，第一个维度表示批量大小，第二个维度表示时间步数。为了将每一个时间步的样本放在一起，我们先对 X 取转置，再调整为长度为 28 的独热编码形式。

# 使用独热编码
X = torch.arange(10).reshape((2, 5))
F.one_hot(X.T, 28).shape # 28表示26个小写字母+<unk>+<pad>

# 此时得到的张量大小为 torch.Size([5, 2, 28])

接下来，我们对循环神经网络模型的参数进行初始化。

# 初始化参数
def get_params(vocab_size, num_hiddens, device):
    # 输入和输出的维度都是词表大小
    num_inputs = num_outputs = vocab_size

    def normal(shape):
        return torch.randn(size=shape, device=device) * 0.01

    # 隐藏层参数
    W_xh = normal((num_inputs, num_hiddens))
    W_hh = normal((num_hiddens, num_hiddens))
    b_h = torch.zeros(num_hiddens, device=device)
    # 输出层参数
    W_hq = normal((num_hiddens, num_outputs))
    b_q = torch.zeros(num_outputs, device=device)
    # 附加梯度
    params = [W_xh, W_hh, b_h, W_hq, b_q]
    for param in params:
        param.requires_grad_(True)
    return params

由于循环神经网络在每个时间步都会返回一个隐变量，定义初始化函数 init_rnn_state 和隐变量的更新函数 rnn。这里的隐变量是一个二维张量，第一个维度表示批量大小，第二个维度表示隐藏单元数。这里，我们在隐变量更新中采用 $\mathrm{tanh}$ 激活函数。

# 全0初始化
def init_rnn_state(batch_size, num_hiddens, device):
    return (torch.zeros((batch_size, num_hiddens), device=device), )

# 隐变量计算
def rnn(inputs, state, params):
    # inputs的形状：(时间步数量，批量大小，词表大小)
    W_xh, W_hh, b_h, W_hq, b_q = params
    H, = state
    outputs = []
    # 对时间步数遍历，得到X的形状：(批量大小，词表大小)
    for X in inputs:
        H = torch.tanh(torch.mm(X, W_xh) + torch.mm(H, W_hh) + b_h)
        Y = torch.mm(H, W_hq) + b_q
        outputs.append(Y)
    return torch.cat(outputs, dim=0), (H,)

接着，我们定义类 RNNModelScratch 以封装函数并实例化

class RNNModelScratch:
    def __init__(self, vocab_size, num_hiddens, device,
                 get_params, init_state, forward_fn):
        self.vocab_size, self.num_hiddens = vocab_size, num_hiddens
        self.params = get_params(vocab_size, num_hiddens, device)
        self.init_state, self.forward_fn = init_state, forward_fn

    def __call__(self, X, state):
        X = F.one_hot(X.T, self.vocab_size).type(torch.float32)
        return self.forward_fn(X, state, self.params)

    def begin_state(self, batch_size, device):
        return self.init_state(batch_size, self.num_hiddens, device)

num_hiddens = 512
net = RNNModelScratch(len(vocab), num_hiddens, d2l.try_gpu(), get_params,
                      init_rnn_state, rnn)
state = net.begin_state(X.shape[0], d2l.try_gpu())

预测训练

现在，我们尝试在未训练的情况下进行序列预测。我们首先定义一个字符串 prefix，并定义函数 predict 来预测我们给定的 prefix 的下文。在循环遍历 prefix 中的字符时，我们不断地将隐状态传递到下一个时间步，但是不生成任何输出。这被称为预热期(Warm-up)，因为在此期间模型会自我更新，但不会进行预测。

预热期结束后，隐状态的值通常比刚开始的初始值更适合预测，从而预测字符并输出它们。

def predict(prefix, num_preds, net, vocab, device):
    state = net.begin_state(batch_size=1, device=device)
    outputs = [vocab[prefix[0]]]
    get_input = lambda: torch.tensor([outputs[-1]], device=device).reshape((1, 1))
    for y in prefix[1:]:  # 预热期
        _, state = net(get_input(), state)
        outputs.append(vocab[y])
    for _ in range(num_preds):  # 预测num_preds步
        y, state = net(get_input(), state)
        outputs.append(int(y.argmax(dim=1).reshape(1)))
    return ''.join([vocab.idx_to_token[i] for i in outputs])

predict('time traveller ', 10, net, vocab, d2l.try_gpu())

由于模型未进行训练，可能会得到太奶级别的梦话。

'time traveller hdrntmapym'

我们尝试对这个神经网络运用小批量随梯度下降法。对于长度为 $T$ 的序列，我们在迭代中计算这 $T$ 个时间步上的梯度，将会在反向传播过程中产生长度为 $O(T)$ 的矩阵乘法链。当 $T$ 较大时，它可能导致数值不稳定，例如可能导致梯度爆炸或梯度消失。因此，循环神经网络模型往往需要额外的方式来支持稳定训练。

先来看梯度消失的情况，我们可以使用前些小节介绍的更换激活函数、更换初始化方式、批量归一化等方法，也可以修改循环神经网络的架构（后续日志会介绍）。

当我们的目标函数 $f$ 具有良好的性质时，比如其对常数 $L>0$ 是 Lipschitz 连续的，那么有

$$|f(x)-f(y)|\leqslant L\|x-y\|$$

此时应用梯度下降，对于梯度更新量 $\eta g$ 有

$$|f(x)-f(x-\eta g)|\leqslant L\eta\|g\|$$

这表明我们不会观察到超过 $L\eta\|g\|$ 的变化，这表示了两件事情

1. 当朝着好的方向更新时，它限制了取得进展的速度
2. 当朝着坏的方向更新时，它限制了事情变糟的程度

现在考虑梯度爆炸的问题，当我们的目标函数 $f$ 满足上述性质时（一般都是满足的），我们可以减小学习率 $\eta$ 来防止梯度过大。然而，当学习率过小时，我们的算法是很难收敛的，此时需要采取一个更聪明的策略——梯度裁剪。给定一个超参数 $\theta$，定义裁剪后的梯度 $\hat{g}$ 为

$$\hat{g}\leftarrow \min\left\{1,\frac{\theta}{\|g\|}\right\}g$$

这就保证了裁剪后的梯度 $\hat{g}$ 的范数一定不超过 $\theta$，并且和 $g$ 保持同向。梯度裁剪还有一个值得拥有的副作用， 即限制了任何给定的小批量数据（以及其中任何给定的样本）对参数向量的影响，这赋予了模型一定程度的稳定性。梯度裁剪的代码如下

def grad_clipping(net, theta):
    if isinstance(net, nn.Module):
        params = [p for p in net.parameters() if p.requires_grad]
    else:
        params = net.params
    norm = torch.sqrt(sum(torch.sum((p.grad ** 2)) for p in params))
    if norm > theta:
        for param in params:
            param.grad[:] *= theta / norm

由于循环神经网络和MLP的差别，我们的单epoch训练方式有三个不同之处

• 序列数据的不同采样方法将导致隐状态初始化的差异
  • 使用随机抽样时，因为每个样本都是在随机位置抽样的， 因此需要为每个迭代周期重新初始化隐状态
  • 使用顺序分区时，在每个迭代周期的开始位置初始化隐状态。由于下一个小批量数据中的第 $i$ 个子序列样本与当前第 $i$ 个子序列样本相邻，因此当前小批量数据最后一个样本的隐状态，将用于初始化下一个小批量数据第一个样本的隐状态
• 在更新模型参数之前需要进行梯度裁剪
• 使用困惑度评价模型

# 单 epoch 迭代函数
def train_epoch(net, train_iter, loss, updater, device, use_random_iter):
    state, timer = None, d2l.Timer()
    metric = d2l.Accumulator(2)  # 训练损失之和,词元数量
    for X, Y in train_iter:
        if state is None or use_random_iter:
            # 在第一次迭代或使用随机抽样时初始化state
            state = net.begin_state(batch_size=X.shape[0], device=device)
        else:
            if isinstance(net, nn.Module) and not isinstance(state, tuple):
                # state对于nn.GRU是个张量
                state.detach_()
            else:
                # state对于nn.LSTM或对于我们从零开始实现的模型是个张量
                for s in state:
                    s.detach_()
        y = Y.T.reshape(-1)
        X, y = X.to(device), y.to(device)
        y_hat, state = net(X, state)
        l = loss(y_hat, y.long()).mean()
        if isinstance(updater, torch.optim.Optimizer):
            updater.zero_grad()
            l.backward()
            grad_clipping(net, 1)
            updater.step()
        else:
            l.backward()
            grad_clipping(net, 1)
            # 因为已经调用了mean函数
            updater(batch_size=1)
        metric.add(l * y.numel(), y.numel())
    return math.exp(metric[0] / metric[1]), metric[1] / timer.stop()

将这些组合在一起，就可以进行训练了！

# 训练函数，默认顺序分区采样
def train(net, train_iter, vocab, lr, num_epochs, device,
              use_random_iter=False):
    loss = nn.CrossEntropyLoss()
    animator = d2l.Animator(xlabel='epoch', ylabel='perplexity',
                            legend=['train'], xlim=[10, num_epochs])
    # 初始化
    if isinstance(net, nn.Module):
        updater = torch.optim.SGD(net.parameters(), lr)
    else:
        updater = lambda batch_size: d2l.sgd(net.params, lr, batch_size)
    predict = lambda prefix: predict_prefix(prefix, 50, net, vocab, device) # 可调节预测步数
    # 训练和预测
    for epoch in range(num_epochs):
        ppl, speed = train_epoch(
            net, train_iter, loss, updater, device, use_random_iter)
        if (epoch + 1) % 10 == 0:
            print(predict('time traveller')) # 可调节 prefix
            animator.add(epoch + 1, [ppl])
    print(f'困惑度 {ppl:.1f}, {speed:.1f} 词元/秒 {str(device)}')
    print(predict('time traveller'))
    print(predict('traveller'))

# 设置较大超参数以加速收敛
num_epochs, lr = 500, 1
train(net, train_iter, vocab, lr, num_epochs, d2l.try_gpu())

课程提供的训练结果如下。

困惑度 1.0, 67212.6 词元/秒 cuda:0
time traveller for so it will be convenient to speak of himwas e
travelleryou can show black is white by argument said filby

每个人得到的预测结果都会不一样，如果想尝试随机采样，创建新的 RNNModelScratch 模型实例并在 train 中修改 use_random_iter=True 即可。限于本训练数据集过小，其训练结果大多是单词组合的形式，较难组合成一段有意义的话。

通过时间反向传播

---

梯度分析

本节将更深入地探讨序列模型反向传播的细节，以及相关的数学原理。前些日志中，我们探讨了多层感知机中的前向传播和反向传播步骤。下面，我们从一个描述循环神经网络工作原理的简化模型开始，此模型忽略了隐状态的特性及其更新方式的细节，以避免符号混淆。

将时间步 $t$ 的隐状态表示为 $h_t$，输入表示为 $x_t$。输出表示为 $o_t$。分别使用 $w_h,w_o$ 表示隐藏层和输出层的权重。于是每个时间步的隐状态和输出可以写为

$$h_t=f(x_t,h_{t-1},w_h)$$

$$o_t=g(h_t,w_o)$$

这样就得到了一个三元组链

$$\{\cdots,(x_{t-1},h_{t-1},o_{t-1}),(x_t,h_t,o_t),\cdots\}$$

前向传播是相当简单的，我们一次一个时间步的遍历所有三元组 $(x_t,h_t,o_t)$，然后通过某个目标函数 $L$ 评估 $T$ 时间步内输出 $o_t$ 和标签 $y_t$ 之间的差别

$$L(x_1,\cdots,x_T;y_1,\cdots,y_T;w_h,w_o)=\frac{1}{T}\sum_{t=1}^T l(y_t,o_t)$$

对于反向传播而言，问题就有点棘手了。按照链式法则计算 $L$ 关于参数 $w_h$ 的梯度时有

$$\frac{\partial L}{\partial w_h}=\frac{1}{T}\sum_{t=1}^T \frac{\partial l(y_t,o_t)}{\partial o_t}\frac{\partial g(h_t,w_o)}{\partial h_t}\frac{\partial h_t}{\partial w_h}$$

前两项因子是容易求得，第三项则是此问题的核心。由于我们需要循环的计算参数 $w_t$ 对隐藏层 $h_t$ 的影响，因此 $h_t$ 的计算既依赖于 $w_h$ 又依赖于 $h_{t-1}$，且 $h_{t-1}$ 的计算依赖于 $w_h$。按照链式法则有

$$\frac{\partial h_t}{\partial w_h}=\frac{\partial f(x_t,h_{t-1},w_h)}{\partial w_h}+\frac{\partial f(x_t,h_{t-1},w_h)}{\partial h_{t-1}}\frac{\partial h_{t-1}}{\partial w_h}$$

上式可以看做某数列递推式 $a_t=b_t+c_ta_{t-1}$，不停地对等式右侧的 $a_{t-1}$ 进行替换得到

$$a_t=b_t+\sum_{i=1}^{t-1}\left(\prod_{j=i+1}^t c_j\right)b_i$$

因此有

$$\frac{\partial h_t}{\partial w_h}=\frac{\partial f(x_t,h_{t-1},w_h)}{\partial w_h}+\sum_{i=1}^{t-1}\left(\prod_{j=i+1}^{t}\frac{\partial f(x_j,h_{j-1},w_h)}{\partial h_{j-1}}\right)\frac{\partial f(x_i,h_{i-1},w_h)}{\partial w_h}$$

这样就可以递归的计算梯度了。

截断计算

显然，想要完全按照上式计算是不显示的。这主要归结于以下几点

• 计算量过于巨大，计算非常缓慢
• 可能存在梯度爆炸现象
• 初始条件的微小变化就可能会对结果产生巨大的影响，影响泛化能力

一个可行的方案是，我们在求和中只考虑最近的 $\tau$ 项，这也符合马尔科夫假设。在实践中，这种方式工作得很好。它通常被称为截断的通过时间反向传播。这样做导致该模型主要侧重于短期影响，而不是长期影响。这在现实中是可取的，因为它会将估计值偏向更简单和更稳定的模型。

另外一种截断方法则引入了随机性，通过随机阶段减少计算复杂度。遗憾的是，虽然随机截断在理论上具有吸引力，但很可能是由于多种因素在实践中并不比常规截断更好。首先，在对过去若干个时间步经过反向传播后，观测结果足以捕获实际的依赖关系。其次，增加的方差抵消了时间步数越多梯度越精确的事实。第三，我们真正想要的是只有短范围交互的模型。因此，模型需要的正是截断的通过时间反向传播方法所具备的轻度正则化效果。

参考网站

课程主页

沐神伟大

课程教材

无需多言

PyTorch安装

PyTorch安装教程