章节4：RNN与序列模型

---

学习目标

• 理解 RNN 循环神经网络原理与隐藏状态传递机制
• 掌握 LSTM 与 GRU 的门控结构及其解决长依赖问题的方法
• 理解 Seq2Seq 架构的 Encoder-Decoder 设计模式
• 完成时间序列预测实战项目

---

4.1 RNN 循环神经网络

4.1.1 基本思想

传统神经网络假设输入之间相互独立，但序列数据（文本、语音、股票价格）具有时间依赖关系。RNN 通过**隐藏状态（Hidden State）**在不同时间步之间传递信息。

4.1.2 RNN 单元结构

在每个时间步 $t$，RNN 单元接收当前输入 $x_t$ 和上一时刻隐藏状态 $h_{t-1}$，计算当前隐藏状态 $h_t$ 和输出 $y_t$。

数学公式：

$$ h_t = \tanh(W_{hh} h_{t-1} + W_{xh} x_t + b_h) $$

$$ y_t = W_{hy} h_t + b_y $$

其中：

• $h_t$：当前隐藏状态
• $h_{t-1}$：上一时刻隐藏状态
• $x_t$：当前输入
• $W_{hh}, W_{xh}, W_{hy}$：权重矩阵

PyTorch 原生 RNN：

import torch
import torch.nn as nn

# 使用 PyTorch 内置 RNN
rnn = nn.RNN(
    input_size=10,   # 每个时间步的输入特征维度
    hidden_size=20,  # 隐藏状态维度
    num_layers=2,    # RNN 层数
    batch_first=True # 输入形状为 (batch, seq_len, feature)
)

x = torch.randn(16, 5, 10)   # (batch=16, seq_len=5, input_size=10)
output, h_n = rnn(x)

print(f"输出形状: {output.shape}")  # [16, 5, 20]
print(f"最终隐藏状态: {h_n.shape}") # [2, 16, 20]

4.1.3 手写简单 RNN 单元

class SimpleRNNCell(nn.Module):
    def __init__(self, input_size, hidden_size):
        super().__init__()
        self.W_xh = nn.Linear(input_size, hidden_size)
        self.W_hh = nn.Linear(hidden_size, hidden_size)
        self.tanh = nn.Tanh()

    def forward(self, x, h_prev):
        return self.tanh(self.W_xh(x) + self.W_hh(h_prev))

class SimpleRNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super().__init__()
        self.rnn_cell = SimpleRNNCell(input_size, hidden_size)
        self.fc = nn.Linear(hidden_size, output_size)
        self.hidden_size = hidden_size

    def forward(self, x):
        batch_size, seq_len, _ = x.shape
        h = torch.zeros(batch_size, self.hidden_size, device=x.device)
        outputs = []
        for t in range(seq_len):
            h = self.rnn_cell(x[:, t, :], h)
            outputs.append(self.fc(h))
        return torch.stack(outputs, dim=1)

model = SimpleRNN(10, 20, 5)
x = torch.randn(16, 5, 10)
y = model(x)
print(f"输出形状: {y.shape}")  # [16, 5, 5]

4.1.4 RNN 的局限

• 梯度消失/爆炸：时间步长时，梯度连乘导致指数级衰减或增长
• 长距离依赖：难以捕捉间隔较远的序列间关系
• 并行性差：必须逐时间步顺序计算

---

4.2 LSTM 与 GRU

4.2.1 LSTM（长短期记忆网络）

LSTM（Hochreiter & Schmidhuber, 1997）通过门控机制和**细胞状态（Cell State）**解决长依赖问题。

核心结构——三个门：

1. 遗忘门（Forget Gate）：决定丢弃哪些历史信息
2. 输入门（Input Gate）：决定写入哪些新信息
3. 输出门（Output Gate）：决定输出哪些信息

数学公式：

$$ \begin{aligned} f_t &= \sigma(W_f \cdot [h_{t-1}, x_t] + b_f) \ i_t &= \sigma(W_i \cdot [h_{t-1}, x_t] + b_i) \ \tilde{C}t &= \tanh(W_C \cdot [h, x_t] + b_C) \ C_t &= f_t \odot C_{t-1} + i_t \odot \tilde{C}t \ o_t &= \sigma(W_o \cdot [h, x_t] + b_o) \ h_t &= o_t \odot \tanh(C_t) \end{aligned} $$

# PyTorch LSTM
lstm = nn.LSTM(
    input_size=10,
    hidden_size=20,
    num_layers=2,
    batch_first=True,
    bidirectional=False
)

