【自然语言处理（NLP）】深度学习架构：Transformer 原理及代码实现

个人主页：道友老李
欢迎加入社区：道友老李的学习社区

介绍

**自然语言处理（Natural Language Processing，NLP）**是计算机科学领域与人工智能领域中的一个重要方向。它研究的是人类（自然）语言与计算机之间的交互。NLP的目标是让计算机能够理解、解析、生成人类语言，并且能够以有意义的方式回应和操作这些信息。

NLP的任务可以分为多个层次，包括但不限于：

1. 词法分析：将文本分解成单词或标记（token），并识别它们的词性（如名词、动词等）。
2. 句法分析：分析句子结构，理解句子中词语的关系，比如主语、谓语、宾语等。
3. 语义分析：试图理解句子的实际含义，超越字面意义，捕捉隐含的信息。
4. 语用分析：考虑上下文和对话背景，理解话语在特定情境下的使用目的。
5. 情感分析：检测文本中表达的情感倾向，例如正面、负面或中立。
6. 机器翻译：将一种自然语言转换为另一种自然语言。
7. 问答系统：构建可以回答用户问题的系统。
8. 文本摘要：从大量文本中提取关键信息，生成简短的摘要。
9. 命名实体识别（NER）：识别文本中提到的特定实体，如人名、地名、组织名等。
10. 语音识别：将人类的语音转换为计算机可读的文字格式。

NLP技术的发展依赖于算法的进步、计算能力的提升以及大规模标注数据集的可用性。近年来，深度学习方法，特别是基于神经网络的语言模型，如BERT、GPT系列等，在许多NLP任务上取得了显著的成功。随着技术的进步，NLP正在被应用到越来越多的领域，包括客户服务、智能搜索、内容推荐、医疗健康等。

Transformer

Transformer是一种深度学习架构，最初在2017年的论文《Attention Is All You Need》中被提出，它在自然语言处理（NLP）等领域取得了巨大的成功，并引发了后续一系列相关研究和技术的发展。

核心组件

• 多头注意力机制（Multi-Head Attention）
  • 原理：将输入的向量表示通过多个头（head）的注意力机制，并行地计算不同位置之间的依赖关系，从而捕捉到更丰富的语义信息。每个头都可以关注输入序列的不同部分，然后将这些头的结果进行拼接和线性变换，得到最终的输出。
  • 公式：$MultiHead(Q,K,V)=Concat(hea{d}_1,\cdots ,hea{d}_h)W^O$，其中$hea{d}_i=Attention(Q{W}_i^Q,K{W}_i^K,V{W}_i^V)$，$Attention(Q,K,V)=softmax(\frac{Q{K}^T}{\sqrt{d_k}})V$。
  • 作用：能够自适应地聚焦于输入序列中的不同位置，有效地捕捉长序列中的依赖关系，相比传统的循环神经网络（RNN）和卷积神经网络（CNN），在处理长序列数据时具有更好的性能和可并行性。
• 编码器和解码器
  • 编码器：由多个堆叠的编码器层组成，每个编码器层包含多头注意力机制和前馈神经网络（Feed Forward Network，FFN），还使用了残差连接和层归一化（Layer Normalization）技术。其作用是将输入序列编码成一个固定长度的向量表示，提取输入序列中的特征。
  • 解码器：同样由多个解码器层堆叠而成，与编码器类似，但在多头注意力机制部分有所不同，解码器还包含一个掩码多头注意力（Masked Multi-Head Attention）层，用于防止解码器在生成当前位置的输出时看到未来的信息。解码器的任务是根据编码器的输出和之前已经生成的输出序列，逐步生成目标序列。

架构图

编码器（Encoder）

• 嵌入层（Embedding Layer）：将输入的离散符号（如单词）转换为连续的向量表示，也就是词嵌入。
• 位置编码（Positional Encoding）：由于Transformer本身不具备捕捉序列顺序信息的能力，位置编码将位置信息加入到词嵌入中，以便模型能区分不同位置的元素。
• 多头注意力（Multi - Head Attention）：通过多个头并行计算注意力，捕捉输入序列不同位置之间的依赖关系和语义信息。
• 逐位前馈网络（Feed - Forward Network）：对多头注意力的输出进行进一步非线性变换，增强模型的表达能力。
• 加 & 规范化（Add & Normalization）：采用残差连接（Add）将输入直接加到注意力或前馈网络的输出上，防止梯度消失等问题；层归一化（Normalization）对数据进行归一化处理，加速训练收敛。

