在深度學(xué)習(xí)中，Transformer 架構(gòu)憑借其卓越的并行性和對序列間長距離依賴關(guān)系的精準(zhǔn)捕捉，在自然語言處理（NLP）領(lǐng)域大放異彩，從機(jī)器翻譯到文本生成，Transformer 正在重塑我們對語言處理的認(rèn)知。對于初學(xué)者來說，理解并掌握基于 PyTorch 和 TorchText 的 Transformer 模型構(gòu)建，是踏入 NLP 高級應(yīng)用的關(guān)鍵一步。本文將帶您從零開始，循序漸進(jìn)地學(xué)習(xí)如何利用 PyTorch 的 nn.Transformer 模塊和 TorchText 實(shí)現(xiàn)序列到序列建模，開啟您的 NLP 之旅。

一、環(huán)境搭建：筑造模型的基石

在開始構(gòu)建基于 Transformer 的序列到序列模型之前，確保您的開發(fā)環(huán)境已正確安裝相關(guān)的依賴庫，這是保證后續(xù)代碼順利運(yùn)行的基礎(chǔ)。

1. 安裝 PyTorch ：訪問Pytorch官方網(wǎng)址，根據(jù)您的系統(tǒng)配置（如操作系統(tǒng)、CUDA 版本等）獲取適合的安裝命令。例如，對于 Linux 系統(tǒng)且 CUDA 版本為 11.8 的用戶，可以使用以下命令安裝 PyTorch：
   • 命令 ：conda install pytorch torchvision torchaudio pytorch-cuda=11.8 -c pytorch -c nvidia

1. 安裝 TorchText ：TorchText 是專門用于處理文本數(shù)據(jù)的 PyTorch 擴(kuò)展庫，它提供了豐富的數(shù)據(jù)處理工具和預(yù)定義的數(shù)據(jù)集，極大地方便了文本相關(guān)任務(wù)的開展。安裝命令如下：
   • 命令 ：pip install torchtext

1. 安裝 Spacy ：為了實(shí)現(xiàn)高效的文本分詞操作，本文推薦使用 Spacy 庫，它支持多種語言的分詞，并且與 TorchText 配合默契。
   • 命令 ：pip install spacy
   • 接下來，您需要下載對應(yīng)語言的分詞模型，例如英語和德語模型：
     • 下載英語模型：python -m spacy download en
     • 下載德語模型：python -m spacy download de

二、數(shù)據(jù)準(zhǔn)備：序列到序列模型的養(yǎng)料

數(shù)據(jù)是構(gòu)建模型的基石，對于序列到序列任務(wù)，我們選擇 Multi30k 數(shù)據(jù)集，該數(shù)據(jù)集包含了約 30,000 個(gè)英語和德語句子對，句子平均長度約為 13 個(gè)單詞，非常適合作為訓(xùn)練和評估序列到序列模型的數(shù)據(jù)基礎(chǔ)。

（一）數(shù)據(jù)集加載與字段定義

TorchText 提供了便捷的數(shù)據(jù)集加載方式，能夠輕松加載 Multi30k 數(shù)據(jù)集并進(jìn)行初步處理。同時(shí)，我們需要定義字段，指定對源語言（德語）和目標(biāo)語言（英語）的預(yù)處理方式，如分詞、添加起始和結(jié)束標(biāo)記等。

from torchtext.datasets import Multi30k
from torchtext.data import Field


## 定義德語字段（源語言）
SRC = Field(tokenize="spacy", tokenizer_language="de", init_token="<sos>", eos_token="<eos>", lower=True)


## 定義英語字段（目標(biāo)語言）
TRG = Field(tokenize="spacy", tokenizer_language="en", init_token="<sos>", eos_token="<eos>", lower=True)


## 加載 Multi30k 數(shù)據(jù)集
train_data, valid_data, test_data = Multi30k(language_pair=("de", "en"))

