13、深入理解Seq2Seq：让我们看看语言翻译是怎么来的

你好，我是独行。

上节课我们一起学习了Word2Vec，Word2Vec的主要能力是把词汇放在多维的空间里，相似的词汇会被放在邻近的位置。这节课我们将进入Seq2Seq的领域，了解这种更为复杂且功能强大的模型，它不仅能理解词汇，还能把这些词汇串联成完整的句子。

Seq2Seq

Seq2Seq（Sequence-to-Sequence），顾名思义是从一个序列到另一个序列的转换。它不仅仅能理解单词之间的关系，而且还能把整个句子的意思打包，并解压成另一种形式的表达。如果说Word2Vec是让我们的机器学会了理解词汇的话，那Seq2Seq则是教会了机器如何理解句子并进行相应地转化。

在这个过程中，我们会遇到两个核心的角色：编码器（Encoder）和解码器（Decoder）。编码器的任务是理解和压缩信息，就像是把一封长信函整理成一个精简的摘要；而解码器则需要将这个摘要展开，翻译成另一种语言或形式的完整信息。这个过程有一定的挑战，比如如何确保信息在这次转换中不丢失精髓，而是以新的面貌精准地呈现出来，这就是我们接下来要探索的内容之一。

基本概念

Seq2Seq也是一种神经网络架构，模型的核心由两部分组成：编码器（Encoder）和解码器（Decoder）。你可以看一下这个架构的示意图。

编码器

编码器的任务是读取并理解输入序列，然后把它转换为一个固定长度的上下文向量，也叫作状态向量。这个向量是输入序列的一种内部表示，捕捉了序列的关键信息。编码器通常是一个循环神经网络（RNN）或其变体，比如长短期记忆网络（LSTM）或门控循环单元（GRU），它们能够处理不同长度的输入序列，并且记住序列中的长期依赖关系。

解码器

解码器的任务是接收编码器生成的上下文向量，并基于这个向量生成目标序列。解码过程是一步步进行的，每一步生成目标序列中的一个元素，比如一个词或字符，直到生成特殊的结束符号，表示输出序列的结束。解码器通常也是一个RNN、LSTM或GRU，它不仅依赖于编码器的上下文向量，还可能依赖于自己之前的输出，来生成下一个输出元素。

注意力机制（可选）

在编码器和解码器之间，可能还会有一个注意力机制（Attention Mechanism）。注意力机制使解码器能够在生成每个输出元素时“关注”输入序列中的不同部分，从而提高模型处理长序列和捕捉复杂依赖关系的能力。编码器、解码器、注意力机制之间是怎样协作的呢？你可以看一下我给出的示意图。

下面我通过一个翻译的例子，来说明Seq2Seq的工作原理。

工作原理

我们先从模型的训练开始，Seq2Seq的训练和Word2Vec不太一样，因为我们讲解的是中英文翻译场景，所以训练的时候，训练数据是中英文数据对。Seq2Seq的训练会比Word2Vec更加复杂一些。上节课的Word2Vec，我们使用的是gensim库提供的基础模型，直接进行训练，这节课我们完全从头写起，训练一个Seq2Seq模型。

模型训练

我们先准备训练数据，可以在网上找公开的翻译数据集，我们用的是 AIchallenger 2017，这个数据集有1000万对中英文数据，不过因为电脑配置问题，我直接从里面中文和英文的部分各取了10000条进行训练。数据集名称是train_1w.zh和train_1w.en。

cn_sentences = []
zh_file_path = "train_1w.zh"
# 使用Python的文件操作逐行读取文件，并将每一行的内容添加到列表中
with open(zh_file_path, "r", encoding="utf-8") as file:
    for line in file:
        # 去除行末的换行符并添加到列表中
        cn_sentences.append(line.strip())

en_sentences = []
en_file_path = "train_1w.en"
# 使用Python的文件操作逐行读取文件，并将每一行的内容添加到列表中
with open(en_file_path, "r", encoding="utf-8") as file:
    for line in file:
        # 去除行末的换行符并添加到列表中
        en_sentences.append(line.strip())

