前言

在上一章【課程總結(jié)】Day17（下）：初始Seq2Seq模型中，我們初步了解了Seq2Seq模型的基本情況及代碼運行效果，本章內(nèi)容將深入了解Seq2Seq模型的代碼，梳理代碼的框架圖、各部分組成部分以及運行流程。

框架圖

工程目錄結(jié)構(gòu)

查看項目目錄結(jié)構(gòu)如下：

seq2seq_demo/
├── data.txt                     # 原始數(shù)據(jù)文件，包含訓(xùn)練或測試數(shù)據(jù)
├── dataloader.py                # 數(shù)據(jù)加載器，負(fù)責(zé)讀取和預(yù)處理數(shù)據(jù)
├── decoder.py                   # 解碼器實現(xiàn)，用于生成輸出序列
├── encoder.py                   # 編碼器實現(xiàn)，將輸入序列編碼為上下文向量
├── main.py                      # 主程序入口，執(zhí)行模型訓(xùn)練和推理
├── seq2seq.py                   # seq2seq 模型的實現(xiàn)，整合編碼器和解碼器
└── tokenizer.py                 # 分詞器實現(xiàn)，將文本轉(zhuǎn)換為模型可處理的格式

查看各個py文件整理關(guān)系圖結(jié)構(gòu)如下：

• main.py 文件是主程序入口，同時其中也定義了 Translation類，用于訓(xùn)練和推理。
• Translation類在 __init__() 方法中調(diào)用 get_tokenizer() 方法實例化tokenizer對象。
• Translation類在 __init__() 方法中調(diào)用 get_model() 實例化seq2seq類對象，進(jìn)而實例化 Encoder 和 Decoder 對象。
• Translation類在 train() 方法中調(diào)用 get_dataloader() 方法實例化dataloader對象。

核心邏輯

初始化過程

• 上述流程中較為重要的代碼主要是 build_dict() 、encoder實例化、decoder實例化初始化過程：

Build_dict()

def build_dict(self):"""構(gòu)建字典"""if os.path.exists(self.saved_dict):self.load()print("加載本地字典成功")returninput_words = {"<UNK>", "<PAD>"}output_words = {"<UNK>", "<PAD>", "<SOS>", "<EOS>"}with open(file=self.data_file, mode="r", encoding="utf8") as f:for line in tqdm(f.readlines()):if line:input_sentence, output_sentence = line.strip().split("\t")input_sentence_words = self.split_input(input_sentence)output_sentence_words = self.split_output(output_sentence)input_words = input_words.union(set(input_sentence_words))output_words = output_words.union(set(output_sentence_words))# 輸入字典self.input_word2idx = {word: idx for idx, word in enumerate(input_words)}self.input_idx2word = {idx: word for word, idx in self.input_word2idx.items()}self.input_dict_len = len(self.input_word2idx)# 輸出字典self.output_word2idx = {word: idx for idx, word in enumerate(output_words)}self.output_idx2word = {idx: word for word, idx in self.output_word2idx.items()}self.output_dict_len = len(self.output_word2idx)# 保存self.save()print("保存字典成功")

代碼解析：

• 首先，判斷本地是否有字典，有的話直接加載；
• 其次，在input_words 和 output_words 集合中添加特殊符號(special tokens)：
  • <UNK>：表示未知單詞，用于表示輸入序列中未在字典中找到的單詞；
  • <PAD>：表示填充符號，用于填充輸入序列和輸出序列，使它們具有相同的長度；
  • <SOS>：表示序列的開始，用于表示輸出序列的起始位置；
  • <EOS>：表示序列的結(jié)束，用于表示輸出序列的結(jié)束位置。
• 然后，讀取data.txt文件，以\t切分?jǐn)?shù)據(jù)并切分單詞：
  • 輸入的英文調(diào)用split_input進(jìn)行預(yù)處理，例如：I’m a student.→[‘i’, ‘m’, ‘a(chǎn)’, ‘student’, ‘.’]
  • 輸出的中文調(diào)用split_output進(jìn)行切分，例如：我愛北京天安門→[‘我’, ‘愛’, ‘北京’, ‘天安門’]
• 最后，調(diào)用self.save() 方法將字典保存到本地文件 self.saved_dict 中。

encoder

