# mypy: 允许未类型化定义
from 打字
导入
任何,
可调用,
可选,
联合
导入
火炬
导入 torch.ao.nn.quantized
是 nnq
导入 torch.ao.nn.quantized.dynamic
是 nnqd
导入 torch.nn
是
神经网络
from torch.ao.quantization 导入
准备
from torch.ao.quantization.quantization_mappings 导入 (
获取默认比较输出模块列表,
)
非叶子模块添加观察者允许列表 = {
氮气.
线性,
氮气.
线性,
氮气.
长短期记忆网络,
神经网络.
长短期记忆网络,
}
def _find_match(
字符串列表:
联盟[
字典[
字符串,
任何
]，
列表[
字符串]],
key_str: 字符串,
后缀:
字符串,
) -> 可选[
字符串
]：
分割字符串 = key_str.
分割(
“。”)
如果
分割字符串[-1] ==
后缀:
匹配字符串 =
输入文本翻译为简体中文为："".
连接(key_str.
分割(
“。”
)0:-1])
for s2 在
字符串列表:
模式 1 =
输入文本翻译为简体中文为："".
连接(s2.
分割(
“。”
)0:-1])
模式 2 =
输入文本翻译为简体中文为："".
连接(s2.
分割(
“。”
)0:-2])
如果
匹配字符串 ==
模式 1:
返回 s2
如果
匹配字符串 ==
模式 2:
返回 s2
用于匹配 "fc.weight" 和 "fc._packed_params._packed_params"
如果
后缀 == "_packed_params":
匹配字符串 =
输入文本翻译为简体中文为："".
连接(key_str.
分割(
“。”
)0:-2])
如果
长度(
匹配字符串) == 0:
返回
无
for s2 在
字符串列表:
模式 1 =
输入文本翻译为简体中文为："".
连接(s2.
分割(
“。”
)0:-1])
模式 2 =
输入文本翻译为简体中文为："".
连接(s2.
分割(
“。”
)0:-2])
如果
匹配字符串 ==
模式 1:
返回 s2
如果
匹配字符串 ==
模式 2:
返回 s2
返回
无
否则:
返回
无
[文档]def
比较权重(
浮点字典:
字典[
字符串,
任何
]，
量化字典:
字典[
字符串,
任何]
) -> 字典[
字符串,
字典[
字符串,
火把.
张量
]]
r比较浮点模块与其相应量化模块的权重
模块。返回一个字典，键对应模块名称，每个条目为
一个包含两个键 'float' 和 'quantized' 的字典，包含浮点数和
量化权重。此字典可用于比较和计算量化
浮点模型和量化模型权重错误。
演示用法：
wt_compare_dict = compare_weights(浮点模型.state_dict(), qmodel.state_dict())
for key in wt_compare_dict:
对于 wt_compare_dict 中的每个键：
print(
key,
compute_error(
wt_compare_dict[key]['float'],
wt_compare_dict[key]['quantized'].dequantize()
)
)
参数：
float_dict: 浮点模型的状态字典
quantized_dict: 量化模型的状态字典
返回：
weight_dict: 与模块名称对应的键的字典，每个条目为
一个包含两个键 'float' 和 'quantized' 的字典，包含浮点数和
量化权重
"文档"
火把._C._log_api_usage_once("quantization_api._numeric_suite.compare_weights")
重量字典:
字典[
字符串,
字典] = {}
for key 在
量化字典:
匹配键 = _find_match(
浮点字典,
键,
重量)
如果
匹配键
是
不
无:
重量字典[
键] = {}
重量字典[
键
]
[浮点数] =
浮点字典[
匹配键]
重量字典[
键
]
[量化] =
量化字典[
键]
continue
用于匹配 "fc.weight" 和 "fc._packed_params._packed_params"
匹配键 = _find_match(
浮点字典,
键, "_packed_params")
如果
匹配键
是
不
无:
重量字典[
键] = {}
重量字典[
键
]
[浮点数] =
浮点字典[
匹配键]
重量字典[
键
]
[量化] =
量化字典[
键
]
[0]
# 对于 LSTM
分割字符串 =
键.
分割(
“。”)
如果
分割字符串[-1] ==
"参数"
并且
分割字符串[-3] ==
"所有权重值":
层 =
分割字符串[-2]
模块名称 =
“。”.
连接(
分割字符串
[-3])
float_weight_ih_key = 模块名称 + ".weight_ih_l" +
层
float_weight_hh_key = 模块名称 + ".weight_hh_l" +
层
如果 float_weight_ih_key
在 float_dict
并且 float_weight_hh_key
在
浮点字典:
重量字典[
键] = {}
重量字典[
键
]
[浮点数] =
浮点字典[
浮点权重_ih 键]
重量字典[
键
]
[量化] = (
量化字典[
键].__getstate__()[0
]
[4
]
[0].__getstate__()[0
]
[0]
)
重量字典[
键
]
[浮点数] =
浮点字典[
浮点权重 hh 键]
重量字典[
键
]
[量化] = (
量化字典[
键].__getstate__()[0
]
[4
]
[1].__getstate__()[0
]
[0]
)
返回
重量字典
def _get_logger_dict_helper(
模块:
神经网络.
模块,
目标字典:
字典[
字符串,
任何
]，
前缀:
字符串 =
输入文本翻译为简体中文为："",
) -> 无:
r这是 get_logger_dict 的辅助函数
参数：
mod: 我们想要保存所有日志统计信息的模块
prefix: 当前模块的前缀
target_dict: 保存所有日志统计信息的字典
"文档"
def 获取前缀(
前缀):
返回
前缀
如果
前缀 ==
请提供需要翻译的文本
否则
前缀 + "."
for 名称,
儿童
在
模块.
命名子项():
如果 isinstance(
儿童,
记录器):
目标字典[
获取前缀(
前缀) +
统计] =
儿童.
统计
断开
for 名称,
儿童
在
模块.
命名子项():
模块前缀 =
获取前缀(
前缀) +
名称
如果
前缀
否则
名称
_get_logger_dict_helper(儿童,
目标字典,
模块前缀)
[文档]def get_logger_dict(mod: nn.Module, prefix: str = "") -> dict[str, dict]:
遍历模块并将所有日志统计信息保存到目标字典中。
这主要用于量化精度调试。
支持的日志记录器类型：
ShadowLogger：用于记录量化模块及其匹配的浮点阴影模块的输出，
OutputLogger：用于记录模块的输出
Args：
mod：我们要保存所有日志统计信息的模块
前缀：当前模块的前缀
返回：
目标字典：用于保存所有日志统计信息的字典
"""
torch._C._log_api_usage_once("量化_api._numeric_suite.get_logger_dict")
target_dict: dict[str, dict] = {}
_get_logger_dict_helper(mod, target_dict, prefix)
return target_dict
[文档]class Logger(nn.Module):
r"""日志统计的基础类"""
def __init__(self):
super().__init__()
self.stats = {}
# 仅当操作使用静态量化时才插入观察者，
# 这通过 activation_observer.dtype == quint8 来识别。这在将 Logger 作为观察者附加到 FX 模式时是必需的
# 当将 Logger 作为观察者附加到 FX 模式时是必需的
self.dtype = torch.quint8
[文档] def forward(self, x):
# fmt: off
"""
空白文档块以使 autodoc 高兴
[文档]类 ShadowLogger(日志记录器):
用于 Shadow 模块的类，用于记录原始输出
阴影模块。
"""
def __init__(self):
super().__init__()
self.stats["float"] = []
self.stats["quantized"] = []
[文档] def forward(self, x, y): # 类型：忽略[覆盖]
# fmt: off
```python
# 输入文本
input_text = '"""'
# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 假设的翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
``` # 空文档块以使自动文档功能满意
# fmt: 开启格式化
如果 len(x) > 1:
x = x[0]
如果 len(y) > 1:
y = y[0]
self.stats["quantized"].append(x.detach())
self.stats["float"].append(y.detach())
[文档]class OutputLogger(Logger):
类用于记录模块的输出
def __init__(self):
super().__init__()
self.stats["tensor_val"] = []
[文档] def forward(self, x):
# fmt: off
"""
""" # 空文档块以使 autodoc 高兴
# fmt: 开启格式化
self.stats["tensor_val"].append(x)
返回 x
任何) ->
任何:
返回 [
转换为列表(x) for x
在 t]
如果
类型(t)
是
元组
否则 t
def _反量化张量列表(t:
任何) ->
任何:
返回 (
[_反量化张量列表(x) for x
在 t]
如果
类型(t)
是
列表
否则 t.
反量化()
如果 t.
是否量化
否则 t
)
[文档]
类
影子(
神经网络.
模块):
r阴影模块将其浮点模块附加到其匹配的量化模块
作为阴影。然后它使用 Logger 模块处理两者的输出
模块
参数：
q_module：从 float_module 量化的模块，我们想要进行阴影
float_module：用于遮蔽 q_module 的浮点模块
logger_cls：用于处理 q_module 输出结果的日志器类型
float_module。可以使用 ShadowLogger 或自定义日志器。
"文档"
def __init__(我, q_module,
浮点模块,
日志类):
超级().__init__()
我.
原模块 =
Q 模块
我.
阴影模块 =
浮动模块
我.
反量化 =
氮气.
反量化化()
我.
记录器 =
日志类()
[文档] def forward(self, *x) -> torch.Tensor:
# fmt: off
"""
""" # 空文档块以使自动文档功能满意
# fmt: 开启格式化
xl = 将元组转换为列表(x)
输出 = self.orig_module(*xl)
xl_float = _dequantize_tensor_list(xl)
shadow_output = self.shadow_module(*xl_float)
self.logger(output, shadow_output)
return output
[文档] def add(self, x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
# fmt: off
```python
# 输入文本
input_text = '"""'
# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 假设的翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
``` # 空文档块以使自动文档功能满意
# fmt: 开启格式化
output = self.orig_module.add(x, y)
x = x.dequantize()
y = y.dequantize()
shadow_output = self.shadow_module.add(x, y)
self.logger(输出, 阴影输出)
返回输出
[文档] def add_scalar(self, x: torch.Tensor, y: float) -> torch.Tensor:
# fmt: off
```python
# 输入文本
input_text = '"""'
# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 假设的翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
``` # 空文档块以使自动文档功能满意
# fmt: 开启格式化
输出 = self.orig_module.add_scalar(x, y)
x = x.dequantize()
shadow_output = self.shadow_module.add_scalar(x, y)
self.logger(output, shadow_output)
return output
[文档] def mul(self, x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
# fmt: off
"""
""" # 空文档块以使自动文档功能正常
# fmt: 开启格式化
输出 = self.orig_module.mul(x, y)
x = x 反量化()
y = y.dequantize()
shadow_output = self.shadow_module.mul(x, y)
self.logger(output, shadow_output)
return output
[文档] def mul_scalar(self, x: torch.Tensor, y: float) -> torch.Tensor:
# fmt: off
```python
# 输入文本
input_text = '"""'
# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 假设的翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
``` # 空文档块以使自动文档功能满意
# fmt: 开启格式化
output = self.orig_module.mul_scalar(x, y)
x = x.dequantize()
shadow_output = self.shadow_module.mul_scalar(x, y)
self.logger(output, shadow_output)
返回输出
[文档] def cat(self, x: list[torch.Tensor], dim: int = 0) -> torch.Tensor:
# fmt: off
"""
# 空的文档块以使 autodoc 高兴
# fmt: on
output = self.orig_module.cat(x, dim)
x = [y.dequantize() for y in x]
shadow_output = self.shadow_module.cat(x, dim)
self.logger(output, shadow_output)
return output
[文档] def add_relu(self, x: torch.Tensor, y: torch.Tensor) -> torch.Tensor:
# fmt: off
```python
# 输入文本
input_text = '"""'
# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 假设的翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
``` # 空文档块以使自动文档功能满意
# fmt: 开启格式化
output = self.orig_module.add_relu(x, y)
x = x.dequantize()
y = y.dequantize()
shadow_output = self.shadow_module.add_relu(x, y)
self.logger(输出, 阴影输出)
返回输出
[文档]def 准备模型_with_stubs(
float_module: nn.Module,
q_module: nn.Module,
module_swap_list: set[type],
logger_cls: Callable,
) -> None:
准备模型，将其匹配的量化模块附加到模型上
模块作为浮点模块类型在模块_swap_list 中的影子。
示例用法：
准备带有占位符的模型(float_model, q_model, module_swap_list, Logger)
q_model(data)
ob_dict = get_logger_dict(q_model)
Args:
float_module: 用于生成 q_module 的浮点模块
q_module：从浮点模块量化的模块
module_swap_list：附加阴影的浮点模块类型列表
logger_cls：阴影模块中使用的日志记录器类型，用于处理
量化和其浮点阴影模块的输出
"""
torch._C._log_api_usage_once(
"量化 API._numeric_suite.prepare_model_with_stubs"
)
float_module_children = {}
for name, mod in float_module.named_children():
float_module_children[name] = mod
reassign = {}
for name, mod in q_module.named_children():
if name not in float_module_children:
continue
float_mod = float_module_children[name]
如果类型 float_mod 不在 module_swap_list 中：
准备带有占位符的模型 prepare_model_with_stubs(float_mod, mod, module_swap_list, logger_cls)
# 仅当模块类型与浮点模块不同时才插入阴影模块
# 不插入与浮点模块相同类型的模块
如果 type(float_mod) 在 module_swap_list 中，并且 not _is_identical_module_type(
mod, float_mod
):
reassign[name] = Shadow(mod, float_mod, logger_cls)
for key, value in reassign.items():
q_module._modules[key] = value
def _is_identical_module_type(模块 1,
模块 2):
# 比较两个模块是否具有相同的 dtype
mod1 模块类型 = [
类型(
模块) for
修饰
在
模块 1.
模块()]
mod2 模块类型 = [
类型(
模块) for
修饰
在
模块 2.
模块()]
返回
mod1 模块类型 ==
mod2 模块类型
[文档]def compare_model_stub(
float_model: nn.Module,
q_model: nn.Module,
module_swap_list: set[type],
*data,
logger_cls=ShadowLogger,
) -> dict[str, dict]:
r"""比较模型中量化模块与其浮点数对应版本，
使用相同的输入对两者进行喂入。返回一个字典，键对应模块名称，
每个条目都是一个字典，包含两个键 'float' 和
'量化，包含量化及其匹配的输出张量'
'浮点阴影模块。此字典可用于比较和计算模块'
'量化误差级别'
'此函数首先调用 prepare_model_with_stubs()以交换量化'
我们想要与 Shadow 模块进行比较的模块，该模块接收量化
模块、相应的浮点模块和记录器作为输入，并在内部创建
路径，以便浮点模块与量化阴影模块共享相同的输入。记录器
可自定义，默认记录器为 ShadowLogger
并且将保存量化模块和浮点模块的输出
可以用来计算模块级别的量化误差。
示例用法::
module_swap_list = [torchvision.models.quantization.resnet.QuantizableBasicBlock]
ob_dict = compare_model_stub(float_model,qmodel,module_swap_list, data)
for key in ob_dict:
print(key, compute_error(ob_dict[key]['float'], ob_dict[key]['quantized'].dequantize()))
Args:
float_model：用于生成 q_model 的浮点模型
q_model：从 float_model 量化的模型
module_swap_list：将附加阴影模块的浮点模块类型列表
将被附加。
输入数据，用于运行准备好的 q_model
日志类，用于在阴影模块中处理量化模块及其浮点阴影模块的输出
量化模块及其浮点阴影模块
"""
torch._C._log_api_usage_once("量化_api._numeric_suite.compare_model_stub")
prepare_model_with_stubs(float_model, q_model, module_swap_list, logger_cls)
q_model(*data)
ob_dict = get_logger_dict(q_model)
返回 ob_dict
[文档]def 获取匹配的激活(
浮点模块：nn.Module,
q 模块：nn.Module,
) -> dict[str, dict[str, torch.Tensor]]:
查找浮点模块和量化模块之间的匹配激活。
Args:
float_module: 用于生成 q_module 的浮点模块
q_module：从浮点模块量化的模块
返回：
act_dict：键对应量化模块名称的字典，每个条目都是一个包含
两个键 'float' 和 'quantized' 的字典
匹配浮点数和量化激活
"""
torch._C._log_api_usage_once(
"量化 API._numeric_suite.get_matching_activations"
)
float_dict = 获取 float_module 的日志字典
quantized_dict = 获取 q_module 的日志字典
act_dict: dict[str, dict] = {}
for key in quantized_dict:
if len(quantized_dict[key]["tensor_val"]) == 0:
continue
match_key = _find_match(sorted(float_dict, reverse=True), key, "stats")
如果 match_key 不为空：
act_dict[key] = {}
act_dict[key]["float"] = float_dict[match_key]["tensor_val"]
act_dict[key]["quantized"] = quantized_dict[key]["tensor_val"]
返回 act_dict
[文档]def prepare_model_outputs(
float_module: nn.Module,
q_module: nn.Module,
logger_cls=OutputLogger,
allow_list=None,
) -> None:
r"""准备模型，将日志记录器附加到 float 模块和量化模块，如果它们在 allow_list 中。
并将日志记录器附加到 float 模块和量化模块，如果它们在 allow_list 中。
Args:
float_module：用于生成 q_module 的浮点模块
q_module：从 float_module 量化的模块
logger_cls：附加到 float_module 和 q_module 的日志器类型
允许列表：附加日志的模块类型
"""
torch._C._log_api_usage_once(
"量化 API._numeric_suite.prepare_model_outputs"
)
如果 allow_list 为 None：
allow_list = get_default_compare_output_module_list()
qconfig_debug = torch.ao.quantization.QConfig(activation=logger_cls, weight=None)
float_module.qconfig = qconfig_debug # 忽略[赋值]警告
prepare(
float_module, inplace=True, allow_list=allow_list, prepare_custom_config_dict={}
)
q_module.qconfig = qconfig_debug # 忽略[赋值]错误
prepare(
q_module,
inplace=True,
allow_list=allow_list,
observer_non_leaf_module_list=NON_LEAF_MODULE_TO_ADD_OBSERVER_ALLOW_LIST,
prepare_custom_config_dict={},
)
[文档]def compare_model_outputs(
float_model: nn.Module,
q_model: nn.Module,
*data,
logger_cls=OutputLogger,
allow_list=None,
) -> dict[str, dict[str, torch.Tensor]]:
r"""比较浮点数和量化模型之间的输出激活
对应相同输入的位置。返回一个字典，键对应
到量化模块名称，每个条目都是一个包含两个键的字典
'float' 和 'quantized'，包含量化模型的激活和
浮点模型在匹配位置的激活。此字典可用于比较和
计算传播量化误差。
示例用法::
act_compare_dict = compare_model_outputs(float_model, qmodel, data)
for key in act_compare_dict:
print(
key,
compute_error(
act_compare_dict[key]['float'],
act_compare_dict[key]['quantized'].反量化()
)
)
Args:
float_model：用于生成 q_model 的浮点模型
q_model：从 float_model 量化的模型
data：用于运行准备好的 float_model 和 q_model 的输入数据
logger_cls：附加到 float_module 和 q_module 的日志记录器类型
允许列表：附加日志记录器的模块类型列表
返回：
act_compare_dict：与量化模块名称对应的键的字典
并且每个条目都是一个包含两个键 'float' 和 'quantized' 的字典
包含匹配的浮点数和量化激活
"""
torch._C._log_api_usage_once(
"量化 API._numeric_suite.compare_model_outputs"
)
如果 allow_list 为 None：
allow_list = get_default_compare_output_module_list()
prepare_model_outputs(float_model, q_model, logger_cls, allow_list)
float_model(*data)
q_model(*data)
act_compare_dict = get_matching_activations(float_model, q_model)
return act_compare_dict
