Transformer是一种用于自然语言处理（NLP）和其他序列到序列任务的深度学习模型架构，在2017年由Vaswani等人首次提出，它作为大语言模型的基石架构，彻底改变了序列建模领域，如机器翻译、文本生成等[^2][^3][^4]。 ### 核心创新 Transformer架构引入的自注意力机制是一个关键创新，它解决了循环神经网络（RNN）无法并行计算和长期依赖的问题，在处理序列数据时表现出色。该机制使模型能够在处理序列中的每个元素时，考虑到序列中其他元素的信息，从而捕获全局依赖关系[^2][^4]。 ### 整体架构 Transformer由编码器（Encoder）和解码器（Decoder）堆叠而成，编码器负责理解输入序列，解码器负责生成输出序列。核心组件包括词嵌入、位置编码、多头注意力和前馈神经网络等。每个层包含多头自注意力（Multi - Head Self - Attention）、前馈网络（Feed - Forward Network, FFN）、残差连接（Residual Connection）和层归一化（LayerNorm） [^4][^5]。 ### 编码器 编码器的主要作用是捕获输入序列的全局依赖关系。其工作流程如下： 1. **词嵌入**：将输入的词转换为对应的向量表示，使得模型能够处理文本信息。 2. **位置编码**：由于Transformer本身没有像RNN那样的顺序结构，位置编码用于为每个词的向量表示添加位置信息，帮助模型理解词在序列中的位置。 3. **多头自注意力**：通过多个不同的注意力头并行计算，模型可以从不同的表示子空间中捕捉序列元素之间的依赖关系，最后将多个头的输出拼接并线性变换得到最终结果。 4. **前馈神经网络**：对多头自注意力的输出进行非线性变换，增强模型的表达能力。 5. **残差连接和层归一化**：残差连接有助于缓解梯度消失问题，使模型更容易训练；层归一化则对每一层的输入进行归一化处理，加速模型收敛。 ### 解码器 解码器通过带掩码的自注意力和编码器 - 解码器注意力生成输出。具体过程如下： 1. **带掩码的自注意力**：在生成输出序列时，为了防止模型看到未来的信息，需要对自注意力进行掩码操作，确保模型只能关注到当前位置及之前的元素。 2. **编码器 - 解码器注意力**：解码器在生成每个输出元素时，会参考编码器的输出，从而结合输入序列的信息进行输出。 3. **前馈神经网络、残差连接和层归一化**：与编码器中的作用类似，进一步处理和优化解码器的输出。 ### 与TensorFlow的结合 理解Transformer的架构，结合TensorFlow的工程实践，开发者能将这一理论快速转化为解决实际问题的生产力，这也是Transformer与TensorFlow组合在工业界长盛不衰的核心原因 [^1]。 ### 代码示例 以下是一个简单的Transformer模型的代码框架（使用TensorFlow）： ```python import tensorflow as tf from tensorflow.keras import layers # 多头注意力层 class MultiHeadAttention(tf.keras.layers.Layer): def __init__(self, d_model, num_heads): super(MultiHeadAttention, self).__init__() self.num_heads = num_heads self.d_model = d_model assert d_model % self.num_heads == 0 self.depth = d_model // self.num_heads self.wq = tf.keras.layers.Dense(d_model) self.wk = tf.keras.layers.Dense(d_model) self.wv = tf.keras.layers.Dense(d_model) self.dense = tf.keras.layers.Dense(d_model) def split_heads(self, x, batch_size): """Split the last dimension into (num_heads, depth). Transpose the result such that the shape is (batch_size, num_heads, seq_len, depth) """ x = tf.reshape(x, (batch_size, -1, self.num_heads, self.depth)) return tf.transpose(x, perm=[0, 2, 1, 3]) def call(self, v, k, q, mask): batch_size = tf.shape(q)[0] q = self.wq(q) # (batch_size, seq_len, d_model) k = self.wk(k) # (batch_size, seq_len, d_model) v = self.wv(v) # (batch_size, seq_len, d_model) q = self.split_heads(q, batch_size) # (batch_size, num_heads, seq_len_q, depth) k = self.split_heads(k, batch_size) # (batch_size, num_heads, seq_len_k, depth) v = self.split_heads(v, batch_size) # (batch_size, num_heads, seq_len_v, depth) # scaled_attention.shape == (batch_size, num_heads, seq_len_q, depth) # attention_weights.shape == (batch_size, num_heads, seq_len_q, seq_len_k) scaled_attention, attention_weights = self.scaled_dot_product_attention( q, k, v, mask) scaled_attention = tf.transpose(scaled_attention, perm=[0, 2, 1, 3]) # (batch_size, seq_len_q, num_heads, depth) concat_attention = tf.reshape(scaled_attention, (batch_size, -1, self.d_model)) # (batch_size, seq_len_q, d_model) output = self.dense(concat_attention) # (batch_size, seq_len_q, d_model) return output, attention_weights def scaled_dot_product_attention(self, q, k, v, mask): """Calculate the attention weights. q, k, v must have matching leading dimensions. k, v must have matching penultimate dimension, i.e.: seq_len_k = seq_len_v. The mask has different shapes depending on its type(padding or look ahead) but it must be broadcastable for addition. Args: q: query shape == (..., seq_len_q, depth) k: key shape == (..., seq_len_k, depth) v: value shape == (..., seq_len_v, depth_v) mask: Float tensor with shape broadcastable to (..., seq_len_q, seq_len_k). Defaults to None. Returns: output, attention_weights """ matmul_qk = tf.matmul(q, k, transpose_b=True) # (..., seq_len_q, seq_len_k) # scale matmul_qk dk = tf.cast(tf.shape(k)[-1], tf.float32) scaled_attention_logits = matmul_qk / tf.math.sqrt(dk) # add the mask to the scaled tensor. if mask is not None: scaled_attention_logits += (mask * -1e9) # softmax is normalized on the last axis (seq_len_k) so that the scores # add up to 1. attention_weights = tf.nn.softmax(scaled_attention_logits, axis=-1) # (..., seq_len_q, seq_len_k) output = tf.matmul(attention_weights, v) # (..., seq_len_q, depth_v) return output, attention_weights # 前馈网络 def point_wise_feed_forward_network(d_model, dff): return tf.keras.Sequential([ tf.keras.layers.Dense(dff, activation='relu'), # (batch_size, seq_len, dff) tf.keras.layers.Dense(d_model) # (batch_size, seq_len, d_model) ]) # 编码器层 class EncoderLayer(tf.keras.layers.Layer): def __init__(self, d_model, num_heads, dff, rate=0.1): super(EncoderLayer, self).__init__() self.mha = MultiHeadAttention(d_model, num_heads) self.ffn = point_wise_feed_forward_network(d_model, dff) self.layernorm1 = tf.keras.layers.LayerNormalization(epsilon=1e-6) self.layernorm2 = tf.keras.layers.LayerNormalization(epsilon=1e-6) self.dropout1 = tf.keras.layers.Dropout(rate) self.dropout2 = tf.keras.layers.Dropout(rate) def call(self, x, training, mask): attn_output, _ = self.mha(x, x, x, mask) # (batch_size, input_seq_len, d_model) attn_output = self.dropout1(attn_output, training=training) out1 = self.layernorm1(x + attn_output) # (batch_size, input_seq_len, d_model) ffn_output = self.ffn(out1) # (batch_size, input_seq_len, d_model) ffn_output = self.dropout2(ffn_output, training=training) out2 = self.layernorm2(out1 + ffn_output) # (batch_size, input_seq_len, d_model) return out2 # 解码器层 class DecoderLayer(tf.keras.layers.Layer): def __init__(self, d_model, num_heads, dff, rate=0.1): super(DecoderLayer, self).__init__() self.mha1 = MultiHeadAttention(d_model, num_heads) self.mha2 = MultiHeadAttention(d_model, num_heads) self.ffn = point_wise_feed_forward_network(d_model, dff) self.layernorm1 = tf.keras.layers.LayerNormalization(epsilon=1e-6) self.layernorm2 = tf.keras.layers.LayerNormalization(epsilon=1e-6) self.layernorm3 = tf.keras.layers.LayerNormalization(epsilon=1e-6) self.dropout1 = tf.keras.layers.Dropout(rate) self.dropout2 = tf.keras.layers.Dropout(rate) self.dropout3 = tf.keras.layers.Dropout(rate) def call(self, x, enc_output, training, look_ahead_mask, padding_mask): # enc_output.shape == (batch_size, input_seq_len, d_model) attn1, attn_weights_block1 = self.mha1(x, x, x, look_ahead_mask) # (batch_size, target_seq_len, d_model) attn1 = self.dropout1(attn1, training=training) out1 = self.layernorm1(attn1 + x) attn2, attn_weights_block2 = self.mha2( enc_output, enc_output, out1, padding_mask) # (batch_size, target_seq_len, d_model) attn2 = self.dropout2(attn2, training=training) out2 = self.layernorm2(attn2 + out1) # (batch_size, target_seq_len, d_model) ffn_output = self.ffn(out2) # (batch_size, target_seq_len, d_model) ffn_output = self.dropout3(ffn_output, training=training) out3 = self.layernorm3(ffn_output + out2) # (batch_size, target_seq_len, d_model) return out3, attn_weights_block1, attn_weights_block2 ```