接下来，基于训练数据集构建中文和英文的词汇表，将每个词映射到一个唯一的索引（integer）。

1
2
3

# cn_sentences 和 en_sentences 分别包含了所有的中文和英文句子
cn_vocab = build_vocab(cn_sentences, tokenize_cn, max_size=10000, min_freq=2)
en_vocab = build_vocab(en_sentences, tokenize_en, max_size=10000, min_freq=2)

我们再来看 biild_vocab的源码。

def build_vocab(sentences, tokenizer, max_size, min_freq):
    token_freqs = Counter()
    for sentence in sentences:
        tokens = tokenizer(sentence)
        token_freqs.update(tokens)
    vocab = {token: idx + 4 for idx, (token, freq) in enumerate(token_freqs.items()) if freq >= min_freq}
    vocab['<unk>'] = 0
    vocab['<pad>'] = 1
    vocab['<sos>'] = 2
    vocab['<eos>'] = 3
    return vocab

思路就是把所有的句子读进去，循环分词，放入字典，放的时候要判断一下是否大于等于min_freq，用来过滤掉出现频率较低的词汇，最后构建出来的词汇表如下：

vocab = {
    '<unk>': 0,
    '<pad>': 1,
    '<sos>': 2,
    '<eos>': 3,
    'i': 4,
    'like': 5,
    'learning': 6,
    'machine': 7,
    'is': 8,
    'very': 9,
    'interesting': 10,
    ...
}

我们来看一下里面比较重要的几个部分。

<unk>：未知单词，表示在训练数据中没有出现过的单词。当模型在处理输入文本时遇到未知单词时，会用这个标记来表示。
<pad>：填充单词，用于将不同长度的序列填充到相同的长度。在处理批次数据时，由于不同序列的长度可能不同，因此需要用这个标记把短序列填充到与最长序列相同的长度，以便进行批次处理。
<sos>：句子起始标记，表示句子的开始位置。在Seq2Seq模型中，通常会在目标句子的开头添加这个标记，以指示解码器开始生成输出。
<eos>：句子结束标记，表示句子的结束位置。在Seq2Seq模型中，通常会在目标句子的末尾添加该标记，以指示解码器生成结束。

创建训练数据集，将数据处理成方便训练的格式：语言序列，比如 [1,2,3,4]。

1 2	dataset = TranslationDataset(cn_sentences, en_sentences, cn_vocab, en_vocab, tokenize_cn, tokenize_en) train_loader = DataLoader(dataset, batch_size=32, collate_fn=collate_fn)

然后检测是否有显卡：

1
2
3

# 检查是否有可用的GPU，如果没有，则使用CPU
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("训练设备为：", device)

创建模型，参数的解释可以参考代码注释。


# 定义一些超参数
INPUT_DIM = 10000  # 输入语言的词汇量
OUTPUT_DIM = 10000  # 输出语言的词汇量
ENC_EMB_DIM = 256  # 编码器嵌入层大小，也就是编码器词向量维度
DEC_EMB_DIM = 256  # 解码器嵌入层大小，解码器词向量维度
HID_DIM = 512  # 隐藏层维度
N_LAYERS = 2  # RNN层的数量
ENC_DROPOUT = 0.5  # 编码器神经元输出的数据有50%会被随机丢掉
DEC_DROPOUT = 0.5  # 解码器同上

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)

model = Seq2Seq(enc, dec, device).to(device)
# 假定模型已经被实例化并移到了正确的设备上
model.to(device)
# 定义优化器和损失函数
optimizer = optim.Adam(model.parameters())
criterion = nn.CrossEntropyLoss(ignore_index=en_vocab['<pad>'])  # 忽略<pad>标记的损失

开始训练：

