通俗《深度学习》—RNN卷积神经网络原理及实现

暗恤

188人浏览 · 2026-07-10 00:48:09

暗恤 · 2026-07-10 00:48:09 发布

🧱 第一步：从"普通网络"说起

为了让 RNN 变得好理解，我们先聊聊你已经可能知道的东西：普通的神经网络（也叫做前馈神经网络，Feedforward Neural Network）。

想象一个标准的神经网络：

输入 → 隐藏层 → 输出
数据从输入端流入，一路向前，最终输出结果

普通神经网络记不住上下文。你输入"我"，它处理"我"，然后忘了；接着输入"今天"，它又从零开始处理"今天"。就像一个人得了"瞬时失忆症"，每看一个字就把前面的话全忘了。

🔁 第二步：RNN 的核心思想——"记忆小纸条"

现在想象一个场景：

你在读一本小说，读到第 100 页时，你需要记住第 1 页的主角是谁吗？ —— 当然需要！否则你根本读不懂故事。

RNN 就是为了解决这个问题而生的。它做了一件很简单但很聪明的事：

RNN 在自己的"脑中"留了一张「小纸条」，每读一个新词，就把之前的关键信息记在这张小纸条上，然后传递给下一步的自己。

让我用一个具体的例子来说明：

假设网络要处理 "我喜欢吃苹果" 这句话：

第1步：看到 "我"  →  小纸条上写："我"第2步：看到 "喜欢" →  结合纸条上的"我" → 知道是"我喜欢" → 更新纸条第3步：看到 "吃"   →  结合纸条上"我喜欢" → 知道是"我喜欢吃" → 更新纸条第4步：看到 "苹果" →  结合纸条上"我喜欢吃" → 知道是"我喜欢吃苹果" → 给出结论

这个机制叫做 "循环"（Recurrent）——网络每一层的输出，会作为一部分输入再次传入自己，形成一个循环。

💡 用类比加深理解

想象你在接力赛跑：

普通网络：每个运动员跑到终点就结束了，不传递任何信息给下一个人。
RNN：每个运动员跑完一段，把接力棒（小纸条/记忆）交给下一个运动员，下一个运动员带着这个接力棒继续跑。

这个"接力棒"在 RNN 中叫做「隐藏状态」（hidden state, 通常用 h 表示）。

🤔 检查一下理解

让我出一个小问题检验一下：

一个 RNN 在处理「今天天气真不错」这句话时，当它看到最后一个词「错」时，它的小纸条（隐藏状态）里最可能包含了什么信息？

只有「错」只记住了当前看到的最后一个字
整句话的浓缩信息从「今天」到「错」整个句子的上下文信息
什么都没有完全空白，和普通网络一样

我相信你会准确选择第二个选项，说明你理解了最核心的概念！是的，当 RNN 处理到最后一个词时，它的隐藏状态（小纸条）里浓缩了从开始到现在的全部上下文信息——虽然信息可能会有一定的"稀释"，但理论上包含了整个句子的精华。

📐 第三步：稍微看一下"数学直觉"

接下来，我们看一个稍微技术一点的层面。

这个"小纸条"（隐藏状态 htht）的更新公式其实很简单：

ht=tanh(当前输入+上一步的小纸条)

翻译成大白话：

新的小纸条 = 对「当前看到的新内容 + 以前记下的内容」做一个整合与压缩

这里的 tanh⁡ 像一个"挤压器"，观察函数图像可知把值控制在 -1 到 1 之间，防止数字越滚越大炸掉。

你不需要现在就能写出这个公式，只是让你知道——这个机制数学上也很简单优雅，并不神秘。

🗺️ 到目前为止，我们学了什么？

让我帮你整理一下今天已经搭建的知识结构：

接下来，使用PyTorch实现RNN对正弦波的预测的一个小实验，

第一步，生成正弦波数据

np.random.seed(42)
torch.manual_seed(42)

SEQ_LEN = 50  # 输入序列长度
N_SAMPLES = 2000  # 总样本数

