循环神经网络

# -*- coding: utf-8 -*-
import torch
from torch import nn
from d2l import torch as d2l
import os

T = 1000
# 生成时间索引（1 ~ T）
time = torch.arange(1, T+1, dtype=torch.float32)
# 生成带噪声的正弦时间序列：
# x(t) = sin(0.01t) + 高斯噪声
# 其中 torch.normal(0, 0.2, (T,)) 表示均值为 0、标准差为 0.2 的噪声
x = torch.sin(0.01 * time) + torch.normal(0, 0.2, (T,))
# 绘制时间序列 x
d2l.plot(time, x, 'time', 'x', xlim=[1, T], figsize=(6, 3))

# 保存图像到本地，便于在非交互式环境（如脚本运行）中查看结果
fig_path = '/home/lxg/code/AI/code/20260414/time_series.png'
d2l.plt.savefig(fig_path, dpi=150)
print(f'时间序列图已保存到: {fig_path}')
d2l.plt.show()

# tau 表示“滞后步数”（look-back window）
# 用前 tau 个时刻的值来预测下一时刻的值
tau = 4
# 构造监督学习输入特征矩阵：
# 形状为 (样本数, tau) = (T-tau, tau)
# 每一行是长度为 tau 的时间窗口
features = torch.zeros((T-tau, tau))
for i in range(tau):
    # 第 i 列对应窗口中第 i 个时间步的值
    # 例如 tau=4 时，一行特征是 [x_t, x_{t+1}, x_{t+2}, x_{t+3}]
    features[:, i] = x[i:T-tau+i]
# 标签是每个窗口后面的下一个点：
# y = x_{t+tau}，并 reshape 为二维列向量，便于后续训练
labels = x[tau:].reshape(-1, 1)

# batch_size：每个小批量样本数
# n_train：用于训练的样本数量（从前面的时间片段中截取）
batch_size, n_train = 16, 600
# d2l.load_array 将张量封装成 DataLoader：
# - 只取前 n_train 条作为训练集
# - is_train=True 表示按小批量打乱后迭代
train_iter = d2l.load_array((features[:n_train], labels[:n_train]),
                            batch_size, is_train=True)

# 初始化网络权重的函数
def init_weights(m):
    # 仅对线性层进行 Xavier 均匀初始化，有助于保持前向/反向传播数值稳定
    if type(m) == nn.Linear:
        nn.init.xavier_uniform_(m.weight)

# 一个简单的多层感知机
def get_net():
    # 输入维度为 4（即 tau=4 的时间窗口特征）
    # 隐藏层 10 个神经元，输出 1 个值（预测下一时刻）
    net = nn.Sequential(nn.Linear(4, 10),
                        nn.ReLU(),
                        nn.Linear(10, 1))
    # 应用自定义初始化
    net.apply(init_weights)
    return net

# 平方损失。注意：MSELoss计算平方误差时不带系数1/2
# reduction='none' 表示先保留每个样本的损失值，后续可手动做 sum/mean
loss = nn.MSELoss(reduction='none')

# 训练函数
# net：待训练模型
# train_iter：训练数据迭代器
# loss：损失函数
# epochs：训练轮数
# lr：学习率
def train(net, train_iter, loss, epochs, lr):
    # 使用 Adam 优化器，lr 控制参数更新步长
    trainer = torch.optim.Adam(net.parameters(), lr)
    for epoch in range(epochs):
        # 遍历一个 epoch 内的所有小批量
        for X, y in train_iter:
            # 清空上一轮迭代累积的梯度
            trainer.zero_grad()
            # 前向传播并计算逐样本损失
            l = loss(net(X), y)
            # 反向传播：这里用 sum 聚合一个批量内的损失
            l.sum().backward()
            # 根据梯度更新参数
            trainer.step()
        # 每个 epoch 结束后，输出当前训练集平均损失
        print(f'epoch {epoch + 1}, '
              f'loss: {d2l.evaluate_loss(net, train_iter, loss):f}')

# 实例化网络并启动训练（训练 5 轮，学习率 0.01）
net = get_net()
train(net, train_iter, loss, 5, 0.01)