num_epochs = 10  # 训练轮数
for epoch in range(num_epochs):
    model.train()
    total_loss = 0
    for src, trg in train_loader:
        src, trg = src.to(device), trg.to(device)
        optimizer.zero_grad()  # 清空梯度
        output = model(src, trg[:-1])  # 输入给模型的是除了最后一个词的目标句子
        # Reshape输出以匹配损失函数期望的输入
        output_dim = output.shape[-1]
        output = output.view(-1, output_dim)
        trg = trg[1:].view(-1)  # 从第一个词开始的目标句子
        loss = criterion(output, trg) # 计算模型输出和实际目标序列之间的损失
        loss.backward()  # 通过反向传播计算损失相对于模型参数的梯度
        optimizer.step()  # 根据梯度更新模型参数，这是优化器的一个步骤
        total_loss += loss.item()
    avg_loss = total_loss / len(train_loader)
    print(f'Epoch {epoch + 1}/{num_epochs}, Average Loss: {avg_loss}')

我拿下面的素材举例，简单解释一下训练过程。

1 2	我喜欢学习机器学习。 I like studying machine learning

在开始训练之前，先把原文本转化成在对应词语表里的语言序列，比如在中文词汇表中，我喜欢学习机器学习 分别对应的是 1,2,3,4,5，那么转化成的语言序列就是 [1,2,3,4,5]，也就是前面讲的train_loader里的格式。

编码器接收到语言序列，经过神经网络GRU单元处理后，生成一个上下文向量，这个上下文向量会作为解码器的初始状态。

解码器接收上下文向量作为输入，并根据当前上下文以及已生成的部分目标语言序列，计算目标词汇表中每个单词的概率分布。例如，在第一个时间步，解码器可能计算出目标词汇表中每个单词的概率分布，如 "I": 0.3, "like": 0.1, "studying": 0.5, "machine": 0.05, "learning": 0.05，根据解码器生成的概率分布，选择概率最高的词studying作为当前时间步的输出。

模型将解码器生成的输出词汇与目标语言句子（“I like studying machine learning.”）中当前时间步对应的词汇进行对比。这里解码器输出的 "studying" 与目标语言句子中的 "I" 进行对比，发现它们之间的差别较大。

根据解码器输出 "studying" 和目标语言句子中的真实词汇 "I" 计算损失，并通过反向传播算法计算梯度。损失值是一个衡量模型预测输出与真实目标之间差异的指标。然后，根据损失值更新模型参数，使模型能够更准确地预测下一个词汇。

重复以上步骤，直到模型达到指定的训练轮数或者满足其他停止训练的条件。在每次训练迭代中，模型都在尝试调整自己的参数，以使其预测输出更接近真实的目标语言序列，从而提高翻译质量。

所以这里就能看出，训练轮数就非常关键，不能太少，也不能太多。

模型验证

def translate_sentence(sentence, src_vocab, trg_vocab, model, device, max_len=50):
    # 将输入句子进行分词并转换为索引序列
    src_tokens = ['<sos>'] + tokenize_cn(sentence) + ['<eos>']
    src_indices = [src_vocab[token] if token in src_vocab else src_vocab['<unk>'] for token in src_tokens]
    # 将输入句子转换为张量并移动到设备上
    src_tensor = torch.LongTensor(src_indices).unsqueeze(1).to(device)
    # 将输入句子传递给编码器以获取上下文张量
    with torch.no_grad():
        encoder_hidden = model.encoder(src_tensor)
    # 初始化解码器输入为<sos>
    trg_token = '<sos>'
    trg_index = trg_vocab[trg_token]
    # 存储翻译结果
    translation = []
    # 解码过程
    for _ in range(max_len):
        # 将解码器输入传递给解码器，并获取输出和隐藏状态
        with torch.no_grad():
            trg_tensor = torch.LongTensor([trg_index]).to(device)
            output, encoder_hidden = model.decoder(trg_tensor, encoder_hidden)
        # 获取解码器输出中概率最高的单词的索引
        pred_token_index = output.argmax(dim=1).item()
        # 如果预测的单词是句子结束符，则停止解码
        if pred_token_index == trg_vocab['<eos>']:
            break
        # 否则，将预测的单词添加到翻译结果中
        pred_token = list(trg_vocab.keys())[list(trg_vocab.values()).index(pred_token_index)]
        translation.append(pred_token)
        # 更新解码器输入为当前预测的单词
        trg_index = pred_token_index
    # 将翻译结果转换为字符串并返回
    translation = ' '.join(translation)
    return translation

