torch.library¶
torch.library 是一个用于扩展 PyTorch 核心库操作符集合的 API 集合。它包含用于测试自定义操作符、创建新自定义操作符以及扩展使用 PyTorch 的 C++ 操作符注册 API(例如 aten 操作符)定义的操作符的工具。
有关如何有效使用这些API的详细指南,请参阅 请参阅 PyTorch 自定义操作符着陆页 以了解更多关于如何有效使用这些API的信息。
测试自定义操作¶
使用 torch.library.opcheck() 测试自定义操作,以检查 Python torch.library 和/或 C++ TORCH_LIBRARY API 的错误用法。此外,如果您的操作支持训练,请使用 torch.autograd.gradcheck() 测试梯度的数学正确性。
- torch.library.opcheck(op, args, kwargs=None, *, test_utils=('test_schema', 'test_autograd_registration', 'test_faketensor', 'test_aot_dispatch_dynamic'), raise_exception=True)[source]¶
给定一个算子和一些样本参数,测试该算子是否正确注册。
也就是说,当你使用torch.library/TORCH_LIBRARY API创建自定义算子时,你需要指定该自定义算子的元数据(例如可变性信息),并且这些API要求你传递给它们的函数必须满足某些属性(例如在假/元/抽象内核中不访问数据指针)。
opcheck测试这些元数据和属性。具体来说,我们测试以下内容: - test_schema: 操作符的模式是否正确。 - test_autograd_registration: autograd 是否已正确注册。 - test_faketensor: 如果操作符具有 FakeTensor 内核 (以及它是否正确)。FakeTensor 内核对于操作符与 PyTorch 编译 APIs (torch.compile/export/FX) 一起工作是必要的 (但不是充分条件)。 - test_aot_dispatch_dynamic: 如果操作符在使用 PyTorch 编译 APIs (torch.compile/export/FX) 时行为正确。 这会检查输出(如果适用还包括梯度)在 eager-mode PyTorch 和 torch.compile 下是否相同。 此测试是
test_faketensor的超集。为了获得最佳效果,请多次调用
opcheck,并使用一组具有代表性的输入。如果您的操作符支持自动微分,请使用opcheck,并输入requires_grad = True;如果您的操作符支持多个设备(例如CPU和CUDA),请使用opcheck,并在所有支持的设备上进行输入。- Parameters
op (Union[OpOverload, OpOverloadPacket, CustomOpDef]) – 要操作的运算符。必须是使用
torch.library.custom_op()装饰的函数,或者是 torch.ops.* 中找到的 OpOverload/OpOverloadPacket (例如 torch.ops.aten.sin, torch.ops.mylib.foo)test_utils (Union[str, Sequence[str]]) – 我们应该运行的测试。默认:全部运行。 示例:(“test_schema”,“test_faketensor”)
raise_exception (bool) – 如果应在第一次出错时抛出异常。如果为False,我们将返回一个字典,其中包含每个测试是否通过的信息。
- Return type
警告
opcheck 和
torch.autograd.gradcheck()测试不同的内容; opcheck 检查你对 torch.library API 的使用是否正确,而torch.autograd.gradcheck()检查你的自动微分公式是否数学上正确。请使用两者来测试支持梯度计算的自定义操作。示例
>>> @torch.library.custom_op("mylib::numpy_mul", mutates_args=()) >>> def numpy_add(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_sin.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_sin.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(foo, args)
用Python创建新的自定义操作¶
使用 torch.library.custom_op() 创建新的自定义操作。
- torch.library.custom_op(name, fn=None, /, *, mutates_args, device_types=None, schema=None)¶
将函数包装为自定义操作符。
您可能希望创建自定义运算的原因包括: - 将第三方库或自定义内核包装起来,以便与 PyTorch 子系统(如自动微分)一起使用。 - 防止 torch.compile/导出/FX 追踪窥探您的函数内部。
此 API 用作函数周围的装饰器(请参见示例)。 提供的函数必须具有类型提示;这些类型提示用于与 PyTorch 的各种子系统进行接口。
- Parameters
名称 (str) – 自定义操作的名称,格式类似于“{命名空间}::{名称}”, 例如 “mylib::my_linear”。该名称在 PyTorch 子系统(如 torch.export、FX 图)中用作操作的稳定标识符。 为避免名称冲突,请使用您的项目名称作为命名空间; 例如,pytorch/fbgemm 中的所有自定义操作都使用 “fbgemm” 作为命名空间。
mutates_args (Iterable[str]) – 函数修改的参数名称. 这必须准确,否则行为是未定义的。
device_types (None | str | Sequence[str]) – 该函数适用的设备类型。如果未提供设备类型,则该函数将作为所有设备类型的默认实现。示例:"cpu","cuda"。
schema (无 | 字符串) – 操作符的模式字符串。如果为None (推荐),我们将从其类型注释中推断出操作符的模式。除非您有特定原因,否则我们建议让系统自动推断模式。 示例:“(Tensor x, int y) -> (Tensor, Tensor)”。
- Return type
注意
我们建议不要传递
schema参数,而是让我们从类型注释中推断。 自己编写模式容易出错。如果您对我们对类型注释的解释不满意, 您可以提供自己的模式。有关如何编写模式字符串的更多信息,请参见 这里- Examples::
>>> import torch >>> from torch import Tensor >>> from torch.library import custom_op >>> import numpy as np >>> >>> @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) >>> >>> x = torch.randn(3) >>> y = numpy_sin(x) >>> assert torch.allclose(y, x.sin()) >>> >>> # Example of a custom op that only works for one device type. >>> @custom_op("mylib::numpy_sin_cpu", mutates_args=(), device_types="cpu") >>> def numpy_sin_cpu(x: Tensor) -> Tensor: >>> x_np = x.numpy() >>> y_np = np.sin(x_np) >>> return torch.from_numpy(y_np) >>> >>> x = torch.randn(3) >>> y = numpy_sin_cpu(x) >>> assert torch.allclose(y, x.sin()) >>> >>> # Example of a custom op that mutates an input >>> @custom_op("mylib::numpy_sin_inplace", mutates_args={"x"}, device_types="cpu") >>> def numpy_sin_inplace(x: Tensor) -> None: >>> x_np = x.numpy() >>> np.sin(x_np, out=x_np) >>> >>> x = torch.randn(3) >>> expected = x.sin() >>> numpy_sin_inplace(x) >>> assert torch.allclose(x, expected)
扩展自定义操作(由Python或C++创建)¶
使用 register.* 方法,例如 torch.library.register_kernel() 和
func:torch.library.register_fake,为任何操作符添加实现
(它们可能通过 torch.library.custom_op() 创建,
或通过 PyTorch 的 C++ 操作符注册 API)。
- torch.library.register_kernel(op, device_types, func=None, /, *, lib=None)[source]¶
为该操作的设备类型注册一个实现。
一些有效的设备类型有:“cpu”,“cuda”,“xla”,“mps”,“ipu”,“xpu”。 此 API 可用作装饰器。
- Parameters
- Examples::
>>> import torch >>> from torch import Tensor >>> from torch.library import custom_op >>> import numpy as np >>> >>> # Create a custom op that works on 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) >>> >>> # Add implementations for the cuda device >>> @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())
- torch.library.register_autograd(op, backward, /, *, setup_context=None, lib=None)[source]¶
为该自定义操作注册反向公式。
为了使算子与自动求导一起工作,你需要注册一个反向传播公式: 1. 在反向传播过程中,你需要告诉我们如何计算梯度,为此需要提供一个“反向”函数。 2. 如果你需要在前向传播中保存任何值以计算梯度,可以使用setup_context来保存这些值用于反向传播。
backward在反向传播过程中运行。它接受(ctx, *grads): -grads是一个或多个梯度。梯度的数量与操作符的输出数量相匹配。ctx对象是相同的 ctx 对象,由torch.autograd.Function使用。backward_fn的语义与torch.autograd.Function.backward()相同。setup_context(ctx, inputs, output)在前向传递期间运行。 请通过将所需的量保存到ctx对象中来准备反向传播, 可以通过torch.autograd.function.FunctionCtx.save_for_backward()或者将其作为ctx的属性进行赋值。如果您的自定义操作具有仅限关键字参数,我们期望setup_context的签名是setup_context(ctx, inputs, keyword_only_inputs, output)。既
setup_context_fn和backward_fn都必须是可追踪的。也就是说, 它们不能直接访问torch.Tensor.data_ptr(),并且它们不能依赖或改变全局状态。如果你需要一个不可追踪的反向传播, 你可以将其作为一个单独的自定义操作,在backward_fn内部调用。示例
>>> import torch >>> import numpy as np >>> from torch import 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()) >>> >>> # Example with a keyword-only arg >>> @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))
- torch.library.register_fake(op, func=None, /, *, lib=None, _stacklevel=1)[source]¶
为该操作注册一个 FakeTensor 实现(“伪实现”)。
有时也被称为“元核”或“抽象实现”。
“‘FakeTensor 实现’”指定了此操作在不携带数据的张量(“FakeTensor”)上的行为。给定一些具有特定属性(尺寸/步幅/存储偏移/设备)的输入张量,它指定了输出张量的属性是什么。
FakeTensor 的实现与操作符具有相同的签名。 它既适用于 FakeTensors 也适用于元张量。要编写 FakeTensor 实现, 假设所有传递给操作符的张量输入都是常规的 CPU/CUDA/元张量, 但它们没有存储,并且你正试图返回常规的 CPU/CUDA/元张量作为输出。 FakeTensor 的实现必须仅由 PyTorch 操作组成(并且不能直接访问任何输入或中间张量的存储或数据)。
此 API 可用作装饰器(参见示例)。
有关自定义操作的详细指南,请参见 https://pytorch.org/tutorials/advanced/custom_ops_landing_page.html
示例
>>> import torch >>> import numpy as np >>> from torch import Tensor >>> >>> # Example 1: an operator without data-dependent output shape >>> @torch.library.custom_op("mylib::custom_linear", mutates_args=()) >>> def custom_linear(x: Tensor, weight: Tensor, bias: Tensor) -> Tensor: >>> raise NotImplementedError("Implementation goes here") >>> >>> @torch.library.register_fake("mylib::custom_linear") >>> def _(x, weight, bias): >>> assert x.dim() == 2 >>> assert weight.dim() == 2 >>> assert bias.dim() == 1 >>> assert x.shape[1] == weight.shape[1] >>> assert weight.shape[0] == bias.shape[0] >>> assert x.device == weight.device >>> >>> return (x @ weight.t()) + bias >>> >>> with 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. >>> # Since we cannot peek at the data in an fake impl, >>> # we use the ctx object to construct a new symint that >>> # represents the data-dependent size. >>> 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() >>> >>> assert torch.allclose(trace(x), torch.ops.mylib.custom_nonzero(x))
- torch.library.impl_abstract(qualname, func=None, *, lib=None, _stacklevel=1)[source]¶
该API在PyTorch 2.4中重命名为
torch.library.register_fake()。 请使用那个代替。
- torch.library.get_ctx()[source]¶
get_ctx() 返回当前的 AbstractImplCtx 对象。
调用
get_ctx()只能在 fake impl 内有效 (有关更多用法细节,请参见torch.library.register_fake())。- Return type
AbstractImplCtx
低级APIs¶
以下 API 是对 PyTorch 的 C++ 低级操作注册 API 的直接绑定。
警告
低级别的算子注册API和PyTorch调度器是PyTorch中的一个复杂概念。我们建议您在可能的情况下使用更高层次的API(这些API不需要torch.library.Library对象)。 这篇博客文章<http://blog.ezyang.com/2020/09/lets-talk-about-the-pytorch-dispatcher/>是一个了解PyTorch调度器的好起点。
在Google Colab上提供了一个教程,通过一些示例向您展示如何使用此API。
- class torch.library.Library(ns, kind, dispatch_key='')[source]¶
一个类,用于创建可以从 Python 中使用的库,以注册新操作符或覆盖现有库中的操作符。 用户可以选择性地传入一个调度键名,以便仅注册与特定调度键相对应的内核。
要创建一个用于重载现有库(名称为 ns)操作符的库,请将类型设置为“IMPL”。 要创建一个新的库(名称为 ns)以注册新的操作符,请将类型设置为“DEF”。 要创建一个可能已存在的库片段以注册操作符(并绕过给定命名空间只能有一个库的限制),请将类型设置为“FRAGMENT”。
- Parameters
ns – 库名称
类型 – “DEF”,“IMPL”(默认:“IMPL”),“FRAGMENT”
dispatch_key – PyTorch 调度键 (默认值:”“)
- define(schema, alias_analysis='', *, tags=())[source]¶
在命名空间 ns 中定义一个新的操作符及其语义。
- Parameters
- Returns
根据模式推断的操作符名称。
- Example::
>>> my_lib = Library("mylib", "DEF") >>> my_lib.define("sum(Tensor self) -> Tensor")
- impl(op_name, fn, dispatch_key='', *, with_keyset=False)[source]¶
为库中定义的操作符注册函数实现。
- Parameters
op_name – 操作符名称(包括重载)或 OpOverload 对象。
fn – 输入分发键的操作实现函数,或
fallthrough_kernel()以注册一个备用操作。dispatch_key – 输入函数应注册的分派键。默认情况下,它使用库创建时的分派键。
- Example::
>>> my_lib = Library("aten", "IMPL") >>> def div_cpu(self, other): >>> return self * (1 / other) >>> my_lib.impl("div.Tensor", div_cpu, "CPU")
- torch.library.define(qualname, schema, *, lib=None, tags=())[source]¶
- torch.library.define(lib, schema, alias_analysis='')
定义一个新的算子。
在 PyTorch 中,定义一个算子(op 的全称为“operator”)是一个两步过程: - 我们需要定义该算子(提供算子名称和模式) - 我们需要实现该算子与各种 PyTorch 子系统的交互行为,比如 CPU/CUDA 张量、自动微分等。
此入口定义了自定义操作符(第一步), 然后您必须通过调用各种
impl_*API 来执行第二步,例如torch.library.impl()或torch.library.register_fake()。- Parameters
qualname (str) – 操作符的限定名。应为类似于“命名空间::名称”的字符串,例如“aten::sin”。 PyTorch中的操作符需要一个命名空间以避免名称冲突;给定的操作符只能创建一次。 如果您正在编写Python库,我们建议命名空间使用顶级模块的名称。
模式 (字符串) – 算子的模式。例如,对于接受一个张量并返回一个张量的操作,其模式为 “(Tensor x) -> Tensor”。 这里不包含操作名称(操作名称通过
qualname传递)。库 (可选[库]) – 如果提供,此操作符的生命周期将与库对象的生命周期绑定。
标签 (标签 | 序列[标签]) – 一个或多个 torch.Tag,应用于此操作符。标记一个操作符会改变该操作符在各种 PyTorch 子系统下的行为;请在应用之前仔细阅读 torch.Tag 的文档。
- Example::
>>> import torch >>> import numpy as np >>> >>> # Define the operator >>> torch.library.define("mylib::sin", "(Tensor x) -> Tensor") >>> >>> # Add implementations for the operator >>> @torch.library.impl("mylib::sin", "cpu") >>> def f(x): >>> return torch.from_numpy(np.sin(x.numpy())) >>> >>> # Call the new operator from torch.ops. >>> x = torch.randn(3) >>> y = torch.ops.mylib.sin(x) >>> assert torch.allclose(y, x.sin())
- torch.library.impl(qualname, types, func=None, *, lib=None)[source]¶
- torch.library.impl(lib, name, dispatch_key='')
为该操作的设备类型注册一个实现。
您可以将“default”传递给
types,以将此实现注册为所有设备类型的默认实现。 请仅在该实现真正支持所有设备类型时使用此选项;例如,如果它是内置的PyTorch操作符组合,则此条件成立。一些有效的类型包括:“cpu”,“cuda”,“xla”,“mps”,“ipu”,“xpu”。
- Parameters
示例
>>> import torch >>> import numpy as np >>> >>> # Define the operator >>> torch.library.define("mylib::mysin", "(Tensor x) -> Tensor") >>> >>> # Add implementations for the cpu device >>> @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())