RNN 的高级用法

至此，你已经学习了以下内容：

• RNN的概念及其工作原理；
• LSTM 的概念，以及为什么它在长序列上的效果比普通 RNN更好；
• 如何使用 Keras 的 RNN层来处理序列数据。
  接下来，我们将介绍RNN 的几个高级功能，它们有助于你充分利用深度学习序列模型。学
  完本节，你将掌握用Keras 实现RNN 的大部分知识。

本节将介绍以下内容。

• 循环dropout（recurrent dropout）：这是dropout的一种变体，用于在循环层中降低过拟合。
• 循环层堆叠（stacking recurrent layers）：这会提高模型的表示能力（代价是更大的计算量）。
• 双向循环层（bidirectional recurrent layer）：它会将相同的信息以不同的方式呈现给RNN，
  可以提高精度并缓解遗忘问题。
  我们将使用这3 种方法来完善温度预测RNN。

利用循环dropout 降低过拟合

我们回头来看10.2.5 节中的基于LSTM 的模型，它是第一个能够超越常识基准的模型。观
察这个模型的训练曲线和验证曲线（图10-5），可以明显看出，尽管模型只有很少的单元，但很快就出现过拟合，训练损失和验证损失在几轮过后就开始明显偏离。你已经熟悉了降低过拟合
的经典方法——dropout，即让某一层的输入单元随机为0，其目的是破坏该层训练数据中的偶
然相关性。但如何在RNN 中正确使用dropout，并不是一个简单的问题。

人们早就知道，在循环层之前使用dropout 会妨碍学习过程，而不会有助于正则化。2016 年，
Yarin Gal 在他关于贝叶斯深度学习的博士论文a 中，确定了在RNN 中使用dropout 的正确方式：
在每个时间步都应该使用相同的dropout 掩码（相同模式的舍弃单元），而不是在不同的时间步
使用随机变化的dropout 掩码。此外，为了对GRU 和LSTM 等层的循环门得到的表示做正则化，
还应该对该层的内部循环激活应用一个不随时间变化的dropout 掩码（循环dropout 掩码）。在
每个时间步使用相同的dropout 掩码，可以让神经网络沿着时间传播其学习误差，而随时间随机
变化的dropout 掩码则会破坏这个误差信号，不利于学习过程。

Yarin Gal 使用Keras 开展这项研究，并帮助将这一机制直接内置于Keras 循环层中。Keras
中的每个循环层都有两个与dropout 相关的参数：一个是dropout，它是一个浮点数，指定该
层输入单元的dropout 比率；另一个是recurrent_dropout，指定循环单元的dropout 比率。
对于第一个LSTM 示例，我们向LSTM 层中添加循环dropout，看一下它对过拟合的影响，如代
码清单10-22 所示。

由于使用了dropout，我们不需要过分依赖网络尺寸来进行正则化，因此我们将使用具有两
倍单元个数的LSTM 层，希望它的表示能力更强（如果不使用dropout，这个网络会马上开始过
拟合，你可以试试看）。由于使用dropout 正则化的网络总是需要更长时间才能完全收敛，因此
我们将模型训练轮数设为原来的5 倍。

代码清单10-22 训练并评估一个使用dropout 正则化的LSTM 模型

模型结果如图10-11 所示。成功！模型在前20 轮中不再过拟合。验证MAE 低至2.27 摄氏
度（比不使用机器学习的基准改进了7%），测试MAE 为2.45 摄氏度（比基准改进了6.5%），
还不错。

循环层堆叠

模型不再过拟合，但似乎遇到了性能瓶颈，所以我们应该考虑增加神经网络的容量和表示
能力。回想一下机器学习的通用工作流程，增大模型容量通常是好的做法，直到过拟合成为主要
障碍（假设你已经采取了基本措施来降低过拟合，比如使用dropout）。只要过拟合不是太严重，
那么模型就很可能容量不足。

增加网络容量的通常做法是增加每层单元个数或添加更多的层。循环层堆叠是构建更加强
大的循环网络的经典方法，比如，不久之前谷歌翻译算法就是7 个大型LSTM 层的堆叠——这
个模型很大。

在Keras 中堆叠循环层，所有中间层都应该返回完整的输出序列（一个3 阶张量），而不
是只返回最后一个时间步的输出。前面说过，这可以通过指定return_sequences=True 来
实现。