sentence = "我喜欢学习机器学习。"
translation = translate_sentence(sentence, cn_vocab, en_vocab, model, device)
print(f"Chinese: {sentence}")
print(f"Translation: {translation}")

程序输出如下：

1 2	Chinese: 我喜欢学习机器学习。 Translation: I a a a . a . . . .

看上去只翻译成功了“我”这个字，其他都没出来，大概率是因为训练数据太少的原因。

推理过程和训练过程很像，区别在于，训练过程中模型会记住参数，推理的时候直接根据这些参数计算下一个词的概率即可。

结尾放一下完整的代码：

import torch
from torch.utils.data import Dataset, DataLoader
import spacy
import jieba
from collections import Counter
from torch.nn.utils.rnn import pad_sequence
import torch.nn as nn
import random
import torch.optim as optim

# 加载英文的Spacy模型
spacy_en = spacy.load('en_core_web_sm')

def tokenize_en(text):
    """
    Tokenizes English text from a string into a list of strings (tokens)
    """
    return [tok.text for tok in spacy_en.tokenizer(text)]

def tokenize_cn(text):
    """
    Tokenizes Chinese text from a string into a list of strings (tokens)
    """
    return list(jieba.cut(text))

def build_vocab(sentences, tokenizer, max_size, min_freq):
    token_freqs = Counter()
    for sentence in sentences:
        tokens = tokenizer(sentence)
        token_freqs.update(tokens)
    vocab = {token: idx + 4 for idx, (token, freq) in enumerate(token_freqs.items()) if freq >= min_freq}
    vocab['<unk>'] = 0
    vocab['<pad>'] = 1
    vocab['<sos>'] = 2
    vocab['<eos>'] = 3
    return vocab

class TranslationDataset(Dataset):
    def __init__(self, src_sentences, trg_sentences, src_vocab, trg_vocab, tokenize_src, tokenize_trg):
        self.src_sentences = src_sentences
        self.trg_sentences = trg_sentences
        self.src_vocab = src_vocab
        self.trg_vocab = trg_vocab
        self.tokenize_src = tokenize_src
        self.tokenize_trg = tokenize_trg
    def __len__(self):
        return len(self.src_sentences)
    def __getitem__(self, idx):
        src_sentence = self.src_sentences[idx]
        trg_sentence = self.trg_sentences[idx]
        src_indices = [self.src_vocab[token] if token in self.src_vocab else self.src_vocab['<unk>']
                       for token in ['<sos>'] + self.tokenize_src(src_sentence) + ['<eos>']]
        trg_indices = [self.trg_vocab[token] if token in self.trg_vocab else self.trg_vocab['<unk>']
                       for token in ['<sos>'] + self.tokenize_trg(trg_sentence) + ['<eos>']]
        return torch.tensor(src_indices), torch.tensor(trg_indices)

def collate_fn(batch):
    src_batch, trg_batch = zip(*batch)
    src_batch = pad_sequence(src_batch, padding_value=1)  # 1 is the index for <pad>
    trg_batch = pad_sequence(trg_batch, padding_value=1)  # 1 is the index for <pad>
    return src_batch, trg_batch

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.GRU(emb_dim, hid_dim, n_layers, dropout=dropout)
        self.dropout = nn.Dropout(dropout)
    def forward(self, src):
        # src: [src_len, batch_size]
        embedded = self.dropout(self.embedding(src))
        outputs, hidden = self.rnn(embedded)
        return hidden

