h_t = tanh(W_hh · h_{t-1} + W_xh · x_t + b_h)
y_t = W_hy · h_t + b_y

import torch
import torch.nn as nn

# 基本RNN
rnn = nn.RNN(input_size=10, hidden_size=20, num_layers=1, batch_first=True)

# 输入：(batch, seq_len, input_size)
x = torch.randn(32, 50, 10)  # 批量32，序列长度50，输入维度10

output, h_n = rnn(x)
print(f"输出形状: {output.shape}")  # (32, 50, 20)
print(f"最终隐藏状态: {h_n.shape}")  # (1, 32, 20)

import numpy as np

class SimpleRNN:
    def __init__(self, input_size, hidden_size):
        self.W_xh = np.random.randn(input_size, hidden_size) * 0.01
        self.W_hh = np.random.randn(hidden_size, hidden_size) * 0.01
        self.b_h = np.zeros(hidden_size)
        self.W_hy = np.random.randn(hidden_size, input_size) * 0.01
        self.b_y = np.zeros(input_size)
    
    def forward(self, x_seq):
        """x_seq: (seq_len, input_size)"""
        seq_len = x_seq.shape[0]
        h = np.zeros(self.W_hh.shape[0])
        
        h_states = [h]
        y_seq = []
        
        for t in range(seq_len):
            x_t = x_seq[t]
            h = np.tanh(x_t @ self.W_xh + h @ self.W_hh + self.b_h)
            y = h @ self.W_hy + self.b_y
            h_states.append(h)
            y_seq.append(y)
        
        return np.array(y_seq), h_states

# 测试
rnn = SimpleRNN(input_size=10, hidden_size=20)
x = np.random.randn(50, 10)
y, h_states = rnn.forward(x)
print(f"输出形状: {y.shape}")
print(f"隐藏状态数量: {len(h_states)}")

# BPTT将RNN在时间上展开，转化为一个很深的前馈网络
# 然后应用标准的反向传播

# 梯度问题：
# ∂L/∂W_hh 涉及连乘 ∏(t=i to j) ∂h_t/∂h_{t-1}
# 每个 ∂h_t/∂h_{t-1} ≈ W_hh * diag(1 - h_t²)
# 连乘导致梯度指数级增长（爆炸）或衰减（消失）

# 梯度裁剪解决梯度爆炸
def clip_gradient_norm(parameters, max_norm):
    total_norm = 0
    for p in parameters:
        if p.grad is not None:
            total_norm += p.grad.data.norm() ** 2
    total_norm = total_norm ** 0.5
    clip_coef = max_norm / (total_norm + 1e-6)
    if clip_coef < 1:
        for p in parameters:
            if p.grad is not None:
                p.grad.data.mul_(clip_coef)

遗忘门：f_t = σ(W_f · [h_{t-1}, x_t] + b_f)
输入门：i_t = σ(W_i · [h_{t-1}, x_t] + b_i)
候选值：C̃_t = tanh(W_C · [h_{t-1}, x_t] + b_C)
细胞状态：C_t = f_t * C_{t-1} + i_t * C̃_t
输出门：o_t = σ(W_o · [h_{t-1}, x_t] + b_o)
隐藏状态：h_t = o_t * tanh(C_t)

import torch
import torch.nn as nn

# LSTM
lstm = nn.LSTM(input_size=10, hidden_size=20, num_layers=1, batch_first=True)

x = torch.randn(32, 50, 10)
output, (h_n, c_n) = lstm(x)

print(f"输出形状: {output.shape}")      # (32, 50, 20)
print(f"隐藏状态: {h_n.shape}")         # (1, 32, 20)
print(f"细胞状态: {c_n.shape}")         # (1, 32, 20)

# 手动实现LSTM单元
class LSTMCell:
    def __init__(self, input_size, hidden_size):
        self.hidden_size = hidden_size
        combined_size = input_size + hidden_size
        
        # 合并所有门的权重
        self.W = np.random.randn(combined_size, 4 * hidden_size) * 0.01
        self.b = np.zeros(4 * hidden_size)
    
    def forward(self, x, h_prev, c_prev):
        combined = np.concatenate([x, h_prev])
        gates = combined @ self.W + self.b
        
        # 分割四个门
        f = self.sigmoid(gates[:self.hidden_size])           # 遗忘门
        i = self.sigmoid(gates[self.hidden_size:2*self.hidden_size])  # 输入门
        g = np.tanh(gates[2*self.hidden_size:3*self.hidden_size])     # 候选值
        o = self.sigmoid(gates[3*self.hidden_size:])         # 输出门
        
        c_new = f * c_prev + i * g
        h_new = o * np.tanh(c_new)
        
        return h_new, c_new
    
    def sigmoid(self, x):
        return 1 / (1 + np.exp(-np.clip(x, -500, 500)))

# LSTM的细胞状态更新：C_t = f_t * C_{t-1} + i_t * C̃_t
# 梯度流：∂C_t/∂C_{t-1} = f_t
# 遗忘门f_t可以学习保持为1，梯度不会消失

# 对比标准RNN：h_t = tanh(W_hh * h_{t-1} + ...)
# 梯度流：∂h_t/∂h_{t-1} = W_hh * diag(1 - h_t²)
# 连乘导致梯度消失

