torch.nn.utils.stateless 的源代码

# mypy: 允许未类型化定义
导入 contextlib
来自 打字 导入 任意, 可选, 联合
来自 typing_extensions 导入 已弃用

导入 火炬
来自 火炬 导入 张量
来自 torch.nn.utils._named_member_accessor 导入 命名成员访问器


__all__ = [功能调用]


def _解绑命名张量映射(
    模块: "torch.nn.Module",
    参数和缓冲区: 字典[字符串, 张量]
) -> 字典[字符串, 张量]：
    ""
解除模块中所有绑定到 parameters_and_buffers 的张量。

此函数返回一个新的未绑定的 untied_parameters_and_buffers 字典，并保持原始
untied_parameters_and_buffers 字典不变。它为模块中未绑定的张量添加新的（缺失的）键
到 untied_parameters_and_buffers。新键的值是用户提供的值
在原始的 parameters_and_buffers 字典中。

如果对于同一个绑定的张量有多个用户提供的值，将会引发错误。

例如，如果模块有两个绑定的权重 self.foo 和 self.tied_foo，并且用户传递了
{'foo': foo_value, ...}，这将返回 {'foo': foo_value, 'tied_foo': foo_value, ...}。如果
用户传递 {'foo': foo_value, 'tied_foo': tied_foo_value, ...}，将引发错误。
用户传递 {'foo': foo_value, 'tied_foo': foo_value, ...}，不会引发错误。

    Args:
模块（torch.nn.Module）：确定哪些张量是绑定的模块。
parameters_and_buffers (Dict[str, Tensor])：用于重新参数化模块的 {name: tensor} 映射。

返回：
参数和缓冲区字典的新未绑定版本。

抛出异常：
ValueError：如果对于同一个绑定的张量有多个用户提供的值。
"源代码"
    模块中所有张量（包括绑定的张量）的{name: tensor}映射。
    所有命名张量: 字典[字符串, 张量] = {}
    所有命名张量.更新(模块.命名参数。(删除重复项=错误))
    所有命名张量.更新(模块.命名缓冲区(删除重复项=错误))

    # 模块中所有张量名称的映射：{张量：all_tied_names 的集合}
    张量到绑定名称映射: 字典[张量, 集合[字符串]] = {}
    为 名称, 张量 在 所有命名张量.项目():
        如果 张量 not 在 张量到绑定名称映射:
            张量到绑定名称映射[张量] = 集合()
        张量到绑定名称映射[张量].添加(名称)

    # 模块中所有张量名称的 {绑定名称: all_tied_names_set} 映射。
    # 如果一个名称未绑定，则不会出现在此映射中。
    tied_names_map: 字典[字符串, 集合[字符串]] = {}
    为 tied_names 在 张量到绑定名称映射.值():
        如果 长度(绑定名称) > 1:
            为 绑定名称 在 绑定名称:
                绑定名称映射[绑定名称] = 绑定名称列表

    确保用户没有为相同的绑定张量传递多个值。
    给定名称 = 集合(参数和缓冲区.键())
    # 与 given_names.intersection(tied_names_map.keys()) 相同，但 dynamo 无法处理
    # 处理该问题
    给定张量的名称: 集合[字符串] = 集合()
    为 名称 在 给定名称:
        如果 名称 在 绑定名称映射:
            给定张量的名称.添加(名称)

    为 给名 在 用于绑定张量的给定名称:
        绑定名称 = 绑定名称映射[given_name]
        如果 (
            检测是否存在同一张张量上的多个键。
            长度(tied_names.交集(给定的张量名称)) > 1
            # 仅当用户为同一关联张量传递多个值时才引发错误。
            # 如果所有给定的值都相同，则不引发错误。
            和 长度({参数和缓冲区[绑定名称] 为 绑定名称 在 绑定名称列表})
            != 1
        ):
            提升 ValueError(
                f"功能调用为键提供了多个值"{排序(绑定的名称)}，"
                f"它们是绑定的。考虑使用 tie_weights=False"
            )

    # 解除给定命名张量映射的绑定
    # 创建一个副本以不修改原始字典
    解绑参数和缓冲区 = 参数和缓冲区.复制()
    为 给定名称 在 给定名称的绑定张量:
        为 绑定名称 在 绑定名称映射[给定名称]：
            未绑定参数和缓冲区[绑定名称] = 参数和缓冲区[
                给定名称
            ]
    返回 解绑参数和缓冲区


