1. Sequence-to-Sequence 簡介

大多數(shù)常見的 sequence-to-sequence (seq2seq) model 為 encoder-decoder model，主要由兩個(gè)部分組成，分別是 Encoder 和 Decoder，而這兩個(gè)部分大多數(shù)是由 recurrent neural network (RNN) 實(shí)現(xiàn)。

Encoder 是將一連串的輸入，如文字、影片、聲音訊號等，編碼為單個(gè)向量，這個(gè)向量可以想像為整個(gè)輸入的抽象表示，包含了整個(gè)輸入的資訊。
Decoder 是將 Encoder 輸出的向量進(jìn)行逐步解碼，一次輸出一個(gè)結(jié)果，直到將最終的目標(biāo)全部輸出為止，每次輸出會(huì)影響下一個(gè)輸出，一般會(huì)在開始輸入 < BOS > 來表示開始解碼，會(huì)在結(jié)尾出輸出 < EOS > 來表示解碼結(jié)束。

2. 任務(wù)介紹

• 英文翻譯為中文
  • 輸入： 一句英文 （e.g. tom is a student .）
  • 輸出： 中文翻譯 （e.g. 湯姆 是 個(gè) 學(xué)生 。）

3. 實(shí)現(xiàn)過程

首先要做的是下載資料，主要是用來下載本次任務(wù)需要的數(shù)據(jù)集

!gdown --id '1r4px0i-NcrnXy1-tkBsIwvYwbWnxAhcg' --output data.tar.gz !tar -zxvf data.tar.gz !mkdir ckpt !ls

之后導(dǎo)入需要用到的包（如果nltk包沒有下載的話，可使用第一段代碼進(jìn)行下載）

!pip3 install --user nltk import torch import torch.nn as nn import torch.optim as optim import torch.nn.functional as F import torch.utils.data as data import torch.utils.data.sampler as sampler import torchvision from torchvision import datasets, transformsimport numpy as np import sys import os import random import jsondevice = torch.device("cuda" if torch.cuda.is_available() else "cpu") # 判斷是用 CPU 還是 GPU 執(zhí)行運(yùn)算

需要注意的是，不同的句子往往有著不同的長度，這無疑給訓(xùn)練帶來了不小的麻煩（因?yàn)?RNN 的輸入維度要進(jìn)行相應(yīng)的改變）。為了解決這個(gè)麻煩，我們使用 <pad> 長度較短的句子進(jìn)行填充。因此這里定義一個(gè)長度轉(zhuǎn)換的類

import numpy as npclass LabelTransform(object):def __init__(self, size, pad):self.size = sizeself.pad = paddef __call__(self, label):label = np.pad(label, (0, (self.size - label.shape[0])), mode='constant', constant_values=self.pad)return label

下一步就是數(shù)據(jù)的準(zhǔn)備了，我們定義一個(gè)Dataset。

• Data (出自manythings 的 cmn-eng):

  • 訓(xùn)練資料：18000句
  • 驗(yàn)證資料： 500句
  • 測試資料： 2636句

• 資料預(yù)處理:

  • 英文：
    • 用 subword-nmt 套件將word轉(zhuǎn)為subword
    • 建立字典：取出標(biāo)簽中出現(xiàn)頻率高于預(yù)定閾值的subword
  • 中文：
    • 用 jieba 將中文句子進(jìn)行斷句
    • 建立字典：取出標(biāo)簽中出現(xiàn)頻率高于預(yù)定閾值的詞
  • 特殊字元： < PAD >, < BOS >, < EOS >, < UNK >
    • < PAD > ：無意義，將句子拓展到相同長度
    • < BOS > ：Begin of sentence, 開始字元
    • < EOS > ：End of sentence, 結(jié)尾字元
    • < UNK > ：單字沒有出現(xiàn)在字典里的字
  • 將字典里出現(xiàn)的 subword (詞) 用一個(gè)整數(shù)表示，分為英文和中文的字典，方便之后轉(zhuǎn)化為 one-hot vector

