21_RNN与LSTM：序列建模的经典方法

输入序列: x1, x2, x3, ..., xt
隐藏状态: h0, h1, h2, ..., ht
输出序列: y1, y2, y3, ..., yt

计算过程:
1. 初始化隐藏状态: h0 = 0
2. 对于每个时间步t:
   ht = tanh(Wxh*x1 + Whh*ht-1 + bh)
   yt = Why*ht + by

ht = f(Wxh*x1 + Whh*ht-1 + bh)
yt = g(Why*ht + by)

[输入] x1 → [RNN单元] → h1 → y1
          ↓
[输入] x2 → [RNN单元] → h2 → y2
          ↓
[输入] x3 → [RNN单元] → h3 → y3
          ↓
          ...

输入: xt（当前输入）, ht-1（前一时刻隐藏状态）, ct-1（前一时刻细胞状态）
输出: ht（当前隐藏状态）, ct（当前细胞状态）

ft = σ(Wf*xt + Uf*ht-1 + bf)

it = σ(Wi*xt + Ui*ht-1 + bi)
ct_hat = tanh(Wc*xt + Uc*ht-1 + bc)

ct = ft * ct-1 + it * ct_hat

ot = σ(Wo*xt + Uo*ht-1 + bo)
ht = ot * tanh(ct)

更新门: zt = σ(Wz*xt + Uz*ht-1 + bz)
重置门: rt = σ(Wr*xt + Ur*ht-1 + br)
候选隐藏状态: ht_hat = tanh(Wh*xt + rt*(Uh*ht-1) + bh)
隐藏状态更新: ht = (1 - zt)*ht-1 + zt*ht_hat

前向LSTM: → h1^f, h2^f, h3^f, ..., ht^f →
反向LSTM: ← h1^b, h2^b, h3^b, ..., ht^b ←

组合隐藏状态: ht = [ht^f, ht^b]

输入层 → [LSTM层1] → [LSTM层2] → ... → [LSTM层n] → 输出层

import torch
import torch.nn as nn
import torch.optim as optim

# 定义RNN模型
class SimpleRNN(nn.Module):
    def __init__(self, input_size, hidden_size, output_size):
        super(SimpleRNN, self).__init__()
        self.hidden_size = hidden_size
        # RNN层
        self.rnn = nn.RNN(input_size, hidden_size, batch_first=True)
        # 输出层
        self.fc = nn.Linear(hidden_size, output_size)
    
    def forward(self, x, hidden=None):
        # x: [batch_size, seq_len, input_size]
        batch_size, seq_len, _ = x.size()
        
        # 初始化隐藏状态
        if hidden is None:
            hidden = torch.zeros(1, batch_size, self.hidden_size).to(x.device)
        
        # RNN前向传播
        out, hidden = self.rnn(x, hidden)
        
        # 输出层
        out = self.fc(out)
        
        return out, hidden

# 示例使用
input_size = 10
hidden_size = 20
output_size = 5
batch_size = 3
seq_len = 7

model = SimpleRNN(input_size, hidden_size, output_size)
input_tensor = torch.randn(batch_size, seq_len, input_size)
output, hidden = model(input_tensor)
print(f"Output shape: {output.shape}")  # [3, 7, 5]
print(f"Hidden shape: {hidden.shape}")  # [1, 3, 20]

import torch
import torch.nn as nn
import torch.optim as optim

