模型量化算法详解：从 PTQ 到 AWQ 的大模型压缩实战指南

import torch
import numpy as np

classLinearQuantizer:
    """线性量化器（支持对称/非对称）"""
    
    def__init__(self, bits=8, symmetric=True):
        self.bits = bits
        self.symmetric = symmetric
        
        # 计算量化范围
        if symmetric:
            self.qmin = -(2 ** (bits - 1))
            self.qmax = 2 ** (bits - 1) - 1
        else:
            self.qmin = 0
            self.qmax = 2 ** bits - 1
    
    defcompute_scale_zero_point(self, tensor):
        """计算 scale 和 zero_point"""
        ifself.symmetric:
            # 对称量化
            abs_max = torch.max(torch.abs(tensor))
            scale = abs_max / self.qmax
            zero_point = 0
        else:
            # 非对称量化
            min_val = torch.min(tensor)
            max_val = torch.max(tensor)
            scale = (max_val - min_val) / (self.qmax - self.qmin)
            zero_point = self.qmin - torch.round(min_val / scale)
        
        return scale, zero_point
    
    defquantize(self, tensor):
        """量化张量"""
        scale, zero_point = self.compute_scale_zero_point(tensor)
        
        # 量化
        quantized = torch.round(tensor / scale + zero_point)
        
        # 截断到量化范围
        quantized = torch.clamp(quantized, self.qmin, self.qmax)
        
        return quantized.to(torch.int8 ifself.bits == 8else torch.int32), scale, zero_point
    
    defdequantize(self, quantized, scale, zero_point):
        """反量化"""
        return scale * (quantized.float() - zero_point)
    
    defquantize_dequantize(self, tensor):
        """量化后立即反量化（模拟量化效果）"""
        quantized, scale, zero_point = self.quantize(tensor)
        returnself.dequantize(quantized, scale, zero_point)

# 使用示例
print("=" * 60)
print("基础线性量化演示")
print("=" * 60)

# 创建测试权重
weight = torch.randn(10, 10) * 2# 标准差为2的正态分布

# 对称 INT8 量化
quantizer_sym = LinearQuantizer(bits=8, symmetric=True)
quantized_sym, scale_sym, zp_sym = quantizer_sym.quantize(weight)
weight_dequant_sym = quantizer_sym.dequantize(quantized_sym, scale_sym, zp_sym)

# 非对称 INT8 量化
quantizer_asym = LinearQuantizer(bits=8, symmetric=False)
quantized_asym, scale_asym, zp_asym = quantizer_asym.quantize(weight)
weight_dequant_asym = quantizer_asym.dequantize(quantized_asym, zp_asym, scale_asym)

# 计算误差
error_sym = torch.mean(torch.abs(weight - weight_dequant_sym)).item()
error_asym = torch.mean(torch.abs(weight - weight_dequant_asym)).item()

print(f"原始权重范围: [{weight.min():.4f}, {weight.max():.4f}]")
print(f"对称量化 - 平均绝对误差: {error_sym:.6f}")
print(f"非对称量化 - 平均绝对误差: {error_asym:.6f}")
print(f"存储节省: {(32 - 8) / 32 * 100:.1f}%")

预训练模型 (FP32/FP16)
    ↓
收集校准数据（少量样本）
    ↓
计算每层 scale 和 zero_point
    ↓
量化权重和激活
    ↓
部署量化模型 (INT8/INT4)

import torch
import torch.quantization

# PyTorch 动态量化示例
model = torch.nn.Sequential(
    torch.nn.Linear(784, 256),
    torch.nn.ReLU(),
    torch.nn.Linear(256, 10)
)

# 应用动态量化
model_quantized = torch.quantization.quantize_dynamic(
    model, 
    {torch.nn.Linear},  # 量化 Linear 层
    dtype=torch.qint8
)

# 查看模型大小变化
print(f"原始模型参数数量: {sum(p.numel() for p in model.parameters())}")
print(f"量化后模型类型: {type(model_quantized)}")

class StaticQuantizationExample:
    """静态量化完整示例"""
    
    def__init__(self, model):
        self.model = model
        
    defprepare_for_quantization(self):
        """准备量化"""
        # 插入观察者（Observer）
        self.model.qconfig = torch.quantization.get_default_qconfig('fbgemm')
        torch.quantization.prepare(self.model, inplace=True)
        returnself.model
    
    defcalibrate(self, calibration_data):
        """使用校准数据收集统计信息"""
        self.model.eval()
        with torch.no_grad():
            for batch in calibration_data:
                self.model(batch)
    
    defconvert_to_quantized(self):
        """转换为量化模型"""
        torch.quantization.convert(self.model, inplace=True)
        returnself.model