import torch
from torch import nnclass Encoder(nn.Module):"""定義一個 編碼器"""def __init__(self, tokenizer):super(Encoder, self).__init__()self.tokenizer = tokenizer# 嵌入層self.embed = nn.Embedding(num_embeddings=self.tokenizer.input_dict_len,embedding_dim=self.tokenizer.input_embed_dim,padding_idx=self.tokenizer.input_word2idx.get("<PAD>"))# GRU單元self.gru = nn.GRU(input_size=self.tokenizer.input_embed_dim,hidden_size=self.tokenizer.input_hidden_size,batch_first=False)def forward(self, x, x_len):# [seq_len, batch_size] --> [seq_len, batch_size, embed_dim]x = self.embed(x)# 壓緊被填充的序列x = nn.utils.rnn.pack_padded_sequence(input=x,lengths=x_len,batch_first=False)out, hn = self.gru(x)# 填充被壓緊的序列out, out_len = nn.utils.rnn.pad_packed_sequence(sequence=out,batch_first=False,padding_value=self.tokenizer.input_word2idx.get("<PAD>"))# out: [seq_len, batch_size, hidden_size]# hn: [1, batch_size, hidden_size]return out, hn

代碼解析：

• encoder是一個典型的RNN結(jié)構(gòu)，其定義了embedding層用于詞嵌入，以及GRU單元進(jìn)行序列處理。
• 在forward方法中，首先將輸入序列進(jìn)行詞嵌入，然后使用pack_padded_sequence將被填充的序列壓緊，以便于GRU單元處理。

decoder

import torch
from torch import nn
import randomdevice = torch.device("cuda" if torch.cuda.is_available() else "cpu")class Decoder(nn.Module):def __init__(self, tokenizer):super(Decoder, self).__init__()self.tokenizer = tokenizer# 嵌入self.embed = nn.Embedding(num_embeddings=self.tokenizer.output_dict_len,embedding_dim=self.tokenizer.output_embed_dim,padding_idx=self.tokenizer.output_word2idx.get("<PAD>"),)# 抽取特征self.gru = nn.GRU(input_size=self.tokenizer.output_embed_dim,hidden_size=self.tokenizer.output_hidden_size,batch_first=False,)# 轉(zhuǎn)換維度，做概率輸出self.fc = nn.Linear(in_features=self.tokenizer.output_hidden_size,out_features=self.tokenizer.output_dict_len,)def forward_step(self, decoder_input, decoder_hidden):"""單步解碼:decoder_input: [1, batch_size]decoder_hidden: [1, batch_size, hidden_size]"""# [1, batch_size] --> [1, batch_size, embedding_dim]decoder_input = self.embed(decoder_input)# 輸入：[1, batch_size, embedding_dim] [1, batch_size, hidden_size]# 輸出：[1, batch_size, hidden_size] [1, batch_size, hidden_size]# 因為只有1步，所以 out 跟 decoder_hidden是一樣的out, decoder_hidden = self.gru(decoder_input, decoder_hidden)# [batch_size, hidden_size]out = out.squeeze(dim=0)# [batch_size, dict_len]out = self.fc(out)# out: [batch_size, dict_len]# decoder_hidden: [1, batch_size, hidden_size]return out, decoder_hiddendef forward(self, encoder_hidden, y, y_len):"""訓(xùn)練時的正向傳播- encoder_hidden: [1, batch_size, hidden_size]- y: [seq_len, batch_size]- y_len: [batch_size]"""# 計算輸出的最大長度（本批數(shù)據(jù)的最大長度）output_max_len = max(y_len.tolist()) + 1# 本批數(shù)據(jù)的批量大小batch_size = encoder_hidden.size(1)# 輸入信號 SOS  讀取第0步，啟動信號# decoder_input: [1, batch_size]# 輸入信號 SOS [1, batch_size]decoder_input = torch.LongTensor([[self.tokenizer.output_word2idx.get("<SOS>")] * batch_size]).to(device=device)# 收集所有的預(yù)測結(jié)果# decoder_outputs: [seq_len, batch_size, dict_len]decoder_outputs = torch.zeros(output_max_len, batch_size, self.tokenizer.output_dict_len)# 隱藏狀態(tài) [1, batch_size, hidden_size]decoder_hidden = encoder_hidden# 手動循環(huán)for t in range(output_max_len):# 輸入：decoder_input: [batch_size, dict_len], decoder_hidden: [1, batch_size, hidden_size]# 返回值：decoder_output_t: [batch_size, dict_len], decoder_hidden: [1, batch_size, hidden_size]decoder_output_t, decoder_hidden = self.forward_step(decoder_input, decoder_hidden)# 填充結(jié)果張量 [seq_len, batch_size, dict_len]decoder_outputs[t, :, :] = decoder_output_t# teacher forcing 教師強(qiáng)迫機(jī)制use_teacher_forcing = random.random() > 0.5# 0.5 概率 實行教師強(qiáng)迫if use_teacher_forcing:# [1, batch_size] 取標(biāo)簽中的下一個詞decoder_input = y[t, :].unsqueeze(0)else:# 取出上一步的推理結(jié)果 [1, batch_size]decoder_input = decoder_output_t.argmax(dim=-1).unsqueeze(0)# decoder_outputs: [seq_len, batch_size, dict_len]return decoder_outputs# ...（其他函數(shù)暫略）

