python手动搭建transformer，并实现自回归推理
以下是添加了详细注释的代码和参数介绍：
Transformer 实现及自回归推理
本文展示了如何手动实现一个简化版的Transformer模型，并用自回归方式实现一个seq2seq任务，例如机器翻译。
导入必要的库
import torch
import torch.nn as nn
import torch.optim as optim
import torch.nn.functional as F
import numpy as np
定义位置编码
Transformer 使用位置编码来捕捉序列中的位置信息。
class PositionalEncoding(nn.Module):def __init__(self, d_model, max_len=5000):super(PositionalEncoding, self).__init__()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() * (-np.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):return x + self.pe[:x.size(0), :]
参数介绍：
d_model: 词嵌入和位置编码的维度。max_len: 序列的最大长度。
定义自注意力机制
class MultiHeadAttention(nn.Module):def __init__(self, d_model, num_heads):super(MultiHeadAttention, self).__init__()self.num_heads = num_headsself.d_model = d_modelassert d_model % num_heads == 0self.depth = d_model // num_headsself.wq = nn.Linear(d_model, d_model)self.wk = nn.Linear(d_model, d_model)self.wv = nn.Linear(d_model, d_model)self.dense = nn.Linear(d_model, d_model)def split_heads(self, x, batch_size):x = x.view(batch_size, -1, self.num_heads, self.depth)return x.permute(0, 2, 1, 3)def forward(self, v, k, q, mask):batch_size = q.size(0)q = self.split_heads(self.wq(q), batch_size)k = self.split_heads(self.wk(k), batch_size)v = self.split_heads(self.wv(v), batch_size)matmul_qk = torch.matmul(q, k.transpose(-1, -2))dk = torch.tensor(k.size(-1)).float()scaled_attention_logits = matmul_qk / torch.sqrt(dk)if mask is not None:scaled_attention_logits += (mask * -1e9)attention_weights = F.softmax(scaled_attention_logits, dim=-1)output = torch.matmul(attention_weights, v)output = output.permute(0, 2, 1, 3).contiguous()output = output.view(batch_size, -1, self.d_model)return self.dense(output)
参数介绍：
d_model: 词嵌入和注意力机制的维度。num_heads: 注意力头的数量。
定义前馈神经网络
class FeedForward(nn.Module):def __init__(self, d_model, dff):super(FeedForward, self).__init__()self.linear1 = nn.Linear(d_model, dff)self.linear2 = nn.Linear(dff, d_model)def forward(self, x):return self.linear2(F.relu(self.linear1(x)))
参数介绍：
d_model: 词嵌入的维度。dff: 前馈神经网络的隐藏层维度。
定义编码器层
class EncoderLayer(nn.Module):def __init__(self, d_model, num_heads, dff, dropout=0.1):super(EncoderLayer, self).__init__()self.mha = MultiHeadAttention(d_model, num_heads)self.ffn = FeedForward(d_model, dff)self.layernorm1 = nn.LayerNorm(d_model)self.layernorm2 = nn.LayerNorm(d_model)self.dropout1 = nn.Dropout(dropout)self.dropout2 = nn.Dropout(dropout)def forward(self, x, mask):attn_output = self.mha(x, x, x, mask)out1 = self.layernorm1(x + self.dropout1(attn_output))ffn_output = self.ffn(out1)out2 = self.layernorm2(out1 + self.dropout2(ffn_output))return out2
参数介绍：
d_model: 词嵌入和注意力机制的维度。num_heads: 注意力头的数量。dff: 前馈神经网络的隐藏层维度。dropout: Dropout 概率。
定义编码器
class Encoder(nn.Module):def __init__(self, num_layers, d_model, num_heads, dff, input_vocab_size, max_len, dropout=0.1):super(Encoder, self).__init__()self.d_model = d_modelself.num_layers = num_layersself.embedding = nn.Embedding(input_vocab_size, d_model)self.pos_encoding = PositionalEncoding(d_model, max_len)self.enc_layers = nn.ModuleList([EncoderLayer(d_model, num_heads, dff, dropout) for _ in range(num_layers)])self.dropout = nn.Dropout(dropout)def forward(self, x, mask):seq_len = x.size(1)x = self.embedding(x)x *= torch.sqrt(torch.tensor(self.d_model, dtype=torch.float32))x = self.pos_encoding(x.permute(1, 0, 2))x = x.permute(1, 0, 2)x = self.dropout(x)for i in range(self.num_layers):x = self.enc_layers[i](x, mask)return x
参数介绍：
num_layers: 编码器层的数量。d_model: 词嵌入和注意力机制的维度。num_heads: 注意力头的数量。dff: 前馈神经网络的隐藏层维度。input_vocab_size: 输入词汇表大小。max_len: 序列的最大长度。dropout: Dropout 概率。
定义解码器层
class DecoderLayer(nn.Module):def __init__(self, d_model, num_heads, dff, dropout=0.1):super(DecoderLayer, self).__init__()self.mha1 = MultiHeadAttention(d_model, num_heads)self.mha2 = MultiHeadAttention(d_model, num_heads)self.ffn = FeedForward(d_model, dff)self.layernorm1 = nn.LayerNorm(d_model)self.layernorm2 = nn.LayerNorm(d_model)self.layernorm3 = nn.LayerNorm(d_model)self.dropout1 = nn.Dropout(dropout)self.dropout2 = nn.Dropout(dropout)self.dropout3 = nn.Dropout(dropout)def forward(self, x, enc_output, look_ahead_mask, padding_mask):attn1 = self.mha1(x, x, x, look_ahead_mask)attn1 = self.dropout1(attn1)out1 = self.layernorm1(attn1 + x)attn2 = self.mha2(enc_output, enc_output, out1, padding_mask)attn2 = self.dropout2(attn2)out2 = self.layernorm2(attn2 + out1)ffn_output = self.ffn(out2)ffn_output = self.dropout3(ffn_output)out3 = self.layernorm3(ffn_output + out2)return out3
参数介绍：
d_model: 词嵌入和注意力机制的维度。num_heads: 注意力头的数量。dff: 前馈神经网络的隐藏层维度。dropout: Dropout 概率。
定义解码器
class Decoder(nn.Module):def __init__(self, num_layers, d_model, num_heads, dff, target_vocab_size, max_len, dropout=0.1):super(Decoder, self).__init__()self.d_model = d_modelself.num_layers = num_layersself.embedding = nn.Embedding(target_vocab_size, d_model)self.pos_encoding = PositionalEncoding(d_model, max_len)self.dec_layers = nn.ModuleList([DecoderLayer(d_model, num_heads, dff, dropout) for _ in range(num_layers)])self.dropout = nn.Dropout(dropout)def forward(self, x, enc_output, look_ahead_mask, padding_mask):seq_len = x.size(1)attention_weights = {}x = self.embedding(x)x *= torch.sqrt(torch.tensor(self.d_model, dtype=torch.float32))x = self.pos_encoding(x.permute(1, 0, 2))x = x.permute(1, 0, 2)x = self.dropout(x)for i in range(self.num_layers):x = self.dec_layers[i](x, enc_output, look_ahead_mask, padding_mask)return x
参数介绍：
num_layers: 解码器层的数量。d_model: 词嵌入和注意力机制的维度。num_heads: 注意力头的数量。dff: 前馈神经网络的隐藏层维度。target_vocab_size: 目标词汇表大小。max_len: 序列的最大长度。dropout: Dropout 概率。
定义Transformer模型
class Transformer(nn.Module):def __init__(self, num_layers, d_model, num_heads, dff, input_vocab_size, target_vocab_size, pe_input, pe_target, dropout=0.1):super(Transformer, self).__init__()self.encoder = Encoder(num_layers, d_model, num_heads, dff, input_vocab_size, pe_input, dropout)self.decoder = Decoder(num_layers, d_model, num_heads, dff, target_vocab_size, pe_target, dropout)self.final_layer = nn.Linear(d_model, target_vocab_size)def forward(self, inp, tar, enc_padding_mask, look_ahead_mask, dec_padding_mask):enc_output = self.encoder(inp, enc_padding_mask)dec_output = self.decoder(tar, enc_output, look_ahead_mask, dec_padding_mask)final_output = self.final_layer(dec_output)return final_output
参数介绍：
num_layers: 编码器和解码器层的数量。d_model: 词嵌入和注意力机制的维度。num_heads: 注意力头的数量。dff: 前馈神经网络的隐藏层维度。input_vocab_size: 输入词汇表大小。target_vocab_size: 目标词汇表大小。pe_input: 输入序列的最大长度。pe_target: 目标序列的最大长度。dropout: Dropout 概率。
创建掩码
def create_padding_mask(seq):seq = torch.eq(seq, 0)return seq[:, None, None, :]def create_look_ahead_mask(size):mask = torch.triu(torch.ones((size, size)), 1)return mask
自回归推理
实现一个简化的自回归推理过程：
def generate_text(model, input_sequence, start_token, max_length, target_vocab_size):generated = [start_token]model.eval()enc_padding_mask = create_padding_mask(input_sequence)with torch.no_grad():enc_output = model.encoder(input_sequence, enc_padding_mask)for _ in range(max_length):dec_input = torch.tensor(generated).unsqueeze(0)look_ahead_mask = create_look_ahead_mask(dec_input.size(1))dec_padding_mask = create_padding_mask(dec_input)with torch.no_grad():output = model.decoder(dec_input, enc_output, look_ahead_mask, dec_padding_mask)output = model.final_layer(output)next_token = torch.argmax(output[:, -1, :], dim=-1).item()generated.append(next_token)if next_token == eos_token:breakreturn generated
参数介绍：
model: 训练好的Transformer模型。input_sequence: 输入的序列张量。start_token: 生成序列的开始标记。max_length: 生成序列的最大长度。target_vocab_size: 目标词汇表大小。
使用示例
创建一个简单的模型并进行文本生成：
input_vocab_size = 1000 # 输入词汇表大小
target_vocab_size = 1000 # 目标词汇表大小
max_len = 50 # 序列最大长度
num_layers = 2 # 编码器和解码器层的数量
d_model = 512 # 词嵌入和注意力机制的维度
num_heads = 8 # 注意力头的数量
dff = 2048 # 前馈神经网络的隐藏层维度# 创建Transformer模型
transformer = Transformer(num_layers, d_model, num_heads, dff, input_vocab_size, target_vocab_size, max_len, max_len)# 输入序列，假设输入序列为[1, 2, 3, 4, 0, 0, 0]
input_sequence = torch.tensor([[1, 2, 3, 4, 0, 0, 0]])# 假设开始标记为1，结束标记为2
start_token = 1
eos_token = 2# 生成序列
generated_sequence = generate_text(transformer, input_sequence, start_token, max_length=20, target_vocab_size=target_vocab_size)
print("Generated sequence:", generated_sequence)
以上代码展示了一个简化的Transformer模型的实现，包括位置编码、自注意力机制、前馈神经网络、编码器层、解码器层、编码器和解码器整体的实现，以及一个基本的自回归推理过程。你可以根据需要进行调整和扩展。
关于mask的解释
以下关于掩码函数 create_padding_mask 和 create_look_ahead_mask 的详细介绍以及示例。
create_padding_mask
该函数用于生成填充掩码，以忽略序列中的填充值（通常是0）。在Transformer模型中，填充掩码用于屏蔽掉填充值在计算注意力时的影响。
代码实现
def create_padding_mask(seq):seq = torch.eq(seq, 0) # 查找填充值（假设填充值为0），返回一个布尔张量return seq[:, None, None, :] # 扩展维度以适配注意力机制中的广播
示例
假设我们有一个输入序列，其中0是填充值：
seq = torch.tensor([[7, 6, 0, 0, 0], [1, 2, 3, 0, 0]])padding_mask = create_padding_mask(seq)
print(padding_mask)
输出
tensor([[[[False, False, True, True, True]]],[[[False, False, False, True, True]]]])
在输出中，True 表示填充值的位置，这些位置将在计算注意力时被忽略。
create_look_ahead_mask
该函数用于生成前瞻掩码，以确保解码器中的每个位置只能看到该位置之前的序列，不能看到未来的信息。在自回归生成中，前瞻掩码用于防止解码器在生成下一个标记时看到未来的标记。
代码实现
def create_look_ahead_mask(size):mask = torch.triu(torch.ones((size, size)), 1) # 生成上三角矩阵，主对角线以上的元素为1return mask # 返回前瞻掩码
示例
假设我们有一个序列长度为5：
size = 5
look_ahead_mask = create_look_ahead_mask(size)
print(look_ahead_mask)
输出
tensor([[0., 1., 1., 1., 1.],[0., 0., 1., 1., 1.],[0., 0., 0., 1., 1.],[0., 0., 0., 0., 1.],[0., 0., 0., 0., 0.]])
在输出中，1 表示被掩盖的位置，这些位置在计算注意力时将被屏蔽。
综合示例
结合以上两种掩码，假设我们有以下输入序列：
seq = torch.tensor([[7, 6, 0, 0, 0], [1, 2, 3, 0, 0]])
size = seq.size(1)padding_mask = create_padding_mask(seq)
look_ahead_mask = create_look_ahead_mask(size)print("Padding Mask:\n", padding_mask)
print("Look Ahead Mask:\n", look_ahead_mask)
输出
Padding Mask:tensor([[[[False, False, True, True, True]]],[[[False, False, False, True, True]]]])
Look Ahead Mask:tensor([[0., 1., 1., 1., 1.],[0., 0., 1., 1., 1.],[0., 0., 0., 1., 1.],[0., 0., 0., 0., 1.],[0., 0., 0., 0., 0.]])
在实际使用中，编码器使用 padding_mask 来屏蔽填充值的影响，解码器则同时使用 look_ahead_mask 和 padding_mask 来屏蔽未来标记和填充值的影响。