https://zh.d2l.ai/chapter_recurrent-neural-networks/rnn-concise.html#
86循环神经网络的简洁.py
import torch
from torch import nn
from torch.nn import functional as F
from d2l import torch as d2l
from API_86 import *
batch_size, num_steps = 32, 35
train_iter, vocab = load_data_time_machine(batch_size, num_steps)
num_hiddens = 256
# 词的长度决定了热编码长度 决定了单个样本的长度
rnn_layer = nn.RNN(len(vocab), num_hiddens)
# H state
state = torch.zeros((1, batch_size, num_hiddens))
#state.shape
#num_steps 35 *32 *len(vocab)
X = torch.rand(size=(num_steps, batch_size, len(vocab)))
Y, state_new = rnn_layer(X, state)
# Y.shape, state_new.shape
#@save
class RNNModel(nn.Module):
"""循环神经网络模型"""
def __init__(self, rnn_layer, vocab_size, **kwargs):
super(RNNModel, self).__init__(**kwargs)
self.rnn = rnn_layer
self.vocab_size = vocab_size
self.num_hiddens = self.rnn.hidden_size
# 如果RNN是双向的（之后将介绍），num_directions应该是2，否则应该是1
if not self.rnn.bidirectional:
self.num_directions = 1
self.linear = nn.Linear(self.num_hiddens, self.vocab_size)
else:
self.num_directions = 2
self.linear = nn.Linear(self.num_hiddens * 2, self.vocab_size)
def forward(self, inputs, state):
X = F.one_hot(inputs.T.long(), self.vocab_size)
X = X.to(torch.float32)
Y, state = self.rnn(X, state)
# 全连接层首先将Y的形状改为(时间步数*批量大小,隐藏单元数)
# 它的输出形状是(时间步数*批量大小,词表大小)。
output = self.linear(Y.reshape((-1, Y.shape[-1])))
return output, state
def begin_state(self, device, batch_size=1):
if not isinstance(self.rnn, nn.LSTM):
# nn.GRU以张量作为隐状态
return torch.zeros((self.num_directions * self.rnn.num_layers,
batch_size, self.num_hiddens),
device=device)
else:
# nn.LSTM以元组作为隐状态
return (torch.zeros((
self.num_directions * self.rnn.num_layers,
batch_size, self.num_hiddens), device=device),
torch.zeros((
self.num_directions * self.rnn.num_layers,
batch_size, self.num_hiddens), device=device))
device = d2l.try_gpu()
net = RNNModel(rnn_layer, vocab_size=len(vocab))
net = net.to(device)
predict_ch8('time traveller', 10, net, vocab, device)
num_epochs, lr = 500, 1
train_ch8(net, train_iter, vocab, lr, num_epochs, device)
API_86.py
import collections
import re
from d2l import torch as d2l
import random
import math
import torch
from torch import nn
from torch.nn import functional as F
import numpy as np
draw_pic=0
#@save
d2l.DATA_HUB['time_machine'] = (d2l.DATA_URL + 'timemachine.txt',
'090b5e7e70c295757f55df93cb0a180b9691891a')
def read_time_machine(): #@save
"""将时间机器数据集加载到文本行的列表中"""
with open(d2l.download('time_machine'), 'r') as f:
lines = f.readlines()
#re.sub('[^A-Za-z]+', ' ', line)：将line字符串中的 连续多个非字母的字符 变成 空格
#re.sub('[^A-Za-z]+', ' ', 'I\n')
#Out[6]: 'I '
return [re.sub('[^A-Za-z]+', ' ', line).strip().lower() for line in lines]
#lines = read_time_machine()
#print(f'# 文本总行数: {len(lines)}')
#print(lines[0])
#print(lines[10])
def tokenize(lines, token='word'): #@save
"""将文本行拆分为单词或字符词元"""
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)
# for i in range(16):
# print("tokens",i,tokens[i])
def count_corpus(tokens):#
"""统计词元的频率"""
# 这里的tokens是1D列表或2D列表
if len(tokens) == 0 or isinstance(tokens[0], list):
# 将词元列表展平成一个列表
# tokens1=[]
# for line in tokens:
# #print("line",line)
# for token in line:
# #print("token",token)
# tokens1.append(token)
# tokens=tokens1
tokens = [token for line in tokens for token in line]
#print(tokens.shape)
#最主要的作用是计算“可迭代序列”中各个元素（element）的数量
'''
#对字符串作用
temp=Counter('aabbcdeffgg')
print(temp) #Counter({'a': 2, 'b': 2, 'f': 2, 'g': 2, 'c': 1, 'd': 1, 'e': 1})
'''
return collections.Counter(tokens)
#count_corpus(tokens)
#词表
#构建一个字典，通常也叫做词表（vocabulary）,用来将字符串类型的词元映射到从开始的数字索引中.
# 我们先将训练集中的所有文档合并在一起，对它们的唯一词元进行统计， 得到的统计结果称之为语料（corpus）。
# 然后根据每个唯一词元的出现频率，为其分配一个数字索引。 很少出现的词元通常被移除，这可以降低复杂性。
# 另外，语料库中不存在或已删除的任何词元都将映射到一个特定的未知词元“<unk>”。
# 我们可以选择增加一个列表，用于保存那些被保留的词元， 例如：填充词元（“<pad>”）； 序列开始词元（“<bos>”）； 序列结束词元
class Vocab: #@save
"""文本词表"""
def __init__(self, tokens=None, min_freq=0, reserved_tokens=None):
if tokens is None:
tokens = []
if reserved_tokens is None:
#reserved_tokens = ["a","b"]
reserved_tokens = []
# 按出现频率排序
'''
#对字符串作用
temp=Counter('aabbcdeffgg')
print(temp) #Counter({'a': 2, 'b': 2, 'f': 2, 'g': 2, 'c': 1, 'd': 1, 'e': 1})
'''
counter = count_corpus(tokens)
#print("词频统计结果",counter)
'''
词频统计结果 Counter({' ': 29927, 'e': 17838, 't': 13515, 'a': 11704,
'i': 10138, 'n': 9917, 'o': 9758, 's': 8486,
'h': 8257, 'r': 7674, 'd': 6337, 'l': 6146, 'm': 4043,
'u': 3805, 'c': 3424, 'f': 3354, 'w': 3225, 'g': 3075,
'y': 2679, 'p': 2427, 'b': 1897, 'v': 1295, 'k': 1087,
'x': 236, 'z': 144, 'j': 97, 'q': 95})
'''
# 排序 已知词汇 reverse=True 从高到低
self._token_freqs = sorted(counter.items(), key=lambda x: x[1],
reverse=True)
# 未知词元的索引为0
self.idx_to_token = ['<unk>'] + reserved_tokens
#self.idx_to_token = ['<unk>'] + reserved_tokens
#{'<unk>': 0}
#{'<unk>': 0, 'a': 1, 'b': 2}
self.token_to_idx = {token: idx
for idx, token in enumerate(self.idx_to_token)}
#print(self.token_to_idx )
'''
1最开头是 未知词元 假设是
词源 idx_to_token['<unk>', ‘a’ ,'b']
词源ID token_to_idx[0,1,2]
2将文本中统计出来的有效词源 加进去 例如 b1
2-1 剔除小于阈值的
2-2 加入新词源
频率 从高到低
词源 idx_to_token['<unk>', ‘a’ ,'b','b1']
词源ID token_to_idx[0,1,2,3]
'''
# 最开头是 未知词元 假设是 idx_to_token['<unk>', ‘a’ ,'b']
# 将文本中统计出来的有效词源 加进去 例如 b1
#
for token, freq in self._token_freqs:
# 从高到低访问
if freq < min_freq:# 剔除小于阈值的词
break
if token not in self.token_to_idx:# 未知词汇剔除
# 未知词元 添加 单词进入列表
# '<unk>' a b 加入 d1
self.idx_to_token.append(token)
# token_to_idx【'd1'】= 4-1=3
self.token_to_idx[token] = len(self.idx_to_token) - 1 #从0开始 添加
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 load_corpus_time_machine(max_tokens=-1): #@save
"""返回时光机器数据集的词元索引列表和词表"""
# 1 读取所有文本 逐行存
lines = read_time_machine()
# 2
# 2-1将所有行的字母拆出来，汇总到一个列表里面
tokens = tokenize(lines, 'char')
# print("tokens",tokens)
'''
tokens= [['a', 'b'],['c' 'd'],...]
'''
# 3 获取词汇列表
'''
(频率 从高到底 前面是未知词源)
词源 idx_to_token['<unk>', ‘a’ ,'b','b1']
词源ID token_to_idx[0,1,2,3]
'''
vocab = Vocab(tokens)
# 因为时光机器数据集中的每个文本行不一定是一个句子或一个段落，
# 所以将所有文本行拆分成一个个字符，然后展平到一个列表中
corpus=[]
'''
tokens= [['a', 'b'],['c' 'd'],...]
'''
for line in tokens:
for token in line:
corpus.append(vocab[token]) # vocab[token] 字母’x‘的编码ID
#corpus = [vocab[token] for line in tokens for token in line]
if max_tokens > 0:
corpus = corpus[:max_tokens]
#corpus 全文变成单独字符的列表结合
#vocab vocab[token] 字母’x‘的编码ID(非次数 ID就是从高到低按照次数排列到的) 集合 {“a”,1,"b","2",...}
return corpus, vocab
#===============================测试1 按照 字符拆分==============================================
#在使用上述函数时，我们将所有功能打包到load_corpus_time_machine函数中，
#该函数返回corpus（词元索引列表）和vocab（时光机器语料库的词表）。
'''
为了简化后面章节中的训练，我们使用字符（而不是单词）实现文本词元化；
时光机器数据集中的每个文本行不一定是一个句子或一个段落，还可能是一个单词，因此返回的corpus仅处理为单个列表，而不是使用多词元列表构成的一个列表。
'''
#corpus, vocab = load_corpus_time_machine()
#170580 28 一共有corpus 170580个字符，其中重复字符的有vocab 28个（ABCD-FG 28个英文字符）
#print(len(corpus), len(vocab))
'''
词频统计结果 Counter({' ': 29927, 'e': 17838, 't': 13515, 'a': 11704,
'i': 10138, 'n': 9917, 'o': 9758, 's': 8486,
'h': 8257, 'r': 7674, 'd': 6337, 'l': 6146, 'm': 4043,
'u': 3805, 'c': 3424, 'f': 3354, 'w': 3225, 'g': 3075,
'y': 2679, 'p': 2427, 'b': 1897, 'v': 1295, 'k': 1087,
'x': 236, 'z': 144, 'j': 97, 'q': 95})
'''
#==================
#当序列变得太长而不能被模型一次性全部处理时， 我们可能希望拆分这样的序列方便模型读取。
# 假设我们将使用神经网络来训练语言模型， 模型中的网络一次处理具有预定义长度 （例如个时间步）的一个小批量序列。
#现在的问题是如何随机生成一个小批量数据的特征和标签以供读取。
#首先，由于文本序列可以是任意长的
#于是任意长的序列可以被我们划分为具有相同时间步数的子序列。
#因为我们可以选择任意偏移量来指示初始位置，所以我们有相当大的自由度。
#如果我们只选择一个偏移量， 那么用于训练网络的、所有可能的子序列的覆盖范围将是有限的。
#因此，我们可以从随机偏移量开始划分序列， 以同时获得覆盖性（coverage）和随机性（randomness）。
def seq_data_iter_sequential(corpus, batch_size, num_steps): #@save
"""使用顺序分区生成一个小批量子序列"""
# 从随机偏移量开始划分序列
#corpus= [0, 1, 2, 3, ...,33,34]
#num_steps=5
#batch_size=2
offset = random.randint(0, num_steps)# 获取一个 0-num_steps 之间的整型随机数
#offset=1
print('offset',offset)
##offset 1
num_tokens = ((len(corpus) - offset - 1) // batch_size) * batch_size
#取整除 ,返回商的整数部分(向下取整)
# （35-1-1）// 2 * 2 = 16 *2 = 32
print('num_tokens',num_tokens)
#num_tokens 32
# [1 : 1+32=33] 最后一个不取==> [1 : 32] 总长度 num_tokens 偶数
Xs = torch.tensor(corpus[offset: offset + num_tokens])
print('Xs',Xs.shape,Xs)
'''
Xs torch.Size([32]) tensor([ 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])
'''
# [2 : 2+32=34]
Ys = torch.tensor(corpus[offset + 1: offset + 1 + num_tokens])
print('Ys',Ys.shape,Ys)
'''
Ys torch.Size([32]) tensor([ 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])
'''
# batch_size 批次 单词导入的样本数目
# Xs 32个
Xs, Ys = Xs.reshape(batch_size, -1), Ys.reshape(batch_size, -1)
print('Xsreshape',Xs.shape,Xs)
'''
Xsreshape torch.Size([2, 16]) tensor(
[[ 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]])
'''
print('Ysreshape',Ys.shape,Ys)
'''
Ysreshape torch.Size([2, 16]) tensor(
[[ 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]])
'''
# Xs.shape[1] 列 长度// num_steps
num_batches = Xs.shape[1] // num_steps
#num_batches= 16//5 = 15/5=3
# 0 - 5*3 - 5
for i in range(0, num_steps * num_batches, num_steps):
# i 0 5 10
X = Xs[:, i: i + num_steps]
Y = Ys[:, i: i + num_steps]
'''
# i=0
Xs [:,0:5]
Ys [:,0:5]
# i=5
Xs [:,5:10]
Ys [:,5:10]
# i=15
Xs [:,10:15]
Ys [:,10:15]
'''
yield X, Y
#my_seq [0, 1, 2, 3, ...,33,34]
#my_seq = list(range(35))
#print("my_seq",my_seq)
#for X, Y in seq_data_iter_sequential(my_seq, batch_size=2, num_steps=5):
# print('X: ', X, '\nY:', Y)
#==================
def seq_data_iter_random(corpus, batch_size, num_steps): #@save
"""使用随机抽样生成一个小批量子序列"""
# 从随机偏移量开始对序列进行分区，随机范围包括num_steps-1
corpus = corpus[random.randint(0, num_steps - 1):]
# 减去1，是因为我们需要考虑标签
num_subseqs = (len(corpus) - 1) // num_steps
# 长度为num_steps的子序列的起始索引
initial_indices = list(range(0, num_subseqs * num_steps, num_steps))
# 在随机抽样的迭代过程中，
# 来自两个相邻的、随机的、小批量中的子序列不一定在原始序列上相邻
random.shuffle(initial_indices)
def data(pos):
# 返回从pos位置开始的长度为num_steps的序列
return corpus[pos: pos + num_steps]
num_batches = num_subseqs // batch_size
for i in range(0, batch_size * num_batches, batch_size):
# 在这里，initial_indices包含子序列的随机起始索引
initial_indices_per_batch = initial_indices[i: i + batch_size]
X = [data(j) for j in initial_indices_per_batch]
Y = [data(j + 1) for j in initial_indices_per_batch]
yield torch.tensor(X), torch.tensor(Y)
#my_seq = list(range(35))
#for X, Y in seq_data_iter_random(my_seq, batch_size=2, num_steps=5):
# print('X: ', X, '\nY:', Y)
#我们将上面的两个采样函数包装到一个类中， 以便稍后可以将其用作数据迭代器。
class SeqDataLoader: #@save
"""加载序列数据的迭代器"""
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 = load_corpus_time_machine(max_tokens)
'''
corpus=['单词1','单词2','单词3',...,'最后单词']
# 全文拆分成行句，据拆分成单词，所有单词汇聚在一个列表中。
vocab = Vocab(tokens)
(频率 从高到底 前面是未知词源)
词源 idx_to_token['<unk>', ‘a’ ,'b','b1']
词源ID token_to_idx[0,1,2,3]
'''
self.batch_size, self.num_steps = batch_size, num_steps
def __iter__(self):
return self.data_iter_fn(self.corpus, self.batch_size, self.num_steps)
# 我们定义了一个函数load_data_time_machine， 它同时返回数据迭代器和词表， 因此可以与其他带有load_data前缀的函数 （如 3.5节中定义的 d2l.load_data_fashion_mnist）类似地使用。
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
#===================训练
'''
序列数据的不同采样方法（随机采样和顺序分区）将导致隐状态初始化的差异。
我们在更新模型参数之前裁剪梯度。 这样的操作的目的是，即使训练过程中某个点上发生了梯度爆炸，也能保证模型不会发散。
我们用困惑度来评价模型。如 8.4.4节所述， 这样的度量确保了不同长度的序列具有可比性。
'''
#@save
def train_epoch_ch8(net, train_iter, loss, updater, device, use_random_iter):
"""训练网络一个迭代周期（定义见第8章）"""
state, timer = None, d2l.Timer()
metric = d2l.Accumulator(2) # 训练损失之和,词元数量
for X, Y in train_iter:
# state==H
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()
#@save
def train_ch8(net, train_iter, vocab, lr, num_epochs, device,
use_random_iter=False):
"""训练模型（定义见第8章）"""
# 1 定义损失
loss = nn.CrossEntropyLoss()
animator = d2l.Animator(xlabel='epoch', ylabel='perplexity',
legend=['train'], xlim=[10, num_epochs])
#2 确定更新步骤策略 初始化
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_ch8(prefix, 50, net, vocab, device)
#pre_result=predict_ch8('time traveller ', 10, net, vocab, d2l.try_gpu())
# 训练和预测
for epoch in range(num_epochs):
ppl, speed = train_epoch_ch8(net, train_iter, loss, updater, device, use_random_iter)
if (epoch + 1) % 10 == 0:
print(predict('time traveller'))
animator.add(epoch + 1, [ppl])
print(f'困惑度 {ppl:.1f}, {speed:.1f} 词元/秒 {str(device)}')
print(predict('time traveller'))
print(predict('traveller'))
#预测
def predict_ch8(prefix, num_preds, net, vocab, device): #@save
"""在prefix后面生成新字符"""
state = net.begin_state(batch_size=1, device=device)
#state=(torch.zeros((batch_size, num_hiddens), device=device),)
# [0,0,...,0 ] 1*512
outputs = [vocab[prefix[0]]]
#prefix=['time traveller ']
#prefix[0]='t'
#vocab['t'] #词排序 {'<unk>': 0, ' ': 1, 'e': 2, 't': 3, 'a': 4,...}
print('outputs',outputs,outputs[-1]) # 3
# 最后一个字母的编号
#lambda 函数只计算一个短表达式将 这个变为函数可以调用 获取最后面的字符
get_input = lambda: torch.tensor([outputs[-1]], device=device).reshape((1, 1))
print('get_input',get_input())
#get_input tensor([[3]], device='cuda:0')
# 使用 输入预测的TXT 跟新下H 到 H（r）
for y in prefix[1:]: # 预热期
# y= i m e t r a v e l l e r
# state [0,0,...,0 ] 1*512
#net=RNNModelScratch
# state 是 H 中间变量
_, state = net(get_input(), state)
# net(get_input(), state)
# 1 根据词ID 变为 热编码X
# X = F.one_hot(X.T, self.vocab_size).type(torch.float32)
# 2 2-1热编码X作为训练数据输入
# 2-2 state作为初H的始化参数 # [0,0,...,0 ] 1*512
# 2-3 self.params 是所有的网络参数
#return self.forward_fn(X, state, self.params)
# 输出
# 预测的结果集合
# return Y_list, (H,)
#vocab[y] 字符 y 的 词编码ID
outputs.append(vocab[y])
#print('outputs',outputs) # 3)
#输入 y=t i m e t r a v e l l e r
#词排序 {'<unk>': 0, ' ': 1, 'e': 2, 't': 3, 'a': 4, 'i': 5, 'n': 6, 'o': 7, 's': 8, 'h': 9,
# 'r': 10, 'd': 11, 'l': 12, 'm': 13, 'u': 14, 'c': 15, 'f': 16, 'w': 17, 'g': 18, 'y': 19,
# 'p': 20, 'b': 21, 'v': 22, 'k': 23, 'x': 24, 'z': 25, 'j': 26, 'q': 27}
#输出 outputs [3, 5, 13, 2, 1, 3, 10, 4, 22, 2, 12, 12, 2, 10, 1]
# 预测步长 num_preds=10
## 使用 输入预测的TXT 跟新下H 到 H（r），开始预测后续的H（...）
for _ in range(num_preds): # 预测num_preds步
# state -- H 中间态
# get_input() 获取最后一个字母
y, state = net(get_input(), state)
# 摘除最大的预测结果 lambda
y_max=int(y.argmax(dim=1).reshape(1))
outputs.append(y_max)
#print('outputs',outputs) #
#outputs [3, 5, 13, 2, 1, 3, 10, 4, 22, 2, 12, 12, 2, 10, 1, 20, 25, 10, 24, 3, 11, 23, 4, 18, 18]
msg=[]
for i in outputs:
# 根据词的id 找字符
c_id=vocab.idx_to_token[i]
# 集合起来
msg.append(c_id)
msg_txt=''.join(msg)
return msg_txt
#====================梯度裁剪
def grad_clipping(net, theta): #@save
"""裁剪梯度"""
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