# 一步预测（1-step prediction）：
# 这里对每个时间点都使用“真实历史观测值”构造输入窗口，
# 因此误差不会在时间上累积，通常曲线会与原始数据更接近。
onestep_preds = net(features)
d2l.plot([time, time[tau:]],
         [x.detach().numpy(), onestep_preds.detach().numpy()], 'time',
         'x', legend=['data', '1-step preds'], xlim=[1, 1000],
         figsize=(6, 3))

# 多步递归预测（multi-step recursive prediction）：
# 思路：先使用一段真实历史作为起点，随后每一步都把“模型上一步预测值”
# 再喂回输入窗口继续预测，因此随着步数增加，误差会逐步累积。
multistep_preds = torch.zeros(T)
# 前 n_train+tau 个点直接使用真实值，作为递归预测的初始上下文
multistep_preds[: n_train + tau] = x[: n_train + tau]
for i in range(n_train + tau, T):
    # 取最近 tau 个值（其中后期可能已包含模型预测值）来预测下一个点
    multistep_preds[i] = net(
        multistep_preds[i - tau:i].reshape((1, -1)))

# 最大预测步长：用于观察“预测步数越长，误差越大”的现象
max_steps = 64

# 这里重新构造一个更宽的特征矩阵：
# - 行数：可用于滚动预测的起点数
# - 列数：tau 个已知观测 + max_steps 个不同步长的预测值
features = torch.zeros((T - tau - max_steps + 1, tau + max_steps))
# 列i（i<tau）是来自x的观测，其时间步从（i）到（i+T-tau-max_steps+1）
for i in range(tau):
    features[:, i] = x[i: i + T - tau - max_steps + 1]

# 列i（i>=tau）是来自（i-tau+1）步的预测，其时间步从（i）到（i+T-tau-max_steps+1）
for i in range(tau, tau + max_steps):
    # 每次使用前 tau 列（可能部分为预测值）作为输入，逐列向右递推
    features[:, i] = net(features[:, i - tau:i]).reshape(-1)

# 选取不同预测步长进行可视化比较：
# 1-step、4-step、16-step、64-step 越往后通常偏差越明显
steps = (1, 4, 16, 64)
d2l.plot([time[tau + i - 1: T - max_steps + i] for i in steps],
         [features[:, (tau + i - 1)].detach().numpy() for i in steps], 'time', 'x',
         legend=[f'{i}-step preds' for i in steps], xlim=[5, 1000],
         figsize=(6, 3))

import collections
import re
from d2l import torch as d2l

# 在 d2l 的数据集注册表中登记《时间机器》数据集：
# - 第一个元素是下载地址
# - 第二个元素是文件校验哈希（用于完整性校验）
d2l.DATA_HUB['time_machine'] = (d2l.DATA_URL + 'timemachine.txt',
                                '090b5e7e70c295757f55df93cb0a180b9691891a')

def read_time_machine():
    """读取时间机器数据集"""
    # d2l.download 会在本地缓存数据文件，若不存在则自动下载
    with open(d2l.download('time_machine'), 'r') as f:
        # 逐行读取原始文本（每个元素是一行字符串）
        lines = f.readlines()
    # 文本预处理：
    # 1) 用正则把非字母字符替换为空格
    # 2) 去掉首尾空白
    # 3) 全部转小写，降低词表稀疏性
    return [re.sub('[^A-Za-z]+', ' ', line).strip().lower() for line in lines]

# 读取并预处理后的文本行列表
lines = read_time_machine()
# 打印数据集总行数与示例行，便于快速确认读取结果
print(f'# 文本总行数: {len(lines)}')
print(lines[0])
print(lines[10])

def tokenize(lines, token='word'):  #@save
    """将文本行拆分为单词或字符词元"""
    # 按“单词”级别切分：
    # 例如 "time traveller" -> ["time", "traveller"]
    if token == 'word':
        return [line.split() for line in lines]
    # 按“字符”级别切分：
    # 例如 "time" -> ["t", "i", "m", "e"]
    elif token == 'char':
        return [list(line) for line in lines]
    else:
        # 若传入未知 token 类型，给出提示信息
        print('错误：未知词元类型：' + token)

