PyTorch学习笔记-Week7

本周从计算机视觉 (CNN) 领域迈入了一个全新的领域：自然语言处理 (NLP)与循环神经网络（RNN）。

核心挑战在于，神经网络不理解文字，只理解数字。首要任务就是学会如何将人类的语言“翻译”成机器能够计算的格式。

通过古诗生成案例，从最基础的文本处理，到构建、训练、并最终实现一个能够生成古诗的循环神经网络 (RNN)

一：自然语言处理 (NLP) 入门

1.1 核心翻译流程

1. 分词 (Tokenization): 将一个完整的句子或段落，拆分成最小的语言单元（“词元”或“Token”）。对于中文，这通常是单个的汉字。
2. 构建词汇表 (Vocabulary): 收集所有在数据集中出现过的独立词元，并为每一个词元分配一个独一无二的整数ID。
3. 数值化 (Numericalization): 使用构建好的词汇表，将原始的词元序列转换成一个整数ID序列。

1.2 文本预处理 (preprocess_poems 函数)

通过一个函数封装了所有预处理步骤，这是一种非常好的编程实践。

• 思考与讨论: 构建词汇表的两种方法：

  1. 在循环中逐行读取，用 set.update() 逐行去重。
  2. 先用 "".join(lines) 将所有行合并成一个大字符串，然后用 set() 一次性去重。
  • 结论: 第二种方法更简洁、更符合Python风格 (Pythonic)，并且通常效率更高。

• 关键实现 (poetry_generator_rnn.py):

  import re
  
  def preprocess_poems(file_path):
      # 1. 逐行读取，清洗数据 (移除标点和空白)
      with open(file_path, 'r', encoding='utf-8') as f:
          lines = [re.sub(f"[，。？！、：]", "", line.strip()) for line in f if line.strip()]
  
      # 2. 构建词汇表
      # a. 合并所有文本并去重
      all_chars = sorted(list(set("".join(lines))))
      # b. 创建 id <-> word 映射
      word2id = {word : i for i, word in enumerate(all_chars)}
      id2word = {i : word for i, word in enumerate(all_chars)}
      # c. 添加未知词标记
      word2id['<UNK>'] = len(word2id)
      id2word[len(id2word)] = '<UNK>'
  
      # 3. 数值化
      # 使用嵌套列表推导式高效地将所有诗句转换为ID序列
      id_seqs = [[word2id.get(char, word2id['<UNK>']) for char in line] for line in lines]
  
      return id_seqs, word2id, id2word

---

二：为语言模型构建数据管道

2.1 自定义 Dataset - PoetryDataset

• 核心任务: 将连续的诗歌文本，转换成适用于语言模型训练的 (输入序列, 目标序列) 数据对。

• 实现方式: 使用滑动窗口 (Sliding Window) 的思想。

• 关键实现 (poetry_generator_rnn.py):

  from torch.utils.data import Dataset
  
  class PoetryDataset(Dataset):
      def __init__(self, id_seqs, seq_len):
          self.data = []
          # 遍历每一首独立的诗
          for id_seq in id_seqs:
              if len(id_seq) < seq_len + 1:
                  continue
              # 在每一首诗内部创建滑动窗口
              for i in range(len(id_seq) - seq_len):
                  # 输入序列 (features)
                  features = id_seq[i : i + seq_len]
                  # 目标序列 (labels)，是输入序列向后错一位
                  labels = id_seq[i + 1 : i + seq_len + 1]
                  self.data.append((features, labels))
  
      def __getitem__(self, index):
          features, labels = self.data[index]
          # 关键：转换发生在 __getitem__ 中，实现“懒加载”
          return torch.tensor(features, dtype=torch.long), torch.tensor(labels, dtype=torch.long)
  
      def __len__(self):
          return len(self.data)

2.2 __getitem__ 中的“懒加载” (Lazy Loading)

• 思考: 为什么不像以前一样，在 __init__ 中就把所有数据都转换成torch.tensor？
• 结论: 这是一种更高级、更节省内存的最佳实践。
  • __init__ 只存储轻量级的Python列表，几乎不消耗内存。
  • 真正的张量转换操作被推迟到 __getitem__ 中。
  • 只有当 DataLoader 需要获取某个特定样本时，__getitem__ 才会被调用，实现**“即时转换”**。
  • 这使得我们能够处理远超内存大小的超大型数据集。