# 使用流程
# model = MyModel()
# quant_example = StaticQuantizationExample(model)
# quant_example.prepare_for_quantization()
# quant_example.calibrate(calibration_loader)
# quantized_model = quant_example.convert_to_quantized()

标准训练: 前向(FP32) → 反向 → 更新
QAT训练:  前向(FP32→模拟量化→FP32) → 反向 → 更新

class FakeQuantize(torch.nn.Module):
    """伪量化模块（支持 STE）"""
    
    def __init__(self, bits=8, symmetric=True):
        super().__init__()
        self.bits = bits
        self.symmetric = symmetric
        
        if symmetric:
            self.qmin = -(2 ** (bits - 1))
            self.qmax = 2 ** (bits - 1) - 1
        else:
            self.qmin = 0
            self.qmax = 2 ** bits - 1
        
        # 可学习的 scale 和 zero_point
        self.scale = torch.nn.Parameter(torch.ones(1))
        self.zero_point = torch.nn.Parameter(torch.zeros(1))
    
    def forward(self, x):
        ifself.training:
            # 训练时：伪量化 + STE
            return FakeQuantizeSTE.apply(x, self.scale, self.zero_point, 
                                          self.qmin, self.qmax)
        else:
            # 推理时：真实量化
            returnself._quantize(x)
    
    def _quantize(self, x):
        x_quant = torch.round(x / self.scale + self.zero_point)
        x_quant = torch.clamp(x_quant, self.qmin, self.qmax)
        returnself.scale * (x_quant - self.zero_point)

class FakeQuantizeSTE(torch.autograd.Function):
    """支持 STE 的伪量化"""
    
    @staticmethod
    def forward(ctx, x, scale, zero_point, qmin, qmax):
        x_quant = torch.round(x / scale + zero_point)
        x_quant = torch.clamp(x_quant, qmin, qmax)
        return scale * (x_quant - zero_point)
    
    @staticmethod
    def backward(ctx, grad_output):
        # STE: 直接传递梯度
        return grad_output, None, None, None, None

class QATModel(torch.nn.Module):
    """支持 QAT 的模型"""
    
    def __init__(self, input_dim, hidden_dim, output_dim):
        super().__init__()
        self.fc1 = torch.nn.Linear(input_dim, hidden_dim)
        self.fake_quant1 = FakeQuantize(bits=8)
        self.relu = torch.nn.ReLU()
        self.fc2 = torch.nn.Linear(hidden_dim, output_dim)
        self.fake_quant2 = FakeQuantize(bits=8)
    
    def forward(self, x):
        x = self.fc1(x)
        x = self.fake_quant1(x)  # 伪量化
        x = self.relu(x)
        x = self.fc2(x)
        x = self.fake_quant2(x)  # 伪量化
        return x

def train_qat(model, train_loader, num_epochs=10):
    """QAT 训练流程"""
    optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
    criterion = torch.nn.CrossEntropyLoss()
    
    # 阶段1：正常训练（可选）
    print("阶段1：正常训练...")
    
    # 阶段2：插入伪量化
    print("阶段2：启用量化感知训练...")
    
    # 阶段3：QAT 微调
    for epoch inrange(num_epochs):
        model.train()
        total_loss = 0
        
        for batch_idx, (data, target) inenumerate(train_loader):
            optimizer.zero_grad()
            output = model(data)
            loss = criterion(output, target)
            loss.backward()
            optimizer.step()
            
            total_loss += loss.item()
        
        print(f"Epoch {epoch+1}/{num_epochs}, Loss: {total_loss/len(train_loader):.4f}")
    
    # 阶段4：转换为真实量化模型
    print("阶段4：转换为量化模型...")
    quantized_model = convert_to_real_quantization(model)

    return quantized_model

def convert_to_real_quantization(model):
    """将 QAT 模型转换为真实量化模型"""
    # 冻结统计信息
    model.eval()
    
    # 替换伪量化为真实量化
    # 实际实现需要遍历模型并替换模块
    
    return model

输入: 预训练权重 W，校准数据 X
输出: 量化后的权重 W_q

1. 计算 Hessian 矩阵: H = X^T @ X
2. 对 H 进行 Cholesky 分解
3. 对每一层:
   a. 按重要性排序权重
   b. 逐个量化权重:
      - 计算量化误差
      - 更新未量化权重进行补偿
      - 更新 Hessian 逆矩阵
4. 返回量化后的权重

import torch
import torch.nn as nn
import numpy as np

