終于把 Seq2Seq 算法搞懂了?。?/h1>
作者：程序員小寒
編碼器是一個循環(huán)神經(jīng)網(wǎng)絡(luò)（RNN）或其變體，如LSTM或GRU，用于接收輸入序列并將其轉(zhuǎn)換為一個固定大小的上下文向量。
Seq2Seq（Sequence-to-Sequence）模型是一種用于處理序列數(shù)據(jù)的神經(jīng)網(wǎng)絡(luò)架構(gòu)，廣泛應(yīng)用于自然語言處理（NLP）任務(wù)，如機器翻譯、文本生成、對話系統(tǒng)等。
它通過編碼器-解碼器架構(gòu)將輸入序列（如一個句子）映射到輸出序列（另一個句子或序列）。
圖片
模型結(jié)構(gòu)
Seq2Seq 模型由兩個主要部分組成。
編碼器（Encoder）
編碼器是一個循環(huán)神經(jīng)網(wǎng)絡(luò)（RNN）或其變體，如LSTM或GRU，用于接收輸入序列并將其轉(zhuǎn)換為一個固定大小的上下文向量。
編碼器逐步處理輸入序列的每個時間步，通過隱藏層狀態(tài)不斷更新輸入信息的表示，直到編碼到達輸入序列的結(jié)尾。
這一過程的最后一個隱藏狀態(tài)通常被認為是整個輸入序列的摘要，傳遞給解碼器。
圖片
class Encoder(nn.Module):
def __init__(self,input_dim,embedding_dim,hidden_size,num_layers,dropout):
super(Encoder,self).__init__()
#note hidden size and num layers
self.hidden_size = hidden_size
self.num_layers = num_layers
#create a dropout layer
self.dropout = nn.Dropout(dropout)
#embedding to convert input token into dense vectors
self.embedding = nn.Embedding(input_dim,embedding_dim)
#bilstm layer
self.lstm = nn.LSTM(embedding_dim,hidden_size,num_layers=num_layers,bidirectinotallow=True,dropout=dropout)
def forward(self,src):
embedded = self.dropout(self.embedding(src))
out,(hidden,cell) = self.lstm(embedded)
return hidden,cell解碼器（Decoder）
解碼器也是一個RNN網(wǎng)絡(luò)，接受編碼器輸出的上下文向量，并生成目標序列。
解碼器在每一步會生成一個輸出，并將上一步的輸出作為下一步的輸入，直到產(chǎn)生特定的終止符。
解碼器的初始狀態(tài)來自編碼器的最后一個隱藏狀態(tài)，因此可以理解為解碼器是基于編碼器生成的全局信息來預測輸出序列。
class Decoder(nn.Module):
def __init__(self,output_dim,embedding_dim,hidden_size,num_layers,dropout):
super(Decoder,self).__init__()
self.output_dim = output_dim
#note hidden size and num layers for seq2seq class
self.hidden_size = hidden_size
self.num_layers = num_layers
self.dropout = nn.Dropout(dropout)
#note inputs of embedding layer
self.embedding = nn.Embedding(output_dim,embedding_dim)
self.lstm = nn.LSTM(embedding_dim,hidden_size,num_layers=num_layers,bidirectinotallow=True,dropout=dropout)
#we apply softmax over target vocab size
self.fc = nn.Linear(hidden_size*2,output_dim)
def forward(self,input_token,hidden,cell):
#adjust dimensions of input token
input_token = input_token.unsqueeze(0)
emb = self.embedding(input_token)
emb = self.dropout(emb)
#note hidden and cell along with output
out,(hidden,cell) = self.lstm(emb,(hidden,cell))
out = out.squeeze(0)
pred = self.fc(out)
return pred,hidden,cell工作流程
Seq2Seq 模型的基本工作流程如下
- 輸入處理
將輸入序列（如源語言句子）逐步傳入編碼器的 RNN 層，編碼器的最后一層的隱藏狀態(tài)會保留輸入序列的上下文信息。 - 生成上下文向量
編碼器輸出的隱藏狀態(tài)向量（通常是最后一個隱藏狀態(tài)）稱為上下文向量，它包含了輸入序列的信息。 - 解碼過程
解碼器接收上下文向量作為初始狀態(tài)，然后通過自身的RNN結(jié)構(gòu)逐步生成目標序列。
每一步解碼器生成一個輸出token，并將其作為下一步的輸入，直到生成結(jié)束token。 - 序列生成
解碼器生成的序列作為模型的最終輸出。
優(yōu)缺點
優(yōu)點
- 通用性強
Seq2Seq 模型可以處理可變長度的輸入和輸出序列，適用于許多任務(wù)，例如機器翻譯、文本摘要、對話生成、語音識別等。
它的編碼器-解碼器結(jié)構(gòu)使得輸入和輸出不必同長，具有高度的靈活性。 - 適應(yīng)復雜序列任務(wù)
Seq2Seq 模型通過編碼器-解碼器的分離，能夠更好地學習序列映射關(guān)系。
編碼器負責捕獲輸入序列的信息，而解碼器則生成符合輸出序列特征的內(nèi)容。
缺點
- 信息壓縮損失
傳統(tǒng) Seq2Seq 模型通過編碼器最后一個隱藏狀態(tài)來表示整個輸入序列信息，當輸入序列較長時，這種單一的上下文向量難以全面表示輸入內(nèi)容，導致信息丟失。這會導致模型在長序列任務(wù)上表現(xiàn)欠佳。 - 對長序列敏感
在沒有注意力機制的情況下，Seq2Seq模型難以有效處理長序列，因為解碼器需要依賴于編碼器的固定向量，而這個向量可能無法完全涵蓋長序列的細節(jié)。 - 訓練難度大
Seq2Seq 模型在訓練時面臨梯度消失和爆炸的問題，尤其是在長序列的情況下。
案例分享
下面是一個使用 Seq2Seq 進行機器翻譯的示例代碼。
首先，我們從 HuggingFace 導入了數(shù)據(jù)集，并將其分為訓練集和測試集
import numpy as np
import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt
import torch
import torch.nn as nn
import torch.nn.functional as F
from torch.utils.data import DataLoader,Dataset
import tqdm,datasets
from torchtext.vocab import build_vocab_from_iterator
from torch.nn.utils.rnn import pad_sequence
import spacy
dataset = datasets.load_dataset('bentrevett/multi30k')
train_data,val_data,test_data = dataset['train'],dataset['validation'],dataset['test']加載源語言和目標語言的 spaCy 模型。
spaCy 是一個功能強大、可用于生產(chǎn)的 Python 高級自然語言處理庫。
與許多其他 NLP 庫不同，spaCy 專為實際使用而設(shè)計，而非研究實驗。
它擅長使用預先訓練的模型進行高效的文本處理，可完成標記化、詞性標記、命名實體識別和依賴性解析等任務(wù)。
en_nlp = spacy.load('en_core_web_sm')
de_nlp = spacy.load('de_core_news_sm')
#tokenizer
def sample_tokenizer(sample,en_nlp,de_nlp,lower,max_length,sos_token,eos_token):
en_tokens = [token.text for token in en_nlp.tokenizer(sample["en"])][:max_length]
de_tokens = [token.text for token in de_nlp.tokenizer(sample["de"])][:max_length]
if lower == True:
en_tokens = [token.lower() for token in en_tokens]
de_tokens = [token.lower() for token in de_tokens]
en_tokens = [sos_token] + en_tokens + [eos_token]
de_tokens = [sos_token] + de_tokens + [eos_token]
return {"en_tokens":en_tokens,"de_tokens":de_tokens}
fn_kwargs = {
"en_nlp":en_nlp,
"de_nlp":de_nlp,
"lower":True,
"max_length":1000,
"sos_token":'<sos>',
"eos_token":'<eos>'
}
train_data = train_data.map(sample_tokenizer,fn_kwargs=fn_kwargs)
val_data = val_data.map(sample_tokenizer,fn_kwargs=fn_kwargs)
test_data = test_data.map(sample_tokenizer,fn_kwargs=fn_kwargs)
min_freq = 2
specials = ['<unk>','<pad>','<sos>','<eos>']
en_vocab = build_vocab_from_iterator(train_data['en_tokens'],specials=specials,min_freq=min_freq)
de_vocab = build_vocab_from_iterator(train_data['de_tokens'],specials=specials,min_freq=min_freq)
assert en_vocab['<unk>'] == de_vocab['<unk>']
assert en_vocab['<pad>'] == de_vocab['<pad>']
unk_index = en_vocab['<unk>']
pad_index = en_vocab['<pad>']
en_vocab.set_default_index(unk_index)
de_vocab.set_default_index(unk_index)
def sample_num(sample,en_vocab,de_vocab):
en_ids = en_vocab.lookup_indices(sample["en_tokens"])
de_ids = de_vocab.lookup_indices(sample["de_tokens"])
return {"en_ids":en_ids,"de_ids":de_ids}
fn_kwargs = {"en_vocab":en_vocab,"de_vocab":de_vocab}
train_data = train_data.map(sample_num,fn_kwargs=fn_kwargs)
val_data = val_data.map(sample_num,fn_kwargs=fn_kwargs)
test_data = test_data.map(sample_num,fn_kwargs=fn_kwargs)
train_data = train_data.with_format(type="torch",columns=['en_ids','de_ids'],output_all_columns=True)
val_data = val_data.with_format(type="torch",columns=['en_ids','de_ids'],output_all_columns=True)
test_data = test_data.with_format(type="torch",columns=['en_ids','de_ids'],output_all_columns=True)
def get_collate_fn(pad_index):
def collate_fn(batch):
batch_en_ids = [sample["en_ids"] for sample in batch]
batch_de_ids = [sample["de_ids"] for sample in batch]
batch_en_ids = pad_sequence(batch_en_ids,padding_value=pad_index)
batch_de_ids = pad_sequence(batch_de_ids,padding_value=pad_index)
batch = {"en_ids":batch_en_ids,"de_ids":batch_de_ids}
return batch
return collate_fn
def get_dataloader(dataset,batch_size,shuffle,pad_index):
collate_fn = get_collate_fn(pad_index)
dataloader = DataLoader(dataset=dataset,
batch_size=batch_size,
shuffle=shuffle,
collate_fn=collate_fn)
return dataloader
train_loader = get_dataloader(train_data,batch_size=512,shuffle=True,pad_index=pad_index)
val_loader = get_dataloader(val_data,batch_size=512,shuffle=True,pad_index=pad_index)
test_loader = get_dataloader(test_data,batch_size=512,shuffle=True,pad_index=pad_index)接下來構(gòu)建 seq2seq 模型。
class Seq2Seq(nn.Module):
def __init__(self,encoder,decoder,device):
super(Seq2Seq,self).__init__()
self.encoder = encoder
self.decoder = decoder
assert encoder.num_layers == decoder.num_layers
assert encoder.hidden_size == decoder.hidden_size
def forward(self,src,trg,teacher_forcing_ratio):
#exctract dim for out vector
trg_len = trg.shape[0]
batch_size = trg.shape[1]
vocab_size = self.decoder.output_dim
outputs = torch.zeros(trg_len,batch_size,vocab_size).to(device)
#get input and hidden
input_token = trg[0,:]
hidden,cell = self.encoder(src)
for t in range(1,trg_len):
out,hidden,cell = self.decoder(input_token,hidden,cell)
outputs[t] = out
#decide what passes as input
top1 = out.argmax(1)
teacher_force = np.random.randn()<teacher_forcing_ratio
input_token = trg[t] if teacher_force else top1
return outputs
input_dim = len(de_vocab)
output_dim = len(en_vocab)
encoder_embedding_dim = 256
decoder_embedding_dim = 256
hidden_size = 512
num_layers = 3
encoder_dropout = 0.5
decoder_dropout = 0.5
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
encoder = Encoder(
input_dim,
encoder_embedding_dim,
hidden_size=hidden_size,
num_layers=num_layers,
dropout=encoder_dropout,
)
decoder = Decoder(
output_dim,
decoder_embedding_dim,
hidden_size=hidden_size,
num_layers=num_layers,
dropout=decoder_dropout,
)
model = Seq2Seq(encoder, decoder, device).to(device)模型訓練
optimizer = torch.optim.Adam(model.parameters())
criterion = nn.CrossEntropyLoss(ignore_index=pad_index)
def train_fn(
model, data_loader, optimizer, criterion, clip, teacher_forcing_ratio, device):
model.train()
epoch_loss = 0
for i, batch in enumerate(data_loader):
src = batch["de_ids"].to(device)
trg = batch["en_ids"].to(device)
# src = [src length, batch size]
# trg = [trg length, batch size]
optimizer.zero_grad()
output = model(src, trg, teacher_forcing_ratio)
# output = [trg length, batch size, trg vocab size]
output_dim = output.shape[-1]
output = output[1:].view(-1, output_dim)
# output = [(trg length - 1) * batch size, trg vocab size]
trg = trg[1:].view(-1)
# trg = [(trg length - 1) * batch size]
loss = criterion(output, trg)
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), clip)
optimizer.step()
epoch_loss += loss.item()
return epoch_loss / len(data_loader)
def evaluate_fn(model, data_loader, criterion, device):
model.eval()
epoch_loss = 0
with torch.no_grad():
for i, batch in enumerate(data_loader):
src = batch["de_ids"].to(device)
trg = batch["en_ids"].to(device)
# src = [src length, batch size]
# trg = [trg length, batch size]
output = model(src, trg, 0) # turn off teacher forcing
# output = [trg length, batch size, trg vocab size]
output_dim = output.shape[-1]
output = output[1:].view(-1, output_dim)
# output = [(trg length - 1) * batch size, trg vocab size]
trg = trg[1:].view(-1)
# trg = [(trg length - 1) * batch size]
loss = criterion(output, trg)
epoch_loss += loss.item()
return epoch_loss / len(data_loader)
n_epochs = 10
clip = 1.0
teacher_forcing_ratio = 1
best_valid_loss = float("inf")
for epoch in tqdm.tqdm(range(n_epochs)):
train_loss = train_fn(
model,
train_loader,
optimizer,
criterion,
clip,
teacher_forcing_ratio,
device,
)
valid_loss = evaluate_fn(
model,
val_loader,
criterion,
device,
)
if valid_loss < best_valid_loss:
best_valid_loss = valid_loss
torch.save(model.state_dict(), "tut1-model.pt")
print(f"\tTrain Loss: {train_loss:7.3f} | Train PPL: {np.exp(train_loss):7.3f}")
print(f"\tValid Loss: {valid_loss:7.3f} | Valid PPL: {np.exp(valid_loss):7.3f}")
model.load_state_dict(torch.load("tut1-model.pt"))
test_loss = evaluate_fn(model, test_loader, criterion, device)
print(f"| Test Loss: {test_loss:.3f} | Test PPL: {np.exp(test_loss):7.3f} |")接下來，我們看一下最終的效果
def translate_sentence(
sentence,
model,
en_nlp,
de_nlp,
en_vocab,
de_vocab,
lower,
sos_token,
eos_token,
device,
max_output_length=25,
):
model.eval()
with torch.no_grad():
if isinstance(sentence, str):
tokens = [token.text for token in de_nlp.tokenizer(sentence)]
else:
tokens = [token for token in sentence]
if lower:
tokens = [token.lower() for token in tokens]
tokens = [sos_token] + tokens + [eos_token]
ids = de_vocab.lookup_indices(tokens)
tensor = torch.LongTensor(ids).unsqueeze(-1).to(device)
hidden, cell = model.encoder(tensor)
inputs = en_vocab.lookup_indices([sos_token])
for _ in range(max_output_length):
inputs_tensor = torch.LongTensor([inputs[-1]]).to(device)
output, hidden, cell = model.decoder(inputs_tensor, hidden, cell)
predicted_token = output.argmax(-1).item()
inputs.append(predicted_token)
if predicted_token == en_vocab[eos_token]:
break
tokens = en_vocab.lookup_tokens(inputs)
return tokens
sentence ='Der Mann ist am Weisheitsspross'
sos_token='<sos>'
eos_token='<eos>'
lower=True
translation = translate_sentence(
sentence,
model,
en_nlp,
de_nlp,
en_vocab,
de_vocab,
lower,
sos_token,
eos_token,
device,
)
print(translation)
#['<sos>', 'the', 'woman', 'is', 'looking', 'at', 'the', 'camera', '.', '<eos>']
責任編輯：武曉燕
來源：
程序員學長