• 核心关系: 理解了 DataLoader 的工作机制——它就像一个图书管理员，隐式地、自动地调用 Dataset 的 __getitem__ 方法来按批次“取书”。

---

2.3 张量创建：torch.tensor vs. torch.LongTensor

• 思考: torch.tensor(data, dtype=torch.long) 和 torch.LongTensor(data) 有什么区别？
• 结论:
  • torch.LongTensor() 是旧版的、特定类型的快捷函数。
  • torch.tensor() 是现代的、统一的、通用的张量构造函数。
  • 使用 torch.tensor(..., dtype=...) 的写法更清晰、更灵活，是官方推荐的最佳实践。

---

三：RNN模型架构与核心原理

3.1 核心组件 (nn.RNN, nn.LSTM)

• nn.RNN vs. nn.RNNCell:

  • 思考与辨析: 深入探讨了这两者的区别。
  • 结论:
    • nn.RNNCell 是一个**“单个工作站”，只处理一个时间步的数据，需要我们手动编写for循环**来迭代整个序列。它提供了最高的灵活性，适用于需要在时间步之间进行复杂干预的高级研究。
    • nn.RNN 是一条**“自动化流水线”，它内部封装了循环逻辑，可以一次性处理整个序列。它更高效、更简洁，是绝大多数标准序列任务的首选**。

• nn.LSTM (长短期记忆网络):

  • 在实践中，选择使用nn.LSTM来替代基础的nn.RNN。LSTM通过引入“门控机制”（输入门、遗忘门、输出门），能够更好地捕捉长距离依赖关系，有效缓解了简单RNN的梯度消失/爆炸问题，是现代NLP任务中更常用的基石。

• 关键组件参数:

  • input_size: 输入序列中，每一个元素的特征维度。在当前项目中，这等于词嵌入的维度 (embedding_dim)。
  • hidden_size: RNN隐藏状态的维度，可以理解为模型的“记忆容量”。
  • num_layers: 堆叠的RNN层数。num_layers > 1可以构建深层RNN，让网络学习到更抽象的时间模式。
  • batch_first=True: 一个至关重要的参数，它指定输入和输出张量的维度顺序为 (批次大小, 序列长度, 特征数)，这更直观，也是现代PyTorch代码的推荐实践。

  关键实现:

  # --- 在模型中定义LSTM层 ---
  self.rnn = nn.LSTM(
  embedding_dim,
  hidden_size,
  num_layers,
  batch_first=True
  )

3.2 forward 方法：一个灵活的设计模式

• 思考与辨析:

  • 问题: forward方法中是否需要手动初始化hidden状态？
  • 结论: 不应该。如果hidden状态为None，PyTorch的RNN层会自动创建一个全零的初始隐藏状态，并且其batch_size和device会自动匹配输入张量，这避免了因最后一个批次大小不匹配或设备不一致而导致的常见错误。
  • 问题: forward方法的签名写成forward(self, x, hidden=None)是否规范？
  • 结论: 这是最专业、最推荐的最佳实践。这个设计模式极其灵活，使得同一个模型可以：
    1. 在无状态的训练/分类任务中，通过 model(features) 调用，让PyTorch自动处理hidden状态。
    2. 在有状态的文本生成任务中，通过 output, hidden = model(input, hidden) 的循环调用，手动传递和更新hidden状态，以维持“记忆”的连贯性。

• 关键实现 (PoetryRNN):

  # --- 灵活的forward方法实现 ---
  class PoetryRNN(nn.Module):
      def __init__(self, ...):
          # ... (Embedding, LSTM, Linear layers) ...
  
      def forward(self, x, hidden=None):
          x = self.embed(x)
          # 将输入和（可能存在的）hidden状态传给RNN
          x, hidden = self.rnn(x, hidden)
          # 将RNN所有时间步的输出都通过全连接层
          x = self.fc(x)
          return x, hidden

3.3 RNN信息流：hidden vs output

