54_模型优化：大模型的压缩与量化

大模型优化技术演进
├── 早期阶段(2018-2020): 基本剪枝、8位量化，性能损失明显
├── 发展阶段(2021-2023): 知识蒸馏、结构化剪枝，平衡效率与精度
└── 成熟阶段(2024-2025): 混合精度量化、参数高效微调，成本降至原1/30

目录
├── 第一章：大模型优化的必要性与挑战
├── 第二章：量化技术深度解析
├── 第三章：知识蒸馏方法学
├── 第四章：模型剪枝策略
├── 第五章：参数高效微调技术
├── 第六章：模型压缩工具与框架
├── 第七章：工程实践案例分析
├── 第八章：2025年最新进展与突破
└── 第九章：未来发展趋势与建议

# 量化过程的简化实现示例
def quantize_tensor(tensor, bits=8):
    # 获取tensor的最小值和最大值
    min_val = tensor.min().item()
    max_val = tensor.max().item()
    
    # 计算缩放因子和零点
    # 对于n位整数，可以表示的范围是[-2^(n-1), 2^(n-1)-1]
    q_min = -2 ** (bits - 1)
    q_max = 2 ** (bits - 1) - 1
    
    # 计算缩放因子
    scale = (max_val - min_val) / (q_max - q_min)
    
    # 计算零点（将浮点数零点映射到整数零点）
    zero_point = q_min - min_val / scale
    
    # 确保zero_point在整数范围内
    zero_point = max(q_min, min(q_max, zero_point))
    zero_point = int(round(zero_point))
    
    # 执行量化
    quantized = (tensor / scale + zero_point).round()
    quantized = quantized.clamp(q_min, q_max).to(torch.int8)
    
    return quantized, scale, zero_point

def dequantize_tensor(quantized, scale, zero_point):
    # 反量化过程
    return (quantized.to(torch.float32) - zero_point) * scale

# 混合精度量化的简化实现示例
import torch
import copy

def mixed_precision_quantization(model, calibration_dataset):
    # 复制原始模型
    quantized_model = copy.deepcopy(model)
    
    # 对不同层应用不同精度量化
    for name, module in quantized_model.named_modules():
        # 注意力层使用INT8量化
        if 'attention' in name or 'query' in name or 'key' in name or 'value' in name:
            # 对注意力层进行INT8量化
            quantized_module = quantize_module(module, bits=8)
            setattr(quantized_model, name.split('.')[-1], quantized_module)
        
        # 其他层可以使用INT4量化
        elif isinstance(module, torch.nn.Linear) and 'attention' not in name:
            # 对普通线性层进行INT4量化
            quantized_module = quantize_module(module, bits=4)
            setattr(quantized_model, name.split('.')[-1], quantized_module)
    
    # 使用校准数据集进行量化校准
    calibrate_model(quantized_model, calibration_dataset)
    
    return quantized_model

# 知识蒸馏的简化实现示例
import torch
import torch.nn as nn
import torch.nn.functional as F

class KnowledgeDistillation:
    def __init__(self, teacher_model, student_model, temperature=4.0, alpha=0.5):
        self.teacher_model = teacher_model
        self.student_model = student_model
        self.temperature = temperature  # 温度参数，控制软标签的平滑程度
        self.alpha = alpha  # 软标签损失的权重
        
        # 设置教师模型为评估模式
        self.teacher_model.eval()
        
    def distillation_loss(self, student_logits, teacher_logits, labels):
        # 软标签损失：学生模型的输出分布与教师模型的输出分布之间的KL散度
        soft_loss = nn.KLDivLoss(reduction='batchmean')(
            F.log_softmax(student_logits / self.temperature, dim=1),
            F.softmax(teacher_logits / self.temperature, dim=1)
        ) * (self.temperature ** 2)
        
        # 硬标签损失：学生模型的输出与真实标签之间的交叉熵
        hard_loss = F.cross_entropy(student_logits, labels)
        
        # 总损失 = 软标签损失 * alpha + 硬标签损失 * (1 - alpha)
        loss = self.alpha * soft_loss + (1 - self.alpha) * hard_loss
        
        return loss
    
    def train_step(self, input_ids, attention_mask, labels, optimizer):
        # 学生模型前向传播
        student_logits = self.student_model(input_ids, attention_mask)
        
        # 教师模型前向传播（不计算梯度）
        with torch.no_grad():
            teacher_logits = self.teacher_model(input_ids, attention_mask)
        
        # 计算蒸馏损失
        loss = self.distillation_loss(student_logits, teacher_logits, labels)
        
        # 反向传播和参数更新
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
        
        return loss.item()

# 大语言模型中间层蒸馏示例
import torch
import torch.nn as nn

