L5-手写RNN、CNN、Transformer等经典架构

RNN

工作原理： 神经网络是由相互连接的节点组成的网络，每个节点都会对输入数据进行某种形式的处理。在传统的神经网络中，所有的输入都是独立处理的，这意味着网络无法在处理当前输入时考虑到之前的输入。而RNN的设计就是为了解决这个问题。

传统的神经网络数据流只在一个方向上流动，即从输入到输出。这些网络在处理与时间或序列无关的问题时表现良好，例如图像识别。但是对于需要考虑数据的时间序列信息的任务（如语言模型），这种一次性处理模式就略显劣势。

RNN的核心思想是使用循环，使得网络能够将信息从一个步骤传递到下一个步骤。这种循环结构使得网络能够保留某种状态，即网络在处理当前输入时，同时考虑之前的输入。在RNN中，每个序列元素都会更新网络的隐藏状态。这个隐藏状态是网络记忆之前信息的关键，它可以被视为网络的“记忆”。

为了处理序列中的每个元素，RNN会对每个输入执行相同的任务，但每一步都会有一些小的改变，因为它包含了之前步骤的信息。这种结构使得RNN在处理序列数据时非常有效，例如，在文本中，当前单词的含义可能取决于之前的单词。

例子：基于 RNN 的正弦波序列预测实现

Setup

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

1、定义简单的RNN模型

这里forward函数表述RNN的前向传播。其中参数x表示输入张量，形状为 (batch_size, seq_len, input_size)，返回值output为输出张量，形状 (batch_size, seq_len, output_size)，而hidden则代表最后一个时间步的隐藏状态。

class SimpleRNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(SimpleRNN, self).__init__()

        self.hidden_size = hidden_size

        # 输入到隐藏层的权重
        self.Wxh = nn.Linear(input_size, hidden_size)
        # 隐藏层到隐藏层的权重
        self.Whh = nn.Linear(hidden_size, hidden_size)
        # 隐藏层到输出层的权重
        self.Why = nn.Linear(hidden_size, output_size)

        # 初始化隐藏状态
        self.hidden = self.init_hidden()
    
    def init_hidden(self, batch_size=1):
        return torch.zeros(batch_size, self.hidden_size)
    
    def forward(self, x):
        batch_size, seq_len, input_size = x.size()
        
        # 重置隐藏状态
        self.hidden = self.init_hidden(batch_size)
        
        # 存储每个时间步的输出
        outputs = []
        
        # 遍历序列中的每个时间步
        for t in range(seq_len):
            # 获取当前时间步的输入
            x_t = x[:, t, :]
            
            # 计算隐藏状态: h_t = tanh(Wxh * x_t + Whh * h_{t-1})
            hidden_t = torch.tanh(self.Wxh(x_t) + self.Whh(self.hidden))
            
            # 计算输出: y_t = Why * h_t
            output_t = self.Why(hidden_t)
            
            # 更新隐藏状态和输出列表
            self.hidden = hidden_t
            outputs.append(output_t)
        
        # 将输出列表转换为张量并调整形状
        outputs = torch.stack(outputs, dim=1)
        return outputs, self.hidden

2、生成测试数据

def generate_sine_data(seq_length=20, num_samples=1000):
    # 生成时间步
    time = np.linspace(0, 30, num_samples + seq_length)

    # 生成正弦波
    data = np.sin(time)
    
    # 创建输入输出序列
    X, y = [], []
    for i in range(len(data) - seq_length):
        X.append(data[i:i+seq_length])
        y.append(data[i+1:i+seq_length+1])
    
    # 转换为PyTorch张量并添加维度 (样本数, 序列长度, 特征数)
    X = torch.FloatTensor(X).unsqueeze(2)
    y = torch.FloatTensor(y).unsqueeze(2)
    
    return X, y

3、训练模型