@contextlib.contextmanager
def 重新参数化模块(
    模块: torch.nn.Module,
    参数和缓冲区: 字典[字符串, 张量]
    绑定权重: 布尔类型 = 错误,
    严格的: 布尔类型 = 错误,
    栈权重: 布尔类型 = 错误,
):
    参数和缓冲区 = 参数和缓冲区
    栈权重 = 栈权重

    如果 绑定权重:
        未绑定的参数和缓冲区 = _untie_named_tensors_map(
            模块, 参数和缓冲区
        )
    否则:
        解绑参数和缓冲区 = 参数和缓冲区

    访问器 = 命名成员访问器(模块)
    如果 严格的:
        缺少键, 预期之外的键 = 访问器.检查键(
            未绑定的参数和缓冲区
        )
        错误消息 = []
        如果 长度(预期之外的键) > 0:
            错误信息.追加(
                f"意外的键："{“，”.加入(地图(表示, 预期之外的键))}。
            )
        如果 长度(缺少键) > 0:
            错误信息.追加(f"缺少的键："{“，”.加入(地图(表示, 缺少键))}.")
        如果 长度(错误信息) > 0:
            提升 运行时错误(
                "重新参数化时出错："{}:\n\t{}".格式(
                    模块._获取名称(), "\n\t".加入(错误信息)
                )
            )

    原始参数和缓冲区: 字典[字符串, 张量] = {}
    try:
        原始参数和缓冲区, _ = 访问器.交换张量字典(
            解绑参数和缓冲区, 允许缺失=真实
        )
        产生
    最后:
        如果 栈权重:
            当启用堆叠时，我们将按后进先出（LIFO）的顺序恢复权重。
            原始参数和缓冲区 = 字典(
                反转(原始参数和缓冲区.项目())
            )
        新参数和缓冲区, _ = 访问器.交换张量字典(
            原始参数和缓冲区, 允许缺失=真实
        )
        有时该模块并非完全无状态，并对 _parameters 和 _buffers 字典进行了一些原地修改。
        将更改后的参数和缓冲区写回原始字典。
        将更改后的参数和缓冲区写回原始字典。
        parameters_and_buffers.更新(
            {
                k: 新参数和缓冲区[k]
                为 k 在 参数和缓冲区
                如果 k 在 新参数和缓冲区
            }
        )


@deprecated(
    "自 PyTorch 2.0 起，`torch.nn.utils.stateless.functional_call` 已弃用"
    "将在未来的 PyTorch 版本中删除。"
    "请使用 `torch.func.functional_call` 代替，这是一个直接替换。",
    分类=未来警告,
)
def functional_call(
    模块: "torch.nn.Module",
    参数和缓冲区: 字典[字符串, 张量]
    参数: 可选[联盟[任意, 元组]] = 无,
    kwargs: 可选[字典[字符串, 任意]] = 无,
    *,
    绑定权重: 布尔类型 = True,
    严格的: 布尔类型 = 错误,
):
    r通过用提供的参数和缓冲区替换模块参数和缓冲区，在模块上执行功能调用。

.. 警告::

该 API 自 PyTorch 2.0 起已弃用，将在未来的版本中删除。
PyTorch 版本。请使用:func:`torch.func.functional_call`代替，
这是对该 API 的直接替换。

.. note:: 如果该模块有活动的参数化，请使用名称设置为常规参数的
attr:`parameters_and_buffers`参数传递一个值。
名称将完全禁用参数化。
如果您想将参数化函数应用于传递的值
请将键设置为 ``{submodule_name}.parametrizations.{parameter_name}.original``。