解码器（Decoder）

• 嵌入层和位置编码：与编码器作用类似，处理目标序列。
• 掩码多头注意力（Masked Multi - Head Attention）：在生成目标序列时，为了防止解码器提前看到未来的信息，使用掩码操作遮盖后续位置，保证生成过程符合自回归特性。
• 多头注意力：这里的多头注意力用于建立目标序列和编码器输出之间的联系，帮助解码器根据源序列信息生成目标序列。
• 逐位前馈网络和加 & 规范化：与编码器中的作用相同，进行非线性变换和归一化等操作。
• 全连接层（Fully - Connected Layer）：将解码器的输出映射到目标词汇表的维度，通过softmax函数得到每个词的生成概率，用于最终的输出预测。

Transformer架构通过编码器和解码器的多层结构，利用多头注意力机制高效地捕捉序列中的依赖关系，在自然语言处理等诸多领域有着广泛且出色的应用。

优点

• 并行计算能力：Transformer可以并行计算所有位置的输出，大大提高了训练和推理的速度，相比需要顺序处理每个时间步的RNN和其变体（如LSTM、GRU），能够更高效地利用现代硬件设备（如GPU、TPU）的并行计算能力。
• 长序列建模能力：通过自注意力机制，Transformer能够直接建模输入序列中任意两个位置之间的依赖关系，而不受序列长度的限制，能够很好地处理长序列数据，避免了RNN在处理长序列时可能出现的梯度消失或爆炸问题。
• 灵活的特征提取能力：多头注意力机制可以同时关注输入序列的不同方面，能够自动学习到文本中的各种语义和句法结构，提取更丰富、更抽象的特征，对各种自然语言任务具有很强的适应性。

应用

• 自然语言处理领域
  • 机器翻译：Transformer在机器翻译任务中取得了显著的成果，能够将一种语言准确地翻译成另一种语言，例如谷歌的GNMT（Google Neural Machine Translation）系统采用了Transformer架构，大大提高了翻译质量和效率。
  • 文本生成：可以用于生成各种类型的文本，如对话生成、故事生成、诗歌生成等，如OpenAI的GPT系列模型基于Transformer架构，能够生成连贯、有逻辑的自然语言文本。
  • 文本分类：对文本进行分类，如情感分类、新闻分类等，通过对文本的特征提取和表示学习，Transformer能够准确地判断文本所属的类别。
• 其他领域
  • 计算机视觉：一些研究将Transformer应用于图像识别、目标检测、图像生成等任务中，如Vision Transformer（ViT）将图像分块后，将其视为序列输入到Transformer中，取得了与传统CNN相当甚至更好的性能。
  • 语音识别：在语音识别任务中，Transformer也被用于对语音信号的特征进行建模和处理，提高语音识别的准确率。

代码实现

导包

import math
import pandas as pd
import torch
from torch import nn
import dltools

基于位置的前馈网络

class PositionWiseFFN(nn.Module):
    def __init__(self, ffn_num_input, ffn_num_hiddens, ffn_num_outputs, **kwargs):
        super().__init__(**kwargs)
        self.dense1 = nn.Linear(ffn_num_input, ffn_num_hiddens)
        self.relu = nn.ReLU()
        self.dense2 = nn.Linear(ffn_num_hiddens, ffn_num_outputs)
        
    def forward(self, X):
        return self.dense2(self.relu(self.dense1(X)))

根据经验, 传入全连接的数据一般要求是二维. 但序列数据一般是三维, ffn中就需要降维. 一般是把batch独立, 其他维度合并.这里因为序列的有效长度不相同，所以把batch_size和num_steps合并, 再做全连接。

残差连接后进行层规范化

class AddNorm(nn.Module):
    def __init__(self, normalized_shape, dropout, **kwargs):
        super().__init__(**kwargs)
        self.dropout = nn.Dropout(dropout)
        self.ln = nn.LayerNorm(normalized_shape)
        
    def forward(self, X, Y):
        return self.ln(self.dropout(Y) + X)