在下面这个示例中，我们尝试堆叠两个使用dropout 正则化的循环层，如代码清单10-23
所示。不同的是，我们将使用门控循环单元（gated recurrent unit，GRU）层代替LSTM 层。
GRU 与LSTM 非常类似，你可以将其看作LSTM 架构的精简版本。它由Kyunghyun Cho 等人
于2014 年提出，当时RNN 刚刚开始在不大的研究群体中重新引起人们的兴趣。

代码清单10-23 训练并评估一个使用dropout 正则化的堆叠GRU 模型

模型结果如图10-12 所示。测试MAE 为2.39 摄氏度（比基准改进了8.8%）。可以看到，增
加一层确实对结果有所改进，但效果并不明显。此时你可能会发现，增加网络容量的回报在逐
渐减小。

使用双向RNN

本节介绍的最后一种方法是双向RNN（bidirectional RNN）。双向RNN 是一种常见的RNN
变体，它在某些任务上的性能比普通RNN 更好。它常用于自然语言处理，可谓深度学习对自然
语言处理的“瑞士军刀”。

RNN 特别依赖于顺序，它按顺序处理输入序列的时间步，而打乱时间步或反转时间步会完
全改变RNN 从序列中提取的表示。正是由于这个原因，如果顺序对问题很重要（比如温度预
测问题），那么RNN 的表现就会很好。双向RNN 利用了RNN 的顺序敏感性：它包含两个普通
RNN（比如前面介绍过的GRU 层和LSTM 层），每个RNN 分别沿一个方向对输入序列进行处
理（按时间正序和按时间逆序），然后将它们的表示合并在一起。通过沿着两个方向处理序列，
双向RNN 能够捕捉到可能被单向RNN 忽略的模式。

值得注意的是，本节所有RNN 层都按时间正序处理序列（较早的时间步在前），这个决定
可能有些随意。至少到目前为止，我们还没有试着去质疑这个决定。如果RNN 按时间逆序处理
输入序列（较晚的时间步在前），能否表现得足够好呢？我们来试一下，看看会发生什么。你只
需编写一个数据生成器的变体，将输入序列沿着时间维度反转（将最后一行代码替换为yield
samples[:, ::-1, :], targets）a。10.2.5 节中的第一个示例用的是基于LSTM 的模型，我
们训练一个与之相同的模型，得到的结果如图10-13 所示。

逆序LSTM 的性能甚至比基于常识的基准还要差很多，这说明在本例中，按时间正序处理
对成功解决问题很重要。这非常合理：底层的LSTM 层通常更善于记住最近的过去，而不是遥
远的过去。对这个问题而言，更晚的天气数据点当然比更早的天气数据点具有更强的预测能力
（这也是基于常识的基准非常强大的原因）。因此，按时间正序的层必然比按时间逆序的层表现
要好。

然而，对于许多其他问题（包括自然语言），情况并非如此：直觉上看，一个单词对理解句
子的重要性，通常并不取决于它在句子中的位置。对于文本数据，逆序处理的效果与正序处理
一样好——你可以倒着阅读文本（试试吧）。虽然单词顺序对理解语言很重要，但使用哪种顺序
并不重要。

重要的是，在逆序序列上训练的RNN 学到的表示不同于在原始序列上训练学到的表示，正
如在现实世界中，如果时间倒流（你的人生是第一天死亡、最后一天出生），那么你的心智模型
也会完全不同。在机器学习中，如果一种数据表示不同但有用，那么总是值得加以利用，并且
这种表示与其他表示的差异越大越好。它提供了观察数据的全新角度，可以捕捉到数据中被其
他方法忽略的内容，因此有助于提高模型在某个任务上的性能。这正是集成（ensembling）方法
背后的原理，第13 章将介绍这一概念。

双向RNN 正是利用这一想法来提高正序RNN 的性能。它从两个方向查看输入序列（如图
10-14 所示），从而得到更加丰富的表示，并捕捉到仅使用正序RNN 时可能忽略的一些模式。

要想在Keras 中将双向RNN 实例化，你可以使用Bidirectional 层，它的第一个参数是
一个循环层实例。Bidirectional 会对这个循环层创建第二个单独的实例，然后用一个实例按
正序处理输入序列，用另一个实例按逆序处理输入序列。你可以在温度预测任务上试试这种方
法，如代码清单10-24 所示。

