# mypy: 允许未类型化装饰器
# mypy: 允许未类型化定义
导入 functools
导入
打字
来自
打字
导入
角色,
可选,
联合
来自 typing_extensions
导入
已弃用
导入
火炬
来自
火炬
导入
张量
来自 torch.utils._foreach_utils
导入 (
_device_has_foreach_support,
按设备类型和数据类型分组张量,
_has_foreach_support,
)
__all__ = [
clip_grad_norm_,
clip_grad_norm,
clip_grad_value_,
]
_tensor_or_tensors = 联盟[
火炬.
张量,
输入法.
迭代器[
火炬.
张量
]
# noqa: UP006 - 需要直到 XLA 的补丁更新
]
def _no_grad(函数):
""
此包装器用于在使用 @torch.no_grad 对公开函数时避免循环导入
clip_grad_norm_ 和 clip_grad_value_ 本身。
"源代码"
def _no_grad_wrapper(*参数, **kwargs):
与
火炬.
不梯度():
返回
函数(*
参数, **kwargs)
functools.更新包装器(_no_grad_wrapper,
函数)
返回 _no_grad_wrapper
@_no_grad
def _get_total_norm(
张量:
索引或索引,
归一化类型:
浮点数 = 2.0,
如果非有限值则报错:
布尔类型 =
错误,
foreach: 可选[
布尔] =
无,
) -> 火炬.
张量:
r计算张量序列的范数。
范数是通过对单个张量范数的计算来计算的,
单个张量被连接成一个单一的向量。
Args:
张量(Iterable[Tensor] 或 Tensor):张量的可迭代对象或单个张量,该张量将被归一化
norm_type(浮点数):使用的 p 范数的类型。可以是 ``'inf'`` 表示无穷范数。
norm_type (float): type of the used p-norm. Can be ``'inf'`` for
无穷范数。
错误如果非有限(布尔值):如果为 True,当总
张量的范数为“nan”、“inf”或“-inf”。
默认:``False``
使用基于 foreach 的更快实现。
如果为空,则使用 CUDA 和 CPU 原生张量的 foreach 实现,并静默地
转回其他设备类型的慢速实现。
默认:`None`
返回:
张量的总范数(视为单个向量)。
"源代码"
如果 isinstance(
张量,
火炬.
张量):
张量 = [
张量]
否则:
张量 =
列表(
张量)
规范类型 = float(
归一化类型)
如果
长度(
张量) == 0:
返回
火炬.
张量(0.0)
first_device = 张量[0].
设备
分组张量:
字典[
元组[
火炬.
设备,
火炬.
数据类型
]
元组[
列表[
列表[
张量]],
列表[int]]
] = 按设备类型和数据类型分组张量(
[张量] # type: ignore[list-item]
) # 类型:忽略[赋值]
规范:
列表[
张量] = []
为 (
设备, _), ([
设备张量
] _)
在
分组张量.
项目():
如果 (foreach
是
无
和
_支持 foreach 循环(
设备张量,
设备))
或 (
foreach 和
设备支持 foreach(
设备)
):
规范.
扩展(
火炬.
foreach 规范(
设备张量,
归一化类型))
elif foreach:
提升
运行时错误(
f"foreach=True 已传递,但不能使用 foreach API 在"{
设备.
类型}
张量上"
)
否则:
范数.
扩展(
[火炬.
线性代数.
向量范数(g,
归一化类型)
为 g
在
设备张量]
)
总规范 =
火炬.
线性代数.
向量范数(
火炬.
栈([
归一化.
到(
首个设备)
为
标准
在
规范
)]
规范类型
)
如果
如果非有限值则报错
和
火炬.
逻辑或(
总规约.isnan(),
总规约.
判断是否为无穷大()):
提升
运行时错误(
f"总范数的阶数"{
归一化类型}
用于从“
"参数`是非有限的,因此不能进行裁剪。要禁用“ "
"此错误并无论如何按非有限范数缩放梯度, "
"设置 `error_if_nonfinite=False`"
)
返回
总规范
@_no_grad
def _clip_grads_with_norm_(
参数: _tensor_or_tensors,
最大范数: float,
总规约:
火炬.
张量,
foreach: 可选[
布尔] =
无,
) -> 无:
r将给定预计算的总体范数和期望的最大范数的可迭代参数的梯度缩放。
梯度将通过以下计算进行缩放
.. math::
grad = grad * \frac{max_norm}{total_norm + 1e-6}
梯度将在原地修改。
此函数与 :func:`torch.nn.utils.clip_grad_norm_` 相当,但预先计算了总范数。
总范数。
Args:
参数 (Iterable[Tensor] 或 Tensor): 一个 Tensors 的可迭代对象或一个 Tensor
单个具有归一化梯度的张量
最大范数(浮点数):梯度的最大范数
总范数(Tensor):用于剪切的梯度总范数
使用基于 foreach 的快速实现(bool)。
如果为 ``None``,则对于 CUDA 和 CPU 原生张量使用 foreach 实现,并静默回退到其他设备类型的慢速实现。
注意:当系数被钳位到 1 时,乘以钳位的系数是多余的,但这样做
默认:`None`
返回:
无
"源代码"
如果 isinstance(
参数,
火炬.
张量):
参数 = [
参数]
梯度 = [p.
梯度
为 p
在
参数
如果 p.
梯度
是 not
无]
max_norm = float(最大范数)
如果
长度(
梯度) == 0:
返回
分组梯度:
字典[
元组[
火炬.
设备,
火炬.
数据类型
]
元组[
列表[
列表[
张量]],
列表[int]]
] = 按设备类型和数据类型分组张量(
[梯度]
) # 类型:忽略[赋值]
clip_coef = max_norm / (总规范 + 1e-6)
# 注意:当系数被钳位到 1 时,乘以钳位的系数是多余的,但这样做
避免了`if clip_coef < 1:`条件判断,这可能需要 CPU 与设备之间的同步
当梯度不在 CPU 内存中时。
clip_coef_clamped = 火炬.
卡钳(clip_coef,
最大值=1.0)
为 (
设备, _), ([
设备梯度
] _)
在
分组梯度.
项目():
如果 (foreach
是
无
和 _has_foreach_support(
设备梯度,
设备))
或 (
foreach 和 _device_has_foreach_support(
设备)
):
火炬._foreach_mul_(
设备梯度, clip_coef_clamped.
到(
设备))
elif foreach:
提升
运行时错误(
f"foreach=True 被传递,但不能使用 foreach API 在"{
设备.
类型}
张量上
)
否则:
clip_coef_clamped_device = clip_coef_clamped.到(
设备)
为 g
在
设备梯度:
g.mul_(clip_coef_clamped_device)
[文档]@_no_grad
def clip_grad_norm_(
参数: _tensor_or_tensors,
最大范数: float,
范数类型: float = 2.0,
如果非有限值则报错: bool = False,
foreach: 可选[bool] = None,
) -> torch.Tensor:
限制一个参数迭代器的梯度范数。
范数是通过对所有参数的各个梯度的范数进行计算得到的,
好像将个体梯度的规范连接成了一个单一的向量。
梯度就地修改。
此函数等价于:func:`torch.nn.utils.get_total_norm` 后跟
func:`torch.nn.utils.clip_grads_with_norm_`,其中 `total_norm` 由 `get_total_norm` 返回。
Args:
参数(Tensor 的可迭代对象或单个 Tensor):将要归一化梯度的 Tensor 的可迭代对象或单个 Tensor
max_norm(浮点数):梯度的最大范数
max_norm(浮点数):梯度的最大范数
norm_type (float): 使用 p-norm 的类型。可以是 ``'inf'`` 表示
无穷范数。
错误如果非有限(布尔值):如果为 True,当总
梯度的范数从 :attr:`parameters` 为 ``nan``
``inf``,或 ``-inf``。默认:False(未来将切换到 True)
foreach(布尔值):使用基于 foreach 的更快实现。
如果为 ``None``,则使用 CUDA 和 CPU 原生张量的 foreach 实现,并静默
回退到其他设备类型的慢速实现。
默认:``None``
返回:
参数梯度的总范数(视为单个向量)。
"""
if isinstance(parameters, torch.Tensor):
parameters = [parameters]
else:
# 防止生成器耗尽
parameters = list(parameters)
grads = [p.grad for p in parameters if p.grad is not None]
total_norm = _get_total_norm(grads, norm_type, error_if_nonfinite, foreach)
_clip_grads_with_norm_(parameters, max_norm, total_norm, foreach)
返回总范数
[文档]@已弃用(
现已弃用 `torch.nn.utils.clip_grad_norm`,
建议使用 `torch.nn.utils.clip_grad_norm_` 代替。
category=FutureWarning,
)
def clip_grad_norm(
parameters: _tensor_or_tensors,
max_norm: float,
norm_type: float = 2.0,
error_if_nonfinite: bool = False,
foreach: Optional[bool] = None,
) -> torch.Tensor:
限制一个参数迭代器的梯度范数。
.. 警告::
此方法现已弃用,推荐使用
`torch.nn.utils.clip_grad_norm_`
"""
返回 clip_grad_norm_(parameters, max_norm, norm_type, error_if_nonfinite, foreach)
[文档]@_no_grad
def clip_grad_value_(
参数: _tensor_or_tensors,
裁剪值: float,
foreach: Optional[bool] = None,
) -> None:
剪切参数迭代器中指定值的梯度。
梯度就地修改。
参数:
参数 (Iterable[Tensor] 或 Tensor): 一个 Tensors 的可迭代对象或一个 Tensor
单个具有归一化梯度的张量
clip_value (float): 最大允许的梯度值。
梯度被裁剪在范围内
`[-clip_value, clip_value]`
foreach (布尔值):使用基于 foreach 的快速实现
如果为 `None`,则使用 CUDA 和 CPU 原生张量的 foreach 实现并
静默回退到其他设备类型的慢速实现。
默认:``None``
``
如果 isinstance(parameters, torch.Tensor):
parameters = [parameters]
clip_value = float(clip_value)
grads = [p.grad for p in parameters if p.grad is not None]
grouped_grads = _group_tensors_by_device_and_dtype([grads])
for (device, _), ([grads], _) in grouped_grads.items():
if (
foreach is None
and _has_foreach_support(cast(list[Tensor], grads), device=device)
) or (foreach and _device_has_foreach_support(device)):
torch._foreach_clamp_min_(将 grads 转换为[Tensor]列表,-clip_value)
torch._foreach_clamp_max_(将 grads 转换为[Tensor]列表,clip_value)
elif foreach:
raise RuntimeError(
foreach=True 参数已传递,但无法在 {device.type} 矩阵上使用 foreach API
)
else:
for grad in grads:
将 Tensor 的梯度裁剪到[-clip_value, clip_value]范围内