读取周杰伦专辑歌词数据集：
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向量。
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向量长度（词典大小）。
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是一个超参数。
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的情况。
def init_rnn_state(batch_size, num_hiddens, device):
return (torch.zeros((batch_size, num_hiddens), device=device), )下面的rnn函数定义了在一个时间步里如何计算隐藏状态和输出。这里的激活函数使用了tanh函数。
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个字符。
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$。
# 本函数已保存在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。
定义模型训练函数
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个迭代周期便根据当前训练的模型创作一段歌词。
num_epochs, num_steps, batch_size, lr, clipping_theta = 250, 35, 32, 1e2, 1e-2
pred_period, pred_len, prefixes = 50, 50, ['分开', '不分开']下面采用随机采样训练模型并创作歌词。
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)接下来采用相邻采样训练模型并创作歌词。
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)
Comments | NOTHING