首先把所有代码放在一块,整体上了解一下transformer的结构和原理,后面继续分段介绍。
以下是一个基于代码的简单示例,演示Transformer模型的原理:
import torch import torch.nn as nn import torch.nn.functional as F class Transformer(nn.Module): def __init__(self, input_size, hidden_size, num_heads, num_layers): super(Transformer, self).__init__() self.embedding = nn.Embedding(input_size, hidden_size) self.positional_encoding = PositionalEncoding(hidden_size) self.encoder = nn.ModuleList([ EncoderLayer(hidden_size, num_heads) for _ in range(num_layers) ]) self.decoder = nn.ModuleList([ DecoderLayer(hidden_size, num_heads) for _ in range(num_layers) ]) self.fc = nn.Linear(hidden_size, input_size) def forward(self, src, trg): src = self.embedding(src) # 输入序列的嵌入表示 trg = self.embedding(trg) # 目标序列的嵌入表示 src = self.positional_encoding(src) # 加入位置编码 # 编码器 for encoder_layer in self.encoder: src = encoder_layer(src) trg = self.positional_encoding(trg) # 加入位置编码 # 解码器 for decoder_layer in self.decoder: trg = decoder_layer(src, trg) output = self.fc(trg) # 最后的线性变换得到输出结果 return F.log_softmax(output, dim=-1) class EncoderLayer(nn.Module): def __init__(self, hidden_size, num_heads): super(EncoderLayer, self).__init__() self.multihead_attention = MultiheadAttention(hidden_size, num_heads) self.feed_forward = FeedForward(hidden_size) def forward(self, x): residual = x x = self.multihead_attention(x) x = x + residual residual = x x = self.feed_forward(x) x = x + residual return x class DecoderLayer(nn.Module): def __init__(self, hidden_size, num_heads): super(DecoderLayer, self).__init__() self.masked_multihead_attention = MultiheadAttention(hidden_size, num_heads) self.multihead_attention = MultiheadAttention(hidden_size, num_heads) self.feed_forward = FeedForward(hidden_size) def forward(self, src, trg): residual = trg trg = self.masked_multihead_attention(trg) trg = trg + residual residual = trg trg = self.multihead_attention(src, trg) trg = trg + residual residual = trg trg = self.feed_forward(trg) trg = trg + residual return trg class MultiheadAttention(nn.Module): def __init__(self, hidden_size, num_heads): super(MultiheadAttention, self).__init__() self.num_heads = num_heads self.head_size = hidden_size // num_heads self.q_linear = nn.Linear(hidden_size, hidden_size) self.k_linear = nn.Linear(hidden_size, hidden_size) self.v_linear = nn.Linear(hidden_size, hidden_size) self.fc = nn.Linear(hidden_size, hidden_size) def forward(self, x): batch_size, seq_len, hidden_size = x.size() q = self.q_linear(x).view(batch_size, seq_len, self.num_heads, self.head_size).transpose(1, 2) k = self.k_linear(x).view(batch_size, seq_len, self.num_heads, self.head_size).transpose(1 , 2) v = self.v_linear(x).view(batch_size, seq_len, self.num_heads, self.head_size).transpose(1, 2) attention_scores = torch.matmul(q, k.transpose(-2, -1)) / torch.sqrt(torch.tensor(self.head_size, dtype=torch.float32)) attention_weights = F.softmax(attention_scores, dim=-1) x = torch.matmul(attention_weights, v).transpose(1, 2).contiguous().view(batch_size, seq_len, hidden_size) x = self.fc(x) return x class FeedForward(nn.Module): def __init__(self, hidden_size, feed_forward_size=2048): super(FeedForward, self).__init__() self.fc1 = nn.Linear(hidden_size, feed_forward_size) self.fc2 = nn.Linear(feed_forward_size, hidden_size) def forward(self, x): x = F.relu(self.fc1(x)) x = self.fc2(x) return x class PositionalEncoding(nn.Module): def __init__(self, hidden_size, max_seq_len=5000): super(PositionalEncoding, self).__init__() self.positional_encoding = self.generate_positional_encoding(hidden_size, max_seq_len) def forward(self, x): x = x + self.positional_encoding[:x.size(0), :] return x def generate_positional_encoding(self, hidden_size, max_seq_len): positional_encoding = torch.zeros(max_seq_len, hidden_size) position = torch.arange(0, max_seq_len, dtype=torch.float32).unsqueeze(1) div_term = torch.exp(torch.arange(0, hidden_size, 2, dtype=torch.float32) * -(math.log(10000.0) / hidden_size)) positional_encoding[:, 0::2] = torch.sin(position * div_term) positional_encoding[:, 1::2] = torch.cos(position * div_term) return positional_encoding
这个代码示例实现了一个简化版的Transformer模型,包括编码器(Encoder)和解码器(Decoder)以及注意力机制(MultiheadAttention)、前馈神经网络(FeedForward)和位置编码(PositionalEncoding)等组件。在模型的前向传播过程中,输入的源语言和目标语言序列经过嵌入层(Embedding)和位置编码层(PositionalEncoding)后,分别经过编码器和解码器的多层堆叠。在每个编码器和解码器层中,都包含了自注意力机制和前馈神经网络层。最终,通过线性变换层(Linear)和softmax函数将解码器的输出转化为概率分布,用于预测目标语言序列。 请注意,这只是一个简化的示例代码,实际的Transformer模型可能包含更多的细节和优化。此示例旨在提供一个基本的理解,帮助解释Transformer模型的核心原理和组成部分。
以下是分段介绍各部分的含义。
import torch import torch.nn as nn import torch.nn.functional as F
这些代码段导入了PyTorch库和所需的模块。
class Transformer(nn.Module): def __init__(self, input_size, hidden_size, num_heads, num_layers): super(Transformer, self).__init__() self.embedding = nn.Embedding(input_size, hidden_size) self.positional_encoding = PositionalEncoding(hidden_size) self.encoder = nn.ModuleList([ EncoderLayer(hidden_size, num_heads) for _ in range(num_layers) ]) self.decoder = nn.ModuleList([ DecoderLayer(hidden_size, num_heads) for _ in range(num_layers) ]) self.fc = nn.Linear(hidden_size, input_size) def forward(self, src, trg): src = self.embedding(src) # 输入序列的嵌入表示 trg = self.embedding(trg) # 目标序列的嵌入表示 src = self.positional_encoding(src) # 加入位置编码 # 编码器 for encoder_layer in self.encoder: src = encoder_layer(src) trg = self.positional_encoding(trg) # 加入位置编码 # 解码器 for decoder_layer in self.decoder: trg = decoder_layer(src, trg) output = self.fc(trg) # 最后的线性变换得到输出结果 return F.log_softmax(output, dim=-1)
这段代码定义了一个名为Transformer的模型类。在初始化方法__init__
中,模型类定义了Transformer的各个组件。具体来说:
embedding
是一个嵌入层(Embedding Layer),用于将输入的整数序列转换为密集向量表示。positional_encoding
是一个位置编码层(PositionalEncoding),用于为序列中的每个位置添加编码信息,以保留序列中的位置顺序信息。encoder
是一个由多个编码器层(EncoderLayer)组成的编码器堆叠。在初始化时,通过循环创建了指定数量的编码器层,并将其存储在nn.ModuleList
中。decoder
是一个由多个解码器层(DecoderLayer)组成的解码器堆叠。与编码器类似,它也通过循环创建了指定数量的解码器层,并将其存储在nn.ModuleList
中。fc
是一个线性变换层(Linear Layer),将解码器的输出转换为与目标词汇表大小相同的张量。最后,通过应用log_softmax
函数,将输出转换为对目标语言词汇表中每个单词的对数概率。
在前向传播方法forward
中,输入的源语言和目标语言序列经过嵌入层和位置编码层后,分别传入编码器和解码器的多层堆叠中。编码器层循环遍历,并依次对源语言序列进行编码。解码器层也类似地循环遍历,并在每个解码器层中,传入源语言序列和目标语言序列,进行解码操作。最后,通过线性变换层将解码器的输出转化为概率分布,并应用log_softmax
函数进行标准化,以便用于预测目标语言序列。
class EncoderLayer(nn.Module): def __init__(self, hidden_size, num_heads): super(EncoderLayer, self).__init__() self.multihead_attention = MultiheadAttention(hidden_size, num_heads) self.feed_forward = FeedForward(hidden_size) def forward(self, x): residual = x x = self.multihead_attention(x) x = x + residual residual = x x = self.feed_forward(x) x = x + residual return x
这段代码定义了编码器层(EncoderLayer)类。编码器层由两个子层组成:
multihead_attention
是一个多头注意力机制层(MultiheadAttention),用于在输入序列内部进行自注意力计算。feed_forward
是一个前馈神经网络层(FeedForward),用于对每个位置的向量进行非线性变换。
在前向传播方法中,输入通过多头注意力机制层,然后与输入进行残差连接(residual connection)并进行加法操作。接着,将结果输入到前馈神经网络层,并再次与之前的结果进行残差连接和加法操作。最后,输出经过编码器层的变换后返回。
class DecoderLayer(nn.Module): def __init__(self, hidden_size, num_heads): super(DecoderLayer, self).__init__() self.masked_multihead_attention = MultiheadAttention(hidden_size, num_heads) self.multihead_attention = MultiheadAttention(hidden_size, num_heads) self.feed_forward = FeedForward(hidden_size) def forward(self, src, trg): residual = trg trg = self.masked_multihead_attention(trg) trg = trg + residual residual = trg trg = self.multihead_attention(src, trg) trg = trg + residual residual = trg trg = self.feed_forward(trg) trg = trg + residual return trg
这段代码定义了解码器层(DecoderLayer)类。解码器层由三个子层组成:
masked_multihead_attention
是一个带有遮蔽的多头注意力机制层(MultiheadAttention),用于在解码器内部进行自注意力计算,并且在计算过程中遮蔽(mask)了未来的信息。multihead_attention
是一个多头注意力机制层(MultiheadAttention),用于使解码器能够关注编码器的输出。feed_forward
是一个前馈神经网络层(FeedForward),用于对每个位置的向量进行非线性变换。
在前向传播方法中,输入的解码器序列通过带有遮蔽的多头注意力机制层。然后将结果与输入进行残差连接并进行加法操作。接下来,将结果传入多头注意力机制层,以便解码器关注编码器的输出。同样地,将结果与之前的结果进行残差连接和加法操作。最后,将结果输入到前馈神经网络层,并再次进行残差连接和加法操作。输出经过解码器层的变换后返回。
class MultiheadAttention(nn.Module): def __init__(self, hidden_size, num_heads): super(MultiheadAttention, self).__init__() self.num_heads = num_heads self.head_size = hidden_size // num_heads self.q_linear = nn.Linear(hidden_size, hidden_size) self.k_linear = nn.Linear(hidden_size, hidden_size) self.v_linear = nn.Linear(hidden_size, hidden_size) self.fc = nn.Linear(hidden_size, hidden_size) def forward(self, x): batch_size, seq_len, hidden_size = x.size() q = self.q_linear(x).view(batch_size, seq_len, self.num_heads, self.head_size).transpose(1, 2) k = self.k_linear(x).view(batch_size, seq_len, self.num_heads, self.head_size).transpose(1, 2) v = self.v_linear(x).view(batch_size, seq_len, self.num_heads, self.head_size).transpose(1, 2) attention_scores = torch.matmul(q, k.transpose(-2, -1)) / torch.sqrt(torch.tensor(self.head_size, dtype=torch.float32)) attention_weights = F.softmax(attention_scores, dim=-1) x = torch.matmul(attention_weights, v).transpose(1, 2).contiguous().view(batch_size, seq_len, hidden_size) x = self.fc(x) return x
这段代码定义了多头注意力机制层(MultiheadAttention)。在初始化方法中,指定了注意力头数(num_heads)和每个注意力头的大小(head_size)。
在前向传播方法中,输入通过三个线性变换层(q_linear
、k_linear
和v_linear
)进行线性变换,并进行形状重塑操作,以便进行多头注意力计算。然后,通过矩阵乘法计算注意力分数。注意力分数除以根号下注意力头大小,以进行缩放。接着,通过softmax函数计算注意力权重。
然后,将注意力权重与值(v
)相乘,得到注意力加权的表示。最后,通过转置和形状重塑操作,将结果重新恢复为原始形状,并通过线性变换层进行最终的变换。
class FeedForward(nn.Module): def __init__(self, hidden_size, feed_forward_size=2048): super(FeedForward, self).__init__() self.fc1 = nn.Linear(hidden_size, feed_forward_size) self.fc2 = nn.Linear(feed_forward_size, hidden_size) def forward(self, x): x = F.relu(self.fc1(x)) x = self.fc2(x) return x
这段代码定义了前馈神经网络层(FeedForward)。它由两个线性变换层(fc1
和fc2
)组成,中间使用了ReLU激活函数。
在前向传播方法中,输入通过第一个线性变换层,并经过ReLU激活函数进行非线性变换。然后,结果通过第二个线性变换层进行最终的线性变换,并返回结果。
class PositionalEncoding(nn.Module): def __init__(self, hidden_size, max_seq_len=5000): super(PositionalEncoding, self).__init__() self.positional_encoding = self.generate_positional_encoding(hidden_size, max_seq_len) def forward(self, x): x = x + self.positional_encoding[:x.size(0), :] return x def generate_positional_encoding(self, hidden_size, max_seq_len): positional_encoding = torch.zeros(max_seq_len, hidden_size) position = torch.arange(0, max_seq_len, dtype=torch.float32).unsqueeze(1) div_term = torch.exp(torch.arange(0, hidden_size, 2, dtype=torch.float32) * -(math.log(10000.0) / hidden_size)) positional_encoding[:, 0::2] = torch.sin(position * div_term) positional_encoding[:, 1::2] = torch.cos(position * div_term) return positional_encoding
这段代码定义了位置编码层(PositionalEncoding)。位置编码用于为序列中的每个位置添加编码信息,以保留序列中的位置顺序信息。
在初始化方法中,通过调用generate_positional_encoding
方法生成了位置编码矩阵。在前向传播方法中,将输入序列与位置编码矩阵相加,以添加位置编码信息。最后,返回添加位置编码后的结果。
generate_positional_encoding
方法根据位置和维度计算位置编码矩阵。首先,创建一个全零矩阵,大小为最大序列长度(max_seq_len
)乘以隐藏维度(hidden_size
)。然后,通过一系列的数学运算计算每个位置的编码值,并将奇数维度和偶数维度分别填充到位置编码矩阵的对应位置。最后,返回生成的位置编码矩阵。
以上是基于代码的Transformer模型的详细介绍。这个模型由编码器、解码器、注意力机制、前馈神经网络和位置编码等组件构成,用于处理序列数据的建模任务。通过堆叠多个编码器和解码器层,Transformer模型能够捕捉序列中的长距离依赖关系,并在机器翻译、文本生成等任务中取得了显著的效果。