# mypy: 允许未类型化装饰器
# mypy: 允许未类型化定义
# 目前暂时跳过此文件的 RUF，我们可以重新启用
# 将仿射量化相关内容移动到 torchao 后
# 无需检查：RUF
""
本模块实现了用于收集统计信息的观察者
校准（PTQ）或训练（QAT）期间观察到的值。
""
导入
正则表达式
导入
警告
from abc 导入 ABCMeta,
抽象方法
from 集合
导入 OrderedDict
from functools 导入
部分信息
from 打字
导入
任何,
可选
导入
火炬
导入 torch.nn
是
神经网络
from torch.ao.quantization.utils 导入 (
计算 qmin_qmax,
检查最小最大值是否有效,
是否按通道,
是否为每张张量,
验证 qmin_qmax,
)
全部 = [
"default_affine_fixed_qparams_observer",
"默认调试观察者",
"默认动态量化观察者",
"默认固定 q 参数范围 0 到 1 观察者",
"默认固定 q 参数范围-1 到 1 观察者",
"默认浮点 q 参数观察者",
"默认浮点 Q 参数观察器 4 位",
"默认直方图观察者",
"默认观察者",
"默认通道权重观察者",
"默认占位符观察者",
"默认重用输入观察者",
"默认对称固定 Q 参数观察者",
"默认权重观察者",
get_observer_state_dict,
"加载观察者状态字典",
"每个通道权重观察者范围-127 到 127",
"weight_observer_range_neg_127_to_127",
"固定 Q 参数观察者",
"直方图观察者",
"最小最大观察者",
移动平均最小最大观察者,
每通道移动平均最小最大观察者,
无操作观察者,
观察者基类,
"按通道最大最小观察者",
"占位符观察者",
"录制观察者",
"重用输入观察者",
"统一量化观察者基",
“仿射量化观察者基”,
“粒度”,
"映射类型",
"沿轴",
"按块",
"按组",
"每行",
"每张张量",
"每个标记",
"PyTorch AOD 类型",
零点域,
获取区块大小,
]
类
_部分包装器:
def __init__(我, p):
我.p = p
我.
可调用参数 = {}
def __调用__(
我, *
参数, **
关键字):
# 调用 callable_args 中的每个参数并添加它们的部分，然后使用关键字运行
# 如果 arg_name 在关键字中，则跳过，这样就可以覆盖
for 变量名
在
我.callable_args:
如果
变量名
不
在 keywords:
关键词 = {**
关键词,
参数名称:
我.
可调用参数[
参数名称]()}
返回
我.p(*
参数, **
关键词)
def __repr__(我):
返回
我.p.__repr__() +
我.
可调用参数.__repr__()
def 带参数的(
我, **kwargs):
返回
_带参数的_(
我, **kwargs)
def 可调用参数(
我, **kwargs):
结果 =
_部分包装器(p=
我.p)
结果.
可调用参数 = {**
我.
可调用参数, **kwargs}
返回
结果
def _with_args(cls_or_self, **kwargs):
r“包装器，允许创建类工厂。
这在需要创建具有相同属性或行为的类时非常有用。
构造函数参数，但为不同的实例。可以与
_callable_args
示例::
>>> # xdoctest: +SKIP("未定义变量")
>>> Foo.with_args = classmethod(_with_args)
>>> foo_builder = Foo.with_args(a=3, b=4).with_args(answer=42)
>>> foo_instance1 = foo_builder()
>>> foo_instance2 = foo_builder()
>>> id(foo_instance1) == id(foo_instance2)
False
"文档"
r = _PartialWrapper(偏函数(cls_or_self, **kwargs))
返回 r
def _with_callable_args(cls_or_self, **kwargs):
r允许创建需要参数的类工厂的包装器
构造时调用。
当需要创建具有相同构造函数参数但不同实例的类时，这可能很有用。
这些参数应在构造时计算，可以与_with_args 一起使用。
可以在构造时计算，可以与_with_args 一起使用。
示例::
>>> # xdoctest: +SKIP("未定义变量")
>>> Foo.with_callable_args = classmethod(_with_callable_args)
>>> Foo.with_args = classmethod(_with_args)
>>> foo_builder = Foo.with_callable_args(cur_time=get_time_func).with_args(name="dan")
>>> foo_instance1 = foo_builder()
>>> # 等待 50 秒
>>> foo_instance2 = foo_builder()
>>> id(foo_instance1.creation_time) == id(foo_instance2.creation_time)
False
"文档"
r = _PartialWrapper(偏函数(cls_or_self))
返回 r.
带可调用参数(**kwargs)
ABC: 任何 = ABCMeta("ABC", (
对象,), {})
# 兼容 Python 2 和 3：
[文档]class ObserverBase(ABC, nn.Module):
r"""基础观察器模块。
任何观察器实现都应从此类派生。
混合观察者应遵循相同的 API。在正向中，它们将更新
被观察张量的统计信息。并且它们应提供
`calculate_qparams` 函数，该函数根据收集到的统计信息计算量化参数。
收集到的统计信息。
参数:
dtype：`quantize`节点所需的`dtype`参数，以实现
参考模型规范
is_dynamic: 表示观察者是否为动态量化的占位符的指示器
或静态量化
"""
def __init__(self, dtype, is_dynamic: bool = False):
super().__init__()
self.dtype = dtype
self.is_dynamic = is_dynamic
@abstractmethod
def forward(self, x):
pass
@abstractmethod
def calculate_qparams(self, **kwargs):
pass
with_args = classmethod(_with_args)
with_callable_args = classmethod(_with_callable_args)
类 UniformQuantizationObserverBase(
观察者基类):
r"""通用基类，用于所有使用均匀量化计算观察器的观察器"""
缩放和零点。
参数：
数据类型：`quantize`节点所需的`dtype`参数
参考模型规范
量化方案：要使用的量化方案。
reduce_range: 减少量化数据类型的范围 1 位。
有时需要这样做以避免指令溢出。
quant_min: 最小量化值。如未指定，将遵循 8 位设置。
quant_max: 最大量化值。如未指定，将遵循 8 位设置。
eps: float32 的 Epsilon 值，默认为`torch.finfo(torch.float32).eps`。
.. 警告::
attr:`dtype`只能取`torch.qint8`或`torch.quint8`。
或`torch.int8`或`torch.uint8`。
.. 警告::
attr:`qscheme`只能取以下选项之一：
- torch.per_tensor_affine
- torch.per_tensor_symmetric
- torch.per_channel_affine
- torch.per_channel_symmetric
"文档"
所有观察者类型的版本是共享的
#
版本 1/无
self
#
版本 2（仅基类，不包括子类缓冲区）
# self
# |--- eps : 张量
#
# 版本 3
# 仅对 HistogramObserver 有效，改变了未初始化的形状
# min_val 和 max_val 缓冲区从 torch.Size([0])变为 torch.Size([])
# 对于 PerChannelObservers，将缓冲区的名称从 min_vals 更改为 min_val，从 max_vals 更改为 max_val。
# 将 min_vals 和 max_vals 分别更改为 min_val 和 max_val。
_version = 3
eps: 火把.
张量
def __init__(
我,
数据类型=
火把.quint8,
q 方案=
火把.
每张张量仿射,
范围缩减=
错误,
量化最小值=
无,
量化最大值=
无,
工厂参数=
无,
eps=火把.finfo(
火把.float32).eps,
是否动态=
错误,
**kwargs,
) -> 无:
工厂参数 =
火把.
神经网络.
工厂参数(
工厂参数)
超级().__init__(
数据类型=
数据类型,
是否动态=
动态的, **kwargs)
我.
虚拟方案 =
虚拟方案
如果
范围缩减:
警告.
警告(
请使用 quant_min 和 quant_max 来指定观察者的范围。\
reduce_range 将在 PyTorch 的未来版本中弃用。
)
我.reduce_range =
缩小范围
我.
注册缓冲区(
eps,
火把.
张量
[eps
]， **
工厂参数))
断言
我.
虚拟方案
在 (
火把.
每张张量仿射,
火把.
每张张量对称,
火把.
每通道仿射变换,
火把.
每通道对称,
火把.
每通道仿射浮点量化参数,
), "默认观察者仅适用于 per_tensor_affine、"\
per_tensor_symmetric、per_channel_affine、\
per_channel_symmetric 以及 per_channel_float_qparams 量化方案
_ALLOWED_DTYPES = (
火把.qint8,
火把.quint8,
火把.quint4x2,
火把.qint32,
火把.int8,
火把.uint8,
火把.int16,
火把.int32,
火把.float8_e5m2,
火把.float8_e4m3fn,
火把.uint16,
)
断言 (
我.dtype
在
允许的数据类型
), f默认观察者仅适用于{
允许的数据类型}
数据类型
我.
已自定义 qrange = (
量化最小值
是
不
无)
并且 (
量化最大值
是
不
无)
如果
我.
已自定义 qrange:
验证 qmin_qmax(
量化最小值,
量化最大值)
我.
量化最小值,
我.
量化最大值 =
计算 qmin_qmax(
量化最小值,
量化最大值,
我.
自定义 q 范围,
我.
数据类型,
我.
范围缩减,
)
def 从状态字典加载(
我,
状态字典,
前缀,
本地元数据,
严格的,
缺少键,
预期之外的键,
错误信息,
):
版本 =
本地元数据.
获取(
版本,
无)
如果
版本
是
无
或者
版本 == 1:
# eps 已在版本 2 中移动到缓冲区
eps = 火把.
张量
[
火把.finfo(
火把.float32).eps])
状态字典[
前缀 +
eps] = eps
超级().
从状态字典加载(
状态字典,
前缀,
本地元数据,
严格的,
缺少键,
预期之外的键,
错误消息,
)
@torch.算子.
导出
def _validate_qmin_qmax(我,
量化最小值:
整数,
量化最大值:
整数) ->
无:
r验证用户指定的量化范围是否正确初始化
并且在观察器数据类型的支持范围内。
为了适应相对于现有 torch.qint8 的更低位量化
torch.quint8 数据类型，用户可以选择通过传递动态量化范围来使用动态量化
在一个初始 qmin 和 qmax 值的元组中。一个用例是这些自定义的 qmin 和 qmax 值用于计算激进低比特伪量化的静态尺度估计和零点。这些估计与通过反向传播学习到的参数进行比较。
这些值用于计算激进低比特伪量化的静态尺度估计和零点。这些估计与通过反向传播学习到的参数进行比较。
通过反向传播进行尺度估计和零点相关的文献如下：
通过反向传播进行尺度估计和零点相关的文献如下：
学习型步长量化：https://openreview.net/pdf?id=rkgO66VKDS
训练量化阈值：https://arxiv.org/pdf/1903.08066.pdf
"文档"
变量名前缀为 "initial"，因为它们的值（qmin 和 qmax）可能会根据量化范围是否减小以及观察者使用的数据类型（有符号/无符号）进行调整。
基于 whether quantization range is reduced and the datatype (signed/unsigned) used by the observer.
断言 (
量化最小值
≤ 0
≤
量化最大值
), "使用的量化范围必须包含 0。"
断言 (
量化最小值 <
量化最大值
), "用户指定的量化范围中，qmin 必须严格小于 qmax。"
@torch.算子.
导出
def _计算_qparams(
我, min_val:
火把.
张量, max_val:
火把.
张量
) -> 元组[
火把.
张量,
火把.
张量
]：
r计算量化参数，给定最小值和最大值
处理值张量。适用于每个张量和每个通道的情况
参数：
min_val: 每个通道的最小值
max_val: 每个通道的最大值
返回：
尺度张量，形状为 (#channels,)
零点张量，形状为 (#channels,)
"文档"
与 utils.py 中的'determine_qparams'功能等效。观察者必须是 torchscriptable 的，并且 qscheme
据我所知，不允许作为参数传递给 torchscript 函数。这使得重构观察者
使用这个工具非常痛苦且非常恶心。目前我选择只是复制这段代码
看起来不太可能改变（上次更新超过 1 年），当 torchscript 完全弃用时我们可以重构。
TODO(jakeszwe, jerryzh168)
如果
不
检查_min_max_valid(min_val, max_val):
返回
火把.
张量
[1.0
]，
设备=min_val.
设备.
类型),
火把.
张量(
[0]，
设备=min_val.
设备.
类型
)
量化最小值,
量化最大值 =
我.
量化最小值,
我.
量化最大值
最小值负 =
火把.
最小(min_val,
火把.
与...相同形状的零(min_val))
最大值正 =
火把.
最大值(max_val,
火把.
与...相同形状的零(max_val))
设备 =
最小值负.
设备
缩放 =
火把.
一(
最小值负.
尺寸(),
数据类型=
火把.float32,
设备=
设备)
零点 =
火把.
零值(
最小值负.
尺寸(),
数据类型=
火把.int64,
设备=
设备)
如果 (
我.
虚拟方案 ==
火把.
每张张量对称
或者
我.
虚拟方案 ==
火把.
单通道对称
):
最大值正 =
火把.
最大值(-
最小值负,
最大值正)
缩放 =
最大值位置 / (float(
量化最大值 -
量化最小值) / 2)
缩放 =
火把.
最大值(
比例,
我.eps)
如果
我.dtype
在 [
火把.quint8,
火把.uint8
]：
如果
我.
是否有自定义量化范围:
当使用自定义量化范围时，选择范围的下舍中点。
零点 =
零点.
新满(
零点.
尺寸(), (
量化最小值 +
量化最大值) // 2
)
否则:
零点 =
零点.
新满(
零点.
尺寸(), 128)
elif 我.dtype
在 [
火把.uint16
]：
零点 =
零点.
新满(
零点.
尺寸(), 2**15)
elif 我.
虚拟方案 ==
火把.
每通道仿射浮点量化参数:
缩放 = (max_val - min_val) / float(
量化最大值 -
量化最小值)
缩放 =
火把.
哪里(
缩放 >
我.eps,
比例,
火把.
喜欢的(
比例))
我们使用量化函数
xq = Round(Xf * inv_scale + zero_point),
将 zero_point 设置为(-1 * min * inv_scale)时，我们得到
# Xq = (Xf - 最小值) * inv_scale
零点 = -1 *
最小值 /
缩放
否则:
缩放 = (
最大值正 -
最小值负) / float(
量化最大值 -
量化最小值)
缩放 =
火把.
最大值(
比例,
我.eps)
零点 =
量化最小值 -
火把.
四舍五入(
最小值负 /
比例).
到(
火把.
整数)
零点 =
火把.
卡钳(
零点,
量化最小值,
量化最大值)
# 对于标量值，将它们转换为大小为 1 的张量以保持默认值与 FakeQuantize 中的默认值一致。
# TODO：在添加 JIT 支持后切换到 scale.item()
如果
长度(
比例.shape) == 0:
# TODO：在添加 JIT 支持后切换到 scale.item()
缩放 =
火把.
张量
[float(
比例
)]，
数据类型=
比例.
数据类型,
设备=
设备)
如果
长度(
零点.shape) == 0:
# TODO: 在添加 JIT 支持后切换到 zero_point.item()
零点 =
火把.
张量(
[整数(
零点
)]，
数据类型=
零点.
数据类型,
设备=
设备
)
如果
我.
虚拟方案 ==
火把.
每通道仿射浮点量化参数:
零点 =
火把.
张量(
[float(零点
)]，
数据类型=
零点.
数据类型,
设备=
设备
)
返回
比例,
零点
@torch.算子.
导出
def 重置最小/最大值(
我):
raise 不支持的操作异常(
"无法在给定的观察器中重置最小/最大值。")
# 最初，这个类被命名为 `_ObserverBase`。保留旧名称以
# 保持向后兼容性。
# TODO(在 v1.13 之后): 删除此内容
_ObserverBase = 均匀量化 ObserverBase
[文档]
类 MinMaxObserver(
均匀量化观察器基):
r观测模块，用于基于量化参数的计算
运行最小和最大值。
此观察者使用张量最小/最大统计信息来计算量化
参数。该模块记录输入张量的运行最小值和最大值，
并使用此统计信息来计算量化参数。
参数：
dtype：`quantize`节点的 dtype 参数，用于实现参考模型规范。
。
量化方案：要使用的量化方案
reduce_range: 通过 1 位减少量化数据类型的范围
quant_min: 量化最小值。如未指定，将遵循 8 位设置。
quant_max: 量化最大值。如未指定，将遵循 8 位设置。
eps: float32 的 Epsilon 值，默认为`torch.finfo(torch.float32).eps`。
给定运行中的最小/最大值 :math:`x_\text{min}` 和 :math:`x_\text{max}`，
比例 :math:`s` 和零点 :math:`z` 的计算方法如下：
运行中的最小/最大值 :math:`x_\text{min/max}` 的计算方法如下：
.. math::
\begin{array}{ll}
\( x_\text{min} = \begin{cases}
\min(X) & \text{if~}x_\text{min} = \text{None} \\
\min\left(x_\text{min}, \min(X)\right) & \text{otherwise}
\end{cases}\\
\( x_\text{max} = \begin{cases} \)
\(\max(X)\) & \text{如果} x_\text{max} = \text{None} \)
\(\max\left(x_\text{max}, \max(X)\right) & \text{否则} \)
\end{cases}\\
\end{array}
其中 :math:`X` 是观测张量。
然后计算尺度 :math:`s` 和零点 :math:`z` 如下：
.. math::
\begin{aligned}
如果对称：\\
&s = 2 \max(|x_\text{min}|, x_\text{max}) /
\left( Q_\text{max} - Q_\text{min} \right) \\
&z = \begin{cases}
0 & \text{如果数据类型是 qint8} \\
128 & \text{否则}
\end{cases}\\
\text{否则：}&\\
(x_max - x_min) /
(Q_max - Q_min) \\
&z = Q_min - round(x_min / s)
\end{aligned}
数学中的最小和最大 Q 值
量化数据类型的最大值。
.. 警告:: :attr:`dtype` 只能取 ``torch.qint8`` 或 ``torch.quint8``。
.. 注意:: 如果运行中的最小值等于运行中的最大值，则比例和零点设置为 1.0 和 0。
并且工厂参数设置为 1.0 和 0。
"文档"
min_val: 火把.
张量
max_val: 火把.
张量
def __init__(
我,
数据类型=
火把.quint8,
q 方案=
火把.
每张张量仿射,
范围缩减=
错误,
量化最小值=
无,
量化最大值=
无,
factory_kwargs=无,
eps=火把.finfo(
火把.float32).eps,
是动态的=
错误,
**kwargs,
) -> 无:
如果
不 is_per_tensor(
q 方案):
raise 不支持的操作异常(
"MinMaxObserver 的 qscheme 仅支持 torch.per_tensor_symmetric"\
和 torch.per_tensor_affine."
)
# TODO: MinMaxObserver 本身不支持动态量化，
如果它继承自 MovingAverageObserver，并且平均常数为 1，则
支持动态量化，我们可能需要在这里进行更好的错误检查
对于 x86 量化内核，我们需要确保 vpmaddubsw 指令不会溢出。我们允许使用 reduce_range 参数来
确保 vpmaddubsw 指令不会溢出。我们允许使用 reduce_range 参数来
观察者将量化范围减少到（0，127）或（-64，63）。
更多详情请参阅 aten/src/ATen/native/quantized/cpu/qconv.cpp。
对于非 x86 后端来说，这不是一个最佳选择，因为它会丢失一些精度。
这会丢失激活的精度。
超级().__init__(
数据类型=
数据类型,
q 方案=
q 方案,
范围缩减=
范围缩减,
量化最小值=
量化最小值,
量化最大值=
量化最大值,
工厂参数=
工厂参数,
eps=eps,
动态的=
动态的,
**kwargs,
)
工厂参数 =
火把.
神经网络.
工厂参数(
工厂参数)
我.
注册缓冲区(
"最小值",
火把.
张量(float(
"无穷大"), **
工厂参数))
我.
注册缓冲区(
"最大值",
火把.
张量(float(
-无穷大), **
工厂参数))
如果 (
我.
虚拟方案 ==
火把.
每张张量对称
并且
我.
减少范围
并且
我.dtype ==
火把.
无符号 8 位整数
):
raise 不支持的操作异常(
无法减少对称范围的值\
量化为 quint8
)
[文档] def forward(self, x_orig):
记录 `x` 的运行最小值和最大值。
如果 x_orig.numel() 等于 0:
返回 x_orig
x = x_orig.detach() # 避免保留 autograd tape
x = x.to(self.min_val.dtype)
min_val_cur, max_val_cur = torch.aminmax(x)
min_val = torch.min(min_val_cur, self.min_val)
max_val = torch.max(max_val_cur, self.max_val)
self.min_val.copy_(min_val)
self.max_val.copy_(max_val)
return x_orig
[文档] @torch.jit.export
def calculate_qparams(self):
计算量化参数。
返回 self._calculate_qparams(self.min_val, self.max_val)
@torch.算子.
导出
def 额外表示(
我):
返回 f
min_val={
我.min_val}, max_val={
我.max_val}"
[文档] @torch.jit.export
def 重置_min_max_vals(self):
重置最小/最大值。
self.min_val.copy_(torch.tensor(float("inf")))
self.max_val.copy_(torch.tensor(float("-inf")))
[文档]
类
移动平均最小最大观察者(
MinMax 观察器):
r"""基于计算量化参数的观察器模块，"""
最小值和最大值的移动平均值。
此观察者根据输入张量的最小值和最大值的平均值计算量化参数。
该模块记录输入张量的平均最小值和最大值，并使用这些值。
记录输入张量的平均最小值和最大值，并使用这些值。
计算量化参数的统计量。
参数：
averaging_constant：min/max 的平均常数。
dtype：`quantize`节点所需的 dtype 参数，用于实现
参考模型规范。
量化方案：要使用的量化方案
reduce_range: 通过 1 位减少量化数据类型的范围
quant_min: 量化最小值。如未指定，将遵循 8 位设置。
quant_max: 量化最大值。如未指定，将遵循 8 位设置。
eps: float32 的 Epsilon 值，默认为`torch.finfo(torch.float32).eps`。
移动平均的最小/最大值计算如下
.. math::
\begin{array}{ll}
x_\text{min} = \begin{cases}
最小值(X) & 如果~x_min = None
(1 - c) x_min + c 最小值(X) & 否则
\end{cases}\\
x_max = \begin{cases}
max(X) & 若 x_max = None
(1 - c) x_max + c max(X) & 否则
结束情况
结束数组
其中 :math:`x_\text{min/max}` 是运行平均最小/最大值，:math:`X` 是
输入张量，而 :math:`c` 是 ``平均常数``。
然后计算缩放和零点，方法与
class:`~torch.ao.quantization.observer.MinMaxObserver` 中所述。
.. note:: 仅适用于 ``torch.per_tensor_affine`` 量化方案。
.. note:: 如果运行最小值等于运行最大值，则比例因子和零点设置为 1.0 和 0。
和零点设置为 1.0 和 0。
"文档"
def __init__(
我,
平均常数=0.01,
数据类型=
火把.quint8,
q 方案=
火把.
每张张量仿射,
范围缩减=
错误,
量化最小值=
无,
量化最大值=
无,
eps=火把.finfo(
火把.float32).eps,
动态的=
错误,
**kwargs,
) -> 无:
如果
不
每个张量(
q 方案):
raise 不支持的操作异常(
f"MovingAverageMinMaxObserver 的 qscheme 只支持\
torch.per_tensor_symmetric 和 torch.per_tensor_affine。\
但是得到了：{
q 方案}"
)
我.
平均常数 =
平均常数
如果
是否动态
并且
我.
平均常数 != 1:
raise 不支持的操作异常(
"MovingAverageMinMaxObserver 不支持动态量化，对于"
f"平均常数"{
我.
平均常数}"
)
超级().__init__(
数据类型=
数据类型,
q 方案=
q 方案,
范围缩减=
范围缩减,
量化最小值=
量化最小值,
量化最大值=
量化最大值,
eps=eps,
动态的=
动态的,
**kwargs,
)
def 前向(
我,
原始 x):
如果
原始 x.
元素数量() == 0:
返回
原始 x
x = 原始 x.detach()
避免保留自动微分带
x = x.到(
我.min_val.
数据类型)
最小值 =
我.
最小值
最大值 =
我.
最大值
如果
最小值 == float(
"无穷大")
并且
最大值 == float(
-无穷大):
min_val, 最大值 =
火把.
阿敏最大(x)
否则:
当前最小值,
当前最大值 =
火把.
阿敏最大(x)
最小值 =
最小值 +
我.
平均常数 * (
当前最小值 - min_val)
最大值 =
最大值 +
我.
平均常数 * (
当前最大值 - max_val)
我.min_val.
复制_(min_val)
我.max_val.
复制_(max_val)
返回
原始 x 坐标
[文档]
类
每通道最大最小值观察者(
均匀量化观察器基):
r观测模块，用于基于量化参数的计算
运行每个通道的最小和最大值。
此观察者使用张量最小/最大统计信息来计算每个通道
量化参数。该模块记录运行中的最小值和最大值
输入张量，并使用此统计信息进行量化
参数
参数：
ch_axis: 通道轴
`dtype` 参数到 `quantize` 节点的实现所需的
参考模型规范。
qscheme：要使用的量化方案
reduce_range：通过 1 位减少量化数据类型的范围
quant_min: 最小量化值。如未指定，将遵循 8 位设置。
quant_max: 最大量化值。如未指定，将遵循 8 位设置。
eps: float32 的 Epsilon 值，默认为`torch.finfo(torch.float32).eps`。
量化参数的计算方式与以下相同
`:class:`~torch.ao.quantization.observer.MinMaxObserver`，与差异
运行中的最小/最大值按通道存储。
刻度和零点因此按通道计算。
.. 注意:: 如果运行最小值等于运行最大值，则比例
and zero_points 被设置为 1.0 和 0.
"文档"
min_val: 火把.
张量
max_val: 火把.
张量
def __init__(
我,
坐标轴=0,
数据类型=
火把.quint8,
q 方案=
火把.
每通道仿射变换,
范围缩减=
错误,
量化最小值=
无,
量化最大值=
无,
工厂参数=
无,
eps=火把.finfo(
火把.float32).eps,
是否动态=
错误,
**kwargs,
) -> 无:
如果
不
是否按通道(
q 方案):
raise 不支持的操作异常(
"PerChannelMinMaxObserver 的 qscheme 仅支持\
torch.per_channel_symmetric、torch.per_channel_affine 和 torch.per_channel_affine_float_qparams."
)
如果
动态的:
raise 不支持的操作异常(
"PerChannelMinMaxObserver 不支持动态量化"
)
超级().__init__(
数据类型=
数据类型,
q 方案=
q 方案,
范围缩减=
范围缩减,
量化最小值=
量化最小值,
量化最大值=
量化最大值,
工厂参数=
工厂参数,
eps=eps,
动态的=
动态的,
**kwargs,
)
工厂参数 =
火把.
神经网络.
工厂参数(
工厂参数)
我.
ch 轴 =
ch 轴
我.
注册缓冲区(
"最小值",
火把.
张量([], **
工厂参数))
我.
注册缓冲区(
"最大值",
火把.
张量([], **
工厂参数))
如果 (
我.
虚拟方案 ==
火把.
单通道对称
并且
我.
减少范围
并且
我.dtype ==
火把.quint8
):
raise 不支持的操作异常(
"无法为 quint8 的对称量化减少范围"
)
def 前向(
我, x_orig):
返回
我._forward(
原始 x)
def 前向(
我,
原始 x):
如果
原始 x.
元素数量() == 0:
返回 x_orig
x = x_orig.detach() 避免保留自动微分带
最小值 =
我.
最小值
最大值 =
我.
最大值
x 维度 = x.
尺寸()
新轴列表 = [i for i
在
范围(
长度(
x 维度))] # noqa: C416
新轴列表[
我.
坐标轴] = 0
new_axis_list[0] = 我.
ch 轴
y = x.排列(new_axis_list)
# 需要匹配 min/max 的数据类型，因为缓冲区的更新
# 是就地完成的，类型需要匹配以进行比较
y = y.到(
我.min_val.
数据类型)
y = 火把.
展平(y, start_dim=1)
如果 min_val.
元素数量() == 0
或者 max_val.
元素数量() == 0:
min_val, 最大值 =
火把.
阿敏最大(y,
暗淡=1)
否则:
当前最小值,
当前最大值 =
火把.
阿敏最大(y,
暗淡=1)
最小值 =
火把.
最小(min_val_cur, min_val)
max_val = 火把.
最大值(max_val_cur, max_val)
我.min_val.
调整大小(min_val.shape)
我.max_val.
调整大小(max_val.shape)
我.min_val.
复制_(min_val)
我.max_val.
复制_(max_val)
返回
x 原始
@torch.算子.
导出
def 计算 q 参数(
我):
返回
我.
_计算 q 参数(
我.min_val,
我.max_val)
def 额外表示(
我):
返回 f
"最小值="{
我.min_val}
, 最大值={
我.max_val}"
def 从状态字典加载(
我,
状态字典:
字典[
字符串,
任何
]，
前缀:
字符串,
本地元数据:
字典[
字符串,
火把.
张量
]，
严格的:
布尔,
缺少键:
列表[
字符串
]，
预期之外的键:
列表[
字符串
]，
错误消息:
列表[
字符串
]，
):
版本 =
本地元数据.
获取(
版本,
无)
如果
版本
是
不
无
并且
版本 < 3:
本地状态 = [
最小值,
最大值]
预期最小名称 = "min_vals"
expected_max_name = "max_vals"
否则:
local_state = ["min 值",
"max 值"]
预期最小名称 =
"min 值"
预期最大名称 =
"最大值"
for 名称
在
本地状态:
key = 前缀 +
名称
如果 key
在
状态字典:
val = 状态字典[
键]
# 自定义处理以允许加载 min_val 或 max_val
将大小为 N 的值放入大小为 0 的未初始化缓冲区中。
在此处调整缓冲区大小，并将值复制进去。
父类的默认 state_dict 加载代码。
如果
名称 == expected_min_name:
我.min_val.
调整大小(val.shape)
elif 名称 ==
预期最大名称:
我.max_val.
调整大小(val.shape)
否则:
警告.
警告(
f"观察者从状态字典加载时得到意外的名称"{
名称}"
)
# 对于 torchscript 模块，我们需要在这里更新属性，因为我们没有调用
# 定义在 module.py 中的`_load_from_state_dict`函数
如果
火把.
算子.
是否正在脚本化():
如果
名称 ==
预期最小名称:
我.min_val.
复制_(val)
elif 名称 ==
预期最大名称:
我.max_val.
复制_(val)
否则:
警告.
警告(
f"观察者从状态字典加载时得到意外的名称"{
名称}"
)
elif 严格的:
缺少键.append(
键)
如果
不
火把.
算子.
是否正在脚本化():
超级().
从状态字典加载(
状态字典,
前缀,
本地元数据,
错误,
缺少键,
预期之外的键,
错误消息,
)
def _从状态字典加载脚本(
我,
状态字典:
字典[
字符串,
任何
]，
前缀:
字符串,
本地元数据:
字典[
字符串,
火把.
张量
]，
严格的:
布尔,
缺少键:
列表[
字符串
]，
预期之外的键:
列表[
字符串
]，
错误消息:
列表[
字符串
]，
):
我.
从状态字典加载(
状态字典,
前缀,
本地元数据,
严格的,
缺少键,
预期之外的键,
错误消息,
)
[文档] @torch.jit.export
def 重置_min_max_vals(self):
"""重置最小/最大值。"""
# 这曾经是 torch.ones，但不起作用因为
# JIT 编译器可以通过公共子表达式消除来优化它
# 在这种情况下，min_val 和 max_val 都指向同一个张量。
self.min_val = torch.rand(
0,
)
self.max_val = torch.rand(
0,
)
[文档]
类
每通道最小最大值观察器(
每通道最小最大值观察器):
r"""基于运行时每通道的最小值和最大值计算量化参数的观察器模块。"""
运行时每通道的最小值和最大值。
此观察者使用张量最小/最大统计信息来计算每个通道
量化参数。该模块记录运行中的最小值和最大值
输入张量，并使用此统计信息进行量化
参数
参数：
平均常数：min/max 的平均常数。
ch_axis：通道轴。
dtype：量化数据类型。
qscheme：要使用的量化方案。
reduce_range: 减少量化数据类型的范围 1 位
quant_min: 最小量化值。如未指定，将遵循 8 位设置。
quant_max: 最大量化值。如未指定，将遵循 8 位设置。
eps: float32 的 Epsilon 值，默认为`torch.finfo(torch.float32).eps`。
量化参数的计算方式与之前相同
`:class:`~torch.ao.quantization.observer.MovingAverageMinMaxObserver`，与
差异在于运行中的最小/最大值是按通道存储的。
刻度和零点因此按通道计算。
.. 注意:: 如果运行中的最小值等于运行中的最大值，则比例和零点设置为 1.0 和 0。
并且设置为 1.0 和 0。
"文档"
def __init__(
我,
平滑常数=0.01,
坐标轴=0,
数据类型=
火把.quint8,
q 方案=
火把.
每通道仿射变换,
范围缩减=
错误,
量化最小值=
无,
量化最大值=
无,
eps=火把.finfo(
火把.float32).eps,
是动态的=
错误,
**kwargs,
) -> 无:
如果
不
是否按通道(
q 方案):
raise 不支持的操作异常(
"移动平均每通道最大最小观察器的 qscheme 仅支持"\
torch.per_channel_symmetric、torch.per_channel_affine 以及 torch.per_channel_affine_float_qparams。
)
如果
是动态的:
raise 不支持的操作异常(
MovingAveragePerChannelMinMaxObserver 不支持动态量化
)
超级().__init__(
坐标轴=
坐标轴,
数据类型=
数据类型,
q 方案=
q 方案,
范围缩减=
范围缩减,
量化最小值=
量化最小值,
量化最大值=
量化最大值,
eps=eps,
动态的=
动态的,
**kwargs,
)
我.
平均常数 =
平均常数
def 前向(
我,
原始 x):
如果
原始 x.
元素数量() == 0:
返回
原始 x
x = 原始 x.detach()
避免保留自动求导带
x = x.到(
我.min_val.
数据类型)
最小值 =
我.
最小值
最大值 =
我.
最大值
x 维度 = x.
尺寸()
新的轴列表 = [i for i
在
范围(
长度(
x 维度))]
# 无需注意：C416
新轴列表[
我.
坐标轴] = 0
新轴列表[0] =
我.
ch 轴
y = x.排列(
新轴列表)
y = 火把.
展平(y, start_dim=1)
如果 min_val.
元素数量() == 0
或者 max_val.
元素数量() == 0:
min_val, 最大值 =
火把.
阿敏最大(y,
暗淡=1)
否则:
当前最小值,
当前最大值 =
火把.
阿敏最大(y,
暗淡=1)
最小值 =
最小值 +
我.
平均常数 * (
当前最小值 - min_val)
最大值 =
最大值 +
我.
平均常数 * (
当前最大值 - max_val)
我.min_val.
调整大小(min_val.shape)
我.max_val.
调整大小(max_val.shape)
我.min_val.
复制_(min_val)
我.max_val.
复制_(max_val)
返回
原始
[文档]
类
直方图观察者(
均匀量化观察者基类):
r""
该模块记录张量值的运行直方图，
最小/最大值。`calculate_qparams`将计算缩放和零点。
参数：
箱数：用于直方图的箱数。
dtype：`quantize`节点所需的 dtype 参数，用于实现。
参考模型规范。
量化方案，用于指定要使用的量化方案
reduce_range: 通过 1 位减少量化数据类型的范围
eps: float32 的 Epsilon 值，默认为`torch.finfo(torch.float32).eps`。
量化和零点的计算方法如下：
1. 创建输入数据的直方图。
直方图是连续计算的，每个 bin 的范围会随着观察到的每个新张量而变化。
3. 在直方图中搜索分布，以找到最优的 min/max 值。
4. 在直方图中搜索分布，以找到最优的 min/max 值。
查找最小/最大值确保相对于浮点模型的量化误差最小化。
3. 以与 :class:`~torch.ao.quantization.MinMaxObserver` 相同的方式计算缩放和零点。
3. 以与 :class:`~torch.ao.quantization.MinMaxObserver` 相同的方式计算缩放和零点。
3. 以与 :class:`~torch.ao.quantization.MinMaxObserver` 相同的方式计算缩放和零点。
"文档"
直方图:
火把.
张量
min_val: 火把.
张量
max_val: 火把.
张量
def __init__(
我,
箱数:
整型 = 2048,
数据类型:
火把.dtype =
火把.quint8,
q 方案=
火把.
每张张量仿射,
范围缩减=
错误,
量化最小值=
无,
量化最大值=
无,
工厂参数=
无,
eps=火把.finfo(
火把.float32).eps,
是动态的=
错误,
**kwargs,
) -> 无:
如果
不
是否为张量(
q 方案):
raise 不支持的操作异常(
"HistogramObserver 的 qscheme 仅支持 torch.per_tensor_symmetric"\
和 torch.per_tensor_affine。
)
如果
是动态的:
raise 不支持的操作异常(
"HistogramObserver 不支持动态量化"
)
bins：用于直方图计算的 bins 数量。
超级().__init__(
数据类型=
数据类型,
q 方案=
q 方案,
范围缩减=
范围缩减,
量化最小值=
量化最小值,
量化最大值=
量化最大值,
工厂参数=
工厂参数,
eps=eps,
是动态的=
是动态的,
**kwargs,
)
工厂参数 =
火把.
神经网络.
工厂参数(
工厂参数)
我.
箱子 =
箱子
我.
注册缓冲区(
直方图,
火把.
零值(
我.
箱数, **
工厂参数))
我.
注册缓冲区(
"最小值",
火把.
张量(float(
"无穷大"), **
工厂参数))
我.
注册缓冲区(
"最大值",
火把.
张量(float(
-无穷大), **factory_kwargs))
我.dst_nbins = 2 **
火把.iinfo(
我.
数据类型).
比特
我.
上采样率 = (
16 # 用于在放大直方图时减少量化误差
)
def 获取规范(
我, delta_begin:
火把.
张量, delta_end:
火把.
张量,
密度:
火把.
张量
) -> 火把.
张量:
r""
计算均匀分布在...的值的范数
delta_begin 和 delta_end.
目前仅支持 L2 范数。
范数 = 密度 * (从 begin 到 end 的 x^2 积分)
= 密度 * (end^3 - begin^3) / 3
"文档"
范数 = (
Δ端 *
Δ端 *
Δ端 -
Δ始 * delta_begin * delta_begin
) / 3
返回
密度 *
规范
def 计算量化误差(
我,
下一个起始 bin:
整数,
下一个结束 bin:
整数):
r""
如果我们使用 start_bin 到 end_bin 作为量化误差的计算范围，则计算量化误差
最小值和最大值用于进行量化。
"文档"
离散化宽度 = (
我.max_val.
项目() -
我.min_val.
项目()) /
我.
箱子
目标箱宽度 =
箱宽度 * (
下一个结束箱 -
下一个起始二进制 + 1) /
我.
目标二进制数
如果
目标二进制宽度 == 0.0:
返回 0.0
源二进制 =
火把.arange(
我.
箱数,
设备=
我.
直方图.
设备)
# 起始于第一个 dst_bin 起始处的距离
# 到 src_bin 起始处的结束
源_bin 开始 = (
源_bin -
下一个开始_bin) *
bin 宽度
src_bin_end = src_bin_begin + bin_width
# 哪些 dst_bins 包含 src_bin 的起始和结束？
dst_bin_of_begin = 火把.
卡钳(
火把.
除法(src_bin_begin, dst_bin_width,
四舍五入模式=
楼板),
0,
我.
目标 n 个 bin - 1,
)
目标起始中心 bin = (
目标起始 bin + 0.5) *
目标 bin 宽度
目标端 dst_bin_of_end =
火把.
卡钳(
火把.
除法(
源端 src_bin_end,
目标端宽度 dst_bin_width,
四舍五入模式=
地板 floor),
0,
我.dst_nbins - 1,
)
密度 =
我.
直方图 /
窗口宽度
规范 =
火把.
零值(
我.
箱数,
设备=
我.
直方图.
设备)
delta_begin = src_bin_begin - dst_bin_of_begin_center
delta_end = 目标二进制宽度 / 2
标准化 +=
我.
_获取标准化(
起始增量,
火把.
一(
我.
箱数,
设备=
我.
直方图.
设备) * delta_end,
密度,
)
norm += (dst_bin_of_end - dst_bin_of_begin - 1) * 我.
获取规范(
火把.
张量(-
目标箱宽度 / 2),
火把.
张量(
目标箱宽度 / 2),
密度
)
目标端中心 dst_bin =
目标端 dst_bin *
目标二进制宽度 +
目标二进制宽度 / 2
起始增量 = -
目标二进制宽度 / 2
delta_end = src_bin_end - dst_bin_of_end_center
norm += 我.
获取规范(
火把.
张量(
起始差值),
结束差值,
密度)
返回
归一化.
总和().
项目()
def 非线性参数搜索(
我) ->
元组[
火把.
张量,
火把.
张量
]：
r非线性参数搜索。
对于选择最小/最大值的 L2 误差最小化的近似。
通过选择新的最小/最大值，我们过滤掉输入分布中的异常值。
这遵循了 NormMinimization::NonlinearQuantizationParamsSearch 的实现。
caffe2/量化/server/规范化最小化.cc
"文档"
断言
我.
直方图.
尺寸()[0] ==
我.
箱数,
"bins 不匹配"
窗口宽度 = (
我.
最大值 -
我.min_val) /
我.
箱子
累计和
总计 =
火把.
总和(
我.
直方图).
项目()
cSum = 火把.
累加和(
我.
直方图,
暗淡=0)
步长 =
0.00001
# 粒度
alpha = 0.0 # 下限
测试版 = 1.0
上限
开始区间 = 0
结束区间 =
我.
箱子 - 1
最小规范 = float(
"无穷大")
当 alpha <
测试版:
# 找到下一步
下一个字母 = alpha +
步长
下一个 beta = beta -
步长
# 在分位数界限之间找到左右分箱
l = 开始分箱
r = 结束分箱
当 l < end_bin
并且 cSum[l] < next_alpha *
总计:
l = l + 1
当 r > start_bin
并且 cSum[r] > next_beta *
总计:
r = r - 1
#决定下一步
next_start_bin = 开始箱子
下一个结束箱子 =
结束箱子
如果 (l -
开始箱子) > (
结束区 - r):
# 移动起始区
下一个起始区 = l
alpha = 下一个字母
否则:
# 移动末端箱子
下一个末端箱子 = r
测试版 =
下一个测试版
如果
下一个启动块 ==
启动块
并且
下一个结束块 ==
结束块:
continue
使用 next_start_bin 和 next_end_bin 计算量化误差
规范 =
我.
_计算量化误差(next_start_bin,
下一个端点)
如果
标准 >
标准最小值:
断开
标准最小值 =
标准
开始分箱 =
下一个开始分箱
结束分箱 =
下一个端点箱
新最小值 =
我.
最小值 +
箱宽 *
开始分箱
新最大值 =
我.
最小值 +
分箱宽度 * (
结束二进制 + 1)
返回
新最小值,
新最大值
def _提升直方图(
我,
直方图:
火把.
张量,
原始最小值:
火把.
张量,
原始最大值:
火把.
张量,
更新最小值:
火把.
张量,
更新最大值:
火把.
张量,
):
这将直方图转换为更精细的直方图以减少
# 算子量化错误
# 直方图 =
直方图.
重复交织(
我.
# 上采样率) /
我.
# 上采样率
比例大小 = (
原始最大值 -
原始最小值) / (
我.
箱子数量 *
我.
上采样率)
中点直方图 = (
火把.
线性空间(
原始最小值,
原始最大值,
我.
箱子 *
我.
上采样率 + 1,
设备=
原始最小值.
设备,
`):`-1].
到(
直方图.
设备)
+ 0.5 * 箱大小
)
边界新直方图 =
火把.
线性空间(
更新最小值,
更新最大值,
我.
箱数 + 1,
设备=
更新最小值.
设备
).到(
直方图.
设备)
将直方图的中间点映射到新直方图的空间
桶分配 = (
火把.
桶化(
中点直方图,
新直方图边界,
正确=
是)
- 1
)
# 这将根据原始直方图的值，将直方图的中点映射到新空间中
# 这只是旧直方图在新直方图空间中的表示
由于数值问题，值可能超出最大/最小范围
桶分配[
桶分配 >=
我.
箱数] =
我.
箱子 - 1
桶分配[
桶分配 < 0] = 0
更新直方图 =
火把.
二进制计数(
桶分配,
权重=
直方图,
最小长度=
我.
垃圾箱
)
返回
更新直方图
def _合并直方图(
我,
原始直方图:
火把.
张量,
原始最小值:
火把.
张量,
原始最大值:
火把.
张量,
更新历史:
火把.
张量,
更新最小值:
火把.
张量,
更新最大值:
火把.
张量,
) -> 火把.
张量:
# 如果新的最小值和最大值与当前的最小值和最大值相同，
# 我们可以直接将新的直方图添加到原始直方图上
如果
更新最小值 ==
原始最小值
并且
更新最大值 ==
原始最大值:
返回
原始历史记录 +
更新历史
# 如果原始历史记录只有一个值（即最小值和最大值相同）
# 我们可以直接将其添加到新的直方图中
如果
原始最小值 ==
原始最大值:
二进制值 =
火把.
总和(
更新历史)
转换后的原始历史 = (
火把.
历史计数器(
原始最小值,
箱数=
我.
箱数,
最小=
更新最小值,
最大值=
更新最大值) # type: ignore[arg-type]
* 二进制值
)
返回
转换后的原始直方图 +
更新历史
# 假设 update_hist 已经在目标范围内，我们将 orig_max 映射到它
断言
更新最小值
≤
原始最小值
断言
更新最大值 >=
原始最大值
# 现在我们需要将旧直方图转换为新直方图的范围
转换后的原始直方图 =
我._upscale_histogram(
orig_hist,
orig_min,
orig_max,
更新最小值,
更新最大值,
)
返回
更新历史 +
转换后的原始历史
def 重置直方图(
我, x:
火把.
张量, min_val:
火把.
张量, max_val:
火把.
张量
) -> 无:
我.min_val.
调整大小(min_val.shape)
我.min_val.
复制_(min_val)
我.max_val.
调整大小(max_val.shape)
我.max_val.
复制_(max_val)
断言 (
min_val.元素数量() == 1
并且 max_val.
元素数量() == 1
), "直方图的最小/最大值必须是标量。"
新建直方图 =
火把.
历史计数器(x,
我.
箱数,
最小=min_val,
最大值=max_val) # type: ignore[arg-type]
我.
直方图.
断开连接().
调整大小(
新建直方图.shape)
我.
直方图.
复制_(
新直方图)
def 前向(
我,
x 原点:
火把.
张量) ->
火把.
张量: # pyre-ignore[14]
如果
原始 x.
元素数量() == 0:
返回
原始 x
x = 原始 x.detach()
x 最小值, x_max =
火把.
阿敏最大(x)
# 想要忽略 torch.inf，因为我们实际上并不需要
# 想要使我们的量化范围无限
# 实际上这些值将被限制
如果 x_min == -
火把.
无
或者 x_max ==
火把.
无穷:
警告.
警告(
输入张量中检测到 torch.inf，忽略输入)
x = x[x.绝对值() !=
火把.
无穷]
如果 x.
元素数量() == 0:
返回
原始 x
x_min, x 最大值 =
火把.
阿敏最大(x)
当前最小值 =
我.
最小值
当前最大值 =
我.max_val
未初始化 =
我.
最小值 == float(
"无穷大")
或者
我.max_val == float(
-无穷大)
如果
未初始化:
我.
重置直方图(x, x_min, x_max)
否则:
update_min, update_max = x_min, x_max
new_min = 火把.
最小(current_min,
更新最小值)
新最大值 =
火把.
最大值(
当前最大值,
更新最大值)
原因不明，此字段要求为通过 torchscript 测试
new_min 和 new_max 应该已经将 requires_grad 设置为 False
new_min, new_max = 新最小值.detach(),
新最大值.detach()
更新直方图 =
火把.
历史计数器(
x, 我.
箱数,
最小=
新最小值,
最大值=
新最大值 # type: ignore[arg-type]
).到(
我.
直方图.
设备)
如果
新最小值 ==
当前最小值
并且
新最大值 ==
当前最大值:
合并直方图 =
我.
直方图 +
更新直方图
我.
直方图.
断开连接().
调整大小(
合并直方图.shape)
我.
直方图.
复制_(
合并直方图)
否则:
合并直方图 =
我.
合并直方图(
我.
直方图,
当前最小值,
当前最大值,
更新直方图,
新最小,
新最大,
)
我.
直方图.
断开连接().
调整大小(
合并直方图.shape)
我.
直方图.
复制_(
合并直方图)
我.min_val.
断开连接().
调整大小(
新最小值.shape)
我.min_val.
复制_(
新最小值)
我.max_val.
断开连接().
调整大小(
新最大值.shape)
我.max_val.
复制_(
新最大值)
返回
原始 x
@torch.算子.
导出
def 计算 q 参数(
我):
未初始化 =
我.
最小值 == float(
"无穷大")
并且
我.
最大值 == float(
"-无穷"
)
如果
未初始化:
警告.
警告(
"必须在调用 calculate_qparams 之前运行 observer。"\
返回默认比例尺和零点。
)
返回
火把.
张量
[1.0
]，
设备=
我.min_val.
设备.
类型),
火把.
张量(
[0]，
设备=
我.min_val.
设备.
类型
)
断言
我.bins ==
长度(
我.
直方图), (
"直方图中的 bins 数量应等于 bins 的数量。"
"在制作此观察者时提供"
)
新_min,
新_max =
我.
_非线性参数搜索()
返回
我.
_计算_qparams(
新最小值,
新最大值)
def _保存到状态字典(
我,
目的地,
前缀,
保留变量):
超级().
保存到状态字典(
目的地,
前缀,
保留变量)
目的地[
前缀 +
min_val] =
我.min_val
目的地[
前缀 +
max_val] =
我.max_val
def 从状态字典加载(
我,
状态字典,
前缀,
本地元数据,
严格的,
缺少键,
预期之外的键,
错误消息,
):
版本 =
本地元数据.
获取(
版本,
无)
如果
版本
是
无
或者
版本 < 3:
# 如果 min_val 和 max_val 未初始化，则更新它们的形状
考虑 v2 和 v3 之间的差异
min_val_name, max_val_name = 前缀 + "min_val",
前缀 + "max_val"
如果
min_val 名称
在
状态字典:
如果
状态字典[
min_val 名称].
形状 ==
火把.
尺寸
[0
)]
状态字典[
min_val 名称] =
火把.
张量(float(
"无穷大"))
如果
最大值名称
在
状态字典:
如果
状态字典[
最大值名称].
形状 ==
火把.
尺寸
[0
)]
状态字典[
最大值名称] =
火把.
张量(float(
-无穷大))
本地状态 = ["min_val", "max_val"]
for 名称
在 local_state:
key = 前缀 +
名称
如果 key
在
状态字典:
val = 状态字典[
键]
setattr(我,
名称, val)
elif 严格的:
缺少键.append(
键)
超级().
从状态字典加载(
状态字典,
前缀,
本地元数据,
严格的,
缺少键,
预期之外的键,
错误消息,
)
def 额外表示(
我):
返回 f
min_val={
我.min_val}, max_val={
我.max_val}"
类
固定 QParams 观察者(
观察者基类):
r""
观察器，用于模拟固定量化和去量化
训练时间中的量化参数。仅针对每个张量
量化功能已支持。
参数：
`scale` (浮点数): 观察者的固定比例
`zero_point` (整数): 观察者的固定零点
`dtype`，`qscheme`，`quant_min`，`quant_max`
"文档"
比例:
火把.
张量
零点:
火把.
张量
def __init__(
我,
比例,
零点,
数据类型=
火把.quint8,
q 方案=
火把.
每张张量仿射,
量化最小值=0,
量化最大值=255,
动态的=
错误,
**kwargs,
):
如果
动态的:
raise 不支持的操作异常(
"FixedQParamsObserver 不支持动态量化"
)
超级().__init__(
数据类型=
数据类型,
动态的=
动态的, **kwargs)
我.
量化最小值 =
量化最小值
我.
量化最大值 =
量化最大值
我.
注册缓冲区(
"缩放",
火把.
张量
[
比例
]，
数据类型=
火把.float))
我.
注册缓冲区(
零点,
火把.
张量
输入文本:
([
翻译:
（[零点
]，
数据类型=
火把.
整数))
我.dtype = dtype
我.
虚拟方案 =
虚拟方案
def 前向(
我, X):
返回 X
@torch.算子.
导出
def 计算 q 参数(
我):
返回
我.
比例,
我.
零点
[文档]
类
占位符观察者(
观察者基类):
r""
不做任何事情，只是将配置传递给量化模块的 ``.from_float()``。
可用于量化到不需要确定浮点数精度的 float16。
可以用于量化到不需要确定浮点数精度的 float16。
范围。
参数：
`quantize` 节点的 `dtype` 参数，用于实现参考模型规范。
参考模型规范。
quant_min：量化域中的最小值（待办：与其他观察者对齐行为）
quant_max：量化域中的最大值
custom_op_name:（临时）指定此观察者用于不需要观察的操作符
(可用于图形模式通行证的特殊情况操作)。
compute_dtype（已弃用）：如果设置，则标记未来的量化函数使用
动态量化而非静态量化。
此字段已弃用，请使用 `is_dynamic=True` 代替。
is_dynamic：如果为 True，则参考模型中的 `quantize` 函数
将从该观察实例获取统计信息的表示形式
使用动态量化。
"文档"
def __init__(
我,
数据类型=
火把.float32,
自定义操作名称=
输入文本翻译为简体中文为："",
计算数据类型=
无,
量化最小值=
无,
量化最大值=
无,
q 方案=
无,
eps=无,
动态的=
错误,
) -> 无:
超级().__init__(
数据类型=
数据类型,
动态的=
动态的)
如果
虚拟方案
是
无:
虚拟方案 =
火把.
每张张量仿射
如果 eps
是
无:
eps = 火把.finfo(
火把.float32).eps
目标算子输入的数据类型，例如对于动态量化
# ops，数据类型将是 float32
我.dtype = dtype
我.
虚拟方案 =
虚拟方案
我.
量化最小值 =
量化最小值
我.
量化最大值 =
量化最大值
我.eps = eps
我.
自定义操作 =
自定义操作名称
# 用于配置动态量化的计算类型
如果
计算数据类型:
动态的 =
真实
警告.
警告(
"请使用 `动态的` 而不是 `compute_dtype`。"\
`compute_dtype` 将在未来的版本中弃用\
PyTorch。的
)
def 前向(
我, x):
返回 x
@torch.算子.
导出
def 额外表示(
我):
返回 f
dtype={
我.
数据类型}, is_dynamic={
我.
动态的}"
@torch.算子.
导出
def 计算 q 参数(
我):
raise 异常(
# 无需注意：TRY002
"calculate_qparams 不应该为 PlaceholderObserver 调用"
)
[文档]class RecordingObserver(ObserverBase):
r"""
模块主要用于调试，并在运行时记录张量值。
Args:
dtype: 量化数据类型
量化方案：要使用的量化方案
reduce_range: 将量化数据类型的范围减少 1 位
```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)
```
__annotations__ = {"tensor_val": 列表[Optional[torch.Tensor]]}
def __init__(self, dtype=torch.quint8):
super().__init__(dtype=dtype, is_dynamic=False)
self.tensor_val = []
def forward(self, x):
self.tensor_val.append(x.clone())
return x
@torch.jit.export
def calculate_qparams(self):
raise Exception( # noqa: TRY002
"calculate_qparams 不应该为 RecordingObserver 调用"
)
@torch.jit.export
def get_tensor_value(self):
return self.tensor_val
[文档]class NoopObserver(ObserverBase):
r"""
观察者不执行任何操作，只是将其配置传递给
量化模块的 ``.from_float()``。
主要用于量化到 float16，无需确定
范围。
Args:
数据类型：量化数据类型
custom_op_name: (临时)指定此观察者用于不需要观察的操作符
（可用于图模式 Pass 的特殊情况操作符）。
"""
"""
def __init__(self, dtype=torch.float16, custom_op_name="") -> None:
super().__init__(dtype=dtype, is_dynamic=False)
self.dtype = dtype
self.custom_op = 自定义操作名称
def forward(self, x):
return x
@torch.jit.export
def calculate_qparams(self):
raise Exception( # noqa: TRY002
"calculate_qparams 应该不适用于 NoopObserver"
)
类
重用输入观察者(
观察者基类):
r此观察者用于当我们想要重用操作符的观察者时
产生输入张量的函数，通常用于 reshape 等操作。
```
x0 = ...
x1 = x0.reshape()
```
如果我们将 x0 配置为被某些观察者观察，比如说 MinMaxObserver，
并且将 reshape 配置为 ReuseInputObserver，我们将重用观察者实例
对于 x0 和 x1（reshape 的输出）。如果 x0 没有被观察，我们也不会观察 x1。
注意：此功能仅在 FX 图形模式量化中启用
"文档"
def __init__(我) ->
无:
超级().__init__(
火把.quint8,
是否动态=
错误)
def 前向(
我, x):
返回 x
@torch.算子.
导出
def 计算 q 参数(
我):
raise 异常(
# 无需注意：TRY002
"calculate_qparams 不应该对 ReuseInputObserver 调用"
)
""
# 实验性仿射量化功能开始
我们计划在将 pt2e 流程移至 torchao 后，与 torchao 仓库合并以下内容
复制自 https://github.com/pytorch/ao/blob/main/torchao/quantization/observer.py
""
from dataclasses 导入
数据类
from 枚举
导入
自动,
枚举
[文档]类 MappingType(Enum):
"""浮点数如何映射到整数
对称映射意味着浮点数范围是对称映射到整数范围
假设我们有一个浮点数范围（-3.5，10.2）和整数范围（-8，7）（int4）
我们将使用（-10.2，10.2）作为浮点数的范围，并将其映射到（-8，7）
例如，缩放比例 = (10.2 - (-10.2)) / (7 - (-8))
SYMMETRIC_NO_CLIPPING_ERR 是对称映射的一种变体，其中缩放比例是 smin 的最大值
和 smax，其中 smin = min_val_neg / quant_min，smax = max_val_pos / quant_max。通过单独计算 smin 和 smax，可以减少负数的舍入误差，并且不会超出所有浮点数的范围。
smin 和 smax，可以减少负数的舍入误差，并且不会超出所有浮点数的范围。
所有浮点值。
非对称映射意味着我们直接将浮点数范围映射到整数范围，
对于上述示例，我们将把 (-3.5, 10.2) 映射到 (-8, 7) 并计算量化参数
基于此映射
例如，缩放比例 = (10.2 - (-3.5)) / (7 - (-8))
"""
SYMMETRIC = 自动()
SYMMETRIC_NO_CLIPPING_ERR = 自动()
ASYMMETRIC = 自动()
[文档]类 ZeroPointDomain(Enum):
枚举类型，表示 zero_point 是否在整数域或浮点域
整数域：quantized_val = (float_val / scale) (整数) + zero_point (整数)
浮点域：quantized_val = (float_val - (zero_point (浮点) - scale * mid_point)) / scale
无域：quantized_val = (float_val / scale)
"""
INT = 自动()
FLOAT = 自动()
NONE = 自动()
[文档]class TorchAODType(Enum):
"""
占位符，用于表示 PyTorch 核心中尚未存在的数据类型。
"""
PyTorch 2.6 中将添加 torch.int1 到 torch.int7
这些将保留以与旧版本的 PyTorch 保持向后兼容
INT1 = auto()
INT2 = auto()
INT3 = 自动()
INT4 = 自动()
INT5 = 自动()
INT6 = 自动()
INT7 = 自动()
[文档]@dataclass(冻结=True)
类 Granularity:
"""
基类，用于表示量化的粒度。
此类作为用于特定粒度类型的父类
量化操作，例如按张量或按轴量化。
```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)
```
[文档]@dataclass(frozen=True)
class PerBlock(粒度):
"""
表示量化中的每个块粒度。参见
`torchao.quantization.quant_primitives.quantize_affine` 的文档说明
`block_size`
属性：
block_size (元组[int, ...]): 每个量化组的尺寸
"""
block_size: 元组[int, ...]
[文档]@dataclass(frozen=True)
class PerTensor(粒度):
"""
表示量化中的张量粒度。
此粒度类型计算基于整个张量的量化参数。
基于整个张量进行量化参数的计算。
"``"
[文档]@dataclass(冻结=True)
class PerAxis(粒度):
"``"
表示量化在各个轴上的粒度。
此粒度类型沿张量的指定轴计算不同的量化参数。
例如，如果输入张量的形状为[8, 16]且 axis=0，那么
对于例如输入张量的形状为[8, 16]且 axis=0 的情况，
量化参数为张量的每一行计算。
总共提供 8 个量化参数。
属性：
轴（int）：执行降维的轴。
"""
轴: 整数
[文档]@dataclass(冻结=True)
class PerGroup(粒度):
"""
表示量化中的通道组粒度。
此粒度类型为每个元素组计算不同的量化参数。
为每个包含个元素的组计算不同的量化参数。
例如，如果输入张量的形状为[8, 16]，并且组大小为 4，那么
输入张量将被重塑为[64, 4]
为每个 4 个元素的组计算量化参数，
总共计算得到 64 个量化参数。
属性：
group_size (int): 每个量化组的尺寸
"""
group_size: int
[文档]class PerRow(粒度):
"""
表示量化中的行粒度。
这是按轴量化的特殊情况，并且是 Float8 矩阵乘法独有的。
输入量被以(1, ..., 输入.shape[-1])的 block_size 量化。权重
以(1, weight.shape[1])的 block_size 量化。
"""
[文档]类 PerToken(Granularity):
"""
表示量化中的按词粒度。
此粒度类型为每个标记计算一组不同的量化参数
，每个标记表示为张量的最后一个维度。
例如，如果输入张量的形状为[2, 3, 4]，那么就有 6 个 token
每个 token 包含 4 个元素，我们将计算 6 组量化参数，
每个 token 一组。
如果输入张量只有两个维度，例如[8, 16]，那么这
等同于 `PerAxis(axis=0)`，返回 8 组量化参数。
"""
[文档]def get_block_size(
输入形状：元组[int, ...]，粒度：Granularity
) -> tuple[int, ...]:
"""根据输入形状和粒度类型获取块大小。
Args:
input_shape: 输入张量形状，可能超过 2 维
粒度：量化的粒度类型
```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)
```
断言 isinstance(
粒度，粒度
请提供一个粒度的实例，而不是它的子类
如果 granuality 是 PerTensor 类型
返回输入形状
elif granuality 是 PerAxis 类型
block_size = list(input_shape)
block_size[granularity.axis] = 1
return tuple(block_size)
elif isinstance(granularity, PerRow):
return (1,) * (len(input_shape) - 1) + (input_shape[-1],)
elif isinstance(granularity, PerGroup):
assert (
len(input_shape) == 2
), 预期输入形状维度为 2 以进行每组量化，实际输入形状：{input_shape}
返回(1, granularity.group_size)
elif isinstance(granularity, PerToken):
block_size = list(input_shape)
block_size[-1] = 输入形状[-1]
返回 tuple(block_size)
抛出 ValueError(f"不支持的粒度: {granularity}")
[文档]
类 AffineQuantizedObserverBase(ABC,
火把.
神经网络.
模块):
观察模块用于仿射量化（https://github.com/pytorch/ao/tree/main/torchao/quantization#affine-quantization）
参数：
`粒度`和`块大小`：量化粒度，必须指定至少一个，如果两者都指定，则`块大小`优先
必须指定至少一个，如果两者都指定则`块大小`优先
当前支持的粒度类型为 `PerTensor` 和 `PerAxis`
其他参数请参阅 `:class:torchao.dtypes.AffineQuantizedTensor`
"文档"
with_args = 类方法(_with_args)
def __init__(
我,
mapping_type: MappingType,
target_dtype: 火把.
数据类型,
粒度:
粒度,
量化最小值:
可选[
整数] =
无,
量化最大值:
可选[
整数] =
无,
eps: 可选[float] =
无,
数据类型:
可选[
火把.
数据类型] =
无,
零点数据类型:
可选[
火把.
数据类型] =
无,
保留零:
布尔 =
是,
零点域:
可选[
零点域] =
零点域.
整数,
# there could be some extra args that's ignored
**kwargs,
):
超级().__init__()
断言
粒度
是
不
无,
"粒度是 None"
我.
映射类型 =
映射类型
我.
目标数据类型 =
目标数据类型
我.
粒度 =
粒度
我.
量化最小值 =
量化最小值
我.
量化最大值 =
量化最大值
我.eps = eps
我.
数据类型 =
数据类型
我.
零点类型 =
零点数据类型
我.
保留零 =
保留零
我.
零点域 =
零点域
前向传播期间填充
我.
块大小 =
无
我.
原始数据类型 =
无
[文档] @abstractmethod
def forward(self, input: torch.Tensor) -> torch.Tensor:
"""forward 函数应接收输入张量
并更新内部统计信息，并返回原始输入张量
"""
[文档] @abstractmethod
def 计算 qparams(self) -> tuple[torch.Tensor, torch.Tensor]:
"""根据附加到观察器模块的统计数据计算量化参数
返回一个包含缩放和零点张量的元组
"""
def _is_observer_script_module(模块, obs_type_name):
"返回 true 表示给定的 mod 是 Observer 脚本模块的实例。"
如果 isinstance(
模块,
火把.
算子.
递归脚本模块):
# 完整的名称看起来像 '__torch__.torch.ao.quantization.observer.___torch_mangle_2.MinMaxObserver'
后缀 =
模块._c.
合法名称.
分割(
“。”, 1
)1]
名称 =
正则表达式.
子(r
"__torch_mangle_\d+",
输入文本翻译为简体中文为："",
后缀)
返回 obs_type_name
在
名称
返回
假
# 实验性仿射量化功能结束
def _is_activation_post_process(模块):
返回 isinstance(
模块,
(
火把.
嗳.
量化.
观察者基类,
火把.
嗳.
量化.
模拟量化基类,
AffineQuantizedObserverBase,
),
) 或者 _is_observer_script_module(
模块,
"量化.观察者")
def _is_per_channel_script_obs_instance(模块):
如果 isinstance(
模块,
火把.
算子.
递归脚本模块):
返回 _is_observer_script_module(
模块,
"量化.观察者.按通道最小最大观察者"
) 或者 _is_observer_script_module(
模块,
"量化.观察者.移动平均按通道最小最大观察者"
)
返回
假
[文档]def get_observer_state_dict(mod):
r"""
返回与观察者状态相对应的状态字典。
遍历模型的状态字典并提取状态。
"""
od = OrderedDict()
if isinstance(mod, torch.jit.RecursiveScriptModule):
for k, v in mod.state_dict().items():
如果 "observer" 在 k 中
则 od[k] = v
否则:
# 图模块和 nn.Module 的路径（急切模式）
for k, v in mod.state_dict().items():
if "activation_post_process" in k:
od[k] = v
od._metadata = mod.state_dict()._metadata # type: ignore[attr-defined]
返回 od
[文档]def load_observer_state_dict(mod, obs_dict):
r"""
给定输入模型和一个包含模型观察者状态的 state_dict，
将状态重新加载回模型。观察者 state_dict 可以保存
使用 torch.ao.quantization.get_observer_state_dict
"""
missing_keys: list[str] = []
unexpected_keys: list[str] = []
for name, module in mod.named_modules():
prefix = name + "."
if _is_activation_post_process(module):
if _is_per_channel_script_obs_instance(module):
对于每个通道的观察者，我们需要调用自定义的 load_from_state_dict 来调整张量的大小。
然而，当模块被脚本化时，这并没有被调用，我们最终调用了 module.py 中的默认一个。
module._load_from_state_dict_script(
obs_dict, prefix, {}, True, missing_keys, unexpected_keys, []
)
else:
模块._load_from_state_dict(
obs_dict, prefix, {}, False, missing_keys, unexpected_keys, []
)
for k in missing_keys:
if "observer" in k or "activation_post_process" in k:
raise Exception( # noqa: TRY002
缺少状态字典中观察者 {k} 的键
)
for k in unexpected_keys:
如果 "observer" 在 k 中或 "activation_post_process" 在 k 中：
raise Exception( # noqa: TRY002
在 state_dict 中观察到意外的键{k}
)
# 限制激活值在范围（0,127）内
默认观察者 =
最小最大观察者.
带参数(
量化最小值=0,
量化最大值=127)
""
静态量化默认观察者，通常用于调试。
""
默认占位符观察者 =
占位符观察者
""
默认占位符观察者，通常用于将 torch.float16 进行量化。
""
默认调试观察者 = RecordingObserver
""
默认仅调试观察者。
""
默认权重观察者 =
最小最大观察者.
带参数(
数据类型=
火把.qint8,
q 方案=
火把.
每张张量对称
)
""
默认权重观察者。
""
负 127 到 127 的权重观察器范围 =
最小最大观察器.
带参数(
数据类型=
火把.qint8,
q 方案=
火把.
每张张量对称,
量化最小值=-127,
量化最大值=127,
eps=2**-12,
)
""
具有对称权重观察器，8 位值限制在[-127, +127]之间，不包括-128。
""
默认直方图观察者 =
直方图观察器.
带参数的(
量化最小值=0,
量化最大值=127)
""
默认直方图观察者，通常用于 PTQ。
""
默认每通道权重观察者 =
每通道最大最小值观察器.
带有参数(
数据类型=
火把.qint8,
q 方案=
火把.
单通道对称
)
""
默认的按通道权重观察者，通常用于支持按通道权重量化的后端，例如 `fbgemm`。
支持按通道权重量化的后端，例如 `fbgemm`。
""
per_channel_weight_observer_range_neg_127_to_127 = PerChannelMinMaxObserver.带参数(
数据类型=
火把.qint8,
q 方案=
火把.
每通道对称,
量化最小值=-127,
量化最大值=127,
eps=2**-12,
)
""
每通道对称权重观察器，8 位值限制在[-127, +127]之间，不包括-128。
""
默认动态量化观察者 =
占位符观察器.
带参数(
数据类型=
火把.quint8,
量化最小值=0,
量化最大值=255,
动态的=
是,
)
""
动态量化的默认观察者。
""
默认浮点零点观察者 = PerChannelMinMaxObserver.
带参数(
数据类型=
火把.quint8,
q 方案=
火把.
每通道仿射浮点量化参数,
坐标轴=0
)
""
浮点零点默认观察者。
""
默认浮点 q 参数观察者_4 位 = PerChannelMinMaxObserver.with_args(
数据类型=
火把.quint4x2,
q 方案=
火把.
每通道仿射浮点量化参数,
坐标轴=0
)
""
默认用于浮点零点和 4 位激活的观察者。
""
# TODO(未来 PR): 删除这些默认值并强制执行激活函数
# 明确指定它们的输出范围
默认固定 qparams 范围-1 到 1 观察者 =
固定 QParams 观察者.
带有参数(
比例=2.0 / 256.0,
零点=128,
数据类型=
火把.quint8,
量化最小值=0,
量化最大值=255
)
默认固定参数范围 0 到 1 观察者 =
固定 Q 参数观察者.
带参数(
比例=1.0 / 256.0,
零点=0,
数据类型=
火把.quint8,
量化最小值=0,
量化最大值=255
)
# TODO: 以下两个变量保留以供向后兼容；在几轮发布后删除
默认对称固定 q 参数观察者 =
默认固定 q 参数范围-1 到 1 观察者
默认仿射固定 q 参数观察者 =
默认固定 q 参数范围 0 到 1 的观察者
""
默认固定 q 参数操作的观察者
""
默认重用输入观察者 = ReuseInputObserver
""
默认观察者，用于重用输入的观察者，适用于 reshape 等操作符
该操作符
""