# 生成带噪声的正弦波
t = np.linspace(0, 40 * np.pi, N_SAMPLES + SEQ_LEN)
wave = np.sin(t) + 0.03 * np.random.randn(N_SAMPLES + SEQ_LEN)

# 构造样本: 用前 SEQ_LEN 个点 → 预测下一个点
X = np.array([wave[i:i + SEQ_LEN] for i in range(N_SAMPLES)])
y = np.array([wave[i + SEQ_LEN] for i in range(N_SAMPLES)])

# 转换为 PyTorch 张量
# X shape: (样本数, 序列长度, 特征维度=1)
X = torch.tensor(X, dtype=torch.float32).unsqueeze(-1)
y = torch.tensor(y, dtype=torch.float32)

# 拆分 (80% 训练, 20% 测试)
split = int(0.8 * N_SAMPLES)
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]

第二步，定义RNN模型

class RNNPredictor(nn.Module):
    """
    用 PyTorch 内置 RNN 层做时间序列预测
    结构: Embed → RNN → Linear → 输出
    """

    def __init__(self, input_size=1, hidden_size=32, output_size=1):
        # 调用父类初始化
        super().__init__()
        # 创建 RNN 层: input_size=1(单特征), hidden_size=32(隐藏维度), batch_first=True(输入格式为batch在前)
        self.rnn = nn.RNN(
            input_size=input_size,
            hidden_size=hidden_size,
            batch_first=True
        )
        # 创建全连接层: 将隐藏状态维度映射到输出维度(1个预测值)
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        # 前向传播: x shape: (batch, seq_len, 1)
        # RNN 前向计算: output存储所有时间步的隐藏状态, hidden是最后一个时间步的隐藏状态
        output, hidden = self.rnn(x)
        # 取最后一个隐藏状态用于预测: hidden[-1] shape: (batch, hidden_size)
        last_hidden = hidden[-1]
        # 全连接层映射到预测值: (batch, hidden_size) → (batch, 1)
        out = self.fc(last_hidden)
        # 去掉多余维度: (batch, 1) → (batch,)
        return out.squeeze()


# 创建模型
INPUT_SIZE = 1
HIDDEN_SIZE = 32
OUTPUT_SIZE = 1

model = RNNPredictor(INPUT_SIZE, HIDDEN_SIZE, OUTPUT_SIZE)

第三步，训练模型

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = model.to(device)
X_train, X_test = X_train.to(device), X_test.to(device)
y_train, y_test = y_train.to(device), y_test.to(device)

# 损失函数: 均方误差 (MSE)
criterion = nn.MSELoss()
# 优化器: Adam (自动调整学习率)
optimizer = optim.Adam(model.parameters(), lr=0.01)

EPOCHS = 300
train_losses = []
test_losses = []

# 循环训练 EPOCHS 轮
for epoch in range(EPOCHS):
    # ---- 训练阶段 ----
    # 设置模型为训练模式(启用dropout/batchnorm等)
    model.train()
    # 清空梯度(防止梯度累积)
    optimizer.zero_grad()
    # 前向传播: 通过模型得到训练集预测值
    predictions = model(X_train)
    # 计算损失: 预测值与真实值的均方误差
    loss = criterion(predictions, y_train)
    # 反向传播: 计算梯度
    loss.backward()
    # 优化器更新参数: 根据梯度调整权重
    optimizer.step()

    # ---- 评估阶段 ----
    # 设置模型为评估模式(关闭dropout/batchnorm等)
    model.eval()
    # 禁用梯度计算(节省内存,加速推理)
    with torch.no_grad():
        # 前向传播测试集
        test_pred = model(X_test)
        # 计算测试集损失
        test_loss = criterion(test_pred, y_test)

    # 记录训练损失和测试损失
    train_losses.append(loss.item())
    test_losses.append(test_loss.item())

    # 每50轮打印一次训练进度
    if (epoch + 1) % 50 == 0:
        print(f"  第 {epoch + 1:3d} 轮 | 训练损失: {loss.item():.6f} | 测试损失: {test_loss.item():.6f}")

第四步，进行自回归预测测试

