# Transformer模型架构解析：自注意力机制在自然语言处理中的创新应用

## 架构设计的核心思想

Transformer模型在2017年由Vaswani等人提出，通过完全基于注意力机制的架构革新了序列建模范式。该架构摒弃了循环神经网络和卷积神经网络的传统设计，采用自注意力机制实现序列元素的全局依赖建模，在处理长距离依赖关系方面展现出显著优势。

## 自注意力机制实现原理

### 缩放点积注意力

自注意力机制的核心是缩放点积注意力，通过查询、键、值三元组实现信息聚合：

```python

import torch

import torch.nn as nn

import torch.nn.functional as F

import math

class ScaledDotProductAttention(nn.Module):

    """缩放点积注意力机制"""

    def __init__(self, dropout_rate=0.1):

        super().__init__()

        self.dropout = nn.Dropout(dropout_rate)

    

    def forward(self, queries, keys, values, mask=None):

        """

        计算缩放点积注意力

        

        参数:

            queries: [batch_size, seq_len_q, d_k]

            keys: [batch_size, seq_len_k, d_k]

            values: [batch_size, seq_len_v, d_v]

            mask: 可选的注意力掩码

            

        返回:

            attention_output: [batch_size, seq_len_q, d_v]

            attention_weights: [batch_size, seq_len_q, seq_len_k]

        """

        batch_size, seq_len_q, d_k = queries.size()

        _, seq_len_k, _ = keys.size()

        

        # 计算点积并缩放

        scores = torch.matmul(queries, keys.transpose(1, 2)) / math.sqrt(d_k)

        

        # 应用掩码（如因果掩码）

        if mask is not None:

            scores = scores.masked_fill(mask == 0, -1e9)

        

        # 计算注意力权重

        attention_weights = F.softmax(scores, dim=-1)

        attention_weights = self.dropout(attention_weights)

        

        # 加权求和

        output = torch.matmul(attention_weights, values)

        

        return output, attention_weights

def causal_attention_mask(seq_len):

    """

    生成因果注意力掩码（用于自回归任务）

    """

    mask = torch.triu(torch.ones(seq_len, seq_len), diagonal=1)

    return mask == 0  # 下三角为True，允许访问；上三角为False，禁止访问

```

### 多头注意力机制

多头注意力通过并行多个注意力头捕获不同类型的依赖关系：

```python

class MultiHeadAttention(nn.Module):

    """多头注意力机制"""

    def __init__(self, d_model=512, num_heads=8, dropout_rate=0.1):

        super().__init__()

        assert d_model % num_heads == 0, "d_model必须能被num_heads整除"

        

        self.d_model = d_model

        self.num_heads = num_heads

        self.d_k = d_model // num_heads

        self.d_v = d_model // num_heads

        

        # 线性变换层

        self.w_q = nn.Linear(d_model, d_model)

        self.w_k = nn.Linear(d_model, d_model)

        self.w_v = nn.Linear(d_model, d_model)

        self.w_o = nn.Linear(d_model, d_model)

        

        self.attention = ScaledDotProductAttention(dropout_rate)

        self.dropout = nn.Dropout(dropout_rate)

    

    def split_heads(self, x, batch_size):

        """将输入分割为多个注意力头"""

        # x形状: [batch_size, seq_len, d_model]

        x = x.view(batch_size, -1, self.num_heads, self.d_k)

        # 重排维度: [batch_size, num_heads, seq_len, d_k]

        return x.transpose(1, 2)

    

    def combine_heads(self, x, batch_size):

        """合并多个注意力头的输出"""

        # x形状: [batch_size, num_heads, seq_len, d_k]

        x = x.transpose(1, 2).contiguous()

        # 合并: [batch_size, seq_len, d_model]

        return x.view(batch_size, -1, self.d_model)

    

    def forward(self, q, k, v, mask=None):

        batch_size = q.size(0)

        

        # 线性变换并分割头

        q = self.split_heads(self.w_q(q), batch_size)

        k = self.split_heads(self.w_k(k), batch_size)

        v = self.split_heads(self.w_v(v), batch_size)

        

        # 计算注意力

        attention_output, attention_weights = self.attention(q, k, v, mask)

        

        # 合并头并输出变换

        combined = self.combine_heads(attention_output, batch_size)

        output = self.w_o(combined)

        output = self.dropout(output)

        

        return output, attention_weights

```

## Transformer编码器实现

### 前馈网络层

