机器翻译：Sequence to Sequence Modeling with nn.Transformer
作者：禅与计算机程序设计艺术
1.简介
在近几年里，基于深度学习的神经网络在自然语言处理（NLP）领域逐渐成为主流，其主要应用领域之一就是机器翻译。其核心思想就是用计算机将一段文本从一种语言翻译成另一种语言，例如英文到中文或者中文到英文。目前最常用的机器翻译模型是 seq2seq 模型，即序列到序列模型。
Seq2seq 模型的基本思路是将输入序列通过编码器进行编码并得到固定长度的上下文表示，然后把此上下文表示作为解码器的初始状态，将目标序列通过解码器生成翻译后的文本。
本文将使用 pytorch 的 nn.Transformer 和 torchtext 来实现一个 seq2seq 模型，用来进行中文到英文的机器翻译任务。
2.基本概念、术语、名词解释
2.1 什么是 NLP？
自然语言处理（NLP）主要研究领域是使计算机能够像人类一样理解和交流自然语言。
该领域涵盖词法解析、语法结构分析以及语义理解等多个方面的技术。
2.2 什么是机器翻译？
机器翻译（Machine Translation）是一种自动化的方式用于将一段文本从一种语言转换为另一种语言。在常规应用中输入和输出均为 plain text 文本形式；然而，在特殊需求下还可以实现对图像、视频或其他多媒体文件内容的文字转译功能。
此外，在某种程度上来说，机器翻译系统的行为与其说是自动化的工具不如说是模仿了人类译者的某些工作模式：它们通过分析源语言的内容并将其重新组织以适应目标语言的文化背景与表达习惯。
这种行为帮助阅读者或听众更直观地理解原文作者意图的同时确保信息传递准确无误。
2.3 Sequence-to-sequence model
Seq2seq模型是近年来的核心议题之一，并且已成为机器学习领域中备受关注的主流模型。从字面意义来看, 其核心机制在于将输入序列转换为输出序列的过程被形象地命名为 Seq2seq 模型.该模型采用编码器-解码器架构设计, 在这一框架下, 编码器的作用是将输入信息转化为固定的表示形式.解码器则基于此表示逐步生成目标序列.此外, 该模型的主要特点还包括支持较长文本处理以及端到端训练机制.值得注意的是, 该方法通常会涉及到复杂的数学推导过程.
- Encoder: 通过编码器模块将输入序列转换为固定长度的上下文表示。
- Decoder: 借助解码器模块基于当前时步的输入token及其历史信息预测下一个token。
- Attention mechanism: 旨在处理输出序列生成过程中的复杂依赖关系。
- Beam search or greedy decoding: 常用于优化机器翻译中的解码过程以选择最优候选词。
2.4 transformer
Transformer 是 Google 在 2017 年提出的 NLP 模型之一,其核心技术在于完全依赖自注意力机制（self-attention）。与 RNN 和 CNN 相比,Transformer 更擅长处理长距离依赖关系,因此特别适合用于短文本翻译任务,展现出显著的效果。
3.核心算法原理和具体操作步骤
3.1 数据集准备
3.1.1 定义数据集和数据预处理流程
第一步命名并获取所需的数据集（如 chinese-english-v1 或 wmt14-en-de），其中每个文件记录了一个对话及其翻译。接着检查并确认目标目录中包含所有相关文档（如 cc-en.txt）。随后将所有相关文档整合到同一目录中以便后续处理工作开展。接下来利用 Python 的内置库 re 去掉所有标点符号空白字符等无关符号以减少对原始文本的依赖性从而防止出现错误的训练结果。最后对整合后的文本进行分词操作并选择合适的中文分词工具（如 jieba pkuseg 等）完成前期预处理工作
import os
import re
from nltk.tokenize import word_tokenize
DATASET = 'chinese-english-v1'
SRC_LANG = 'zh'
TRG_LANG = 'en'
data_path = '/home/user/' + DATASET + '/' + SRC_LANG + '-' + TRG_LANG
for filename in os.listdir(data_path):
filepath = os.path.join(data_path, filename)
file = open(filepath, mode='r', encoding='utf-8')
text = file.read()
clean_text = re.sub('[^a-zA-Z\u4e00-\u9fff]+','', text).lower()
tokens = word_tokenize(clean_text)
print(tokens[:10])
file.close()
代码解读此处采用jieba分词器对文本进行分词，获得一串token。
3.1.2 构建数据迭代器
有两类数据迭代器构建方式：一类是手动搭建迭代器的方式。其中一种方法涉及逐行读取文本文件中的每一句话。这种情况下会按源语言和目标语言分门别类地整理成对应的字典结构。随后将其编码为数值形式并将其放入列表中。最后利用collate函数将这些列表打包成一个批次的数据集送入网络训练
def collate_fn(batch):
src_sentences = [item['src'] for item in batch]
trg_sentences = [item['trg'] for item in batch]
src_tensor = tokenizer.batch_encode_plus(
src_sentences,
max_length=MAX_LEN,
padding="longest",
truncation=True,
return_tensors="pt"
)
trg_tensor = tokenizer.batch_encode_plus(
trg_sentences,
max_length=MAX_LEN,
padding="longest",
truncation=True,
return_tensors="pt"
)
labels_tensor = trg_tensor["input_ids"][:, :-1].contiguous().clone()
labels_tensor[labels_tensor == tokenizer.pad_token_id] = -100
return {
"src": src_tensor["input_ids"],
"src_mask": src_tensor["attention_mask"],
"trg": trg_tensor["input_ids"],
"trg_mask": trg_tensor["attention_mask"],
"labels": labels_tensor
}
代码解读DataLoader 在每次模型调用时会自动化地批量读取所需的数据，并无需自行搭建数据迭代器系统。直接访问 DataLoader 对象即可完成操作。
train_dataset = datasets.TranslationDataset(
data_dir='/home/user/' + DATASET + '/' + SRC_LANG + '-' + TRG_LANG,
tokenizer=tokenizer,
exts=('.zh', '.en'),
fields=(SRC_LANG, TRG_LANG),
)
train_loader = torch.utils.data.DataLoader(
dataset=train_dataset,
batch_size=BATCH_SIZE,
shuffle=False,
collate_fn=collate_fn,
drop_last=True
)
代码解读此处我们通过TranslationDataset函数导入数据集并调用传入指定分隔符为.zh与.en的文件路径分别对应源语言与目标语言文件。随后我们设置好所需字段名以完成数据对齐工作
3.2 模型搭建
考虑到本文采用了 transformer 技术，并通过这种方法实现了一个高效的数据处理框架。该框架通过将输入序列映射到高层次抽象表示而提升了性能，并且能够有效处理长距离依赖关系。
3.2.1 编码器
3.2.1.1 Embedding layer
首先，在输入序列上应用 embedding 技术，通过将每个单词转换为固定长度的向量来表示。其权重矩阵的大小通常等于词表大小与嵌入维度的乘积。
class Embeddings(nn.Module):
def __init__(self, d_model, vocab):
super().__init__()
self.embedding = nn.Embedding(len(vocab), d_model)
self.dropout = nn.Dropout(p=0.1)
def forward(self, x):
emb = self.embedding(x)
emb = self.dropout(emb)
return emb
代码解读3.2.1.2 Positional Encoding
接着，在对词向量进行位置编码处理后，并非简单的相加操作而是采用更为复杂的线性变换方式以确保后续的自注意力机制能够有效地捕捉到各个时间步之间的相对关系
通过正弦和余弦函数构建的位置编码方案能够有效地表示各个时间步之间的相对信息
设 pos 表示不同序列的位置索引，并令 d_model 是嵌入维度
其中 i 表示序列中的第 i 个元素的位置索引
则 position_encoding函数在输入 pos 和 i 后返回对应的数值
class PositionalEncoding(nn.Module):
def __init__(self, d_model, dropout=0.1, 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)
pe = pe.unsqueeze(0).transpose(0, 1)
self.register_buffer("pe", pe)
def forward(self, x):
x = x + self.pe[: x.size(0), :]
return self.dropout(x)
代码解读3.2.1.3 Self-Attention Layer
然后我们搭建自注意力层 并通过ATTENDING其他位置的词对输入序列进行加权平均 这一步相当于识别不同位置间的关联关系 并利用这种关联关系来提升模型的能力 请注意这里的层级越多模型的复杂度也会随之提升
class MultiHeadedSelfAttention(nn.Module):
def __init__(self, h, d_model, dropout=0.1):
super().__init__()
assert d_model % h == 0
self.d_k = d_model // h
self.h = h
self.linear_q = nn.Linear(d_model, d_model)
self.linear_k = nn.Linear(d_model, d_model)
self.linear_v = nn.Linear(d_model, d_model)
self.attn = None
self.dropout = nn.Dropout(p=dropout)
def forward(self, q, k, v, mask=None):
bs = q.size(0)
if mask is not None:
mask = mask.unsqueeze(1)
n_heads = self.h
q = self.linear_q(q).view(bs, -1, self.h, self.d_k).permute(0, 2, 1, 3)
k = self.linear_k(k).view(bs, -1, self.h, self.d_k).permute(0, 2, 3, 1)
v = self.linear_v(v).view(bs, -1, self.h, self.d_k).permute(0, 2, 1, 3)
scores = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.d_k)
if mask is not None:
scores = scores.masked_fill(mask==0, -1e9)
attn = F.softmax(scores, dim=-1)
attn = self.dropout(attn)
output = torch.matmul(attn, v).transpose(1, 2).contiguous()\
.view(bs, -1, self.h * self.d_k)
output = self.w_o(output)
return output
代码解读3.2.1.4 Feed Forward Network
最后一步中，我们建立了前馈神经网络模型，并对编码后的输入序列依次应用了一系列非线性变换操作（如ReLU、sigmoid、tanh），最终获得输出结果。
class PositionwiseFeedForward(nn.Module):
def __init__(self, d_model, d_ff, dropout=0.1):
super().__init__()
self.w_1 = nn.Linear(d_model, d_ff)
self.w_2 = nn.Linear(d_ff, d_model)
self.dropout = nn.Dropout(dropout)
def forward(self, x):
return self.w_2(self.dropout(F.relu(self.w_1(x))))
代码解读3.2.1.5 Encoder Block
该 encoder block 可以由四个子层组成：嵌入器（Embedder）、位置编码器（Positional Encoder）、多头自注意力机制（Multi-Head Self-Attention）和前馈网络（Feed Forward Network）。其输入为一个序列数据；其输出与输入相同。
class EncoderBlock(nn.Module):
def __init__(self, hidden, head, ff_dim, dropout=0.1):
super().__init__()
self.mhsa = MultiHeadedSelfAttention(hidden, head, dropout=dropout)
self.pwf = PositionwiseFeedForward(head * hidden, ff_dim, dropout=dropout)
self.lnorm = nn.LayerNorm(head * hidden)
self.dropout = nn.Dropout(dropout)
def forward(self, x, mask=None):
x = self.mhsa(x, x, x, mask=mask)
x = x + self.dropout(self.pwf(self.lnorm(x)))
return x
代码解读3.2.1.6 Encoder
目前,我们能够构建完整的编码器系统。为此,我们设计了一种新的架构方案,该方案通过将复杂的编码过程分解为多个独立模块来进行高效管理。具体而言,首先我们需要对输入序列进行分割,将其分解为若干个独立的blocks,然后对每个block中的数据进行系统性处理,通过这种方式,在每个block处
理后传递给下一个block作为输入数据,直到所有blocks都完成处理工作为止。最后我们将所有blocks的输出结果整合在一起形成最终输出数据体
class Encoder(nn.Module):
def __init__(self, input_dim, hidden, head, ff_dim, num_blocks, dropout=0.1):
super().__init__()
self.block_list = nn.ModuleList([
EncoderBlock(hidden, head, ff_dim, dropout=dropout)
for _ in range(num_blocks)])
self.embedding = nn.Embedding(input_dim, hidden)
self.pos_enc = PositionalEncoding(hidden, dropout=dropout)
self.dropout = nn.Dropout(dropout)
def forward(self, src, mask=None):
x = self.embedding(src)
x = self.dropout(x) + self.pos_enc(x)
for block in self.block_list:
x = block(x, mask=mask)
return x
代码解读3.2.2 解码器
解码器的核心功能是将经过处理后的上下文信息对应还原到输出序列中。其主要区别在于，解码器能够基于输入的 token 推测输出序列中的下一个 token。
3.2.2.1 Embedding layer
首先，在处理输入序列时完成嵌入过程。具体而言，在自然语言处理中，
我们将每个单词转换为预定维度的空间中的向量表示。
其权重矩阵的规模通常等于词表数量与嵌入维度的乘积。
class Embeddings(nn.Module):
def __init__(self, d_model, vocab):
super().__init__()
self.embedding = nn.Embedding(len(vocab), d_model)
self.dropout = nn.Dropout(p=0.1)
def forward(self, x):
emb = self.embedding(x)
emb = self.dropout(emb)
return emb
代码解读3.2.2.2 Positional Encoding
随后，在词向量之后添加位置编码PE_{(pos,i)}（其中i为序列中的第i个元素），这样在后续的自注意力机制中就能够关注不同序列元素之间的相对位置信息）。具体来说，在构建这种位置信息时可以通过正弦函数和余弦函数来生成频率不同的正弦波和余弦波序列（即PE_{(pos,2i)}=\sin(\frac{pos}{10000^{\frac{2i}{d_model}}})和PE_{(pos,2i+1)}=\cos(\frac{pos}{10000^{\frac{2i}{d_model}}})），其中每一个PE_{(pos,i)}都代表了一个特定频率的位置编码
class PositionalEncoding(nn.Module):
def __init__(self, d_model, dropout=0.1, 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)
pe = pe.unsqueeze(0).transpose(0, 1)
self.register_buffer("pe", pe)
def forward(self, x):
x = x + self.pe[: x.size(0), :]
return self.dropout(x)
代码解读3.2.2.3 Masked Multi-Headed Self-Attention Layer
接着，在模型架构中我们引入了一个带有遮蔽机制的多头自注意力层（Masked Multi-Head Self-Attention Layer），其主要区别是该层引入了一个mask参数来遮蔽超出输出序列位置的信息；这一参数决定了哪些位置的信息会被遮蔽以避免对未来数据的泄露；由于我们无法预知这些未知位置的具体数值信息因此也无法对其进行预测或推断
class MultiHeadedSelfAttention(nn.Module):
def __init__(self, h, d_model, dropout=0.1):
super().__init__()
assert d_model % h == 0
self.d_k = d_model // h
self.h = h
self.linear_q = nn.Linear(d_model, d_model)
self.linear_k = nn.Linear(d_model, d_model)
self.linear_v = nn.Linear(d_model, d_model)
self.attn = None
self.dropout = nn.Dropout(p=dropout)
def forward(self, q, k, v, mask=None):
bs = q.size(0)
if mask is not None:
mask = mask.unsqueeze(1)
n_heads = self.h
q = self.linear_q(q).view(bs, -1, self.h, self.d_k).permute(0, 2, 1, 3)
k = self.linear_k(k).view(bs, -1, self.h, self.d_k).permute(0, 2, 3, 1)
v = self.linear_v(v).view(bs, -1, self.h, self.d_k).permute(0, 2, 1, 3)
scores = torch.matmul(q, k.transpose(-2, -1)) / math.sqrt(self.d_k)
if mask is not None:
scores = scores.masked_fill(mask==0, -1e9)
attn = F.softmax(scores, dim=-1)
attn = self.dropout(attn)
output = torch.matmul(attn, v).transpose(1, 2).contiguous()\
.view(bs, -1, self.h * self.d_k)
output = self.w_o(output)
return output
代码解读3.2.2.4 Decoder Block
最后，我们搭建了一个 decoder 块，并将其设计为包含 embedder、positional encoder、masked multi-headed self-attention 以及 feed forward network 四个组件。其输入由上一个 token 和 encoder 输出构成，并通过一系列计算流程生成当前时间步的预测值。
class DecoderBlock(nn.Module):
def __init__(self, hidden, head, ff_dim, dropout=0.1):
super().__init__()
self.mhsa = MultiHeadedSelfAttention(hidden, head, dropout=dropout)
self.pwf = PositionwiseFeedForward(head * hidden, ff_dim, dropout=dropout)
self.lnorm = nn.LayerNorm(head * hidden)
self.dropout = nn.Dropout(dropout)
self.embedding = nn.Embedding(hidden, hidden)
self.pos_enc = PositionalEncoding(hidden, dropout=dropout)
def forward(self, dec_inputs, enc_outputs, mask=None):
tgt = self.embedding(dec_inputs)
tgt += self.pos_enc(tgt)
tgt = self.dropout(tgt)
tgt = self.mhsa(tgt, enc_outputs, enc_outputs, mask=mask)
out = tgt + self.dropout(self.pwf(self.lnorm(tgt)))
return out
代码解读3.2.2.5 Decoder
当前阶段，在构建解码器的过程中
class Decoder(nn.Module):
def __init__(self, output_dim, hidden, head, ff_dim, num_blocks, dropout=0.1):
super().__init__()
self.block_list = nn.ModuleList([
DecoderBlock(hidden, head, ff_dim, dropout=dropout)
for _ in range(num_blocks)])
self.embedding = nn.Embedding(output_dim, hidden)
self.pos_enc = PositionalEncoding(hidden, dropout=dropout)
self.dropout = nn.Dropout(dropout)
def forward(self, tgt, memory, src_mask=None, tgt_mask=None):
x = self.embedding(tgt)
x += self.pos_enc(x)
x = self.dropout(x)
for block in self.block_list:
x = block(x, memory, mask=tgt_mask & src_mask)
return x
代码解读3.2.3 Seq2Seq Model
在本节所述的系统设计中包含一个完整的序列到序列模型。该模型由编码器模块和解码器模块组成，并采用贪婪搜索以及 beam search等搜索机制实现对目标的预测。
class Seq2SeqModel(nn.Module):
def __init__(self, encoder, decoder, device):
super().__init__()
self.encoder = encoder
self.decoder = decoder
self.device = device
def make_masks(self, src, tgt):
src_mask = (src!= 1).unsqueeze(-2).bool()
tgt_mask = (tgt!= 1).unsqueeze(-2).bool() & subsequent_mask(tgt.shape[-1]).type_as(src_mask.data).to(self.device)
return src_mask, tgt_mask
def forward(self, src, tgt):
src_mask, tgt_mask = self.make_masks(src, tgt)
memory = self.encoder(src, src_mask)
outputs = self.decoder(tgt[:-1], memory, src_mask, tgt_mask)
outputs = outputs.contiguous().view(-1, outputs.shape[-1])
return outputs
代码解读3.3 模型训练
模型训练的步骤如下所示：
- 初始化模型实例，并传递编码器、解码器以及设备对象作为必要参数。
- 配置优化器与损失函数（criterion）。其中损失函数用于评估预测结果与实际目标之间的差异程度。
- 遍历训练数据集中的每一个样本集合（batch），将输入样本及其对应的标签输入到模型中进行处理。随后计算当前批次的损失值，并对损失函数进行回传以计算梯度。最终根据计算得到的梯度信息更新优化器中的权重参数。
- 在独立的验证集中评估模型的性能表现。若验证结果较之前有所提升，则记录当前最优的权重参数设置。
MAX_LEN = 100
BATCH_SIZE = 64
EPOCHS = 100
DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
MODEL_PATH = './models/' + DATASET + '_' + str(EPOCHS) + '_epochs_' + str(BATCH_SIZE) + '_batch_size.pth'
encoder = Encoder(input_dim=len(zh_field.vocab), hidden=EMBEDDING_DIM, head=NUM_HEADS, ff_dim=FF_DIM, num_blocks=NUM_BLOCKS).to(DEVICE)
decoder = Decoder(output_dim=len(en_field.vocab), hidden=EMBEDDING_DIM, head=NUM_HEADS, ff_dim=FF_DIM, num_blocks=NUM_BLOCKS).to(DEVICE)
seq2seq = Seq2SeqModel(encoder, decoder, device=DEVICE).to(DEVICE)
optimizer = optim.AdamW(seq2seq.parameters())
criterion = nn.CrossEntropyLoss(ignore_index=-100)
best_valid_loss = float('inf')
for epoch in range(EPOCHS):
start_time = time.time()
train_loss = train(seq2seq, train_iterator, optimizer, criterion)
valid_loss = evaluate(seq2seq, val_iterator, criterion)
end_time = time.time()
epoch_mins, epoch_secs = epoch_time(start_time, end_time)
if valid_loss < best_valid_loss:
best_valid_loss = valid_loss
torch.save(seq2seq.state_dict(), MODEL_PATH)
print(f'Epoch: {epoch+1:02} | Time: {epoch_mins}m {epoch_secs}s')
print(f' Train Loss: {train_loss:.3f} | Train PPL: {math.exp(train_loss):7.3f}')
print(f' Val. Loss: {valid_loss:.3f} | Val. PPL: {math.exp(valid_loss):7.3f}')
代码解读3.4 模型推断
测试模型的步骤如下所示：
- 生成相应的训练样本集合以及构建相应的迭代器用于遍历样本。
- 初始化机器学习模型框架，并导入最优参数设置以优化性能。
- 基于训练好的测试样本集合运行推理运算以获取预测结果，并计算预测误差值后输出评估指标数值。
test_iterator = DataIterator(test_dataset, batch_size=1, train=False, sort=False)
seq2seq.load_state_dict(torch.load(MODEL_PATH))
test_loss = evaluate(seq2seq, test_iterator, criterion)
print(f'| Test Loss: {test_loss:.3f} | Test PPL: {math.exp(test_loss):7.3f} |')
test_loss_history.append(test_loss)
代码解读4.总结
本文阐述了机器翻译的基本原理及相关概念，并具体包括序列到序列模型这一核心内容。文章系统阐述了数据预处理步骤、模型设计方案以及超参数选取策略等关键环节，并涵盖模型搭建过程中的训练方法与推理机制。在完成整个系统的构建后，在实验部分展示了其应用效果并进行了性能评估指标的设计与应用。