import numpy as np

class NeuralNetwork:
    """从零实现的多层感知机"""
    
    def __init__(self, layer_sizes):
        """
        初始化网络
        Args:
            layer_sizes: 列表，包含每层的神经元数量
                        例如 [784, 256, 128, 10] 表示输入784维，
                        两个隐藏层分别为256和128维，输出10维
        """
        self.num_layers = len(layer_sizes)
        self.layer_sizes = layer_sizes
        
        # He初始化
        self.weights = []
        self.biases = []
        for i in range(self.num_layers - 1):
            w = np.random.randn(layer_sizes[i], layer_sizes[i+1]) * np.sqrt(2.0 / layer_sizes[i])
            b = np.zeros((1, layer_sizes[i+1]))
            self.weights.append(w)
            self.biases.append(b)
    
    def relu(self, x):
        return np.maximum(0, x)
    
    def relu_derivative(self, x):
        return (x > 0).astype(float)
    
    def softmax(self, x):
        # 数值稳定的softmax
        exp_x = np.exp(x - np.max(x, axis=1, keepdims=True))
        return exp_x / np.sum(exp_x, axis=1, keepdims=True)
    
    def forward(self, X):
        """
        前向传播
        Args:
            X: 输入数据，形状为 (batch_size, input_dim)
        Returns:
            output: 输出预测
        """
        self.activations = [X]  # 保存每层的激活值
        self.pre_activations = []  # 保存每层的预激活值
        
        current_input = X
        for i in range(self.num_layers - 1):
            z = current_input @ self.weights[i] + self.biases[i]
            self.pre_activations.append(z)
            
            if i < self.num_layers - 2:
                # 隐藏层使用ReLU
                a = self.relu(z)
            else:
                # 输出层使用Softmax
                a = self.softmax(z)
            
            self.activations.append(a)
            current_input = a
        
        return self.activations[-1]
    
    def backward(self, X, y_onehot, learning_rate=0.01):
        """
        反向传播和参数更新
        Args:
            X: 输入数据
            y_onehot: one-hot编码的真实标签
            learning_rate: 学习率
        """
        batch_size = X.shape[0]
        
        # 计算输出层梯度
        output = self.activations[-1]
        delta = output - y_onehot  # Softmax + CrossEntropy的梯度
        
        # 反向传播
        for i in range(self.num_layers - 2, -1, -1):
            # 计算当前层的权重和偏置梯度
            dw = self.activations[i].T @ delta / batch_size
            db = np.sum(delta, axis=0, keepdims=True) / batch_size
            
            # 计算传播到前一层的梯度
            if i > 0:
                delta = (delta @ self.weights[i].T) * self.relu_derivative(self.pre_activations[i-1])
            
            # 更新参数
            self.weights[i] -= learning_rate * dw
            self.biases[i] -= learning_rate * db
    
    def compute_loss(self, y_pred, y_onehot):
        """计算交叉熵损失"""
        eps = 1e-15
        y_pred_clipped = np.clip(y_pred, eps, 1 - eps)
        loss = -np.sum(y_onehot * np.log(y_pred_clipped)) / y_onehot.shape[0]
        return loss
    
    def train(self, X, y, epochs=100, batch_size=32, learning_rate=0.01):
        """训练循环"""
        num_samples = X.shape[0]
        
        for epoch in range(epochs):
            # Shuffle数据
            indices = np.random.permutation(num_samples)
            X_shuffled = X[indices]
            y_shuffled = y[indices]
            
            total_loss = 0
            num_batches = 0
            
            for start in range(0, num_samples, batch_size):
                end = min(start + batch_size, num_samples)
                X_batch = X_shuffled[start:end]
                y_batch = y_shuffled[start:end]
                
                # 前向传播
                y_pred = self.forward(X_batch)
                
                # 计算损失
                loss = self.compute_loss(y_pred, y_batch)
                total_loss += loss
                num_batches += 1
                
                # 反向传播
                self.backward(X_batch, y_batch, learning_rate)
            
            if (epoch + 1) % 10 == 0:
                avg_loss = total_loss / num_batches
                print(f"Epoch {epoch+1}/{epochs}, Loss: {avg_loss:.4f}")
    
    def predict(self, X):
        """预测类别"""
        probs = self.forward(X)
        return np.argmax(probs, axis=1)


# 测试
if __name__ == "__main__":
    # 生成简单的分类数据
    from sklearn.datasets import make_classification
    from sklearn.preprocessing import StandardScaler
    
    X, y = make_classification(n_samples=1000, n_features=20, n_classes=2, random_state=42)
    scaler = StandardScaler()
    X = scaler.fit_transform(X)
    
    # One-hot编码
    y_onehot = np.zeros((len(y), 2))
    y_onehot[np.arange(len(y)), y] = 1
    
    # 创建并训练网络
    nn = NeuralNetwork([20, 64, 32, 2])
    nn.train(X, y_onehot, epochs=50, batch_size=32, learning_rate=0.01)
    
    # 评估
    predictions = nn.predict(X)
    accuracy = np.mean(predictions == y)
    print(f"Training Accuracy: {accuracy:.4f}")

import torch
import torch.nn as nn

class SwishFunction(torch.autograd.Function):
    """
    自定义Swish激活函数的手动梯度实现
    Swish: f(x) = x * sigmoid(x)
    """
    
    @staticmethod
    def forward(ctx, x):
        # 保存输入以供反向传播使用
        ctx.save_for_backward(x)
        return x * torch.sigmoid(x)
    
    @staticmethod
    def backward(ctx, grad_output):
        x, = ctx.saved_tensors
        sigmoid_x = torch.sigmoid(x)
        
        # Swish的导数: f'(x) = sigmoid(x) + x * sigmoid(x) * (1 - sigmoid(x))
        grad_input = grad_output * (sigmoid_x + x * sigmoid_x * (1 - sigmoid_x))
        return grad_input