```python

class PositionwiseFeedForward(nn.Module):

    """位置感知前馈网络"""

    def __init__(self, d_model=512, d_ff=2048, dropout_rate=0.1):

        super().__init__()

        self.linear1 = nn.Linear(d_model, d_ff)

        self.linear2 = nn.Linear(d_ff, d_model)

        self.dropout = nn.Dropout(dropout_rate)

        self.relu = nn.ReLU()

    

    def forward(self, x):

        # 两层线性变换，中间有ReLU激活

        return self.linear2(self.dropout(self.relu(self.linear1(x))))

class EncoderLayer(nn.Module):

    """Transformer编码器层"""

    def __init__(self, d_model=512, num_heads=8, d_ff=2048, dropout_rate=0.1):

        super().__init__()

        

        self.self_attention = MultiHeadAttention(d_model, num_heads, dropout_rate)

        self.feed_forward = PositionwiseFeedForward(d_model, d_ff, dropout_rate)

        

        # 层归一化

        self.norm1 = nn.LayerNorm(d_model)

        self.norm2 = nn.LayerNorm(d_model)

        

        # Dropout层

        self.dropout1 = nn.Dropout(dropout_rate)

        self.dropout2 = nn.Dropout(dropout_rate)

    

    def forward(self, x, mask=None):

        # 自注意力子层（含残差连接）

        attention_output, _ = self.self_attention(x, x, x, mask)

        x = x + self.dropout1(attention_output)

        x = self.norm1(x)

        

        # 前馈网络子层（含残差连接）

        ff_output = self.feed_forward(x)

        x = x + self.dropout2(ff_output)

        x = self.norm2(x)

        

        return x

```

### 位置编码机制

```python

class PositionalEncoding(nn.Module):

    """正弦位置编码"""

    def __init__(self, d_model, max_len=5000):

        super().__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() * 

                           (-math.log(10000.0) / d_model))

        

        pe[:, 0::2] = torch.sin(position * div_term)

        pe[:, 1::2] = torch.cos(position * div_term)

        

        # 注册为缓冲区（不参与训练）

        self.register_buffer('pe', pe.unsqueeze(0))

    

    def forward(self, x):

        # x形状: [batch_size, seq_len, d_model]

        seq_len = x.size(1)

        return x + self.pe[:, :seq_len, :]

class LearnedPositionalEncoding(nn.Module):

    """可学习的位置编码"""

    def __init__(self, d_model, max_len=512):

        super().__init__()

        self.position_embedding = nn.Embedding(max_len, d_model)

        

    def forward(self, x):

        batch_size, seq_len = x.size(0), x.size(1)

        positions = torch.arange(seq_len, device=x.device).unsqueeze(0).expand(batch_size, seq_len)

        position_embeddings = self.position_embedding(positions)

        return x + position_embeddings

```

### 完整编码器实现

```python

class TransformerEncoder(nn.Module):

    """Transformer编码器"""

    def __init__(self, num_layers=6, d_model=512, num_heads=8, 

                 d_ff=2048, vocab_size=30000, max_len=512, 

                 dropout_rate=0.1, use_learned_pos=True):

        super().__init__()

        

        self.d_model = d_model

        

        # 词嵌入层

        self.token_embedding = nn.Embedding(vocab_size, d_model)

        

        # 位置编码

        if use_learned_pos:

            self.position_encoding = LearnedPositionalEncoding(d_model, max_len)

        else:

            self.position_encoding = PositionalEncoding(d_model, max_len)

        

        # 编码器层堆叠

        self.layers = nn.ModuleList([

            EncoderLayer(d_model, num_heads, d_ff, dropout_rate)

            for _ in range(num_layers)

        ])

        

        # 输出层归一化

        self.norm = nn.LayerNorm(d_model)

        self.dropout = nn.Dropout(dropout_rate)

    

    def forward(self, token_ids, mask=None):

        # 词嵌入

        x = self.token_embedding(token_ids) * math.sqrt(self.d_model)

        

        # 位置编码

        x = self.position_encoding(x)

        x = self.dropout(x)

        

        # 通过编码器层

        for layer in self.layers:

            x = layer(x, mask)

        

        # 最终归一化

        return self.norm(x)

```

## Transformer解码器实现

### 解码器层设计

