09-循环神经网络¶

import torch
import torch.nn as nn

# PyTorch RNN基本用法
rnn = nn.RNN(input_size=10, hidden_size=64, num_layers=1, batch_first=True)

# 输入: (batch_size, seq_len, input_size)
x = torch.randn(32, 20, 10)  # batch=32, 序列长度=20, 输入维度=10
h0 = torch.zeros(1, 32, 64)  # 初始隐藏状态

output, hn = rnn(x, h0)
# output: (32, 20, 64) — 每个时间步的隐藏状态
# hn: (1, 32, 64) — 最后时间步的隐藏状态

# PyTorch梯度裁剪
torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=5.0)

# PyTorch LSTM
lstm = nn.LSTM(
    input_size=128,     # 输入维度
    hidden_size=256,    # 隐藏状态维度
    num_layers=2,       # 层数
    batch_first=True,   # (batch, seq, feature)
    dropout=0.3,        # 层间dropout
    bidirectional=False  # 单向
)

x = torch.randn(32, 50, 128)  # (batch, seq_len, input_size)
output, (hn, cn) = lstm(x)
# output: (32, 50, 256)  每个时间步的隐藏状态
# hn: (2, 32, 256)       每层最后时间步的隐藏状态
# cn: (2, 32, 256)       每层最后时间步的细胞状态

# PyTorch GRU
gru = nn.GRU(input_size=128, hidden_size=256, num_layers=2,
             batch_first=True, dropout=0.3)

x = torch.randn(32, 50, 128)
output, hn = gru(x)  # 没有cn（无细胞状态）
# output: (32, 50, 256)
# hn: (2, 32, 256)

bilstm = nn.LSTM(input_size=128, hidden_size=256, num_layers=2,
                 batch_first=True, bidirectional=True)

x = torch.randn(32, 50, 128)
output, (hn, cn) = bilstm(x)
# output: (32, 50, 512)  — 2*256
# hn: (4, 32, 256)       — 2层*2方向

class Encoder(nn.Module):  # 继承nn.Module定义神经网络层
    def __init__(self, vocab_size, embed_dim, hidden_dim, num_layers=1):  # __init__构造方法，创建对象时自动调用
        super().__init__()  # super()调用父类方法
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, num_layers, batch_first=True)

    def forward(self, x):
        embedded = self.embedding(x)
        outputs, (hidden, cell) = self.lstm(embedded)
        return outputs, hidden, cell

class Decoder(nn.Module):
    def __init__(self, vocab_size, embed_dim, hidden_dim, num_layers=1):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim)
        self.lstm = nn.LSTM(embed_dim, hidden_dim, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_dim, vocab_size)

    def forward(self, x, hidden, cell):
        embedded = self.embedding(x.unsqueeze(1))  # unsqueeze增加一个维度
        output, (hidden, cell) = self.lstm(embedded, (hidden, cell))
        prediction = self.fc(output.squeeze(1))  # squeeze去除大小为1的维度
        return prediction, hidden, cell

# 简单的时序预测LSTM
class TimeSeriesLSTM(nn.Module):
    def __init__(self, input_dim=1, hidden_dim=64, output_dim=1, num_layers=2):
        super().__init__()
        self.lstm = nn.LSTM(input_dim, hidden_dim, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_dim, output_dim)

    def forward(self, x):
        # x: (batch, seq_len, input_dim)
        lstm_out, _ = self.lstm(x)
        prediction = self.fc(lstm_out[:, -1, :])  # 取最后一步
        return prediction

import torch
import torch.nn as nn
import torch.optim as optim
from torch.utils.data import Dataset, DataLoader
from collections import Counter
import re

# ==================== 1. 数据预处理 ====================
class TextDataset(Dataset):
    """简易文本数据集（以IMDb为例）"""
    def __init__(self, texts, labels, vocab, max_len=256):
        self.texts = texts
        self.labels = labels
        self.vocab = vocab
        self.max_len = max_len

    def __len__(self):  # __len__定义len()的行为
        return len(self.texts)

    def __getitem__(self, idx):  # __getitem__定义索引访问行为
        tokens = tokenize(self.texts[idx])
        # 将token转为索引
        indices = [self.vocab.get(t, self.vocab['<UNK>']) for t in tokens]
        # 截断或填充
        if len(indices) > self.max_len:
            indices = indices[:self.max_len]
        else:
            indices = indices + [self.vocab['<PAD>']] * (self.max_len - len(indices))
        return torch.tensor(indices, dtype=torch.long), torch.tensor(self.labels[idx], dtype=torch.long)

def tokenize(text):
    """简单分词"""
    text = text.lower()
    text = re.sub(r'[^a-zA-Z\s]', '', text)
    return text.split()

def build_vocab(texts, max_vocab=25000):
    """构建词表"""
    counter = Counter()  # Counter统计元素出现次数
    for text in texts:
        counter.update(tokenize(text))
    # 保留高频词
    most_common = counter.most_common(max_vocab - 2)
    vocab = {'<PAD>': 0, '<UNK>': 1}
    for word, _ in most_common:
        vocab[word] = len(vocab)
    return vocab