# 默认使用单词粒度进行分词（可改为 token='char' 观察字符级分词效果）
tokens = tokenize(lines)
# 打印前 11 行分词结果，便于检查分词是否符合预期
for i in range(11):
    # 每一行输出的是一个词元列表（token list）
    print(tokens[i])

class Vocab:  #@save
    """文本词表"""
    # tokens：分词后的语料（1D 或 2D 列表）
    # min_freq：最小词频阈值，低于该频次的词元会被过滤
    # reserved_tokens：保留词元列表（如 <pad>、<bos>、<eos> 等）
    def __init__(self, tokens=None, min_freq=0, reserved_tokens=None):
        if tokens is None:
            tokens = []
        if reserved_tokens is None:
            reserved_tokens = []
        # 按出现频率排序
        counter = count_corpus(tokens)
        self._token_freqs = sorted(counter.items(), key=lambda x: x[1],
                                   reverse=True)
        # 词表结构采用“双向映射”：
        # 1) idx_to_token：索引 -> 词元（列表）
        # 2) token_to_idx：词元 -> 索引（字典）
        # 其中未知词元 <unk> 固定放在索引 0
        self.idx_to_token = ['<unk>'] + reserved_tokens
        self.token_to_idx = {token: idx
                             for idx, token in enumerate(self.idx_to_token)}
        # 按词频从高到低把词元加入词表
        for token, freq in self._token_freqs:
            # 遇到低于最小词频阈值的词元后可提前终止（后续词频只会更低）
            if freq < min_freq:
                break
            if token not in self.token_to_idx:
                self.idx_to_token.append(token)
                self.token_to_idx[token] = len(self.idx_to_token) - 1

    def __len__(self):
        # 返回词表大小（总词元数）
        return len(self.idx_to_token)

    def __getitem__(self, tokens):
        # 支持“词元 -> 索引”的两种输入：
        # 1) 单个词元（字符串） -> 单个索引（int）
        # 2) 词元列表（list/tuple） -> 索引列表
        if not isinstance(tokens, (list, tuple)):
            # 若词元不存在于词表，返回 unk 索引（0）
            return self.token_to_idx.get(tokens, self.unk)
        return [self.__getitem__(token) for token in tokens]

    def to_tokens(self, indices):
        # 支持“索引 -> 词元”的两种输入：
        # 1) 单个索引（int） -> 单个词元（str）
        # 2) 索引列表（list/tuple） -> 词元列表
        if not isinstance(indices, (list, tuple)):
            return self.idx_to_token[indices]
        return [self.idx_to_token[index] for index in indices]

    @property
    def unk(self):  # 未知词元的索引为0
        return 0

    @property
    def token_freqs(self):
        return self._token_freqs

def count_corpus(tokens):  #@save
    """统计词元的频率"""
    # 这里的tokens是1D列表或2D列表
    if len(tokens) == 0 or isinstance(tokens[0], list):
        # 将词元列表展平成一个列表
        tokens = [token for line in tokens for token in line]
    # Counter 会返回类似 {'time': 42, 'the': 2261, ...} 的词频统计结果
    return collections.Counter(tokens)

# 基于分词结果构建词表（默认不过滤低频词）
vocab = Vocab(tokens)
# 打印词元到索引映射的前 10 项，便于观察词表内容
print(list(vocab.token_to_idx.items())[:10])

# 展示“文本词元列表 -> 索引列表”的映射示例
for i in [0, 10]:
    print('文本:', tokens[i])
    print('索引:', vocab[tokens[i]])

def load_corpus_time_machine(max_tokens=-1):  #@save
    """返回时光机器数据集的词元索引列表和词表"""
    lines = read_time_machine()
    tokens = tokenize(lines, 'char')
    vocab = Vocab(tokens)
    # 因为时光机器数据集中的每个文本行不一定是一个句子或一个段落，
    # 所以将所有文本行展平到一个列表中
    corpus = [vocab[token] for line in tokens for token in line]
    if max_tokens > 0:
        corpus = corpus[:max_tokens]
    return corpus, vocab