.. 注意:: 如果模块在参数/缓冲区上执行原地操作，这些操作将反映出来
在`parameters_and_buffers`输入中。

示例::

            >>> a = {'foo': torch.zeros(())}
            >>> # xdoctest: +SKIP
>>> mod = Foo()  # 对象 self 中的 foo 属性自增 1
>>> print(mod.foo)  # 索引器(0.)
            >>> functional_call(mod, a, torch.ones(()))
            >>> print(mod.foo)  # tensor(0.)
            >>> print(a['foo'])  # tensor(1.)

.. note:: 如果模块具有绑定权重，functional_call 是否尊重绑定由
tie_weights 标志。

示例::

            >>> a = {'foo': torch.zeros(())}
            >>> # xdoctest: +SKIP
            >>> mod = Foo()  # has both self.foo and self.foo_tied which are tied. Returns x + self.foo + self.foo_tied
>>> print(mod.foo)  # 矩阵(1.)
>>> mod(torch.zeros(()))  # 矩阵(2.)
>>> functional_call(mod, a, torch.zeros(()))  # 矩阵(0.) 因为它也会改变 self.foo_tied
>>> functional_call(mod, a, torch.zeros(()), tie_weights=False)  # 矩阵(1.)--self.foo_tied 不会被更新
            >>> new_a = {'foo': torch.zeros(()), 'foo_tied': torch.zeros(())}
>>> functional_call(mod, new_a, torch.zeros()) # 矩阵(0.)

    Args:
模块 (torch.nn.Module)：要调用的模块
parameters_and_buffers (字符串和 Tensor 的字典)：将要使用的参数
模块调用。
args (Any 或元组): 传递给模块调用的参数。如果不是元组，则视为单个参数。
kwargs (字典): 传递给模块调用的关键字参数
tie_weights (bool, 可选): 如果为 True，则原始模型中绑定的参数和缓冲区在重新参数化版本中将被视为
绑定。因此，如果为 True 且传递了不同的值，则
参数和缓冲区，将出错。如果为 False，则不会尊重原本绑定的参数和
除非传递给两个权重的值相同，否则不缓冲。默认：True。
strict (bool, 可选): 如果为 True，则传入的参数和缓冲区必须与参数和
原始模块中的缓冲区。因此，如果为 True 并且存在任何缺失或意外的键，它将
错误。默认：False。

返回：
Any：调用 `module` 的结果。
"源代码"
    返回 _functional_call(
        模块,
        参数和缓冲区,
        参数,
        kwargs,
        绑定权重=绑定权重,
        严格的=严格的,
    )



def _功能调用(
    模块: torch.nn.Module,
    参数和缓冲区: 字典[字符串, 张量]
    参数: 可选[联盟[任意, 元组]] = 无,
    kwargs: 可选[字典[字符串, 任意]] = 无,
    *,
    绑定权重: 布尔类型 = True,
    严格的: 布尔类型 = 错误,
):
    # TODO 允许使用 kwargs 参数，如 unsafe 和其他参数进行配置
    如果 (
        火炬.算子.is_tracing()
        或 火炬.算子.是否正在脚本化()
        或 isinstance(
            模块,
            (
                火炬.算子.递归脚本模块,
                火炬.算子.脚本模块,
                火炬.算子.脚本函数,
            ),
        )
    ):
        提升 运行时错误(无状态 API 无法与 Jitted 模块一起使用)
    如果 isinstance(模块, 火炬.nn.数据并行):
        提升 运行时错误(
            无状态 API 无法与 nn.DataParallel 模块一起使用
        )
    如果 kwargs 是 无:
        kwargs = {}
    如果 args 是 无:
        args = ()
    elif not isinstance(参数, 元组):
        args = (参数,)
    与 _reparametrize_module(
        模块, 参数和缓冲区, 绑定权重=绑定权重, 严格的=严格的
    ):
        返回 模块(*参数, **kwargs)