📊 循环神经网络（RNN）全面解析

---

1. 🎯 Motivation（动机）

传统神经网络的局限性

前馈神经网络（FNN）的问题：

• 固定输入大小：输入维度必须预先确定

• 无记忆功能：每次预测独立，不考虑历史信息

• 无法处理变长序列：难以处理句子、语音等长度不定的数据

示例对比：

# 传统神经网络 - 处理句子情感分析的问题
sentence = "这个电影真的非常非常非常好看"
# FNN看到的是独立的词，无法理解"非常"的重复强调作用

# RNN可以捕捉这种序列模式
# 每个"非常"都会增强情感强度

为什么需要RNN？

核心需求：处理具有时间依赖或顺序关系的数据

数据类型	传统网络问题	RN优势
时间序列	忽略时间顺序	考虑历史趋势
自然语言	无法理解词序	捕捉语法结构
语音信号	处理独立帧	理解连续发音
视频数据	分析单张图片	理解动作连贯性

生物学启发

人脑处理信息的方式是序列化的：

• 阅读时逐个单词理解

• 听话时连续解析语音

• 决策时考虑历史经验

---

2. 📈 Sequential Data（序列数据）

序列数据的特性

定义：数据点之间存在顺序依赖关系的集合

主要类型

# 1. 时间序列（Temporal Sequences）
stock_prices = [100, 102, 105, 103, 108]  # 时间顺序重要

# 2. 文本序列（Text Sequences）
sentence = ["我", "爱", "深度", "学习"]  # 词序决定语义

# 3. 语音序列（Speech Sequences）
audio_frames = [frame1, frame2, frame3, ...]  # 连续发音

# 4. 行为序列（Action Sequences）
user_actions = ["登录", "浏览", "点击", "购买"]  # 行为模式

序列数据的数学表示

一般形式：

序列S = {x₁, x₂, x₃, ..., x_T}
其中x_t ∈ R^d 表示第t个时间步的数据点

关键特性：

1. 变长性：序列长度T不固定

2. 依赖性：x_t 依赖于 x{t-1}, x{t-2}, ...

3. 动态性：数据分布可能随时间变化

序列建模的挑战

1. 长期依赖（Long-term Dependencies）

# 例子：语言模型预测
text = "我在北京长大...（中间省略100字）...所以我说的方言带有___口音"
# 需要记忆"北京"这个关键信息

2. 可变长度处理

# 不同长度的序列需要统一处理
short_seq = ["你好"]  # 长度2
long_seq = ["今天天气很好我们一起去公园玩"]  # 长度10

3. 计算效率

# 序列长度可能很大（如长文档、长时间序列）
# 需要高效的并行计算

---

3. 🔄 Recurrent Neural Network（循环神经网络）

核心思想：参数共享 + 循环连接

与传统神经网络对比：

# 前馈神经网络（无记忆）
output_t = f(x_t)  # 只依赖当前输入

# 循环神经网络（有记忆）  
output_t = f(x_t, h_{t-1})  # 依赖当前输入+历史状态

RNN基本结构

数学定义

h_t = σ(W_hh * h_{t-1} + W_xh * x_t + b_h)  # 隐藏状态更新
y_t = W_hy * h_t + b_y                       # 输出计算

代码实现

import torch
import torch.nn as nn

class SimpleRNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super().__init__()
        self.hidden_size = hidden_size
        
        # 权重矩阵
        self.W_xh = nn.Linear(input_size, hidden_size)  # 输入到隐藏层
        self.W_hh = nn.Linear(hidden_size, hidden_size) # 隐藏层到隐藏层
        self.W_hy = nn.Linear(hidden_size, output_size) # 隐藏层到输出层
        
        self.tanh = nn.Tanh()
    
    def forward(self, x_sequence):
        """处理整个序列"""
        batch_size, seq_len, input_size = x_sequence.shape
        hidden_states = []
        outputs = []
        
        # 初始化隐藏状态
        h_t = torch.zeros(batch_size, self.hidden_size)
        
        # 按时间步处理序列
        for t in range(seq_len):
            x_t = x_sequence[:, t, :]  # 当前时间步输入
            
            # RNN核心计算
            h_t = self.tanh(self.W_xh(x_t) + self.W_hh(h_t))
            y_t = self.W_hy(h_t)
            
            hidden_states.append(h_t)
            outputs.append(y_t)
        
        # 堆叠所有时间步的输出
        outputs = torch.stack(outputs, dim=1)  # [batch, seq_len, output_size]
        return outputs, hidden_states