# ==================== 2. 模型定义 ====================
class LSTMClassifier(nn.Module):
    """LSTM情感分类模型"""
    def __init__(self, vocab_size, embed_dim=128, hidden_dim=256,
                 num_layers=2, num_classes=2, dropout=0.5, bidirectional=True):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_dim, padding_idx=0)
        self.lstm = nn.LSTM(
            embed_dim, hidden_dim, num_layers,
            batch_first=True, dropout=dropout if num_layers > 1 else 0,
            bidirectional=bidirectional
        )
        self.dropout = nn.Dropout(dropout)

        direction_factor = 2 if bidirectional else 1
        self.fc = nn.Sequential(
            nn.Linear(hidden_dim * direction_factor, hidden_dim),
            nn.ReLU(),
            nn.Dropout(dropout),
            nn.Linear(hidden_dim, num_classes)
        )

    def forward(self, x):
        # x: (batch, seq_len)
        embedded = self.dropout(self.embedding(x))  # (batch, seq_len, embed_dim)
        lstm_out, (hn, cn) = self.lstm(embedded)     # lstm_out: (batch, seq_len, hidden_dim*2)

        # 方式1：取最后时间步
        # 对于双向LSTM，拼接两个方向最后时间步的隐藏状态
        if self.lstm.bidirectional:
            hidden = torch.cat([hn[-2], hn[-1]], dim=1)  # (batch, hidden_dim*2)  # [-1]负索引取最后一个元素
        else:
            hidden = hn[-1]  # (batch, hidden_dim)

        # 方式2（替代）：对所有时间步做平均池化
        # hidden = lstm_out.mean(dim=1)

        output = self.fc(self.dropout(hidden))
        return output

# ==================== 3. 训练流程 ====================
def train_sentiment_model():
    """完整的训练流程（需要配合真实数据集使用）"""
    DEVICE = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

    # 假设已有train_texts, train_labels, test_texts, test_labels
    # 实际使用时可加载IMDb数据集:
    # from torchtext.datasets import IMDB
    # 或 from datasets import load_dataset
    # dataset = load_dataset("imdb")

    # 示例数据（实际使用需替换）
    train_texts = ["This movie is great!", "Terrible waste of time."] * 1000
    train_labels = [1, 0] * 1000

    vocab = build_vocab(train_texts)
    VOCAB_SIZE = len(vocab)

    train_dataset = TextDataset(train_texts, train_labels, vocab, max_len=256)
    train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)  # DataLoader批量加载数据，支持shuffle和多进程

    model = LSTMClassifier(
        vocab_size=VOCAB_SIZE,
        embed_dim=128,
        hidden_dim=256,
        num_layers=2,
        num_classes=2,
        dropout=0.5,
        bidirectional=True
    ).to(DEVICE)

    criterion = nn.CrossEntropyLoss()
    optimizer = optim.Adam(model.parameters(), lr=1e-3, weight_decay=1e-5)
    scheduler = optim.lr_scheduler.ReduceLROnPlateau(optimizer, patience=3, factor=0.5)

    # 训练循环
    for epoch in range(20):
        model.train()  # train()开启训练模式
        total_loss = 0
        correct = 0
        total = 0

        for inputs, labels in train_loader:
            inputs, labels = inputs.to(DEVICE), labels.to(DEVICE)

            optimizer.zero_grad()  # 清零梯度，防止梯度累积
            outputs = model(inputs)
            loss = criterion(outputs, labels)
            loss.backward()  # 反向传播计算梯度

            # 梯度裁剪
            torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=5.0)

            optimizer.step()  # 根据梯度更新模型参数

            total_loss += loss.item()  # .item()将单元素张量转为Python数值
            _, predicted = outputs.max(1)
            total += labels.size(0)
            correct += predicted.eq(labels).sum().item()  # 链式调用，连续执行多个方法

        train_acc = 100. * correct / total
        avg_loss = total_loss / len(train_loader)
        scheduler.step(avg_loss)

        print(f"Epoch [{epoch+1}/20] Loss: {avg_loss:.4f} | Acc: {train_acc:.2f}%")

# train_sentiment_model()

import numpy as np

def load_pretrained_embeddings(vocab, embed_path='glove.6B.100d.txt', embed_dim=100):
    """加载GloVe预训练词向量"""
    embeddings = np.random.normal(0, 0.1, (len(vocab), embed_dim))
    embeddings[0] = 0  # PAD向量为零

    with open(embed_path, 'r', encoding='utf-8') as f:  # with open自动管理文件打开和关闭
        for line in f:
            parts = line.strip().split()
            word = parts[0]
            if word in vocab:
                embeddings[vocab[word]] = np.array(parts[1:], dtype=np.float32)  # 切片操作取子序列  # np.array创建NumPy数组

    return torch.tensor(embeddings, dtype=torch.float32)

# 使用预训练嵌入初始化
# pretrained = load_pretrained_embeddings(vocab)
# model.embedding.weight.data.copy_(pretrained)
# model.embedding.weight.requires_grad = True  # 可选：是否微调