Seq2Seq 任务实战：基于 PyTorch 实现英法翻译模型（BLEU 值 0.35+）

Seq2Seq 任务实战：基于 PyTorch 实现英法翻译模型（BLEU 值 0.35+）

机器翻译是自然语言处理中最具挑战性的任务之一，而 Seq2Seq（Sequence-to-Sequence）模型则是解决这一问题的经典架构。本文将带你从零开始实现一个基于 PyTorch 的英法翻译模型，涵盖数据预处理、模型构建、训练与评估全流程，最终达到 BLEU 值 0.35+ 的翻译效果。

1. 环境准备与数据加载

首先确保已安装 PyTorch 1.8+ 和 torchtext 0.9+。我们将使用 torchtext 提供的 Multi30k 数据集，这是一个包含约 30,000 条英法平行句对的数据集。

import torch import torch.nn as nn import torch.optim as optim from torchtext.datasets import Multi30k from torchtext.data import Field, BucketIterator # 设置随机种子保证可复现性 SEED = 1234 torch.manual_seed(SEED) torch.backends.cudnn.deterministic = True # 定义字段处理器 SRC = Field(tokenize="spacy", tokenizer_language="en", init_token="<sos>", eos_token="<eos>", lower=True) TRG = Field(tokenize="spacy", tokenizer_language="fr", init_token="<sos>", eos_token="<eos>", lower=True) # 加载数据集 train_data, valid_data, test_data = Multi30k.splits(exts=(".en", ".fr"), fields=(SRC, TRG)) # 构建词汇表 SRC.build_vocab(train_data, min_freq=2) TRG.build_vocab(train_data, min_freq=2) # 创建数据迭代器 BATCH_SIZE = 128 device = torch.device("cuda" if torch.cuda.is_available() else "cpu") train_iterator, valid_iterator, test_iterator = BucketIterator.splits( (train_data, valid_data, test_data), batch_size=BATCH_SIZE, device=device )

关键参数说明：

• min_freq=2：过滤掉出现次数少于2次的单词
• BucketIterator：自动将相似长度的样本分到同一批次，减少填充数量
• init_token和eos_token：分别表示序列的开始和结束标记

2. 模型架构设计

我们将实现一个基于 LSTM 的 Encoder-Decoder 架构，并加入注意力机制来提升长序列的翻译效果。

2.1 Encoder 实现

编码器将源语言句子编码为上下文向量：

class Encoder(nn.Module): def __init__(self, input_dim, emb_dim, hid_dim, n_layers, dropout): super().__init__() self.embedding = nn.Embedding(input_dim, emb_dim) self.rnn = nn.LSTM(emb_dim, hid_dim, n_layers, dropout=dropout) self.dropout = nn.Dropout(dropout) def forward(self, src): # src shape: [src_len, batch_size] embedded = self.dropout(self.embedding(src)) # embedded shape: [src_len, batch_size, emb_dim] outputs, (hidden, cell) = self.rnn(embedded) # outputs shape: [src_len, batch_size, hid_dim * n_directions] # hidden/cell shape: [n_layers * n_directions, batch_size, hid_dim] return hidden, cell

2.2 Attention 机制

注意力机制帮助解码器在生成每个词时关注源句子中最相关的部分：

class Attention(nn.Module): def __init__(self, hid_dim): super().__init__() self.attn = nn.Linear(hid_dim * 2, hid_dim) self.v = nn.Linear(hid_dim, 1, bias=False) def forward(self, hidden, encoder_outputs): # hidden shape: [batch_size, hid_dim] # encoder_outputs shape: [src_len, batch_size, hid_dim] src_len = encoder_outputs.shape[0] hidden = hidden.unsqueeze(1).repeat(1, src_len, 1) encoder_outputs = encoder_outputs.permute(1, 0, 2) energy = torch.tanh(self.attn(torch.cat((hidden, encoder_outputs), dim=2))) attention = self.v(energy).squeeze(2) return torch.softmax(attention, dim=1)

2.3 Decoder 实现

解码器利用编码器的输出和注意力机制生成目标语言：