corpus, vocab = load_corpus_time_machine()
print(f'# 时光机器数据集总词元数: {len(corpus)}')
print(f'# 时光机器数据集总词表大小: {len(vocab)}')
print(f'# 时光机器数据集前10个词元索引示例:')
for i in range(30):
    print(corpus[i])
print(f'# 时光机器数据集前10个词元示例:')
for i in range(30):
    print(vocab.to_tokens(corpus[i]))

import random
import torch
import re
from d2l import torch as d2l

# 在 d2l 的数据集注册表中登记《时间机器》数据集：
# - 第一个元素是下载地址
# - 第二个元素是文件校验哈希（用于完整性校验）
d2l.DATA_HUB['time_machine'] = (d2l.DATA_URL + 'timemachine.txt',
                                '090b5e7e70c295757f55df93cb0a180b9691891a')

def read_time_machine():
    """读取时间机器数据集"""
    # d2l.download 会在本地缓存数据文件，若不存在则自动下载
    with open(d2l.download('time_machine'), 'r') as f:
        # 逐行读取原始文本（每个元素是一行字符串）
        lines = f.readlines()
    # 文本预处理：
    # 1) 用正则把非字母字符替换为空格
    # 2) 去掉首尾空白
    # 3) 全部转小写，降低词表稀疏性
    return [re.sub('[^A-Za-z]+', ' ', line).strip().lower() for line in lines]

tokens = d2l.tokenize(read_time_machine())
# 因为每个文本行不一定是一个句子或一个段落，因此我们把所有文本行拼接到一起
corpus = [token for line in tokens for token in line]
vocab = d2l.Vocab(corpus)
print(vocab.token_freqs[:10])

freqs = [freq for token, freq in vocab.token_freqs]
d2l.plot(freqs, xlabel='token: x', ylabel='frequency: n(x)',
         xscale='log', yscale='log')

# 将二元词组从元组转为字符串，避免 d2l.Vocab 内部排序时报 tuple/str 类型冲突
bigram_tokens = [' '.join(pair) for pair in zip(corpus[:-1], corpus[1:])]
bigram_vocab = d2l.Vocab(bigram_tokens)
print(bigram_vocab.token_freqs[:10])

# 将三元词组从元组转为字符串，确保词表构建时词元类型一致
trigram_tokens = [' '.join(triple) for triple in zip(
    corpus[:-2], corpus[1:-1], corpus[2:])]
trigram_vocab = d2l.Vocab(trigram_tokens)
print(trigram_vocab.token_freqs[:10])

bigram_freqs = [freq for token, freq in bigram_vocab.token_freqs]
trigram_freqs = [freq for token, freq in trigram_vocab.token_freqs]
d2l.plot([freqs, bigram_freqs, trigram_freqs], xlabel='token: x',
         ylabel='frequency: n(x)', xscale='log', yscale='log',
         legend=['unigram', 'bigram', 'trigram'])

d2l.plt.show()

def seq_data_iter_random(corpus, batch_size, num_steps):  #@save
    """使用随机抽样生成一个小批量子序列"""
    # 从随机偏移量开始对序列进行分区，随机范围包括num_steps-1
    corpus = corpus[random.randint(0, num_steps - 1):]
    # 减去1，是因为我们需要考虑标签
    num_subseqs = (len(corpus) - 1) // num_steps
    # 长度为num_steps的子序列的起始索引
    initial_indices = list(range(0, num_subseqs * num_steps, num_steps))
    # 在随机抽样的迭代过程中，
    # 来自两个相邻的、随机的、小批量中的子序列不一定在原始序列上相邻
    random.shuffle(initial_indices)

    def data(pos):
        # 返回从pos位置开始的长度为num_steps的序列
        return corpus[pos: pos + num_steps]

    num_batches = num_subseqs // batch_size
    for i in range(0, batch_size * num_batches, batch_size):
        # 在这里，initial_indices包含子序列的随机起始索引
        initial_indices_per_batch = initial_indices[i: i + batch_size]
        X = [data(j) for j in initial_indices_per_batch]
        Y = [data(j + 1) for j in initial_indices_per_batch]
        yield torch.tensor(X), torch.tensor(Y)

