查看全课大纲（第 17 / 40 节）

Seq2Seq 与编码器-解码器

Seq2Seq 与编码器-解码器：如何让神经网络学会“翻译”？

小象实战讲义 · 深度学习基础：从神经网络到 Transformer

在上一节，我们学习了循环神经网络（RNN）及其变种LSTM，它们擅长处理序列信息。但如何让一个序列（如一句英文）转换成另一个序列（如一句中文）呢？这就是序列到序列（Sequence-to-Sequence, Seq2Seq）模型要解决的核心问题。本节将深入剖析Seq2Seq的经典架构——编码器-解码器（Encoder-Decoder），揭示其如何通过一个“中间表示”来完成翻译、摘要等任务。学完本节，你将能亲手搭建一个简易的Seq2Seq模型，并理解其数据流动与梯度传播的完整路径。

💡 核心导读

Seq2Seq是什么：一种将任意长度输入序列映射为任意长度输出序列的通用框架，是机器翻译、对话生成等任务的基石。
编码器-解码器架构：编码器将输入序列压缩为一个固定维度的上下文向量（Context Vector）；解码器以此向量为起点，逐步生成输出序列。
核心挑战与应对：上下文向量是信息瓶颈；通过引入注意力机制（下节内容）可极大缓解，但本节聚焦于基础架构。
实现细节：使用RNN（或LSTM/GRU）作为编码器和解码器，处理变长序列需要pack_padded_sequence等技巧。
训练与推理模式：训练时使用“教师强制”（Teacher Forcing）；推理时使用自回归生成。

🧭 前置知识

循环神经网络（RNN/LSTM）：理解其按时间步展开的计算过程及隐藏状态h_t的传递。
张量形状：熟悉(batch_size, sequence_length, feature_dim)的表示法。
softmax与交叉熵损失：用于对解码器每一步的输出进行多分类。
PyTorch基础：包括nn.Module的定义、nn.Embedding层、torch.optim的使用。

从RNN到Seq2Seq：架构的演进

基本的RNN适用于输入与输出序列等长（如序列标注）或单输入多输出等任务。但对于机器翻译这种输入输出长度可变且不等的任务，我们需要一个更灵活的架构。

Seq2Seq模型（Sutskever et al., 2014, arXiv:1409.3215）应运而生。其核心思想是：使用一个RNN（编码器）读取整个输入序列，并将其编码成一个固定大小的上下文向量（通常就是编码器最后一个时间步的隐藏状态）。然后，使用另一个RNN（解码器）以该上下文向量为初始状态，逐步生成输出序列。

输入序列: [x1, x2, x3] (编码器)
          ↓
上下文向量: c = encoder_last_hidden
          ↓
输出序列: [y1, y2, y3, y4] (解码器)

这种“编码-解码”的范式，使得模型能够先理解整个输入句子的语义，再据此生成目标句子。

编码器-解码器架构详解

编码器（Encoder）

编码器通常是一个RNN（实践中多用LSTM或GRU）。它逐时间步处理输入序列，并更新其隐藏状态。处理完最后一个输入后，其最终的隐藏状态h_T（对于双向RNN则是前后向状态的拼接）被视为整个输入序列的摘要，即上下文向量 c。

import torch
import torch.nn as nn
import torch.nn.functional as F

torch.manual_seed(0)

# 定义超参数
vocab_size = 20  # 假设源语言词表大小
tgt_vocab_size = 25 # 假设目标语言词表大小
embedding_dim = 8
hidden_dim = 16
batch_size = 2
src_seq_len = 5
tgt_seq_len = 6