def train_rnn():
    # 超参数
    input_size = 1       # 输入特征数（正弦波值）
    hidden_size = 32     # 隐藏层大小
    output_size = 1      # 输出特征数
    seq_length = 20      # 序列长度
    num_epochs = 100     # 训练轮数
    learning_rate = 0.01 # 学习率
    
    # 生成数据
    X, y = generate_sine_data(seq_length=seq_length)
    
    # 划分训练集和测试集
    split_idx = int(0.8 * len(X))
    X_train, y_train = X[:split_idx], y[:split_idx]
    X_test, y_test = X[split_idx:], y[split_idx:]
    
    # 创建模型、损失函数和优化器
    model = SimpleRNN(input_size, hidden_size, output_size)
    criterion = nn.MSELoss()
    optimizer = torch.optim.Adam(model.parameters(), lr=learning_rate)
    
    # 训练循环
    for epoch in range(num_epochs):
        # 前向传播
        outputs, _ = model(X_train)
        loss = criterion(outputs, y_train)
        
        # 反向传播和优化
        optimizer.zero_grad()  # 清零梯度
        loss.backward()        # 反向传播
        optimizer.step()       # 更新参数
        
        # 打印训练进度
        if (epoch + 1) % 10 == 0:
            print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.6f}')
    
    # 在测试集上评估
    model.eval()
    with torch.no_grad():
        test_outputs, _ = model(X_test)
        test_loss = criterion(test_outputs, y_test)
        print(f'Test Loss: {test_loss.item():.6f}')
    
    # 可视化预测结果
    plt.figure(figsize=(12, 6))
    
    # 绘制一个测试样本的预测结果
    sample_idx = 0
    plt.plot(y_test[sample_idx, :, 0], label='真实值')
    plt.plot(test_outputs[sample_idx, :, 0], label='预测值', linestyle='--')
    plt.title('RNN正弦波序列预测')
    plt.xlabel('时间步')
    plt.ylabel('值')
    plt.legend()
    plt.show()
    
    return model

4、运行训练

if __name__ == "__main__":
    model = train_rnn()

CNN

工作原理： CNN，即Convolutional Neural Network，卷积神经网络。它是一种专为处理具有类似网格结构的数据（如图像、音频、时序信号）而设计的深度神经网络。其核心思想是通过卷积操作自动提取局部特征，实现空间不变性和参数高效性。它的主要结构包括：

卷积层（Convolutional Layer）：通过卷积核（filter/kernel）滑动提取局部特征。

激活层（Activation Layer）：常用ReLU等非线性函数。

池化层（Pooling Layer）：如最大池化（Max Pooling）、平均池化（Average Pooling），实现下采样和特征压缩。

全连接层（Fully Connected Layer, FC）：用于整合高层语义特征，输出分类或回归结果。

数学表述：

1、卷积操作

设输入特征图为 $X$，卷积核为 $W$，偏置为 $b$，输出特征图为 $Y$，则二维卷积可表示为： $Y_{i,j}^{(k)} = f \left( \sum_{m=1}^{M} \sum_{n=1}^{N} W_{m,n}^{(k)} \cdot X_{i+m-1, j+n-1} + b^{(k)} \right)$ 其中 $f$ 为激活函数，$k$ 表示第 $k$ 个卷积核。

2、池化操作

以最大池化为例，$P$ 为池化窗口： $Y_{i,j} = \max_{(m,n) \in P} X_{i+m, j+n}$

3、前向传播流程

假设网络有 $L$ 层卷积/池化，最后接全连接层，最终输出为 $\hat{y}$：

$$\begin{align*} a^{(0)} &= X \\ a^{(l)} &= f^{(l)} \bigl( \text{Conv/Pool}(a^{(l-1)}) \bigr), \quad l = 1, 2, \ldots, L \\ \hat{y} &= \text{Softmax} \bigl( W^{(fc)} a^{(L)} + b^{(fc)} \bigr) \end{align*}$$

例子：用CNN实现MNIST手写数字分类

Setup

import torch
import torch.nn as nn
import torch.nn.functional as F
import matplotlib.pyplot as plt
from torchvision import datasets, transforms

1、数据加载与预处理

transform = transforms.Compose([
    transforms.ToTensor(),
    transforms.Normalize((0.5,), (0.5,))
])
trainset = datasets.MNIST(root='./data', train=True, download=True, transform=transform)
testset = datasets.MNIST(root='./data', train=False, download=True, transform=transform)
trainloader = torch.utils.data.DataLoader(trainset, batch_size=64, shuffle=True)
testloader = torch.utils.data.DataLoader(testset, batch_size=1000, shuffle=False)

