6.4 循环神经网络的从零开始实现

发表于 更新于 分类于 ，

读取周杰伦专辑歌词数据集：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import time
import math
import numpy as np
import torch
from torch import nn, optim
import torch.nn.functional as F
import zipfile
import random

# import d2lzh_pytorch as d2l
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
print(device)

def load_data_jay_lyrics():
    """加载周杰伦歌词数据集"""
    with zipfile.ZipFile('机器学习/data/jaychou_lyrics.txt.zip') as zin:
        with zin.open('jaychou_lyrics.txt') as f:
            corpus_chars = f.read().decode('utf-8')
    corpus_chars = corpus_chars.replace('\n', ' ').replace('\r', ' ')
    # corpus_chars = corpus_chars[0:10000]
    idx_to_char = list(set(corpus_chars))
    char_to_idx = dict([(char, i) for i, char in enumerate(idx_to_char)])
    vocab_size = len(char_to_idx)
    corpus_indices = [char_to_idx[char] for char in corpus_chars]
    return corpus_indices, char_to_idx, idx_to_char, vocab_size

(corpus_indices, char_to_idx, idx_to_char, vocab_size) = load_data_jay_lyrics()

one-hot向量

假设词典中不同字符的数量为\(N\)（即词典大小vocab_size），每个字符已经同一个从0到\(N-1\)的连续整数值索引一一对应。如果一个字符的索引是整数\(i\), 那么创建一个全0的长为\(N\)的向量，并将其位置为\(i\)的元素设成1。该向量就是对原字符的one-hot向量。

1
2
3
4
5
6
def one_hot(x, n_class, dtype=torch.float32): 
    # X shape: (batch), output shape: (batch, n_class)
    x = x.long()
    res = torch.zeros(x.shape[0], n_class, dtype=dtype, device=x.device)
    res.scatter_(1, x.view(-1, 1), 1)
    return res

每次采样的小批量的形状是(批量大小, 时间步数)。下面的函数将这样的小批量变换成数个可以输入进网络的形状为(批量大小, 词典大小)的矩阵，矩阵个数等于时间步数。也就是说，时间步\(t\)的输入为\(\boldsymbol{X}_t \in \mathbb{R}^{n \times d}\)，其中\(n\)为批量大小，\(d\)为输入个数，即one-hot向量长度（词典大小）。

1
2
3
def to_onehot(X, n_class):  
    # X shape: (batch, seq_len), output: seq_len elements of (batch, n_class)
    return [one_hot(X[:, i], n_class) for i in range(X.shape[1])]

初始化模型参数

隐藏单元个数 num_hiddens是一个超参数。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
num_inputs, num_hiddens, num_outputs = vocab_size, 256, vocab_size
print('will use', device)

def get_params():
    def _one(shape):
        ts = torch.tensor(np.random.normal(0, 0.01, size=shape), device=device, dtype=torch.float32)
        return torch.nn.Parameter(ts, requires_grad=True)

    # 隐藏层参数
    W_xh = _one((num_inputs, num_hiddens))
    W_hh = _one((num_hiddens, num_hiddens))
    b_h = torch.nn.Parameter(torch.zeros(num_hiddens, device=device, requires_grad=True))
    # 输出层参数
    W_hq = _one((num_hiddens, num_outputs))
    b_q = torch.nn.Parameter(torch.zeros(num_outputs, device=device, requires_grad=True))
    return nn.ParameterList([W_xh, W_hh, b_h, W_hq, b_q])

定义模型

根据循环神经网络的计算表达式实现该模型。首先定义init_rnn_state函数来返回初始化的隐藏状态。它返回由一个形状为(批量大小, 隐藏单元个数)的值为0的NDArray组成的元组。使用元组是为了更便于处理隐藏状态含有多个NDArray的情况。

1
2
def init_rnn_state(batch_size, num_hiddens, device):
    return (torch.zeros((batch_size, num_hiddens), device=device), )

下面的rnn函数定义了在一个时间步里如何计算隐藏状态和输出。这里的激活函数使用了tanh函数。

1
2
3
4
5
6
7
8
9
10
11
def rnn(inputs, state, params):
    # inputs和outputs皆为num_steps个形状为(batch_size, vocab_size)的矩阵
    W_xh, W_hh, b_h, W_hq, b_q = params
    H, = state
    outputs = []
    for X in inputs: # 每次同时处理所有批量的同一个时间步
        # print(X)
        H = torch.tanh(torch.matmul(X, W_xh) + torch.matmul(H, W_hh) + b_h)
        Y = torch.matmul(H, W_hq) + b_q
        outputs.append(Y)
    return outputs, (H,)