class IntermediateLayerDistillation:
    def __init__(self, teacher_model, student_model, layer_mapping):
        self.teacher_model = teacher_model
        self.student_model = student_model
        self.layer_mapping = layer_mapping  # 学生层到教师层的映射关系
        self.feature_loss = nn.MSELoss()
        
        # 注册钩子收集中间层输出
        self.teacher_features = {}
        self.student_features = {}
        
        # 为教师模型中间层注册钩子
        def get_teacher_hook(name):
            def hook(module, input, output):
                self.teacher_features[name] = output[0]  # 通常取hidden_states
            return hook
        
        # 为学生模型中间层注册钩子
        def get_student_hook(name):
            def hook(module, input, output):
                self.student_features[name] = output[0]  # 通常取hidden_states
            return hook
        
        # 注册钩子
        for student_layer, teacher_layer in layer_mapping.items():
            getattr(self.teacher_model.model, f'layer_{teacher_layer}').register_forward_hook(
                get_teacher_hook(f'teacher_{teacher_layer}')
            )
            getattr(self.student_model.model, f'layer_{student_layer}').register_forward_hook(
                get_student_hook(f'student_{student_layer}')
            )
    
    def compute_distillation_loss(self, input_ids, attention_mask, labels, alpha=0.5):
        # 前向传播获取输出和中间特征
        student_output = self.student_model(input_ids, attention_mask, labels=labels)
        student_loss = student_output.loss
        
        with torch.no_grad():
            self.teacher_model(input_ids, attention_mask)
        
        # 计算中间层特征损失
        feature_loss = 0.0
        for student_layer, teacher_layer in self.layer_mapping.items():
            s_feat = self.student_features[f'student_{student_layer}']
            t_feat = self.teacher_features[f'teacher_{teacher_layer}']
            feature_loss += self.feature_loss(s_feat, t_feat)
        
        # 平均特征损失
        feature_loss /= len(self.layer_mapping)
        
        # 总损失
        total_loss = (1 - alpha) * student_loss + alpha * feature_loss
        
        return total_loss, student_loss, feature_loss

# 结构化剪枝的简化实现示例
import torch
import torch.nn as nn

def structured_pruning(model, pruning_ratio=0.3):
    # 对模型中的每个线性层进行剪枝
    for name, module in model.named_modules():
        if isinstance(module, nn.Linear):
            # 计算权重的重要性（这里使用L1范数）
            weight_importance = torch.norm(module.weight.data, dim=0, p=1)
            
            # 确定要保留的通道数量
            num_channels = module.out_features
            num_channels_to_keep = int(num_channels * (1 - pruning_ratio))
            
            # 选择重要性最高的通道
            _, top_indices = torch.topk(weight_importance, num_channels_to_keep)
            
            # 创建掩码
            mask = torch.zeros(num_channels, dtype=torch.bool)
            mask[top_indices] = True
            
            # 应用掩码剪枝权重和偏置
            module.weight.data = module.weight.data[mask]
            if module.bias is not None:
                module.bias.data = module.bias.data[mask]
                
            # 记录剪枝信息
            print(f"Pruned layer {name}: {num_channels - num_channels_to_keep} channels removed")
    
    return model

def iterative_pruning(model, train_loader, val_loader, pruning_steps=3, pruning_ratio=0.1):
    """迭代式剪枝：逐步增加稀疏度"""
    for step in range(pruning_steps):
        # 剪枝
        model = structured_pruning(model, pruning_ratio)
        
        # 微调恢复性能
        print(f"Fine-tuning after pruning step {step+1}/{pruning_steps}")
        fine_tune(model, train_loader, val_loader, epochs=3)
        
        # 评估模型性能
        evaluate(model, val_loader)
    
    return model

# LoRA实现的简化示例
import torch
import torch.nn as nn

class LoRALayer(nn.Module):
    def __init__(self, in_dim, out_dim, rank=8, alpha=1):
        super().__init__()
        self.rank = rank
        self.alpha = alpha  # 缩放因子
        
        # 初始化低秩矩阵
        self.A = nn.Parameter(torch.randn(in_dim, rank))
        self.B = nn.Parameter(torch.zeros(rank, out_dim))
        
        # 缩放因子，用于调整更新幅度
        self.scaling = self.alpha / self.rank
        
        # 初始化A矩阵
        nn.init.kaiming_uniform_(self.A, a=math.sqrt(5))
    
    def forward(self, x):
        # x形状: [batch_size, seq_len, in_dim]
        # 计算低秩分解的输出: x * A * B * scaling
        result = x @ self.A @ self.B * self.scaling
        return result

# 将LoRA层集成到Transformer模型中
class LinearWithLoRA(nn.Module):
    def __init__(self, linear_layer, rank=8, alpha=1):
        super().__init__()
        # 原始线性层（冻结）
        self.linear = linear_layer
        for param in self.linear.parameters():
            param.requires_grad = False
        
        # 添加LoRA层
        self.lora = LoRALayer(linear_layer.in_features, linear_layer.out_features, rank, alpha)
        self.use_lora = True
    
    def forward(self, x):
        # 原始线性层输出
        out = self.linear(x)
        
        # 如果启用LoRA，添加LoRA输出
        if self.use_lora:
            out = out + self.lora(x)
        
        return out

