7.3.4. Heterogeneous Model QAT¶
7.3.4.1. Differences Between Heterogeneous Models and Non-heterogeneous Models¶
Heterogeneous models are those that run partly on the BPU and partly on the CPU during deployment, while non-heterogeneous models run completely on the BPU during deployment. Generally, the following two types of models will become heterogeneous when deployed:
Models with operators that are not supported by BPU.
Models of which some operators are specified to run on the CPU by the user due to excessive quantization quantization accuracy errors.
The difference between plugin’s support for heterogeneous models and non-heterogeneous models is as follows:
heterogeneous |
non-heterogeneous |
|
|---|---|---|
operator |
Align with horizon_nn, and the horizon_nn support operator shall prevail. The model can include CPU operator. |
It is directly aligned with the compiler, subject to the horizon_plugin_pytorch support operators, and the model cannot include CPU operators. |
interface |
prepare_qat_fx: Specify the hybrid mode and set dict as required. |
Refer to the usage document of non-heterogeneous mode |
technological process |
|
|
7.3.4.2. Description of Main Interface Parameters¶
The usage of heterogeneous interfaces is basically consistent with the usage of non-heterogeneous interfaces. Only a few parameters such as hybrid are added. For detailed interface parameter descriptions, see the API documentation . Here we focus on the key parameters.
7.3.4.2.1. horizon_plugin_pytorch.quantization.prepare_qat_fx¶
Enable the hybrid parameter. If the BPU operator is not specified to return to the CPU, you can not set the hybrid_dict .
def prepare_qat_fx(
model: Union[torch.nn.Module, GraphModule],
qconfig_dict: Dict[str, Any] = None,
prepare_custom_config_dict: Dict[str, Any] = None,
optimize_graph: bool = False,
hybrid: bool = False,
hybrid_dict: Dict[str, List] = None,
) -> ObservedGraphModule:
"""Prepare QAT model
`model`: torch.nn.Module or GraphModule(model after using fuse_fx)
`qconfig_dict`: Define QConfig. In addition to qconfig_dict, if the eager mode is used to define the qconfig in the module, the qconfig defined in the module takes precedence. The configuration format of qconfig_dict is as follows
qconfig_dict = {
# Optional, global configuration
"": qconfig,
# Optional, configured by module type
"module_type": [(torch.nn.Conv2d, qconfig), ...],
# Optional, configured by module name
"module_name": [("foo.bar", qconfig),...],
# Priority: global < module_type < module_name < module.qconfig
# The qconfig of an operator of non module type is consistent with the qconfig of its parent module by default. If you need to set it separately, please encapsulate it as a module.
}
`prepare_custom_config_dict`: Custom configuration dictionary
prepare_custom_config_dict = {
# Only preserved_attributes are supported temporarily. Generally speaking, all attributes will be automatically retained. This option is just a precaution and is rarely used.
"preserved_attributes": ["preserved_attr"],
}
`optimize_graph`: Keep the scale of cat input and output consistent. At present, it is only valid under the Bernoulli architecture.
`hybrid`:Whether to use the heterogeneous mode. Heterogeneous mode must be enabled in the following cases:
1. The model contains operators that are not supported by BPU or the user wants to specify some BPU operators to return to the CPU.
2. Users want the QAT model to be docked with the horizon_nn for fixed-point.
`hybrid_dict`: Define the CPU operator specified by users actively.
hybrid_dict = {
# Optional, configured by module type
"module_type": [torch.nn.Conv2d, ...],
# Optional, configured by module name
"module_name": ["foo.bar", ...],
# Priority: module_type < module_name
# Similar to qconfigdict, if you want non module operators to run on the CPU, you need to package this part as a module separately.
}
"""
7.3.4.2.2. horizon_plugin_pytorch.quantization.prepare_calibraiton_fx¶
The usage is exactly the same as prepare_qat_fx . Note that qconfig uses calibration qconfig.
def prepare_calibration_fx(
model,
qconfig_dict: Dict[str, Any] = None,
prepare_custom_config_dict: Dict[str, Any] = None,
optimize_graph: bool = False,
hybrid: bool = False,
hybrid_dict: Dict[str, List] = None,
)-> ObservedGraphModule:
7.3.4.2.3. horizon_plugin_pytorch.quantization.convert_fx¶
The convert interface in heterogeneous mode is used in the same way as that in non-heterogeneous mode, but the fixed-point model obtained by the heterogeneous model convert is only used to evaluate the accuracy, not to obtain the final deployed model.
def convert_fx(
graph_module: GraphModule,
convert_custom_config_dict: Dict[str, Any] = None,
_remove_qconfig: bool = True,
) -> QuantizedGraphModule:
"""Convert the QAT model, which is only used to evaluate the fixed-point model.
`graph_module`: Models after prepare ->(calibration) ->train
`convert_custom_config_dict`: Customize the configuration dictionary
convert_custom_config_dict = {
# Only preserved_attributes are supported temporarily.Generally speaking, all attributes will be automatically retained. This option is just a precaution and is rarely used.
"preserved_attributes": ["preserved_attr"],
}
`_remove_qconfig`: Whether to delete the qconfig after the convert is not generally used
"""
7.3.4.2.4. horizon_plugin_pytorch.utils.onnx_helper.export_to_onnx¶
In non-heterogeneous mode, this interface is only used for visualization; In heterogeneous mode, this interface can also be used to export the onnx for docking to hb_mapper.
def export_to_onnx(
model,
args,
f,
export_params=True,
verbose=False,
training=TrainingMode.EVAL,
input_names=None,
output_names=None,
operator_export_type=OperatorExportTypes.ONNX_FALLTHROUGH,
opset_version=11,
do_constant_folding=True,
example_outputs=None,
strip_doc_string=True,
dynamic_axes=None,
keep_initializers_as_inputs=None,
custom_opsets=None,
enable_onnx_checker=False,
):
"""This interface is basically the same as torch.onnx.export, which hides the parameters that do not need to be modified. The following parameters should be noted:
`model`: Models to be exported
`args`: Model input, used for trace model
`f`: Saved onnx file name or file descriptor
`operator_export_type`: Export type of operators
1. For non-heterogeneous models, onnx is only used for visualization, and it is not necessary to ensure actual availability. The default value is OperatorExportTypes.ONNX_ FALLTHROUGH
2. For heterogeneous models, onnx needs to be available in practice. Use None to ensure that the exported operator is a standard onnx operator.
`opset_version`: It can only be 11. Plugin has registered specific mapping rules in opset 11.
Note: If you use the public version torchonnxexport, you need to ensure that the above parameters are set correctly, and use import horizon_ plugin_ pytorch.utils._ register_ onnx_ Ops to register specific mapping rules with the opset 11.
"""
Use Process
Transform the floating point model.
Insert
QuantSubandDeQuantSubto keep consistent with non-heterogeneous usage.If the first op is a
cpu op, you do not need to insertQuantSub.If the last op is a
cpu op, you do not need to insertDeQuantSub.
For non
moduleoperations, if you need to setqconfigseparately or specify it to run on the CPU, you need to package it as amodule. Refer to the_ SeluModule.
Set the
march.Set the
qconfig. Keep the configuration mode of setting qconfig in themodulein non-heterogeneous mode. In addition, you can also pass inqconfigthrough theqconfig_dictparameter of theprepare_qat_fxinterface. For specific usage, see the interface parameter description.For
BPU op, you must ensure that there isqconfig. If the input op is notQuantSub, you must also ensure that the input op hasactivation qconfig.For
CPU op,qconfigwill not affect it in any way. However, ifBPU opis followed,qconfigmust be available.Recommended setting method: first set the global
qconfigtohorizon_plugin_pytorch.quantization.get_default_qat_qconfig(), based on which, it can be modified according to requirements. Generally speaking, you only need to setqconfigseparately for int16 and high-precision output op.
Set the
hybrid_dict. Optional. Please refer to the interface parameter description for specific usage. If there is no actively specified CPU operator, you need not sethybrid_dict.Call
prepare_calibration_fx. Optional. If the task is simple, the accuracy can be achieved after QAT, or you can skip to step 7. Generally speaking,calibrationis beneficial to the QAT accuracy. After printing thecalibrationmodel, you can see thatCalibFakeQuantizeis inserted where the quantization parameters need to be counted. Theconv4structure in the example is as follows:(conv4): Conv2d( 3, 3, kernel_size=(1, 1), stride=(1, 1) (weight_fake_quant): CalibFakeQuantize( (activation_post_process): NoopObserver() ) (activation_post_process): CalibFakeQuantize( (activation_post_process): CalibObserver(CalibObserver() calib_bin_edges=tensor([]) calib_hist=tensor([])) ) )
The
calibrationprocess requires the model to runforwardseveral times in theevalstate.Call
prepare_ qat_ fx. After printing the QAT model, you can see thatFakeQuantizeis inserted where fake quantization is required. Theconv4structure in the example is as follows:(conv4): Conv2d( 3, 3, kernel_size=(1, 1), stride=(1, 1) (weight_fake_quant): FakeQuantize( fake_quant_enabled=tensor([1], dtype=torch.uint8), observer_enabled=tensor([1], dtype=torch.uint8), quant_min=-128, quant_max=127, dtype=qint8, qscheme=torch.per_channel_symmetric, ch_axis=0, scale=tensor([0.0025, 0.0037, 0.0029]), zero_point=tensor([0, 0, 0]) (activation_post_process): MovingAveragePerChannelMinMaxObserver(min_val=tensor([-0.2484, -0.4718, -0.3689]), max_val=tensor([ 0.3239, -0.0056, 0.3312])) ) (activation_post_process): None )
To verify the correctness of the model, you can skip steps 8 and 9 after
prepare_qat_fx. First export onnx to view the model structure according to step 10, and then execute step 8 after verifying that there is no problem.training.Call
convert_fx. Optional. It can be skipped when there is no requirement to evaluate the precision of the fixed-point model.Call
export_ to_ onnx. You can also usetorch.onnx.export, but you need to follow the precautions in theexport_to_onnxinterface description.Use
hb_mapperto transform the onnx model. After conversion, it is necessary to check whether the operator is running on the expected device. In some cases,hb_mapperstill needs to set therun_on_cpuparameter. For example: althoughconvis not quantized in the QAT stage, since its input (the output of the previous operator) has been fake quantized,hb_mapperwill still quantize it by default.