RNN的三种展开模式

1. 一对一（One-to-One）

单个输入 → 单个输出
应用：图像分类

2. 一对多（One-to-Many）

单个输入 → 序列输出
应用：图像描述生成

3. 多对一（Many-to-One）

序列输入 → 单个输出
应用：情感分析、股票价格预测

4. 多对多（Many-to-Many）

序列输入 → 序列输出
应用：机器翻译、语音识别

时间反向传播（BPTT）

训练挑战：梯度通过时间传播可能消失或爆炸

def backward_bptt(self, x_sequence, y_target, loss_fn):
    """随时间反向传播"""
    # 前向传播（保存所有中间状态）
    outputs, hidden_states = self.forward(x_sequence)
    
    # 计算总损失
    total_loss = 0
    for t in range(len(outputs)):
        total_loss += loss_fn(outputs[t], y_target[:, t, :])
    
    # 反向传播（沿时间维度）
    dW_xh, dW_hh, dW_hy = 0, 0, 0
    db_h, db_y = 0, 0
    
    # 从最后时间步开始
    dh_next = torch.zeros_like(hidden_states[0])
    
    for t in reversed(range(len(outputs))):
        # 计算当前时间步梯度
        dy = gradients(outputs[t], y_target[:, t, :])
        
        # 输出层梯度
        dW_hy += hidden_states[t].T @ dy
        db_y += dy.sum(dim=0)
        
        # 隐藏层梯度（包含来自下一时间步的梯度）
        dh = dy @ self.W_hy.weight.T + dh_next
        dh_raw = dh * (1 - hidden_states[t]**2)  # tanh导数
        
        # 权重梯度
        if t > 0:
            dW_hh += hidden_states[t-1].T @ dh_raw
        dW_xh += x_sequence[:, t, :].T @ dh_raw
        db_h += dh_raw.sum(dim=0)
        
        # 传播到前一时间步
        dh_next = dh_raw @ self.W_hh.weight.T
    
    return total_loss, (dW_xh, dW_hh, dW_hy, db_h, db_y)

---

4. 🧠 Long Short-Term Memory（长短期记忆网络）

解决RNN的长期依赖问题

传统RNN的局限性：

# 梯度消失示例：长期记忆衰减
h_100 = W_hh^100 * h_0 + ...  # 早期信息几乎消失

LSTM核心创新：门控机制

LSTM单元结构

class LSTMCell(nn.Module):
    def __init__(self, input_size, hidden_size):
        super().__init__()
        self.input_size = input_size
        self.hidden_size = hidden_size
        
        # 输入门、遗忘门、输出门、候选细胞状态
        self.W_xi = nn.Linear(input_size, hidden_size)
        self.W_hi = nn.Linear(hidden_size, hidden_size)
        self.b_i = nn.Parameter(torch.zeros(hidden_size))
        
        self.W_xf = nn.Linear(input_size, hidden_size)
        self.W_hf = nn.Linear(hidden_size, hidden_size)
        self.b_f = nn.Parameter(torch.ones(hidden_size))  # 初始偏置为1
        
        self.W_xo = nn.Linear(input_size, hidden_size)
        self.W_ho = nn.Linear(hidden_size, hidden_size)
        self.b_o = nn.Parameter(torch.zeros(hidden_size))
        
        self.W_xc = nn.Linear(input_size, hidden_size)
        self.W_hc = nn.Linear(hidden_size, hidden_size)
        self.b_c = nn.Parameter(torch.zeros(hidden_size))
    
    def forward(self, x_t, h_prev, c_prev):
        """LSTM前向传播"""
        # 输入门：控制新信息流入
        i_t = torch.sigmoid(self.W_xi(x_t) + self.W_hi(h_prev) + self.b_i)
        
        # 遗忘门：控制历史信息保留
        f_t = torch.sigmoid(self.W_xf(x_t) + self.W_hf(h_prev) + self.b_f)
        
        # 输出门：控制信息输出
        o_t = torch.sigmoid(self.W_xo(x_t) + self.W_ho(h_prev) + self.b_o)
        
        # 候选细胞状态
        c_tilde = torch.tanh(self.W_xc(x_t) + self.W_hc(h_prev) + self.b_c)
        
        # 更新细胞状态：遗忘旧信息 + 添加新信息
        c_t = f_t * c_prev + i_t * c_tilde
        
        # 更新隐藏状态
        h_t = o_t * torch.tanh(c_t)
        
        return h_t, c_t

