Pytorch官方教程：含注意力的seq2seq2機器翻譯
引入
資料下載：傳送門
這個教程的任務目標是將輸入的法文翻譯成英文。因為翻譯前後句子可能不等長，所以用之前的RNN、LSTM就不太合適，所以這裡就是用的Encoder-Decoder結構。其結構如下圖所示：
資料處理
所需要的庫：
from __future__ import unicode_literals, print_function, division
from io import open
import unicodedata
import string
import re
import random
import torch
import torch.nn as nn
from torch import optim
import torch.nn.functional as F
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
我們之後需要將每個單詞對應唯一的索引作為神經網路的輸入和目標.為了追蹤這些索引我們使用一個幫助類Lang類中有 詞 → 索引 (word2index) 和 索引 → 詞(index2word) 的字典, 以及每個詞word2count用來替換稀疏詞彙。
SOS_token = 0
EOS_token = 1
class Lang:
def __init__(self, name):
self.name = name
self.word2index = {}
self.word2count = {}
self.index2word = {0: "SOS", 1: "EOS"}
self.n_words = 2 # Count SOS and EOS
def addSentence(self, sentence):
for word in sentence.split(' '):
self.addWord(word)
def addWord(self, word):
if word not in self.word2index:
self.word2index[word] = self.n_words
self.word2count[word] = 1
self.index2word[self.n_words] = word
self.n_words += 1
else:
self.word2count[word] += 1
這些檔案全部採用Unicode編碼，為了簡化起見，我們將Unicode字元轉換成ASCII編碼、所有內容小寫、並修剪大部分標點符號。
def unicodeToAscii(s):
return ''.join(
c for c in unicodedata.normalize('NFD', s)
if unicodedata.category(c) != 'Mn'
)
def normalizeString(s):
s = unicodeToAscii(s.lower().strip())
s = re.sub(r"([.!?])", r" \1", s)
s = re.sub(r"[^a-zA-Z.!?]+", r" ", s)
return s
我們將按行分開並將每一行分成兩列來讀取檔案。這些檔案都是英語 -> 其他語言，所以如果我們想從其他語言翻譯 -> 英語，新增reverse標誌來翻轉詞語對。
def readLangs(lang1, lang2, reverse=False):
print("Reading lines...")
# Read the file and split into lines
lines = open('data/%s-%s.txt' % (lang1, lang2), encoding='utf-8').\
read().strip().split('\n')
# Split every line into pairs and normalize
pairs = [[normalizeString(s) for s in l.split('\t')] for l in lines]
# Reverse pairs, make Lang instances
if reverse:
pairs = [list(reversed(p)) for p in pairs]
input_lang = Lang(lang2)
output_lang = Lang(lang1)
else:
input_lang = Lang(lang1)
output_lang = Lang(lang2)
return input_lang, output_lang, pairs
為方便訓練，句子的最大長度是10個單詞(包括標點符號），同時我們將那些翻譯為“I am”或“he is”等形式的句子進行了修改(考慮到之前清除的標點符號——'）
MAX_LENGTH = 10
eng_prefixes = (
"i am ", "i m ",
"he is", "he s ",
"she is", "she s ",
"you are", "you re ",
"we are", "we re ",
"they are", "they re "
)
def filterPair(p):
return len(p[0].split(' ')) < MAX_LENGTH and \
len(p[1].split(' ')) < MAX_LENGTH and \
p[1].startswith(eng_prefixes)
def filterPairs(pairs):
return [pair for pair in pairs if filterPair(pair)]
完整的資料準備過程：
- 按行讀取文字檔案，將行拆分成對
- 規範文字，按長度和內容過濾
- 從句子中成對列出單詞列表
def prepareData(lang1, lang2, reverse=False):
input_lang, output_lang, pairs = readLangs(lang1, lang2, reverse)
print("Read %s sentence pairs" % len(pairs))
pairs = filterPairs(pairs)
print("Trimmed to %s sentence pairs" % len(pairs))
print("Counting words...")
for pair in pairs:
input_lang.addSentence(pair[0])
output_lang.addSentence(pair[1])
print("Counted words:")
print(input_lang.name, input_lang.n_words)
print(output_lang.name, output_lang.n_words)
return input_lang, output_lang, pairs
input_lang, output_lang, pairs = prepareData('eng', 'fra', True)
print(random.choice(pairs))
輸出：
Reading lines... Read 135842 sentence pairs Trimmed to 10599 sentence pairs Counting words... Counted words: fra 4345 eng 2803 ['ils ne sont pas encore chez eux .', 'they re not home yet .']
seq2seq
編碼器
input：torch.size([1])
hidden:torch.size([1,1,hidden_size])
輸入input的seq_len為1，且batch_size也為1，所以input進行詞嵌入後維度變為torch.size([1,hidden_size])，經過view(1,1,-1)變為torch.size([1,1,hidden_size])，之所以這麼做，是因為後面要放入GRU訓練。
output：torch.size([1,1,hidden_size])
hidden：torch.size([1,1,hidden_size])
class EncoderRNN(nn.Module):
def __init__(self, input_size, hidden_size):
super(EncoderRNN, self).__init__()
self.hidden_size = hidden_size
self.embedding = nn.Embedding(input_size, hidden_size)
self.gru = nn.GRU(hidden_size, hidden_size)
def forward(self, input, hidden):
embedded = self.embedding(input).view(1, 1, -1)
output = embedded
output, hidden = self.gru(output, hidden)
return output, hidden
def initHidden(self):
return torch.zeros(1, 1, self.hidden_size, device=device)
簡單解碼器
在最簡單的seq2seq解碼器中，我們只使用編碼器的最後輸出。這最後一個輸出有時稱為上下文向量因為它從整個序列中編碼上下文。該上下文向量用作解碼器的初始隱藏狀態。
在解碼的每一步,解碼器都被賦予一個輸入指令和隱藏狀態. 初始輸入指令字串開始的<SOS>指令,第一個隱藏狀態是上下文向量(編碼器的最後隱藏狀態).
class DecoderRNN(nn.Module):
def __init__(self, hidden_size, output_size):
super(DecoderRNN, self).__init__()
self.hidden_size = hidden_size
self.embedding = nn.Embedding(output_size, hidden_size)
self.gru = nn.GRU(hidden_size, hidden_size)
self.out = nn.Linear(hidden_size, output_size)
self.softmax = nn.LogSoftmax(dim=1)
def forward(self, input, hidden):
output = self.embedding(input).view(1, 1, -1)
output = F.relu(output)
output, hidden = self.gru(output, hidden)
output = self.softmax(self.out(output[0]))
return output, hidden
def initHidden(self):
return torch.zeros(1, 1, self.hidden_size, device=device)
帶注意力機制的解碼器
如果僅在編碼器和解碼器之間傳遞上下文向量,則該單個向量承擔編碼整個句子的負擔.
注意力機制允許解碼器網路針對解碼器自身輸出的每一步”聚焦”編碼器輸出的不同部分. 首先我們計算一組注意力權重. 這些將被乘以編碼器輸出向量獲得加權的組合. 結果(在程式碼中為attn_applied) 應該包含關於輸入序列的特定部分的資訊, 從而幫助解碼器選擇正確的輸出單詞.
注意權值的計算是用另一個前饋層attn進行的, 將解碼器的輸入和隱藏層狀態作為輸入. 由於訓練資料中的輸入序列(語句）長短不一,為了實際建立和訓練此層, 我們必須選擇最大長度的句子(輸入長度,用於編碼器輸出),以適用於此層. 最大長度的句子將使用所有注意力權重,而較短的句子只使用前幾個.
這裡的input和encoder一樣，seq_len = batch_size = 1，詞嵌入後維度變為torch.size([1,hidden_size])，經過view(1,1,-1)變為torch.size([1,1,hidden_size])。
embedded[0]和之前的輸出hidden[0]都是torch.size([1,hidden_size])，拼接在一起後變成torch.size([1,hidden_size*2])，之後經過attn這個全連線層，經過該層之後輸出維度為max_length，因為句子的長度最大為max_length，在encoder中每個單詞都有output，該輸出經過softmax後就變成了對每個output的權重。
接下來權重乘以encoder每個單詞輸出的值，在這裡encoder_outputs的維度為torch.size([max_length，hidden_size])，如果輸出的句子的長度小於max_length，則多出來的那部分為0。得到attn_applied的維度為torch.size([1,1,hidden_size])。
接下來embedded[0]和attn_applied[0]拼接後經過一個全連線層。
class AttnDecoderRNN(nn.Module):
def __init__(self, hidden_size, output_size, dropout_p=0.1, max_length=MAX_LENGTH):
super(AttnDecoderRNN, self).__init__()
self.hidden_size = hidden_size
self.output_size = output_size
self.dropout_p = dropout_p
self.max_length = max_length
self.embedding = nn.Embedding(self.output_size, self.hidden_size)
self.attn = nn.Linear(self.hidden_size * 2, self.max_length)
self.attn_combine = nn.Linear(self.hidden_size * 2, self.hidden_size)
self.dropout = nn.Dropout(self.dropout_p)
self.gru = nn.GRU(self.hidden_size, self.hidden_size)
self.out = nn.Linear(self.hidden_size, self.output_size)
def forward(self, input, hidden, encoder_outputs):
embedded = self.embedding(input).view(1, 1, -1)
embedded = self.dropout(embedded)
attn_weights = F.softmax(
self.attn(torch.cat((embedded[0], hidden[0]), 1)), dim=1)
attn_applied = torch.bmm(attn_weights.unsqueeze(0),
encoder_outputs.unsqueeze(0))
output = torch.cat((embedded[0], attn_applied[0]), 1)
output = self.attn_combine(output).unsqueeze(0)
output = F.relu(output)
output, hidden = self.gru(output, hidden)
output = F.log_softmax(self.out(output[0]), dim=1)
return output, hidden, attn_weights
def initHidden(self):
return torch.zeros(1, 1, self.hidden_size, device=device)
模型訓練
為了訓練,對於每一對我們都需要輸入張量(輸入句子中的詞的索引)和 目標張量(目標語句中的詞的索引). 在建立這些向量時,我們會將EOS標記新增到兩個序列中。
def indexesFromSentence(lang, sentence):
return [lang.word2index[word] for word in sentence.split(' ')]
def tensorFromSentence(lang, sentence):
indexes = indexesFromSentence(lang, sentence)
indexes.append(EOS_token)
return torch.tensor(indexes, dtype=torch.long, device=device).view(-1, 1)
def tensorsFromPair(pair):
input_tensor = tensorFromSentence(input_lang, pair[0])
target_tensor = tensorFromSentence(output_lang, pair[1])
return (input_tensor, target_tensor)
為了訓練我們通過編碼器執行輸入序列,並跟蹤每個輸出和最新的隱藏狀態. 然後解碼器被賦予<SOS>標誌作為其第一個輸入, 並將編碼器的最後一個隱藏狀態作為其第一個隱藏狀態.
“Teacher forcing” 是將實際目標輸出用作每個下一個輸入的概念,而不是將解碼器的 猜測用作下一個輸入.使用“Teacher forcing” 會使其更快地收斂,但是當訓練好的網路被利用時,它可能表現出不穩定性..
您可以觀察“Teacher forcing”網路的輸出，這些網路使用連貫的語法閱讀，但遠離正確的翻譯 - 直覺上它已經學會表示輸出語法，並且一旦老師告訴它前幾個單詞就可以“提取”意義，但是 它沒有正確地學習如何從翻譯中建立句子。
由於PyTorch的autograd給我們的自由,我們可以隨意選擇使用“Teacher forcing”或不使用簡單的if語句. 調高teacher_forcing_ratio來更好地使用它.
teacher_forcing_ratio = 0.5
def train(input_tensor, target_tensor, encoder, decoder, encoder_optimizer, decoder_optimizer, criterion, max_length=MAX_LENGTH):
encoder_hidden = encoder.initHidden()
encoder_optimizer.zero_grad()
decoder_optimizer.zero_grad()
input_length = input_tensor.size(0)
target_length = target_tensor.size(0)
encoder_outputs = torch.zeros(max_length, encoder.hidden_size, device=device)
loss = 0
for ei in range(input_length):
encoder_output, encoder_hidden = encoder(
input_tensor[ei], encoder_hidden)
encoder_outputs[ei] = encoder_output[0, 0]
decoder_input = torch.tensor([[SOS_token]], device=device)
decoder_hidden = encoder_hidden
use_teacher_forcing = True if random.random() < teacher_forcing_ratio else False
if use_teacher_forcing:
# Teacher forcing: 將目標作為下一個輸入
for di in range(target_length):
decoder_output, decoder_hidden, decoder_attention = decoder(
decoder_input, decoder_hidden, encoder_outputs)
loss += criterion(decoder_output, target_tensor[di])
decoder_input = target_tensor[di] # Teacher forcing
else:
# 不適用 teacher forcing: 使用自己的預測作為下一個輸入
for di in range(target_length):
decoder_output, decoder_hidden, decoder_attention = decoder(
decoder_input, decoder_hidden, encoder_outputs)
topv, topi = decoder_output.topk(1)
decoder_input = topi.squeeze().detach() # detach from history as input
loss += criterion(decoder_output, target_tensor[di])
if decoder_input.item() == EOS_token:
break
loss.backward()
encoder_optimizer.step()
decoder_optimizer.step()
return loss.item() / target_length
這是一個幫助函式，用於在給定當前時間和進度%的情況下列印經過的時間和估計的剩餘時間。
import time
import math
def asMinutes(s):
m = math.floor(s / 60)
s -= m * 60
return '%dm %ds' % (m, s)
def timeSince(since, percent):
now = time.time()
s = now - since
es = s / (percent)
rs = es - s
return '%s (- %s)' % (asMinutes(s), asMinutes(rs))
整個訓練過程如下所示:
- 啟動計時器
- 初始化優化器和準則
- 建立一組訓練隊
- 為進行繪圖啟動空損失陣列
之後我們多次呼叫train函式，偶爾列印進度 (樣本的百分比，到目前為止的時間，狙擊的時間) 和平均損失。
def trainIters(encoder, decoder, n_iters, print_every=1000, plot_every=100, learning_rate=0.01):
start = time.time()
plot_losses = []
print_loss_total = 0 # Reset every print_every
plot_loss_total = 0 # Reset every plot_every
encoder_optimizer = optim.SGD(encoder.parameters(), lr=learning_rate)
decoder_optimizer = optim.SGD(decoder.parameters(), lr=learning_rate)
training_pairs = [tensorsFromPair(random.choice(pairs))
for i in range(n_iters)]
criterion = nn.NLLLoss()
for iter in range(1, n_iters + 1):
training_pair = training_pairs[iter - 1]
input_tensor = training_pair[0]
target_tensor = training_pair[1]
loss = train(input_tensor, target_tensor, encoder,
decoder, encoder_optimizer, decoder_optimizer, criterion)
print_loss_total += loss
plot_loss_total += loss
if iter % print_every == 0:
print_loss_avg = print_loss_total / print_every
print_loss_total = 0
print('%s (%d %d%%) %.4f' % (timeSince(start, iter / n_iters),
iter, iter / n_iters * 100, print_loss_avg))
if iter % plot_every == 0:
plot_loss_avg = plot_loss_total / plot_every
plot_losses.append(plot_loss_avg)
plot_loss_total = 0
showPlot(plot_losses)
繪製結果
使用matplotlib完成繪圖，使用plot_losses儲存訓練時的陣列。
import matplotlib.pyplot as plt
plt.switch_backend('agg')
import matplotlib.ticker as ticker
import numpy as np
def showPlot(points):
plt.figure()
fig, ax = plt.subplots()
# 該定時器用於定時記錄時間
loc = ticker.MultipleLocator(base=0.2)
ax.yaxis.set_major_locator(loc)
plt.plot(points)
評估
評估與訓練大部分相同,但沒有目標,因此我們只是將解碼器的每一步預測反饋給它自身. 每當它預測到一個單詞時,我們就會將它新增到輸出字串中,並且如果它預測到我們在那裡停止的EOS指令. 我們還儲存解碼器的注意力輸出以供稍後顯示.
def evaluate(encoder, decoder, sentence, max_length=MAX_LENGTH):
with torch.no_grad():
input_tensor = tensorFromSentence(input_lang, sentence)
input_length = input_tensor.size()[0]
encoder_hidden = encoder.initHidden()
encoder_outputs = torch.zeros(max_length, encoder.hidden_size, device=device)
for ei in range(input_length):
encoder_output, encoder_hidden = encoder(input_tensor[ei],
encoder_hidden)
encoder_outputs[ei] += encoder_output[0, 0]
decoder_input = torch.tensor([[SOS_token]], device=device) # SOS
decoder_hidden = encoder_hidden
decoded_words = []
decoder_attentions = torch.zeros(max_length, max_length)
for di in range(max_length):
decoder_output, decoder_hidden, decoder_attention = decoder(
decoder_input, decoder_hidden, encoder_outputs)
decoder_attentions[di] = decoder_attention.data
topv, topi = decoder_output.data.topk(1)
if topi.item() == EOS_token:
decoded_words.append('<EOS>')
break
else:
decoded_words.append(output_lang.index2word[topi.item()])
decoder_input = topi.squeeze().detach()
return decoded_words, decoder_attentions[:di + 1]
我們可以從訓練集中對隨機句子進行評估，並打印出輸入、目標和輸出，從而做出一些主觀的質量判斷：
def evaluateRandomly(encoder, decoder, n=10):
for i in range(n):
pair = random.choice(pairs)
print('>', pair[0])
print('=', pair[1])
output_words, attentions = evaluate(encoder, decoder, pair[0])
output_sentence = ' '.join(output_words)
print('<', output_sentence)
print('')
訓練和評估
有了所有這些幫助函式(它看起來像是額外的工作，但它使執行多個實驗更容易)，我們實際上可以初始化一個網路並開始訓練。
請記住輸入句子被嚴重過濾, 對於這個小資料集,我們可以使用包含256個隱藏節點 和單個GRU層的相對較小的網路.在MacBook CPU上約40分鐘後,我們會得到一些合理的結果.
注
如果你執行這個筆記本，你可以訓練，中斷核心，評估，並在以後繼續訓練。 註釋編碼器和解碼器初始化的行並再次執行trainIters.
hidden_size = 256 encoder1 = EncoderRNN(input_lang.n_words, hidden_size).to(device) attn_decoder1 = AttnDecoderRNN(hidden_size, output_lang.n_words, dropout_p=0.1).to(device) trainIters(encoder1, attn_decoder1, 75000, print_every=5000)
參考