第N8周:seq2seq翻譯實戰-Pytorch復現

🍨 本文為🔗365天深度學習訓練營中的學習記錄博客
🍖 原作者：K同學啊 | 接輔導、項目定制

一、前期準備

from __future__ import unicode_literals, print_function, division
from io import open
import unicodedata
import string
import re
import randomimport torch
import torch.nn as nn
from torch import optim
import torch.nn.functional as Fdevice = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(device)

1、搭建語言類

SOS_token = 0
EOS_token = 1# 語言類，方便對語料庫進行操作
class Lang:def __init__(self, name):self.name = nameself.word2index = {}self.word2count = {}self.index2word = {0: "SOS", 1: "EOS"}self.n_words    = 2  # Count SOS and EOSdef 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_wordsself.word2count[word] = 1self.index2word[self.n_words] = wordself.n_words += 1else:self.word2count[word] += 1

定義了一個名為Lang的類，用于處理語料庫。Lang類包含了兩個函數，addSentence和addWord，用于向語料庫中添加句子和單詞。類中包含了一些屬性，word2index、word2count、index2word、n_words，分別用于存儲單詞到索引的映射、單詞累計次數、索引到單詞的映射以及單詞總數。

2、文本處理函數?

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

unicodeToAscii 函數的通過使用 unicodedata.normalize('NFD', s) 將字符串進行規范化分解，然后通過列表推導式保留所有不屬于 'Mn' 類別的字符，最后將這些字符拼接成一個新的字符串返回。

normalizeString 函數先調用 unicodeToAscii 函數將輸入的字符串轉換為 ASCII 字符串，然后使用正則表達式替換掉所有的句號、感嘆號和問號，以及所有非字母、非空格、非句號、非感嘆號、非問號的字符，最后返回處理后的字符串。

3、文件讀取函數?

def readLangs(lang1,lang2,reverse=False):print("Reading lines...")lines = open('D:/%s-%s.txt'%(lang1,lang2),encoding='utf-8').\read().strip().split('\n')pairs = [[normalizeString(s) for s in l.split('\t')] for l in lines]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

readLangs接受三個參數：lang1、lang2和reverse。函數的主要功能是從一個文本文件中讀取語言對數據，并根據需要對數據進行預處理。

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):# 選取僅僅包含 eng_prefixes 開頭的語料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,pairsinput_lang,output_lang,pairs = prepareData('eng','fra',True)
print(random.choice(pairs))

二、Seq2Seq模型?

1.編碼器（Encoder）

class EncoderRNN(nn.Module):def __init__(self, input_size, hidden_size):super(EncoderRNN, self).__init__()self.hidden_size = hidden_sizeself.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         = embeddedoutput, hidden = self.gru(output, hidden)return output, hiddendef initHidden(self):return torch.zeros(1, 1, self.hidden_size, device=device)

2.解碼器（Decoder）

class DecoderRNN(nn.Module):def __init__(self, hidden_size, output_size):super(DecoderRNN, self).__init__()self.hidden_size = hidden_sizeself.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, hiddendef initHidden(self):return torch.zeros(1, 1, self.hidden_size, device=device)

三、訓練

1.數據預處理

#將句子中的每個單詞轉換為對應的索引值，并將這些索引值存儲在一個列表中。
def indexesFromSentence(lang, sentence):return [lang.word2index[word] for word in sentence.split(' ')]# 將數字化的文本，轉化為tensor數據
def tensorFromSentence(lang, sentence):indexes = indexesFromSentence(lang, sentence)indexes.append(EOS_token)return torch.tensor(indexes, dtype=torch.long, device=device).view(-1, 1)# 輸入pair文本，輸出預處理好的數據
def tensorsFromPair(pair):input_tensor  = tensorFromSentence(input_lang, pair[0])target_tensor = tensorFromSentence(output_lang, pair[1])return (input_tensor, target_tensor)

2.訓練函數