LSTM的三个门控机制

1. 遗忘门（Forget Gate）

作用：决定从细胞状态中丢弃哪些信息

f_t = σ(W_f · [h_{t-1}, x_t] + b_f)

2. 输入门（Input Gate）

作用：决定哪些新信息存入细胞状态

i_t = σ(W_i · [h_{t-1}, x_t] + b_i)

3. 输出门（Output Gate）

作用：决定输出哪些信息

o_t = σ(W_o · [h_{t-1}, x_t] + b_o)

GRU（门控循环单元）

LSTM的简化版本：

class GRUCell(nn.Module):
    def __init__(self, input_size, hidden_size):
        super().__init__()
        # 重置门和更新门
        self.W_xz = nn.Linear(input_size, hidden_size)
        self.W_hz = nn.Linear(hidden_size, hidden_size)
        
        self.W_xr = nn.Linear(input_size, hidden_size)
        self.W_hr = nn.Linear(hidden_size, hidden_size)
        
        self.W_xh = nn.Linear(input_size, hidden_size)
        self.W_hh = nn.Linear(hidden_size, hidden_size)
    
    def forward(self, x_t, h_prev):
        # 更新门：控制历史信息保留
        z_t = torch.sigmoid(self.W_xz(x_t) + self.W_hz(h_prev))
        
        # 重置门：控制历史信息忽略
        r_t = torch.sigmoid(self.W_xr(x_t) + self.W_hr(h_prev))
        
        # 候选隐藏状态
        h_tilde = torch.tanh(
            self.W_xh(x_t) + self.W_hh(r_t * h_prev)
        )
        
        # 最终隐藏状态：新旧信息融合
        h_t = (1 - z_t) * h_prev + z_t * h_tilde
        
        return h_t

---

5. ⏰ Time-series Applications（时间序列应用）

金融时间序列预测

股票价格预测

class StockPredictor(nn.Module):
    """基于LSTM的股票预测"""
    def __init__(self, feature_dim, hidden_dim, predict_days=1):
        super().__init__()
        self.lstm = nn.LSTM(
            input_size=feature_dim,      # 特征维度（开盘、收盘、成交量等）
            hidden_size=hidden_dim,
            num_layers=2,
            batch_first=True,
            dropout=0.2
        )
        self.regressor = nn.Sequential(
            nn.Linear(hidden_dim, 64),
            nn.ReLU(),
            nn.Dropout(0.3),
            nn.Linear(64, predict_days)  # 预测未来n天
        )
    
    def forward(self, x_historical):
        # x_historical: [batch, seq_len, feature_dim]
        lstm_out, (h_n, c_n) = self.lstm(x_historical)
        
        # 取最后一个时间步的隐藏状态
        last_hidden = lstm_out[:, -1, :]
        
        # 预测未来价格
        prediction = self.regressor(last_hidden)
        return prediction  # [batch, predict_days]

# 使用示例
model = StockPredictor(feature_dim=5, hidden_dim=128, predict_days=5)
historical_data = torch.randn(32, 30, 5)  # 32支股票，30天历史，5个特征
future_prices = model(historical_data)   # 预测未来5天

异常检测应用

工业设备异常检测

class AnomalyDetector(nn.Module):
    """基于RNN的时序异常检测"""
    def __init__(self, input_dim, hidden_dim):
        super().__init__()
        self.encoder = nn.LSTM(input_dim, hidden_dim, batch_first=True)
        self.decoder = nn.LSTM(hidden_dim, input_dim, batch_first=True)
        
    def forward(self, x):
        # 编码器：序列→隐藏表示
        _, (hidden, _) = self.encoder(x)
        
        # 解码器：隐藏表示→重建序列
        # 使用编码器最后状态初始化解码器
        seq_len = x.size(1)
        decoder_input = torch.zeros(x.size(0), seq_len, hidden.size(2))
        reconstructed, _ = self.decoder(decoder_input, (hidden, torch.zeros_like(hidden)))
        
        return reconstructed
    
    def detect_anomalies(self, x, threshold=0.1):
        """检测异常点"""
        reconstructed = self.forward(x)
        
        # 计算重建误差
        reconstruction_error = torch.abs(x - reconstructed)
        
        # 标记异常
        anomalies = reconstruction_error > threshold
        return anomalies, reconstruction_error