import re import jsonclass EN2CNDataset(data.Dataset):def __init__(self, root, max_output_len, set_name):self.root = rootself.word2int_cn, self.int2word_cn = self.get_dictionary('cn')self.word2int_en, self.int2word_en = self.get_dictionary('en')# 載入資料self.data = []with open(os.path.join(self.root, f'{set_name}.txt'), "r") as f:for line in f:self.data.append(line)print (f'{set_name} dataset size: {len(self.data)}')self.cn_vocab_size = len(self.word2int_cn)self.en_vocab_size = len(self.word2int_en)self.transform = LabelTransform(max_output_len, self.word2int_en['<PAD>'])def get_dictionary(self, language):# 載入字典with open(os.path.join(self.root, f'word2int_{language}.json'), "r") as f:word2int = json.load(f)with open(os.path.join(self.root, f'int2word_{language}.json'), "r") as f:int2word = json.load(f)return word2int, int2worddef __len__(self):return len(self.data)def __getitem__(self, Index):# 先將中英文詞分開sentences = self.data[Index]sentences = re.split('[\t\n]', sentences)sentences = list(filter(None, sentences))#print (sentences)assert len(sentences) == 2# 特殊字元BOS = self.word2int_en['<BOS>']EOS = self.word2int_en['<EOS>']UNK = self.word2int_en['<UNK>']# 在開頭添加 <BOS>，在結(jié)尾添加 <EOS> ，不在字典的 subword (詞) 用 <UNK> 取代en, cn = [BOS], [BOS]# 將句子拆解為 subword 并轉(zhuǎn)為整數(shù)sentence = re.split(' ', sentences[0])sentence = list(filter(None, sentence))#print (f'en: {sentence}')for word in sentence:en.append(self.word2int_en.get(word, UNK))en.append(EOS)# 將句子拆解為 subword 并轉(zhuǎn)為整數(shù)# e.g. < BOS >, we, are, friends, < EOS > --> 1, 28, 29, 205, 2sentence = re.split(' ', sentences[1])sentence = list(filter(None, sentence))#print (f'cn: {sentence}')for word in sentence:cn.append(self.word2int_cn.get(word, UNK))cn.append(EOS)en, cn = np.asarray(en), np.asarray(cn)# 用 <PAD> 將將句子拓展到相同長度en, cn = self.transform(en), self.transform(cn)en, cn = torch.LongTensor(en), torch.LongTensor(cn)return en, cn

接下來就是構(gòu)建自己的模型

Encoder

• seq2seq模型的編碼器為RNN。對于每個(gè)輸入，Encoder 會(huì)輸出一個(gè)向量和一個(gè)隱藏層狀態(tài)(hidden state)，并將隱藏層狀態(tài)用于下一個(gè)輸入，換句話說，Encoder 會(huì)逐步讀入輸入序列。
• 參數(shù):
  • en_vocab_size 是英文字典的大小，也就是英文的 subword 的個(gè)數(shù)
  • emb_dim 是 embedding 的維度，主要將 one-hot vector 的單詞向量壓縮到指定的維度，可以使用預(yù)先訓(xùn)練好的 word embedding，如 Glove 和 word2vector
  • hid_dim 是 RNN 輸出和隱藏狀態(tài)的維度
  • n_layers 是 RNN 要疊多少層
  • dropout 是決定有多少的機(jī)率將某某個(gè)節(jié)點(diǎn)變?yōu)?0，主要是為了防止 overfitting ，一般來說是在訓(xùn)練集使用，測試集不使用
• Encoder 的輸入和輸出:
  • 輸入:
    • 英文的整數(shù)序列 e.g. 1, 28, 29, 205, 2
  • 輸出:
    • outputs: 最上層 RNN 全部的輸出，可以用 Attention 再進(jìn)行處理
    • hidden: 每層最后的隱藏狀態(tài)，將傳輸?shù)胶竺娴?Decoder 進(jìn)行解碼

class Encoder(nn.Module):def __init__(self, en_vocab_size, emb_dim, hid_dim, n_layers, dropout):super().__init__()self.embedding = nn.Embedding(en_vocab_size, emb_dim)self.hid_dim = hid_dimself.n_layers = n_layersself.rnn = nn.GRU(emb_dim, hid_dim, n_layers, dropout=dropout, batch_first=True, bidirectional=True)self.dropout = nn.Dropout(dropout)def forward(self, input):# input = [batch size, sequence len, vocab size]embedding = self.embedding(input)outputs, hidden = self.rnn(self.dropout(embedding))# outputs = [batch size, sequence len, hid dim * directions]# hidden = [num_layers * directions, batch size , hid dim]# outputs 是最上層RNN的輸出return outputs, hidden

Decoder

• Decoder 是另一個(gè) RNN，在最簡單的 seq2seq decoder 中，僅使用 Encoder 對每一層最后的隱藏狀態(tài)來進(jìn)行解碼，而這最好的的隱藏狀態(tài)有些被稱為 “content vector”，因?yàn)榭梢韵胂笏鼘φ麄€(gè)前文序列進(jìn)行了編碼， 此 “content vector” 用作 Decoder 的初始隱藏狀態(tài)， 而 Encoder 的輸出通常用于 Attention Mechanism 產(chǎn)生相應(yīng)的 Attention。

