[​AI学习笔记]神经网络架构演进：从MLP到DeepSeek的混合专家系统

# 2025年预测的典型模型架构代码示意
class 2025_MoE(nn.Module):
    def __init__(self):
        self.quantum_attention = QuantumAttentionLayer() # 量子注意力层
        self.sparse_experts = DynamicSparseExperts()     # 动态稀疏专家
        self.neuromorphic_act = SNNActivation()          # 脉冲神经网络激活
        
    def forward(self, x):
        x = self.quantum_attention(x)
        x = self.sparse_experts(x) 
        return self.neuromorphic_act(x)

class LATRouting(nn.Module):
    def __init__(self, dim, num_experts, k=2):
        super().__init__()
        self.query = nn.Linear(dim, dim)
        self.expert_keys = nn.Parameter(torch.randn(num_experts, dim))
        self.k = k
        
    def forward(self, x):
        # x: [B, T, D]
        queries = self.query(x)  # [B, T, D]
        scores = torch.einsum('btd, ed -> bte', queries, self.expert_keys)  # [B, T, E]
        topk_scores, topk_idx = torch.topk(scores, k=self.k, dim=-1)  # [B, T, k]
        
        # 动态容量计算
        expert_counts = torch.bincount(topk_idx.flatten(), minlength=num_experts)
        capacity = (expert_counts.float() / x.size(1)).cumsum(0)
        
        # 门控权重归一化
        mask = torch.zeros_like(scores).scatter(-1, topk_idx, 1)
        gate_weights = F.softmax(topk_scores, dim=-1) * mask.sum(-1, keepdim=True)
        
        return gate_weights, topk_idx, capacity

class DynamicCapacityScheduler:  
    def __init__(self, total_capacity, max_util=0.9):  
        self.total_cap = total_capacity  
        self.util_thresh = max_util  
        
    def update(self, expert_usage):  
        """ expert_usage: [E] 各专家当前利用率 """  
        over_utilized = expert_usage > self.util_thresh  
        under_utilized = expert_usage < 0.3 * self.util_thresh  
        
        # 容量重分配  
        delta = torch.where(over_utilized,  
                           self.total_cap * 0.1 / sum(over_utilized),  
                           -self.total_cap * 0.05 / sum(under_utilized))  
        new_capacity = self.base_capacity + delta  
        
        return torch.clamp(new_capacity, min=32, max=512)  

class HierarchicalGradientCompression:  
    def compress(self, grad):  
        if grad.type == 'router':  
            # 8-bit量化  
            grad_q = torch.quantize_per_tensor(grad, scale=1e-3, zero_point=0, dtype=torch.qint8)  
            return grad_q.dequantize().float()  
        elif grad.type == 'expert':  
            # Top-K稀疏化  
            k = int(0.1 * grad.numel())  
            values, indices = torch.topk(grad.abs().flatten(), k)  
            return (values, indices)  
        else:  
            # 块浮点压缩  
            block_size = 64  
            blocks = grad.view(-1, block_size)  
            max_val = blocks.abs().max(dim=1).values  
            scale = max_val / 127.0  
            scaled_blocks = (blocks / scale.unsqueeze(1)).round()  
            return (scaled_blocks.char(), scale)  

class ExpertProfiler:  
    def __init__(self, num_experts):  
        self.hist = torch.zeros(num_experts)  
        self.sparsity = []  
        
    def update(self, gate_output):  
        active_experts = torch.unique(gate_output.indices)  
        self.hist[active_experts] += 1  
        sparsity = 1 - len(active_experts)/self.hist.size(0)  
        self.sparsity.append(sparsity)  
        
    def visualize(self):  
        plt.figure(figsize=(12,4))  
        plt.subplot(121)  
        plt.bar(range(len(self.hist)), self.hist.cpu())  
        plt.subplot(122)  
        plt.plot(self.sparsity)  

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

# 定义专家网络
class ExpertNetwork(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim):
        super(ExpertNetwork, self).__init__()
        self.layer1 = nn.Linear(input_dim, hidden_dim)
        self.relu = nn.ReLU()
        self.layer2 = nn.Linear(hidden_dim, output_dim)
    
    def forward(self, x):
        x = self.layer1(x)
        x = self.relu(x)
        x = self.layer2(x)
        return x