# 可视化门控值
lstm = nn.LSTM(10, 20, batch_first=True)
x = torch.randn(1, 100, 10)
output, (h, c) = lstm(x)

# 提取门控值需要修改内部实现
# 这里展示细胞状态的变化
print(f"细胞状态范围: [{c.min().item():.3f}, {c.max().item():.3f}]")
print(f"隐藏状态范围: [{h.min().item():.3f}, {h.max().item():.3f}]")

# GRU是LSTM的简化版本，只有两个门
# 重置门：r_t = σ(W_r · [h_{t-1}, x_t])
# 更新门：z_t = σ(W_z · [h_{t-1}, x_t])
# 候选值：h̃_t = tanh(W · [r_t * h_{t-1}, x_t])
# 隐藏状态：h_t = (1 - z_t) * h_{t-1} + z_t * h̃_t

gru = nn.GRU(input_size=10, hidden_size=20, batch_first=True)
x = torch.randn(32, 50, 10)
output, h_n = gru(x)

print(f"GRU输出: {output.shape}")
print(f"GRU隐藏状态: {h_n.shape}")

# 双向RNN：同时从前往后和从后往前处理序列
birnn = nn.RNN(input_size=10, hidden_size=20, bidirectional=True, batch_first=True)
x = torch.randn(32, 50, 10)
output, h_n = birnn(x)
print(f"双向RNN输出: {output.shape}")  # (32, 50, 40) - 前向20 + 后向20

# 深度RNN：堆叠多层
deep_rnn = nn.LSTM(input_size=10, hidden_size=20, num_layers=3, batch_first=True)
output, (h_n, c_n) = deep_rnn(x)
print(f"3层LSTM隐藏状态: {h_n.shape}")  # (3, 32, 20)

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

# 简单的字符级语言模型
class CharRNN(nn.Module):
    def __init__(self, vocab_size, embed_size, hidden_size, num_layers):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_size)
        self.lstm = nn.LSTM(embed_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, vocab_size)
    
    def forward(self, x, hidden=None):
        x = self.embedding(x)
        output, hidden = self.lstm(x, hidden)
        output = self.fc(output)
        return output, hidden

# 训练数据（简单的文本）
text = "hello world, this is a simple character level language model example"
chars = list(set(text))
char2idx = {c: i for i, c in enumerate(chars)}
idx2char = {i: c for i, c in enumerate(chars)}
vocab_size = len(chars)

# 准备数据
seq_length = 20
inputs, targets = [], []
for i in range(len(text) - seq_length):
    inputs.append([char2idx[c] for c in text[i:i+seq_length]])
    targets.append([char2idx[c] for c in text[i+1:i+seq_length+1]])

inputs = torch.tensor(inputs)
targets = torch.tensor(targets)

# 训练
model = CharRNN(vocab_size, embed_size=32, hidden_size=64, num_layers=2)
criterion = nn.CrossEntropyLoss()
optimizer = optim.Adam(model.parameters(), lr=0.005)

for epoch in range(100):
    total_loss = 0
    for i, (inp, tgt) in enumerate(zip(inputs, targets)):
        optimizer.zero_grad()
        output, _ = model(inp.unsqueeze(0))
        loss = criterion(output.view(-1, vocab_size), tgt)
        loss.backward()
        torch.nn.utils.clip_grad_norm_(model.parameters(), 5.0)
        optimizer.step()
        total_loss += loss.item()
    
    if (epoch + 1) % 20 == 0:
        print(f"Epoch {epoch+1}, Loss: {total_loss/len(inputs):.4f}")

# 生成文本
def generate(model, start_char, length=100):
    model.eval()
    x = torch.tensor([[char2idx[start_char]]])
    generated = [start_char]
    hidden = None
    
    for _ in range(length):
        output, hidden = model(x, hidden)
        probs = torch.softmax(output[0, -1], dim=0)
        next_idx = torch.multinomial(probs, 1).item()
        generated.append(idx2char[next_idx])
        x = torch.tensor([[next_idx]])
    
    return ''.join(generated)

print("\n生成文本:")
print(generate(model, 'h', 200))

import torch
import torch.nn as nn

class SequenceClassifier(nn.Module):
    def __init__(self, input_size, hidden_size, num_classes, num_layers=2):
        super().__init__()
        self.lstm = nn.LSTM(input_size, hidden_size, num_layers, 
                           batch_first=True, bidirectional=True, dropout=0.2)
        self.fc = nn.Linear(hidden_size * 2, num_classes)  # 双向
    
    def forward(self, x):
        output, (h_n, _) = self.lstm(x)
        # 拼接前向和后向的最终隐藏状态
        h_forward = h_n[-2]   # 最后一层的前向
        h_backward = h_n[-1]  # 最后一层的后向
        h_combined = torch.cat([h_forward, h_backward], dim=1)
        return self.fc(h_combined)

# 使用示例
model = SequenceClassifier(input_size=10, hidden_size=32, num_classes=5)
x = torch.randn(32, 50, 10)  # 批量32，序列长度50，特征10
output = model(x)
print(f"分类输出: {output.shape}")  # (32, 5)