my_seq = list(range(35))
for X, Y in seq_data_iter_random(my_seq, batch_size=2, num_steps=5):
    print('X: ', X, '\nY:', Y)

def seq_data_iter_sequential(corpus, batch_size, num_steps):  #@save
    """使用顺序分区生成一个小批量子序列"""
    # 从随机偏移量开始划分序列
    offset = random.randint(0, num_steps)
    num_tokens = ((len(corpus) - offset - 1) // batch_size) * batch_size
    Xs = torch.tensor(corpus[offset: offset + num_tokens])
    Ys = torch.tensor(corpus[offset + 1: offset + 1 + num_tokens])
    Xs, Ys = Xs.reshape(batch_size, -1), Ys.reshape(batch_size, -1)
    num_batches = Xs.shape[1] // num_steps
    for i in range(0, num_steps * num_batches, num_steps):
        X = Xs[:, i: i + num_steps]
        Y = Ys[:, i: i + num_steps]
        yield X, Y

for X, Y in seq_data_iter_sequential(my_seq, batch_size=2, num_steps=5):
    print('X: ', X, '\nY:', Y)

class SeqDataLoader:  #@save
    """加载序列数据的迭代器"""
    def __init__(self, batch_size, num_steps, use_random_iter, max_tokens):
        if use_random_iter:
            self.data_iter_fn = d2l.seq_data_iter_random
        else:
            self.data_iter_fn = d2l.seq_data_iter_sequential
        self.corpus, self.vocab = d2l.load_corpus_time_machine(max_tokens)
        self.batch_size, self.num_steps = batch_size, num_steps

    def __iter__(self):
        return self.data_iter_fn(self.corpus, self.batch_size, self.num_steps)
    
def load_data_time_machine(batch_size, num_steps,  #@save
                           use_random_iter=False, max_tokens=10000):
    """返回时光机器数据集的迭代器和词表"""
    data_iter = SeqDataLoader(
        batch_size, num_steps, use_random_iter, max_tokens)
    return data_iter, data_iter.vocab

# -*- coding: utf-8 -*-
import torch
from d2l import torch as d2l

# 本示例用于说明：
# 在循环神经网络（RNN）中，
# X @ W_xh + H @ W_hh 这一计算，
# 可以等价改写为一次拼接后的矩阵乘法。

# X：当前时间步输入，形状 (batch_size, input_size) = (3, 1)
# W_xh：输入到隐藏层的权重，形状 (input_size, hidden_size) = (1, 4)
X, W_xh = torch.normal(0, 1, (3, 1)), torch.normal(0, 1, (1, 4))
# H：上一时间步隐藏状态，形状 (batch_size, hidden_size) = (3, 4)
# W_hh：隐藏状态到隐藏状态的权重，形状 (hidden_size, hidden_size) = (4, 4)
H, W_hh = torch.normal(0, 1, (3, 4)), torch.normal(0, 1, (4, 4))

# 方式一：分别做两次矩阵乘法，再相加
# 结果形状：(3, 4)
Y1 = torch.matmul(X, W_xh) + torch.matmul(H, W_hh)

# 打印方式一结果
print(Y1)

# 方式二：先拼接，再做一次矩阵乘法
# torch.cat((X, H), 1)      -> 在特征维拼接，形状由 (3,1)+(3,4) 变为 (3,5)
# torch.cat((W_xh, W_hh),0) -> 在行维拼接，形状由 (1,4)+(4,4) 变为 (5,4)
# 相乘后结果仍为 (3,4)，理论上与 Y1 完全相同
Y2 = torch.matmul(torch.cat((X, H), 1), torch.cat((W_xh, W_hh), 0))

# 打印方式二结果（应与 Y1 对应元素一致）
print(Y2)

# 额外验证：数值是否几乎完全一致（浮点误差范围内）
print("Y1 与 Y2 是否近似相等:", torch.allclose(Y1, Y2))