从单词翻译到聊天机器人：用PyTorch搭建你的第一个Seq2Seq模型实战指南

从单词翻译到聊天机器人：用PyTorch搭建你的第一个Seq2Seq模型实战指南

在自然语言处理领域，序列到序列（Seq2Seq）模型已经成为处理文本转换任务的核心技术架构。无论是简单的单词翻译，还是复杂的对话系统，Seq2Seq模型都展现出了强大的适应能力。本文将带你从零开始，用PyTorch构建一个完整的Seq2Seq模型，并逐步扩展其功能，最终实现一个微型聊天机器人原型。

1. Seq2Seq模型基础架构

Seq2Seq模型的核心思想是将一个序列转换为另一个序列，这种架构天然适合机器翻译、文本摘要和对话系统等任务。典型的Seq2Seq模型由两个主要组件构成：编码器（Encoder）和解码器（Decoder）。

1.1 编码器设计

编码器负责将输入序列压缩为一个固定维度的上下文向量（context vector）。我们使用RNN（循环神经网络）作为基础架构：

import torch import torch.nn as nn class Encoder(nn.Module): def __init__(self, input_size, hidden_size): super(Encoder, self).__init__() self.hidden_size = hidden_size self.embedding = nn.Embedding(input_size, hidden_size) self.gru = nn.GRU(hidden_size, hidden_size) def forward(self, input, hidden): embedded = self.embedding(input).view(1, 1, -1) output, hidden = self.gru(embedded, hidden) return output, hidden def init_hidden(self): return torch.zeros(1, 1, self.hidden_size)

关键参数说明：

• input_size: 输入词汇表大小
• hidden_size: 隐藏层维度，决定模型容量
• embedding: 将离散的单词索引转换为连续向量表示

1.2 解码器实现

解码器接收编码器生成的上下文向量，逐步生成目标序列。基础解码器实现如下：

class Decoder(nn.Module): def __init__(self, hidden_size, output_size): super(Decoder, self).__init__() self.hidden_size = hidden_size self.embedding = nn.Embedding(output_size, hidden_size) self.gru = nn.GRU(hidden_size, hidden_size) self.out = nn.Linear(hidden_size, output_size) self.softmax = nn.LogSoftmax(dim=1) def forward(self, input, hidden): output = self.embedding(input).view(1, 1, -1) output = torch.relu(output) output, hidden = self.gru(output, hidden) output = self.softmax(self.out(output[0])) return output, hidden

注意：基础Seq2Seq模型在处理长序列时会出现信息瓶颈问题，后续我们会引入注意力机制来解决这个限制。

2. 数据准备与预处理

构建有效的训练数据管道是模型成功的关键。我们需要设计合理的数据预处理流程，特别是对于变长序列的处理。

2.1 构建词汇表

首先创建一个词汇表管理器，处理文本到索引的转换：

class Lang: def __init__(self, name): self.name = name self.word2index = {} self.word2count = {} self.index2word = {0: "SOS", 1: "EOS"} self.n_words = 2 # 包含起始和结束标记 def add_sentence(self, sentence): for word in sentence.split(' '): self.add_word(word) def add_word(self, word): if word not in self.word2index: self.word2index[word] = self.n_words self.word2count[word] = 1 self.index2word[self.n_words] = word self.n_words += 1 else: self.word2count[word] += 1

2.2 序列标准化处理

处理变长序列的常用方法是填充（padding）和截断（truncation）。我们定义一个标准化函数：

def normalize_string(s): s = s.lower().strip() s = re.sub(r"([.!?])", r" \1", s) s = re.sub(r"[^a-zA-Z.!?]+", r" ", s) return s def prepare_data(pairs, max_length=10): input_lang = Lang('input') output_lang = Lang('output') # 过滤过长的句子并构建词汇表 filtered_pairs = [] for pair in pairs: input_sentence = normalize_string(pair[0]) output_sentence = normalize_string(pair[1]) if len(input_sentence.split(' ')) < max_length and \ len(output_sentence.split(' ')) < max_length: filtered_pairs.append((input_sentence, output_sentence)) input_lang.add_sentence(input_sentence) output_lang.add_sentence(output_sentence) return input_lang, output_lang, filtered_pairs

3. 引入注意力机制

基础Seq2Seq模型的瓶颈在于编码器需要将整个输入序列压缩为一个固定长度的向量。注意力机制通过允许解码器在生成每个词时"关注"输入序列的不同部分来解决这个问题。

3.1 注意力层实现

class Attn(nn.Module): def __init__(self, hidden_size): super(Attn, self).__init__() self.hidden_size = hidden_size self.attn = nn.Linear(self.hidden_size * 2, hidden_size) self.v = nn.Parameter(torch.rand(hidden_size)) def forward(self, hidden, encoder_outputs): max_len = encoder_outputs.size(0) attn_energies = torch.zeros(max_len) for i in range(max_len): attn_energies[i] = self.score(hidden, encoder_outputs[i]) return torch.softmax(attn_energies, dim=0).unsqueeze(0) def score(self, hidden, encoder_output): energy = self.attn(torch.cat((hidden[0], encoder_output), 0)) energy = self.v.dot(energy) return energy

3.2 带注意力的解码器

将注意力机制整合到解码器中：

