# 导包
import os
import torch
from d2l import torch as d2l
D:\ana3\envs\nlp_prac\lib\site-packages\numpy\_distributor_init.py:32: UserWarning: loaded more than 1 DLL from .libs:
D:\ana3\envs\nlp_prac\lib\site-packages\numpy\.libs\libopenblas.IPBC74C7KURV7CB2PKT5Z5FNR3SIBV4J.gfortran-win_amd64.dll
D:\ana3\envs\nlp_prac\lib\site-packages\numpy\.libs\libopenblas.XWYDX2IKJW2NMTWSFYNGFUWKQU3LYTCZ.gfortran-win_amd64.dll
stacklevel=1)
d2l.DATA_HUB['fra-eng'] = (d2l.DATA_URL + 'fra-eng.zip',
'94646ad1522d915e7b0f9296181140edcf86a4f5')
#@save
def read_data_nmt():
"""载⼊“英语－法语”数据集"""
data_dir = d2l.download_extract('fra-eng')
with open(os.path.join(data_dir, 'fra.txt'), 'r', encoding='utf-8') as f:
return f.read()
raw_text = read_data_nmt()
print(raw_text[:75])
# 下载成功！
Downloading ..\data\fra-eng.zip from http://d2l-data.s3-accelerate.amazonaws.com/fra-eng.zip...
Go. Va !
Hi. Salut !
Run! Cours !
Run! Courez !
Who? Qui ?
Wow! Ça alors !
# 预处理
def preprocess_nmt(text):
def no_space(char, prev_char):
return char in set(',.!?') and prev_char != ' '
# 空格替换长空格、转小写, xa0是不间断空格
text =text.replace('\u202f', ' ').replace('\xa0', ' ').lower()
# 单词和标点符号之间插入空格
out = [' ' + char if i > 0 and no_space(char, text[i-1]) else char
for i, char in enumerate(text)]
return ''.join(out)
text = preprocess_nmt(raw_text)
print(text[:80])
go . va !
hi . salut !
run ! cours !
run ! courez !
who ? qui ?
wow ! ça alors !
# 可以设置样本数量
def tokenize_nmt(text, num_examples = None):
source, target = [], []
for i, line in enumerate(text.split('\n')):
if num_examples and i > num_examples:
break
parts = line.split('\t')
if len(parts) == 2:
source.append(parts[0].split(' '))
target.append(parts[1].split(' '))
return source, target
source, target = tokenize_nmt(text)
source[:6], target[:6]
([['go', '.'],
['hi', '.'],
['run', '!'],
['run', '!'],
['who', '?'],
['wow', '!']],
[['va', '!'],
['salut', '!'],
['cours', '!'],
['courez', '!'],
['qui', '?'],
['ça', 'alors', '!']])
# 绘制包含词元数量的直方图
def show_list_len_pair_hist(legend, xlabel, ylabel, xlist, ylist):
d2l.set_figsize(figsize=(5,3))
_, _, patches = d2l.plt.hist(
[[len(l) for l in xlist], [len(l) for l in ylist]])
d2l.plt.xlabel(xlabel)
d2l.plt.ylabel(ylabel)
for patch in patches[0].patches:
patch.set_hatch('\')
for patch in patches[1].patches:
patch.set_hatch('/')
d2l.plt.legend(legend)
show_list_len_pair_hist(['source', 'target'], '# tokens per sequence',
'count', source, target)
# 构建词表，添加各种词元
src_vocab = d2l.Vocab(source, min_freq=2, reserved_tokens = ['' , '' , '' ])
len(src_vocab)
10012
# 测试
src_vocab['hello'], src_vocab.to_tokens(1807)
(1807, 'hello')
# 截断和填充
def truncate_pad(line, num_steps, padding_token):
if len(line) > num_steps: # 截断
return line[:num_steps]
return line + [padding_token] * (num_steps - len(line))
truncate_pad(src_vocab[source[0]], 10, src_vocab['' ])
[47, 4, 1, 1, 1, 1, 1, 1, 1, 1]
# 构建小批量数据集
def build_array_nmt(lines, vocab, num_steps):
lines = [vocab[l] for l in lines]
lines = [l + [vocab['' ]] for l in lines]
array = torch.tensor([truncate_pad(
l, num_steps, vocab['' ]) for l in lines])
# 挺巧妙的方法计算长度
valid_len = (array != vocab['' ]).type(torch.int32).sum(1)
return array, valid_len
def load_data_nmt(batch_size, num_steps, num_examples=600):
text = preprocess_nmt(read_data_nmt()) # 文本预处理
source, target = tokenize_nmt(text, num_examples) # 生成序列对
# 构造词表
src_vocab = d2l.Vocab(source, min_freq=2,
reserved_tokens=['' , '' , '' ])
tgt_vocab = d2l.Vocab(target, min_freq=2,
reserved_tokens=['' , '' , '' ])
# 构造小批量数据
src_array, src_valid_len = build_array_nmt(source, src_vocab, num_steps)
tgt_array, tgt_valid_len = build_array_nmt(target, tgt_vocab, num_steps)
data_arrays = (src_array, src_valid_len, tgt_array, tgt_valid_len)
data_iter = d2l.load_array(data_arrays, batch_size)
# 数据迭代器，词表
return data_iter, src_vocab, tgt_vocab
# 读出第一个小批量数据
train_iter, src_vocab, tgt_vocab = load_data_nmt(batch_size=2, num_steps=8)
for x, x_valid_len, y, y_valid_len in train_iter:
print('x: ', x.type(torch.int32))
print('x的有效长度：', x_valid_len)
print('y: ', y.type(torch.int32))
print('y的有效长度：', y_valid_len)
break
x: tensor([[ 90, 19, 4, 3, 1, 1, 1, 1],
[136, 15, 4, 3, 1, 1, 1, 1]], dtype=torch.int32)
x的有效长度： tensor([4, 4])
y: tensor([[ 0, 12, 5, 3, 1, 1, 1, 1],
[ 0, 5, 3, 1, 1, 1, 1, 1]], dtype=torch.int32)
y的有效长度： tensor([4, 3])
from torch import nn
class Encoder(nn.Module):
"""基本编码器接口"""
def __init__(self, **kwargs):
super(Encoder, self).__init__(**kwargs)
def forward(self, x, *args):
raise NotImplementedError
class Decoder(nn.Module):
"""基本编码器接口"""
def __init__(self, **kwargs):
super(Decoder, self).__init__(**kwargs)
def init_state(self, enc_outputs, *args):
raise NotImplementedError
def forward(self, x, state):
raise NotImplementedError
class EncoderDecoder(nn.Module):
"""编码器-解码器架构的基类"""
def __init__(self, encoder, decoder, **kwargs):
super(EncoderDecoder, self).__init__(**kwargs)
self.encoder = encoder
self.decoder = decoder
def forward(self, enc_x, dec_x, *args):
enc_outputs = self.encoder(enc_x, *args) # 灌入数据等参数
dec_state = self.decoder.init_state(enc_outputs, *args) # 思想向量灌入解码器
return self.decoder(dec_x, dec_state)
import collections
import math
import torch
from torch import nn
from d2l import torch as d2l
class Seq2SeqEncoder(d2l.Encoder):
"""Seq2Seq的RNN编码器"""
def __init__(self, vocab_size, embed_size, num_hiddens, num_layers,
dropout=0, **kwargs):
super(Seq2SeqEncoder, self).__init__(**kwargs)
# 嵌入层
self.embedding = nn.Embedding(vocab_size, embed_size)
self.rnn = nn.GRU(embed_size, num_hiddens, num_layers, dropout=dropout)
def forward(self, x, *args):
# 输出x的形状：batch_size, num_steps, embed_size
x = self.embedding(x)
# rnn中，第一个轴为时间步
x = x.permute(1, 0, 2)
# 状态初始默认为0
output, state = self.rnn(x)
# output维度： (num_steps, batch_size, num_hiddens)
# state[0]的形状：(num_layers, batch_size, num_hiddens)
return output, state
# 实例化上述编码器 词典大小 词向量维度 隐藏层单元数 编码器层数(隐藏层数量)
encoder = Seq2SeqEncoder(vocab_size=10, embed_size=8, num_hiddens=16, num_layers=2)
encoder.eval()
x = torch.zeros((4, 7), dtype=torch.long) # 4个长度为7的句子,即批量大小、时间步
output, state = encoder(x)
output.shape
torch.Size([7, 4, 16])
state.shape
torch.Size([2, 4, 16])
class Seq2SeqDecoder(d2l.Decoder):
"""用于Seq2Seq学习的rnn解码器"""
def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, dropout=0, **kwargs):
super(Seq2SeqDecoder, self).__init__(**kwargs)
self.embedding = nn.Embedding(vocab_size, embed_size)
self.rnn = nn.GRU(embed_size + num_hiddens, num_hiddens, num_layers, dropout=dropout)
self.dense = nn.Linear(num_hiddens, vocab_size)
def init_state(self, enc_outputs, *args):
return enc_outputs[1]
def forward(self, x, state):
# x的维度：batch_size, num_steps, embed_size
x = self.embedding(x).permute(1, 0, 2)
# 广播context，使具有与x系统的num_steps
context = state[-1].repeat(x.shape[0], 1, 1)
x_and_context = torch.cat((x, context), 2)
output, state = self.rnn(x_and_context, state)
output = self.dense(output).permute(1, 0, 2) # 换回去
# output维度： batch_size, num_steps, vocab_size
# state[0]维度：num_layers, batch_size, num_hiddens
return output, state
# 实例化解码器
decoder = Seq2SeqDecoder(vocab_size=10, embed_size=8, num_hiddens=16, num_layers=2)
decoder.eval()
state = decoder.init_state(encoder(x))
output, state = decoder(x, state)
output.shape, state.shape
(torch.Size([4, 7, 10]), torch.Size([2, 4, 16]))
# 先要屏蔽不相关的项
def sequence_mask(x, valid_len, value=0):
"""在序列中屏蔽不相关的项"""
maxlen = x.size(1) # 这里决定了总是 *** 作第二维的长度！！
# print(maxlen)
mask = torch.arange((maxlen), dtype=torch.float32,
device=x.device)[None, :] < valid_len[:, None]
# 在行维度上提升一维度， 在列维度上提升一个维度
# print(torch.arange((maxlen), dtype=torch.float32,
# device=x.device)[None, :])
# print(mask, ~mask, sep='\n')
# print(valid_len[:, None])
x[~mask] = value # 把0给需要屏蔽的项目
return x
# 测试
x = torch.tensor([[1, 2, 3], [4, 5, 6]])
sequence_mask(x, torch.tensor([1, 2]))
tensor([[1, 0, 0],
[4, 5, 0]])
# 测试广播
torch.tensor([[0., 1., 2.]]) < torch.tensor([[1],
[2]])
~torch.tensor([[True, True, False]])
tensor([[False, False, True]])
# 屏蔽最后几个轴上的所有项；也可以指定非零值来替换这些项
x = torch.ones(2, 3, 4)
sequence_mask(x, torch.tensor([1, 2]), value=-1) # 第二维度上的前1和前2项保留
tensor([[[ 1., 1., 1., 1.],
[-1., -1., -1., -1.],
[-1., -1., -1., -1.]],
[[ 1., 1., 1., 1.],
[ 1., 1., 1., 1.],
[-1., -1., -1., -1.]]])
# 计算最终的交叉熵损失（屏蔽了不相关的预测）
class MaskedSoftmaxCELoss(nn.CrossEntropyLoss):
# pred形状：batch_size, num_steps, vocab_size
# label: batch_size, num_steps (其实就是batch_size个句子，每个句子长度为num_steps)
# valid_len:batch_size （每个句子的长度大小）
def forward(self, pred, label, valid_len):
weights = torch.ones_like(label)
weights = sequence_mask(weights, valid_len)
self.reduction = 'none'
unweighted_loss = super(MaskedSoftmaxCELoss, self).forward(
pred.permute(0, 2, 1), label)
weighted_loss = (unweighted_loss * weights).mean(dim=1)
return weighted_loss
# 测试
loss = MaskedSoftmaxCELoss()
loss(torch.ones(3, 4, 10), torch.ones((3, 4), dtype=torch.long), torch.tensor([4, 2, 0]))
tensor([2.3026, 1.1513, 0.0000])
def train_seq2seq(net, data_iter, lr, num_epochs, tgt_vocab, device):
"""训练seq2seq模型"""
# xavier初始化权重
def xavier_init_weights(m):
if type(m) == nn.Linear:
nn.init.xavier_uniform_(m.weight)
if type(m) == nn.GRU:
for param in m._flat_weights_names:
if 'weight' in param:
nn.init.xavier_uniform_(m._parameters[param])
net.apply(xavier_init_weights) # 初始化网络参数
net.to(device) # 使用gpu训练
optimizer = torch.optim.Adam(net.parameters(), lr=lr) # 使用adam优化网络的参数，灌入学习率
loss = MaskedSoftmaxCELoss() # 创建损失函数
net.train() # 网络训练模式
animator = d2l.Animator(xlabel='epoch', ylabel='loss', xlim=[10, num_epochs]) # 打印数据
# 按照轮次进行训练
for epoch in range(num_epochs):
timer = d2l.Timer() # 计时器
metric = d2l.Accumulator(2) # 训练损失总和，词元数量
for batch in data_iter: # 批量数据训练
optimizer.zero_grad() # 优化器梯度置为0
x, x_valid_len, y, y_valid_len = [x.to(device) for x in batch] # 获取本次小批量数据
bos = torch.tensor([tgt_vocab['' ]] * y.shape[0], # 加上开始标签
device=device).reshape(-1, 1)
dec_input = torch.cat([bos, y[:, :-1]], 1) # 强制教学 teacher forcing, 加上开始标签和原始输出序列
y_hat, _ = net(x, dec_input, x_valid_len) # 输入灌入网络，获得预测值
l = loss(y_hat, y, y_valid_len) # 计算损失
l.sum().backward() # 损失函数的标量进行“反向传播”
d2l.grad_clipping(net, 1) # 防止梯度爆炸
num_tokens = y_valid_len.sum() # 标签的词元总数
optimizer.step() # 优化器优化
with torch.no_grad():
metric.add(l.sum(), num_tokens) # 损失总数，词元总数
# 画图
if (epoch + 1) % 10 == 0:
animator.add(epoch + 1, (metric[0]) / metric[1], ) # 计算平均损失
print(f'loss {metric[0] / metric[1]:.3f}, {metric[1] / timer.stop():.1f} '
f'tokens / sec on {str(device)}')
# 在机器翻译数据集上，创建和训练一个rnn“编码器-解码器”模型用于Seq2Seq的学习
# 初始化参数
embed_size, num_hiddens, num_layers, dropout = 64, 64, 2, 0.3
batch_size, num_steps = 64, 10
lr, num_epochs, device = 0.005, 500, d2l.try_gpu()
train_iter, src_vocab, tgt_vocab = load_data_nmt(batch_size, num_steps) # 载入数据和词表
# 构造网络模型
encoder = Seq2SeqEncoder(len(src_vocab), embed_size, num_hiddens, num_layers, dropout)
decoder = Seq2SeqDecoder(len(tgt_vocab), embed_size, num_hiddens, num_layers, dropout)
net = d2l.EncoderDecoder(encoder, decoder)
# 训练
train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)
loss 0.019, 17781.7 tokens / sec on cuda:0
## 保存模型
torch.save(net, 'mt_model_gru1.pth')
# seq2seq模型的预测
def predict_seq2seq(net, src_sentence, src_vocab, tgt_vocab, num_steps, device,
save_attention_weights=False):
# 预测的时候，设置net为评估模式
net.eval()
# 原始句子转为tokens
src_tokens = src_vocab[src_sentence.lower().split(' ')] + [src_vocab['' ]]
enc_valid_len = torch.tensor([len(src_tokens)], device=device)
# 用pad补全
src_tokens = d2l.truncate_pad(src_tokens, num_steps, src_vocab['' ])
# 添加批量轴
enc_x = torch.unsqueeze(torch.tensor(src_tokens, dtype=torch.long, device=device), dim=0)
# 计算出思想向量
enc_outputs = net.encoder(enc_x, enc_valid_len)
# 用思想向量初始化解码器
dec_state = net.decoder.init_state(enc_outputs, enc_valid_len)
# 添加批量轴
dec_x = torch.unsqueeze(torch.tensor([tgt_vocab['' ]], dtype=torch.long, device=device), dim=0)
output_seq, attention_weight_seq = [], []
for _ in range(num_steps):
# 解码器中灌入初始词元和思想向量初始化的解码器
y, dec_state = net.decoder(dec_x, dec_state)
# 使用具有预测最高可能性的词元（贪心搜索），作为解码器在下一时间步的输入
dec_x = y.argmax(dim=2)
pred = dec_x.squeeze(dim=0).type(torch.int32).item() # 拿到这个标量
# 保存注意力权重
if save_attention_weights:
attention_weight_seq.append(net.decoder.attention_weights)
# 一旦序列结束词元被预测，输出序列的生成就完成了
if pred == tgt_vocab['' ]:
break
output_seq.append(pred) # 将每次预测token加入输出序列
return ' '.join(tgt_vocab.to_tokens(output_seq)), attention_weight_seq
- 前半部分为惩罚系数，惩罚短的预测序列
- 后半部分为精确度
def bleu(pred_seq, label_seq, k): # 累加到k元
pred_tokens, label_tokens = pred_seq.split(' '), label_seq.split(' ')
len_pred, len_label = len(pred_tokens), len(label_tokens)
score = math.exp(min(0, 1 - len_label / len_pred)) # 计算惩罚项
# 循环统计n元准确率，使用字典来实现
for n in range(1, k + 1):
num_matches, label_subs = 0, collections.defaultdict(int)
for i in range(len_label - n + 1):
label_subs[' '.join(label_tokens[i: i + n])] += 1
for i in range(len_pred - n + 1):
if label_subs[' '.join(pred_tokens[i: i + n])] > 0:
num_matches += 1
label_subs[' '.join(pred_tokens[i: i + n])] -= 1
score *= math.pow(num_matches / (len_pred - n + 1), math.pow(0.5, n))
return score
# 测试bleu最终结果
engs = ['go .', "i lost .", 'he\'s calm .', 'i\'m home .']
fras = ['va !', 'j\'ai perdu .', 'il est calme .', 'je suis chez moi .']
for eng, fra in zip(engs, fras):
translation, attention_weight_seq = predict_seq2seq(
net, eng, src_vocab, tgt_vocab, num_steps, device)
print(f'{eng} => {translation}, bleu {bleu(translation, fra, k=2):.3f}')
go . => va !, bleu 1.000
i lost . => j'ai perdu ., bleu 1.000
he's calm . => il est mouillé ., bleu 0.658
i'm home . => je suis chez ., bleu 0.752
欢迎分享，转载请注明来源：内存溢出
微信扫一扫
支付宝扫一扫
评论列表（0条）