model.eval()
with torch.no_grad():
    # 4a. 单步预测效果
    test_pred = model(X_test).cpu().numpy()
    y_test_np = y_test.cpu().numpy()
    mse = np.mean((test_pred - y_test_np) ** 2)
    print(f"  单步预测 MSE: {mse:.6f}")

    # 4b. 自回归预测未来 100 步
    # 取第一个测试样本作为种子序列: shape (1, 50, 1)
    x_seed = X_test[0:1]
    # 预测未来的步数
    future_steps = 100
    # 存储预测结果的列表
    predictions = []

    # 先用种子序列通过RNN得到初始隐藏状态
    output, hidden = model.rnn(x_seed)
    # 当前输入设为种子序列的最后一个时间步
    curr_input = x_seed[:, -1:, :]
    # 当前隐藏状态
    curr_h = hidden

    # 循环预测未来每一步
    for _ in range(future_steps):
        # RNN 单步前向: 输入当前值和隐藏状态,输出新的隐藏状态
        output, curr_h = model.rnn(curr_input, curr_h)
        # 全连接层: 将最后一个隐藏状态映射到预测值
        pred = model.fc(output[:, -1, :])
        # 将预测值添加到结果列表
        predictions.append(pred.item())
        # 将预测值作为下一步的输入: reshape为(1, 1, 1)格式
        curr_input = pred.reshape(1, 1, 1)

    # 获取真实的未来值用于对比
    true_future = wave[split + SEQ_LEN: split + SEQ_LEN + future_steps]

最后进行可视化，完整代码我放在文末供大家取用，展示一下最后的结果

🔬 你能做的实验（改一行参数看效果变化）

改什么参数	预期效果
`HIDDEN_SIZE = 8`	记忆容量变小，预测变差
`HIDDEN_SIZE = 128`	记忆容量变大，可能过拟合
`SEQ_LEN = 10`	输入太短，难以捕捉波形周期
`SEQ_LEN = 200`	输入太长，训练变慢但预测更准
`lr = 0.1`	学习率太大，损失爆炸
`lr = 0.0001`	学习率太小，收敛极慢

本实验完整代码如下：

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

# 设置字体为SimHei（黑体），解决中文显示问题
plt.rcParams['font.sans-serif'] = ['SimHei']
# 解决负号显示为方块的问题
plt.rcParams['axes.unicode_minus'] = False

# ================================================================
#  1. 生成数据
# ================================================================
print("\n[1/4] 生成正弦波数据...")

np.random.seed(42)
torch.manual_seed(42)

SEQ_LEN = 50  # 输入序列长度
N_SAMPLES = 2000  # 总样本数

# 生成带噪声的正弦波
t = np.linspace(0, 40 * np.pi, N_SAMPLES + SEQ_LEN)
wave = np.sin(t) + 0.03 * np.random.randn(N_SAMPLES + SEQ_LEN)

# 构造样本: 用前 SEQ_LEN 个点 → 预测下一个点
X = np.array([wave[i:i + SEQ_LEN] for i in range(N_SAMPLES)])
y = np.array([wave[i + SEQ_LEN] for i in range(N_SAMPLES)])

# 转换为 PyTorch 张量
# X shape: (样本数, 序列长度, 特征维度=1)
X = torch.tensor(X, dtype=torch.float32).unsqueeze(-1)
y = torch.tensor(y, dtype=torch.float32)

# 拆分 (80% 训练, 20% 测试)
split = int(0.8 * N_SAMPLES)
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]

print(f"  训练集: {len(X_train)} 条")
print(f"  测试集: {len(X_test)} 条")
print(f"  任务: 用前 {SEQ_LEN} 个点 → 预测第 {SEQ_LEN + 1} 个点")

# ================================================================
#  2. 定义 RNN 模型
# ================================================================
print("\n[2/4] 定义 RNN 模型...")