定义预测函数

以下函数基于前缀prefix（含有数个字符的字符串）来预测接下来的num_chars个字符。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
def predict_rnn(prefix, num_chars, rnn, params, init_rnn_state,
                num_hiddens, vocab_size, device, idx_to_char, char_to_idx):
    # prefix: 长度为prefix的字符串
    # num_chars: 预测后续的字符个数
    # rnn: 循环神经网络模型
    # params: 循环神经网络模型的参数
    # init_rnn_state: 初始化隐藏状态的函数
    # num_hiddens: 隐藏单元个数
    # vocab_size: 词典大小，即one-hot向量长度
    # device: 设备名，如'cpu'或'cuda'
    # idx_to_char: 索引到字符的映射
    # char_to_idx: 字符到索引的映射
    state = init_rnn_state(1, num_hiddens, device) # 初始化隐藏状态
    output = [char_to_idx[prefix[0]]] # output记录prefix加上预测的num_chars个字符的索引，这里只记录了prefix的第一个字符
    for t in range(num_chars + len(prefix) - 1):
        # 将上一时间步的输出作为当前时间步的输入
        X = to_onehot(torch.tensor([[output[-1]]], device=device), vocab_size)
        # 计算输出和更新隐藏状态
        (Y, state) = rnn(X, state, params)
        # 下一个时间步的输入是prefix里的字符或者当前的最佳预测字符
        if t < len(prefix) - 1:
            output.append(char_to_idx[prefix[t + 1]])
        else:
            output.append(int(Y[0].argmax(dim=1).item()))
    return ''.join([idx_to_char[i] for i in output])

裁剪梯度

循环神经网络中较容易出现梯度衰减或梯度爆炸。为了应对梯度爆炸，可以裁剪梯度（clip gradient）。假设把所有模型参数梯度的元素拼接成一个向量 \(\boldsymbol{g}\)，并设裁剪的阈值是\(\theta\)。裁剪后的梯度

\[ \min\left(\frac{\theta}{\|\boldsymbol{g}\|}, 1\right)\boldsymbol{g} \]

的\(L_2\)范数不超过\(\theta\)。

如果梯度的范数 \(\|\boldsymbol{g}\|\) 小于或等于阈值 \(\theta\)，则不需要裁剪，直接使用原始梯度 \(\boldsymbol{g}\)。

如果梯度的范数 \(\|\boldsymbol{g}\|\) 大于阈值 \(\theta\)，则需要对梯度进行缩放，裁剪后的梯度为：

\[ \frac{\theta}{\|\boldsymbol{g}\|} \boldsymbol{g} \]

这相当于将梯度向量按比例缩小，使其范数变为 \(\theta\)。

1
2
3
4
5
6
7
8
9
# 本函数已保存在d2lzh_pytorch包中方便以后使用
def grad_clipping(params, theta, device):
    norm = torch.tensor([0.0], device=device)
    for param in params:
        norm += (param.grad.data ** 2).sum()
    norm = norm.sqrt().item()
    if norm > theta:
        for param in params:
            param.grad.data *= (theta / norm)

困惑度

通常使用困惑度（perplexity）来评价语言模型的好坏。困惑度是对交叉熵损失函数做指数运算后得到的值。特别地，

  • 最佳情况下，模型总是把标签类别的概率预测为1，此时困惑度为1；
  • 最坏情况下，模型总是把标签类别的概率预测为0，此时困惑度为正无穷；
  • 基线情况下，模型总是预测所有类别的概率都相同，此时困惑度为类别个数。

显然，任何一个有效模型的困惑度必须小于类别个数。在本例中，困惑度必须小于词典大小vocab_size。