class Decoder(nn.Module):
    def __init__(self, output_dim, emb_dim, hid_dim, n_layers, dropout):
        super().__init__()
        self.output_dim = output_dim
        self.embedding = nn.Embedding(output_dim, emb_dim)
        self.rnn = nn.GRU(emb_dim, hid_dim, n_layers, dropout=dropout)
        self.fc_out = nn.Linear(hid_dim, output_dim)
        self.dropout = nn.Dropout(dropout)
    def forward(self, input, hidden):
        input = input.unsqueeze(0)  # input: [1, batch_size]
        embedded = self.dropout(self.embedding(input))
        output, hidden = self.rnn(embedded, hidden)
        prediction = self.fc_out(output.squeeze(0))
        return prediction, hidden

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):
        # src: [src_len, batch_size]
        # trg: [trg_len, batch_size]
        # teacher_forcing_ratio是使用真实标签的概率
        trg_len = trg.shape[0]
        batch_size = trg.shape[1]
        trg_vocab_size = self.decoder.output_dim
        # 存储解码器输出
        outputs = torch.zeros(trg_len, batch_size, trg_vocab_size).to(self.device)
        # 编码器的最后一个隐藏状态用作解码器的初始隐藏状态
        hidden = self.encoder(src)
        # 解码器的第一个输入是<sos> tokens
        input = trg[0, :]
        for t in range(1, trg_len):
            output, hidden = self.decoder(input, hidden)
            outputs[t] = output
            # 决定是否使用teacher forcing
            teacher_force = random.random() < teacher_forcing_ratio
            top1 = output.argmax(1)
            input = trg[t] if teacher_force else top1
        return outputs

cn_sentences = []
zh_file_path = "train_1w.zh"
# 使用Python的文件操作逐行读取文件，并将每一行的内容添加到列表中
with open(zh_file_path, "r", encoding="utf-8") as file:
    for line in file:
        # 去除行末的换行符并添加到列表中
        cn_sentences.append(line.strip())
en_sentences = []
en_file_path = "train_1w.en"
# 使用Python的文件操作逐行读取文件，并将每一行的内容添加到列表中
with open(en_file_path, "r", encoding="utf-8") as file:
    for line in file:
        # 去除行末的换行符并添加到列表中
        en_sentences.append(line.strip())
# cn_sentences 和 en_sentences 分别包含了所有的中文和英文句子
cn_vocab = build_vocab(cn_sentences, tokenize_cn, max_size=10000, min_freq=2)
en_vocab = build_vocab(en_sentences, tokenize_en, max_size=10000, min_freq=2)

# cn_vocab 和 en_vocab 是已经创建的词汇表
dataset = TranslationDataset(cn_sentences, en_sentences, cn_vocab, en_vocab, tokenize_cn, tokenize_en)
train_loader = DataLoader(dataset, batch_size=32, collate_fn=collate_fn)
# 检查是否有可用的GPU，如果没有，则使用CPU
device = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print("训练设备为：", device)
# 定义一些超参数
INPUT_DIM = 10000  # 输入语言的词汇量
OUTPUT_DIM = 10000  # 输出语言的词汇量
ENC_EMB_DIM = 256  # 编码器嵌入层大小
DEC_EMB_DIM = 256  # 解码器嵌入层大小
HID_DIM = 512  # 隐藏层维度
N_LAYERS = 2  # RNN层的数量
ENC_DROPOUT = 0.5  # 编码器中dropout的比例
DEC_DROPOUT = 0.5  # 解码器中dropout的比例
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)
model = Seq2Seq(enc, dec, device).to(device)
# 假定模型已经被实例化并移到了正确的设备上
model.to(device)
# 定义优化器和损失函数
optimizer = optim.Adam(model.parameters())
criterion = nn.CrossEntropyLoss(ignore_index=en_vocab['<pad>'])  # 忽略<pad>标记的损失
num_epochs = 10  # 训练轮数
for epoch in range(num_epochs):
    model.train()
    total_loss = 0
    for src, trg in train_loader:
        src, trg = src.to(device), trg.to(device)
        optimizer.zero_grad()  # 清空梯度
        output = model(src, trg[:-1])  # 输入给模型的是除了最后一个词的目标句子
        # Reshape输出以匹配损失函数期望的输入
        output_dim = output.shape[-1]
        output = output.view(-1, output_dim)
        trg = trg[1:].view(-1)  # 从第一个词开始的目标句子
        loss = criterion(output, trg)
        loss.backward()  # 反向传播
        optimizer.step()  # 更新参数
        total_loss += loss.item()
    avg_loss = total_loss / len(train_loader)
    print(f'Epoch {epoch + 1}/{num_epochs}, Average Loss: {avg_loss}')
    # 可以在这里添加验证步骤

