# mypy: 允许未类型化定义
导入 contextlib
导入 functools
导入
检查
导入
正则表达式
导入
系统
导入
跟踪回溯
导入
弱引用
from collections.abc 导入
序列
from 打字
导入 (
任何,
可调用,
直接,
可选,
过载,
类型检查,
类型变量,
联合,
)
from typing_extensions 导入
已弃用,
参数规范
导入
火炬
导入
torch._库 as
库
from torch._库.custom_ops
导入 (
_类型转换,
可能获取操作定义,
自定义操作,
自定义运算定义,
设备类型_t,
)
from torch._library.infer_schema 导入 infer_schema # noqa: F401
from torch._library.triton 导入
triton 操作,
triton 包装
from torch._ops 导入
运算符重载
from torch 的类型
导入 _dtype
全部 = [
图书馆,
实现,
定义,
"fallthrough_kernel",
"impl_abstract",
"register_autocast",
"register_fake",
注册 torch 调度,
注册 vmap,
获取上下文,
自定义操作,
triton 操作,
wrap_triton,
推理模式,
]
_T = 类型变量("_T")
_P = 参数规范("_P")
# 为新注册的内核设置的（命名空间，操作符，DispatchKey）组合集合
# 集合中的键的形式为 `namespace + "/" + op_name + "/" + dispatch_key`。
# 维护此集合是为了确保两个库不会尝试覆盖完全相同的功能，以避免
调用非预期被调用的内核的库。
_impls: 设置[str] =
设置()
_定义:
设置[str] =
设置()
# prim 是由 TorchScript 解释器保留的。
_保留命名空间 = [
prim]
[文档]def fallthrough_kernel():
"""
用于注册跳转的示例函数，传递给 `Library.impl`
""
抛出未实现异常("fallthrough_kernel() 不应该被调用。")
[文档]
类
库:
""
用于创建库的类，这些库可以用来注册新的操作符或
在现有的 Python 库中重载运算符。
用户可以选择传递一个调度键名，如果他们只想注册对应一个特定调度键的内核。
要创建一个库以重载现有库（名称为 ns）中的运算符，请将类型设置为"IMPL"。
To create a library to override operators in an existing library (with name ns), set the kind to "IMPL".
创建一个新的库（名称为 ns）以注册新操作员，将类型设置为"DEF"。
要创建可能存在的库的片段以注册操作员（并绕过
给定命名空间只有一个库的限制），将类型设置为
"FRAGMENT"。
参数：
库名称
类型: "DEF", "IMPL"（默认: "IMPL"）, "FRAGMENT"
调度键: PyTorch 调度键（默认: ""）
"""
定义 __init__(self, ns,
类型,
分发键=
输入文本翻译为简体中文为：""):
如果
仁慈 not
进入 (
实现, "DEF",
"碎片"):
抛出
值错误(
"不支持的类型：",
类型)
如果
命名空间
进入
_保留命名空间
以及 (
仁慈 == "DEF"
或者
仁慈 ==
"碎片"):
抛出
值错误(
ns,
"是一个保留的命名空间。请尝试使用其他名称创建库。",
)
如果
火炬.
_运行时使用部署():
库.
工具.
警告部署()
返回
框架 =
跟踪回溯.
提取堆栈(
限制=3
)]0]
文件名,
行号 = frame.
文件名, frame.
行号
self.m: 可选[
任何] =
火炬._C.
_调度库(
类型, ns,
分发键,
文件名,
行号
)
self.命名空间 =
命名空间
self._操作定义:
设置[str] =
设置()
self._op_impls: 设置[str] =
设置()
self._注册处理句柄:
列表[
火炬.
库.
工具.
注册句柄] =
输入文本为空，请提供需要翻译的文本
self.仁慈 =
仁慈
self.分发键 =
分发键
使用终结器设置“析构函数”而不是 __del__。
Python 中的 __del__ 可能会导致奇怪的事情发生（全局变量和局部变量可能在 __del__ 实际被调用时已经消失！）。
终结器有助于这种情况，因为它让我们捕获引用并保持它们存活。
# 最终化器有助于这种情况，因为它让我们捕获引用并保持它们存活。
弱引用.
完成(
self,
_del_library,
_impls,
self._op_impls,
_定义,
self._操作定义,
self._注册处理句柄,
)
定义 __repr__(self):
返回 f
"库(类型={self.
类型}
, 命名空间={self.ns}
,调度键={self.
分发键})>"
[文档]
定义
定义(self,
架构,
别名分析=
输入文本翻译为简体中文为："", *,
标签=()):
r在 ns 命名空间中定义了一个新操作符及其语义。
参数：
schema：定义新操作符的函数 schema。
别名分析（可选）：指示操作符参数的别名属性是否可以
从模式推断（默认行为）或不可推断（"CONSERVATIVE"）。
标签（Tag | Sequence[Tag]）：应用于此的一个或多个 torch.Tag
标记操作符会改变操作符在各种 PyTorch 子系统中的行为
请在应用 torch.Tag 之前仔细阅读文档。
请仔细阅读 torch.Tag 文档后再应用。
返回值：
从模式推断出的算子名称。
示例::
>>> my_lib = Library("mylib", "DEF")
>>> my_lib.define("sum(Tensor self) -> Tensor")
"""
如果
火炬.
_运行时使用部署():
库.
工具.
警告部署()
返回
# 这是因为我们还想禁止 PURE_FUNCTION 别名分析，这是一个有效的
# C++中的 AliasAnalysis 类型
如果
别名分析 not
进入 [
输入文本翻译为简体中文为："",
FROM_SCHEMA,
保守的
]
抛出 RuntimeError(f
无效的别名分析类型{
别名分析}")
断言 self.m
是 not
无
如果 isinstance(
标签,
火炬.
标签):
标签 = (
标签,)
名称 =
架构.
分割(
“（”
)]0]
数据包名称 =
姓名.
分割(
“。”
)]0]
如果 "."
进入
名称
否则
名称
已存在数据包 =
有属性(
火炬.
操作, self.ns)
以及
有属性(
getattr(火炬.
操作, self.ns),
数据包名称
)
结果 = self.m.
定义(
架构,
别名分析,
元组(
标签))
名称 =
架构.
分割(
“（”
)]0]
qualname = self.命名空间 +
“::” +
名称
如果 OpOverloadPacket 已经存在，那么这意味着我们在添加
为它添加新的 OpOverload。刷新数据包以包含新的 OpOverload。
如果
已存在数据包:
命名空间 = getattr(
火炬.
操作, self.ns)
数据包 = getattr(ns,
数据包名称)
火炬.
操作符.
_刷新包(
数据包)
self._操作定义.
添加(qualname)
_定义.
添加(qualname)
返回
结果
定义
注册模拟实现(self,
操作符名称, fn, _stacklevel=1):
r注册库中定义的操作符的模拟实现。
如果
火炬.
_运行时使用部署():
库.
工具.
警告部署()
返回
源 =
火炬.
库.
工具.
获取源(
_堆栈级别 + 1)
框架 =
系统模块._getframe(_stacklevel)
caller_module = 检查.
获取模块(frame)
# 可能为空，如果从没有模块的地方调用 register_fake
# （例如 __main__）
调用模块名称 =
无
如果 caller_module
是
无
否则 caller_module.__name__
# TODO(rzou): 我们需要将这个更改与 torchvision 一起进行阶段化，
# 因为 torchvision 是 GitHub 优先。
如果
调用模块名称
是 not
无
以及
调用模块名称.
以...开头(
"torchvision."
):
调用模块名称 =
无
qualname = f"{self.ns}::{操作符名称}"
条目 =
火炬.
库.
简单注册表.
单例.
找到(qualname)
如果
调用模块名称
是 not
无:
注册函数 = _check_pystubs_once(fn, qualname,
调用模块名称)
否则:
注册函数 = fn
handle = 条目.
模拟实现.
注册(
注册函数,
源)
self._注册处理句柄.
添加(
处理)
定义
_注册 torch 调度规则(self,
操作符名称,
torch 调度类, fn):
r注册给定操作符和 torch_dispatch_class 的 torch_dispatch 规则。
这允许开放注册以指定操作员之间的行为
无需修改 torch_dispatch_class 即可使用 torch_dispatch_class
或直接操作员。
torch_dispatch_class 是一个具有 `__torch_dispatch__` 的 Tensor 子类或
TorchDispatchMode.
如果它是一个 Tensor 子类，我们期望 fn 具有以下签名：
(cls, func: OpOverload, types: Tuple[type, ...], args, kwargs) -> Any
如果它是一个 TorchDispatchMode，我们期望 fn 具有以下签名：
(mode, func: OpOverload, types: Tuple[type, ...], args, kwargs) -> Any
(模式，func: 操作重载，types: 类型元组[type, ...]，args，kwargs) -> 任意
"""
如果
火炬.
_运行时使用部署():
库.
工具.
警告部署()
返回
qualname = f"{self.ns}::{操作符名称}"
条目 =
火炬.
库.
简单注册表.
单例.
找到(qualname)
handle = 条目.
PyTorch 调度规则.
注册(
torch 调度类, fn)
self._注册处理句柄.
添加(
处理)
定义
_使用 AOT 编译实现(self,
操作符名称,
分发键=
输入文本翻译为简体中文为：""):
r注册操作符以使用 AOTI 编译的实现。
参数：
操作名称：操作符名称（包括重载）或 OpOverload 对象。
dispatch_key: 输入函数应注册的 dispatch key。默认情况下，它使用
创建库时使用的 dispatch key。
示例::
>>> my_lib = Library("aten", "IMPL")
>>> my_lib._impl_with_aoti_compile("div.Tensor", "CPU")
"""
如果
火炬.
_运行时使用部署():
库.
工具.
警告部署()
返回
如果
分发键 ==
输入文本翻译为简体中文为："":
分发键 = self.
分发键
断言
火炬.
发送键集(
分发键).
有(
火炬._C.
发送键.
稠密)
如果 isinstance(
操作符名称, str):
名称 =
操作名称
elif isinstance(操作符名称,
操作符重载):
名称 =
操作符名称.
_模式.
名称
重载名称 =
操作符名称.
_模式.
重载名称
如果
重载名称 !=
输入文本翻译为简体中文为："":
名称 =
名称 + "." +
重载名称
否则:
抛出 RuntimeError(
"_impl_with_aoti_compile 应传递一个名称或一个 OpOverload 对象"
作为第一个参数
)
键 = self.
命名空间 +
根目录 +
姓名.
分割(
双冒号
)]-1] +
根目录 +
分发键
如果
键
进入 _impls:
# TODO: 在未来，添加更多关于现有函数注册位置的信息（此信息是
# 在调用 _impl_with_aoti_compile 时已由 C++ 警告返回，但我们在此之前出错）
抛出 RuntimeError(
"这不被允许，因为已经有一个从 Python 注册的内核覆盖了{}"
的行为了{}
发送键和{}
命名空间。".
格式(
姓名.
分割(
双冒号
)]-1
]
分发键, self.
命名空间
)
)
断言 self.m
是 not
无
实现函数:
可调用 = self.m.
实现带 AOT 编译
实现函数(self.ns,
姓名.
分割(
双冒号
)]-1
]
分发键)
_impls.添加(
键)
self._op_impls.添加(
键)
[文档]
定义
实现(self,
操作符名称, fn,
分发键=
输入文本翻译为简体中文为："", *,
带有密钥集=False):
r注册库中定义的操作符的功能实现。
参数：
操作名称：操作符名称（包括重载）或 OpOverload 对象。
fn: 输入分发键的运算符实现函数或：func:`~fallthrough_kernel`
注册一个跳转。
dispatch_key: 输入函数应注册的 dispatch key。默认情况下，它使用
创建库时使用的 dispatch key。
with_keyset: 控制当前调度器调用 keyset 是否应作为第一个参数传递的标志
在调用时使用：attr:`fn`。这应该用于创建适当的键集以进行重派调用。
示例::
>>> my_lib = Library("aten", "IMPL")
>>> def div_cpu(self, other):
>>> 返回 self * (1 / other)
>>> my_lib.impl("div.Tensor", div_cpu, "CPU")
"""
如果
火炬.
_运行时使用部署():
库.
工具.
警告部署()
返回
如果 not
可调用(fn):
抛出
类型错误(
f"输入函数必须是一个可调用对象，但找到的类型为"{
类型(fn)}"
)
如果
分发键 ==
输入文本翻译为简体中文为："":
分发键 = self.
分发键
如果 isinstance(
操作符名称, str):
名称 =
操作名称
elif isinstance(操作符名称,
操作符重载):
名称 =
操作符名称.
_模式.
名称
重载名称 =
操作符名称.
_模式.
重载名称
如果
重载名称 !=
输入文本翻译为简体中文为："":
名称 =
名称 + "." +
重载名称
否则:
抛出 RuntimeError(
"实现应该将名称或 OpOverload 对象作为第一个参数传递"
)
键 = self.
命名空间 +
根目录 +
姓名.
分割(
双冒号
)]-1] +
根目录 +
分发键
如果
键
进入 _impls:
# TODO: 在未来，添加更多关于现有函数注册位置的信息（此信息是
# 当前的 C++警告在调用 impl 时返回的，但我们在此之前就出错）
抛出 RuntimeError(
"这不被允许，因为已经有一个从 Python 注册的内核覆盖了{}"
的行为了{}
发送键和{}
命名空间。".
格式(
姓名.
分割(
双冒号
)]-1
]
分发键, self.
命名空间
)
)
如果
分发键 ==
元数据:
分发器操作名称 =
名称
如果
“::” not
进入 dispatcher_op_name:
分发器操作名称 = f"{self.ns}::{dispatcher_op_name}"
我们内部不应该为任何具有复合隐式自动微分核的算子注册元核。
相反，我们应该让这些分解运行，并且只为基本算子编写元核。
而不是，我们应该让那些分解运行，并且只为基本算子编写元核。
只为基本算子编写元核。
如果
火炬._C._dispatch_has_kernel_for_dispatch_key(
dispatcher_op_name, "复合隐式自动微分"
):
抛出 RuntimeError(
f"我们不应直接将元内核注册到算子'"{
姓名}
'，'
"因为它在核心中有一个 CompositeImplicitAutograd 内核。"
"相反，我们应该让操作符分解，并确保我们有元内核"
"用于它分解成的基操作。"
)
断言 self.m
是 not
无
self.m.实现(
姓名,
分发键
如果
分发键 !=
请提供需要翻译的文本
否则
复合隐式自动微分,
fn,
带有密钥集,
)
_impls.添加(
键)
self._op_impls.添加(
键)
[文档] def fallback(self, fn, dispatch_key="", *, with_keyset=False):
r"""将函数实现注册为给定键的回退函数。"""
此函数仅适用于具有全局命名空间（"_"）的库。
参数:
fn: 作为给定分发键或 :func:`~fallthrough_kernel` 的后备函数
注册一个穿透。
输入函数应注册的 dispatch key。默认情况下，它使用
分配给库创建时的按键。
with_keyset: 控制当前分发器调用是否将键集作为第一个参数传递的标志
在调用时使用 `to :attr:`fn`。这应该用于创建适当的键集以进行重定向调用。
示例::
>>> my_lib = Library("_", "IMPL")
>>> def fallback_kernel(op, *args, **kwargs):
>>> # 处理所有自动类型转换操作
>>> # ...
>>> my_lib.fallback(fallback_kernel, "Autocast")
"""
if torch._running_with_deploy():
_library.utils.warn_deploy()
return
如果 dispatch_key == ""
dispatch_key = self.dispatch_key
if self.ns != "_":
raise RuntimeError(
Fallback 只能使用全局命名空间"_"上的库片段进行注册，但它是{self.ns}
)
断言 dispatch_key 不为空字符串
断言 self.m 不为 None
self.m.fallback(dispatch_key, fn, with_keyset)
定义 _destroy(self):
如果 self.m
是 not
无:
self.m.重置()
self.m = 无
对于 handle
进入 self.
_注册处理句柄:
处理.
摧毁()
self._注册处理句柄.
清晰()
全局 _impls
_impls -= self._op_impls
对于
名称
进入 self.
_操作定义:
# 删除已注册的 torch.ops.ns.foo 缓存。
否则，访问它会导致段错误。
有可能我们只在这个库中注册了一个重载。
另一个库拥有一个活动的重载。
没关系 - 下次调用 torch.ops.ns.foo 时，它将会是
# recomputed to point at the right collection of overloads.
ns, 带重载的名称 =
姓名.
分割(
双冒号)
名称 = name_with_overload.
分割(
“。”
)]0]
如果 not
有属性(
火炬.
操作, ns):
continue
命名空间 = getattr(
火炬.
操作, ns)
如果 not
有属性(
命名空间,
姓名):
continue
delattr(命名空间,
姓名)
命名空间._dir.
删除(
姓名)
定义 _del_library(
捕获实现,
操作实现,
捕获定义,
操作定义,
注册句柄,
):
捕获实现 -=
操作实现
捕获定义 -=
操作定义
对于 handle
进入
注册句柄:
处理.
摧毁()
@contextlib.contextmanager
定义
_scoped_library
（无翻译内容）(*参数, **
关键字参数):
尝试:
库 =
库(*
参数, **
关键字参数)
产生
库
最后:
库._destroy()
_保持活跃:
列表[
库] =
输入文本为空，请提供需要翻译的文本
NAMELESS_SCHEMA = 正则表达式.
编译(r
\(.*\) -> .*)
[文档]@functools.
单例分发
定义
定义(qualname,
架构, *,
库=
无,
标签=()):
r定义一个新的运算符。
在 PyTorch 中，定义一个运算符（简称“op”）是一个两步过程：
- 我们需要定义运算符（通过提供运算符名称和模式）
- 我们需要实现运算符如何与其他元素交互的行为
各种 PyTorch 子系统，如 CPU/CUDA 张量、Autograd 等。
此入口定义自定义操作（第一步）
然后您必须通过调用各种
``impl_*`` API，如 :func:`torch.library.impl` 或
`torch.library.register_fake`.
参数：
qualname (str): 操作符的限定名称。应为
看起来像 "命名空间::名称" 的字符串，例如 "aten::sin"。
PyTorch 中的操作符需要一个命名空间来
避免名称冲突；一个操作符只能创建一次。
如果您正在编写一个 Python 库，我们建议命名空间为
您顶级模块的名称。
schema (str): 操作符的模式。例如：(Tensor x) -> Tensor
用于接受一个张量并返回一个张量的操作。它不包含操作名称（该名称通过 `qualname` 传入）。
不包含操作名称（即通过 `qualname` 传入的操作名称）。
lib（可选[Library]）：如果提供，则此操作的生命周期将与 Library 对象的生命周期绑定。
将会绑定到库对象的生命周期。
标签（Tag | Sequence[Tag]）：应用于此的一个或多个 torch.Tag
标记操作符会改变操作符在各种 PyTorch 子系统中的行为
请在应用 torch.Tag 之前仔细阅读文档。
请仔细阅读 torch.Tag 文档后再应用。
示例::
>>> 导入 torch
>>> import numpy as np
...
>>> # 定义操作符
>>> torch.library.define("mylib::sin", "(Tensor x) -> Tensor")
...
>>> # 添加操作符的实现
>>> @torch.library.impl("mylib::sin", "cpu")
>>> def f(x):
>>> return torch.from_numpy(np.sin(x.numpy()))
...
>>> # 从 torch.ops 调用新操作符。
>>> x = torch.randn(3)
>>> y = torch.ops.mylib.sin(x)
>>> assert torch.allclose(y, x.sin())
"""
如果 not isinstance(qualname, str):
抛出
值错误(
f"define(qualname, schema): 预期 qualname "
f"为 str 的实例，却得到了 "{
类型(qualname)}"
)
命名空间,
名称 =
火炬.
库.
工具.
解析命名空间(qualname)
如果
库
是
无:
库 =
库(
命名空间,
"碎片")
_保持活跃.
添加(
库)
如果 not
无名称模式.
完全匹配(
架构):
抛出
值错误(
f"define(qualname, 模式, ...): 预期模式 "
f'看起来像 e.g. "(Tensor x) -> Tensor" 但 '
f'得到 '{
架构}
''
)
库.
定义(
名称 +
架构,
别名分析=
输入文本翻译为简体中文为："",
标签=
标签)
@define.注册
定义 _(
库:
库,
架构,
别名分析=
输入文本翻译为简体中文为：""):
"""旧版的 torch.library.define.
我们保留这部分是为了向后兼容的原因
"""
定义
包裹(f):
名称 =
库.
定义(
架构,
别名分析)
库.
实现(
姓名, f)
返回 f
返回
包装
@overload
定义
实现(
qualname: str,
类型:
联合[str,
序列[str]],
函数:
直接[
无] =
无,
*,
库:
可选[
库] =
无,
) -> 可调用[[
可调用[...,
对象]],
无
] ...
@overload
定义
实现(
qualname: str,
类型:
联合[str,
序列[str]],
函数:
可调用[...,
对象
]
*,
库:
可选[
库] =
无,
) -> 无: ...
# 已废弃的 BC API
@overload
定义
实现(
库:
库,
姓名: str,
分发键:
字符串 =
输入文本翻译为简体中文为："",
) -> 可调用[[
可调用[_P, _T]],
可调用[_P, _T
]] ...
[文档]@functools.
单例分发
定义
实现(
qualname: str,
类型:
联合[str,
序列[str]],
函数:
可选[
可调用[_P, _T]] =
无,
*,
库:
可选[
库] =
无,
) -> 对象:
为此操作员注册设备类型的实现。
您可以将 "default" 传递给 ``types`` 以将此实现注册为所有设备类型的默认实现。
请仅在此实现真正支持所有设备类型时使用。
请仅在此实现真正支持所有设备类型时使用。
例如，如果它是内置 PyTorch 运算符的组合，则此条件为真。
此 API 可以用作装饰器。您可以使用嵌套装饰器
只要它们返回一个函数并且放在此 API 内部
（请参阅示例 2）。
一些有效的类型包括："cpu"、"cuda"、"xla"、"mps"、"ipu"、"xpu"。
参数：
qualname（字符串）：应是一个类似于 "namespace::operator_name" 的字符串。
types（字符串 | 字符串序列）：要将 impl 注册到的设备类型。
lib（可选[库]）：如果提供，则此注册的生命周期
将会绑定到库对象的生命周期。
示例：
>>> 导入 torch
>>> import numpy as np
>>> # 示例 1：注册函数。
>>> # 定义操作符
>>> torch.library.define("mylib::mysin", "(Tensor x) -> Tensor")
...
>>> # 为 cpu 设备添加实现
>>> @torch.library.impl("mylib::mysin", "cpu")
>>> def f(x):
>>> return torch.from_numpy(np.sin(x.numpy()))
...
>>> x = torch.randn(3)
>>> y = torch.ops.mylib.mysin(x)
>>> assert torch.allclose(y, x.sin())
...
>>> # 示例 2：使用装饰器注册函数。
>>> def custom_decorator(func):
>>> def wrapper(*args, **kwargs):
>>> 返回 func(*args, **kwargs) + 1
>>> 返回 wrapper
...
>>> # 定义操作符
>>> torch.library.define("mylib::sin_plus_one", "(Tensor x) -> Tensor")
...
>>> # 添加操作符的实现
>>> @torch.library.impl("mylib::sin_plus_one", "cpu")
>>> @custom_decorator
>>> def f(x):
>>> return torch.from_numpy(np.sin(x.numpy()))
...
>>> # 从 torch.ops 调用新操作符。
>>> x = torch.randn(3)
...
>>> y1 = torch.ops.mylib.sin_plus_one(x)
>>> y2 = torch.sin(x) + 1
>>> assert torch.allclose(y1, y2)
"""
返回
实现(qualname,
类型,
函数,
库=
库,
禁用 Dynamo=False)
如果 not
类型检查:
@impl.注册
定义 _(
库:
库,
姓名: str,
分发键:
字符串 =
请提供需要翻译的文本
) -> 可调用[[
可调用[_P, _T]],
可调用[_P, _T
]]
旧版 torch.library.impl API。保留以供向后兼容
定义
包裹(f:
可调用[_P, _T]) ->
可调用[_P, _T
]
库.
实现(
姓名, f,
分发键)
返回 f
返回
包装
@overload
定义
实现(
qualname: str,
类型:
联合[str,
序列[str]],
函数:
直接[
无] =
无,
*,
库:
可选[
库] =
无,
禁用 Dynamo:
布尔值 = False,
) -> 可调用[[
可调用[...,
对象]],
无
] ...
@overload
定义
实现(
qualname: str,
类型:
联合[str,
序列[str]],
函数:
可调用[...,
对象
]
*,
库:
可选[
库] =
无,
禁用 Dynamo:
布尔值 = False,
) -> 无: ...
定义
实现(
qualname: str,
类型:
联合[str,
序列[str]],
函数:
可选[
可调用[...,
对象]] =
无,
*,
库:
可选[
库] =
无,
禁用 Dynamo:
布尔值 = False,
) -> 可选[
可调用[[
可调用[...,
对象]],
无
]]
# See impl()
如果 isinstance(
类型, str):
类型 = (
类型,)
键 =
设置({})
对于
类型
进入
类型:
is_dispatch_key = 火炬._C._parse_dispatch_key(typ)
如果
是否为分发键:
我们还支持传递一个 DispatchKey 给 impl。请优先使用更高层次的 torch.library APIs，并且只将 DispatchKey 传递给 torch.library.impl，请谨慎使用（或者更好的选择，不要使用这个选项，并在 GitHub 上提交一个关于你需要解决的问题的 issue）。
# the higher-level torch.library APIs and only pass DispatchKey to
# torch.library.impl with caution (or even better, don't use this
# option and file an issue on GitHub for what you need).
我们不向用户宣传这一点，因为
# 很容易自己打自己的脚。
键.
添加(typ)
否则:
键.
添加(
_设备类型转键(typ))
定义
注册_(
函数:
可调用[...,
对象]) ->
无:
命名空间, _ =
火炬.
库.
工具.
解析命名空间(qualname)
如果
库
是
无:
使用库 =
库(
命名空间,
"碎片")
_保持活跃.
添加(
使用库)
否则:
使用库 =
库
如果
禁用 Dynamo:
@torch.禁用 dynamo
定义
无 Dynamo 功能(*
参数, **
关键字参数):
返回
函数(*
参数, **
关键字参数)
对于
键
进入
键:
使用库.
实现(qualname,
无 Dynamo 功能,
键)
否则:
对于
键
进入
键:
使用库.
实现(qualname,
函数,
键)
如果
函数
是
无:
返回
注册_
否则:
注册_(
函数)
返回
无
定义
_设备类型转键(
设备类型: str) -> str:
如果
设备类型 ==
默认:
这不是技术上的正确做法，因为尽管所有设备类型
# 处理键包含在 CompositeExplicitAutograd 中
# CompositeExplicitAutograd 中并非所有内容都与显式自动微分相关联
设备类型。我并不真的那么在乎这些区别。
返回
"复合显式自动微分"
返回
火炬._C.
_设备调度键(
设备类型)
[文档]@已弃用(
"``torch.library.impl_abstract`` 已重命名为 ``torch.library.register_fake``。请使用它 "
未来版本中，我们将移除 `torch.library.impl_abstract`。
category=未来警告，
)
def impl_abstract(qualname, func=None, *, lib=None, _stacklevel=1):
此 API 在 PyTorch 2.4 版本中更名为：func:`torch.library.register_fake`。
请使用它代替。
""
如果 func 不是 None：
_stacklevel = _stacklevel + 1
返回 register_fake(qualname, func, lib=lib, _stacklevel=_stacklevel)
_op_identifier = 联合[
str, "torch._ops.操作重载",
"torch._library.custom_ops.自定义操作定义"
]
[文档]
定义
注册内核(
操作:
_op 标识符,
设备类型:
设备类型_t,
函数:
可选[
可调用] =
无,
/,
*,
库:
可选[
库] =
无,
):
为此操作员注册设备类型的实现。
一些有效的设备类型包括："cpu"、"cuda"、"xla"、"mps"、"ipu"、"xpu"。
此 API 可以用作装饰器。
参数：
op (str | OpOverload): 要注册实现的运算符。
device_types (None | str | Sequence[str]): 要注册实现的设备类型。
如果为空，我们将注册到所有设备类型 -- 请仅使用
此选项适用于您的实现真正与设备类型无关。
func (Callable): 将注册为实现的函数
给定的设备类型。
lib（可选[库]）：如果提供，则此注册的生命周期
示例::
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA)
>>> 导入 torch
>>> 从 torch 导入 Tensor
>>> from torch.library import custom_op
>>> import numpy as np
...
>>> # 创建一个在 cpu 上工作的自定义操作
>>> @custom_op("mylib::numpy_sin", mutates_args=(), device_types="cpu")
>>> def numpy_sin(x: Tensor) -> Tensor:
>>> x_np = x.numpy()
>>> y_np = np.sin(x_np)
>>> return torch.from_numpy(y_np)
...
>>> # 添加对 cuda 设备的实现
>>> @torch.library.register_kernel("mylib::numpy_sin", "cuda")
>>> def _(x):
>>> x_np = x.cpu().numpy()
>>> y_np = np.sin(x_np)
>>> return torch.from_numpy(y_np).to(device=x.device)
...
>>> x_cpu = torch.randn(3)
>>> x_cuda = x_cpu.cuda()
>>> assert torch.allclose(numpy_sin(x_cpu), x_cpu.sin())
>>> assert torch.allclose(numpy_sin(x_cuda), x_cuda.sin())
"""
如果 not isinstance(
操作, (str,
火炬.
操作符.
操作符重载,
火炬.
库.
自定义运算.
自定义运算定义)
):
抛出
值错误(
f"注册内核（"{
操作}
): 预期外的类型对于 op:{
类型(
操作)}"
)
如果 isinstance(
操作,
火炬.
操作符.
操作符重载):
操作符 =
操作.
名称
操作定义 =
可能获取操作定义(
操作)
如果
操作定义
是 not
无:
返回
操作定义.
注册内核(
设备类型,
函数)
断言 isinstance(
操作, str)
如果
设备类型
是
无:
设备类型 =
"复合显式自动微分"
返回
实现(
操作,
设备类型,
函数,
库=
库,
禁用 Dynamo=True)
[文档]
定义
注册自动转换(
操作:
_op 标识符,
设备类型: str,
输入设备:
数据类型,
/,
*,
库:
可选[
库] =
无,
):
r注册此自定义操作的自动广播调度规则。
有效的 `device_type` 包括: "cpu" 和 "cuda"。
参数：
op (str | OpOverload): 要注册自动广播调度规则的运算符。
device_type(str): 要使用的设备类型。'cuda' 或 'cpu'。
类型与 :class:`torch.device` 的 `type` 属性相同。
因此，您可以使用 `Tensor.device.type` 获取张量的设备类型。
cast_inputs (:class:`torch.dtype`): 当自定义操作在自动类型转换区域运行时，
将传入的浮点型张量转换为目标数据类型（非浮点型张量）
这些不受影响），然后执行禁用自动转换的自定义操作。
lib（可选[库]）：如果提供，则此注册的生命周期
示例::
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA)
>>> 导入 torch
>>> 从 torch 导入 Tensor
>>> from torch.library import custom_op
...
>>> # 创建一个在 cuda 上工作的自定义操作
>>> @torch.library.custom_op("mylib::my_sin", mutates_args=())
>>> def my_sin(x: Tensor) -> Tensor:
>>> return torch.sin(x)
...
>>> # 注册针对 cuda 设备的 autocast 调度规则
>>> torch.library.register_autocast("mylib::my_sin", "cuda", torch.float16)
...
>>> x = torch.randn(3, dtype=torch.float32, device="cuda")
>>> with torch.autocast("cuda", dtype=torch.float16):
>>> y = torch.ops.mylib.my_sin(x)
>>> assert y.dtype == torch.float16
"""
如果 not isinstance(
操作, (str,
火炬.
操作符.
操作符重载,
火炬.
库.
自定义运算.
自定义运算定义)
):
抛出
值错误(
fregister_autocast({
操作}
): 预期外的类型对于 op:{
类型(
操作)}"
)
如果
设备类型 not
进入 ["cpu",
cuda
]
抛出
值错误(f
"未知设备类型："{
设备类型}")
如果 isinstance(
操作,
火炬.
操作符.
操作符重载):
操作符 =
操作.
名称
操作定义 =
可能获取操作定义(
操作)
如果
操作定义
是 not
无:
返回
操作定义.
注册自动转换(
设备类型,
输入设备)
断言 isinstance(
操作, str)
qualname = 操作符
_操作 =
火炬.
库.
工具.
查找操作(qualname)
命名空间,
操作名称 =
火炬.
库.
工具.
解析命名空间(qualname)
如果
库
是
无:
库 =
库(
命名空间,
"碎片")
_保持活跃.
添加(
库)
定义
内核(_, *
参数, **
关键字参数):
断言
长度(
关键字参数) == 0,
"自定义操作目前尚不支持 kwargs。"
autocast_keyset = 火炬._C.
发送键集(
火炬._C.
发送键.
自动转换 CPU
) | 火炬._C.
发送键集(
火炬._C.
发送键.
自动将 CUDA)
替换为
火炬._C.
_排除调度键保护(
自动铸件键集):
返回
_操作(*
_类型转换(
参数,
设备类型,
输入设备))
如果
设备类型 ==
cuda:
返回
库.
实现(
操作名称,
内核,
自动转换 CUDA,
带有密钥集=True)
否则:
# 设备类型为 "cpu"
返回
库.
实现(
操作名称,
内核,
自动转换 CPU,
带有密钥集=True)
[文档]
定义
注册伪造(
操作:
_op 标识符,
函数:
可选[
可调用] =
无,
/,
*,
库:
可选[
库] =
无,
_stacklevel: 整型 = 1,
):
r注册此运算符的 FakeTensor 实现（"fake impl"）。
也被称为“元内核”、“抽象实现”。
"FakeTensor 实现"指定了此运算符在
不携带数据（"FakeTensor"）的 Tensors 上的行为。
指定某些属性（大小/步长/存储偏移/设备），它指定了
输出张量的属性是什么。
FakeTensor 实现与操作符具有相同的签名。
它对 FakeTensors 和元张量都进行运行。为了编写一个 FakeTensor
实现，假设操作符的所有 Tensor 输入都是
正常的 CPU/CUDA/Meta 张量，但它们没有存储，并且
您尝试返回常规的 CPU/CUDA/Meta 张量作为输出。
FakeTensor 的实现必须仅由 PyTorch 操作组成
（并且不能直接访问任何输入或
中间张量）。
此 API 可以用作装饰器（请参阅示例）。
有关自定义操作的详细指南，请参阅
https://maskerprc.github.io/tutorials/advanced/custom_ops_landing_page.html
示例：
>>> 导入 torch
>>> import numpy as np
>>> 从 torch 导入 Tensor
...
>>> # 示例 1：一个没有数据依赖输出形状的操作符
>>> @torch.library.custom_op("mylib::custom_linear", mutates_args=())
>>> def custom_linear(x: Tensor, weight: Tensor, bias: Tensor) -> Tensor:
>>> raise NotImplementedError("实现将在此处进行")
...
>>> @torch.library.register_fake("mylib::custom_linear")
>>> def _(x, weight, bias):
>>> assert x.dim() == 2
>>> assert weight.dim() == 2
>>> 断言 bias 的维度等于 1
>>> 断言 x 的形状[1] 等于 weight 的形状[1]
>>> 断言 weight 的形状[0] 等于 bias 的形状[0]
>>> 断言 x 的设备等于 weight 的设备
...
>>> 返回 (x @ weight.t()) + bias
...
>>> 使用 torch._subclasses.fake_tensor.FakeTensorMode()
>>> x = torch.randn(2, 3)
>>> w = torch.randn(3, 3)
>>> b = torch.randn(3)
>>> y = torch.ops.mylib.custom_linear(x, w, b)
...
>>> assert y.shape == (2, 3)
...
>>> # Example 2: an operator with data-dependent output shape
>>> @torch.library.custom_op("mylib::custom_nonzero", mutates_args=())
>>> def custom_nonzero(x: Tensor) -> Tensor:
>>> x_np = x.numpy(force=True)
>>> res = np.stack(np.nonzero(x_np), axis=1)
>>> return torch.tensor(res, device=x.device)
...
>>> @torch.library.register_fake("mylib::custom_nonzero")
>>> def _(x):
>>> # Number of nonzero-elements is data-dependent.
>>> # 由于我们无法查看模拟实现中的数据，
>>> # 我们使用 ctx 对象来构建一个新的 symint，
>>> # 表示数据依赖的大小。
>>> ctx = torch.library.get_ctx()
>>> nnz = ctx.new_dynamic_size()
>>> shape = [nnz, x.dim()]
>>> result = x.new_empty(shape, dtype=torch.int64)
>>> return result
...
>>> from torch.fx.experimental.proxy_tensor import make_fx
...
>>> x = torch.tensor([0, 1, 2, 3, 4, 0])
>>> trace = make_fx(torch.ops.mylib.custom_nonzero, tracing_mode="symbolic")(x)
>>> trace.print_readable()
...
>>> 断言 torch.allclose(trace(x), torch.ops.mylib.custom_nonzero(x))
"""
如果 not isinstance(
操作, (str,
火炬.
操作符.
操作符重载,
火炬.
库.
自定义运算.
自定义运算定义)
):
抛出
值错误(f
"注册伪造（"{
操作}
): 预期外的类型对于 op:{
类型(
操作)}")
如果 isinstance(
操作,
火炬.
操作符.
操作符重载):
操作符 =
操作.
名称
操作定义 =
可能获取操作定义(
操作)
如果
操作定义
是 not
无:
如果
函数
是
无:
返回
操作定义.
注册伪造
否则:
返回
操作定义.
注册伪造(
函数)
断言 isinstance(
操作, str)
栈级别 =
_堆栈级别
定义
注册(
函数):
命名空间,
操作名称 =
火炬.
库.
工具.
解析命名空间(
操作)
如果
库
是
无:
使用库 =
库(
命名空间,
"碎片")
_保持活跃.
添加(
使用库)
否则:
使用库 =
库
使用库.
注册模拟实现(
操作符名称,
函数, _stacklevel=
栈级别 + 1)
返回
函数
如果
函数
是
无:
返回
注册
否则:
栈级别 += 1
返回
注册(
函数)
[文档]
定义
注册自动微分(
操作:
_op 标识符,
反向:
可调用,
/,
*,
设置上下文:
可选[
可调用] =
无,
库=
无,
) -> 无:
r注册此自定义操作的逆向公式。
为了使操作符能够与 autograd 一起工作，您需要注册
一个反向公式：
1. 您必须告诉我们如何在反向传播过程中计算梯度
通过为我们提供一个“反向”函数。
2. 如果您需要从正向计算梯度时使用任何正向的值，您可以
使用 `setup_context` 来保存向后传递的值。
``backward`` 在反向传播过程中运行。它接受 `(ctx, *grads)`:
- ``grads`` 是一个或多个梯度。梯度的数量与
操作符的输出数量相匹配。
``ctx``对象与``_中使用的`ctx`对象相同。
`torch.autograd.Function`的语义与`backward_fn`相同。
`setup_context(ctx, inputs, output)`在正向传播期间运行。
`torch.autograd.Function.backward`的语义相同。
请将所需的数量反向保存到“ctx”对象中
either :meth:`torch.autograd.function.FunctionCtx.save_for_backward`
翻译为：either :meth:`torch.autograd.function.FunctionCtx.save_for_backward`
将它们作为 ``ctx`` 的属性分配。如果你的自定义操作
仅支持关键字参数，我们期望`setup_context`的签名
`setup_context(ctx, inputs, keyword_only_inputs, output)` 需要设置上下文。
`setup_context_fn` 和 `backward_fn` 必须可追踪。也就是说，
它们不能直接访问 :meth:`torch.Tensor.data_ptr`，并且不能
依赖于或修改全局状态。如果您需要不可追踪的反向操作，
你可以将其制作为一个独立的 custom_op，在 backward_fn 中调用。
如果您需要在不同的设备上实现不同的 autograd 行为，那么我们建议创建两个不同的自定义算子，每个设备一个，并在运行时切换它们。
推荐为需要不同行为的每个设备创建两个不同的自定义算子，并在运行时进行切换。
需要不同行为的设备，建议创建两个不同的自定义算子，并在运行时进行切换。
示例：
>>> 导入 torch
>>> import numpy as np
>>> 从 torch 导入 Tensor
...
>>> @torch.library.custom_op("mylib::numpy_sin", mutates_args=())
>>> def numpy_sin(x: Tensor) -> Tensor:
>>> x_np = x.cpu().numpy()
>>> y_np = np.sin(x_np)
>>> return torch.from_numpy(y_np).to(device=x.device)
...
>>> def setup_context(ctx, inputs, output) -> Tensor:
>>> x, = inputs
>>> ctx.save_for_backward(x)
...
>>> def backward(ctx, grad):
>>> x, = ctx.saved_tensors
>>> return grad * x.cos()
...
>>> torch.library.register_autograd(
... "mylib::numpy_sin", backward, setup_context=setup_context
... )
...
>>> x = torch.randn(3, requires_grad=True)
>>> y = numpy_sin(x)
>>> (grad_x,) = torch.autograd.grad(y, x, torch.ones_like(y))
>>> assert torch.allclose(grad_x, x.cos())
...
>>> # 示例：仅使用关键字参数
>>> @torch.library.custom_op("mylib::numpy_mul", mutates_args=())
>>> def numpy_mul(x: Tensor, *, val: float) -> Tensor:
>>> x_np = x.cpu().numpy()
>>> y_np = x_np * val
>>> return torch.from_numpy(y_np).to(device=x.device)
...
>>> def setup_context(ctx, inputs, keyword_only_inputs, output) -> Tensor:
>>> ctx.val = keyword_only_inputs["val"]
...
>>> def backward(ctx, grad):
>>> return grad * ctx.val
...
>>> torch.library.register_autograd(
... "mylib::numpy_mul", backward, setup_context=setup_context
... )
...
>>> x = torch.randn(3, requires_grad=True)
>>> y = numpy_mul(x, val=3.14)
>>> (grad_x,) = torch.autograd.grad(y, x, torch.ones_like(y))
>>> assert torch.allclose(grad_x, torch.full_like(x, 3.14))
"""
如果 not isinstance(
操作, (str,
火炬.
操作符.
操作符重载,
火炬.
库.
自定义运算.
自定义运算定义)
):
抛出
值错误(
fregister_autograd({
操作}
): 预期外的类型对于 op:{
类型(
操作)}"
)
如果 isinstance(
操作,
火炬.
操作符.
操作符重载):
操作符 =
操作.
名称
操作定义 =
可能获取操作定义(
操作)
如果
操作定义
是 not
无:
操作定义.
注册自动微分(
反向,
设置上下文=
设置上下文)
返回
断言 isinstance(
操作, str)
qualname = 操作符
操作符 =
火炬.
库.
工具.
查找操作(qualname)
架构 =
操作.
_架构
如果 not
库.
工具.
功能性模式(
架构):
抛出 RuntimeError(
f"无法为非功能性算子注册自动微分公式"
f"{操作}
与模式{
架构}
请创建 "
f一个功能操作符并为该操作符注册一个自动求导公式。
)
如果
库.
工具.
只有关键字参数的张量(
架构):
抛出
不支持的操作异常(
f使用仅关键字参数的 Tensor 参数注册自动求导。在原始 "
f的定义中，请确保您的张量不是仅关键字参数的。
f"获取："{
架构}"
)
信息 =
库.
自动微分.
信息(
反向,
设置上下文)
自动微分内核 =
库.
自动微分.
创建自动微分实现(
操作,
信息)
命名空间,
操作名称 =
火炬.
库.
工具.
解析命名空间(qualname)
如果
库
是
无:
库 =
库(
命名空间,
"碎片")
_保持活跃.
添加(
库)
库.
实现(
操作名称,
自动微分内核,
自动微分,
带有密钥集=True)
[文档]
定义
注册 torch 分派(
操作:
_op 标识符,
torch 调度类:
任何,
函数:
可选[
可调用] =
无,
/,
*,
库:
可选[
库] =
无,
):
r注册给定操作符和 `torch_dispatch_class` 的 torch_dispatch 规则。
这允许开放注册以指定操作员之间的行为
无需修改的 `torch_dispatch_class`
或直接操作员。
torch_dispatch_class 可以是具有__torch_dispatch__的 Tensor 子类
TorchDispatchMode.
如果它是一个 Tensor 子类，我们期望 func 具有以下签名：
(cls, func: OpOverload, types: Tuple[type, ...], args, kwargs) -> Any
如果它是一个 TorchDispatchMode，我们期望 func 具有以下签名：
``(模式，func: OpOverload，types: Tuple[type, ...]，args，kwargs) -> Any``
``args`` 和 ``kwargs`` 将以与它们相同的方式进行归一化
在 ``__torch_dispatch__`` 中（参见 :ref:`torch-dispatch-calling-convention`）。
示例：
>>> 导入 torch
...
>>> @torch.library.custom_op("mylib::foo", mutates_args={})
>>> def foo(x: torch.Tensor) -> torch.Tensor:
>>> return x.clone()
...
>>> class MyMode(torch.utils._python_dispatch.TorchDispatchMode):
>>> def __torch_dispatch__(self, func, types, args=(), kwargs=None):
>>> 返回 func(*args, **kwargs)
...
>>> @torch.library.register_torch_dispatch("mylib::foo", MyMode)
>>> def _(mode, func, types, args, kwargs):
>>> x, = args
>>> 返回 x + 1
...
>>> x = torch.randn(3)
>>> y = foo(x)
>>> 断言 torch.allclose(y, x)
...
>>> with MyMode():
>>> y = foo(x)
>>> assert torch.allclose(y, x + 1)
"""
如果 not isinstance(
操作, (str,
火炬.
操作符.
操作符重载,
火炬.
库.
自定义运算.
自定义运算定义)
):
抛出
值错误(
f"注册 torch 分派("{
操作}
): 预期外的类型对于 op:{
类型(
操作)}"
)
如果 isinstance(
操作,
火炬.
操作符.
操作符重载):
操作符 =
操作.
名称
操作定义 =
可能获取操作定义(
操作)
如果
操作定义
是 not
无:
返回
操作定义.
注册 torch 分派(
torch 调度类,
函数)
断言 isinstance(
操作, str)
定义
注册(
函数):
命名空间,
操作名称 =
火炬.
库.
工具.
解析命名空间(
操作)
如果
库
是
无:
使用库 =
库(
命名空间,
"碎片")
_保持活跃.
添加(
使用库)
否则:
使用库 =
库
使用库.
_注册 torch 调度规则(
操作符名称,
torch 调度类,
函数)
返回
函数
如果
函数
是
无:
返回
注册
否则:
返回
注册(
函数)
[文档]
定义
注册_vmap(
操作:
_op 标识符,
函数:
可选[
可调用] =
无,
/,
*,
库=
无,
):
r注册一个 vmap 实现，以支持此自定义操作的:func:`torch.vmap`。
此 API 可以用作装饰器（请参阅示例）。
为了使操作员能够使用 :func:`torch.vmap`，您可能需要注册
vmap 实现如下签名：
vmap_func(info, in_dims: 可选[int] 元组, *args, **kwargs)
``*args`` 和 ``**kwargs`` 是 ``op`` 的参数和关键字参数。
我们不支持仅关键字参数的 Tensor 参数。
它指定了在给定具有额外维度（由`in_dims`指定）的输入时，如何计算`op`的批处理版本。
对于`args`中的每个参数，`in_dims`都有一个对应的`Optional[int]`。它是`None`。
对于`args`中的每个参数，`in_dims`都有一个对应的`Optional[int]`。它是`None`。
如果参数不是张量或者参数没有被 vmapped，否则它是一个整数
指定正在被 vmapped 操作覆盖的张量的哪个维度
``info`` 是一个可能包含有用额外元数据的集合：
``info.batch_size`` 指定了正在被 vmapped 覆盖的维度的尺寸，同时
``info.randomness`` 是传递给 :func:`torch.vmap` 的 ``randomness`` 选项。
函数 ``func`` 的返回值是一个包含 ``output, out_dims`` 的元组。与 ``in_dims`` 类似，
``out_dims`` 应与 ``output`` 具有相同的结构，并包含一个 ``out_dim``，
每个输出指定该输出是否具有 vmapped 维度以及它在其中的索引。
示例：
>>> 导入 torch
>>> import numpy as np
>>> 从 torch 导入 Tensor
>>> 从 typing 导入 Tuple
...
>>> 定义 to_numpy(tensor)函数
>>> 返回 tensor.cpu().numpy()
...
>>> lib = torch.library.Library("mylib", "FRAGMENT")
>>> @torch.library.custom_op("mylib::numpy_cube", mutates_args=())
>>> def numpy_cube(x: Tensor) -> Tuple[Tensor, Tensor]:
>>> x_np = to_numpy(x)
>>> dx = torch.tensor(3 * x_np ** 2, device=x.device)
>>> return torch.tensor(x_np ** 3, device=x.device), dx
...
>>> def numpy_cube_vmap(info, in_dims, x):
>>> result = numpy_cube(x)
>>> 返回结果，(in_dims[0]，in_dims[0])
...
>>> torch.library.register_vmap(numpy_cube, numpy_cube_vmap)
...
>>> x = torch.randn(3)
>>> torch.vmap(numpy_cube)(x)
...
>>> @torch.library.custom_op("mylib::numpy_mul", mutates_args=())
>>> def numpy_mul(x: Tensor, y: Tensor) -> Tensor:
>>> return torch.tensor(to_numpy(x) * to_numpy(y), device=x.device)
...
>>> @torch.library.register_vmap("mylib::numpy_mul")
>>> def numpy_mul_vmap(info, in_dims, x, y):
>>> x_bdim, y_bdim = in_dims
>>> x = x.movedim(x_bdim, -1) if x_bdim is not None else x.unsqueeze(-1)
>>> y = y.movedim(y_bdim, -1) if y_bdim is not None else y.unsqueeze(-1)
>>> result = x * y
>>> result = result.movedim(-1, 0)
结果 = 结果.movedim(-1, 0)
>>> 返回结果，0
...
...
>>> x = torch.randn(3)
>>> y = torch.randn(3)
>>> torch.vmap(numpy_mul)(x, y)
.. 注意::
vmap 函数应旨在保留整个自定义算子的语义。
即，应将 `grad(vmap(op))` 替换为 `grad(map(op))`。
如果您的自定义算子在反向传播中具有任何自定义行为，请记住这一点。
请注意这一点。
"""
如果 not isinstance(
操作, (str,
火炬.
操作符.
操作符重载,
火炬.
库.
自定义运算.
自定义运算定义)
):
抛出
值错误(f
register_vmap({
操作}
): 预期外的类型对于 op:{
类型(
操作)}")
如果 isinstance(
操作,
火炬.
操作符.
操作符重载):
操作符 =
操作.
名称
操作定义 =
可能获取操作定义(
操作)
如果
操作定义
是 not
无:
返回
操作定义.
注册_vmap(
函数)
断言 isinstance(
操作, str)
qualname = 操作符
操作符 =
火炬.
库.
工具.
查找操作(qualname)
架构 =
操作.
_架构
如果
库.
工具.
只有关键字参数的张量(
架构):
抛出
不支持的操作异常(
f在原始操作符的定义中，请确保您的张量不是仅关键字参数的。
f的定义中，请确保您的张量不是仅关键字参数的。
f"获取："{
架构}"
)
定义
注册(
函数):
非局部
操作,
库
命名空间,
操作名称 =
火炬.
库.
工具.
解析命名空间(qualname)
如果
库
是
无:
库 =
库(
命名空间,
"碎片")
_保持活跃.
添加(
库)
from torch._functorch.autograd_function 导入 custom_function_call_vmap_helper
from torch._functorch.pyfunctorch 导入 retrieve_current_functorch_interpreter
定义 wrapped_func(
键集, *
参数, **
关键字参数):
解释器 =
获取当前 functorch 解释器()
返回
自定义函数调用_vmap 辅助器(
解释器,
函数,
操作, *
参数, **kwargs
)
库.
实现(
操作名称, wrapped_func, "FuncTorchBatched",
带有密钥集=True)
如果
函数
是
无:
返回
注册
否则:
返回
注册(
函数)
如果操作是在 C++中定义的，那么我们需要确保存在
m.set_python_module(module, ...) 调用，并且该模块是
与调用 torch.library.register_fake 的模块相同。
定义 _check_pystubs_once(
函数, qualname,
实际模块名):
已检查 =
假
定义
内部(*
参数, **
关键字参数):
非局部
已检查
如果
已检查:
返回
函数(*
参数, **
关键字参数)
操作符 =
火炬.
库.
工具.
查找操作(qualname)
如果
操作.
在 Python 中定义:
已检查 =
真实
返回
函数(*
参数, **
关键字参数)
maybe_pystub = 火炬._C._dispatch_pystub(
操作.
_模式.
姓名,
操作.
_模式.
重载名称
)
如果 maybe_pystub
是
无:
如果
火炬.
库.
工具.requires_set_python_module():
命名空间 =
操作.
命名空间
cpp_filename = 操作.
处理.
调试()
抛出 RuntimeError(
f"操作员 '"{qualname}
'在 C++中被定义，并且 Python '
f"需要也存在一个“假的实现。在这种情况下，我们要求"
f'的 C++ `m.set_python_module("'{
实际模块名}
")` ' "
f"调用，但我们找不到一个。请将其添加到 "
f"到 C++ TORCH_LIBRARY() 的顶部 "{
命名空间}
， ...) 阻塞的
f"操作符已注册在 ("{
cpp 文件名})"
)
否则:
pystub_module = maybe_pystub[0]
如果 actual_module_name !=
pystub 模块:
cpp_filename = 操作.
处理.
调试()
抛出 RuntimeError(
f"操作员 '"{qualname}
'指定了其 Python 模拟实现" '
f"实际上位于 Python 模块 '"{
pystub 模块}
'但实际上找到的是 " '
f'在 '{
实际模块名}
'. 请移动假的实现 "
f"或修正 m.set_python_module 调用 ('{
cpp 文件名})"
)
已检查 =
真实
返回
函数(*
参数, **
关键字参数)
返回
内部
# 注意 [ctx 在伪造实现内部]
如果用户有一个具有数据依赖输出形状的操作符，那么在编写一个伪造实现时，他们必须查询当前 ctx 并使用 ctx 上的方法来构造一个新的未绑定 symint。
这通过我们在每次设置 fake ctx 时设置全局 ctx_getter 函数来完成。
这通过我们在每次设置 fake ctx 时设置全局 ctx_getter 函数来完成。
#
这通过我们在每次设置 fake ctx 时设置全局 ctx_getter 函数来完成。
调用实现。
[文档]def get_ctx() -> "torch._library.fake_impl.FakeImplCtx":
"""get_ctx() 返回当前的 AbstractImplCtx 对象。
在模拟实现内部调用 ``get_ctx()`` 是有效的。
（详见 :func:`torch.library.register_fake` 以获取更多使用细节。
"""
返回 torch._library.fake_impl.global_ctx_getter()
_OPCHECK_DEFAULT_UTILS = (
测试模式,
测试自动注册,
测试_faketensor,
测试_aot_dispatch_dynamic,
)
[文档]
定义
操作检查(
操作:
联合[
火炬.
操作符.
操作符重载,
火炬.
操作符.
操作重载数据包,
自定义运算定义
]
参数:
元组[
任何, ...
]
关键字参数:
可选[
字典[str,
任何]] =
无,
*,
测试工具:
联合[str,
序列[str]] = _OPCHECK_DEFAULT_UTILS,
抛出异常:
布尔值 = True,
atol=无,
相对误差=
无,
) -> 字典[str, str
]
给定一个操作符和一些示例参数，测试该操作符是否
注册正确。
这就是说，当您使用 torch.library/TORCH_LIBRARY API 创建
自定义操作，您指定了关于自定义操作的元数据（例如可变性信息）
并且这些 API 要求您传递给它们的函数满足某些属性（例如在 fake/meta/abstract 内核中不允许访问数据指针）
``opcheck`` 测试这些元数据和属性。
``opcheck`` 测试这些元数据和属性。
具体来说，我们测试以下内容：
- test_schema: 如果模式与操作符的实现匹配
操作符。例如：如果模式指定 Tensor 被修改，那么我们检查实现是否修改了 Tensor。如果模式
指定 Tensor 被修改，那么我们检查实现是否修改了 Tensor。如果模式
指定我们返回一个新的张量，然后我们检查该
实现返回一个新的张量（而不是现有的一个或
一个现有视图的查看。
- test_autograd_registration：如果算子支持训练
(autograd)：我们检查其自动微分公式是否已注册
torch.library.register_autograd 或手动注册到一
或更多 DispatchKey::Autograd 键。任何其他基于 DispatchKey 的
注册可能导致未定义的行为。
- test_faketensor：如果运算符具有 FakeTensor 内核
（并且它是正确的）。FakeTensor 内核对于运算符与 PyTorch 编译
API（torch.compile/export/FX）的配合工作是必要的（
但并不充分）。我们检查一个 FakeTensor 内核
（有时也称为元内核）已注册用于
操作符，并且它是正确的。这个测试将操作符在真实张量上的运行结果与在 FakeTensors 上的运行结果进行比较
运行操作符在真实张量上的结果与运行
操作符在 FakeTensors 上的结果进行比较，以检查它们是否相同
索引元数据（大小/步长/数据类型/设备等）。
- test_aot_dispatch_dynamic：如果操作符具有正确的行为
使用 PyTorch 编译 API（torch.compile/export/FX）。
这将检查输出（以及如果适用的话，梯度）是否是
在 eager-mode PyTorch 和 torch.compile 下相同。
此测试是`test_faketensor`的超集，并且是一个端到端测试；
其他它测试的是操作符支持
函数化和反向传播（如果存在）也
支持 FakeTensor 和函数化。
为了获得最佳结果，请多次调用`opcheck`，使用具有代表性的输入集。
如果您的操作符支持 autograd，请使用`opcheck`与`requires_grad = True`的输入。
如果您的操作符支持 autograd，请使用`opcheck`与`requires_grad = True`的输入。
如果您的运营商支持多设备（例如 CPU 和 CUDA），请
使用 `opcheck` 在所有支持的设备上输入。
参数：
操作符：操作符。必须是装饰过的函数。
`:func:`torch.library.custom_op` 或 OpOverload/OpOverloadPacket
在 torch.ops.*中找到（例如 torch.ops.aten.sin, torch.ops.mylib.foo）
args：算子的参数
kwargs：算子的关键字参数
test_utils：我们应该运行的测试。默认：所有测试
示例：("test_schema", "test_faketensor")
raise_exception：如果我们在第一次错误时引发异常
error。如果为 False，我们将返回一个字典，其中包含有关每个测试是否通过的信息
的信息。
rtol（可选浮点数）：浮点数比较的相对容差。
如果指定了，则必须也指定 atol。
如果省略，则根据 dtype 选择默认值。
（请参阅 :func:`torch.testing.assert_close` 中的表格）。
atol（可选 float）：浮点数比较的绝对容差。
如果指定了 ``rtol``，则必须也指定 ``rtol``。
如果省略，则根据 dtype 选择默认值。
（请参阅 :func:`torch.testing.assert_close` 中的表格）。
..警告：
opcheck 和 :func:`torch.autograd.gradcheck` 测试的是不同的事情；
opcheck 测试的是您对 torch.library API 的使用是否正确，
func:`torch.autograd.gradcheck` 测试的是您的自动微分公式是否正确，
数学上正确。两者都用来测试支持自定义操作的梯度计算。
梯度计算。
示例：
>>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA)
>>> @torch.library.custom_op("mylib::numpy_mul", mutates_args=())
>>> def numpy_mul(x: Tensor, y: float) -> Tensor:
>>> x_np = x.numpy(force=True)
>>> z_np = x_np * y
>>> return torch.from_numpy(z_np).to(x.device)
...
>>> @numpy_mul.register_fake
>>> def _(x, y):
>>> return torch.empty_like(x)
...
>>> def setup_context(ctx, inputs, output):
>>> y, = inputs
>>> ctx.y = y
...
>>> def backward(ctx, grad):
>>> return grad * ctx.y, None
...
>>> numpy_mul.register_autograd(backward, setup_context=setup_context)
...
>>> sample_inputs = [
>>> (torch.randn(3), 3.14),
>>> (torch.randn(2, 3, device='cuda'), 2.718),
>>> (torch.randn(1, 10, requires_grad=True), 1.234),
>>> (torch.randn(64, 64, device='cuda', requires_grad=True), 90.18),
>>> ]
...
>>> for args in sample_inputs:
>>> torch.library.opcheck(numpy_mul, 参数)
"""
导入
torch.testing._internal.选项测试 as
选项测试
返回
选项测试.
操作检查(
操作,
参数,
关键字参数,
测试工具=
测试工具,
抛出异常=
抛出异常,
相对误差=
相对误差,
atol=atol,
)