代碼解析：

• decoder定義了三個層：embed（詞嵌入）、gru和fc（全鏈接層）。
• 全鏈接層用于輸出的是字典長度，即每個位置代表著每個字的概率。
• decoder的forward_step方法，用于一步一步地執(zhí)行，屬于手動循環(huán)；forward方法，把所有步都執(zhí)行完進(jìn)行推理，屬于自動循環(huán)。
• 在forward方法中：
  • 首先，計算本批數(shù)據(jù)的最大長度(用于標(biāo)簽對齊)
  • 其次，使用encoder_hidden.size(1)獲取批量大小
  • 然后，增加啟動信號，即<SOS>
  • 然后，準(zhǔn)備全0的張量 decoder_outputs
  • 然后，開始循環(huán)
    • 在循環(huán)每一步中，將輸入和隱藏狀態(tài)傳給forward_step進(jìn)行處理，得到輸出概率decoder_output_t
    • 將結(jié)果概率放在decoder_outputs中
    • 啟用教師強(qiáng)迫機(jī)制(teacher forcing)：
      • 即有50%概率，使用標(biāo)準(zhǔn)答案作為下一步的輸入；
      • 否則，使用上一步的推理結(jié)果中概率最大的詞作為下一步的輸入。
  • 最后，返回結(jié)果概率張量 decoder_outputs

訓(xùn)練過程

• 上述流程中較為重要的代碼主要是 調(diào)用collate_fn、具體訓(xùn)練過程、手動循環(huán)進(jìn)行正向推理

調(diào)用collate_fn

def collate_fn(batch, tokenizer):# 根據(jù) x 的長度來 倒序排列batch = sorted(batch, key=lambda ele: ele[1], reverse=True)# 合并整個批量的每一部分input_sentences, input_sentence_lens, output_sentences, output_sentence_lens = zip(*batch)# 轉(zhuǎn)索引【按本批量最大長度來填充】input_sentence_len = input_sentence_lens[0]input_idxes = []for input_sentence in input_sentences:input_idxes.append(tokenizer.encode_input(input_sentence, input_sentence_len))# 轉(zhuǎn)索引【按本批量最大長度來填充】output_sentence_len = max(output_sentence_lens)output_idxes = []for output_sentence in output_sentences:output_idxes.append(tokenizer.encode_output(output_sentence, output_sentence_len))# 轉(zhuǎn)張量 [seq_len, batch_size]input_idxes = torch.LongTensor(input_idxes).t()output_idxes = torch.LongTensor(output_idxes).t()input_sentence_lens = torch.LongTensor(input_sentence_lens)output_sentence_lens = torch.LongTensor(output_sentence_lens)return input_idxes, input_sentence_lens, output_idxes, output_sentence_lens

代碼解析：

• 當(dāng)文字長度不一樣齊的時候，需要進(jìn)行補充<PAD>，以保持所有序列長度一致

例如：
I’m a student.
I’m OK.
Here is your change.

• 但是補充<PAD>本身對訓(xùn)練過程會造成干擾，所以我們需要采用一種機(jī)制：既保證對齊數(shù)據(jù)批量化訓(xùn)練，又能消除填充對訓(xùn)練過程的影響。
• 這種機(jī)制原理：在訓(xùn)練時知道實際的數(shù)據(jù)長度，這樣在訓(xùn)練時就可以略過<PAD>。
• torch提供了相應(yīng)的API，其大致過程是：
  • 首先，根據(jù) x(上句) 的長度倒序排序
  • 其次，獲取本批量最大的長度
  • 然后，將數(shù)據(jù)填充到本批量最大長度
  • 最后，在返回數(shù)據(jù)時，不知返回數(shù)據(jù)，還會帶著真實長度

具體訓(xùn)練過程

    # （其他部分代碼略）# 訓(xùn)練過程is_complete = Falsefor epoch in range(self.epochs):self.model.train()for batch_idx, (x, x_len, y, y_len) in enumerate(train_dataloader):x = x.to(device=self.device)y = y.to(device=self.device)results = self.model(x, x_len, y, y_len)loss = self.get_loss(decoder_outputs=results, y=y)# 簡單判定一下，如果損失小于0.5，則訓(xùn)練提前完成if loss.item() < 0.3:is_complete = Trueprint(f"訓(xùn)練提前完成, 本批次損失為：{loss.item()}")breakloss.backward()self.optimizer.step()self.optimizer.zero_grad()# 過程監(jiān)控with torch.no_grad():if batch_idx % 100 == 0:print(f"第 {epoch + 1} 輪 {batch_idx + 1} 批, 當(dāng)前批次損失: {loss.item()}")x_true = self.get_real_input(x)y_pred = self.model.batch_infer(x, x_len)y_true = self.get_real_output(y)samples = random.sample(population=range(x.size(1)), k=2)for idx in samples:print("\t真實輸入：", x_true[idx])print("\t真實結(jié)果：", y_true[idx])print("\t預(yù)測結(jié)果：", y_pred[idx])print("\t----------------------------------------------------------")# 外層提前退出if is_complete:# print("訓(xùn)練提前完成")break# 保存模型torch.save(obj=self.model.state_dict(), f="./model.pt")

