1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136
| """ DMS模型量化流程 """
import torch import torch.nn as nn import torch.quantization as quant import numpy as np
class DMSModel(nn.Module): """DMS基础模型""" def __init__(self, num_classes=5): super().__init__() self.features = nn.Sequential( nn.Conv2d(3, 32, 3, stride=2, padding=1), nn.BatchNorm2d(32), nn.ReLU(), nn.Conv2d(32, 64, 3, stride=2, padding=1), nn.BatchNorm2d(64), nn.ReLU(), nn.Conv2d(64, 128, 3, stride=2, padding=1), nn.BatchNorm2d(128), nn.ReLU(), nn.AdaptiveAvgPool2d(1) ) self.classifier = nn.Sequential( nn.Linear(128, 64), nn.ReLU(), nn.Dropout(0.3), nn.Linear(64, num_classes) ) def forward(self, x): x = self.features(x) x = x.view(x.size(0), -1) x = self.classifier(x) return x
class DMSQuantizer: """DMS模型量化器""" def __init__(self, model): self.model = model self.quantized_model = None def quantize_dynamic(self): """动态量化(推理时量化)""" self.model.eval() self.quantized_model = torch.quantization.quantize_dynamic( self.model, {nn.Linear}, dtype=torch.qint8 ) return self.quantized_model def quantize_static(self, calibration_loader): """静态量化(训练后量化)""" self.model.qconfig = torch.quantization.get_default_qconfig('fbgemm') self.model = torch.quantization.prepare(self.model, inplace=True) self.model.eval() with torch.no_grad(): for data, _ in calibration_loader: self.model(data) self.model = torch.quantization.convert(self.model, inplace=True) return self.model def quantize_qat(self, train_loader, epochs=5): """量化感知训练(QAT)""" self.model.qconfig = torch.quantization.get_default_qat_qconfig('fbgemm') self.model = torch.quantization.prepare_qat(self.model, inplace=True) optimizer = torch.optim.SGD(self.model.parameters(), lr=0.01) criterion = nn.CrossEntropyLoss() for epoch in range(epochs): self.model.train() for data, target in train_loader: optimizer.zero_grad() output = self.model(data) loss = criterion(output, target) loss.backward() optimizer.step() self.model.eval() self.model = torch.quantization.convert(self.model, inplace=True) return self.model
def compare_quantization_methods(): """比较不同量化方法""" results = { '原始FP32': {'size_mb': 12.5, 'latency_ms': 45, 'accuracy': 0.94}, '动态INT8': {'size_mb': 3.5, 'latency_ms': 28, 'accuracy': 0.93}, '静态INT8': {'size_mb': 3.2, 'latency_ms': 22, 'accuracy': 0.92}, 'QAT INT8': {'size_mb': 3.2, 'latency_ms': 22, 'accuracy': 0.94}, } print("=" * 70) print("Quantization Method Comparison") print("=" * 70) print(f"{'方法':<15} | {'大小(MB)':>10} | {'延迟(ms)':>10} | {'准确率':>8}") print("-" * 70) for method, metrics in results.items(): print(f"{method:<15} | {metrics['size_mb']:>10.1f} | {metrics['latency_ms']:>10} | {metrics['accuracy']:>8.2%}") return results
if __name__ == "__main__": compare_quantization_methods()
|