（二）詞匯表構(gòu)建與數(shù)據(jù)迭代器創(chuàng)建

在對數(shù)據(jù)進(jìn)行編碼和解碼之前，我們需要構(gòu)建詞匯表，將文本單詞映射到數(shù)值索引。此外，為了高效地將數(shù)據(jù)喂入模型進(jìn)行訓(xùn)練和評估，我們還需要使用數(shù)據(jù)迭代器。TorchText 提供的 BucketIterator 能夠根據(jù)序列長度將相似長度的樣本劃分到一個(gè)批次中，從而減少填充操作，提高訓(xùn)練效率。

from torchtext.data import BucketIterator


## 構(gòu)建德語詞匯表
SRC.build_vocab(train_data, min_freq=2)


## 構(gòu)建英語詞匯表
TRG.build_vocab(train_data, min_freq=2)


## 設(shè)置設(shè)備（GPU 或 CPU）
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")


## 定義批次大小
BATCH_SIZE = 128


## 創(chuàng)建訓(xùn)練、驗(yàn)證和測試數(shù)據(jù)迭代器
train_iterator, valid_iterator, test_iterator = BucketIterator.splits(
    (train_data, valid_data, test_data),
    batch_size=BATCH_SIZE,
    device=device
)

三、模型構(gòu)建：序列到序列的核心引擎

（一）Transformer 模型架構(gòu)概覽

Transformer 模型主要由編碼器（Encoder）和解碼器（Decoder）組成。編碼器負(fù)責(zé)將輸入序列轉(zhuǎn)換為上下文表示，解碼器則基于編碼器的輸出生成目標(biāo)序列。Transformer 的核心創(chuàng)新在于自注意力機(jī)制（Self-Attention），它允許模型在處理序列中的每個(gè)位置時(shí)，動(dòng)態(tài)地關(guān)注序列中其他位置的相關(guān)信息，從而捕捉序列中的長距離依賴關(guān)系。

（二）編碼器與解碼器的實(shí)現(xiàn)

PyTorch 提供了 nn.TransformerEncoder 和 nn.TransformerDecoder 模塊，簡化了 Transformer 編碼器和解碼器的構(gòu)建過程。以下是編碼器和解碼器的代碼實(shí)現(xiàn)：

import torch.nn as nn


class TransformerSeq2Seq(nn.Module):
    def __init__(self, src_vocab_size, trg_vocab_size, emb_dim, hid_dim, n_layers, n_heads, dropout, device, max_len=100):
        super().__init__()
        self.device = device

        
        # 編碼器嵌入層和位置編碼
        self.src_embedding = nn.Embedding(src_vocab_size, emb_dim)
        self.src_pos_encoder = PositionalEncoding(emb_dim, dropout, max_len)

        
        # 編碼器層
        encoder_layer = nn.TransformerEncoderLayer(d_model=emb_dim, nhead=n_heads, dim_feedforward=hid_dim, dropout=dropout)
        self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=n_layers)

        
        # 解碼器嵌入層和位置編碼
        self.trg_embedding = nn.Embedding(trg_vocab_size, emb_dim)
        self.trg_pos_encoder = PositionalEncoding(emb_dim, dropout, max_len)

        
        # 解碼器層
        decoder_layer = nn.TransformerDecoderLayer(d_model=emb_dim, nhead=n_heads, dim_feedforward=hid_dim, dropout=dropout)
        self.decoder = nn.TransformerDecoder(decoder_layer, num_layers=n_layers)

        
        # 輸出層
        self.out = nn.Linear(emb_dim, trg_vocab_size)

        
    def forward(self, src, trg, src_mask=None, trg_mask=None, src_padding_mask=None, trg_padding_mask=None):
        # 編碼器部分
        src_emb = self.src_embedding(src) * math.sqrt(self.emb_dim)
        src_emb = self.src_pos_encoder(src_emb)
        memory = self.encoder(src_emb, src_key_padding_mask=src_padding_mask)

        
        # 解碼器部分
        trg_emb = self.trg_embedding(trg) * math.sqrt(self.emb_dim)
        trg_emb = self.trg_pos_encoder(trg_emb)
        output = self.decoder(trg_emb, memory, tgt_mask=trg_mask, memory_key_padding_mask=src_padding_mask, tgt_key_padding_mask=trg_padding_mask)

        
        # 輸出層
        output = self.out(output)
        return output