定义模型训练函数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def train_and_predict_rnn(rnn, get_params, init_rnn_state, num_hiddens,
                          vocab_size, device, corpus_indices, idx_to_char,
                          char_to_idx, is_random_iter, num_epochs, num_steps,
                          lr, clipping_theta, batch_size, pred_period,
                          pred_len, prefixes):
    # rnn : 循环神经网络模型
    # get_params : 获取模型参数的函数
    # init_rnn_state : 初始化隐藏状态的函数
    # num_hiddens : 隐藏单元个数
    # vocab_size : 词典大小，即one-hot向量长度
    # device : 设备名，如'cpu'或'cuda'
    # corpus_indices : 字符索引序列
    # idx_to_char : 索引到字符的映射
    # char_to_idx : 字符到索引的映射
    # is_random_iter : 是否使用随机采样
    # num_epochs : 迭代次数
    # num_steps : 时间步数
    # lr : 学习率
    # clipping_theta : 梯度裁剪阈值
    # batch_size : 批量大小
    # pred_period : 预测周期，每间隔多少个迭代周期后预测一次
    # pred_len : 预测长度
    # prefixes : 预测时使用的前缀

    if is_random_iter:
        data_iter_fn = data_iter_random # 6.3中的随机采样函数
    else:
        data_iter_fn = data_iter_consecutive # 6.3中的相邻采样函数
    params = get_params() # 获取模型参数
    loss = nn.CrossEntropyLoss() # 定义交叉熵损失函数

    for epoch in range(num_epochs):
        if not is_random_iter:  # 如使用相邻采样，在epoch开始时初始化隐藏状态
            state = init_rnn_state(batch_size, num_hiddens, device) # 初始化隐藏状态
        l_sum, n, start = 0.0, 0, time.time() # 初始化损失，字符数，开始时间
        data_iter = data_iter_fn(corpus_indices, batch_size, num_steps, device) # 获取数据迭代器
        for X, Y in data_iter:
            if is_random_iter:  # 如使用随机采样，在每个小批量更新前初始化隐藏状态
                state = init_rnn_state(batch_size, num_hiddens, device)
            else:  
            # 否则需要使用detach函数从计算图分离隐藏状态, 这是为了
            # 使模型参数的梯度计算只依赖一次迭代读取的小批量序列(防止梯度计算开销太大)
                for s in state:
                    s.detach_() # 从计算图分离隐藏状态
        
            inputs = to_onehot(X, vocab_size) # one-hot向量
            (outputs, state) = rnn(inputs, state, params) # outputs有num_steps个形状为(batch_size, vocab_size)的矩阵
            outputs = torch.cat(outputs, dim=0) # 拼接之后形状为(num_steps * batch_size, vocab_size)
            y = torch.transpose(Y, 0, 1).contiguous().view(-1)  # Y的形状是(batch_size, num_steps)，转置后再变成长度为 batch * num_steps 的向量，这样跟输出的行一一对应
            l = loss(outputs, y.long()) # 使用交叉熵损失计算平均分类误差
        
            # 梯度清0
            if params[0].grad is not None:
                for param in params:
                    param.grad.data.zero_()
            l.backward() # 反向传播
            grad_clipping(params, clipping_theta, device)  # 裁剪梯度
            sgd(params, lr, 1)  # 因为误差已经取过均值，梯度不用再做平均
            l_sum += l.item() * y.shape[0] # 统计总的损失
            n += y.shape[0] # 统计总的预测数量

        if (epoch + 1) % pred_period == 0:
            print('第 %d 个迭代周期，困惑度 %.2f，耗时 %.2f 秒' % (
                epoch + 1, math.exp(l_sum / n), time.time() - start)) # l_sum / n 是平均损失
            for prefix in prefixes: # 使用预测模型
                print(' -', predict_rnn(prefix, pred_len, rnn, params, init_rnn_state,
                    num_hiddens, vocab_size, device, idx_to_char, char_to_idx))

训练模型并创作歌词

设置模型超参数。将根据前缀“分开”和“不分开”分别创作长度为50个字符（不考虑前缀长度）的一段歌词。每过50个迭代周期便根据当前训练的模型创作一段歌词。

1
2
num_epochs, num_steps, batch_size, lr, clipping_theta = 250, 35, 32, 1e2, 1e-2
pred_period, pred_len, prefixes = 50, 50, ['分开', '不分开']

下面采用随机采样训练模型并创作歌词。

1
2
3
4
5
train_and_predict_rnn(rnn, get_params, init_rnn_state, num_hiddens,
                      vocab_size, device, corpus_indices, idx_to_char,
                      char_to_idx, True, num_epochs, num_steps, lr,
                      clipping_theta, batch_size, pred_period, pred_len,
                      prefixes)

接下来采用相邻采样训练模型并创作歌词。

1
2
3
4
5
train_and_predict_rnn(rnn, get_params, init_rnn_state, num_hiddens,
                      vocab_size, device, corpus_indices, idx_to_char,
                      char_to_idx, False, num_epochs, num_steps, lr,
                      clipping_theta, batch_size, pred_period, pred_len,
                      prefixes)