• batch normalization 和layer normalization的区别：
  • bn是每个通道样本间进行归一化, LN是每个样本通道间归一化, 就相当于对一句话做norm
  • 假如现在有b句话, 每句话的长度(即序列长度)len, 每个词有d个特征表示(embed_size), bn就是对所有句子所有词的某个特征做归一化.
  • LN就是对某一句话所有的词的所有特征做归一化.

示例：

ln = nn.LayerNorm(2)
bn = nn.BatchNorm1d(2)
X = torch.tensor([[1, 2], [2, 3]], dtype=torch.float32)
# 计算ln和bn的结果
print('LN:', ln(X), '\n BN:', bn(X))

LN: tensor([[-1.0000,  1.0000],
        [-1.0000,  1.0000]], grad_fn=<NativeLayerNormBackward0>) 
BN: tensor([[-1.0000, -1.0000],
        [ 1.0000,  1.0000]], grad_fn=<NativeBatchNormBackward0>)

一定要注意, 使用了残差结构, 输入数据的维度一定要相同！

编码器 Block

class EncoderBlock(nn.Module):
    def __init__(self, key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, dropout, use_bias=False, **kwargs):
        super().__init__(**kwargs)
        self.attention = dltools.MultiHeadAttention(key_size, query_size, value_size, num_hiddens, num_heads, dropout, use_bias)
        self.addnorm1 = AddNorm(norm_shape, dropout)
        self.ffn = PositionWiseFFN(ffn_num_input, ffn_num_hiddens, num_hiddens)
        self.addnorm2 = AddNorm(norm_shape, dropout)
        
    def forward(self, X, valid_lens):
        Y = self.addnorm1(X, self.attention(X, X, X, valid_lens))
        return self.addnorm2(Y, self.ffn(Y))

使用示例：

X = torch.ones((2, 100, 24))
valid_lens = torch.tensor([3, 2])
encoder_blk = EncoderBlock(24, 24, 24, 24, [100, 24], 24, 48, 8, 0.5)
encoder_blk.eval()
encoder_blk(X, valid_lens).shape

torch.Size([2, 100, 24])

编码器

class TransformerEncoder(dltools.Encoder):
    def __init__(self, vocab_size, key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, num_layers, dropout, use_bias=False, **kwargs):
        super().__init__(**kwargs)
        self.num_hiddens = num_hiddens
        self.embedding = nn.Embedding(vocab_size, num_hiddens)
        self.pos_encoding = dltools.PositionalEncoding(num_hiddens, dropout)
        self.blks = nn.Sequential()
        for i in range(num_layers):
            self.blks.add_module('block' + str(i), EncoderBlock(key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, dropout, use_bias))
            
    def forward(self, X, valid_lens, *args):
        # 对embedding之后的数据进行缩放, 有助于收敛
        X = self.pos_encoding(self.embedding(X) * math.sqrt(self.num_hiddens))
        self.attention_weights = [None] * len(self.blks)
        for i, blk in enumerate(self.blks):
            X = blk(X, valid_lens)
            self.attention_weights[i] = blk.attention.attention.attention_weights
        return X

使用示例：

encoder = TransformerEncoder(200, 24, 24, 24, 24, [100, 24], 24, 48, 8, 2, 0.3)
encoder.eval()
X = torch.ones((2, 100), dtype=torch.long)
encoder(X, valid_lens).shape

torch.Size([2, 100, 24])

解码器 Block