• 參數(shù)

  • en_vocab_size 是英文字典的大小，也就是英文的 subword 的個(gè)數(shù)
  • emb_dim 是 embedding 的維度，主要將 one-hot vector 的單詞向量壓縮到指定的維度，可以使用預(yù)先訓(xùn)練好的 word embedding，如 Glove 和 word2vector
  • hid_dim 是 RNN 輸出和隱藏狀態(tài)的維度
  • output_dim 是最終輸出的維度，一般來說是將 hid_dim 轉(zhuǎn)到 one-hot vector 的單詞向量
  • n_layers 是 RNN 要疊多少層
  • dropout 是決定有多少的機(jī)率將某某個(gè)節(jié)點(diǎn)變?yōu)?0，主要是為了防止 overfitting ，一般來說是在訓(xùn)練集使用，測試集不使用
  • isatt 是來決定是否使用 Attention Mechanism

• Decoder 的輸入和輸出:

  • 輸入:
    • 前一次解碼出來的單詞的整數(shù)表示
  • 輸出:
    • hidden: 根據(jù)輸入和前一次的隱藏轉(zhuǎn)態(tài)，現(xiàn)在的隱藏轉(zhuǎn)態(tài)的更新的結(jié)果
    • output: 每個(gè)字有多少概率是這次解碼的結(jié)果

class Decoder(nn.Module):def __init__(self, cn_vocab_size, emb_dim, hid_dim, n_layers, dropout, isatt):super().__init__()self.cn_vocab_size = cn_vocab_sizeself.hid_dim = hid_dim * 2self.n_layers = n_layersself.embedding = nn.Embedding(cn_vocab_size, config.emb_dim)self.isatt = isattself.attention = Attention(hid_dim)# 如果使用 Attention Mechanism 會(huì)使得輸入維度變化，請?jiān)谶@裡修改# e.g. Attention 接在輸入後面會(huì)使得維度變化，所以輸入維度改為# self.input_dim = emb_dim + hid_dim * 2 if isatt else emb_dimself.input_dim = emb_dimself.rnn = nn.GRU(self.input_dim, self.hid_dim, self.n_layers, dropout = dropout, batch_first=True)self.embedding2vocab1 = nn.Linear(self.hid_dim, self.hid_dim * 2)self.embedding2vocab2 = nn.Linear(self.hid_dim * 2, self.hid_dim * 4)self.embedding2vocab3 = nn.Linear(self.hid_dim * 4, self.cn_vocab_size)self.dropout = nn.Dropout(dropout)def forward(self, input, hidden, encoder_outputs):# input = [batch size, vocab size]# hidden = [batch size, n layers * directions, hid dim]# Decoder 只會(huì)是單向，所以 directions=1input = input.unsqueeze(1)embedded = self.dropout(self.embedding(input))# embedded = [batch size, 1, emb dim]if self.isatt:attn = self.attention(encoder_outputs, hidden)# TODO: 在這裡決定如何使用 Attention，e.g. 相加 或是 接在後面， 請注意維度變化output, hidden = self.rnn(embedded, hidden)# output = [batch size, 1, hid dim]# hidden = [num_layers, batch size, hid dim]# 將 RNN 的輸出轉(zhuǎn)為每個(gè)詞出現(xiàn)的機(jī)率output = self.embedding2vocab1(output.squeeze(1))output = self.embedding2vocab2(output)prediction = self.embedding2vocab3(output)# prediction = [batch size, vocab size]return prediction, hidden

Attention

• 當(dāng)輸入過長時(shí)，或是單獨(dú)靠 “content vector” 無法獲取整個(gè)輸入的意思時(shí)，用 Attention Mechanism 來提供 Decoder 更多的資訊

• 主要是根據(jù)現(xiàn)在 Decoder hidden state ，去計(jì)算在 Encoder outputs 中，那些與其有較高的關(guān)系，根據(jù)關(guān)系的數(shù)值來決定傳給 Decoder 哪些額外的資訊

• 常見 Attention 的操作是用 Neural Network / Dot Product 來計(jì)算 Decoder hidden state 和 Encoder outputs 之間的關(guān)系，再對所有算出來的數(shù)值做 softmax ，最后根據(jù)過完 softmax 的值對 Encoder outputs 做 weight sum

• 李宏毅老師的課程在此處并沒有給出具體的代碼，需要大家自己補(bǔ)充。大家可以參考這篇文章 Seq2Seq (Attention) 的 PyTorch 實(shí)現(xiàn) 或者B站的視頻 PyTorch35——基于注意力機(jī)制的Seq2Seq的PyTorch實(shí)現(xiàn)示例。