class RNNPredictor(nn.Module):
    """
    用 PyTorch 内置 RNN 层做时间序列预测
    结构: Embed → RNN → Linear → 输出
    """

    def __init__(self, input_size=1, hidden_size=32, output_size=1):
        # 调用父类初始化
        super().__init__()
        # 创建 RNN 层: input_size=1(单特征), hidden_size=32(隐藏维度), batch_first=True(输入格式为batch在前)
        self.rnn = nn.RNN(
            input_size=input_size,
            hidden_size=hidden_size,
            batch_first=True
        )
        # 创建全连接层: 将隐藏状态维度映射到输出维度(1个预测值)
        self.fc = nn.Linear(hidden_size, output_size)

    def forward(self, x):
        # 前向传播: x shape: (batch, seq_len, 1)
        # RNN 前向计算: output存储所有时间步的隐藏状态, hidden是最后一个时间步的隐藏状态
        output, hidden = self.rnn(x)
        # 取最后一个隐藏状态用于预测: hidden[-1] shape: (batch, hidden_size)
        last_hidden = hidden[-1]
        # 全连接层映射到预测值: (batch, hidden_size) → (batch, 1)
        out = self.fc(last_hidden)
        # 去掉多余维度: (batch, 1) → (batch,)
        return out.squeeze()


# 创建模型
INPUT_SIZE = 1
HIDDEN_SIZE = 32
OUTPUT_SIZE = 1

model = RNNPredictor(INPUT_SIZE, HIDDEN_SIZE, OUTPUT_SIZE)
print(f"  模型: RNN(hidden={HIDDEN_SIZE}) → Linear(1)")
print(f"  参数量: {sum(p.numel() for p in model.parameters()):,}")

# ================================================================
#  3. 训练
# ================================================================
print("\n[3/4] 训练中 (300轮)...")

device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = model.to(device)
X_train, X_test = X_train.to(device), X_test.to(device)
y_train, y_test = y_train.to(device), y_test.to(device)

# 损失函数: 均方误差 (MSE)
criterion = nn.MSELoss()
# 优化器: Adam (自动调整学习率)
optimizer = optim.Adam(model.parameters(), lr=0.01)

EPOCHS = 300
train_losses = []
test_losses = []

# 循环训练 EPOCHS 轮
for epoch in range(EPOCHS):
    # ---- 训练阶段 ----
    # 设置模型为训练模式(启用dropout/batchnorm等)
    model.train()
    # 清空梯度(防止梯度累积)
    optimizer.zero_grad()
    # 前向传播: 通过模型得到训练集预测值
    predictions = model(X_train)
    # 计算损失: 预测值与真实值的均方误差
    loss = criterion(predictions, y_train)
    # 反向传播: 计算梯度
    loss.backward()
    # 优化器更新参数: 根据梯度调整权重
    optimizer.step()

    # ---- 评估阶段 ----
    # 设置模型为评估模式(关闭dropout/batchnorm等)
    model.eval()
    # 禁用梯度计算(节省内存,加速推理)
    with torch.no_grad():
        # 前向传播测试集
        test_pred = model(X_test)
        # 计算测试集损失
        test_loss = criterion(test_pred, y_test)

    # 记录训练损失和测试损失
    train_losses.append(loss.item())
    test_losses.append(test_loss.item())

    # 每50轮打印一次训练进度
    if (epoch + 1) % 50 == 0:
        print(f"  第 {epoch + 1:3d} 轮 | 训练损失: {loss.item():.6f} | 测试损失: {test_loss.item():.6f}")

# 训练完成,打印最终测试损失
print(f"\n  ✅ 训练完成！最终测试损失: {test_losses[-1]:.6f}")

# ================================================================
#  4. 测试 —— 自回归预测未来
# ================================================================
print("\n[4/4] 自回归预测测试...")

