目錄

• 前言
• 1. 數據集
• • 1.1 下載數據集并處理
  • 1.2 將數據集分為source和target
  • 1.3 定義詞匯類
  • 1.4 獲取訓練集
• 2. 定義模型
• • 2.1 導入相關工具包
  • 2.2. 定義Encoder模型
  • 2.3.定義Decoder模型
  • 2.4.定義seq2seq模型
  • 2.5.定義loss
• 3.訓練函數
• 4.預測函數
• 5.測試
• • 5.1定義參數
  • 5.2.訓練
  • 5.3.預測

前言

這兩天學習了RNN，就想實現一下機器翻譯，如果有什么問題，可以隨時交流！

1. 數據集

本文的數據集來自李沐老師的《動手學深度學習》

數據集下載 : https://github.com/codefly-xtl/translation/tree/main/data

首先看看數據集啥樣子：左邊為英語，右邊為法語

1.1 下載數據集并處理

在這一部分，我們首先讀取出數據存放到 raw_data 中，再使用空格替換不間斷字符，再小寫。再對標點符號前添加空格，最后返回數據。

def process_data():# 判斷標點前是否有空格def no_peace(char, pre_char):return char in set(',.!?') and pre_char != ' '# 加載數據with open('./data/fra.txt', encoding='utf-8') as f:raw_data = f.read()# 對數據進行處理:變小寫,在標點符號前插入空格raw_data = raw_data.replace('\u202f', ' ').replace('\xa0', ' ').lower()out = [' ' + char if i > 0 and no_peace(char, raw_data[i - 1]) else char for i, char in enumerate(raw_data)]data = ''.join(out)return data

1.2 將數據集分為source和target

在這一部分，我們獲取原文以及譯文的句子，source存放原文，target存放譯文。根據 \n 取出數據的每一行，再根據 \t 將數據分為原文以及譯文，再分別對原文以及譯文根據 空格 將句子按詞分開組成列表。source例子如下：source = [[‘i’, ‘am’, ‘person’],[‘i’, ‘like’, ‘you’]]

def get_sentence(data):# 存儲兩種語言source = []target = []# 取出每一行for line in data.split('\n'):# 取出每一行的兩個部分parts = line.split('\t')if len(parts) == 2:# 將英語放入sourcesource.append(parts[0].split(' '))# 將法語放入targettarget.append(parts[1].split(' '))# source 的樣子如下：# source = [['i', 'am', 'person'],['i', 'like', 'you']]return source, target

1.3 定義詞匯類

這個類的創建過程如下：

先從句子列表中讀取到每一個單詞，得到all_words

再按單詞出現的頻率排序得到word_preq

接下來就實現index_to_word以及word_to_index，這兩個可以實現單個index和對應的word之間的互轉

接下來實現to_index以及wo_word，這兩個可以實現index列表和對應的word列表之間的互相轉換，例如：index = [88,102,562,4850] 轉為 word = [‘點’,‘個’,‘贊’,‘😀’]，to_index輸出的是tensor格式

為了可以對一句話實現翻譯，因此設計了一個prase的方法，可以將句子轉換為對應的index的tensor格式，返回的數據是二維的，shape為 (batch_size , num+steps)

實現獲取詞匯表大小的**len()**方法

# 詞匯類 class Vocab:# reserved_tokens 是一個預留token，比如預留開始字符<bos> def __init__(self, sentence, min_freq=0, reserved_tokens=None):if reserved_tokens is None:reserved_tokens = []# 取出所有單詞self.all_words = [word for words in sentence for word in words]# 統計詞頻，并按頻率大小從大到小排序self.word_preq = self.get_word_preq()# 首先將<unk>以及預留字放入index_to_word以及word_to_index，index_to_word可以根據index獲得word，word_to_index根據word獲得indexself.index_to_word = ['<unk>'] + reserved_tokensself.word_to_index = {word: index for index, word in enumerate(self.index_to_word)}# 再將所有詞放入index_to_word以及word_to_indexfor word, freq in self.word_preq:if freq < min_freq:breakself.index_to_word.append(word)self.word_to_index[word] = len(self.word_to_index)# 統計詞頻def get_word_preq(self):word_preq = {}for word in self.all_words:if word not in word_preq:word_preq[word] = 1else:word_preq[word] += 1# 排序word_preq = sorted(word_preq.items(), key=lambda x: x[1], reverse=True)return word_preq# 獲取詞數def __len__(self):return len(self.index_to_word)# 將index列表轉為word列表def to_word(self, indexs):return [self.index_to_word[i] for i in indexs]# 將word列表轉為index列表def to_index(self, words):output = []for word in words:if word not in self.index_to_word:output.append(self.word_to_index['<unk>'])else:output.append(self.word_to_index[word])return torch.tensor(output)# 將一句話轉為對應的tensor數據def prase(self, raw_data, num_steps):raw_data = raw_data.replace('\u202f', ' ').replace('\xa0', ' ').lower()out = [' ' + char if i > 0 and no_peace(char, raw_data[i - 1]) else char for i, char in enumerate(raw_data)]data = ''.join(out)source = data.split(' ')source.append('<eos>')source_valid_len =len(source)source_word = truncate_or_pad(source, num_steps)source_index = self.to_index(source_word)print(source_index)return torch.tensor(source_index).unsqueeze(0), torch.tensor(source_valid_len).reshape(-1, 1)