class GPTQQuantizer:
    """GPTQ 量化器"""
    
    def__init__(self, bits=4, group_size=128):
        self.bits = bits
        self.group_size = group_size
        self.qmin = -(2 ** (bits - 1))
        self.qmax = 2 ** (bits - 1) - 1
    
    def quantize_weight_group(self, W, H_inv, quantize_indices):
        """
        量化一个权重组，使用 OBS 补偿
        
        W: [out_features, in_features] 权重矩阵
        H_inv: Hessian 逆矩阵
        quantize_indices: 要量化的权重索引
        """
        W_quant = W.clone()
        
        for idx in quantize_indices:
            # 计算当前权重的量化值
            w = W_quant[idx]
            scale = (w.max() - w.min()) / (self.qmax - self.qmin)
            w_quant = torch.round(w / scale).clamp(self.qmin, self.qmax) * scale
            
            # 计算量化误差
            error = w - w_quant
            
            # OBS 补偿：更新其他权重
            # delta_w = -error / H_inv[idx, idx] * H_inv[:, idx]
            compensation = -error / H_inv[idx, idx] * H_inv[:, idx]
            W_quant += compensation.unsqueeze(0)
            
            # 设置量化后的值
            W_quant[idx] = w_quant
        
        return W_quant
    
    def quantize_layer(self, layer, calibration_data):
        """量化一个线性层"""
        weight = layer.weight.data
        
        # 计算 Hessian 矩阵（简化版）
        with torch.no_grad():
            # 收集激活
            activations = []
            for batch in calibration_data:
                activations.append(batch)
            X = torch.cat(activations, dim=0)  # [num_samples, in_features]
            
            # 计算 Hessian: H = X^T @ X
            H = X.T @ X
            
            # 添加阻尼项确保可逆
            H += torch.eye(H.shape[0]) * 0.01
            
            # 计算 Hessian 逆
            H_inv = torch.linalg.inv(H)
        
        # 分组量化
        out_features, in_features = weight.shape
        num_groups = in_features // self.group_size
        
        W_quantized = torch.zeros_like(weight)
        
        for g inrange(num_groups):
            start = g * self.group_size
            end = start + self.group_size
            
            W_group = weight[:, start:end]
            H_group = H[start:end, start:end]
            H_inv_group = torch.linalg.inv(H_group + torch.eye(self.group_size) * 0.01)
            
            # 确定量化顺序（按 Hessian 对角线元素排序）
            quantize_order = torch.argsort(torch.diag(H_inv_group))
            
            # 量化该组
            W_quant_group = self.quantize_weight_group(
                W_group, H_inv_group, quantize_order
            )
            
            W_quantized[:, start:end] = W_quant_group
        
        return W_quantized

# 使用示例
print("=" * 60)
print("GPTQ 量化演示")
print("=" * 60)

# 创建测试层
linear_layer = nn.Linear(512, 256)

# 模拟校准数据
calibration_data = [torch.randn(32, 512) for _ inrange(10)]

# GPTQ 量化
gptq = GPTQQuantizer(bits=4, group_size=128)
quantized_weight = gptq.quantize_layer(linear_layer, calibration_data)

# 计算误差
original_weight = linear_layer.weight.data
error = torch.mean(torch.abs(original_weight - quantized_weight)).item()

print(f"原始权重形状: {original_weight.shape}")
print(f"量化后平均绝对误差: {error:.6f}")
print(f"存储节省: {(16 - 4) / 16 * 100:.1f}%")

输入: 预训练模型，校准数据
输出: 量化后的模型

1. 收集激活统计信息
   - 计算每个通道的最大激活值
   
2. 计算缩放因子
   - s_i = (max_activation_i / mean_max_activation)^alpha
   
3. 应用缩放并量化
   - W_scaled = W * s
   - W_quant = Quantize(W_scaled)
   - W_final = W_quant / s
   
4. 返回量化模型

class AWQQuantizer:
    """AWQ 量化器（激活感知）"""
    
    def__init__(self, bits=4, group_size=128, alpha=0.5):
        self.bits = bits
        self.group_size = group_size
        self.alpha = alpha
        self.qmin = -(2 ** (bits - 1))
        self.qmax = 2 ** (bits - 1) - 1
    
    defcompute_scales(self, activations):
        """
        计算逐通道缩放因子
        
        activations: [num_samples, num_features]
        """
        # 计算每个通道的最大激活值
        max_activations = torch.max(torch.abs(activations), dim=0)[0]
        
        # 计算缩放因子
        mean_max = torch.mean(max_activations)
        scales = (max_activations / mean_max) ** self.alpha
        
        return scales
    
    defquantize_with_scaling(self, weight, scales):
        """
        应用缩放后进行量化
        
        weight: [out_features, in_features]
        scales: [in_features] 逐输入通道缩放
        """
        # 应用缩放
        weight_scaled = weight * scales.unsqueeze(0)
        
        # 计算量化参数
        w_max = torch.max(torch.abs(weight_scaled))
        scale = w_max / self.qmax
        
        # 量化
        weight_quant = torch.round(weight_scaled / scale)
        weight_quant = torch.clamp(weight_quant, self.qmin, self.qmax)
        
        # 反量化并去除缩放
        weight_dequant = weight_quant * scale
        weight_final = weight_dequant / scales.unsqueeze(0)
        
        return weight_final
    
    defquantize_layer(self, layer, calibration_data):
        """量化一个线性层"""
        weight = layer.weight.data
        
        # 收集激活
        with torch.no_grad():
            activations = []
            for batch in calibration_data:
                # 模拟前向传播获取激活
                activations.append(batch)
            X = torch.cat(activations, dim=0)
        
        # 计算缩放因子
        scales = self.compute_scales(X)
        
        # 分组量化（每组有自己的缩放）
        out_features, in_features = weight.shape
        num_groups = in_features // self.group_size
        
        W_quantized = torch.zeros_like(weight)
        
        for g inrange(num_groups):
            start = g * self.group_size
            end = start + self.group_size
            
            W_group = weight[:, start:end]
            scales_group = scales[start:end]
            
            W_quant_group = self.quantize_with_scaling(W_group, scales_group)
            W_quantized[:, start:end] = W_quant_group
        
        return W_quantized