• 思考与辨析:
  • 问题: “每一个output传给下一个cell作为hidden输入”，这个理解对吗？
  • 结论: 这个理解只在单层RNN中成立。在多层RNN中，hidden和output有明确的分工：
    • hidden state (隐藏状态): 负责将信息水平地传递给同一层的下一个时间步。它是层的“内部记忆”。
    • output: 指的是最后一层（最顶层）在每一个时间步的隐藏状态。它负责将信息垂直地传递出去，作为整个RNN模块的最终输出。

---

四：训练与生成

4.1 训练循环

• 梯度清零的位置:

  • 思考: optimizer.zero_grad() 应该放在循环的开头还是末尾？
  • 结论: 两种写法在功能上等价，但将其放在循环开头是更现代、更清晰的最佳实践。这符合“每一次迭代都是一个全新的开始，先擦干净黑板再写新东西”的逻辑。

• Seq-to-Seq损失计算:

  • 问题: 在PoetryDataset中，labels的形状是 [BATCH_SIZE, SEQ_LEN]，而模型outputs的形状是 [BATCH_SIZE, SEQ_LEN, VOCAB_SIZE]。如何计算损失？
  • 解决方案: nn.CrossEntropyLoss在处理这种时间序列任务时，期望的输入outputs形状是 [BATCH_SIZE, VOCAB_SIZE, SEQ_LEN]。因此，我们需要对模型的输出进行维度转换。

  关键实现:

  # --- 训练循环中的损失计算 ---
  model.train()
  for features, labels in dataloader:
  features, labels = features.to(DEVICE), labels.to(DEVICE)
  
  optimizer.zero_grad()
  	outputs, _ = model(features) # outputs: [B, S, V]
  
  # 交换SeqLen和Vocab_Size维度以匹配CrossEntropyLoss的要求
  	loss = criterion(outputs.transpose(1, 2), labels)
  
  loss.backward()
  	optimizer.step()

4.2 文本生成 (generate_poem 函数)

• 核心原理: 自回归 (Auto-regressive) 和 状态延续 (Stateful)。

  1. 用“开头”的词或句子“预热”RNN，得到一个初始的hidden状态。

  2. 将最后一个词作为下一步的输入。

  3. 在循环中：

     a. 将当前词和上一步的hidden状态一起送入模型。

     b. 从模型输出的概率分布中，随机采样出下一个词的ID。

     c. 将这个新生成的词ID作为下一次循环的输入。

     d. 将模型返回的新的hidden状态保存，用于下一次循环。

• 创造性采样 (Creative Sampling):

  • 问题: 如果总是选择概率最高的词（贪婪搜索），生成的文本会非常单调和重复。

  • 解决方案: 使用温度 (Temperature) 和 多项式采样 (torch.multinomial)。

    关键实现:

    # --- 带温度的随机采样 ---
    # output是模型在当前步的logits输出, shape: [1, 1, vocab_size]
    output_dist = output.view(-1).div(temperature).exp()
    next_id_tensor = torch.multinomial(output_dist, 1)[0]

    temperature值越低，采样结果越保守，接近贪婪搜索；值越高，结果越随机，更具创造性。

• 结构化生成逻辑:

  • 思考与调试: 最初尝试了一个非常复杂的单行for循环来控制诗句长度，但通过调试发现它在处理空start_token时存在边界bug。
  • 最终方案: 采纳了一种更健壮、更清晰的逻辑，即在每个诗句生成前，显式地计算出还需要生成多少个字num_chars_to_gen，然后用一个简单的for循环来执行固定次数的生成。这完美地解决了所有边界问题。

  关键实现:

  # --- 健壮的结构化生成循环 ---
  for i in range(line_num):
  for j, interpunction in enumerate(["，", "。\n"]):
    # 只有在诗歌最开头才计算偏移
    offset = len(poem) if i == 0 and j == 0 else 0
    num_chars_to_gen = line_length - offset
    for _ in range(num_chars_to_gen):
        # ... (执行一次单字生成) ...

五：完整代码

import torch
import numpy as np
from torch import nn, optim
from torch.utils.data import Dataset, DataLoader
import re

SEQ_LEN = 24
BATCH_SIZE = 64
EPOCHS = 30
LR = 1e-3
DEVICE = torch.device("cuda" if torch.cuda.is_available() else "cpu")
print(f"Using device: {DEVICE}")