（三）位置編碼模塊

位置編碼模塊用于向模型注入序列中單詞的位置信息，因?yàn)?Transformer 模型本身并不像 RNN 那樣對序列順序敏感。位置編碼的維度與嵌入維度相同，因此可以將兩者相加。在這里，我們使用不同頻率的 sine 和 cosine 函數(shù)來實(shí)現(xiàn)位置編碼。

import math


class PositionalEncoding(nn.Module):
    def __init__(self, d_model, dropout=0.1, max_len=5000):
        super().__init__()
        self.dropout = nn.Dropout(p=dropout)
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2).float() * (-math.log(10000.0) / d_model))
        pe[:, 0::2] = torch.sin(position * div_term)
        pe[:, 1::2] = torch.cos(position * div_term)
        pe = pe.unsqueeze(0).transpose(0, 1)
        self.register_buffer("pe", pe)

        
    def forward(self, x):
        x = x + self.pe[:x.size(0), :]
        return self.dropout(x)

四、模型訓(xùn)練與評估：打磨序列到序列模型的利器

（一）超參數(shù)設(shè)置與模型初始化

在訓(xùn)練模型之前，我們需要設(shè)置一些超參數(shù)，這些超參數(shù)將影響模型的訓(xùn)練過程和最終性能。接下來，我們根據(jù)超參數(shù)初始化編碼器、解碼器和序列到序列模型，并定義優(yōu)化器來更新模型參數(shù)。

## 超參數(shù)設(shè)置
SRC_VOCAB_SIZE = len(SRC.vocab)
TRG_VOCAB_SIZE = len(TRG.vocab)
EMB_DIM = 512
HID_DIM = 1024
N_LAYERS = 3
N_HEADS = 8
DROPOUT = 0.1


## 初始化模型
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
model = TransformerSeq2Seq(SRC_VOCAB_SIZE, TRG_VOCAB_SIZE, EMB_DIM, HID_DIM, N_LAYERS, N_HEADS, DROPOUT, device).to(device)


## 定義優(yōu)化器
optimizer = torch.optim.Adam(model.parameters())

（二）損失函數(shù)定義與訓(xùn)練過程

為了評估模型的性能并指導(dǎo)模型訓(xùn)練，我們需要定義損失函數(shù)。在序列到序列任務(wù)中，通常使用交叉熵?fù)p失函數(shù)，并忽略填充部分的損失計(jì)算。訓(xùn)練過程是模型學(xué)習(xí)數(shù)據(jù)模式并不斷提升翻譯性能的關(guān)鍵階段。在每個(gè)訓(xùn)練周期（epoch）中，模型會(huì)處理整個(gè)訓(xùn)練數(shù)據(jù)集，并根據(jù)計(jì)算得到的損失更新模型參數(shù)。

import torch
import math
import time


## 定義損失函數(shù)
PAD_IDX = TRG.vocab.stoi["<pad>"]
criterion = nn.CrossEntropyLoss(ignore_index=PAD_IDX)


def train(model, iterator, optimizer, criterion, clip):
    model.train()
    epoch_loss = 0
    for _, batch in enumerate(iterator):
        src = batch.src
        trg = batch.trg

        
        optimizer.zero_grad()

        
        # 創(chuàng)建掩碼
        src_mask = (src != SRC.vocab.stoi["<pad>"]).unsqueeze(-2)
        trg_mask = (trg != TRG.vocab.stoi["<pad>"]).unsqueeze(-2)
        trg_mask = trg_mask & (torch.triu(torch.ones(trg.shape[0], trg.shape[0])) == 0).to(trg_mask.dtype).to(trg_mask.device)

        
        output = model(src, trg[:-1], src_mask=src_mask, trg_mask=trg_mask)

        
        output = output.contiguous().view(-1, output.shape[-1])
        trg = trg[1:].contiguous().view(-1)

        
        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(iterator)