# 定义LSTM模型
class SimpleLSTM(nn.Module):
    def __init__(self, input_size, hidden_size, output_size, num_layers=1, bidirectional=False):
        super(SimpleLSTM, self).__init__()
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.bidirectional = bidirectional
        self.num_directions = 2 if bidirectional else 1
        
        # LSTM层
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, 
                           batch_first=True, bidirectional=bidirectional)
        # 输出层
        self.fc = nn.Linear(hidden_size * self.num_directions, output_size)
    
    def forward(self, x, hidden=None):
        # x: [batch_size, seq_len, input_size]
        batch_size, seq_len, _ = x.size()
        
        # 初始化隐藏状态和细胞状态
        if hidden is None:
            h0 = torch.zeros(self.num_layers * self.num_directions, batch_size, 
                            self.hidden_size).to(x.device)
            c0 = torch.zeros(self.num_layers * self.num_directions, batch_size, 
                            self.hidden_size).to(x.device)
        else:
            h0, c0 = hidden
        
        # LSTM前向传播
        out, (h_n, c_n) = self.lstm(x, (h0, c0))
        
        # 输出层
        out = self.fc(out)
        
        return out, (h_n, c_n)

# 示例使用
input_size = 10
hidden_size = 20
output_size = 5
batch_size = 3
seq_len = 7

# 单向LSTM
model_uni = SimpleLSTM(input_size, hidden_size, output_size)
input_tensor = torch.randn(batch_size, seq_len, input_size)
output_uni, (h_uni, c_uni) = model_uni(input_tensor)
print(f"单向LSTM输出形状: {output_uni.shape}")  # [3, 7, 5]
print(f"单向LSTM隐藏状态形状: {h_uni.shape}")  # [1, 3, 20]

# 双向LSTM
model_bi = SimpleLSTM(input_size, hidden_size, output_size, bidirectional=True)
output_bi, (h_bi, c_bi) = model_bi(input_tensor)
print(f"双向LSTM输出形状: {output_bi.shape}")  # [3, 7, 5]
print(f"双向LSTM隐藏状态形状: {h_bi.shape}")  # [2, 3, 20]

import torch
import torch.nn as nn
import torch.optim as optim
import torchtext
from torchtext.datasets import IMDB
from torchtext.data.utils import get_tokenizer
from collections import Counter
from torchtext.vocab import Vocab
import numpy as np
from torch.utils.data import DataLoader
from torch.nn.utils.rnn import pad_sequence

# 1. 数据预处理
# 分词器
tokenizer = get_tokenizer('basic_english')

# 构建词汇表
def build_vocab(dataset, max_tokens=10000):
    counter = Counter()
    for label, text in dataset:
        counter.update(tokenizer(text))
    return Vocab(counter, max_size=max_tokens, specials=("<unk>", "<pad>", "<bos>", "<eos>"))

# 加载数据集
train_iter = IMDB(split='train')
vocab = build_vocab(train_iter)

# 数据转换函数
def text_pipeline(x):
    return [vocab[token] for token in tokenizer(x)]

def label_pipeline(x):
    return 1 if x == 'pos' else 0

# 批量处理函数
def collate_batch(batch):
    label_list, text_list, lengths = [], [], []
    for (_label, _text) in batch:
        label_list.append(label_pipeline(_label))
        processed_text = torch.tensor(text_pipeline(_text), dtype=torch.int64)
        text_list.append(processed_text)
        lengths.append(len(processed_text))
    # 填充序列
    padded_text = pad_sequence(text_list, batch_first=True, padding_value=vocab["<pad>"])
    return torch.tensor(label_list), padded_text, torch.tensor(lengths)

# 数据加载器
train_iter, test_iter = IMDB()
train_dataset = list(train_iter)
test_dataset = list(test_iter)
train_dataloader = DataLoader(train_dataset, batch_size=64, shuffle=True, collate_fn=collate_batch)
test_dataloader = DataLoader(test_dataset, batch_size=64, shuffle=False, collate_fn=collate_batch)

# 2. 定义模型
class LSTMClassifier(nn.Module):
    def __init__(self, vocab_size, embedding_dim, hidden_dim, output_dim):
        super(LSTMClassifier, self).__init__()
        self.embedding = nn.Embedding(vocab_size, embedding_dim)
        self.lstm = nn.LSTM(embedding_dim, hidden_dim, batch_first=True)
        self.fc = nn.Linear(hidden_dim, output_dim)
        self.sigmoid = nn.Sigmoid()
    
    def forward(self, text, text_lengths):
        embedded = self.embedding(text)
        packed = nn.utils.rnn.pack_padded_sequence(embedded, text_lengths, batch_first=True, enforce_sorted=False)
        output, (hidden, cell) = self.lstm(packed)
        hidden = hidden.squeeze(0)
        return self.sigmoid(self.fc(hidden))