2、定义CNN模型

class SimpleCNN(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = nn.Conv2d(1, 16, kernel_size=3, padding=1)
        self.pool = nn.MaxPool2d(2, 2)
        self.conv2 = nn.Conv2d(16, 32, kernel_size=3, padding=1)
        self.fc1 = nn.Linear(32 * 7 * 7, 128)
        self.fc2 = nn.Linear(128, 10)
    def forward(self, x):
        x = self.pool(F.relu(self.conv1(x)))  # 28x28 -> 14x14
        x = self.pool(F.relu(self.conv2(x)))  # 14x14 -> 7x7
        x = x.view(-1, 32 * 7 * 7)
        x = F.relu(self.fc1(x))
        x = self.fc2(x)
        return x

model = SimpleCNN()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
criterion = nn.CrossEntropyLoss()

3、训练模型并记录损失

losses = []
epochs = 5
for epoch in range(epochs):
    running_loss = 0.0
    for images, labels in trainloader:
        optimizer.zero_grad()
        outputs = model(images)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()
        running_loss += loss.item()
    avg_loss = running_loss / len(trainloader)
    losses.append(avg_loss)
    print(f"Epoch {epoch+1}/{epochs}, Loss: {avg_loss:.4f}")

4、可视化训练损失曲线

plt.figure(figsize=(7, 4))
plt.plot(range(1, epochs+1), losses, marker='o')
plt.title('Training Loss Curve (CNN on MNIST)')
plt.xlabel('Epoch')
plt.ylabel('Loss')
plt.grid(True)
plt.tight_layout()
plt.show()

5、测试集准确率

model.eval()
correct = 0
total = 0
with torch.no_grad():
    for images, labels in testloader:
        outputs = model(images)
        _, predicted = torch.max(outputs, 1)
        total += labels.size(0)
        correct += (predicted == labels).sum().item()
print(f"Test Accuracy: {100 * correct / total:.2f}%")

6、可视化部分测试样本及预测结果

examples = enumerate(testloader)
batch_idx, (example_data, example_targets) = next(examples)
output = model(example_data)
_, preds = torch.max(output, 1)

plt.figure(figsize=(10, 3))
for i in range(10):
    plt.subplot(2, 5, i+1)
    plt.imshow(example_data[i][0].cpu().numpy(), cmap='gray')
    plt.title(f"Label: {example_targets[i]}\nPred: {preds[i].item()}")
    plt.axis('off')
plt.suptitle('CNN Predictions on MNIST Test Samples')
plt.tight_layout(rect=[0, 0, 1, 0.92])
plt.show()

Transformer

Transformer是一种在自然语言处理任务中广泛使用的模型，它基于注意力机制，特别是自注意力机制和多头注意力机制，来捕捉输入序列中的长距离依赖关系。Transformer模型的主要优点是它可以并行处理整个输入序列，而不是像循环神经网络（RNN）那样逐个处理序列中的元素。这使得Transformer模型在处理长序列时具有更高的效率和更好的性能。值得一提的是，后面产生的大语言模型很多都是基于Transformer结构的，如Bert、Gpt。

下面我们来分别编写几个Transformer重要部分。

Setup

import torch
import torch.nn as nn
import math

1、PositionalEncoding

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super().__init__()
        # 创建位置编码矩阵：shape [max_len, d_model]
        pe = torch.zeros(max_len, d_model)
        position = torch.arange(0, max_len).unsqueeze(1)
        div_term = torch.exp(torch.arange(0, d_model, 2) * (-math.log(10000.0) / d_model))

        # 偶数位置编码使用sin，奇数位置使用cos
        pe[:, 0::2] = torch.sin(position * div_term)  # even
        pe[:, 1::2] = torch.cos(position * div_term)  # odd

        # 添加 batch 维度：shape [1, max_len, d_model]
        pe = pe.unsqueeze(0)
        self.register_buffer('pe', pe)

    def forward(self, x):
        # x: [batch_size, seq_len, d_model]
        x = x + self.pe[:, :x.size(1), :]
        return x

2、Multi-Head Attention

class MultiHeadAttention(nn.Module):
    def __init__(self, d_model, n_heads):
        super().__init__()
        assert d_model % n_heads == 0

        self.n_heads = n_heads
        self.head_dim = d_model // n_heads

        self.wq = nn.Linear(d_model, d_model)
        self.wk = nn.Linear(d_model, d_model)
        self.wv = nn.Linear(d_model, d_model)
        self.fc_out = nn.Linear(d_model, d_model)

    def forward(self, x, mask=None):
        B, S, D = x.size()
        Q = self.wq(x)
        K = self.wk(x)
        V = self.wv(x)

        # 拆成多头
        Q = Q.view(B, S, self.n_heads, self.head_dim).transpose(1, 2)
        K = K.view(B, S, self.n_heads, self.head_dim).transpose(1, 2)
        V = V.view(B, S, self.n_heads, self.head_dim).transpose(1, 2)

        # 注意力
        scores = torch.matmul(Q, K.transpose(-2, -1)) / math.sqrt(self.head_dim)
        if mask is not None:
            scores = scores.masked_fill(mask == 0, float('-inf'))
        attention = torch.softmax(scores, dim=-1)
        out = torch.matmul(attention, V)

        # 拼接多头
        out = out.transpose(1, 2).contiguous().view(B, S, D)
        return self.fc_out(out)

3、EncoderLayer

class TransformerEncoderLayer(nn.Module):
    def __init__(self, d_model, n_heads, dim_feedforward, dropout):
        super().__init__()
        self.self_attn = MultiHeadAttention(d_model, n_heads)
        self.linear1 = nn.Linear(d_model, dim_feedforward)
        self.linear2 = nn.Linear(dim_feedforward, d_model)
        self.dropout = nn.Dropout(dropout)
        self.norm1 = nn.LayerNorm(d_model)
        self.norm2 = nn.LayerNorm(d_model)
        self.activation = nn.ReLU()

    def forward(self, src, src_mask=None):
        # 多头注意力 + 残差连接 + LayerNorm
        attn_output = self.self_attn(src, src_mask)
        src = self.norm1(src + self.dropout(attn_output))

        # FFN + 残差连接 + LayerNorm
        ffn_output = self.linear2(self.dropout(self.activation(self.linear1(src))))
        src = self.norm2(src + self.dropout(ffn_output))

        return src

4、Encoder

class TransformerEncoder(nn.Module):
    def __init__(self, num_layers, d_model, n_heads, dim_feedforward, dropout, max_len=5000):
        super().__init__()
        self.pos_encoder = PositionalEncoding(d_model, max_len)
        self.layers = nn.ModuleList([
            TransformerEncoderLayer(d_model, n_heads, dim_feedforward, dropout)
            for _ in range(num_layers)
        ])
        self.dropout = nn.Dropout(dropout)

    def forward(self, src, src_mask=None):
        # 添加位置编码
        src = self.pos_encoder(src)
        src = self.dropout(src)

        # 逐层传递
        for layer in self.layers:
            src = layer(src, src_mask)
        return src

下面是一个实例：

d_model = 512
n_heads = 8
dim_ff = 2048
dropout = 0.1
num_layers = 6

model = TransformerEncoder(num_layers, d_model, n_heads, dim_ff, dropout)

src = torch.rand(32, 100, d_model)
output = model(src)

总结

本教程以Python手写实现为核心，完成了RNN、CNN、Transformer三大经典架构的代码实现，深化对深度学习本质的理解。深度学习的核心是让模型从数据中自主学习特征，而非依赖人工设计，这一点在实操中得以体现出来：RNN通过循环结构捕捉时序特征，CNN凭卷积与池化提取空间特征，Transformer以自注意力建立全局关联，三者分别适配不同场景，印证了架构为任务服务的设计逻辑。

参考文献：

• https://blog.csdn.net/Iconicdusk/article/details/136520154
• https://blog.csdn.net/ai_aijiang/article/details/149255318
• https://blog.csdn.net/weixin_64110589/article/details/149603632