整體結構
1. 嵌入層(Embedding Layer)
生活中的例子:字典查找
想象你在讀一本書,你不認識某個單詞,于是你查閱字典。字典為每個單詞提供了一個解釋,幫助你理解這個單詞的意思。嵌入層就像這個字典,它將每個單詞(或輸入序列中的每個標記)映射到一個高維向量(解釋),這個向量包含了單詞的各種語義信息。
import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import DataLoader, TensorDataset
import mathclass EmbeddingLayer(nn.Module):def __init__(self, vocab_size, d_model, max_seq_length=512):super(EmbeddingLayer, self).__init__()# vocab_size: 詞匯表的大小,即輸入序列中可能的不同標記的總數。# d_model: 每個嵌入向量的維度,即詞嵌入向量的長度。# max_seq_length: 序列的最大長度,用于位置嵌入。self.embedding = nn.Embedding(vocab_size, d_model) # 詞嵌入層self.pos_embedding = nn.Embedding(max_seq_length, d_model) # 位置嵌入層self.d_model = d_model# 初始化位置編碼pe = torch.zeros(max_len, d_model)# 生成詞位置列表position = torch.arange(0, max_len).unsqueeze(1)# 根據公式計算詞位置參數div_term = torch.exp(torch.arange(0, d_model, 2) * -(math.log(10000.0) / d_model))# 生成詞位置矩陣my_matmulres = position * div_term# 給位置編碼矩陣奇數列,賦值sin曲線特征pe[:, 0::2] = torch.sin(my_matmulres)# 給位置編碼矩陣偶數列,賦值cos曲線特征pe[:, 1::2] = torch.cos(my_matmulres)# 形狀變化 [max_seq_length,d_model]-->[1,max_seq_length,d_model]pe = pe.unsqueeze(0)# 把pe位置編碼矩陣 注冊成模型的持久緩沖區buffer; 模型保存再加載時,可以根模型參數一樣,一同被加載# 什么是buffer: 對模型效果有幫助的,但是卻不是模型結構中超參數或者參數,不參與模型訓練self.register_buffer('pe', pe)def forward(self, x):seq_length = x.size(1) # 序列長度pos = torch.arange(0, seq_length, device=x.device).unsqueeze(0) # 生成位置索引return self.embedding(x) * math.sqrt(self.d_model) + self.pe[:,:x.size()[-1], :] # 詞嵌入和位置嵌入相加
2. 多頭自注意力機制(Multi-Head Self-Attention)
生活中的例子:小組討論
想象你在一個小組討論中,每個人(每個位置上的單詞)都提出自己的觀點(Query),并聽取其他人的意見(Key和Value)。每個人對所有其他人的觀點進行加權平均,以形成自己的新觀點。多頭注意力機制類似于多個小組同時進行討論,每個小組從不同的角度(頭)討論問題,然后將所有討論結果合并在一起。
class MultiHeadSelfAttention(nn.Module):def __init__(self, d_model, nhead):super(MultiHeadSelfAttention, self).__init__()# d_model: 輸入和輸出的維度,即每個位置的特征向量的長度。# nhead: 注意力頭的數量,多頭注意力機制中并行的注意力計算數。self.nhead = nheadself.d_model = d_model# 定義線性變換層self.q_linear = nn.Linear(d_model, d_model)self.k_linear = nn.Linear(d_model, d_model)self.v_linear = nn.Linear(d_model, d_model)self.out_linear = nn.Linear(d_model, d_model)self.scale = (d_model // nhead) ** 0.5 # 縮放因子def forward(self, x):batch_size = x.size(0) # 獲取批大小# 線性變換并分成多頭q = self.q_linear(x).view(batch_size, -1, self.nhead, self.d_model // self.nhead).transpose(1, 2)k = self.k_linear(x).view(batch_size, -1, self.nhead, self.d_model // self.nhead).transpose(1, 2)v = self.v_linear(x).view(batch_size, -1, self.nhead, self.d_model // self.nhead).transpose(1, 2)# 計算注意力得分scores = torch.matmul(q, k.transpose(-2, -1)) / self.scaleattn = torch.nn.functional.softmax(scores, dim=-1) # 計算注意力權重context = torch.matmul(attn, v).transpose(1, 2).contiguous().view(batch_size, -1, self.d_model) # 加權求和out = self.out_linear(context) # 最后一層線性變換return out
3. 前饋神經網絡(Feed-Forward Network)
生活中的例子:信息過濾和處理
想象你在整理會議紀要,需要對會議地錄音進行歸納、總結和補充。前饋神經網絡類似于這個過程,它對輸入的信息進行進一步處理和轉換,以提取重要特征。
class FeedForwardNetwork(nn.Module):def __init__(self, d_model, dim_feedforward, dropout=0.1):super(FeedForwardNetwork, self).__init__()# d_model: 輸入和輸出的維度,即每個位置的特征向量的長度。# dim_feedforward: 前饋神經網絡的隱藏層維度。# dropout: 在前饋神經網絡中使用的dropout比率,用于正則化。self.linear1 = nn.Linear(d_model, dim_feedforward) # 第一個線性層self.dropout = nn.Dropout(dropout) # dropout層self.linear2 = nn.Linear(dim_feedforward, d_model) # 第二個線性層def forward(self, x):return self.linear2(self.dropout(torch.nn.functional.relu(self.linear1(x)))) # 激活函數ReLU和dropout
4. 層歸一化(Layer Normalization)
生活中的例子:團隊合作中的標準化
想象你在一個團隊中工作,每個人都有不同的工作習慣和標準。為了更好地合作,團隊決定采用統一的工作標準(如文檔格式、命名規范等)。層歸一化類似于這種標準化過程,它將輸入歸一化,使得每個特征的均值為0,標準差為1,以穩定和加速訓練。
class LayerNorm(nn.Module):def __init__(self, d_model, eps=1e-6):super(LayerNorm, self).__init__()# d_model: 輸入和輸出的維度,即每個位置的特征向量的長度。# eps: 用于數值穩定的小值,防止除以零。self.gamma = nn.Parameter(torch.ones(d_model)) # 縮放參數self.beta = nn.Parameter(torch.zeros(d_model)) # 偏移參數self.eps = eps # epsilon,用于數值穩定def forward(self, x):mean = x.mean(dim=-1, keepdim=True) # 計算均值std = x.std(dim=-1, keepdim=True) # 計算標準差return self.gamma * (x - mean) / (std + self.eps) + self.beta # 歸一化
5. 殘差連接(Residual Connection)
生活中的例子:備忘錄
想象你在會議上記了很多筆記。為了確保不會遺漏任何重要信息,你在總結時會參照這些筆記。殘差連接類似于這個過程,它將每層的輸入直接加到輸出上,確保信息不會在層與層之間丟失。
class ResidualConnection(nn.Module):def __init__(self, d_model, dropout=0.1):super(ResidualConnection, self).__init__()# d_model: 輸入和輸出的維度,即每個位置的特征向量的長度。# dropout: 在殘差連接中使用的dropout比率,用于正則化。self.norm = LayerNorm(d_model) # 層歸一化self.dropout = nn.Dropout(dropout) # dropout層def forward(self, x, sublayer):return x + self.dropout(sublayer(self.norm(x))) # 殘差連接
6. 編碼器層(Encoder Layer)
生活中的例子:多輪面試
想象你在參加多輪面試,每輪面試都有不同的考官,考察不同的方面(如專業知識、溝通能力等)。每輪面試都幫助你更全面地展示自己。編碼器層類似于這種多輪面試的過程,每層處理輸入序列的不同方面,逐層提取和增強特征。
class EncoderLayer(nn.Module):def __init__(self, d_model, nhead, dim_feedforward, dropout=0.1):super(EncoderLayer, self).__init__()# d_model: 輸入和輸出的維度,即每個位置的特征向量的長度。# nhead: 注意力頭的數量,多頭注意力機制中并行的注意力計算數。# dim_feedforward: 前饋神經網絡的隱藏層維度。# dropout: 在各層中使用的dropout比率,用于正則化。self.self_attn = MultiHeadSelfAttention(d_model, nhead) # 多頭自注意力機制self.feed_forward = FeedForwardNetwork(d_model, dim_feedforward, dropout) # 前饋神經網絡self.sublayers = nn.ModuleList([ResidualConnection(d_model, dropout) for _ in range(2)]) # 兩個子層(注意力和前饋網絡)def forward(self, src):src = self.sublayers[0](src, lambda x: self.self_attn(x)) # 應用自注意力機制src = self.sublayers[1](src, self.feed_forward) # 應用前饋神經網絡return src
7. 解碼器層(Decoder Layer)
生活中的例子:逐步解謎
想象你在玩一個解謎游戲,每解決一個謎題(每層解碼器),你都會得到新的線索,逐步解開整個謎題。解碼器層類似于這種逐步解謎的過程,每層結合當前解碼的結果和編碼器的輸出,逐步生成目標序列。
class DecoderLayer(nn.Module):def __init__(self, d_model, nhead, dim_feedforward, dropout=0.1):super(DecoderLayer, self).__init__()# d_model: 輸入和輸出的維度,即每個位置的特征向量的長度。# nhead: 注意力頭的數量,多頭注意力機制中并行的注意力計算數。# dim_feedforward: 前饋神經網絡的隱藏層維度。# dropout: 在各層中使用的dropout比率,用于正則化。self.self_attn = MultiHeadSelfAttention(d_model, nhead) # 多頭自注意力機制self.cross_attn = MultiHeadSelfAttention(d_model, nhead) # 編碼器-解碼器注意力self.feed_forward = FeedForwardNetwork(d_model, dim_feedforward, dropout) # 前饋神經網絡self.sublayers = nn.ModuleList([ResidualConnection(d_model, dropout) for _ in range(3)]) # 三個子層(自注意力、交叉注意力、前饋網絡)def forward(self, tgt, memory):tgt = self.sublayers[0](tgt, lambda x: self.self_attn(x)) # 應用自注意力機制tgt = self.sublayers[1](tgt, lambda x: self.cross_attn(x, memory)) # 應用編碼器-解碼器注意力tgt = self.sublayers[2](tgt, self.feed_forward) # 應用前饋神經網絡return tgt
8. 編碼器(Encoder)
class Encoder(nn.Module):def __init__(self, num_layers, d_model, nhead, dim_feedforward, dropout=0.1):super(Encoder, self).__init__()# num_layers: 編碼器層的數量,即堆疊的編碼器層數。# d_model: 輸入和輸出的維度,即每個位置的特征向量的長度。# nhead: 注意力頭的數量,多頭注意力機制中并行的注意力計算數。# dim_feedforward: 前饋神經網絡的隱藏層維度。# dropout: 在各層中使用的dropout比率,用于正則化。self.layers = nn.ModuleList([EncoderLayer(d_model, nhead, dim_feedforward, dropout) for _ in range(num_layers)]) # 堆疊多個編碼器層def forward(self, src):for layer in self.layers:src = layer(src) # 依次通過每個編碼器層return src
9. 解碼器(Decoder)
class Decoder(nn.Module):def __init__(self, num_layers, d_model, nhead, dim_feedforward, dropout=0.1):super(Decoder, self).__init__()# num_layers: 解碼器層的數量,即堆疊的解碼器層數。# d_model: 輸入和輸出的維度,即每個位置的特征向量的長度。# nhead: 注意力頭的數量,多頭注意力機制中并行的注意力計算數。# dim_feedforward: 前饋神經網絡的隱藏層維度。# dropout: 在各層中使用的dropout比率,用于正則化。self.layers = nn.ModuleList([DecoderLayer(d_model, nhead, dim_feedforward, dropout) for _ in range(num_layers)]) # 堆疊多個解碼器層def forward(self, tgt, memory):for layer in self.layers:tgt = layer(tgt, memory) # 依次通過每個解碼器層return tgt
10. Transformer模型
class TransformerModel(nn.Module):def __init__(self, vocab_size, d_model, nhead, num_encoder_layers, num_decoder_layers, dim_feedforward, dropout=0.1):super(TransformerModel, self).__init__()# vocab_size: 詞匯表的大小,即輸入序列中可能的不同標記的總數。# d_model: 每個嵌入向量的維度,即詞嵌入向量的長度。# nhead: 注意力頭的數量,多頭注意力機制中并行的注意力計算數。# num_encoder_layers: 編碼器層的數量,即堆疊的編碼器層數。# num_decoder_layers: 解碼器層的數量,即堆疊的解碼器層數。# dim_feedforward: 前饋神經網絡的隱藏層維度。# dropout: 在各層中使用的dropout比率,用于正則化。self.embedding = EmbeddingLayer(vocab_size, d_model) # 嵌入層self.encoder = Encoder(num_encoder_layers, d_model, nhead, dim_feedforward, dropout) # 編碼器self.decoder = Decoder(num_decoder_layers, d_model, nhead, dim_feedforward, dropout) # 解碼器self.fc = nn.Linear(d_model, vocab_size) # 最后一層線性變換,將輸出維度映射到詞匯表大小def forward(self, src, tgt):src = self.embedding(src) # 嵌入輸入序列tgt = self.embedding(tgt) # 嵌入目標序列memory = self.encoder(src) # 編碼器處理輸入序列output = self.decoder(tgt, memory) # 解碼器處理目標序列output = self.fc(output) # 映射到詞匯表大小return output
訓練示例
# 參數
# vocab_size: 詞匯表的大小,即輸入序列中可能的不同標記的總數。
# d_model: 每個嵌入向量的維度,即詞嵌入向量的長度。
# nhead: 注意力頭的數量,多頭注意力機制中并行的注意力計算數。
# num_encoder_layers: 編碼器層的數量,即堆疊的編碼器層數。
# num_decoder_layers: 解碼器層的數量,即堆疊的解碼器層數。
# dim_feedforward: 前饋神經網絡的隱藏層維度。
# dropout: 在各層中使用的dropout比率,用于正則化。
# batch_size: 每個訓練批次中的樣本數量。
# seq_length: 輸入序列的長度。
# num_epochs: 訓練的輪數,即遍歷整個訓練集的次數。
vocab_size = 1000
d_model = 512
nhead = 8
num_encoder_layers = 6
num_decoder_layers = 6
dim_feedforward = 2048
dropout = 0.1
batch_size = 32
seq_length = 10
num_epochs = 10# 數據集
src = torch.randint(0, vocab_size, (batch_size, seq_length))
tgt = torch.randint(0, vocab_size, (batch_size, seq_length))dataset = TensorDataset(src, tgt)
dataloader = DataLoader(dataset, batch_size=batch_size, shuffle=True)# 模型實例
model = TransformerModel(vocab_size, d_model, nhead, num_encoder_layers, num_decoder_layers, dim_feedforward, dropout)# 損失函數和優化器
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)# 訓練
for epoch in range(num_epochs):for src_batch, tgt_batch in dataloader:tgt_input = tgt_batch[:, :-1] # 目標輸入tgt_output = tgt_batch[:, 1:] # 目標輸出optimizer.zero_grad()output = model(src_batch, tgt_input) # 前向傳播output = output.permute(1, 2, 0) # 調整形狀以匹配損失函數loss = criterion(output, tgt_output) # 計算損失loss.backward() # 反向傳播optimizer.step() # 更新參數print(f"Epoch {epoch+1}/{num_epochs}, Loss: {loss.item()}")print("訓練完成")
代碼說明
- EmbeddingLayer:將輸入序列和位置嵌入映射到高維空間。
- MultiHeadSelfAttention:實現多頭自注意力機制,包括查詢、鍵和值的線性變換和注意力計算。
- FeedForwardNetwork:前饋神經網絡,用于進一步處理特征。
- LayerNorm:層歸一化,用于穩定訓練過程。
- ResidualConnection:殘差連接,幫助訓練更深的網絡。
- EncoderLayer:將多頭自注意力機制和前饋神經網絡組合在一起,形成編碼器層。
- DecoderLayer:包括多頭自注意力機制、編碼器-解碼器注意力和前饋神經網絡,形成解碼器層。
- Encoder:由多個編碼器層堆疊而成。
- Decoder:由多個解碼器層堆疊而成。
- TransformerModel:將編碼器和解碼器組合在一起,形成完整的Transformer模型。