手動循環(huán)進(jìn)行正向推理

    #（其他部分略）def batch_infer(self, encoder_hidden):"""推理時的正向傳播- encoder_hidden: [1, batch_size, hidden_size]"""# 推理時，設(shè)定一個最大的固定長度output_max_len = self.tokenizer.output_max_len# 獲取批量大小batch_size = encoder_hidden.size(1)# 輸入信號 SOS [1, batch_size]decoder_input = torch.LongTensor([[self.tokenizer.output_word2idx.get("<SOS>")] * batch_size]).to(device=device)# print(decoder_input)results = []# 隱藏狀態(tài)# encoder_hidden: [1, batch_size, hidden_size]decoder_hidden = encoder_hiddenwith torch.no_grad():# 手動循環(huán)for t in range(output_max_len):# decoder_input: [1, batch_size]# decoder_hidden: [1, batch_size, hidden_size]decoder_output_t, decoder_hidden = self.forward_step(decoder_input, decoder_hidden)# 取出結(jié)果 [1, batch_size]decoder_input = decoder_output_t.argmax(dim=-1).unsqueeze(0)results.append(decoder_input)# [seq_len, batch_size]results = torch.cat(tensors=results, dim=0)return results

代碼解析：

• 相比訓(xùn)練的時候，推理的時候函數(shù)入?yún)]有y標(biāo)準(zhǔn)答案。
• 推理的過程：
  • (與訓(xùn)練類似)獲取最大長度、獲取批量大小、構(gòu)建啟動信號。
  • (與訓(xùn)練不同)在無梯度環(huán)境里，調(diào)用forward_step函數(shù)，進(jìn)行循環(huán)推理。
  • (與訓(xùn)練不同)因為推理時不需要teacher forcing機(jī)制，所以直接使用貪心思想獲得概率最大的詞。
  • 循環(huán)結(jié)束后，將結(jié)果拼接起來，返回。

補充知識

tqdm

定義

tqdm 是一個用于在 Python 中顯示進(jìn)度條的庫，非常適合在長時間運行的循環(huán)中使用。

安裝方法

pip install tqdm

使用方法

from tqdm import tqdm
import time# 示例：在一個簡單的循環(huán)中使用 tqdm
for i in tqdm(range(10)):time.sleep(1)  # 模擬某個耗時操作

運行結(jié)果：

OpenCC

定義

OpenCC（Open Chinese Convert）是一個用于簡體中文和繁體中文之間轉(zhuǎn)換的工具

安裝方法

pip install OpenCC

使用方法

import opencc# 創(chuàng)建轉(zhuǎn)換器，使用簡體到繁體的配置
converter = opencc.OpenCC('s2t')  # s2t: 簡體到繁體# 輸入簡體中文
simplified_text = "我愛編程"# 進(jìn)行轉(zhuǎn)換
traditional_text = converter.convert(simplified_text)print(traditional_text)  
# 輸出結(jié)果：我愛編程

內(nèi)容小結(jié)

• Seq2Seq項目整體組成由tokenizer（分詞器）、dataloader(數(shù)據(jù)加載)、encoder（編碼器）、decoder（解碼器）、seq2seq和main六個部分組成
• 在分詞器中重點工作是構(gòu)建自定義字典，并添加特殊符號(special tokens)
  • <UNK>：表示未知單詞，用于表示輸入序列中未在字典中找到的單詞；
  • <PAD>：表示填充符號，用于填充輸入序列和輸出序列，使它們具有相同的長度；
  • <SOS>：表示序列的開始，用于表示輸出序列的起始位置；上文不會增加。
  • <EOS>：表示序列的結(jié)束，用于表示輸出序列的結(jié)束位置，上文不會增加。
• 在decoder的forward函數(shù)中，增加了一個teacher_forcing_ratio參數(shù)，用于控制是否使用教師強(qiáng)迫機(jī)制。
  • 有50%概率，使用標(biāo)準(zhǔn)答案作為下一步的輸入；
  • 有50%概率，使用上一步的推理結(jié)果中概率最大的詞作為下一步的輸入。
  • 該機(jī)制用于提升訓(xùn)練速度。
• 在訓(xùn)練過程中會使用collate_fn用于數(shù)據(jù)對齊時消除PAD的影響。

參考資料

(暫無)