代码清单10-24 训练并评估双向LSTM

可以看到，双向LSTM 的性能不如普通LSTM 层。原因很容易理解：所有的预测能力肯定
都来自于正序的那一半网络，因为我们已经知道逆序的那一半网络在这项任务上的性能很差（再
次强调，这是因为在本例中，最近的过去比遥远的过去更重要）。同时，逆序的那一半网络使网
络容量加倍，从而导致更早开始出现过拟合。

然而，双向RNN 非常适合用于文本数据或任何其他类型的数据，其中顺序很重要，但使用
哪种顺序并不重要。事实上，在2016 年的一段时间里，双向LSTM 被认为是许多自然语言处理
任务中最先进的方法（这是在Transformer 架构兴起之前，第11 章会介绍这种架构）

进一步实验

为了提高模型在温度预测问题上的性能，你还可以尝试下面这些做法。

• 调节堆叠循环层中每层的单元个数和dropout比率。当前取值在很大程度上是随意选择的，
  因此可能不是最优的。
• 调节 RMSprop优化器的学习率，或者尝试使用其他优化器。
• 在循环层上使用 Dense层堆叠作为回归器，而不是单一的Dense层。
• 改进模型输入：尝试使用更长或更短的序列，或者尝试使用不同的采样率，再或者做特
  征工程。

如前所述，深度学习更像是一门艺术，而不是一门科学。我们可以提供指导，对于某个问
题哪些方法可能有效、哪些方法可能无效，但归根结底，每个数据集都是独一无二的，你必须
根据经验来评估不同的策略。目前没有任何理论能够提前准确地告诉你，怎样做才能最优地解
决问题。你必须不断迭代。

根据我的经验，在没有使用机器学习的基准上改进10% 左右，可能是你在这个数据集上能
得到的最佳结果。这不算很好，但这些结果是合理的：如果你能获得来自不同地点（范围很广）
的数据，那么近期的天气是高度可预测的，但如果你只有单一地点的测量结果，那么近期天气
就很难预测。你所在地点的天气演变，取决于周围地区当前的天气模式。

完整代码

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from tensorflow import keras
from tensorflow.keras import layers
import os


# ====================
# 1. 创建模拟数据
# ====================

def create_simulated_data(n_samples=500000, n_features=14):
    """创建模拟的耶拿气候数据"""
    print("正在创建模拟数据...")

    # 创建一个时间序列的基础模式（温度）
    time = np.arange(n_samples)
    base_temp = 10 + 10 * np.sin(2 * np.pi * time / (24 * 6 * 365))  # 年周期
    daily_variation = 5 * np.sin(2 * np.pi * time / (24 * 6))  # 日周期
    noise = np.random.normal(0, 2, n_samples)  # 随机噪声

    temperature = base_temp + daily_variation + noise

    # 创建其他特征（与温度相关）
    data = np.zeros((n_samples, n_features))

    # 第一列：日期时间（模拟）
    data[:, 0] = time / (24 * 6 * 365) + 2009  # 从2009年开始

    # 第二列：温度（主要特征）
    data[:, 1] = temperature

    # 其他特征：创建与温度相关的其他气象数据
    for i in range(2, n_features):
        # 每个特征都有一些与温度的相关性，加上一些噪声
        correlation = np.random.uniform(-0.8, 0.8)
        noise_level = np.random.uniform(0.1, 0.5)
        data[:, i] = correlation * temperature + np.random.normal(0, noise_level, n_samples)

    # 创建特征名称（模拟原始数据集）
    column_names = [
        'Date Time', 'T (degC)', 'p (mbar)', 'Tpot (K)', 'Tdew (degC)',
        'rh (%)', 'VPmax (mbar)', 'VPact (mbar)', 'VPdef (mbar)', 'sh (g/kg)',
        'H2OC (mmol/mol)', 'rho (g/m**3)', 'wv (m/s)', 'max. wv (m/s)'
    ]

    # 确保列名数量匹配
    if len(column_names) > n_features:
        column_names = column_names[:n_features]
    elif len(column_names) < n_features:
        # 如果特征更多，添加通用名称
        for i in range(len(column_names), n_features):
            column_names.append(f'feature_{i}')

    df = pd.DataFrame(data, columns=column_names)

    print(f"模拟数据创建完成：{n_samples}个样本，{n_features}个特征")
    return df