def translate_sentence(sentence, src_vocab, trg_vocab, model, device, max_len=50):
    # 将输入句子进行分词并转换为索引序列
    src_tokens = ['<sos>'] + tokenize_cn(sentence) + ['<eos>']
    src_indices = [src_vocab[token] if token in src_vocab else src_vocab['<unk>'] for token in src_tokens]
    # 将输入句子转换为张量并移动到设备上
    src_tensor = torch.LongTensor(src_indices).unsqueeze(1).to(device)
    # 将输入句子传递给编码器以获取上下文张量
    with torch.no_grad():
        encoder_hidden = model.encoder(src_tensor)
    # 初始化解码器输入为<sos>
    trg_token = '<sos>'
    trg_index = trg_vocab[trg_token]
    # 存储翻译结果
    translation = []
    # 解码过程
    for _ in range(max_len):
        # 将解码器输入传递给解码器，并获取输出和隐藏状态
        with torch.no_grad():
            trg_tensor = torch.LongTensor([trg_index]).to(device)
            output, encoder_hidden = model.decoder(trg_tensor, encoder_hidden)
        # 获取解码器输出中概率最高的单词的索引
        pred_token_index = output.argmax(dim=1).item()
        # 如果预测的单词是句子结束符，则停止解码
        if pred_token_index == trg_vocab['<eos>']:
            break
        # 否则，将预测的单词添加到翻译结果中
        pred_token = list(trg_vocab.keys())[list(trg_vocab.values()).index(pred_token_index)]
        translation.append(pred_token)
        # 更新解码器输入为当前预测的单词
        trg_index = pred_token_index
    # 将翻译结果转换为字符串并返回
    translation = ' '.join(translation)
    return translation

sentence = "我喜欢学习机器学习。"
translation = translate_sentence(sentence, cn_vocab, en_vocab, model, device)
print(f"Chinese: {sentence}")
print(f"Translation: {translation}")

小结

这节课我们自己动手训练了一个Seq2Seq模型，Seq2Seq可以算是一种高级的神经网络模型了，除了做语言翻译外，甚至可以做基本的问答系统了。但是，Seq2Seq缺点也比较明显，首先Seq2Seq使用固定上下文长度，所以长距离依赖能力较弱。此外，Seq2Seq训练和推理通常需要逐步处理输入和输出序列，所以处理长序列可能会有限制。最后Seq2Seq参数量通常较少，所以面对复杂场景，模型性能可能会受限。

带着这些问题，下一节课我将会向你介绍终极大boss：Transformer，我们学习了这么多基础概念，就是为学习Transformer做铺垫，从ML->NLP->Word2Vec->Seq2Seq->Transformer一步一步递进。

注：en_core_web_sm、train_1w.zh、train_1w.en 链接: https://pan.baidu.com/s/1_GG3bIAjqpPGLGugHEI5Dg?pwd=fm8j 提取码: fm8j

思考题

我刚刚讲过，推理的时候模型会使用训练过程中记住的参数来进行概率预测，你可以思考一下，模型的参数到底是什么？欢迎在评论区留言，我们一起讨论学习，如果你觉得这节课的内容对你有帮助的话，也欢迎你分享给需要的朋友，邀TA一起学习，我们下节课再见！