class Attention(nn.Module):def __init__(self, hid_dim):super(Attention, self).__init__()self.hid_dim = hid_dimdef forward(self, encoder_outputs, decoder_hidden):# encoder_outputs = [batch size, sequence len, hid dim * directions]# decoder_hidden = [num_layers, batch size, hid dim]# 一般來說是取最後一層的 hidden state 來做 attention######### TODO #########attention=Nonereturn attention

Seq2seq模型

• 由 Encoder 和 Decoder 組成
• 接收輸入并傳給 Encoder
• 將 Encoder 的輸出傳給 Decoder
• 不斷地將 Decoder 的輸出傳回 Decoder ，進(jìn)行解碼
• 當(dāng)解碼完成，將 Decoder 的輸出傳回

class Seq2Seq(nn.Module):def __init__(self, encoder, decoder, device):super().__init__()self.encoder = encoderself.decoder = decoderself.device = deviceassert encoder.n_layers == decoder.n_layers, \"Encoder and decoder must have equal number of layers!"def forward(self, input, target, teacher_forcing_ratio):# input = [batch size, input len, vocab size]# target = [batch size, target len, vocab size]# teacher_forcing_ratio 是有多少機(jī)率使用正確答案來訓(xùn)練batch_size = target.shape[0]target_len = target.shape[1]vocab_size = self.decoder.cn_vocab_size# 準(zhǔn)備一個(gè)儲(chǔ)存空間來儲(chǔ)存輸出outputs = torch.zeros(batch_size, target_len, vocab_size).to(self.device)# 將輸入放入 Encoderencoder_outputs, hidden = self.encoder(input)# Encoder 最後的隱藏層(hidden state) 用來初始化 Decoder# encoder_outputs 主要是使用在 Attention# 因?yàn)?Encoder 是雙向的RNN，所以需要將同一層兩個(gè)方向的 hidden state 接在一起# hidden = [num_layers * directions, batch size , hid dim] --> [num_layers, directions, batch size , hid dim]hidden = hidden.view(self.encoder.n_layers, 2, batch_size, -1)hidden = torch.cat((hidden[:, -2, :, :], hidden[:, -1, :, :]), dim=2)# 取的 <BOS> tokeninput = target[:, 0]preds = []for t in range(1, target_len):output, hidden = self.decoder(input, hidden, encoder_outputs)outputs[:, t] = output# 決定是否用正確答案來做訓(xùn)練teacher_force = random.random() <= teacher_forcing_ratio# 取出機(jī)率最大的單詞top1 = output.argmax(1)# 如果是 teacher force 則用正解訓(xùn)練，反之用自己預(yù)測的單詞做預(yù)測input = target[:, t] if teacher_force and t < target_len else top1preds.append(top1.unsqueeze(1))preds = torch.cat(preds, 1)return outputs, predsdef inference(self, input, target):######### TODO ########## 在這裡實(shí)施 Beam Search# 此函式的 batch size = 1 # input = [batch size, input len, vocab size]# target = [batch size, target len, vocab size]batch_size = input.shape[0]input_len = input.shape[1] # 取得最大字?jǐn)?shù)vocab_size = self.decoder.cn_vocab_size# 準(zhǔn)備一個(gè)儲(chǔ)存空間來儲(chǔ)存輸出outputs = torch.zeros(batch_size, input_len, vocab_size).to(self.device)# 將輸入放入 Encoderencoder_outputs, hidden = self.encoder(input)# Encoder 最後的隱藏層(hidden state) 用來初始化 Decoder# encoder_outputs 主要是使用在 Attention# 因?yàn)?Encoder 是雙向的RNN，所以需要將同一層兩個(gè)方向的 hidden state 接在一起# hidden = [num_layers * directions, batch size , hid dim] --> [num_layers, directions, batch size , hid dim]hidden = hidden.view(self.encoder.n_layers, 2, batch_size, -1)hidden = torch.cat((hidden[:, -2, :, :], hidden[:, -1, :, :]), dim=2)# 取的 <BOS> tokeninput = target[:, 0]preds = []for t in range(1, input_len):output, hidden = self.decoder(input, hidden, encoder_outputs)# 將預(yù)測結(jié)果存起來outputs[:, t] = output# 取出機(jī)率最大的單詞top1 = output.argmax(1)input = top1preds.append(top1.unsqueeze(1))preds = torch.cat(preds, 1)return outputs, preds

總結(jié)

以上是生活随笔為你收集整理的Pytorch实战_Seq2seq模型的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。