def evaluate(model, iterator, criterion):
    model.eval()
    epoch_loss = 0
    with torch.no_grad():
        for _, batch in enumerate(iterator):
            src = batch.src
            trg = batch.trg

            
            src_mask = (src != SRC.vocab.stoi["<pad>"]).unsqueeze(-2)
            trg_mask = (trg != TRG.vocab.stoi["<pad>"]).unsqueeze(-2)
            trg_mask = trg_mask & (torch.triu(torch.ones(trg.shape[0], trg.shape[0])) == 0).to(trg_mask.dtype).to(trg_mask.device)

            
            output = model(src, trg[:-1], src_mask=src_mask, trg_mask=trg_mask)

            
            output = output.contiguous().view(-1, output.shape[-1])
            trg = trg[1:].contiguous().view(-1)

            
            loss = criterion(output, trg)

            
            epoch_loss += loss.item()
    return epoch_loss / len(iterator)


## 訓(xùn)練模型
N_EPOCHS = 10
CLIP = 1


best_valid_loss = float("inf")


for epoch in range(N_EPOCHS):
    start_time = time.time()

    
    train_loss = train(model, train_iterator, optimizer, criterion, CLIP)
    valid_loss = evaluate(model, valid_iterator, criterion)

    
    end_time = time.time()

    
    epoch_mins, epoch_secs = divmod(end_time - start_time, 60)

    
    print(f"Epoch: {epoch + 1:02} | Time: {epoch_mins}m {epoch_secs:.2f}s")
    print(f"\tTrain Loss: {train_loss:.3f} | Train PPL: {math.exp(train_loss):7.3f}")
    print(f"\t Val. Loss: {valid_loss:.3f} |  Val. PPL: {math.exp(valid_loss):7.3f}")

    
    if valid_loss < best_valid_loss:
        best_valid_loss = valid_loss
        torch.save(model.state_dict(), "transformer_model.pt")


## 加載最佳模型并在測試集上評估
model.load_state_dict(torch.load("transformer_model.pt"))
test_loss = evaluate(model, test_iterator, criterion)
print(f"| Test Loss: {test_loss:.3f} | Test PPL: {math.exp(test_loss):7.3f} |")

五、總結(jié)與展望

通過本文，您已經(jīng)學(xué)習(xí)了如何利用 PyTorch 和 TorchText 構(gòu)建一個(gè)基于 Transformer 的序列到序列模型，從環(huán)境搭建、數(shù)據(jù)準(zhǔn)備、模型構(gòu)建到訓(xùn)練與評估，每一步都至關(guān)重要。Transformer 模型憑借其強(qiáng)大的并行計(jì)算能力和對長距離依賴關(guān)系的捕捉能力，在眾多序列到序列任務(wù)中表現(xiàn)出色。在實(shí)際應(yīng)用中，您可以根據(jù)需求進(jìn)一步優(yōu)化模型，如調(diào)整超參數(shù)、使用更大的模型架構(gòu)、采用數(shù)據(jù)增強(qiáng)技術(shù)等，以提升模型性能。

未來，隨著深度學(xué)習(xí)技術(shù)的不斷發(fā)展，Transformer 架構(gòu)及其變體將在自然語言處理領(lǐng)域發(fā)揮更加重要的作用，為人們的語言交流和信息處理提供更強(qiáng)大的支持。編程獅將持續(xù)為您提供更多優(yōu)質(zhì)的技術(shù)教程和資源，助力您的編程學(xué)習(xí)之旅。