# 使用示例
print("=" * 60)
print("AWQ 量化演示")
print("=" * 60)

# 创建测试层
linear_layer = nn.Linear(512, 256)

# 模拟校准数据（含异常值）
calibration_data = [torch.randn(32, 512) for _ inrange(10)]
# 添加异常值
calibration_data[0][0, :10] *= 10

# AWQ 量化
awq = AWQQuantizer(bits=4, group_size=128, alpha=0.5)
quantized_weight_awq = awq.quantize_layer(linear_layer, calibration_data)

# 对比普通 INT4 量化
quantizer_int4 = LinearQuantizer(bits=4, symmetric=True)
quantized_int4, _, _ = quantizer_int4.quantize(linear_layer.weight.data)
quantized_weight_int4 = quantizer_int4.dequantize(quantized_int4, 
                                                   quantizer_int4.compute_scale_zero_point(linear_layer.weight.data)[0],
                                                   0)

# 计算误差
error_awq = torch.mean(torch.abs(linear_layer.weight.data - quantized_weight_awq)).item()
error_int4 = torch.mean(torch.abs(linear_layer.weight.data - quantized_weight_int4)).item()

print(f"普通 INT4 量化误差: {error_int4:.6f}")
print(f"AWQ 量化误差: {error_awq:.6f}")
print(f"误差改善: {(error_int4 - error_awq) / error_int4 * 100:.1f}%")

注意力权重 → 更高精度（如 6-bit）
FFN 权重   → 较低精度（如 4-bit）
嵌入层     → 最高精度（如 8-bit）

# 1. 克隆 llama.cpp
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp

# 2. 编译
cmake --build build --config Release
# 3. 转换 Hugging Face 模型为 GGUF
python convert_hf_to_gguf.py \
    --outfile models/llama-7b-f16.gguf \
    ../Llama-2-7b-hf

# 4. 量化
./llama-quantize \
    models/llama-7b-f16.gguf \
    models/llama-7b-q4_k_m.gguf \
    q4_k_m

# 5. 推理
./llama-cli \
    -m models/llama-7b-q4_k_m.gguf \
    -p "Once upon a time" \
    -n 128

from llama_cpp import Llama

# 加载量化模型
llm = Llama(
    model_path="models/llama-7b-q4_k_m.gguf",
    n_ctx=2048,        # 上下文长度
    n_threads=8,       # CPU 线程数
    n_gpu_layers=0     # GPU 层数（0 表示纯 CPU）
)

# 生成文本
output = llm(
    "Q: 什么是模型量化？

A: ",
    max_tokens=256,
    temperature=0.7,
    stop=["Q:", "

"]
)

print(output['choices'][0]['text'])

需要部署大模型（>7B）？
├── 是 → 需要 4-bit 量化？
│   ├── 是 → 追求极致速度？
│   │   ├── 是 → AWQ（支持 fused kernel）
│   │   └── 否 → GPTQ（精度更高）
│   └── 否 → 使用 8-bit
│       └── 选择：LLM.int8() 或 bitsandbytes
│
└── 否 → 中小型模型（<7B）
    ├── 追求零训练成本？
    │   ├── 是 → PTQ（TensorRT, ONNX Runtime）
    │   └── 否 → QAT（精度最优）
    │
    └── 边缘设备部署？
        ├── 是 → TensorFlow Lite, Core ML
        └── 否 → PyTorch Quantization

PTQ (2018)
    ↓ 精度不足
QAT (2019)
    ↓ 训练成本高
GPTQ (2022)
    ↓ 大模型专用
AWQ (2023)
    ↓ 激活感知
GGUF/K-quant (2023)
    ↓ 混合精度
??? (2024+)