1.4 獲取訓練集

在這一部分，用于獲取訓練集，步驟如下:

首先獲取數據，在將數據分為source_sentences, target_sentences

分別對這上述兩個句子列表形成詞匯表source_Vocab 和target_Vocab

根據句子的多少以及batch_size計算出整個數據集可以形成多少個batch

每一個batch，存放四部分：source_batch，source_len_batch，target_batch，target_len_batch，其中，_batch存放句子，_len_batch用于存放每一個句子中有效長度為多少。

# 用于填充字符或者截斷句子 def truncate_or_pad(line, num_steps):# 例: line = ['i','am','person']# 超出后進行截斷if len(line) > num_steps:return line[:num_steps]# 沒有超出就padfor i in range(num_steps - len(line)):line.append('<pad>')return linedef get_train_iter(batch_size, num_steps):data = process_data()# source_sentences 例: source_sentences = [['i am person'],['i like you']]source_sentences, target_sentences = get_sentence(data)source_Vocab = Vocab(source_sentences, min_freq=0, reserved_tokens=['<pad>', '<bos>', '<eos>'])target_Vocab = Vocab(target_sentences, min_freq=0, reserved_tokens=['<pad>', '<bos>', '<eos>'])database = []batch_num = len(source_sentences) // batch_size# 每一個batch放在database里面for i in range(batch_num):source_batch = []source_len_batch = []target_batch = []target_len_batch = []for j in range(batch_size):# 獲取一個句子以及翻譯source_sentence = source_sentences[i * batch_size + j] + ['<eos>']target_sentence = target_sentences[i * batch_size + j] + ['<eos>']source_valid_len = len(source_sentence)target_valid_len = len(target_sentence)# 將句子變為單詞列表,超過num_steps的截斷,不夠num_steps的補齊source_word = truncate_or_pad(source_sentence, num_steps)target_word = truncate_or_pad(target_sentence, num_steps)# 獲取單詞對應的標號source_index = source_Vocab.to_index(source_word)target_index = target_Vocab.to_index(target_word)# 存放起來source_batch.append(source_index)source_len_batch.append(source_valid_len)target_batch.append(target_index)target_len_batch.append(target_valid_len)source_batch_tensor = torch.stack(source_batch)target_batch_tensor = torch.stack(target_batch)source_len_batch_tensor = torch.tensor(source_len_batch)target_len_batch_tensor = torch.tensor(target_len_batch)database.append((source_batch_tensor, source_len_batch_tensor, target_batch_tensor, target_len_batch_tensor))return database, source_Vocab, target_Vocab

2. 定義模型

2.1 導入相關工具包

import torch from torch import nn import utils

2.2. 定義Encoder模型

class Encoder(nn.Module):def __init__(self, vocab_size, embed_size, num_hiddens, num_layers, bidirectional=False):super().__init__()self.embedding = nn.Embedding(vocab_size, embed_size)self.rnn = nn.LSTM(embed_size, num_hiddens, num_layers, bidirectional=bidirectional)self.num_layers = num_layersself.num_hiddens = num_hiddensself.bidirectional = bidirectionalif bidirectional:# 由于每一層有兩個方向,因此需要將兩個方向進行合并self.linear_hidden = nn.Linear(self.num_hiddens * 2, self.num_hiddens)self.linear_content = nn.Linear(self.num_hiddens * 2, self.num_hiddens)def forward(self, X):X = self.embedding(X)X = X.permute(1, 0, 2)output, state = self.rnn(X)hidden_state, content_state = stateif self.bidirectional:# 將每一層的正反state拼在一起,再放入神經網絡中,使得與decoder的num_hiddens一致hidden_state = torch.cat([hidden_state[:self.num_layers * 2:2, :, :], hidden_state[1:self.num_layers * 2 + 1:2, :, :]], dim=2)content_state = torch.cat([content_state[:self.num_layers * 2:2, :, :], content_state[1:self.num_layers * 2 + 1:2, :, :]], dim=2)hidden_state = self.linear_hidden(hidden_state)content_state = self.linear_content(content_state)return hidden_state, content_state

2.3.定義Decoder模型