# 1. 编码器实现
class EncoderRNN(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.rnn = nn.GRU(embed_dim, hidden_dim, batch_first=True)

    def forward(self, src_tokens):
        # src_tokens shape: (batch_size, src_seq_len)
        embedded = self.embedding(src_tokens)  # (batch, src_seq_len, embed_dim)
        # GRU返回：所有时间步的输出outputs，最后一个时间步的隐藏状态hidden
        outputs, hidden = self.rnn(embedded)   # outputs: (batch, src_seq_len, hidden_dim)
                                               # hidden: (1, batch, hidden_dim)
        # 我们取最后一个时间步的隐藏状态作为上下文向量
        context = hidden.squeeze(0)  # (batch, hidden_dim) 去掉num_layers维度
        return context, outputs

# 构造模拟输入数据
src_data = torch.randint(0, vocab_size, (batch_size, src_seq_len))  # 随机整数代表词ID
encoder = EncoderRNN(vocab_size, embedding_dim, hidden_dim)
context_vector, encoder_outputs = encoder(src_data)

print("src_data.shape =", tuple(src_data.shape))
print("context_vector.shape =", tuple(context_vector.shape))
print("encoder_outputs.shape =", tuple(encoder_outputs.shape))
print("context_vector[0] (前4维) =", [round(v.item(), 4) for v in context_vector[0][:4]])

src_data.shape = (2, 5)
context_vector.shape = (2, 16)
encoder_outputs.shape = (2, 5, 16)
context_vector[0] (前4维) = [-0.3879, 0.3692, -0.2767, 0.0329]

解码器（Decoder）

解码器是另一个RNN，它的任务是以上下文向量c为初始隐藏状态，并逐步生成目标序列。在训练阶段，我们使用“教师强制”（Teacher Forcing）：将真实的目标序列（虽然偏移了一位）作为解码器的输入，以加速收敛。每一步，解码器接收上一个时间步的真实词嵌入和当前隐藏状态，输出当前时间步对目标词表的预测分数。

import torch
import torch.nn as nn

torch.manual_seed(0)

tgt_vocab_size = 25
embedding_dim = 8
hidden_dim = 16
batch_size = 2
tgt_seq_len = 6

# 2. 解码器实现（训练模式，使用教师强制）
class DecoderRNN(nn.Module):
    def __init__(self, tgt_vocab_size, embed_dim, hidden_dim):
        super().__init__()
        self.embedding = nn.Embedding(tgt_vocab_size, embed_dim)
        self.rnn = nn.GRU(embed_dim, hidden_dim, batch_first=True)
        # 线性层将RNN隐藏状态映射到目标词表空间
        self.fc_out = nn.Linear(hidden_dim, tgt_vocab_size)

    def forward(self, tgt_tokens, hidden_init):
        # tgt_tokens shape: (batch_size, tgt_seq_len) 在训练时是真实目标序列
        # hidden_init shape: (batch_size, hidden_dim) 来自编码器的上下文向量
        embedded = self.embedding(tgt_tokens)  # (batch, tgt_seq_len, embed_dim)
        # 需要将hidden_init变形为RNN期望的格式: (num_layers, batch, hidden_dim)
        hidden_init = hidden_init.unsqueeze(0)  # (1, batch, hidden_dim)
        # 解码器RNN前向传播
        outputs, _ = self.rnn(embedded, hidden_init)  # outputs: (batch, tgt_seq_len, hidden_dim)
        # 将每个时间步的隐藏状态映射为词表分数 (logits)
        logits = self.fc_out(outputs)  # (batch, tgt_seq_len, tgt_vocab_size)
        return logits

# 构造模拟目标数据（训练时）
tgt_data = torch.randint(0, tgt_vocab_size, (batch_size, tgt_seq_len))
# 用一个随机向量代替编码器给出的上下文向量（真实流程里它来自编码器最后一步的隐状态）
context_vector = torch.randn(batch_size, hidden_dim)
decoder = DecoderRNN(tgt_vocab_size, embedding_dim, hidden_dim)

# 解码器输入：通常是目标序列的起始符+序列[:-1]（教师强制，练习里会改造成标准写法）
decoder_logits = decoder(tgt_data, context_vector)

print("tgt_data.shape =", tuple(tgt_data.shape))
print("decoder_logits.shape =", tuple(decoder_logits.shape))
print("logits[0, 0, :5] =", [round(v.item(), 4) for v in decoder_logits[0, 0, :5]])

tgt_data.shape = (2, 6)
decoder_logits.shape = (2, 6, 25)
logits[0, 0, :5] = [-0.1076, 0.6024, -0.1926, -0.5967, 0.2415]

组合与训练

将编码器和解码器组合，并定义损失函数。对于序列生成，我们通常计算每个时间步的交叉熵损失，然后求和或平均。

import torch
import torch.nn as nn

torch.manual_seed(0)

vocab_size, tgt_vocab_size = 20, 25
embedding_dim, hidden_dim = 8, 16
batch_size, src_seq_len, tgt_seq_len = 2, 5, 6

# 为保证每个代码块都能单独运行，这里把编码器、解码器一并定义
class EncoderRNN(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.rnn = nn.GRU(embed_dim, hidden_dim, batch_first=True)

    def forward(self, src_tokens):
        outputs, hidden = self.rnn(self.embedding(src_tokens))
        return hidden.squeeze(0), outputs  # (batch, hidden), (batch, seq, hidden)

class DecoderRNN(nn.Module):
    def __init__(self, tgt_vocab_size, embed_dim, hidden_dim):
        super().__init__()
        self.embedding = nn.Embedding(tgt_vocab_size, embed_dim)
        self.rnn = nn.GRU(embed_dim, hidden_dim, batch_first=True)
        self.fc_out = nn.Linear(hidden_dim, tgt_vocab_size)

    def forward(self, tgt_tokens, hidden_init):
        outputs, _ = self.rnn(self.embedding(tgt_tokens), hidden_init.unsqueeze(0))
        return self.fc_out(outputs)

# 3. 组合编码器-解码器并计算损失
class Seq2Seq(nn.Module):
    def __init__(self, encoder, decoder):
        super().__init__()
        self.encoder = encoder
        self.decoder = decoder

    def forward(self, src, tgt):
        context, _ = self.encoder(src)
        # 真实训练中 tgt_input = <sos> + tgt[:-1]；这里先用完整 tgt 演示前向形状
        return self.decoder(tgt, context)

src_data = torch.randint(0, vocab_size, (batch_size, src_seq_len))
tgt_data = torch.randint(0, tgt_vocab_size, (batch_size, tgt_seq_len))

model = Seq2Seq(EncoderRNN(vocab_size, embedding_dim, hidden_dim),
                DecoderRNN(tgt_vocab_size, embedding_dim, hidden_dim))
criterion = nn.CrossEntropyLoss()

logits = model(src_data, tgt_data)
# 交叉熵要求 (样本数, 类别数) 对 (样本标签)：把 batch 与 seq 两维摊平
loss = criterion(logits.view(-1, tgt_vocab_size), tgt_data.view(-1))

print("logits.shape =", tuple(logits.shape))
print("loss.item() =", round(loss.item(), 4))

logits.shape = (2, 6, 25)
loss.item() = 3.1815

推理：自回归生成

在推理（或测试）阶段，我们没有真实的目标序列作为解码器输入。解码器必须自回归地生成：从起始符<sos>开始，将当前时间步的预测结果（通常取argmax）作为下一个时间步的输入，直到生成结束符<eos>或达到最大长度。

import torch
import torch.nn as nn

torch.manual_seed(0)

vocab_size, tgt_vocab_size = 20, 25
embedding_dim, hidden_dim = 8, 16

class EncoderRNN(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.rnn = nn.GRU(embed_dim, hidden_dim, batch_first=True)

    def forward(self, src_tokens):
        outputs, hidden = self.rnn(self.embedding(src_tokens))
        return hidden.squeeze(0), outputs

class DecoderRNN(nn.Module):
    def __init__(self, tgt_vocab_size, embed_dim, hidden_dim):
        super().__init__()
        self.embedding = nn.Embedding(tgt_vocab_size, embed_dim)
        self.rnn = nn.GRU(embed_dim, hidden_dim, batch_first=True)
        self.fc_out = nn.Linear(hidden_dim, tgt_vocab_size)

    def forward(self, tgt_tokens, hidden_init):
        outputs, _ = self.rnn(self.embedding(tgt_tokens), hidden_init.unsqueeze(0))
        return self.fc_out(outputs)

class Seq2Seq(nn.Module):
    def __init__(self, encoder, decoder):
        super().__init__()
        self.encoder = encoder
        self.decoder = decoder

model = Seq2Seq(EncoderRNN(vocab_size, embedding_dim, hidden_dim),
                DecoderRNN(tgt_vocab_size, embedding_dim, hidden_dim))

# 4. 推理模式下的解码器（贪婪搜索）
def greedy_decode(model, src, max_len=10, start_token=0):
    model.eval()
    with torch.no_grad():
        context, _ = model.encoder(src)          # (batch, hidden)
        batch_size = src.size(0)
        decoder_input = torch.full((batch_size, 1), start_token, dtype=torch.long)
        hidden = context
        all_predictions = []
        for t in range(max_len):
            embedded = model.decoder.embedding(decoder_input)     # (batch, 1, embed)
            output, hidden_state = model.decoder.rnn(embedded, hidden.unsqueeze(0))
            hidden = hidden_state.squeeze(0)                      # (batch, hidden)
            logits = model.decoder.fc_out(output)                 # (batch, 1, vocab)
            next_token = logits.argmax(dim=-1)                    # (batch, 1)
            all_predictions.append(next_token)
            decoder_input = next_token                            # 自回归：上一步输出=这一步输入
        return torch.cat(all_predictions, dim=1)                  # (batch, max_len)

# 模拟一个源序列进行推理
src_single = torch.randint(0, vocab_size, (1, 4))
pred_tokens = greedy_decode(model, src_single, max_len=6, start_token=0)
print("src_single =", src_single.tolist())
print("pred_tokens =", pred_tokens.tolist())

src_single = [[4, 12, 2, 1]]
pred_tokens = [[22, 9, 19, 19, 20, 17]]

注意：模型是随机初始化的、没经过任何训练，所以生成的 6 个词基本是随机的（其中出现了相邻重复，如连续两个 19），没有任何语义。这恰恰说明训练的重要性。

📝 动手练一练

问题：在教师强制训练中，我们通常不会将整个目标序列tgt直接输入解码器，而是输入<sos> + tgt序列去掉最后一个词。请修改上面第2部分解码器的代码，模拟这一过程。假设起始符<sos>的ID为0，并构造相应的decoder_input。

👉 点击查看参考答案

思路：解码器的输入应比目标序列（即标签）在时间上提前一步。我们需要构造一个decoder_input，其第一个位置是起始符0，后续位置是目标序列tgt_data的前tgt_seq_len-1个词。

import torch
import torch.nn as nn

torch.manual_seed(0)

vocab_size, tgt_vocab_size = 20, 25
embedding_dim, hidden_dim = 8, 16
batch_size, src_seq_len, tgt_seq_len = 2, 5, 6

class EncoderRNN(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.rnn = nn.GRU(embed_dim, hidden_dim, batch_first=True)

    def forward(self, src_tokens):
        outputs, hidden = self.rnn(self.embedding(src_tokens))
        return hidden.squeeze(0), outputs

class DecoderRNN(nn.Module):
    def __init__(self, tgt_vocab_size, embed_dim, hidden_dim):
        super().__init__()
        self.embedding = nn.Embedding(tgt_vocab_size, embed_dim)
        self.rnn = nn.GRU(embed_dim, hidden_dim, batch_first=True)
        self.fc_out = nn.Linear(hidden_dim, tgt_vocab_size)

    def forward(self, tgt_tokens, hidden_init):
        outputs, _ = self.rnn(self.embedding(tgt_tokens), hidden_init.unsqueeze(0))
        return self.fc_out(outputs)

encoder = EncoderRNN(vocab_size, embedding_dim, hidden_dim)
decoder = DecoderRNN(tgt_vocab_size, embedding_dim, hidden_dim)
criterion = nn.CrossEntropyLoss()

src_data = torch.randint(0, vocab_size, (batch_size, src_seq_len))
tgt_data = torch.randint(1, tgt_vocab_size, (batch_size, tgt_seq_len))  # 避免0，因为0是<sos>

# 1. 获取上下文向量
context, _ = encoder(src_data)

# 2. 构造正确的解码器输入 (teacher forcing)：[<sos>, y_1, ..., y_{T-1}]
sos_token = 0
decoder_input = torch.cat([
    torch.full((batch_size, 1), sos_token, dtype=torch.long),
    tgt_data[:, :-1]
], dim=1)

# 3. 前向传播
logits = decoder(decoder_input, context)

print("tgt_data (ground truth) =", tgt_data.tolist())
print("decoder_input (teacher forcing) =", decoder_input.tolist())
print("logits.shape =", tuple(logits.shape))
# 计算损失时，输出与完整 tgt_data 对齐
loss = criterion(logits.view(-1, tgt_vocab_size), tgt_data.view(-1))
print("loss with teacher forcing =", round(loss.item(), 4))

tgt_data (ground truth) = [[15, 1, 24, 3, 16, 22], [18, 8, 3, 5, 8, 4]]
decoder_input (teacher forcing) = [[0, 15, 1, 24, 3, 16], [0, 18, 8, 3, 5, 8]]
logits.shape = (2, 6, 25)
loss with teacher forcing = 3.2911

常见错误

错误理解或写法	现象或报错	原因	正确做法
训练和推理使用相同的解码逻辑	推理时输出混乱或重复，模型无法生成合理序列。	训练时使用教师强制（已知真实下一词），推理时需自回归（用上一时刻的预测）。	明确区分`model.train()`和`model.eval()`模式，并为解码器实现独立的`decode`方法。
编码器上下文向量形状不匹配	`RuntimeError: Expected hidden[0] size (1, 2, 16), got (2, 16)`	PyTorch RNN期望初始隐藏状态形状为`(num_layers, batch, hidden_dim)`，但编码器输出是`(batch, hidden_dim)`。	使用`hidden_init.unsqueeze(0)`增加一个维度，或初始化解码器RNN时指定`num_layers`。
忽略变长序列的填充	损失计算时，填充符（如`<pad>`）也参与了计算，降低了有效信号权重。	交叉熵损失对所有的位置一视同仁。	使用`nn.CrossEntropyLoss(ignore_index=pad_idx)`或在计算损失前通过mask将填充位置的贡献置零。
解码器输入未做偏移	模型在训练时“偷看”了当前要预测的词，导致训练与推理差距巨大。	直接将目标序列`tgt`作为解码器输入，相当于给了模型答案。	解码器输入应为`<sos>` + `tgt[:-1]`，确保每一步预测只依赖于之前的词。

本章小结

Seq2Seq模型通过编码器-解码器架构，解决了输入输出序列长度可变且不等的映射问题。
编码器（通常为RNN/LSTM）将整个输入序列编码为一个上下文向量，作为输入序列的语义摘要。
解码器（另一个RNN）以该上下文向量为初始状态，自回归地生成输出序列。
教师强制是训练Seq2Seq模型的关键技巧，它使用真实目标序列作为解码器输入，以加速训练，但需注意输入要相对于目标偏移一步。
基础Seq2Seq的瓶颈在于上下文向量需要承载整个输入序列的信息，对于长序列效果受限，这引出了下一节的核心——注意力机制。

📋 行动清单

运行本节所有代码块，确保理解从编码器、解码器到完整Seq2Seq模型前向传播的每一步张量形状变化。
修改解码器推理代码greedy_decode，添加简单的终止条件（例如，当所有序列都预测出结束符<eos>时停止生成）。
尝试将编码器和解码器中的nn.GRU替换为nn.LSTM，注意LSTM的隐藏状态是一个元组(h_n, c_n)，需要同时传递。