# 3. 训练模型
# 超参数设置
vocab_size = len(vocab)
embedding_dim = 100
hidden_dim = 256
output_dim = 1
learning_rate = 1e-3
num_epochs = 5

# 初始化模型、损失函数和优化器
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = LSTMClassifier(vocab_size, embedding_dim, hidden_dim, output_dim).to(device)
criterion = nn.BCELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# 训练函数
def train(model, iterator, optimizer, criterion):
    model.train()
    epoch_loss = 0
    epoch_acc = 0
    
    for labels, texts, lengths in iterator:
        labels = labels.to(device).float().unsqueeze(1)
        texts = texts.to(device)
        lengths = lengths.to(device)
        
        optimizer.zero_grad()
        
        predictions = model(texts, lengths)
        loss = criterion(predictions, labels)
        
        # 计算准确率
        rounded_preds = torch.round(predictions)
        correct = (rounded_preds == labels).float()
        acc = correct.sum() / len(correct)
        
        loss.backward()
        optimizer.step()
        
        epoch_loss += loss.item()
        epoch_acc += acc.item()
    
    return epoch_loss / len(iterator), epoch_acc / len(iterator)

# 评估函数
def evaluate(model, iterator, criterion):
    model.eval()
    epoch_loss = 0
    epoch_acc = 0
    
    with torch.no_grad():
        for labels, texts, lengths in iterator:
            labels = labels.to(device).float().unsqueeze(1)
            texts = texts.to(device)
            lengths = lengths.to(device)
            
            predictions = model(texts, lengths)
            loss = criterion(predictions, labels)
            
            # 计算准确率
            rounded_preds = torch.round(predictions)
            correct = (rounded_preds == labels).float()
            acc = correct.sum() / len(correct)
            
            epoch_loss += loss.item()
            epoch_acc += acc.item()
    
    return epoch_loss / len(iterator), epoch_acc / len(iterator)

# 训练循环
for epoch in range(num_epochs):
    train_loss, train_acc = train(model, train_dataloader, optimizer, criterion)
    valid_loss, valid_acc = evaluate(model, test_dataloader, criterion)
    
    print(f'Epoch: {epoch+1:02}')
    print(f'\tTrain Loss: {train_loss:.3f} | Train Acc: {train_acc*100:.2f}%')
    print(f'\t Val. Loss: {valid_loss:.3f} |  Val. Acc: {valid_acc*100:.2f}%')

# 4. 模型应用
def predict_sentiment(model, sentence, vocab, tokenizer):
    model.eval()
    tokens = tokenizer(sentence)
    indexed = [vocab[token] for token in tokens]
    tensor = torch.LongTensor(indexed).unsqueeze(0).to(device)
    length = torch.LongTensor([len(indexed)]).to(device)
    prediction = model(tensor, length)
    return prediction.item()

# 测试示例
example_sentences = [
    "This movie was fantastic! I really enjoyed it and would watch it again.",
    "I hated this movie. It was boring and predictable."
]

for sentence in example_sentences:
    sentiment = predict_sentiment(model, sentence, vocab, tokenizer)
    print(f'Sentence: {sentence}')
    print(f'Sentiment Score: {sentiment:.4f}')
    print(f'Predicted Class: {"Positive" if sentiment >= 0.5 else "Negative"}\n')

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler

# 1. 生成模拟数据
def generate_time_series(n_samples=1000, seq_length=50):
    # 生成正弦波数据
    t = np.linspace(0, 20 * np.pi, n_samples)
    data = np.sin(t) + np.sin(2*t) + 0.1*np.random.randn(n_samples)
    
    # 创建序列数据
    X = []
    y = []
    for i in range(len(t) - seq_length):
        X.append(data[i:i+seq_length])
        y.append(data[i+seq_length])
    
    return np.array(X), np.array(y)