class AttnDecoder(nn.Module): def __init__(self, hidden_size, output_size, dropout_p=0.1, max_length=10): super(AttnDecoder, self).__init__() self.hidden_size = hidden_size self.output_size = output_size self.dropout_p = dropout_p self.max_length = max_length self.embedding = nn.Embedding(self.output_size, self.hidden_size) self.attn = nn.Linear(self.hidden_size * 2, self.max_length) self.attn_combine = nn.Linear(self.hidden_size * 2, self.hidden_size) self.dropout = nn.Dropout(self.dropout_p) self.gru = nn.GRU(self.hidden_size, self.hidden_size) self.out = nn.Linear(self.hidden_size, self.output_size) def forward(self, input, hidden, encoder_outputs): embedded = self.embedding(input).view(1, 1, -1) embedded = self.dropout(embedded) attn_weights = torch.softmax( self.attn(torch.cat((embedded[0], hidden[0]), 1)), dim=1) attn_applied = torch.bmm(attn_weights.unsqueeze(0), encoder_outputs.unsqueeze(0)) output = torch.cat((embedded[0], attn_applied[0]), 1) output = self.attn_combine(output).unsqueeze(0) output = torch.relu(output) output, hidden = self.gru(output, hidden) output = torch.log_softmax(self.out(output[0]), dim=1) return output, hidden, attn_weights

4. 训练策略与技巧

训练Seq2Seq模型需要特殊的技巧来处理序列生成任务的特点。以下是几个关键训练策略：

4.1 Teacher Forcing

Teacher Forcing是一种训练技术，它使用真实的目标输出作为下一步的输入，而不是使用解码器自己的预测：

def train(input_tensor, target_tensor, encoder, decoder, encoder_optimizer, decoder_optimizer, criterion, max_length=10): encoder_hidden = encoder.init_hidden() encoder_optimizer.zero_grad() decoder_optimizer.zero_grad() input_length = input_tensor.size(0) target_length = target_tensor.size(0) encoder_outputs = torch.zeros(max_length, encoder.hidden_size) loss = 0 for ei in range(input_length): encoder_output, encoder_hidden = encoder( input_tensor[ei], encoder_hidden) encoder_outputs[ei] = encoder_output[0, 0] decoder_input = torch.tensor([[SOS_token]]) decoder_hidden = encoder_hidden use_teacher_forcing = True if random.random() < teacher_forcing_ratio else False if use_teacher_forcing: # Teacher forcing: 使用真实目标作为下一个输入 for di in range(target_length): decoder_output, decoder_hidden, decoder_attention = decoder( decoder_input, decoder_hidden, encoder_outputs) loss += criterion(decoder_output, target_tensor[di]) decoder_input = target_tensor[di] # Teacher forcing else: # 不使用teacher forcing: 使用自己的预测作为下一个输入 for di in range(target_length): decoder_output, decoder_hidden, decoder_attention = decoder( decoder_input, decoder_hidden, encoder_outputs) topv, topi = decoder_output.topk(1) decoder_input = topi.squeeze().detach() # detach from history loss += criterion(decoder_output, target_tensor[di]) if decoder_input.item() == EOS_token: break loss.backward() encoder_optimizer.step() decoder_optimizer.step() return loss.item() / target_length

4.2 超参数设置

训练Seq2Seq模型需要仔细调整以下超参数：

5. 从翻译到聊天机器人

现在我们已经构建了一个完整的Seq2Seq模型，可以将其扩展到对话系统。关键在于准备合适的对话数据集和调整模型架构。

5.1 对话数据准备

对话数据通常采用"问题-回答"对的形式。我们可以使用Cornell Movie Dialogs Corpus等公开数据集：

def load_conversations(data_path): # 加载对话数据 lines = {} with open(os.path.join(data_path, "movie_lines.txt"), encoding='iso-8859-1') as f: for line in f: parts = line.strip().split(" +++$+++ ") lines[parts[0]] = parts[4] conversations = [] with open(os.path.join(data_path, "movie_conversations.txt"), encoding='iso-8859-1') as f: for line in f: parts = line.strip().split(" +++$+++ ") conv = eval(parts[3]) # 对话ID列表 for i in range(len(conv) - 1): input_line = lines[conv[i]] target_line = lines[conv[i+1]] conversations.append([input_line, target_line]) return conversations

5.2 对话生成策略

生成对话时需要考虑上下文连贯性。我们可以实现一个交互式对话循环：

def evaluate(encoder, decoder, sentence, max_length=10): with torch.no_grad(): input_tensor = tensor_from_sentence(input_lang, sentence) input_length = input_tensor.size()[0] encoder_hidden = encoder.init_hidden() encoder_outputs = torch.zeros(max_length, encoder.hidden_size) for ei in range(input_length): encoder_output, encoder_hidden = encoder(input_tensor[ei], encoder_hidden) encoder_outputs[ei] += encoder_output[0, 0] decoder_input = torch.tensor([[SOS_token]]) decoder_hidden = encoder_hidden decoded_words = [] decoder_attentions = torch.zeros(max_length, max_length) for di in range(max_length): decoder_output, decoder_hidden, decoder_attention = decoder( decoder_input, decoder_hidden, encoder_outputs) decoder_attentions[di] = decoder_attention.data topv, topi = decoder_output.data.topk(1) if topi.item() == EOS_token: decoded_words.append('<EOS>') break else: decoded_words.append(output_lang.index2word[topi.item()]) decoder_input = topi.squeeze().detach() return decoded_words, decoder_attentions[:di+1] def chat(encoder, decoder): while True: try: input_sentence = input('> ') if input_sentence.lower() in ['quit', 'exit']: break output_words, _ = evaluate(encoder, decoder, input_sentence) print('Bot:', ' '.join(output_words[:-1])) # 去掉<EOS> except KeyError: print("抱歉，我不理解这句话。")

在实际项目中，我发现使用Beam Search策略可以显著提高生成质量，特别是在处理开放域对话时。此外，引入最大互信息（MMI）等技术可以避免生成通用性回复。