class Swish(nn.Module):
    """Swish激活函数模块"""
    def forward(self, x):
        return SwishFunction.apply(x)


# 验证自定义实现的正确性
def test_swish():
    x = torch.randn(10, requires_grad=True)
    
    # 使用自定义实现
    y_custom = SwishFunction.apply(x)
    y_custom.sum().backward()
    grad_custom = x.grad.clone()
    
    # 重置梯度
    x.grad = None
    
    # 使用数值梯度验证
    eps = 1e-5
    grad_numerical = torch.zeros_like(x)
    for i in range(len(x)):
        x_plus = x.clone()
        x_minus = x.clone()
        x_plus[i] += eps
        x_minus[i] -= eps
        
        y_plus = (x_plus * torch.sigmoid(x_plus)).sum()
        y_minus = (x_minus * torch.sigmoid(x_minus)).sum()
        
        grad_numerical[i] = (y_plus - y_minus) / (2 * eps)
    
    print("Analytical gradient:", grad_custom)
    print("Numerical gradient:", grad_numerical)
    print("Difference:", torch.abs(grad_custom - grad_numerical).max().item())


# 使用内置的SiLU（等同于Swish）
def test_builtin():
    x = torch.randn(10, requires_grad=True)
    silu = nn.SiLU()
    y = silu(x)
    y.sum().backward()
    print("Built-in SiLU gradient:", x.grad)


if __name__ == "__main__":
    test_swish()
    test_builtin()

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

class RNNWithGradientClipping(nn.Module):
    """使用梯度裁剪的RNN模型"""
    
    def __init__(self, vocab_size, embed_size, hidden_size, num_layers):
        super().__init__()
        self.embedding = nn.Embedding(vocab_size, embed_size)
        self.rnn = nn.LSTM(embed_size, hidden_size, num_layers, batch_first=True)
        self.fc = nn.Linear(hidden_size, vocab_size)
        self.hidden_size = hidden_size
        self.num_layers = num_layers
    
    def forward(self, x, hidden=None):
        embedded = self.embedding(x)
        output, hidden = self.rnn(embedded, hidden)
        logits = self.fc(output)
        return logits, hidden
    
    def init_hidden(self, batch_size, device):
        h0 = torch.zeros(self.num_layers, batch_size, self.hidden_size).to(device)
        c0 = torch.zeros(self.num_layers, batch_size, self.hidden_size).to(device)
        return (h0, c0)


def train_with_clipping(model, data_loader, criterion, optimizer, device, max_grad_norm=1.0):
    """
    带梯度裁剪的训练函数
    Args:
        model: RNN模型
        data_loader: 数据加载器
        criterion: 损失函数
        optimizer: 优化器
        device: 计算设备
        max_grad_norm: 梯度范数的最大允许值
    """
    model.train()
    total_loss = 0
    
    for batch_idx, (inputs, targets) in enumerate(data_loader):
        inputs = inputs.to(device)
        targets = targets.to(device)
        batch_size = inputs.size(0)
        
        # 初始化隐藏状态
        hidden = model.init_hidden(batch_size, device)
        
        # 前向传播
        logits, _ = model(inputs, hidden)
        
        # 计算损失（忽略填充位置）
        loss = criterion(logits.view(-1, logits.size(-1)), targets.view(-1))
        
        # 反向传播
        optimizer.zero_grad()
        loss.backward()
        
        # 梯度裁剪
        torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=max_grad_norm, norm_type=2)
        
        # 参数更新
        optimizer.step()
        
        total_loss += loss.item()
        
        if (batch_idx + 1) % 50 == 0:
            avg_loss = total_loss / (batch_idx + 1)
            print(f"Batch {batch_idx+1}, Avg Loss: {avg_loss:.4f}")
    
    return total_loss / len(data_loader)


def monitor_gradients(model):
    """监控梯度统计信息"""
    grad_norms = []
    for name, param in model.named_parameters():
        if param.grad is not None:
            grad_norm = param.grad.norm().item()
            grad_norms.append((name, grad_norm))
    
    # 打印梯度范数最大的前5个参数
    grad_norms.sort(key=lambda x: x[1], reverse=True)
    print("\nTop 5 gradient norms:")
    for name, norm in grad_norms[:5]:
        print(f"  {name}: {norm:.6f}")


# 使用示例
if __name__ == "__main__":
    # 创建模拟数据
    vocab_size = 1000
    seq_length = 50
    batch_size = 32
    
    inputs = torch.randint(0, vocab_size, (batch_size, seq_length))
    targets = torch.randint(0, vocab_size, (batch_size, seq_length))
    
    # 创建模型
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    model = RNNWithGradientClipping(vocab_size, 128, 256, 2).to(device)
    
    criterion = nn.CrossEntropyLoss(ignore_index=0)
    optimizer = optim.Adam(model.parameters(), lr=0.001)
    
    # 模拟训练步骤
    logits, _ = model(inputs.to(device))
    loss = criterion(logits.view(-1, vocab_size), targets.view(-1).to(device))
    loss.backward()
    
    # 监控梯度
    monitor_gradients(model)
    
    # 应用梯度裁剪
    torch.nn.utils.clip_grad_norm_(model.parameters(), max_norm=1.0)
    print("\nAfter clipping:")
    monitor_gradients(model)