# 生成数据
X, y = generate_time_series()

# 数据预处理
scaler = MinMaxScaler(feature_range=(-1, 1))
X_scaled = scaler.fit_transform(X.reshape(-1, X.shape[-1])).reshape(X.shape)
y_scaled = scaler.fit_transform(y.reshape(-1, 1)).flatten()

# 划分训练集和测试集
train_size = int(0.8 * len(X))
X_train, X_test = X_scaled[:train_size], X_scaled[train_size:]
y_train, y_test = y_scaled[:train_size], y_scaled[train_size:]

# 转换为PyTorch张量
X_train = torch.FloatTensor(X_train).unsqueeze(2)
X_test = torch.FloatTensor(X_test).unsqueeze(2)
y_train = torch.FloatTensor(y_train)
y_test = torch.FloatTensor(y_test)

# 2. 定义LSTM模型
class LSTMTimeSeries(nn.Module):
    def __init__(self, input_size, hidden_size, num_layers, output_size):
        super(LSTMTimeSeries, self).__init__()
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, output_size)
    
    def forward(self, x):
        # x: [batch_size, seq_len, input_size]
        batch_size = x.size(0)
        
        # 初始化隐藏状态和细胞状态
        h0 = torch.zeros(self.num_layers, batch_size, self.hidden_size).to(x.device)
        c0 = torch.zeros(self.num_layers, batch_size, self.hidden_size).to(x.device)
        
        # LSTM前向传播
        out, _ = self.lstm(x, (h0, c0))
        
        # 取最后一个时间步的输出
        out = self.fc(out[:, -1, :])
        
        return out

# 3. 训练模型
# 超参数设置
input_size = 1
hidden_size = 64
num_layers = 2
output_size = 1
learning_rate = 1e-3
num_epochs = 100
batch_size = 64

# 初始化模型、损失函数和优化器
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
model = LSTMTimeSeries(input_size, hidden_size, num_layers, output_size).to(device)
criterion = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# 创建数据加载器
train_data = torch.utils.data.TensorDataset(X_train, y_train)
train_loader = torch.utils.data.DataLoader(train_data, batch_size=batch_size, shuffle=True)

# 训练循环
for epoch in range(num_epochs):
    model.train()
    epoch_loss = 0
    
    for X_batch, y_batch in train_loader:
        X_batch = X_batch.to(device)
        y_batch = y_batch.to(device).unsqueeze(1)
        
        optimizer.zero_grad()
        
        outputs = model(X_batch)
        loss = criterion(outputs, y_batch)
        
        loss.backward()
        optimizer.step()
        
        epoch_loss += loss.item()
    
    if (epoch+1) % 10 == 0:
        print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {epoch_loss/len(train_loader):.4f}')

# 4. 评估模型
model.eval()
with torch.no_grad():
    X_test = X_test.to(device)
    y_pred = model(X_test).cpu().numpy()

# 逆变换预测结果
y_pred_original = scaler.inverse_transform(y_pred.reshape(-1, 1)).flatten()
y_test_original = scaler.inverse_transform(y_test.reshape(-1, 1)).flatten()

# 计算评估指标
from sklearn.metrics import mean_squared_error, mean_absolute_error

mse = mean_squared_error(y_test_original, y_pred_original)
mae = mean_absolute_error(y_test_original, y_pred_original)

print(f'MSE: {mse:.4f}')
print(f'MAE: {mae:.4f}')

# 5. 可视化结果
plt.figure(figsize=(12, 6))
plt.plot(y_test_original, label='True Values')
plt.plot(y_pred_original, label='Predicted Values')
plt.title('Time Series Prediction with LSTM')
plt.xlabel('Time')
plt.ylabel('Value')
plt.legend()
plt.show()

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