x = torch.randn(16, 5, 10)
output, (h_n, c_n) = lstm(x)

print(f"输出: {output.shape}")        # [16, 5, 20]
print(f"隐藏状态: {h_n.shape}")       # [2, 16, 20]
print(f"细胞状态: {c_n.shape}")       # [2, 16, 20]

4.2.2 GRU（门控循环单元）

GRU（Cho et al., 2014）是 LSTM 的简化版本，合并了遗忘门和输入门为更新门，参数量更少、训练更快。

两个门：

1. 重置门（Reset Gate）：控制忽略上一隐藏状态的程度
2. 更新门（Update Gate）：控制上一隐藏状态保留的比例

数学公式：

$$ \begin{aligned} r_t &= \sigma(W_r \cdot [h_{t-1}, x_t] + b_r) \ z_t &= \sigma(W_z \cdot [h_{t-1}, x_t] + b_z) \ \tilde{h}t &= \tanh(W_h \cdot [r_t \odot h, x_t] + b_h) \ h_t &= (1 - z_t) \odot h_{t-1} + z_t \odot \tilde{h}_t \end{aligned} $$

# PyTorch GRU
gru = nn.GRU(
    input_size=10,
    hidden_size=20,
    num_layers=2,
    batch_first=True
)

x = torch.randn(16, 5, 10)
output, h_n = gru(x)

print(f"GRU 输出: {output.shape}")  # [16, 5, 20]
print(f"GRU 隐藏状态: {h_n.shape}") # [2, 16, 20]

4.2.3 RNN vs LSTM vs GRU

特性	RNN	LSTM	GRU
门控数	0	3	2
参数量	少	多	中
长依赖	差	好	好
训练速度	快	慢	中
过拟合风险	低	高	中

---

4.3 Seq2Seq 架构（Encoder-Decoder）

4.3.1 架构概述

Seq2Seq（Sequence-to-Sequence）用于输入和输出都是变长序列的任务，如机器翻译、文本摘要、语音识别。

核心思想：

1. Encoder：读取输入序列，压缩成上下文向量（Context Vector）
2. Decoder：基于上下文向量，逐步生成输出序列

输入：["我", "爱", "你"] → [Encoder] → 上下文向量 → [Decoder] → ["I", "love", "you"]

class Encoder(nn.Module):
    def __init__(self, vocab_size, embed_size, hidden_size, num_layers=1):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_size)
        self.lstm = nn.LSTM(embed_size, hidden_size, num_layers, batch_first=True)

    def forward(self, x):
        embedded = self.embedding(x)
        output, (hidden, cell) = self.lstm(embedded)
        return hidden, cell  # 返回最终隐藏状态给 Decoder


class Decoder(nn.Module):
    def __init__(self, vocab_size, embed_size, hidden_size, num_layers=1):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_size)
        self.lstm = nn.LSTM(embed_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, vocab_size)

    def forward(self, x, hidden, cell):
        embedded = self.embedding(x)
        output, (hidden, cell) = self.lstm(embedded, (hidden, cell))
        prediction = self.fc(output)
        return prediction, hidden, cell


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

    def forward(self, src, trg):
        batch_size, trg_len = trg.shape

        # Encoder 编码源序列
        hidden, cell = self.encoder(src)

        # Decoder 初始输入为 <SOS> token
        decoder_input = trg[:, 0:1]  # 第一个 token（<SOS>）
        outputs = torch.zeros(batch_size, trg_len, self.decoder.fc.out_features)

        for t in range(trg_len):
            decoder_output, hidden, cell = self.decoder(decoder_input, hidden, cell)
            outputs[:, t:t+1, :] = decoder_output
            # 教师强制（Teacher Forcing）：用真实标签作为下一步输入
            decoder_input = trg[:, t:t+1]

        return outputs

4.3.2 Attention 机制（简介）

基础 Seq2Seq 将所有信息压缩到一个固定向量中，长序列时信息丢失严重。Attention 机制让 Decoder 在每个时间步"关注" Encoder 的不同位置。

$$ \text{Attention}(Q, K, V) = \text{softmax}\left(\frac{QK^T}{\sqrt{d_k}}\right)V $$

（Attention 的详细内容将在第5章 Transformer 中深入讲解）

---

4.4 时间序列预测实战

使用 LSTM 预测股票价格（或正弦波等任意时间序列）。

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import matplotlib.pyplot as plt

# 1. 生成合成数据（正弦波 + 噪声）
np.random.seed(42)
t = np.linspace(0, 100, 1000)
data = np.sin(t * 0.1) + np.random.randn(1000) * 0.1