```python

class DecoderLayer(nn.Module):

    """Transformer解码器层"""

    def __init__(self, d_model=512, num_heads=8, d_ff=2048, dropout_rate=0.1):

        super().__init__()

        

        # 掩码自注意力

        self.self_attention = MultiHeadAttention(d_model, num_heads, dropout_rate)

        

        # 编码器-解码器注意力

        self.cross_attention = MultiHeadAttention(d_model, num_heads, dropout_rate)

        

        # 前馈网络

        self.feed_forward = PositionwiseFeedForward(d_model, d_ff, dropout_rate)

        

        # 层归一化

        self.norm1 = nn.LayerNorm(d_model)

        self.norm2 = nn.LayerNorm(d_model)

        self.norm3 = nn.LayerNorm(d_model)

        

        # Dropout层

        self.dropout1 = nn.Dropout(dropout_rate)

        self.dropout2 = nn.Dropout(dropout_rate)

        self.dropout3 = nn.Dropout(dropout_rate)

    

    def forward(self, x, encoder_output, self_mask=None, cross_mask=None):

        # 掩码自注意力子层

        attention_output, _ = self.self_attention(x, x, x, self_mask)

        x = x + self.dropout1(attention_output)

        x = self.norm1(x)

        

        # 编码器-解码器注意力子层

        cross_output, _ = self.cross_attention(x, encoder_output, encoder_output, cross_mask)

        x = x + self.dropout2(cross_output)

        x = self.norm2(x)

        

        # 前馈网络子层

        ff_output = self.feed_forward(x)

        x = x + self.dropout3(ff_output)

        x = self.norm3(x)

        

        return x

```

### 完整解码器实现

```python

class TransformerDecoder(nn.Module):

    """Transformer解码器"""

    def __init__(self, num_layers=6, d_model=512, num_heads=8,

                 d_ff=2048, vocab_size=30000, max_len=512,

                 dropout_rate=0.1, use_learned_pos=True):

        super().__init__()

        

        self.d_model = d_model

        

        # 词嵌入层

        self.token_embedding = nn.Embedding(vocab_size, d_model)

        

        # 位置编码

        if use_learned_pos:

            self.position_encoding = LearnedPositionalEncoding(d_model, max_len)

        else:

            self.position_encoding = PositionalEncoding(d_model, max_len)

        

        # 解码器层堆叠

        self.layers = nn.ModuleList([

            DecoderLayer(d_model, num_heads, d_ff, dropout_rate)

            for _ in range(num_layers)

        ])

        

        # 输出层

        self.norm = nn.LayerNorm(d_model)

        self.output_projection = nn.Linear(d_model, vocab_size)

        self.dropout = nn.Dropout(dropout_rate)

    

    def forward(self, target_ids, encoder_output, 

                target_mask=None, memory_mask=None):

        # 目标序列嵌入

        x = self.token_embedding(target_ids) * math.sqrt(self.d_model)

        

        # 位置编码

        x = self.position_encoding(x)

        x = self.dropout(x)

        <"cec.a8k1.org.cn"><"vds.a8k1.org.cn">

        # 通过解码器层

        for layer in self.layers:

            x = layer(x, encoder_output, target_mask, memory_mask)

        

        # 最终归一化和线性投影

        x = self.norm(x)

        logits = self.output_projection(x)

        

        return logits

```

## 完整Transformer模型

```python

class Transformer(nn.Module):

    """完整的Transformer模型（编码器-解码器架构）"""

    def __init__(self, src_vocab_size, tgt_vocab_size,

                 num_layers=6, d_model=512, num_heads=8,

                 d_ff=2048, max_len=512, dropout_rate=0.1):

        super().__init__()

        

        # 编码器

        self.encoder = TransformerEncoder(

            num_layers=num_layers,

            d_model=d_model,

            num_heads=num_heads,

            d_ff=d_ff,

            vocab_size=src_vocab_size,

            max_len=max_len,

            dropout_rate=dropout_rate

        )

        

        # 解码器

        self.decoder = TransformerDecoder(

            num_layers=num_layers,

            d_model=d_model,

            num_heads=num_heads,

            d_ff=d_ff,

            vocab_size=tgt_vocab_size,

            max_len=max_len,

            dropout_rate=dropout_rate

        )

    

    def encode(self, src_ids, src_mask=None):

        """编码源序列"""

        return self.encoder(src_ids, src_mask)

    

    def decode(self, tgt_ids, encoder_output, 

               tgt_mask=None, memory_mask=None):

        """解码目标序列"""

        return self.decoder(tgt_ids, encoder_output, tgt_mask, memory_mask)

    

    def forward(self, src_ids, tgt_ids, 

                src_mask=None, tgt_mask=None, memory_mask=None):

        """前向传播"""

        # 编码

        encoder_output = self.encode(src_ids, src_mask)

        <"kfc.a8k1.org.cn"><"doro.a8k1.org.cn"><"cctc.a8k1.org.cn">

        # 解码

        decoder_output = self.decode(tgt_ids, encoder_output, 

                                   tgt_mask, memory_mask)

        

        return decoder_output

```