# ====================
# 2. 加载数据
# ====================

# 检查是否存在真实数据文件
real_data_path = "jena_climate_2009_2016.csv"
use_real_data = False

if os.path.exists(real_data_path):
    print(f"找到真实数据文件: {real_data_path}")
    try:
        df = pd.read_csv(real_data_path)
        use_real_data = True
        print("成功加载真实数据")
    except Exception as e:
        print(f"加载真实数据失败: {e}")
        print("将使用模拟数据")
        df = create_simulated_data(n_samples=100000, n_features=14)
else:
    print(f"未找到真实数据文件: {real_data_path}")
    print("将使用模拟数据")
    df = create_simulated_data(n_samples=100000, n_features=14)

# ====================
# 3. 数据预处理
# ====================

print(f"\n数据形状: {df.shape}")
print(f"数据列名: {df.columns.tolist()}")

# 提取温度数据（第二列）
if 'T (degC)' in df.columns:
    temperature_col = 'T (degC)'
elif df.shape[1] >= 2:
    temperature_col = df.columns[1]
else:
    temperature_col = df.columns[0]

print(f"\n使用 '{temperature_col}' 作为温度列")

temperature = df[temperature_col].values

# 数据标准化
train_size = min(200000, len(temperature) // 2)
mean = temperature[:train_size].mean()
std = temperature[:train_size].std()
temperature = (temperature - mean) / std

print(f"数据标准化: mean={mean:.2f}, std={std:.2f}")

# 准备完整数据（多变量）
if use_real_data:
    raw_data = df.values
else:
    # 对于模拟数据，使用所有列
    raw_data = df.values

# 标准化所有特征
if len(raw_data) > train_size:
    data_mean = raw_data[:train_size].mean(axis=0)
    data_std = raw_data[:train_size].std(axis=0)
    # 避免除零
    data_std[data_std == 0] = 1.0
    raw_data = (raw_data - data_mean) / data_std
else:
    # 如果数据太少，使用整体统计
    data_mean = raw_data.mean(axis=0)
    data_std = raw_data.std(axis=0)
    data_std[data_std == 0] = 1.0
    raw_data = (raw_data - data_mean) / data_std

print(f"多变量数据形状: {raw_data.shape}")


# ====================
# 4. 创建数据生成器
# ====================

def generator(data, lookback, delay, min_index, max_index,
              shuffle=False, batch_size=128, step=6):
    """生成时间序列批数据"""
    if max_index is None:
        max_index = len(data) - delay - 1
    i = min_index + lookback

    while True:
        if shuffle:
            rows = np.random.randint(
                min_index + lookback, max_index, size=batch_size
            )
        else:
            if i + batch_size >= max_index:
                i = min_index + lookback
            rows = np.arange(i, min(i + batch_size, max_index))
            i += len(rows)

        samples = np.zeros((len(rows), lookback // step, data.shape[-1]))
        targets = np.zeros((len(rows),))

        for j, row in enumerate(rows):
            indices = range(rows[j] - lookback, rows[j], step)
            samples[j] = data[indices]
            targets[j] = data[rows[j] + delay][1]  # 温度是第二列

        yield samples, targets


# 参数设置
lookback = 720  # 5天（5*24*6） - 使用较小的值以便快速训练
step = 6  # 每小时一个数据点
delay = 144  # 24小时后
batch_size = 64  # 使用较小的批次以便快速训练

# 数据集划分
data_len = len(raw_data)
train_end = int(data_len * 0.5)
val_end = int(data_len * 0.75)

print(f"\n数据划分: 训练集(0-{train_end}), 验证集({train_end}-{val_end}), 测试集({val_end}-{data_len})")

train_gen = generator(
    raw_data,
    lookback=lookback,
    delay=delay,
    min_index=0,
    max_index=train_end,
    shuffle=True,
    step=step,
    batch_size=batch_size
)

val_gen = generator(
    raw_data,
    lookback=lookback,
    delay=delay,
    min_index=train_end,
    max_index=val_end,
    step=step,
    batch_size=batch_size
)

test_gen = generator(
    raw_data,
    lookback=lookback,
    delay=delay,
    min_index=val_end,
    max_index=None,
    step=step,
    batch_size=batch_size
)

# 计算每个数据集的步数
val_steps = max(1, (val_end - train_end - lookback) // batch_size)
test_steps = max(1, (data_len - val_end - lookback) // batch_size)

print(f"验证步数: {val_steps}, 测试步数: {test_steps}")


# ====================
# 5. 基准模型（朴素预测）
# ====================

def evaluate_naive_method():
    batch_maes = []
    for step in range(min(10, val_steps)):  # 只评估前10个批次以节省时间
        samples, targets = next(val_gen)
        # 预测最后一个时间步的温度
        preds = samples[:, -1, 1]
        mae = np.mean(np.abs(preds - targets))
        batch_maes.append(mae)
    print(f'基准MAE: {np.mean(batch_maes):.4f}')


evaluate_naive_method()


# ====================
# 6. 创建和训练模型
# ====================

def create_and_train_model(model_type='lstm_dropout', epochs=5):
    """创建和训练指定类型的模型"""
    print(f"\n=== 训练 {model_type} 模型 ===")

    if model_type == 'lstm_dropout':
        model = keras.Sequential([
            layers.LSTM(16, dropout=0.2, recurrent_dropout=0.2,
                        input_shape=(None, raw_data.shape[-1])),
            layers.Dense(1)
        ])
    elif model_type == 'stacked_gru':
        model = keras.Sequential([
            layers.GRU(16, dropout=0.1, recurrent_dropout=0.5,
                       return_sequences=True,
                       input_shape=(None, raw_data.shape[-1])),
            layers.GRU(32, dropout=0.1, recurrent_dropout=0.5),
            layers.Dense(1)
        ])
    elif model_type == 'bidirectional':
        model = keras.Sequential([
            layers.Bidirectional(
                layers.LSTM(16),
                input_shape=(None, raw_data.shape[-1])
            ),
            layers.Dense(1)
        ])
    else:
        raise ValueError(f"未知的模型类型: {model_type}")

    model.compile(
        optimizer=keras.optimizers.RMSprop(learning_rate=0.001),
        loss='mae',
        metrics=['mae']
    )

    # 使用较少的训练步数以加快训练
    history = model.fit(
        train_gen,
        steps_per_epoch=50,
        epochs=epochs,
        validation_data=val_gen,
        validation_steps=min(10, val_steps),
        verbose=1
    )

    return model, history


# 训练所有模型
models = {}
histories = {}

for model_type in ['lstm_dropout', 'stacked_gru', 'bidirectional']:
    model, history = create_and_train_model(model_type, epochs=5)
    models[model_type] = model
    histories[model_type] = history

# ====================
# 7. 可视化结果
# ====================

fig, axes = plt.subplots(1, 3, figsize=(15, 4))

for idx, (model_type, history) in enumerate(histories.items()):
    ax = axes[idx]
    ax.plot(history.history['mae'], label='训练MAE')
    ax.plot(history.history['val_mae'], label='验证MAE')
    ax.set_title(model_type)
    ax.set_xlabel('Epochs')
    ax.set_ylabel('MAE')
    ax.legend()
    ax.grid(True)

plt.suptitle('不同RNN架构在模拟温度数据上的表现', fontsize=14)
plt.tight_layout()
plt.show()

# ====================
# 8. 评估所有模型
# ====================

print("\n=== 模型评估结果 ===")
print("=" * 40)

for model_type, model in models.items():
    # 在测试集上评估
    test_loss, test_mae = model.evaluate(
        test_gen, steps=min(10, test_steps), verbose=0
    )
    print(f"{model_type:15s} - 测试MAE: {test_mae:.4f}")

print("=" * 40)

# ====================
# 9. 进行预测示例
# ====================

print("\n=== 预测示例 ===")

# 获取一个测试批次
samples, true_values = next(test_gen)

# 使用每个模型进行预测
for model_type, model in models.items():
    predictions = model.predict(samples[:5], verbose=0)  # 只预测前5个样本
    print(f"\n{model_type} 预测示例:")
    for i in range(min(3, len(predictions))):
        print(f"  样本 {i + 1}: 真实值={true_values[i]:.4f}, 预测值={predictions[i][0]:.4f}")

# ====================
# 10. 模型结构总结
# ====================

print("\n=== 模型结构总结 ===")
for model_type, model in models.items():
    print(f"\n{model_type}:")
    model.summary()

print("\n训练完成！")