# 2. 构建序列样本（用前 look_back 步预测后一步）
def create_sequences(data, look_back=20):
    X, y = [], []
    for i in range(len(data) - look_back):
        X.append(data[i:i+look_back])
        y.append(data[i+look_back])
    return np.array(X), np.array(y)

look_back = 20
X, y = create_sequences(data, look_back)

# 分割训练/测试
split = int(0.8 * len(X))
X_train, y_train = X[:split], y[:split]
X_test, y_test = X[split:], y[split:]

# 转换为 PyTorch 张量 [samples, seq_len, features]
X_train = torch.FloatTensor(X_train).unsqueeze(-1)
y_train = torch.FloatTensor(y_train).unsqueeze(-1)
X_test = torch.FloatTensor(X_test).unsqueeze(-1)
y_test = torch.FloatTensor(y_test).unsqueeze(-1)

print(f"训练集: {X_train.shape}, 测试集: {X_test.shape}")

# 3. 定义 LSTM 预测模型
class TimeSeriesLSTM(nn.Module):
    def __init__(self, input_size=1, hidden_size=64, num_layers=2):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size, hidden_size, num_layers,
            batch_first=True, dropout=0.2
        )
        self.fc = nn.Linear(hidden_size, 1)

    def forward(self, x):
        # x: [batch, seq_len, 1]
        output, (hidden, cell) = self.lstm(x)
        # 取最后一个时间步的输出预测下一个值
        return self.fc(output[:, -1, :])

model = TimeSeriesLSTM()
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.01)

# 4. 训练
epochs = 100
batch_size = 32
dataset = torch.utils.data.TensorDataset(X_train, y_train)
loader = torch.utils.data.DataLoader(dataset, batch_size=batch_size, shuffle=True)

losses = []
for epoch in range(epochs):
    model.train()
    epoch_loss = 0.0
    for X_batch, y_batch in loader:
        optimizer.zero_grad()
        y_pred = model(X_batch)
        loss = criterion(y_pred, y_batch)
        loss.backward()
        optimizer.step()
        epoch_loss += loss.item()

    if (epoch + 1) % 10 == 0:
        avg_loss = epoch_loss / len(loader)
        losses.append(avg_loss)
        print(f"Epoch {epoch+1}/{epochs}, Loss: {avg_loss:.6f}")

# 5. 预测与可视化
model.eval()
with torch.no_grad():
    train_pred = model(X_train)
    test_pred = model(X_test)

train_pred = train_pred.numpy().flatten()
test_pred = test_pred.numpy().flatten()
y_train_np = y_train.numpy().flatten()
y_test_np = y_test.numpy().flatten()

plt.figure(figsize=(12, 5))

plt.subplot(1, 2, 1)
plt.plot(y_train_np[:200], label='真实值')
plt.plot(train_pred[:200], label='预测值')
plt.title('训练集预测对比')
plt.legend()

plt.subplot(1, 2, 2)
plt.plot(y_test_np[:200], label='真实值')
plt.plot(test_pred[:200], label='预测值')
plt.title('测试集预测对比')
plt.legend()

plt.tight_layout()
plt.show()

# 计算测试集 MSE
test_mse = np.mean((y_test_np - test_pred) ** 2)
print(f"测试集 MSE: {test_mse:.6f}")

---

小结

概念	关键要点
RNN	隐藏状态传递，适合序列数据，存在梯度消失问题
LSTM	遗忘门/输入门/输出门 + 细胞状态，解决长依赖
GRU	重置门/更新门，LSTM 的简化高效替代
Seq2Seq	Encoder-Decoder 架构，用于变长序列转换
时间序列预测	用历史窗口预测未来值，LSTM 效果优于 RNN
Attention	解决信息瓶颈，让 Decoder 关注源序列的特定位置

练习

1. RNN 对比实验：在相同的时间序列数据上对比 RNN、LSTM、GRU 的预测效果（MSE）。
2. 手写 LSTM：不使用 nn.LSTM，仅用 nn.Linear 实现一个 LSTM 单元并验证正确性。
3. 文本生成：用 LSTM 训练一个字符级语言模型（输入：莎士比亚文本，输出：生成类似风格的文本）。
4. Seq2Seq 翻译：实现一个简单的英译中 Seq2Seq 模型（使用小规模平行语料）。
5. 超参数调优：改变 look_back 窗口大小和 hidden_size，观察对时间序列预测精度的影响。