class Decoder(nn.Module): def __init__(self, output_dim, emb_dim, hid_dim, n_layers, dropout, attention): super().__init__() self.output_dim = output_dim self.attention = attention self.embedding = nn.Embedding(output_dim, emb_dim) self.rnn = nn.LSTM(emb_dim + hid_dim, hid_dim, n_layers, dropout=dropout) self.fc_out = nn.Linear(emb_dim + hid_dim * 2, output_dim) self.dropout = nn.Dropout(dropout) def forward(self, input, hidden, cell, encoder_outputs): input = input.unsqueeze(0) embedded = self.dropout(self.embedding(input)) a = self.attention(hidden[-1], encoder_outputs) a = a.unsqueeze(1) encoder_outputs = encoder_outputs.permute(1, 0, 2) weighted = torch.bmm(a, encoder_outputs) weighted = weighted.permute(1, 0, 2) rnn_input = torch.cat((embedded, weighted), dim=2) output, (hidden, cell) = self.rnn(rnn_input, (hidden, cell)) embedded = embedded.squeeze(0) output = output.squeeze(0) weighted = weighted.squeeze(0) prediction = self.fc_out(torch.cat((output, weighted, embedded), dim=1)) return prediction, hidden, cell

2.4 Seq2Seq 整合

将编码器和解码器组合成完整的 Seq2Seq 模型：

class Seq2Seq(nn.Module): def __init__(self, encoder, decoder, device): super().__init__() self.encoder = encoder self.decoder = decoder self.device = device def forward(self, src, trg, teacher_forcing_ratio=0.5): batch_size = trg.shape[1] trg_len = trg.shape[0] trg_vocab_size = self.decoder.output_dim outputs = torch.zeros(trg_len, batch_size, trg_vocab_size).to(self.device) encoder_outputs, hidden, cell = self.encoder(src) input = trg[0, :] for t in range(1, trg_len): output, hidden, cell = self.decoder(input, hidden, cell, encoder_outputs) outputs[t] = output teacher_force = random.random() < teacher_forcing_ratio top1 = output.argmax(1) input = trg[t] if teacher_force else top1 return outputs

3. 模型训练与优化

3.1 初始化模型与优化器

INPUT_DIM = len(SRC.vocab) OUTPUT_DIM = len(TRG.vocab) ENC_EMB_DIM = 256 DEC_EMB_DIM = 256 HID_DIM = 512 N_LAYERS = 2 ENC_DROPOUT = 0.5 DEC_DROPOUT = 0.5 attn = Attention(HID_DIM) enc = Encoder(INPUT_DIM, ENC_EMB_DIM, HID_DIM, N_LAYERS, ENC_DROPOUT) dec = Decoder(OUTPUT_DIM, DEC_EMB_DIM, HID_DIM, N_LAYERS, DEC_DROPOUT, attn) model = Seq2Seq(enc, dec, device).to(device) optimizer = optim.Adam(model.parameters()) criterion = nn.CrossEntropyLoss(ignore_index=TRG.vocab.stoi[TRG.pad_token])

3.2 训练循环实现

def train(model, iterator, optimizer, criterion, clip): model.train() epoch_loss = 0 for i, batch in enumerate(iterator): src = batch.src trg = batch.trg optimizer.zero_grad() output = model(src, trg) output_dim = output.shape[-1] output = output[1:].view(-1, output_dim) trg = trg[1:].view(-1) loss = criterion(output, trg) loss.backward() torch.nn.utils.clip_grad_norm_(model.parameters(), clip) optimizer.step() epoch_loss += loss.item() return epoch_loss / len(iterator)

3.3 评估函数实现

def evaluate(model, iterator, criterion): model.eval() epoch_loss = 0 with torch.no_grad(): for i, batch in enumerate(iterator): src = batch.src trg = batch.trg output = model(src, trg, 0) # 关闭teacher forcing output_dim = output.shape[-1] output = output[1:].view(-1, output_dim) trg = trg[1:].view(-1) loss = criterion(output, trg) epoch_loss += loss.item() return epoch_loss / len(iterator)

3.4 BLEU 分数计算

from torchtext.data.metrics import bleu_score def calculate_bleu(data, src_field, trg_field, model, device, max_len=50): trgs = [] pred_trgs = [] model.eval() with torch.no_grad(): for i, example in enumerate(data): src = vars(example)["src"] trg = vars(example)["trg"] pred_trg = translate_sentence(src, src_field, trg_field, model, device, max_len) pred_trgs.append(pred_trg) trgs.append([trg]) return bleu_score(pred_trgs, trgs)

4. 模型训练与结果分析

4.1 训练参数设置

