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 137 138 139 140 141 142 143
| import torch import torch.nn as nn import torch.quantization as quant
class DMSModel(nn.Module): """ DMS模型示例(简化版) """ def __init__(self): 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.Linear(64, 2) ) def forward(self, x): x = self.features(x) x = x.view(x.size(0), -1) x = self.classifier(x) return x
class PTQPipeline: """ 训练后量化流水线 """ def __init__(self, model): self.model = model def prepare_for_quantization(self): """ 准备量化 """ self.model.eval() for i in range(len(self.model.features)): if isinstance(self.model.features[i], nn.Conv2d): if i+2 < len(self.model.features): if isinstance(self.model.features[i+1], nn.BatchNorm2d) and \ isinstance(self.model.features[i+2], nn.ReLU): torch.quantization.fuse_modules( self.model.features, [str(i), str(i+1), str(i+2)], inplace=True ) self.model.qconfig = quant.get_default_qconfig('fbgemm') quant.prepare(self.model, inplace=True) def calibrate(self, calibration_data): """ 校准(统计激活范围) """ with torch.no_grad(): for batch in calibration_data: self.model(batch) def convert_to_int8(self): """ 转换为INT8模型 """ quant.convert(self.model, inplace=True) def export_onnx(self, output_path, input_shape=(1, 3, 224, 224)): """ 导出ONNX """ dummy_input = torch.randn(input_shape) torch.onnx.export( self.model, dummy_input, output_path, opset_version=13, input_names=['input'], output_names=['output'], dynamic_axes={ 'input': {0: 'batch_size'}, 'output': {0: 'batch_size'} } )
def ptq_workflow(): """ PTQ完整流程 """ model = DMSModel() model.load_state_dict(torch.load('dms_model.pth')) ptq = PTQPipeline(model) ptq.prepare_for_quantization() calibration_data = [torch.randn(1, 3, 224, 224) for _ in range(100)] ptq.calibrate(calibration_data) ptq.convert_to_int8() ptq.export_onnx('dms_model_int8.onnx') print("PTQ完成,模型已导出到 dms_model_int8.onnx")
if __name__ == "__main__": ptq_workflow()
|