## 训练与推理框架

### 训练循环实现

```python

class TransformerTrainer:

    """Transformer训练器"""

    def __init__(self, model, optimizer, criterion, device):

        self.model = model

        self.optimizer = optimizer

        self.criterion = criterion

        self.device = device

        self.model.to(device)

    

    def train_step(self, src_batch, tgt_batch, src_mask, tgt_mask):

        """单个训练步骤"""

        self.model.train()

        

        # 准备数据

        src_batch = src_batch.to(self.device)

        tgt_batch = tgt_batch.to(self.device)

        

        # 前向传播（使用教师强制）

        decoder_input = tgt_batch[:, :-1]

        decoder_target = tgt_batch[:, 1:]

        

        logits = self.model(src_batch, decoder_input, src_mask, tgt_mask)

        

        # 计算损失

        loss = self.criterion(

            logits.view(-1, logits.size(-1)),

            decoder_target.contiguous().view(-1)

        )

        

        # 反向传播

        self.optimizer.zero_grad()

        loss.backward()

        

        # 梯度裁剪

        torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)

        

        # 参数更新

        self.optimizer.step()

        

        return loss.item()

    

    def evaluate(self, dataloader):

        """模型评估"""

        self.model.eval()

        total_loss = 0

        total_tokens = 0

        

        with torch.no_grad():

            for src_batch, tgt_batch, src_mask, tgt_mask in dataloader:

                src_batch = src_batch.to(self.device)

                tgt_batch = tgt_batch.to(self.device)

                

                decoder_input = tgt_batch[:, :-1]

                decoder_target = tgt_batch[:, 1:]

                

                logits = self.model(src_batch, decoder_input, src_mask, tgt_mask)

                

                loss = self.criterion(

                    logits.view(-1, logits.size(-1)),

                    decoder_target.contiguous().view(-1)

                )

                

                batch_tokens = decoder_target.ne(self.criterion.ignore_index).sum().item()

                total_loss += loss.item() * batch_tokens

                total_tokens += batch_tokens

        

        return total_loss / total_tokens if total_tokens > 0 else float('inf')

```

### 推理生成策略