teacher_forcing_ratio = 0.5def train(input_tensor, target_tensor, encoder, decoder, encoder_optimizer, decoder_optimizer, criterion, max_length=MAX_LENGTH):# 編碼器初始化encoder_hidden = encoder.initHidden()# grad屬性歸零encoder_optimizer.zero_grad()decoder_optimizer.zero_grad()input_length  = input_tensor.size(0)target_length = target_tensor.size(0)# 用于創建一個指定大小的全零張量（tensor），用作默認編碼器輸出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_hiddenuse_teacher_forcing = True if random.random() < teacher_forcing_ratio else False# 將編碼器處理好的輸出送入解碼器if use_teacher_forcing:# Teacher forcing: Feed the target as the next inputfor di in range(target_length):decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)loss         += criterion(decoder_output, target_tensor[di])decoder_input = target_tensor[di]  # Teacher forcingelse:# Without teacher forcing: use its own predictions as the next inputfor di in range(target_length):decoder_output, decoder_hidden = decoder(decoder_input, decoder_hidden)topv, topi    = decoder_output.topk(1)decoder_input = topi.squeeze().detach()  # detach from history as inputloss         += criterion(decoder_output, target_tensor[di])if decoder_input.item() == EOS_token:breakloss.backward()encoder_optimizer.step()decoder_optimizer.step()return loss.item() / target_length

在序列生成的任務中，如機器翻譯或文本生成，解碼器（decoder）的輸入通常是由解碼器自己生成的預測結果，即前一個時間步的輸出。然而，這種自回歸方式可能存在一個問題，即在訓練過程中，解碼器可能會產生累積誤差，并導致輸出與目標序列逐漸偏離。
為了解決這個問題，引入了一種稱為"Teacher Forcing"的技術。在訓練過程中，Teacher Forcing將目標序列的真實值作為解碼器的輸入，而不是使用解碼器自己的預測結果。這樣可以提供更準確的指導信號，幫助解碼器更快地學習到正確的輸出。
在這段代碼中，use_teacher_forcing變量用于確定解碼器在訓練階段使用何種策略作為下一個輸入。
當use_teacher_forcing為True時，采用"Teacher Forcing"的策略，即將目標序列中的真實標簽作為解碼器的下一個輸入。而當use_teacher_forcing為False時，采用"Without Teacher Forcing"的策略，即將解碼器自身的預測作為下一個輸入。

import time
import mathdef asMinutes(s):m = math.floor(s / 60)s -= m * 60return '%dm %ds' % (m, s)def timeSince(since, percent):now = time.time()s = now - sincees = s / (percent)rs = es - sreturn '%s (- %s)' % (asMinutes(s), asMinutes(rs))

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_everyplot_loss_total  = 0  # Reset every plot_everyencoder_optimizer = optim.SGD(encoder.parameters(), lr=learning_rate)decoder_optimizer = optim.SGD(decoder.parameters(), lr=learning_rate)# 在 pairs 中隨機選取 n_iters 條數據用作訓練集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 += lossplot_loss_total  += lossif iter % print_every == 0:print_loss_avg   = print_loss_total / print_everyprint_loss_total = 0print('%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_everyplot_losses.append(plot_loss_avg)plot_loss_total = 0return plot_losses

四、訓練與評估

hidden_size   = 256
encoder1      = EncoderRNN(input_lang.n_words, hidden_size).to(device)
attn_decoder1 = DecoderRNN(hidden_size, output_lang.n_words).to(device)plot_losses = trainIters(encoder1, attn_decoder1, 100000, print_every=5000)

import matplotlib.pyplot as plt
#隱藏警告
import warnings
warnings.filterwarnings("ignore")               # 忽略警告信息
# plt.rcParams['font.sans-serif']    = ['SimHei'] # 用來正常顯示中文標簽
plt.rcParams['axes.unicode_minus'] = False      # 用來正常顯示負號
plt.rcParams['figure.dpi']         = 100        # 分辨率epochs_range = range(len(plot_losses))plt.figure(figsize=(8, 3))plt.subplot(1, 1, 1)
plt.plot(epochs_range, plot_losses, label='Training Loss')
plt.legend(loc='upper right')
plt.title('Training Loss')
plt.show()