7.3.4.3. Example¶
import numpy as np
import torch
from horizon_plugin_pytorch.march import March, set_march
from horizon_plugin_pytorch.nn import qat
from horizon_plugin_pytorch.quantization import (
get_default_calib_qconfig,
get_default_qat_qconfig,
get_default_qat_out_qconfig,
prepare_calibration_fx,
prepare_qat_fx,
convert_fx,
)
from torch import nn
from torch.quantization import DeQuantStub, QuantStub
from horizon_plugin_pytorch.utils.onnx_helper import export_to_onnx
class _ConvBlock(nn.Module):
def __init__(self, channels=3):
super().__init__()
self.conv = nn.Conv2d(channels, channels, 1)
self.prelu = torch.nn.PReLU()
def forward(self, input):
x = self.conv(input)
x = self.prelu(x)
return torch.nn.functional.selu(x)
# Encapsulate functional selu as a module, which is easy to set separately
class _SeluModule(nn.Module):
def forward(self, input):
return torch.nn.functional.selu(input)
class HybridModel(nn.Module):
def __init__(self, channels=3):
super().__init__()
# Insert QuantStub
self.quant = QuantStub()
self.conv0 = nn.Conv2d(channels, channels, 1)
self.prelu = torch.nn.PReLU()
self.conv1 = _ConvBlock(channels)
self.conv2 = nn.Conv2d(channels, channels, 1)
self.conv3 = nn.Conv2d(channels, channels, 1)
self.conv4 = nn.Conv2d(channels, channels, 1)
self.selu = _SeluModule()
# Insert DequantStub
self.dequant = DeQuantStub()
self.identity = torch.nn.Identity()
def forward(self, input):
x = self.quant(input)
x = self.conv0(x)
x = self.identity(x)
x = self.prelu(x)
x = torch.nn.functional.selu(x)
x = self.conv1(x)
x = self.conv2(x)
x = self.conv3(x)
x = self.identity(x)
x = self.conv4(x)
x = self.selu(x)
return self.dequant(x)
# Set march
set_march(March.XXX)
data_shape = [1, 3, 224, 224]
data = torch.rand(size=data_shape)
model = HybridModel()
# prepare_qat_fx will make inplace changes to the float model. Do not run inference on the float model until after prepare_qat_fx.
float_res = model(data)
calibration_model = prepare_calibration_fx(
model,
{
# Calibration fakes quant is only used for statistics. The unused calibration fake quant in the QAT stage will be automatically removed, and no special setting for high-precision output op is required
"": get_default_calib_qconfig(),
},
hybrid=True,
hybrid_dict={
"module_name": ["conv1.conv", "conv3"],
"module_type": [_SeluModule],
},
)
# The calibration phase should ensures that the original model will not change
calibration_model.eval()
for i in range(5):
calibration_model(torch.rand(size=data_shape))
qat_model = prepare_qat_fx(
calibration_model,
{
"": get_default_qat_qconfig(),
# Selu is the CPU operator. Conv4 is actually the output of the bpu model.
# It is set to high-precision output
"module_name": [("conv4", get_default_qat_out_qconfig())]
},
hybrid=True,
hybrid_dict={
"module_name": ["conv1.conv", "conv3"],
"module_type": [_SeluModule],
},
)
# prepare_qat_fx will make inplace changes to the float model. Do not run inference on the QAT prepare_qat_fx.
qat_res = qat_model(data)
# qat training start
# ......
# qat training end
# export qat.onnx
export_to_onnx(
qat_model,
data,
"qat.onnx",
enable_onnx_checker=True,
operator_export_type=None,
)
# Evaluate the fixed-point model
quantize_model = convert_fx(qat_model)
quantize_res = quantize_model(data)
Print the result of the calibration model.
HybridModel(
(quant): QuantStub(
(activation_post_process): CalibFakeQuantize(
(activation_post_process): CalibObserver(CalibObserver() calib_bin_edges=tensor([]) calib_hist=tensor([]))
)
)
(conv0): Conv2d(
3, 3, kernel_size=(1, 1), stride=(1, 1)
(weight_fake_quant): CalibFakeQuantize(
(activation_post_process): NoopObserver()
)
(activation_post_process): CalibFakeQuantize(
(activation_post_process): CalibObserver(CalibObserver() calib_bin_edges=tensor([]) calib_hist=tensor([]))
)
)
(prelu): PReLU(num_parameters=1)
(conv1): _ConvBlock(
(conv): Conv2d(3, 3, kernel_size=(1, 1), stride=(1, 1))
(prelu): PReLU(num_parameters=1)
)
(conv2): Conv2d(
3, 3, kernel_size=(1, 1), stride=(1, 1)
(weight_fake_quant): CalibFakeQuantize(
(activation_post_process): NoopObserver()
)
(activation_post_process): CalibFakeQuantize(
(activation_post_process): CalibObserver(CalibObserver() calib_bin_edges=tensor([]) calib_hist=tensor([]))
)
)
(conv3): Conv2d(3, 3, kernel_size=(1, 1), stride=(1, 1))
(conv4): Conv2d(
3, 3, kernel_size=(1, 1), stride=(1, 1)
(weight_fake_quant): CalibFakeQuantize(
(activation_post_process): NoopObserver()
)
(activation_post_process): CalibFakeQuantize(
(activation_post_process): CalibObserver(CalibObserver() calib_bin_edges=tensor([]) calib_hist=tensor([]))
)
)
(selu): _SeluModule()
(dequant): DeQuantStub()
(identity): Identity()
(prelu_input_dequant): DeQuantStub()
(selu_1_activation_post_process): CalibFakeQuantize(
(activation_post_process): CalibObserver(CalibObserver() calib_bin_edges=tensor([]) calib_hist=tensor([]))
)
(conv3_activation_post_process): CalibFakeQuantize(
(activation_post_process): CalibObserver(CalibObserver() calib_bin_edges=tensor([]) calib_hist=tensor([]))
)
(conv3_input_dequant): DeQuantStub()
(selu_2_input_dequant): DeQuantStub()
)
def forward(self, input):
input_1 = input
quant = self.quant(input_1); input_1 = None
conv0 = self.conv0(quant); quant = None
identity = self.identity(conv0); conv0 = None
prelu_input_dequant_0 = self.prelu_input_dequant(identity); identity = None
prelu = self.prelu(prelu_input_dequant_0); prelu_input_dequant_0 = None
selu = torch.nn.functional.selu(prelu, inplace = False); prelu = None
conv1_conv = self.conv1.conv(selu); selu = None
conv1_prelu = self.conv1.prelu(conv1_conv); conv1_conv = None
selu_1 = torch.nn.functional.selu(conv1_prelu, inplace = False); conv1_prelu = None
selu_1_activation_post_process = self.selu_1_activation_post_process(selu_1); selu_1 = None
conv2 = self.conv2(selu_1_activation_post_process); selu_1_activation_post_process = None
conv3_input_dequant_0 = self.conv3_input_dequant(conv2); conv2 = None
conv3 = self.conv3(conv3_input_dequant_0); conv3_input_dequant_0 = None
conv3_activation_post_process = self.conv3_activation_post_process(conv3); conv3 = None
identity_1 = self.identity(conv3_activation_post_process); conv3_activation_post_process = None
conv4 = self.conv4(identity_1); identity_1 = None
selu_2_input_dequant_0 = self.selu_2_input_dequant(conv4); conv4 = None
selu_2 = torch.nn.functional.selu(selu_2_input_dequant_0, inplace = False); selu_2_input_dequant_0 = None
dequant = self.dequant(selu_2); selu_2 = None
return dequant
Print the result of the QAT model.
HybridModel(
(quant): QuantStub(
(activation_post_process): FakeQuantize(
fake_quant_enabled=tensor([1], dtype=torch.uint8), observer_enabled=tensor([1], dtype=torch.uint8), quant_min=-128, quant_max=127, dtype=qint8, qscheme=torch.per_tensor_symmetric, ch_axis=-1, scale=tensor([0.0078]), zero_point=tensor([0])
(activation_post_process): MovingAverageMinMaxObserver(min_val=tensor([-0.9995]), max_val=tensor([0.9995]))
)
)
(conv0): Conv2d(
3, 3, kernel_size=(1, 1), stride=(1, 1)
(weight_fake_quant): FakeQuantize(
fake_quant_enabled=tensor([1], dtype=torch.uint8), observer_enabled=tensor([1], dtype=torch.uint8), quant_min=-128, quant_max=127, dtype=qint8, qscheme=torch.per_channel_symmetric, ch_axis=0, scale=tensor([0.0038, 0.0041, 0.0016]), zero_point=tensor([0, 0, 0])
(activation_post_process): MovingAveragePerChannelMinMaxObserver(min_val=tensor([-0.4881, -0.4944, 0.0787]), max_val=tensor([-0.1213, 0.5284, 0.1981]))
)
(activation_post_process): FakeQuantize(
fake_quant_enabled=tensor([1], dtype=torch.uint8), observer_enabled=tensor([1], dtype=torch.uint8), quant_min=-128, quant_max=127, dtype=qint8, qscheme=torch.per_tensor_symmetric, ch_axis=-1, scale=tensor([0.0064]), zero_point=tensor([0])
(activation_post_process): MovingAverageMinMaxObserver(min_val=tensor([-0.8159]), max_val=tensor([0.8159]))
)
)
(prelu): PReLU(num_parameters=1)
(conv1): _ConvBlock(
(conv): Conv2d(3, 3, kernel_size=(1, 1), stride=(1, 1))
(prelu): PReLU(num_parameters=1)
)
(conv2): Conv2d(
3, 3, kernel_size=(1, 1), stride=(1, 1)
(weight_fake_quant): FakeQuantize(
fake_quant_enabled=tensor([1], dtype=torch.uint8), observer_enabled=tensor([1], dtype=torch.uint8), quant_min=-128, quant_max=127, dtype=qint8, qscheme=torch.per_channel_symmetric, ch_axis=0, scale=tensor([0.0040, 0.0044, 0.0040]), zero_point=tensor([0, 0, 0])
(activation_post_process): MovingAveragePerChannelMinMaxObserver(min_val=tensor([-0.5044, -0.4553, -0.5157]), max_val=tensor([0.1172, 0.5595, 0.4104]))
)
(activation_post_process): FakeQuantize(
fake_quant_enabled=tensor([1], dtype=torch.uint8), observer_enabled=tensor([1], dtype=torch.uint8), quant_min=-128, quant_max=127, dtype=qint8, qscheme=torch.per_tensor_symmetric, ch_axis=-1, scale=tensor([0.0059]), zero_point=tensor([0])
(activation_post_process): MovingAverageMinMaxObserver(min_val=tensor([-0.7511]), max_val=tensor([0.7511]))
)
)
(conv3): Conv2d(3, 3, kernel_size=(1, 1), stride=(1, 1))
(conv4): Conv2d(
3, 3, kernel_size=(1, 1), stride=(1, 1)
(weight_fake_quant): FakeQuantize(
fake_quant_enabled=tensor([1], dtype=torch.uint8), observer_enabled=tensor([1], dtype=torch.uint8), quant_min=-128, quant_max=127, dtype=qint8, qscheme=torch.per_channel_symmetric, ch_axis=0, scale=tensor([0.0025, 0.0037, 0.0029]), zero_point=tensor([0, 0, 0])
(activation_post_process): MovingAveragePerChannelMinMaxObserver(min_val=tensor([-0.2484, -0.4718, -0.3689]), max_val=tensor([ 0.3239, -0.0056, 0.3312]))
)
(activation_post_process): None
)
(selu): _SeluModule()
(dequant): DeQuantStub()
(identity): Identity()
(prelu_input_dequant): DeQuantStub()
(selu_1_activation_post_process): _WrappedCalibFakeQuantize(
(activation_post_process): FakeQuantize(
fake_quant_enabled=tensor([1], dtype=torch.uint8), observer_enabled=tensor([1], dtype=torch.uint8), quant_min=-128, quant_max=127, dtype=qint8, qscheme=torch.per_tensor_symmetric, ch_axis=-1, scale=tensor([0.0042]), zero_point=tensor([0])
(activation_post_process): MovingAverageMinMaxObserver(min_val=tensor([-0.5301]), max_val=tensor([0.5301]))
)
)
(conv3_activation_post_process): _WrappedCalibFakeQuantize(
(activation_post_process): FakeQuantize(
fake_quant_enabled=tensor([1], dtype=torch.uint8), observer_enabled=tensor([1], dtype=torch.uint8), quant_min=-128, quant_max=127, dtype=qint8, qscheme=torch.per_tensor_symmetric, ch_axis=-1, scale=tensor([0.0072]), zero_point=tensor([0])
(activation_post_process): MovingAverageMinMaxObserver(min_val=tensor([-0.9156]), max_val=tensor([0.9156]))
)
)
(conv3_input_dequant): DeQuantStub()
(selu_2_input_dequant): DeQuantStub()
)
def forward(self, input):
input_1 = input
quant = self.quant(input_1); input_1 = None
conv0 = self.conv0(quant); quant = None
identity = self.identity(conv0); conv0 = None
prelu_input_dequant_0 = self.prelu_input_dequant(identity); identity = None
prelu = self.prelu(prelu_input_dequant_0); prelu_input_dequant_0 = None
selu = torch.nn.functional.selu(prelu, inplace = False); prelu = None
conv1_conv = self.conv1.conv(selu); selu = None
conv1_prelu = self.conv1.prelu(conv1_conv); conv1_conv = None
selu_1 = torch.nn.functional.selu(conv1_prelu, inplace = False); conv1_prelu = None
selu_1_activation_post_process = self.selu_1_activation_post_process(selu_1); selu_1 = None
conv2 = self.conv2(selu_1_activation_post_process); selu_1_activation_post_process = None
conv3_input_dequant_0 = self.conv3_input_dequant(conv2); conv2 = None
conv3 = self.conv3(conv3_input_dequant_0); conv3_input_dequant_0 = None
conv3_activation_post_process = self.conv3_activation_post_process(conv3); conv3 = None
identity_1 = self.identity(conv3_activation_post_process); conv3_activation_post_process = None
conv4 = self.conv4(identity_1); identity_1 = None
selu_2_input_dequant_0 = self.selu_2_input_dequant(conv4); conv4 = None
selu_2 = torch.nn.functional.selu(selu_2_input_dequant_0, inplace = False); selu_2_input_dequant_0 = None
dequant = self.dequant(selu_2); selu_2 = None
return dequant
The exported onnx is shown in the figure. The part circled in red is the CPU operator.