```python

class GreedyDecoder:

    """贪心解码器"""

    def __init__(self, model, max_len=100, start_token=1, end_token=2):

        self.model = model

        self.max_len = max_len

        self.start_token = start_token

        self.end_token = end_token

    

    def decode(self, src_sequence, src_mask=None):

        """贪心搜索解码"""

        batch_size = src_sequence.size(0)

        

        # 编码输入序列

        encoder_output = self.model.encode(src_sequence, src_mask)

        

        # 初始化目标序列（以起始符开始）

        decoder_input = torch.full((batch_size, 1), 

                                 self.start_token,

                                 dtype=torch.long,

                                 device=src_sequence.device)

        

        # 逐个生成标记

        for step in range(self.max_len):

            # 生成注意力掩码（防止看到未来信息）

            target_mask = causal_attention_mask(decoder_input.size(1))

            target_mask = target_mask.to(src_sequence.device)

            

            # 解码

            logits = self.model.decode(decoder_input, encoder_output, target_mask)

            next_token = logits[:, -1, :].argmax(dim=-1, keepdim=True)

            

            # 添加到序列

            decoder_input = torch.cat([decoder_input, next_token], dim=1)

            

            # 检查是否所有序列都已生成结束符

            if (next_token == self.end_token).all():

                break

        

        return decoder_input

class BeamSearchDecoder:

    """束搜索解码器"""

    def __init__(self, model, beam_size=5, max_len=100, 

                 start_token=1, end_token=2, length_penalty=0.6):

        self.model = model

        self.beam_size = beam_size

        self.max_len = max_len

        self.start_token = start_token

        self.end_token = end_token

        self.length_penalty = length_penalty

    

    def decode(self, src_sequence, src_mask=None):

        """束搜索解码"""

        batch_size = src_sequence.size(0)

        device = src_sequence.device

        

        # 编码输入序列

        encoder_output = self.model.encode(src_sequence, src_mask)

        

        # 扩展编码器输出用于束搜索

        encoder_output = encoder_output.repeat_interleave(self.beam_size, dim=0)

        

        # 初始化束

        beams = torch.full((batch_size * self.beam_size, 1),

                          self.start_token,

                          dtype=torch.long,

                          device=device)

        

        beam_scores = torch.zeros(batch_size, self.beam_size, device=device)

        beam_scores[:, 1:] = -float('inf')

        beam_scores = beam_scores.view(-1)

        

        finished_beams = [[] for _ in range(batch_size)]

        

        for step in range(self.max_len):

            # 生成注意力掩码

            target_mask = causal_attention_mask(beams.size(1))

            target_mask = target_mask.to(device)

            

            # 解码

            logits = self.model.decode(beams, encoder_output, target_mask)

            next_token_logits = logits[:, -1, :]

            

            # 计算分数

            vocab_size = next_token_logits.size(-1)

            scores = F.log_softmax(next_token_logits, dim=-1)

            scores = scores + beam_scores.unsqueeze(1)

            

            # 重新组织分数以便选择top-k

            if step == 0:

                scores = scores.view(batch_size, vocab_size)

            else:

                scores = scores.view(batch_size, self.beam_size * vocab_size)

            

            # 选择top-k

            topk_scores, topk_indices = scores.topk(self.beam_size, dim=1)

            

            # 计算束索引和词索引

            beam_indices = topk_indices // vocab_size

            token_indices = topk_indices % vocab_size

            

            # 更新束

            new_beams = []

            for i in range(batch_size):

                batch_beams = []

                for j in range(self.beam_size):

                    beam_idx = beam_indices[i, j].item()

                    token_idx = token_indices[i, j].item()

                    

                    # 获取之前的序列

                    if step == 0:

                        prev_beam = beams[i]

                    else:

                        prev_beam = beams[i * self.beam_size + beam_idx]

                    

                    # 添加新标记

                    new_beam = torch.cat([prev_beam, torch.tensor([token_idx], device=device)])

                    

                    # 检查是否结束

                    if token_idx == self.end_token:

                        # 计算最终分数（带长度惩罚）

                        length = len(new_beam)

                        score = topk_scores[i, j].item() / (length ** self.length_penalty)

                        finished_beams[i].append((score, new_beam))

                    else:

                        batch_beams.append(new_beam)

                

                new_beams.extend(batch_beams)

            

            if not new_beams:

                break

            

            beams = torch.stack(new_beams)

            beam_scores = topk_scores.view(-1)

        

        # 选择每个批次的最佳序列

        outputs = []

        for i in range(batch_size):

            if finished_beams[i]:

                best_beam = max(finished_beams[i], key=lambda x: x[0])[1]

            else:

                best_beam = beams[i * self.beam_size]

            outputs.append(best_beam)

        

        return torch.stack(outputs)

```

## 模型优化技术

```python

class TransformerOptimizer:

    """Transformer专用优化器（含学习率预热）"""

    def __init__(self, optimizer, d_model, warmup_steps=4000):

        self.optimizer = optimizer

        self.d_model = d_model

        self.warmup_steps = warmup_steps

        self.current_step = 0

    

    def step(self):

        """执行优化步骤"""

        self.current_step += 1

        lr = self.get_learning_rate()

        

        for param_group in self.optimizer.param_groups:

            param_group['lr'] = lr

        

        self.optimizer.step()

    

    def get_learning_rate(self):

        """计算当前学习率"""

        step = self.current_step

        warmup = self.warmup_steps

        

        # 学习率预热公式

        return self.d_model ** (-0.5) * min(step ** (-0.5), step * warmup ** (-1.5))

    

    def zero_grad(self):

        """清空梯度"""

        self.optimizer.zero_grad()

```

## 结语

Transformer架构通过自注意力机制彻底改变了序列建模的方法论，其并行计算特性和长距离依赖建模能力为自然语言处理带来了革命性进步。从机器翻译到大规模预训练模型，Transformer已经成为现代自然语言处理的基石架构。

理解Transformer的架构细节和实现原理，对于深入掌握当前自然语言处理技术发展趋势具有重要意义。随着研究的深入，Transformer架构的变体和改进不断涌现，但核心的自注意力机制和编码器-解码器框架仍然是这一领域的重要基础。