# mypy: 允许未类型化定义
""
本模块包含用于比较模型权重和激活的工具
在模型之间。示例用法：
导入 copy
导入 torch
导入 torch.ao.quantization.quantize_fx 作为 quantize_fx
导入 torch.ao.ns._numeric_suite_fx 作为 ns
m = torch.nn.Sequential(torch.nn.Conv2d(1, 1, 1)).eval()
mp = quantize_fx.prepare_fx(m, {'': torch.ao.quantization.default_qconfig})
# 我们转换一个副本，因为我们需要原始的已准备模型
# 用于比较，并且 `quantize_fx.convert_fx` 是原地转换。
mq = quantize_fx.convert_fx(copy.deepcopy(mp))
#
# 比较权重
#
# 提取权重对
weight_comparison = ns.extract_weights('a', mp, 'b', mq)
为每个比较添加 SQNR，就地
ns.extend_logger_results_with_comparison(
weight_comparison, 'a', 'b', torch.ao.ns.fx.utils.compute_sqnr,
'sqnr')
# 权重比较包含从 `mp` 和 `mq` 存储的权重，并可用于进一步分析。
以成对形式出现，可用于进一步分析。
#
比较激活，带有错误传播
#
添加日志记录器
mp_ns, mq_ns = ns.add_loggers(
'a', 深拷贝(mp),
'b', 深拷贝(mq),
ns.OutputLogger)
# 发送一个示例数据以捕获中间激活
datum = torch.randn(1, 1, 1, 1)
mp_ns(数据)
mq_ns(数据)
提取中间激活
act_comparison = ns.extract_logger_info(
mp_ns, mq_ns, ns.OutputLogger, 'b')
为每个比较添加 SQNR，就地
ns.extend_logger_results_with_comparison(
act_comparison, 'a', 'b', torch.ao.ns.fx.utils.compute_sqnr,
'sqnr')
# act_comparison 包含了 `mp_ns` 和 `mq_ns` 的激活，并存储成对，可用于进一步分析。
以成对形式出现，可用于进一步分析。
#
# 比较激活，不进行错误传播
#
创建阴影模型
mp_shadows_mq = ns.add_shadow_loggers(
'a', 深拷贝(mp),
'b', 深拷贝(mq),
ns.OutputLogger)
# 发送一个示例数据以捕获中间激活
datum = torch.randn(1, 1, 1, 1)
mp_shadows_mq(datum)
提取中间激活
shadow_act_comparison = ns.extract_shadow_logger_info(
mp_shadows_mq, ns.OutputLogger, 'b')
为每个比较添加 SQNR，就地
ns.extend_logger_results_with_comparison(
shadow_act_comparison, 'a', 'b', torch.ao.ns.fx.utils.compute_sqnr,
'sqnr')
# shadow_act_comparison 包含从 `mp_ns` 和 `mq_ns` 存储的激活
以成对形式出现，可用于进一步分析。
""
导入
集合
从
打字
导入
任何,
可调用,
可选,
类型检查
导入
火炬
导入 torch.ao.quantization.quantize_fx
是
量化效果
导入 torch.nn
是
神经网络
from torch.ao.ns.fx.graph_matcher 导入
获取匹配的子图对
from torch.ao.ns.fx.mappings 导入
获取基本名称到相关操作集合的映射
from torch.ao.ns.fx.n_shadows_utils 导入 (
获取去重子图,
创建添加日志记录器的图,
创建 n 个转换并记录的子图副本,
创建结果比较,
提取权重比较,
按子图分组结果,
输出属性,
打印 n 个阴影摘要,
阴影包装节点名称前缀,
)
from torch.ao.ns.fx.qconfig_multi_mapping 导入 QConfigMultiMapping
from torch.ao.quantization 导入
QConfig 映射
from torch.ao.quantization.backend_config 导入
后端配置
from torch.ao.quantization.backend_config.utils 导入 (
get_fusion_pattern_to_root_node_getter,
)
from torch.ao.quantization.fx.graph_module 导入 _get_observed_graph_module_attr
from torch.ao.quantization.fx.量化匹配工具
导入 _find_matches
from torch.ao.quantization.fx.量化配置映射工具
导入 (
_generate_node_name_to_qconfig,
)
from torch.ao.quantization.fx.quantize_handler 导入 _get_pattern_to_quantize_handlers
from torch.fx 导入
图模块
从 torch.fx.graph
导入
节点
来自 .fx.graph_passes
导入 add_loggers_to_model,
创建阴影_b
from .fx.ns_types 导入
NS 节点目标类型,
NS 结果类型, NSSingleResultValuesType
from .fx.utils 导入 (
获取目标类型字符串,
可能添加缺失的全限定名,
在模型节点名称上重新键入日志信息,
)
from .fx.权重工具
导入
从节点提取权重
如果
类型检查:
from torch.ao.quantization.qconfig 导入 QConfigAny
RNN 返回类型 =
元组[
火把.
张量,
元组[
火把.
张量,
火把.
张量]]
[文档]
类
输出记录器(
神经网络.
模块):
""
捕获中间值的基类。
"文档"
统计:
列表[torch.
张量]
统计 RNN:
列表[
RNN 返回类型]
# 标记为不纯，以便在 DCE 期间调用它时不会被删除。
_is_impure = 真实
def __init__(
我,
参考节点名称:
字符串,
前节点名称:
字符串,
模型名称:
字符串,
参考名称:
字符串,
前节点目标类型:
字符串,
引用节点目标类型:
字符串,
结果类型:
字符串,
参数内索引:
整数,
索引参数:
整数,
完全限定名:
可选[
字符串
]，
qconfig 字符串:
可选[
字符串] =
输入文本翻译为简体中文为："",
):
超级().__init__()
我.
统计:
列表[
PyTorch.
张量] =
输入文本为空，请提供需要翻译的文本
我.
统计 RNN:
列表[
RNN 返回类型] =
输入文本为空，请提供需要翻译的文本
负责添加此日志记录器的节点名称
备注：
- 如果我们正在记录节点输出，则此名称与 prev_node_name 相同
- 如果我们正在记录节点输入，则此名称为节点名称
# 该记录器记录的输入。
#
# 例如，logger1 记录 op1 的输入，logger2 记录
# op1 的输出：
#
# x1 -> logger1 -> op1 -> logger2 -> x2
#
# 在此示例中，
# - logger1 的前节点名称是 x1，引用节点名称是 op1
# - logger2 的前节点名称是 op1，引用节点名称是 op1
我.
节点引用名称 =
节点引用名称
# 该 Logger 捕获输出的节点名称
我.
前节点名称 =
前节点名称
模型节点的来源名称
我.
模型名称 =
模型名称
# 参考名称，用于匹配来自不同模型的记录器
互相
我.
引用名称 =
引用名称
该节点输出此记录器记录的目标的类型
我.prev_node_target_type = prev_node_target_type
# 节点负责添加此内容的类型
# 日志记录器
我.
引用节点目标类型 =
引用节点目标类型
# 统计中包含哪些值
我.
结果类型 =
结果类型
# 输入/输出节点参数中的此节点索引
# 例如，在 cat([x1, x2, x3], dim=0) 中，x2 的 index_within_arg 等于 1
我.
输入索引 =
输入索引
# 输入/输出节点参数中的节点索引
# 例如，在 add(x1, x2) 中，x2 的 index_of_arg 等于 1
我.
索引参数 =
索引参数
完全限定名称
我.
完全限定名 =
完全限定名
如果在 prepare_fx 之前添加了日志记录器，但我们不想
仅收集校准结果，仅收集 convert_fx 后的结果
因此，我们添加一个标志来控制这个日志记录器是否收集数据
我.
启用 =
真实
qconfig 的字符串表示
我.qconfig_str = qconfig_str
# 这可以在校准期间关闭以减少内存使用
我.
保存激活 =
真实
# 注意：无法注释 x 的类型，因为 TorchScript 不支持
# 联合类型。
[文档] def forward(self, x):
# fmt: off
"""
""" # 空文档块以使自动文档功能满意
# fmt: 开启格式化
# TODO(future PR): 考虑设计得更好，因为差异
这两个标志之间的差异微妙且不明显。
如果未 self.enabled:
return x
if not self.save_activations:
return x
# TODO(future PR): consider refactoring this to better reuse the parent
类
如果 isinstance(x, torch.Tensor):
self.stats.append(x.detach())
elif isinstance(x, tuple) and len(x) == 2 and len(x[1]) == 2:
new_res = (x[0].detach(), (x[1][0].detach(), x[1][1].detach()))
self.stats_rnn.append(new_res)
return x
def __repr__(我):
清理字典 = {
k: v
for k, v 在
我.
字典.
项目()
# 跳过 nn.Module 键
如果 (k !=
"训练")
并且
不 k.
以...开头(
“_”)
}
返回 f
"输出记录器("{
清洁字典})"
[文档]类 OutputComparisonLogger(OutputLogger):
"""
与 OutputLogger 相同，但还需要原始激活
为了在标定时间计算比较
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
def __init__(self, *参数, **关键字参数):
super().__init__(*args, **kwargs)
# TODO(未来 PR): 使比较函数可配置
self.comparison_fn = torch.ao.ns.fx.utils.compute_sqnr
self.comparison_fn_name = "sqnr"
# 预计算的日志输出与参考的比较
self.comparisons = []
# 预计算的比较函数
[文档] def forward(self, x, x_ref): # type: ignore[override]
# fmt: off
"""
""" # 空的文档块以使自动文档满意
# 开启格式化
如果没有 self.enabled:
return x
assert isinstance(x, torch.Tensor), "非张量输入尚未支持"
if self.save_activations:
# 保存激活，用于调试
self.stats.append(x.detach())
# 保存比较
self.comparisons.append(self.comparison_fn(x, x_ref))
return x
def __repr__(self): clean_dict = { k: v for k, v in self.__dict__.items() # 跳过 nn.Module 键 if (k != "training") and not k.startswith("_") } return f"OutputComparisonLogger({clean_dict})"
[文档]class NSTracer(quantize_fx.量化追踪器):
"""
就像常规的 FX 量化追踪器一样，但处理观察者和 fake_quantize
模块作为叶子模块。
"""
[文档] def is_leaf_module(self, m: torch.nn.Module, module_qualified_name: str) -> bool:
# fmt: off
"""
"""
""" # 空的文档块以使自动文档满意
# 开启格式化
如果 isinstance(m, torch.ao.quantization.ObserverBase):
返回 True
如果 isinstance(m, torch.ao.quantization.FakeQuantizeBase):
返回 True
返回 super().is_leaf_module(m, module_qualified_name)
字符串,
模型: GraphModule,
nodes_and_names_to_instrument: 列表[
元组[
节点,
字符串]],
结果:
NS 结果类型,
op_to_type_to_weight_extraction_fn: 可选[
字典[
字符串,
字典[
可调用,
可调用]]
] = 无,
) -> 无:
火把._C._log_api_usage_once(
"量化_api._数值套件_fx._从单一模型提取权重"
)
for node, 引用名称
在 nodes_and_names_to_instrument:
资源类型 =
单一结果值类型.
重量.value
提取的重量 =
从节点提取权重(
node, 模型,
操作符到类型到权重提取函数
)
如果
提取重量:
如果
参考名称
非
在
结果:
结果[
参考名称] = {
资源类型: {}}
结果[
参考名称
]
[资源类型
]
[模型名称] = [
提取权重]
def _提取权重实现(
模型名_a:
字符串,
gm_a: 图模块,
模型名称_b:
字符串,
gm_b: GraphModule,
基础名称到相关操作集合映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] =
无,
无法匹配的类型映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] =
无,
op_to_type_to_weight_extraction_fn: 可选[
字典[
字符串,
字典[
可调用,
可调用]]
] = 无,
) -> NS 结果类型:
火把._C._log_api_usage_once(
"量化 API._数值套件 FX._提取权重实现"
)
匹配的子图对 =
获取匹配的子图对(
gm_a, gm_b, 基础名称到相关操作集合映射,
无法匹配的类型映射
)
将子图对分割成每个模型的一个数据结构
要监控的节点和名称:
列表[
元组[
节点,
字符串]] =
输入文本为空，请提供需要翻译的文本
节点与名称要检测_b:
列表[
元组[
节点,
字符串]] =
输入文本为空，请提供需要翻译的文本
for 匹配名称,
匹配
在
匹配的子图对.
项目():
subgraph_a, 子图_b =
匹配
要监控的节点和名称.append((subgraph_a.
基础操作节点,
匹配名称))
节点与名称要检测_b.append((subgraph_b.
基础操作节点,
匹配名称))
# 逐个模型填充结果
结果:
NS 结果类型 = {}
_从单个模型提取权重(
模型名_a,
gm_a,
要监控的节点和名称,
结果,
op_to_type_to_weight_extraction_fn,
)
_从单个模型提取权重(
模型名_b,
gm_b,
节点与名称要检测_b,
结果,
op_to_type_to_weight_extraction_fn,
)
# 填充缺失的 fqn 条目
可能添加缺失的全限定名(
结果)
在 gm_b 中对节点名称进行重新键值
结果 =
在模型节点名称上重新键值 logger_info(
结果,
模型名_b)
返回
结果
def _添加日志器_单个模型(
模型名称:
字符串,
模型: GraphModule,
要监控的节点和名称:
列表[
元组[
节点,
字符串,
字符串]],
要仪器化的节点和名称:
列表[
元组[
节点,
字符串,
字符串]],
日志类:
可调用,
) -> 神经网络.
模块:
PyTorch._C._log_api_usage_once(
"量化 API._numeric_suite_fx._添加单个模型的日志记录器"
)
# TODO(未来 PR): 不要观察我们不关心的节点
# 关于（包括 fp32、拒绝列表等）
将节点到仪器输入映射到参考名称:
字典[
节点,
元组[
字符串,
字符串]] = {}
节点将输出映射到引用名称:
字典[
节点,
元组[
字符串,
字符串]] = {}
为 node,
参考名称,
参考节点类型
在
要监控的节点和名称:
将节点到仪器输入映射到参考名称[node] = (
参考名称,
参考节点类型)
为 node,
参考名称,
参考节点类型
在
要仪器化的节点和名称:
节点将输出映射到引用名称[node] = (
参考名称,
参考节点类型)
模型 = add_loggers_to_model(
模型,
将节点到仪器输入映射到参考名称,
节点将输出映射到引用名称,
日志类,
模型名称,
)
返回
模型
def _添加日志实现(
name_a: 字符串,
gm_a: GraphModule,
name_b: 字符串,
gm_b: GraphModule,
日志类:
可调用,
应记录输入:
布尔,
基础名称到相关操作集合映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] = None,
无法匹配的类型映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] = None,
) -> 元组[
神经网络.
模块,
神经网络.
模块
]：
PyTorch._C._log_api_usage_once("quantization_api._numeric_suite_fx._add_loggers_impl")
匹配的子图对 =
获取匹配的子图对(
gm_a, gm_b, 基础名称到相关操作集合映射,
无法匹配的类型映射
)
节点和名称到要仪器化的输入的映射_a =
输入文本为空，请提供需要翻译的文本
节点与名称到要监控的输入 =
输入文本为空，请提供需要翻译的文本
节点与名称到要监控的输出_a =
输入文本为空，请提供需要翻译的文本
节点与名称到要监控的输出_b =
输入文本为空，请提供需要翻译的文本
为
匹配名称, (subgraph_a, subgraph_b)
在
匹配的子图对.
项目():
ref_node_type_a = 获取目标类型字符串(subgraph_a.
基础操作节点, gm_a)
ref_node_type_b = 获取目标类型字符串(subgraph_b.
基础操作节点, gm_b)
# 注意：对于匹配输入，我们使用起始节点，例如观察
# 线性-relu 中的线性输入
如果
应记录输入:
节点及名称到仪器输入_a.append(
(subgraph_a.开始节点,
匹配名称,
参考节点类型_a)
)
节点与名称到仪器输入_b.append(
(subgraph_b.开始节点,
匹配名称,
参考节点类型 B)
)
# 注意：对于匹配激活，我们始终使用 end_node，
# 例如观察线性-relu 的输出中的 relu，
节点及名称到仪器输出_a.append(
(subgraph_a.结束节点,
匹配名称,
参考节点类型_a)
)
节点与名称到仪器输出_b.append(
(subgraph_b.结束节点,
匹配名称,
参考节点类型 B)
)
新模型 A =
_添加日志器_单个模型(
name_a,
gm_a,
节点及名称到仪器输入_a,
节点及名称到仪器输出_a,
日志类,
)
新模型_b =
_添加日志器_单个模型(
name_b,
gm_b,
节点与名称到仪器输入_b,
节点与名称到仪器输出_b,
日志类,
)
返回 (
新模型_a,
新模型_b)
[文档]def
添加日志记录器(
name_a: 字符串,
model_a: 神经网络.
模块,
name_b: 字符串,
模型_b:
神经网络.
模块,
logger 类:
可调用,
应记录输入:
布尔类型 =
错误,
基础名称到相关操作集合:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] =
无,
无法匹配的类型映射:
可选[
字典[
字符串,
集合[
NS 节点目标类型]]] =
无,
) -> 元组[
神经网络.
模块,
神经网络.
模块
]：
""
使用记录器对仪器模型 A 和模型 B 进行建模。
参数：
name_a: 字符串 模型 A 在结果中使用的名称
model_a: 模型 A
name_b: 字符串 模型 B 在结果中使用的名称
model_b: 模型 B
logger_cls: 要使用的 Logger 类
base_name_to_sets_of_related_ops: 可选覆盖子图基本节点，可能发生变化
unmatchable_types_map: 可选覆盖不可匹配的类型，可能发生变化
返回：
返回一个包含(model_a_with_loggers, model_b_with_loggers)的元组。就地修改两个模型。
"文档"
torch._C._log_api_usage_once(quantization_api._numeric_suite_fx.add_loggers)
# TODO(未来 PR): 公开这些
跳过的模块名称:
列表[
字符串] =
输入文本为空，请提供需要翻译的文本
跳过的模块类:
列表[
可调用] =
输入文本为空，请提供需要翻译的文本
tracer_a = NS 跟踪器(
跳过的模块名称,
跳过的模块类)
跟踪器_b =
NS 跟踪器(
跳过的模块名称,
跳过的模块类)
gm_a = GraphModule(model_a, 跟踪器_a.
跟踪(model_a))
maybe_model_a 节点名称到作用域 =
_获取观察到的图模块属性(
model_a, "节点名称到作用域"
)
if maybe_model_a 节点名称到作用域
是
不 None:
gm_a._节点名称到作用域_ =
maybe_model_a 节点名称到作用域
gm_b = GraphModule(模型_b,
跟踪器_b.
跟踪(
模型_b))
maybe_model_b_node_name_to_scope = _获取观察到的图模块属性(
模型_b,
"节点名称到作用域"
)
如果 maybe_model_b_node_name_to_scope
是
不 None:
gm_b._节点名称到作用域_ = maybe_model_b_node_name_to_scope
返回
_添加日志实现(
name_a,
gm_a,
name_b,
gm_b,
日志类,
应记录输入=
应记录输入,
基础名称到相关操作集合映射=
基础名称到相关操作集合映射,
无法匹配的类型映射=
无法匹配的类型映射,
)
def _提取一个模型的信息(
模型:
神经网络.
模块,
结果:
NS 结果类型,
日志类:
可调用,
) -> None:
PyTorch._C._log_api_usage_once(
"quantization_api._numeric_suite_fx._extract_logger_info_one_model"
)
for _gm_name, 修饰
在
模型.
命名模块():
# TODO(未来 PR): 更好的脚本检查
is_logger = isinstance(模块,
日志类)
或者 ( # type: ignore[arg-type]
isinstance(模块,
火把.
算子.
递归脚本模块)
并且
模块.
原始名称 ==
输出记录器
)
如果
是否为记录器:
key = 模块.
引用名称
如果 key
不
在
结果:
结果[
键] = {}
断言 (
模块.
模型名称
不
在
结果[
键]
), f"{模块.
模型名称}
已存在于结果中
如果
模块.
结果类型
不
在
结果[
键
]：
结果[
键
]
[模块.
结果类型] = {}
如果
模块.
模型名称
不
在
结果[
键
]
[模块.
结果类型
]
结果[
键
] [
模块.
结果类型
]
[模块.
模型名称] =
输入文本为空，请提供需要翻译的文本
可用统计 =
模块.
统计
如果
长度(
模块.
统计 RNN) > 0:
可用统计 =
模块.stats_rnn
数据 = {
类型:
模块.
结果类型,
"值":
要使用的统计数据,
"引用节点名称":
模块.
参考节点名称,
"节点目标类型":
模块.
引用节点目标类型,
"前节点名称":
模块.
前节点名称,
"prev_node_target_type": 模块.
前节点目标类型,
"index_within_arg": 模块.
参数内索引,
"参数索引":
模块.
索引参数,
"全限定名":
模块.
完全限定名,
"qconfig 字符串":
模块.
qconfig 字符串,
}
if 有属性(
模块, "comparisons"):
数据["comparisons"] =
模块.
比较
数据["comparison_fn_name"] =
模块.comparison_fn_name
否则:
数据["comparisons"] =
输入文本为空，请提供需要翻译的文本
数据["comparison_fn_name"] =
请提供需要翻译的文本
结果[
键
]
[模块.
结果类型
]
[模块.
模型名称].
追加(
数据)
# 确保列表保持排序
结果[
键
]
[模块.
结果类型
]
[模块.
模型名称].
排序(
键=lambda
资源: f"{
资源[
参数索引]}:{
资源[
参数内索引]}"
)
# TODO(未来 PR): 在命名上达成一致
这部分相当于 `ns.compare_model_outputs` 的比较提取部分
def _add_shadow_loggers_impl(
name_a: 字符串,
gm_a: GraphModule,
name_b: 字符串,
gm_b: GraphModule,
日志类:
可调用,
应记录输入:
布尔,
基础名称到相关操作集合映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] = None,
节点类型到输入输出类型映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] = None,
无法匹配的类型映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] = None,
) -> 神经网络.
模块:
PyTorch._C._log_api_usage_once(
"quantization_api._numeric_suite_fx._add_shadow_loggers_impl"
)
匹配的子图对 =
获取匹配的子图对(
gm_a, gm_b, 基础名称到相关操作集合映射,
无法匹配的类型映射
)
gm_a_shadows_b = 创建阴影_b(
name_a,
gm_a,
name_b,
gm_b,
匹配的子图对,
日志类,
应记录输入=
应记录输入,
节点类型到输入输出类型映射=
节点类型到输入输出类型映射,
)
返回 gm_a_shadows_b
[文档]def
添加阴影记录器(
name_a: 字符串,
model_a: 神经网络.
模块,
name_b: 字符串,
模型_b:
神经网络.
模块,
日志类:
可调用,
应记录输入:
布尔 =
假的,
基础名称到相关操作集合映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] = None,
节点类型到输入输出类型映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] = None,
无法匹配的类型映射:
可选[
字典[
字符串,
设置[
NS 节点目标类型]]] = None,
) -> 神经网络.
模块:
""
使用阴影记录器对 A 型仪器和 B 型仪器进行模型测试。
参数：
name_a: 字符串 模型 A 在结果中使用的名称
model_a: 模型 A
name_b: 字符串 模型 B 在结果中使用的名称
model_b: 模型 B
logger_cls: 要使用的 Logger 类
should_log_inputs: 是否记录输入
base_name_to_sets_of_related_ops: 可选覆盖子图基本节点，可能发生变化
unmatchable_types_map: 可选覆盖不可匹配的类型，可能发生变化
"文档"
PyTorch._C._log_api_usage_once(
"量化 API._numeric_suite_fx 添加阴影记录器"
)
# TODO(未来 PR): 公开这些
跳过的模块名称:
列表[
字符串] =
输入文本为空，请提供需要翻译的文本
跳过的模块类:
列表[
可调用] =
输入文本为空，请提供需要翻译的文本
tracer_a = NS 跟踪器(
跳过的模块名称,
跳过的模块类)
跟踪器_b =
NS 跟踪器(
跳过的模块名称,
跳过的模块类)
gm_a = GraphModule(model_a, 跟踪器_a.
跟踪(model_a))
maybe_model_a 节点名称到作用域 =
_获取观察到的图模块属性(
model_a, "节点名称到作用域"
)
如果
maybe_model_a 节点名称到作用域
是
不 None:
gm_a._节点名称到作用域_ =
maybe_model_a 节点名称到作用域
gm_b = GraphModule(模型_b,
跟踪器_b.
跟踪(
模型_b))
maybe_model_b_node_name_to_scope = _获取观察到的图模块属性(
模型_b,
"节点名称到作用域"
)
如果 maybe_model_b_node_name_to_scope
是
不 None:
gm_b._节点名称到作用域_ = maybe_model_b_node_name_to_scope
返回 _add_shadow_loggers_impl(
name_a,
gm_a,
name_b,
gm_b,
日志类,
应记录输入=
应记录输入,
基础名称到相关操作集合映射=
基础名称到相关操作集合映射,
节点类型到输入输出类型映射=
节点类型到输入输出类型映射,
无法匹配的类型映射=
无法匹配的类型映射,
)
[文档]def extract_shadow_logger_info(
model_a_shadows_b: nn.Module,
logger_cls: Callable,
model_name_to_use_for_layer_names: str,
) -> NSResultsType:
```python
# 假设输入文本为：
input_text = '"""'
# 翻译函数（此处仅为示例，实际翻译功能需要调用真实的翻译 API）
def translate_to_simplified_chinese(text):
# 这里应该调用真实的翻译 API 进行翻译
# 由于示例中不使用真实的 API，以下为模拟翻译结果
return text
# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
遍历影子模型中的所有记录器，并提取记录内容
信息。
参数:
model_a_shadows_b: 影子模型
logger_cls: 要使用的 Logger 类
model_name_to_use_for_layer_names: 输出层名称中要使用的模型名称
layer_names_in_the_output: 输出层名称
返回：
包含已记录比较的 NSResultsType
"""
torch._C._log_api_usage_once(
quantization_api._numeric_suite_fx.extract_shadow_logger_info
)
results: NSResultsType = collections.defaultdict(dict)
_extract_logger_info_one_model(model_a_shadows_b, results, logger_cls)
填充缺失的 fqn 条目
可能添加缺失的 fqn(results)
在模型 b 的名称上重新键
results = 在模型节点的名称上重新键入日志信息(rekey_logger_info_on_node_name_of_model(
结果，用于层名称的模型名称
)
返回字典（results）
[文档]def
扩展日志结果以包含比较(
结果:
NS 结果类型,
模型名称_1:
字符串,
模型名称_2:
字符串,
比较函数:
可调用[[
PyTorch.
张量,
PyTorch.
张量
]，
PyTorch.
张量
]，
比较名称:
字符串,
) -> None:
""
将 `model_name_2` 的日志值与 `model_name_1` 中相应的值进行比较
使用 `comparison_fn` 记录结果
在`model_name_2`的`comparison_name`结果下。就地修改`results`。
参数：
结果：来自`extract_logger_info`或`extract_shadow_logger_info`的结果数据结构。
`extract_shadow_logger_info`。
model_name_1：模型 1 的字符串名称。
模型名称_2：模型 2 的字符串名称
比较函数：比较两个张量的函数
比较名称：用于比较的模型的字符串名称
输出层的名称
"文档"
为
结果类型到结果
在
结果.
值():
为
模型名称到结果
在
结果类型到结果.
值():
断言 (
模型名称_1
在
模型名称到结果
), f"{模型名称_1}
未在结果中找到
断言 (
模型名称_2
在
模型名称到结果
), f"{模型名称 2}
未在结果中找到
结果_1 =
模型名称到结果[
模型名称 1]
结果 2 =
模型名称到结果[
模型名称_2]
为
结果_2
在
结果_2:
参数内的索引_2 =
结果_2["index_within_arg"]
arg_2 的索引 =
结果_2[
"参数索引"]
查找对应的结果_1
结果_1 =
无
为
当前结果_1
在
结果_1:
参数_1 内的索引 =
当前结果_1["index_within_arg"]
参数_1 的索引 = cur_result_1[
"参数索引"]
如果 (index_within_arg_1 == index_within_arg_2)
和 (
index_of_arg_1 == 参数 2 的索引
):
结果 1 =
当前结果 1
断开
断言
结果 1
是
否。
无
values_1 = result_1["值"]
values_2 = 结果_2[
"值"]
结果_2[
比较名称] =
输入文本为空，请提供需要翻译的文本
为 value_1, value_2
在 zip(values_1, values_2):
比较结果 =
比较函数(value_1, value_2)
结果_2[
比较名称].
追加(
比较结果)
[文档]def
准备_n_shadows 模型(
模型:
火把.
神经网络.
模块,
示例输入:
任何,
qconfig 多映射:
QConfig 多映射,
后端配置:
后端配置,
自定义准备函数:
可选[
可调用] = None,
自定义准备参数:
可选[
字典[
字符串,
任何]] = None,
custom_tracer
自定义追踪器: 任何 = None,
) -> GraphModule:
""
给定一个包含 M 个操作的图模型，
args_kwargs_m -> op_m -> output_m
并且为每个操作创建一组 N 个 qconfig，创建一个新的模型，
将`op_m`的每个子图转换为
.. 代码块::
|---------> op_m_n -> log_m_n
| /
args_kwargs_m ---------> op_m -> log_m_0
其中 op_m_n 是将 op_m 封装在子模块中并进行转换的
qconfig_n 及其内部图看起来像
.. 代码块::
args_m -------- op_m_prepared_with_qconfig_n -> out_m_n
/
kwargs_m ---
这对于测试多层不同量化很有用
一次通过模型。
未来 Pull Requests 的高级待办事项：
找出更好的命名输出结构的方法
返回一个结果数据结构而不是打印出来
添加示例到文档块
"文档"
如果
custom_tracer
自定义跟踪器 是 None:
追踪器 =
量化效果.
量化追踪器
[]
[]
否则:
追踪器 =
custom_tracer
自定义跟踪器
沉浸式翻译 =
PyTorch.fx.GraphModule(
模型,
追踪器.
跟踪(
模型))
这是必要的，以确保记录器 FQNs 被填充。
沉浸式翻译.
_节点名称到作用域_ =
追踪器.
节点名称到范围映射
# 类型：忽略[赋值]
运行示例输入传播，我们需要这个来调用 prepare_fx
单个子图
输出传播 =
输出属性(
沉浸式翻译)
输出属性.
传播(*
示例输入)
找到原始图中我们需要
考虑。
模块 =
字典(
沉浸式翻译.
命名模块(
删除重复项=
假的))
模式 =
_获取量化处理器的模式(
后端配置)
根节点获取器映射 = get_fusion_pattern_to_root_node_getter(
后端配置)
独立模块名称:
列表[
字符串] =
输入文本为空，请提供需要翻译的文本
独立模块类:
列表[
类型] =
输入文本为空，请提供需要翻译的文本
自定义模块类:
列表[
类型] =
输入文本为空，请提供需要翻译的文本
匹配项 = _find_matches(
沉浸式翻译.
图,
模块,
模式,
根节点获取映射,
独立模块名称,
独立模块类,
自定义模块类,
)
子图去重:
字典[
字符串,
列表[
节点]] =
获取去重子图(
匹配)
为每个子图生成节点到 qconfig
# TODO(未来 PR): 去重重复条目
节点名到 qconfig 的列表:
列表[
字典[
字符串, QConfigAny]] =
输入文本为空，请提供需要翻译的文本
为
qconfig 映射
在
qconfig 多映射.
Q 配置映射列表:
节点名称到 qconfig = _generate_node_name_to_qconfig(
沉浸式翻译,
模块,
沉浸式翻译.
图,
QConfig 映射,
追踪器.
节点名称到范围映射
)
节点名到 qconfig 的列表.
追加(
节点名称到 Q 配置映射)
对于模型中的每个区域，执行以下操作：
对于该区域的每个 qconfig，执行以下操作：
1. 创建一个模块包装区域的副本
2. 传递原始参数、原始关键字参数和预期输出到模块
3. 添加一个输出比较记录器并将其连接到比较
实际输出到预期输出
在模块上运行 `prepare_fx`
为 subgraph_idx, (
匹配名称,
节点数)
在
列举(
子图去重.
项目()
):
创建 n 个转换并记录的子图副本(
沉浸式翻译,
subgraph_idx,
匹配名称,
节点数,
qconfig 多映射.
Q 配置映射列表,
节点名到 qconfig 的列表,
自定义准备函数,
自定义准备参数, # type: ignore[arg-type]
)
返回
沉浸式翻译
# TODO(future PR): 我们应该重新思考所有 PNP API 的命名
def _准备_n_阴影添加日志记录器模型(
模型:
PyTorch.
神经网络.
模块,
示例输入:
任何,
QConfig 映射:
QConfig 映射,
后端配置:
后端配置,
) -> PyTorch.
神经网络.
模块:
r""
注意：此 API 不建议广泛使用，它仅
为需要从 `add_loggers` 迁移的客户提供
API.
这将创建一个模型，该模型为以下问题提供日志记录：
问题是：如果我们使用 `qconfig_mapping` 对 `model` 进行量化并输入
通过两种模型进行相同输入，记录比较结果。
对应的中间层。
问题通过单个模型解决。具体来说，我们将模型 `model` 划分为 N 个子图，为每个相关的子图创建一个副本。
将 `model` 划分为 N 个子图，为每个相关的子图创建一个副本。
将子图封装在模块中，并应用量化 API
将模块连接到日志记录器以测量比较
示例起始图：
x0 -> op0 -> x1 -> op1 -> x2
示例配置：将 op0 量化为 int8，对 op1 不做任何操作。
以下将创建的图：
.. 代码块::
x0_0 -> op0_0 -> x1_0 -> log -----> op1_0 -> x2_0 -> log
\ \ \ # noqa: W605
---> op0_1 -> x1_1 ----> clog -> op1_0 -> x2_1 ----> clog
其中 op0_0 是 op0，op0_1 是封装在子模块中的 op0 并进行了量化
op1_0 是 op1（在图中出现两次），log 是记录器，
并且 clog 是一个比较日志记录器。
"文档"
追踪器 =
量化效果.
量化追踪器([],
[]
沉浸式翻译 =
PyTorch.fx.GraphModule(
模型,
追踪器.
跟踪(
模型))
这是必要的，以确保记录器 FQNs 被填充。
沉浸式翻译.
_节点名称到作用域_ =
追踪器.
节点名称到范围映射
# 类型：忽略[赋值]
运行示例输入传播，我们需要这个来调用 prepare_fx
单个子图
输出传播 =
输出属性(
沉浸式翻译)
输出属性.
传播(*
示例输入)
找到原始图中我们需要
考虑。
模块 =
字典(
沉浸式翻译.
命名模块(
删除重复项=
假的))
模式 =
_获取量化处理器的模式(
后端配置)
根节点获取器映射 = get_fusion_pattern_to_root_node_getter(
后端配置)
独立模块名称:
列表[
字符串] =
输入文本为空，请提供需要翻译的文本
独立模块类:
列表[
类型] =
输入文本为空，请提供需要翻译的文本
自定义模块类:
列表[
类型] =
输入文本为空，请提供需要翻译的文本
匹配项 = _find_matches(
沉浸式翻译.
图,
模块,
模式,
根节点获取映射,
独立模块名称,
独立模块类,
自定义模块类,
)
子图去重:
字典[
字符串,
列表[
节点]] =
获取去重子图(
匹配)
为每个子图生成节点到 qconfig
节点名称到 qconfig = _generate_node_name_to_qconfig(
沉浸式翻译,
模块,
沉浸式翻译.
图,
QConfig 映射,
追踪器.
节点名称到范围映射
)
现在，将图变异为具有传播的 add_loggers 图
错误。
创建添加日志记录器的图(
沉浸式翻译,
子图去重,
QConfig 映射,
节点名称到 Q 配置映射)
返回
沉浸式翻译
# TODO(future PR): 我们应该重新思考所有 PNP API 的命名
def _n_shadows_compare_weights(
模型:
PyTorch.
神经网络.
模块,
示例输入:
任何,
QConfig 映射:
QConfig 映射,
后端配置:
后端配置,
) -> NS 结果类型:
""
注意：此 API 不建议广泛使用，它仅
为需要从 `add_loggers` 迁移的客户提供
API.
"文档"
qconfig 多映射 =
QConfig 多映射.
从列表中获取 QConfig 映射(
[QConfig 映射]
)
mp = 准备_n_shadows 模型(
模型,
示例输入,
qconfig 多映射, backend_config
)
# 通过模型传递输入是填充
# 观察具有实数值的权重观察者
mp(*示例输入)
mq = 转换_n_阴影模型(mp)
重量比较 =
提取权重比较(mq)
返回
重量比较
# TODO(future PR): 考虑将 API 签名与其他类似量化对齐
# 函数（enable_fake_quant 等）
[文档]def loggers_set_enabled(model: torch.nn.Module, enabled: bool) -> None:
"""
设置`model`的日志记录器的`enabled`设置
"""
for _, child in model.named_modules():
if isinstance(child, OutputLogger):
child.enabled = enabled
# TODO(future PR): 考虑将 API 签名与其他类似量化对齐
# 函数（enable_fake_quant 等）
[文档]def loggers_set_save_activations(
模型: torch.nn.Module,
save_activations: 布尔型，
)
-> None:
"""
设置`model`的日志记录器的`save_activations`设置
"""
for _name, child in model.named_modules():
if isinstance(child, OutputLogger):
child.save_activations = save_activations
[文档]def convert_n_shadows_model(
model: 图模块,
custom_convert_fn: 可选[可调用对象] = None,
custom_convert_kwargs: 可选[dict[str, Any]] = None,
) -> 图模块:
""
给定一个来自 `prepare_n_shadows_model` 的模型，运行 `convert_fx`
在每个阴影子模块上。
""
for node in model.graph.nodes:
# TODO(future PR): 考虑以更安全的方式匹配
# 节点名字符串匹配
if node.name.startswith(SHADOW_WRAPPER_NODE_NAME_PREFIX):
orig_mod = getattr(model, node.name)
if custom_convert_fn is None:
converted_mod = torch.ao.quantization.quantize_fx.convert_fx(orig_mod)
else:
如果 custom_convert_kwargs 为 None：
custom_convert_kwargs = {}
converted_mod = custom_convert_fn(orig_mod, **custom_convert_kwargs)
setattr(model, node.name, converted_mod)
返回模型
[文档]def extract_results_n_shadows_model(model: torch.nn.Module) -> NSResultsType:
"""
从 `model` 中提取日志结果。
"``"
results: NSResultsType = {}
_extract_logger_info_one_model(model, results, OutputLogger)
return results
[文档]def 打印比较和阴影模型(results: NSResultsType) -> None:
"""
打印提取的 `results` 的摘要。
"""
results_grouped = 按子图分组结果(results)
results_comparison = 创建结果比较(results_grouped)
print_n_shadows_summary(results_comparison)