# 组合式参数高效微调示例
from peft import get_peft_model, LoraConfig, TaskType
import transformers

def setup_combined_peft(model, task_type=TaskType.CAUSAL_LM):
    # 配置LoRA
    lora_config = LoraConfig(
        task_type=task_type,
        inference_mode=False,
        r=16,  # 秩参数
        lora_alpha=32,
        lora_dropout=0.1,
        # 只对注意力层应用LoRA
        target_modules=["q_proj", "v_proj", "k_proj", "o_proj"],
        # 冻结其他参数
        bias="none",
    )
    
    # 获取PEFT模型
    peft_model = get_peft_model(model, lora_config)
    
    # 另外，只微调顶层分类器（如果存在）
    if hasattr(model, "lm_head"):
        for param in model.lm_head.parameters():
            param.requires_grad = True
    
    return peft_model

def count_trainable_params(model):
    """计算可训练参数数量"""
    trainable_params = 0
    all_param = 0
    for _, param in model.named_parameters():
        num_params = param.numel()
        if param.requires_grad:
            trainable_params += num_params
        all_param += num_params
    
    print(f"可训练参数: {trainable_params / 1000000:.2f}M")
    print(f"总参数: {all_param / 1000000:.2f}M")
    print(f"参数效率: {100 * trainable_params / all_param:.2f}%")
    
    return trainable_params, all_param

# PyTorch静态量化示例
import torch
import torchvision.models as models

# 1. 准备模型
model = models.resnet18(pretrained=True)
model.eval()

# 2. 融合量化友好的操作（如Conv+BN+ReLU）
model_fused = torch.quantization.fuse_modules(model, [['conv1', 'bn1', 'relu']])
for module_name, module in model_fused.named_children():
    if isinstance(module, torch.nn.Sequential):
        torch.quantization.fuse_modules(module, [['conv1', 'bn1', 'relu1'], ['conv2', 'bn2']])

# 3. 指定量化配置
model_prepared = torch.quantization.prepare(model_fused)

# 4. 使用校准数据进行校准
calibration_data = get_calibration_data()  # 加载校准数据
for data in calibration_data:
    model_prepared(data)

# 5. 转换为量化模型
model_quantized = torch.quantization.convert(model_prepared)

# 6. 保存量化模型
torch.jit.save(torch.jit.script(model_quantized), "quantized_model.pt")

# PyTorch训练时量化示例
import torch
import torch.nn as nn

# 1. 定义量化友好的模型
class QuantizableModel(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv = nn.Conv2d(3, 64, kernel_size=3, padding=1)
        self.bn = nn.BatchNorm2d(64)
        self.relu = nn.ReLU()
        self.fc = nn.Linear(64 * 224 * 224, 1000)
    
    def forward(self, x):
        x = self.conv(x)
        x = self.bn(x)
        x = self.relu(x)
        x = x.view(x.size(0), -1)
        x = self.fc(x)
        return x

# 2. 准备训练
model = QuantizableModel()
model.qconfig = torch.quantization.get_default_qconfig('fbgemm')
model_prepared = torch.quantization.prepare_qat(model)

# 3. 训练模型（正常训练流程）
optimizer = torch.optim.SGD(model_prepared.parameters(), lr=0.01)
criterion = nn.CrossEntropyLoss()

for epoch in range(10):
    for inputs, labels in train_loader:
        optimizer.zero_grad()
        outputs = model_prepared(inputs)
        loss = criterion(outputs, labels)
        loss.backward()
        optimizer.step()

# 4. 转换为量化模型
model_prepared.eval()
model_quantized = torch.quantization.convert(model_prepared)

# Hugging Face PEFT LoRA示例
from transformers import AutoModelForCausalLM, AutoTokenizer
from peft import get_peft_model, LoraConfig, TaskType

# 加载基础模型
tokenizer = AutoTokenizer.from_pretrained("facebook/opt-125m")
model = AutoModelForCausalLM.from_pretrained("facebook/opt-125m")

# 配置LoRA
lora_config = LoraConfig(
    task_type=TaskType.CAUSAL_LM,
    inference_mode=False,
    r=8,
    lora_alpha=32,
    lora_dropout=0.1,
    target_modules=["q_proj", "v_proj"],
    bias="none",
)

# 创建PEFT模型
model = get_peft_model(model, lora_config)

# 打印可训练参数
model.print_trainable_parameters()

# 正常训练流程...
# 训练完成后保存适配器
model.save_pretrained("lora-adapter")

# 加载适配器进行推理
from peft import PeftModel

base_model = AutoModelForCausalLM.from_pretrained("facebook/opt-125m")
peft_model = PeftModel.from_pretrained(base_model, "lora-adapter")

# 合并适配器到基础模型（可选，用于部署）
merged_model = peft_model.merge_and_unload()

大模型优化成功要素
明确目标 → 技术选型 → 工程实施 → 持续优化 → 价值实现