导入 abc
导入 contextlib
导入 functools
导入
记录日志
导入
线程
来自
集合
导入
默认字典,
双端队列
来自 collections.abc
导入
生成器,
迭代器,
迭代器,
可变映射,
序列
来自
打字
导入 (
任何,
可调用,
角色,
直接,
命名元组,
可选,
类型检查,
联合,
)
来自 typing_extensions
导入
类型别名
来自
弱引用
导入 WeakKeyDictionary,
弱值字典
导入
火炬
来自 torch.autograd.variable
导入
变量
来自 torch.utils._python_dispatch
导入
Torch 分发模式
来自 torch.utils.hooks
导入
可移除句柄
如果
类型检查:
来自 torch._ops
导入
运算符重载
全部 = [
"保存张量钩子",
"在 CPU 上保存",
"禁用保存张量钩子",
"注册多梯度钩子",
"允许在保存的 tensor 上执行修改",
"Node",
"梯度边",
"获取梯度边",
"增加版本",
]
日志 =
记录.
获取日志记录器(__name__)
类
节点(abc.ABC):
[文档] @abc.抽象方法
def 名称(self) -> str:
返回名称。
示例::
导入 torch 库
创建一个需要梯度的 tensor：a = torch.tensor([0., 0., 0.], requires_grad=True)
>>> b = a.clone()
>>> assert isinstance(b.grad_fn, torch.autograd.graph.Node)
>>> print(b.grad_fn.name())
CloneBackward0
"``"
抛出未实现异常
@property
@abc.抽象方法
def 下一个函数(self) ->
元组[
元组[
可选["Node"
] int
] ...
]
抛出
未实现异常
[文档] @abc.abstractmethod
def 元数据(self) -> dict:
r"""返回元数据。"""
抛出未实现异常
@property
@abc.抽象方法
def 输入元数据(self) ->
列表[
任何
]
抛出
未实现异常
@abc.抽象方法
def _注册钩子字典(self,
张量:
PyTorch.
张量) ->
无:
抛出
未实现异常
[文档] @abc.abstractmethod
def 注册钩子(self, fn: Callable[..., Any]) -> RemovableHandle:
r"""注册反向钩子。
每次计算相对于梯度的导数时，都会调用此钩子。
节点已计算。钩子应具有以下签名::
hook(grad_inputs: 张量元组, grad_outputs: 张量元组) -> 张量元组或 None
钩子不应修改其参数，但可以可选地返回
一个新的梯度，该梯度将用于替换 :attr:`grad_inputs`。
此函数返回一个带有方法 `handle.remove()` 的句柄
该方法从模块中移除钩子
.. 注意::
有关此钩子何时执行的更多信息，请参阅 :ref:`backward-hooks-execution`
执行过程以及其相对于其他钩子的执行顺序。
.. 注意::
在非常罕见的情况下，当钩子在节点已经开始执行时注册，就不再保证 :attr:`grad_outputs`。
在钩子在节点已经开始执行时注册的罕见情况下，不再有关于 :attr:`grad_outputs` 的任何保证。
内容（可能因其他因素而正常或为空）。
钩子仍然可以选择性地返回一个新梯度，以替换使用
:attr:`grad_inputs` 与 :attr:`grad_outputs` 独立。
示例：
>>> 导入 torch 库
>>> a = torch.tensor([0., 0., 0.], requires_grad=True)
>>> b = a.clone()
>>> assert isinstance(b.grad_fn, torch.autograd.graph.Node)
>>> handle = b.grad_fn.register_hook(lambda gI, gO: (gO[0] * 2,))
>>> b.sum().backward(retain_graph=True)
>>> print(a.grad)
tensor([2., 2., 2.])
>>> handle.remove() # 移除钩子
>>> a.grad = None
>>> b.sum().backward(retain_graph=True)
>>> print(a.grad)
tensor([1., 1., 1.])
"""
抛出未实现异常
[文档] @abc.abstractmethod
def 注册预钩子(self, fn: Callable[..., Any]) -> RemovableHandle:
r"""注册反向预钩子。
每次计算相对于梯度的导数时，都会调用此钩子。
节点已计算。钩子应具有以下签名::
钩子应不修改其参数，但可以可选地返回
一个新的梯度，该梯度将用于替换 :attr:`grad_outputs`。
如果返回新的梯度，则它将被用于代替 :attr:`grad_outputs`。
该函数返回一个带有方法 `handle.remove()` 的句柄
该方法从模块中移除钩子
.. 注意::
有关此钩子何时执行的更多信息，请参阅 :ref:`backward-hooks-execution`
执行过程及其相对于其他钩子的执行顺序。
示例::
>>> a = torch.tensor([0., 0., 0.], requires_grad=True)
>>> b = a.clone()
>>> 断言 b.grad_fn 是 torch.autograd.graph.Node 的实例
>>> handle = b.grad_fn.register_prehook(lambda gI: (gI[0] * 2,))
>>> b.sum().backward(retain_graph=True)
>>> 打印(a.grad)
tensor([2., 2., 2.])
>>> handle.remove()
>>> a.grad = None
>>> b.sum().backward(retain_graph=True)
>>> 打印(a.grad)
tensor([1., 1., 1.])
"""
抛出未实现异常
@classmethod
def __subclasshook__(类,
子类:
类型) -> bool:
如果
类
是
节点
和 (
(
子类
是
不
无
和
子类
是 getattr(
PyTorch._C.
函数,
子类.__name__,
无)
)
或者
派生类(
子类,
PyTorch.
自动微分.
函数.
向后 C 函数)
):
返回
真实
返回 NotImplemented
def _获取梯度函数或梯度累积器(t:
联合[
PyTorch.
张量,
"梯度边"]) ->
节点:
如果 isinstance(t,
梯度边缘):
返回 t.
节点
如果 t.
需要梯度
和 t.grad_fn
是
无:
替换为
PyTorch.
启用梯度():
节点 = t.
以查看方式(t).grad_fn.
下一个函数[0
]
[0] # 类型：忽略[联合属性]
否则:
节点 = t.grad_fn
断言
节点
是
不
无
返回
节点
[文档]类 GradientEdge(NamedTuple):
表示 autograd 图中给定梯度边的对象。
获取给定张量梯度将计算的梯度边缘
您可以使用 `edge = autograd.graph.get_gradient_edge(tensor)` 获取张量的梯度边。
```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)
```
node: 节点
output_nr: 整数
[docs]def get_gradient_edge(tensor: torch.Tensor) -> GradientEdge:
"""获取给定张量的梯度边，用于计算该张量的梯度。
特别是，它等价于调用
`g = autograd.grad(loss, input)` 和 `g = autograd.grad(loss, get_gradient_edge(input))`.
```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)
```
如果 tensor 不需要梯度
引发 RuntimeError(
无法获取张量的梯度边缘
"不需要梯度"
)
grad_fn = _get_grad_fn_or_grad_acc(tensor)
# Note that output_nr default to 0 which is the right value
# for the AccumulateGrad node.
return GradientEdge(grad_fn, tensor.output_nr)
[docs]def increment_version(tensor: Union[torch.Tensor, Iterable[torch.Tensor]]) -> None:
"""更新自动微分元数据跟踪给定的 Tensor 是否被就地修改。
这是为了在自动微分引擎中实现更精确的错误检查。
这已经由 PyTorch 函数和自定义 Function 自动完成。
当适当地调用 mark_dirty() 时，你只需显式调用此函数
如果你在以 PyTorch 不支持的方式对 Tensor 数据进行原地操作
了解。例如，一个自定义内核读取 Tensor 数据指针并修改
基于此指针的内存原地。可以接受一个张量或张量列表。
请注意，对单个就地操作多次增加版本计数器
不是问题。
请注意，如果传递在 torch.inference_mode()下构建的张量，
我们不会增加其版本计数器（因为你的张量没有版本计数器）。
"""
如果 tensor 是 torch.Tensor 类型：
tensor = (tensor,)
torch._C._increment_version(tensor)
[文档]
类
保存张量钩子:
上下文管理器，用于为保存的张量设置一对打包/解包钩子。
使用此上下文管理器来定义操作中间结果的打包方式，以便在保存前进行打包，并在检索时进行解包。
在此上下文中，每当进行操作时，都会调用 `pack_hook` 函数。
在此上下文中，`pack_hook` 函数将在每次保存张量之前被调用。
操作保存张量以供反向传播（这包括中间结果）
使用
func:`~torch.autograd.function._ContextMethodMixin.save_for_backward` 但
也包括由 PyTorch 定义的操作记录的输出）。
``pack_hook``随后被存储在计算图中，而不是原始张量。
``unpack_hook``在需要访问已保存的张量时被调用，即当执行:func:`torch.Tensor.backward()`时。
当需要访问已保存的张量时，会调用``unpack_hook``，即当执行:func:`torch.Tensor.backward()`时。
当需要访问已保存的张量时，会调用``unpack_hook``，即当执行:func:`torch.Tensor.backward()`时。
`torch.autograd.grad()` 函数接受一个 *打包* 对象作为参数。
由 `pack_hook` 返回的对象，并应返回一个与原始张量（作为对应 `pack_hook` 输入传递的张量）具有相同内容的张量。
`pack_hook` 的输入传递的张量）的内容。
`pack_hook` 的输入传递的张量）的内容。
插件应具有以下签名：
pack_hook(tensor: Tensor) -> Any
unpack_hook(Any) -> Tensor
其中 `pack_hook` 的返回值是 `unpack_hook` 的有效输入。
通常情况下，你希望从数值、大小、数据类型和设备等方面来看，`unpack_hook(pack_hook(t))` 等于 `t`。
在此方面。
示例::
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_AUTOGRAD)
>>> def pack_hook(x):
... 打印("打包", x)
... 返回 x
...
>>> 定义 unpack_hook(x):
... 打印("解包", x)
... 返回 x
...
>>> a = torch.ones(5, requires_grad=True)
>>> b = torch.ones(5, requires_grad=True) * 2
>>> with torch.autograd.graph.saved_tensors_hooks(pack_hook, unpack_hook):
... y = a * b
打包 tensor([1., 1., 1., 1., 1.], requires_grad=True)
打包 tensor([2., 2., 2., 2., 2.], grad_fn=)
>>> y.sum().backward()
解包 tensor([1., 1., 1., 1., 1.], requires_grad=True)
解包 tensor([2., 2., 2., 2., 2.], grad_fn=)
..警告::
对输入的 hooks 进行就地操作可能会导致未定义的行为。
可能导致未定义行为。
..警告::
一次只能使用一对钩子。当递归嵌套时
上下文管理器，只有最内层的钩子会被应用。
"""
def __init__(
self,
打包钩子:
可调用[[
PyTorch.
张量
]
任何
]
解包钩子:
可调用[[
任何
]
PyTorch.
张量
]
) -> 无:
self.打包钩子 =
打包钩子
self.解包钩子 =
解包钩子
def __进入__(self) ->
无:
PyTorch._C.
_自动微分._push_saved_tensors_default_hooks(
self.打包钩子, self.
解包钩子
)
def __退出__(self, *
参数:
对象) ->
无:
PyTorch._C.
_自动微分.
默认钩子保存 pop 张量()
[文档]
类
保存到 CPU(
保存张量钩子):
在此上下文中管理器下，前向传递保存的张量将被存储在 CPU 上，然后用于反向检索。
当在此上下文中管理器内执行操作时，中间结果
结果将在前向传播过程中保存在图中，将被移动到 CPU
当需要反向传播时，会将其复制回原始设备。
如果图已经位于 CPU 上，则不执行张量复制。
使用此上下文管理器以 GPU 内存使用量换取计算（例如
当您的模型在训练过程中不适用于 GPU 内存时）。
参数：
pin_memory（布尔值）：如果为 ``True``，张量将被保存到 CPU 固定内存
在打包过程中异步复制到 GPU，在解包过程中异步从 GPU 复制。
默认为 ``False``。
也请参阅：:ref:`cuda-memory-pinning`。
示例::
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA)
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_AUTOGRAD)
>>> a = torch.randn(5, requires_grad=True, device="cuda")
>>> b = torch.randn(5, requires_grad=True, device="cuda")
>>> c = torch.randn(5, requires_grad=True, device="cuda")
...
>>> def f(a, b, c):
... prod_1 = a * b # a 和 b 存储在 GPU 上
... with torch.autograd.graph.save_on_cpu():
... prod_2 = prod_1 * c # prod_1 和 c 存储在 CPU 上
... y = prod_2 * a # prod_2 和 a 存储在 GPU 上
... 返回 y
...
>>> y = f(a, b, c)
>>> del a, b, c # 仅用于说明
>>> # a, b 和 prod_2 的内容在 GPU 上仍然存活
>>> # prod_1 和 c 的内容仅存在于 CPU 上
>>> y.sum().backward() # 所有 CPU 张量都移动回 GPU，用于反向传播
>>> # 在调用 backward()之后，所有中间张量都会被释放（删除）
"""
def __init__(self, 持久化内存:
布尔值 = False,
设备类型:
字符串 =
cuda) ->
无:
设备模块 = getattr(
PyTorch,
设备类型,
PyTorch.cuda)
def 打包到 CPU(
张量:
PyTorch.
张量) ->
元组[
PyTorch.
设备,
PyTorch.
张量
]
如果
不
持久化内存:
返回 (
张量.
设备,
张量.cpu())
打包 =
PyTorch.
空的(
张量.
尺寸(),
数据类型=
张量.
数据类型,
布局=
张量.
布局,
持久化内存=(
设备模块.
是否可用()
和
不
张量.is_sparse),
)
拼装.
复制_(
张量)
返回 (
张量.
设备,
拼装)
def 从 CPU 解包(
拼装:
元组[
PyTorch.
设备,
PyTorch.
张量]) ->
PyTorch.
张量:
设备,
张量 =
打包
返回
张量.
到(
设备,
非阻塞=
持久化内存)
超级().__init__(
打包到 CPU,
从 CPU 解包)
[文档]@contextlib.contextmanager
def disable_saved_tensors_hooks(error_message: str) -> 生成器[None, None, None]:
"""禁用保存张量默认钩子功能的上下文管理器。
适用于您正在创建一个不与保存的
张量默认钩子不起作用的特性。
参数：
error_message (str): 当使用保存的张量默认钩子时，
已被禁用，出现了一个 RuntimeError 错误
错误信息被引发。
示例：
>>> # xdoctest: +忽略（失败）
>>> message = "已保存张量的默认钩子被禁用"
>>> with torch.autograd.graph.disable_saved_tensors_hooks(message):
... # 抛出 RuntimeError: 已保存张量的默认钩子被禁用
... with torch.autograd.graph.save_on_cpu():
...通过
"""
maybe_prev_message = None
try:
maybe_prev_message = (
torch._C._autograd._saved_tensors_hooks_get_disabled_error_message()
)
torch._C._autograd._saved_tensors_hooks_disable(error_message)
yield
finally:
# 注意：[disabled_error_message 不变]
if maybe_prev_message is None:
torch._C._autograd._saved_tensors_hooks_enable()
else:
torch._C._autograd._saved_tensors_hooks_disable(maybe_prev_message)
类 _MultiHandle(
可移除句柄):
处理:
元组[
可移除句柄, ...]
def __init__(self, 处理:
元组[
可移除句柄, ...]) ->
无:
self.处理 =
处理
def 删除(self) ->
无:
for handle 在 self.
处理:
处理.
删除()
def __getstate__(self) -> 元组[
可移除句柄, ...
]
返回 self.
处理
def __setstate__(self, 状态:
元组[
可移除句柄, ...]) ->
无:
self.处理 =
状态
[文档]def
注册多梯度钩子(
张量:
序列[
PyTorch.
张量
]
函数:
联合[
可调用[[
序列[
可选[
PyTorch.
张量
]]，
无
]
可调用[[
PyTorch.
张量
]
无
]
]
*,
模式:
直接[
所有,
任何] =
所有,
) -> 可移除句柄:
r"注册一个多梯度反向钩子。"
有两种支持的模式：`"all"` 和 `"any"`。
在“all”模式下，钩子将在相对于每个张量的梯度之后被调用
`tensors` 已计算。如果一个张量在 :attr:`tensors` 中但
不是图的一部分，或者如果不需要张量来计算梯度
对于当前 ``.backward()`` 或 ``.grad()`` 调用指定的任何 ``inputs``，这个张量将被忽略，钩子将不会等待其梯度被计算。
在每个非忽略张量的梯度被计算后，:attr:`fn` 将被调用。
计算完毕。
在每个非忽略张量的梯度计算完毕后，:attr:`fn` 将被调用。
调用这些梯度。对于未传递张量的情况，将传递 ``None``。
计算它们的梯度。
在“any”模式下，钩子将在第一次梯度之后被调用
关于在 :attr:`tensors` 中的张量已计算。钩子
将以此梯度作为其参数调用。
钩子不应修改其参数。
此函数返回一个带有方法 `handle.remove()` 的句柄，用于移除钩子。
.. 注意::
查看：:ref:`backward-hooks-execution` 了解有关此钩子何时执行的更多信息
的执行方式以及相对于其他钩子的执行顺序。
示例::
>>> 导入 torch
...
>>> a = torch.rand(2, 3, requires_grad=True)
>>> b = torch.rand(2, 3, requires_grad=True)
>>> c = a * b
>>> d = a * b
...
>>> 定义函数 fn(grads):
... 打印([g is not None for g in grads])
...
>>> torch.autograd.graph.register_multi_grad_hook((a, b, c, d), fn)
...
>>> c.sum().backward(retain_graph=True)
[True, True, True, False]
>>> c.sum().backward(inputs=(a,), retain_graph=True)
[True, False, True, False]
...
"""
supported_modes = (所有,
任何)
锁定 =
线程.
锁()
如果
模式
不
在
支持的模式:
抛出
值错误(f
"期望模式为以下之一"{
支持的模式}
但是得到了{
模式}")
如果
模式 ==
所有:
数量:
字典[int, int] = {}
调用次数 =
无
缓冲区:
字典[int,
列表[
可选[
PyTorch.
张量]]] = {}
梯度函数 =
列表(
地图(
_获取梯度函数或梯度累积器,
张量))
张量长度 =
长度(
张量)
def get_inner_hook(索引: int) ->
可调用[[
PyTorch.
张量
]
无
]
def 内部钩子(
研究生:
PyTorch.
张量) ->
无:
非局部
数量,
调用次数,
缓冲区, fn
标识符 =
PyTorch._C.
_当前图任务 ID()
断言 (
标识符 != -1
), "期望这个钩子在反向调用中被调用"
数量[id] =
数量.
获取(id, 0)
缓冲区[id] =
缓冲区.
获取(id, [
无] *
张量长度)
替换为
锁:
当前计数,
数量[id] =
数量[id
]
数量[id] + 1
如果
当前计数 == 0:
第一次调用时，计算实际的 nb_calls 和 buffer
调用次数 =
总和(
地图(
PyTorch._C._will_engine_execute_node,
梯度函数)
)
缓冲区[id
]
[索引] =
梯度
断言
调用次数
是
不
无
如果
当前计数 ==
调用次数 - 1:
fn = 角色(
可调用[[
序列[
可选[
PyTorch.
张量
]]，
无
]
函数)
函数(
缓冲区[id])
删除
数量[id]
删除
缓冲区[id]
返回
内部钩子
处理 =
元组(
t.注册钩子(get_inner_hook(i)) for i, t
在
列举(
张量)
)
如果...否则
模式 ==
任何:
fn = 角色(
可调用[[
PyTorch.
张量
]
无
]
函数)
挂钩:
字典[int, bool] =
默认字典(bool)
@functools.包装(
函数)
def wrapped_fn(研究生:
PyTorch.
张量) ->
无:
非局部
运行钩子
标识符 =
PyTorch._C.
_当前图任务 ID()
断言
标识符 != -1,
"期望这个钩子在反向调用中被调用"
替换为
锁:
上一页,
挂钩[id] =
挂钩[id
]
真实
如果
上一页:
返回
函数(
研究生)
处理 =
元组(
张量.
注册钩子(wrapped_fn)
for 张量
在
张量
如果
张量.
需要梯度
)
返回 _MultiHandle(
处理) # type: ignore[possibly-undefined]
# 允许对保存用于反向传播的张量进行修改
#
# 1. 张量被保存用于反向传播
# - 记录 Python 对象 ID 和张量的版本
# - 记录别名信息（基类数据指针+版本）
# - 保存原始数据，以便我们控制其生命周期
# 2. 任何时间张量进行原地修改
对于每个被它别名的张量：
使用其对象 ID 和版本检查是否已保存
如果已保存，则克隆它
删除对原始对象的引用
# 3. 在反向过程中
# - 如果克隆存在，张量必须在原地被修改
_允许在保存的张量上启用修改:
布尔值 =
假
_TID: 类型别名 =
元组[int, int, int]
_SID: 类型别名 =
元组[int, int]
def _get_tid(张量:
PyTorch.
张量) -> _TID:
# 错误：这几乎肯定是一个错误。
如果 isinstance(
张量,
(
PyTorch.
子类.fake_tensor.
假 Tensor,
PyTorch.
子类.functional_tensor.FunctionalTensor,
),
):
数据指针 = 0
否则:
数据指针 =
张量.
数据指针()
返回 (id(
张量),
数据指针,
张量.
_版本)
def 获取会话 ID(
张量:
PyTorch.
张量) -> _SID:
# 错误：这几乎肯定是一个错误。
如果 isinstance(
张量,
(
PyTorch.
子类.fake_tensor.
假 Tensor,
PyTorch.
子类.functional_tensor.FunctionalTensor,
),
):
数据指针 = 0
否则:
数据指针 =
张量.
数据指针()
返回 (
数据指针,
张量.
_版本)
类
处理:
通过
类
_与克隆体交换(
保存张量钩子):
def __init__(self, ctx: "_允许在保存的上下文中进行突变") ->
无:
def 打包钩子(
张量:
PyTorch.
张量) ->
处理:
tid = _get_tid(张量)
sid = 获取会话 ID(
张量)
# Tensors 保存用于反向传播的条目在 _tid_to_weakhandle 中
处理:
可选[
处理] =
无
# 保存别名信息
ctx.sid_to_tid[sid].添加(tid)
# NB: 同一个版本的同种张量可以被保存多次
如果 tid
不
在 ctx.tid_to_weakhandle:
handle = 处理()
ctx.tid_to_weakhandle[tid] = handle
ctx.原始[
处理] =
张量
否则:
存储对句柄的额外强引用
handle = ctx.tid_to_weakhandle[tid]
返回 handle
def 解包钩子(
处理:
处理) ->
PyTorch.
张量:
错误信息 = (
"尝试在 'allow_mutation_on_saved_tensors' 上下文之外回溯"
"其中图最初被记录的上下文中。"
)
断言
_允许在保存的张量上启用修改,
错误信息
如果 handle
在 ctx.
克隆的:
res = ctx.克隆的[
处理]
否则:
断言 handle
在 ctx.
原始,
错误信息
res = ctx.原始[
处理]
返回 res
超级().__init__(
打包钩子,
解包钩子)
类 _CloneArgBeforeMutateMode(
火炬调度模式):
def __init__(self, ctx: "_允许在保存的上下文中进行突变") ->
无:
self.ctx = ctx
def __torch_dispatch__(
self,
函数:
"运算符重载",
类型:
迭代器[
类型
]
参数:
元组[
任何, ...] = (),
kwargs: 可选[
字典[
任何,
任何]] =
无,
) -> 任何:
kwargs = kwargs 或者 {}
def 可能克隆(t:
PyTorch.
张量) ->
无:
tid = _get_tid(t)
sid = 获取会话 ID(t)
ctx = self.ctx
如果 sid
在 ctx.sid_to_tid:
for tid 在 ctx.sid_to_tid[sid
]
如果 tid
不
在 ctx.tid_to_weakhandle:
我们知道，如果 tid 在 sid_to_tid 中，那么它也必须在
# tid_to_weakhandle。然而，张量可能是
# 保存过一次，但在修改之前被回退清除了
就地。考虑以下示例：
#
# >>> a = torch.randn(2, 3, requires_grad=True).clone()
# >>> out = (a**2).sum()
# >>> out.backward()
# >>> a.sin_()
continue
handle = ctx.tid_to_weakhandle[tid]
如果 handle
在 ctx.
克隆的:
# 同样的精确张量已经被克隆过了
continue
ctx.克隆的[
处理] = ctx.
原始[
处理].
克隆()
删除 ctx.
原始[
处理]
for 索引,
参数
在
列举(
函数.
_模式.
参数):
如果
参数.
别名信息
是
不
无
和
参数.
别名信息.
是否写入:
如果
参数.
是否输出:
可能克隆(kwargs["out"])
如果...否则 isinstance(
参数[
索引
]
列表):
# 每个案例。（可能的优化：如果大多数张量需要克隆，使用 for each clone？）
# 如果需要克隆张量，使用 for each clone？
for t 在
参数[
索引
]
可能克隆(t)
否则:
可能克隆(
参数[
索引])
返回
函数(*
参数, **kwargs)
类
_允许在保存的上下文中进行突变:
def __init__(self) -> 无:
self.克隆的:
可变映射[
处理,
PyTorch.
张量] = WeakKeyDictionary()
self.原始:
可变映射[
处理,
PyTorch.
张量] = WeakKeyDictionary()
self.tid_to_weakhandle: 可变映射[_TID,
处理] =
弱值字典()
self.sid_to_tid: 字典[_SID,
设置[_TID]] =
默认字典(
设置)
def 清晰(self) ->
无:
self.克隆的.
清晰()
self.原始.
清晰()
self.tid_to_weakhandle.清晰()
self.sid_to_tid.清晰()
[文档]@contextlib.contextmanager
def allow_mutation_on_saved_tensors() -> (
生成器[_AllowMutationOnSavedContext, None, None]
):
在反向传播过程中允许保存可变张量的上下文管理器。
在此上下文管理器中，为反向传播保存的张量在变异时被克隆
因此，原始版本在回溯时仍然可以使用。通常，在回溯时修改张量
保存用于反向操作的内容，在反向操作中使用时将引发错误。
确保正确行为，正向和反向都应该在相同的上下文管理器下运行
在同一个上下文管理器下。
返回：
一个存储此上下文管理器所管理状态的 _AllowMutationOnSavedContext 对象。
环境管理器。此对象在调试过程中可能很有用。
环境管理器管理的状态在退出时自动清除。
示例::
>>> 导入 torch
>>> with torch.autograd.graph.allow_mutation_on_saved_tensors():
... # forward
... a = torch.ones(2, 3, requires_grad=True)
... b = a.clone()
... out = (b**2).sum()
... b.sin_()
... # backward
... out.sum().backward()
...
tensor([[0.8415, 0.8415, 0.8415],
[0.8415, 0.8415, 0.8415]], grad_fn=)
"""
global _allow_mutation_on_saved_tensors_enabled
ctx = _AllowMutationOnSavedContext()
with _swap_with_cloned(ctx), _CloneArgBeforeMutateMode(ctx):
try:
if _allow_mutation_on_saved_tensors_enabled:
raise RuntimeError(
"允许在已保存张量上执行突变的环境不能嵌套"
)
_允许在保存的 tensor 上启用突变 = True
yield ctx
finally:
ctx.clear()
_allow_mutation_on_saved_tensors_enabled = False
def _register_logging_hooks_on_whole_graph(
t_outputs: 序列[
联合[
PyTorch.
张量,
梯度边缘]],
) -> 可调用
[]
无
]
梯度函数 =
列表(
地图(
_获取梯度函数或梯度累积器, t_outputs))
def 迭代图(
根:
列表[
节点]) ->
迭代器[
节点
]
如果
不
根:
返回
已见:
设置[
节点] =
设置()
q: 双端队列[
节点] =
双端队列()
for 节点
在
根:
如果
节点
是
不
无:
已见.
添加(
节点)
q.添加(
节点)
当 q:
节点 = q.popleft()
for 函数, _
在
节点.
下一个函数:
如果 fn
在
看过
或者 fn
是
无:
continue
已见.
添加(
函数)
q.添加(
函数)
产生
节点
def 格式化(t:
可选[
PyTorch.
张量]) -> str:
避免循环导入
来自 torch.testing._internal.common_utils
导入 dtype_abbrs
如果 t
是
无:
返回 "None"
返回 f"{
数据类型缩写[t.
数据类型]}[{
“，”.
连接(
地图(str, t.
形状))}]"
def 预钩子(grad_outputs:
序列[
可选[
PyTorch.
张量]]) ->
无:
节点 =
PyTorch._C.
当前自动微分节点()
梯度输出字符串 = f"[{
逗号.
连接(
格式化(t) for t
在 grad_outputs)}]"
日志字符串 = f
"执行："{
节点}
使用 grad_outputs：{grad_outputs_str}"
日志.
调试(log_str)
处理 = [
节点.
注册预钩子(
预钩子) for
节点
在
迭代图(
梯度函数)]
def 取消注册钩子() ->
无:
for handle 在
处理:
处理.
删除()
返回
注销钩子
def _engine_run_backward(
t_outputs: 序列[
联合[
PyTorch.
张量,
梯度边缘]],
*参数:
任何,
**kwargs: 任何,
) -> 元组[
PyTorch.
张量, ...
]
添加日志钩子 =
日志.
获取有效级别() <=
记录.
调试
如果
添加日志钩子:
注销钩子 = _register_logging_hooks_on_whole_graph(t_outputs)
尝试:
返回
变量._execution_engine.
向后运行(
调用 C++引擎执行反向传播
t_outputs, *参数, **kwargs
) 调用 C++引擎执行反向传播
最后:
如果
添加日志钩子:
取消注册钩子() # type: ignore[possibly-undefined]