class Decoder(nn.Module):def __init__(self, vocab_size, embed_size, num_hiddens, num_layers):super().__init__()self.embedding = nn.Embedding(vocab_size, embed_size)self.rnn = nn.LSTM(embed_size + num_hiddens * 2, num_hiddens, num_layers)self.linear = nn.Linear(num_hiddens, vocab_size)def init_state(self, encoder_output_state):return encoder_output_statedef forward(self, X, state, predict=False):if not predict:X = self.embedding(X).permute(1, 0, 2)# 由于decoder的信息全由encoder的最后一個時間state得到,# 因此最后一個state的最后一層很重要,要盡可能的充分利用,# 因此將最后一個state的最后一層也作為decoder的輸入hidden_state, content_state = statenew_hidden_state = hidden_state[-1].unsqueeze(0).repeat(target.shape[0], 1, 1)new_content_state = content_state[-1].unsqueeze(0).repeat(target.shape[0], 1, 1)X = torch.cat([new_hidden_state, new_content_state, X], dim=2)# X 的shape為:(num_steps, batch_size, decoder_embed_size + encoder_hidden_num * 2)output, state = self.rnn(X, state)output = self.linear(output).permute(1, 0, 2)return output, state

2.4.定義seq2seq模型

class EncoderDecoder(nn.Module):def __init__(self, encoder, decoder):super().__init__()self.encoder = encoderself.decoder = decoderdef forward(self, source, target):encoder_output_state = self.encoder(source)decoder_init_state = self.decoder.init_state(encoder_output_state)return self.decoder(target, decoder_init_state)

2.5.定義loss

由于損失矩陣形狀為 (batch_size, steps_num)，每一個句子后邊有一部分是填充過的，因此不能計算填充數據的損失

class Myloss(nn.CrossEntropyLoss):def value_mask(self, X, valid_len):mask = torch.arange(X.shape[1], dtype=torch.float32, device=X.device)[None, :] > valid_len[:, None]X[mask] = 0return Xdef forward(self, predict, target, valid_len=None):weights = torch.ones_like(target)weights = self.value_mask(weights, valid_len)self.reduction = 'none'unweighted_loss = super().forward(predict.permute(0, 2, 1), target)weighted_loss = unweighted_loss * weightsreturn weighted_loss.mean()

3.訓練函數

def train(net, data_iter, lr, num_epochs, device):net.to(device)optimizer = torch.optim.Adam(net.parameters(), lr=lr)loss = Myloss()net.train()for epoch in range(num_epochs):for batch in data_iter:optimizer.zero_grad()# 將數據放到device上source, source_valid_len, target, target_valid_len = [x.to(device) for x in batch]# 再每一個句子前面添加<bos>的index,bos的index為2bos = torch.tensor([2] * target.shape[0], device=device).reshape(-1, 1)decoder_input = torch.cat([bos, target[:, :-1]], dim=1)# 進行優化Y_hat, _ = net(source, decoder_input)l = loss(Y_hat, target, target_valid_len)l.backward()optimizer.step()print(l)

4.預測函數

def predict(net, source_sentence, source_Vocab, target_Vocab, num_steps, device):# 用于存儲譯文result = []# 原文source, source_valid_len = source_Vocab.prase(source_sentence, num_steps)source, source_valid_len = source.to(device), source_valid_len.to(device)# 獲取最后一個狀態state = net.encoder(source)# 獲取encoder的最后一個state的信息hidden_state, content_state = statenew_hidden_state = hidden_state[-1].unsqueeze(0)new_content_state = content_state[-1].unsqueeze(0)# 初始化decoder的第一個狀態state = net.decoder.init_state(state)# 構造翻譯的第一個詞X = torch.tensor(target_Vocab.word_to_index['<eos>']).reshape(-1, 1).to(device)X = net.decoder.embedding(X).permute(1, 0, 2)X = torch.cat([new_hidden_state, new_content_state, X], dim=2)for i in range(num_steps):# 開啟預測模式,進行預測Y, state = net.decoder(X, state, True)X = Y.argmax(dim=2)# 獲取最大概率的indexpred = X.squeeze(dim=0).type(torch.int32).item()# 如果index為eos,則停止預測if pred == target_Vocab.word_to_index['<eos>']:breakX = net.decoder.embedding(X).permute(1, 0, 2)X = torch.cat([new_hidden_state, new_content_state, X], dim=2)result.append(pred)return ' '.join(target_Vocab.to_word(result))

5.測試

5.1定義參數

batch_size = 64 num_steps = 20 train_iter, source_Vocab, target_Vocab = utils.get_train_iter(batch_size, num_steps) encoder_embed_size = 300 decoder_embed_size = 300 hidden_size = 64 num_layers = 2 encoder = Encoder(len(source_Vocab), decoder_embed_size, hidden_size, num_layers, True) decoder = Decoder(len(target_Vocab), decoder_embed_size, hidden_size, num_layers) net = EncoderDecoder(encoder, decoder) num_epoch = 100 lr = 0.001 device = 'cuda'

5.2.訓練

train(net, train_iter, lr, num_epoch, device) # 顯示如下: tensor(0.0147, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0137, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0139, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0128, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0126, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0126, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0123, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0120, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0128, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0121, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0117, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0122, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0119, device='cuda:0', grad_fn=<MeanBackward0>) tensor(0.0124, device='cuda:0', grad_fn=<MeanBackward0>)

5.3.預測

predict(net, 'He did it just for fun.', source_Vocab, target_Vocab, num_steps, device)

總結

以上是生活随笔為你收集整理的手把手教你使用pytorch实现双向LSTM机器翻译的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。