# 定义gating网络
class GatingNetwork(nn.Module):
    def __init__(self, input_dim, num_experts):
        super(GatingNetwork, self).__init__()
        self.layer = nn.Linear(input_dim, num_experts)
        self.softmax = nn.Softmax(dim=1)
    
    def forward(self, x):
        x = self.layer(x)
        x = self.softmax(x)
        return x

# 定义DeepSeek混合专家系统
class DeepSeekMixtureOfExperts(nn.Module):
    def __init__(self, input_dim, hidden_dim, output_dim, num_experts):
        super(DeepSeekMixtureOfExperts, self).__init__()
        self.experts = nn.ModuleList([ExpertNetwork(input_dim, hidden_dim, output_dim) for _ in range(num_experts)])
        self.gating = GatingNetwork(input_dim, num_experts)
        self.output_layer = nn.Linear(output_dim, 1)  # 情感分析为二分类任务，输出1个概率值
    
    def forward(self, x):
        # 计算gating权重
        gating_weights = self.gating(x)
        
        # 将输入数据分配给各个专家网络
        expert_outputs = []
        for i, expert in enumerate(self.experts):
            expert_output = expert(x)
            expert_outputs.append(expert_output)
        
        # 组合专家网络的输出
        combined_output = torch.zeros_like(expert_outputs[0])
        for i in range(len(self.experts)):
            combined_output += gating_weights[:, i].unsqueeze(1) * expert_outputs[i]
        
        # 通过任务特定输出层得到最终预测结果
        final_output = self.output_layer(combined_output)
        final_output = torch.sigmoid(final_output)  # 使用sigmoid函数将输出映射到[0,1]区间，表示情感概率
        
        return final_output

# 设置超参数
input_dim = 100  # 词向量维度
hidden_dim = 64
output_dim = 32
num_experts = 2
learning_rate = 0.001
num_epochs = 10
batch_size = 32

# 初始化模型、损失函数和优化器
model = DeepSeekMixtureOfExperts(input_dim, hidden_dim, output_dim, num_experts)
criterion = nn.BCELoss()  # 二分类交叉熵损失
optimizer = optim.Adam(model.parameters(), lr=learning_rate)

# 假设已经准备好了训练数据和标签，分别为train_data和train_labels
# 将数据转换为 DataLoader
train_dataset = torch.utils.data.TensorDataset(train_data, train_labels)
train_loader = torch.utils.data.DataLoader(train_dataset, batch_size=batch_size, shuffle=True)

# 训练模型
for epoch in range(num_epochs):
    model.train()
    for batch_data, batch_labels in train_loader:
        optimizer.zero_grad()
        outputs = model(batch_data)
        loss = criterion(outputs, batch_labels)
        loss.backward()
        optimizer.step()
    
    print(f'Epoch [{epoch+1}/{num_epochs}], Loss: {loss.item():.4f}')

# 测试模型
model.eval()
with torch.no_grad():
    test_outputs = model(test_data)
    test_predictions = (test_outputs >= 0.5).float()  # 阈值设为0.5，大于等于0.5预测为积极情感，否则为消极情感
    accuracy = (test_predictions == test_labels).sum().item() / test_labels.size(0)
    print(f'Test Accuracy: {accuracy:.4f}')

# 保存模型
torch.save(model.state_dict(), 'deepseek_sentiment_analysis.pth')

# 加载模型
model = DeepSeekMixtureOfExperts(input_dim, hidden_dim, output_dim, num_experts)
model.load_state_dict(torch.load('deepseek_sentiment_analysis.pth'))
model.eval()

# 应用示例
def analyze_sentiment(text):
    # 对文本进行预处理，转换为模型输入形式
    processed_text = preprocess_text(text)
    input_tensor = torch.tensor(processed_text, dtype=torch.float32)
    
    # 进行情感分析
    output = model(input_tensor)
    sentiment_prob = output.item()
    
    # 判断情感倾向
    if sentiment_prob >= 0.5:
        return "积极情感", sentiment_prob
    else:
        return "消极情感", sentiment_prob