class DecoderBlock(nn.Module):
    def __init__(self, key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, dropout, i, **kwargs):
        super().__init__(**kwargs)
        self.i = i
        self.attention1 = dltools.MultiHeadAttention(key_size, query_size, value_size, num_hiddens, num_heads, dropout)
        self.addnorm1 = AddNorm(norm_shape, dropout)
        self.attention2 = dltools.MultiHeadAttention(key_size, query_size, value_size, num_hiddens, num_heads, dropout)
        self.addnorm2 = AddNorm(norm_shape, dropout)
        self.ffn = PositionWiseFFN(ffn_num_input, ffn_num_hiddens, num_hiddens)
        self.addnorm3 = AddNorm(norm_shape, dropout)
        
    def forward(self, X, state):
        enc_outputs, enc_valid_lens = state[0], state[1]
        if state[2][self.i] is None:
            key_values = X
        else:
            # 预测: 预测需要把前面时刻预测得到的信息和当前block的输出得到的信息拼到一起. 
            key_values = torch.cat((state[2][self.i], X), axis=1)
        state[2][self.i] = key_values
        # 在训练的时候需要对真实值进行遮蔽
        if self.training:
            # (batch_size, num_steps), 每一行是[1, 2, ..., num_steps]
            batch_size, num_steps, _ = X.shape
            dec_valid_lens = torch.arange(1, num_steps + 1, device=X.device).repeat(batch_size, 1)
        else:
            dec_valid_lens = None
         
        # 自注意力
        X2 = self.attention1(X, key_values, key_values, dec_valid_lens)
        Y = self.addnorm1(X, X2)
        Y2 = self.attention2(Y, enc_outputs, enc_outputs, enc_valid_lens)
        Z = self.addnorm2(Y, Y2)
        return self.addnorm3(Z, self.ffn(Z)), state

使用示例：

decoder_blk = DecoderBlock(24, 24, 24, 24, [100, 24], 24, 48, 8, 0.5, 0)
decoder_blk.eval()
X = torch.ones((2, 100, 24))
state = [encoder_blk(X, valid_lens), valid_lens, [None]]
result1, result2 = decoder_blk(X, state)
result1.shape

torch.Size([2, 100, 24])

解码器

class TransformerDecoder(dltools.AttentionDecoder):
    def __init__(self, vocab_size, key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, num_layers, dropout, **kwargs):
        super().__init__(**kwargs)
        self.num_hiddens = num_hiddens
        self.num_layers = num_layers
        self.embedding = nn.Embedding(vocab_size, num_hiddens)
        self.pos_embedding = dltools.PositionalEncoding(num_hiddens, dropout)
        self.blks = nn.Sequential()
        for i in range(num_layers):
            self.blks.add_module('block' + str(i), DecoderBlock(key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, dropout, i))
            
        self.dense = nn.Linear(num_hiddens, vocab_size)
        
    def init_state(self, enc_outputs, enc_valid_lens, *args):
        return [enc_outputs, enc_valid_lens, [None]*self.num_layers]

    def forward(self, X, state):
        X = self.pos_embedding(self.embedding(X) * math.sqrt(self.num_hiddens))
        self._attention_weights = [[None] * len(self.blks) for _ in range(2)]
        for i, blk in enumerate(self.blks):
            X, state = blk(X, state)
            self._attention_weights[0][i] = blk.attention1.attention.attention_weights
            self._attention_weights[1][i] = blk.attention2.attention.attention_weights
        return self.dense(X), state
    
    @property
    def attention_weights(self):
        return self._attention_weights

训练

num_hiddens, num_layers, dropout, batch_size, num_steps = 32, 2, 0.1, 64, 10
lr, num_epochs, device = 0.005, 200, dltools.try_gpu()
ffn_num_input, ffn_num_hiddens, num_heads = 32, 64, 4
key_size, query_size, value_size = 32, 32, 32
norm_shape = [32]

train_iter, src_vocab, tgt_vocab = dltools.load_data_nmt(batch_size, num_steps)

encoder = TransformerEncoder(len(src_vocab), key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, num_layers, dropout)
decoder = TransformerDecoder(len(tgt_vocab), key_size, query_size, value_size, num_hiddens, norm_shape, ffn_num_input, ffn_num_hiddens, num_heads, num_layers, dropout)
net = dltools.EncoderDecoder(encoder, decoder)
dltools.train_seq2seq(net, train_iter, lr, num_epochs, tgt_vocab, device)

预测

engs = ['go .', 'i lost .', 'he\'s calm .', 'i\'m home .']
fras = ['va !', 'j\'ai perdu .', 'il est calme .', 'je suis chez moi .']
for eng, fra in zip(engs, fras):
    translation = dltools.predict_seq2seq(net, eng, src_vocab, tgt_vocab, num_steps, device)
    print(f'{eng} => {translation}, bleu {dltools.bleu(translation[0], fra, k=2):.3f}')

go . => ('va !', []), bleu 1.000
i lost . => ("j'ai perdu .", []), bleu 1.000
he's calm . => ('il est calme .', []), bleu 1.000
i'm home . => ('je suis chez moi .', []), bleu 1.000