model.eval()
with torch.no_grad():
    # 4a. 单步预测效果
    test_pred = model(X_test).cpu().numpy()
    y_test_np = y_test.cpu().numpy()
    mse = np.mean((test_pred - y_test_np) ** 2)
    print(f"  单步预测 MSE: {mse:.6f}")

    # 4b. 自回归预测未来 100 步
    # 取第一个测试样本作为种子序列: shape (1, 50, 1)
    x_seed = X_test[0:1]
    # 预测未来的步数
    future_steps = 100
    # 存储预测结果的列表
    predictions = []

    # 先用种子序列通过RNN得到初始隐藏状态
    output, hidden = model.rnn(x_seed)
    # 当前输入设为种子序列的最后一个时间步
    curr_input = x_seed[:, -1:, :]
    # 当前隐藏状态
    curr_h = hidden

    # 循环预测未来每一步
    for _ in range(future_steps):
        # RNN 单步前向: 输入当前值和隐藏状态,输出新的隐藏状态
        output, curr_h = model.rnn(curr_input, curr_h)
        # 全连接层: 将最后一个隐藏状态映射到预测值
        pred = model.fc(output[:, -1, :])
        # 将预测值添加到结果列表
        predictions.append(pred.item())
        # 将预测值作为下一步的输入: reshape为(1, 1, 1)格式
        curr_input = pred.reshape(1, 1, 1)

    # 获取真实的未来值用于对比
    true_future = wave[split + SEQ_LEN: split + SEQ_LEN + future_steps]

# ================================================================
#  可视化
# ================================================================
fig, axes = plt.subplots(2, 2, figsize=(15, 9))

# 1. 损失曲线
axes[0, 0].plot(train_losses, label='训练损失', color='#2196F3')
axes[0, 0].plot(test_losses, label='测试损失', color='#FF5722', linewidth=2)
axes[0, 0].set_title('损失曲线', fontsize=13, fontweight='bold')
axes[0, 0].set_xlabel('训练轮次')
axes[0, 0].set_ylabel('MSE 损失')
axes[0, 0].legend()
axes[0, 0].grid(alpha=0.3)
axes[0, 0].set_yscale('log')

# 2. 测试集预测效果
plot_n = min(100, len(y_test_np))
axes[0, 1].plot(y_test_np[:plot_n], 'g-', linewidth=1.5, alpha=0.8, label='真实值')
axes[0, 1].plot(test_pred[:plot_n], 'r--', linewidth=1.5, alpha=0.8, label='预测值')
axes[0, 1].set_title(f'测试集单步预测 (MSE={mse:.6f})', fontsize=13, fontweight='bold')
axes[0, 1].legend()
axes[0, 1].grid(alpha=0.3)

# 3. 自回归预测
x_seed_np = X_test[0, :, 0].cpu().numpy()
axes[1, 0].plot(range(SEQ_LEN), x_seed_np, 'b-', linewidth=2, label=f'输入 ({SEQ_LEN}步)')
axes[1, 0].plot(range(SEQ_LEN, SEQ_LEN + future_steps), true_future, 'g--',
                linewidth=1.5, alpha=0.6, label='真实未来')
axes[1, 0].plot(range(SEQ_LEN, SEQ_LEN + future_steps), predictions, 'r-',
                linewidth=2, label='RNN 预测')
axes[1, 0].axvline(x=SEQ_LEN, color='gray', linestyle=':', alpha=0.5)
axes[1, 0].set_title(f'自回归预测: 输入{SEQ_LEN}步 → 预测未来{future_steps}步',
                     fontsize=13, fontweight='bold')
axes[1, 0].legend()
axes[1, 0].grid(alpha=0.3)

# 4. 散点图
axes[1, 1].scatter(y_test_np[:plot_n], test_pred[:plot_n], alpha=0.5, s=15, c='#FF5722')
lims = [-1.5, 1.5]
axes[1, 1].plot(lims, lims, 'b--', linewidth=2, alpha=0.6, label='完美预测')
axes[1, 1].set_xlim(lims)
axes[1, 1].set_ylim(lims)
axes[1, 1].set_title('预测值 vs 真实值', fontsize=13, fontweight='bold')
axes[1, 1].set_xlabel('真实值')
axes[1, 1].set_ylabel('预测值')
axes[1, 1].legend()
axes[1, 1].grid(alpha=0.3)
axes[1, 1].set_aspect('equal')

plt.tight_layout()
plt.savefig('rnn_sine_result.png', dpi=150)
print(f"\n✅ 图片已保存为 rnn_sine_result.png")
plt.show()