N_EPOCHS = 10 CLIP = 1 best_valid_loss = float("inf") for epoch in range(N_EPOCHS): train_loss = train(model, train_iterator, optimizer, criterion, CLIP) valid_loss = evaluate(model, valid_iterator, criterion) if valid_loss < best_valid_loss: best_valid_loss = valid_loss torch.save(model.state_dict(), "seq2seq-model.pt") print(f"Epoch: {epoch+1:02} | Train Loss: {train_loss:.3f} | Val. Loss: {valid_loss:.3f}") # 加载最佳模型并在测试集上评估 model.load_state_dict(torch.load("seq2seq-model.pt")) test_loss = evaluate(model, test_iterator, criterion) print(f"Test Loss: {test_loss:.3f}") # 计算BLEU分数 bleu = calculate_bleu(test_data, SRC, TRG, model, device) print(f"BLEU Score: {bleu*100:.2f}")

4.2 性能优化技巧

1. 学习率调度：使用 ReduceLROnPlateau 动态调整学习率

   scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, "min", patience=2)

2. 标签平滑：缓解模型过度自信问题

   criterion = nn.CrossEntropyLoss(ignore_index=TRG.vocab.stoi[TRG.pad_token], label_smoothing=0.1)

3. 梯度裁剪：防止梯度爆炸

   torch.nn.utils.clip_grad_norm_(model.parameters(), CLIP)

4. 早停机制：当验证集损失不再下降时停止训练

4.3 典型训练输出

Epoch: 01 | Train Loss: 4.512 | Val. Loss: 3.987 Epoch: 02 | Train Loss: 3.856 | Val. Loss: 3.542 Epoch: 03 | Train Loss: 3.421 | Val. Loss: 3.215 Epoch: 04 | Train Loss: 3.087 | Val. Loss: 2.976 Epoch: 05 | Train Loss: 2.814 | Val. Loss: 2.801 Epoch: 06 | Train Loss: 2.587 | Val. Loss: 2.674 Epoch: 07 | Train Loss: 2.393 | Val. Loss: 2.587 Epoch: 08 | Train Loss: 2.224 | Val. Loss: 2.531 Epoch: 09 | Train Loss: 2.075 | Val. Loss: 2.498 Epoch: 10 | Train Loss: 1.942 | Val. Loss: 2.481 Test Loss: 2.463 BLEU Score: 35.72

5. 模型部署与推理

5.1 单句翻译函数

def translate_sentence(sentence, src_field, trg_field, model, device, max_len=50): model.eval() if isinstance(sentence, str): nlp = spacy.load("en") tokens = [token.text.lower() for token in nlp(sentence)] else: tokens = [token.lower() for token in sentence] tokens = [src_field.init_token] + tokens + [src_field.eos_token] src_indexes = [src_field.vocab.stoi[token] for token in tokens] src_tensor = torch.LongTensor(src_indexes).unsqueeze(1).to(device) with torch.no_grad(): encoder_outputs, hidden, cell = model.encoder(src_tensor) trg_indexes = [trg_field.vocab.stoi[trg_field.init_token]] for i in range(max_len): trg_tensor = torch.LongTensor([trg_indexes[-1]]).to(device) with torch.no_grad(): output, hidden, cell = model.decoder(trg_tensor, hidden, cell, encoder_outputs) pred_token = output.argmax(1).item() trg_indexes.append(pred_token) if pred_token == trg_field.vocab.stoi[trg_field.eos_token]: break trg_tokens = [trg_field.vocab.itos[i] for i in trg_indexes] return trg_tokens[1:]

5.2 交互式翻译演示

while True: sentence = input("Enter English sentence (or 'q' to quit): ") if sentence.lower() == "q": break translation = translate_sentence(sentence, SRC, TRG, model, device) print(f"French Translation: {' '.join(translation[:-1])}")

5.3 实际翻译示例

6. 进阶优化方向

6.1 模型架构改进

1. Transformer 架构：替换 LSTM 为自注意力机制
2. 双向编码器：捕获前后文信息
3. 深度解码器：增加解码器层数提升表达能力

6.2 训练策略优化

1. 反向翻译：利用目标语言到源语言的翻译增强数据
2. 课程学习：从简单样本开始逐步增加难度
3. 多任务学习：联合训练其他相关 NLP 任务

6.3 推理优化

1. 束搜索：保留多个候选序列而非贪心选择

   def beam_search_decode(model, src, src_field, trg_field, device, beam_size=5, max_len=50): # 实现束搜索解码 pass

2. 长度惩罚：平衡生成长度与质量

3. 集成方法：组合多个模型的预测结果

在实际项目中，通过结合这些优化技术，我们能够将 BLEU 分数进一步提升到 40+ 的水平。特别是在 Transformer 架构下，配合大规模预训练，现代机器翻译系统已经能够达到接近人类水平的翻译质量。