def preprocess_poems(file_path):
    with open(file_path, 'r', encoding='utf-8') as f:
        lines = [re.sub(f"[，。？！、：]", "", line.strip()) for line in f if line.strip()]
    
    all_chars = sorted(list(set("".join(lines))))

    word2id = {word : i for i, word in enumerate(all_chars)}
    id2word = {i : word for i, word in enumerate(all_chars)}
    word2id['<UNK>'] = len(word2id)
    id2word[len(word2id)] = '<UNK>'

    id_seqs = [[word2id.get(char, word2id['<UNK>']) for char in line] for line in lines]

    return id_seqs, word2id, id2word

id_seqs, word2id, id2word = preprocess_poems('../datasets/poems.txt')
VOCAB_SIZE = len(id2word)

class PoetryDataset(Dataset):
    def __init__(self, id_seqs, seq_len):
        self.data = []
        for id_seq in id_seqs:
            if len(id_seq) < seq_len + 1:
                continue
            for i in range(len(id_seq) - seq_len):
                features = id_seq[i : i + seq_len]
                labels = id_seq[i + 1 : i + 1 + seq_len]
                self.data.append((features, labels))
                
    def __getitem__(self, index):
        features, labels = self.data[index]
        return torch.tensor(features, dtype=torch.long), torch.tensor(labels, dtype=torch.long)
    
    def __len__(self):
        return len(self.data)
    
class PoetryRNN(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_size, num_layers=1):
        super().__init__()
        self.embed = nn.Embedding(vocab_size, embedding_dim)
        self.rnn = nn.LSTM(embedding_dim, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, vocab_size)

    def forward(self, x, hidden=None):
        x = self.embed(x)
        x, hidden = self.rnn(x, hidden)
        x = self.fc(x)
        return x, hidden

dataset = PoetryDataset(id_seqs, SEQ_LEN)
dataloader = DataLoader(dataset=dataset, batch_size=BATCH_SIZE, shuffle=True)

model = PoetryRNN(vocab_size=VOCAB_SIZE, embedding_dim=128, hidden_size=256, num_layers=2).to(DEVICE)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=LR)

model.train()
for epoch in range(EPOCHS):
    running_loss = 0.0
    for features, labels in dataloader:
        features, labels = features.to(DEVICE), labels.to(DEVICE)

        optimizer.zero_grad()
        outputs, _ = model(features)

        loss = criterion(outputs.transpose(1, 2), labels)

        loss.backward()
        optimizer.step()

        running_loss += loss.item()
    
    avg_loss = running_loss / len(dataloader)
    print(f"Epoch {epoch+1}/{EPOCHS} | Loss {avg_loss:.4f}")

def generate_poem(model, start_token, line_length=5, line_num=2, temperature=0.8):
    model.eval()
    poem = list(start_token)
    hidden = None

    with torch.no_grad():
        if start_token:
            input_seq = [word2id.get(w, word2id['<UNK>']) for w in start_token]
            input_tensor = torch.tensor([input_seq], dtype=torch.long).to(DEVICE)
            _, hidden = model(input_tensor, None)
            next_input_id = input_seq[-1]
        else:
            next_input_id = np.random.randint(0, VOCAB_SIZE)
            poem.append(id2word[next_input_id])

    with torch.no_grad():
        for i in range(line_num):
            for j, interpunction in enumerate(["，", "。\n"]):

                offset = len(poem) if i ==0 and j ==0 else 0
                num_chars_to_gen = line_length - offset

                for _ in range(num_chars_to_gen):
                    input_tensor = torch.tensor([[next_input_id]], dtype=torch.long).to(DEVICE)
                    output, hidden = model(input_tensor, hidden)

                    output_dist = output.view(-1).div(temperature).exp()
                    next_input_id = torch.multinomial(output_dist, 1)[0]

                    next_char = id2word.get(int(next_input_id.item()), '<UNK>')
                    poem.append(next_char)
    
                poem.append(interpunction)
    
    return "".join(poem)

print("\n--- 模型生成古诗 ---")
for i in range(5):
    start = "春江"
    poem = generate_poem(model, start, line_length=5, line_num=2)
    print(f"({i+1})\n{poem.strip()}")

for i in range(5):
    start = "明月"
    poem = generate_poem(model, start, line_length=7, line_num=4)
    print(f"({i+6})\n{poem.strip()}")