自然语言处理应用

文本情感分析

class SentimentAnalyzer(nn.Module):
    """基于RNN的情感分析"""
    def __init__(self, vocab_size, embed_dim, hidden_dim, num_classes):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, batch_first=True, bidirectional=True)
        self.classifier = nn.Linear(hidden_dim * 2, num_classes)  # 双向→2倍
        self.dropout = nn.Dropout(0.5)
    
    def forward(self, text_sequences):
        # text_sequences: [batch, seq_len]
        embedded = self.embedding(text_sequences)  # [batch, seq_len, embed_dim]
        
        # 双向LSTM处理
        lstm_out, _ = self.lstm(embedded)  # [batch, seq_len, hidden_dim*2]
        
        # 取最后一个时间步（包含前后文信息）
        last_hidden = lstm_out[:, -1, :]
        last_hidden = self.dropout(last_hidden)
        
        # 情感分类
        logits = self.classifier(last_hidden)  # [batch, num_classes]
        return logits

医疗时间序列分析

心电图异常检测

class ECGAnalyzer(nn.Module):
    """心电图信号分析"""
    def __init__(self, signal_dim, hidden_dim, num_classes):
        super().__init__()
        # 心电图信号通常是多导联的
        self.conv = nn.Sequential(
            nn.Conv1d(signal_dim, 64, 5, padding=2),  # 局部特征提取
            nn.ReLU(),
            nn.MaxPool1d(2),
            nn.Conv1d(64, 128, 5, padding=2),
            nn.ReLU(),
            nn.MaxPool1d(2)
        )
        
        # RNN处理时序依赖
        self.gru = nn.GRU(128, hidden_dim, batch_first=True, bidirectional=True)
        self.classifier = nn.Linear(hidden_dim * 2, num_classes)
    
    def forward(self, ecg_signals):
        # ecg_signals: [batch, seq_len, leads]
        x = ecg_signals.transpose(1, 2)  # [batch, leads, seq_len]
        
        # CNN提取局部特征
        conv_out = self.conv(x)  # [batch, 128, seq_len/4]
        conv_out = conv_out.transpose(1, 2)  # [batch, seq_len/4, 128]
        
        # RNN捕捉长期依赖
        gru_out, _ = self.gru(conv_out)  # [batch, seq_len/4, hidden_dim*2]
        
        # 分类
        last_hidden = gru_out[:, -1, :]
        output = self.classifier(last_hidden)
        return output

实际应用考虑因素

1. 数据预处理

def preprocess_time_series(data, seq_length=50):
    """时间序列数据预处理"""
    sequences = []
    labels = []
    
    for i in range(len(data) - seq_length):
        seq = data[i:i+seq_length]  # 输入序列
        label = data[i+seq_length]  # 预测目标
        sequences.append(seq)
        labels.append(label)
    
    return torch.tensor(sequences), torch.tensor(labels)

2. 训练策略

def train_rnn_with_early_stopping(model, train_loader, val_loader, patience=10):
    """早停法训练"""
    best_val_loss = float('inf')
    patience_counter = 0
    
    for epoch in range(100):
        # 训练阶段
        model.train()
        for batch_x, batch_y in train_loader:
            # ... 训练代码
        
        # 验证阶段
        model.eval()
        val_loss = 0
        with torch.no_grad():
            for batch_x, batch_y in val_loader:
                # ... 验证代码
        
        # 早停判断
        if val_loss < best_val_loss:
            best_val_loss = val_loss
            patience_counter = 0
            # 保存最佳模型
            torch.save(model.state_dict(), 'best_model.pth')
        else:
            patience_counter += 1
            if patience_counter >= patience:
                print(f"早停在epoch {epoch}")
                break

---

🎯 总结

RNN及其变体（LSTM、GRU）为序列数据处理提供了强大的工具。从动机到实际应用，RNN的核心价值在于能够建模序列依赖关系，这在时间序列分析、自然语言处理、语音识别等领域具有不可替代的作用。

关键要点：

1. 序列数据普遍存在且重要

2. RNN通过循环连接实现记忆功能

3. LSTM/GRU通过门控机制解决长期依赖问题

4. 实际应用需要考虑数据特性和业务需求

虽然Transformer在某些任务上表现更好，但RNN仍然是理解序列建模基础和学习深度学习时序处理的重要阶梯。