torch.compiler 的源代码

# mypy: 允许未类型化定义
from 打字 导入 任何, 可调用, 可选, 类型检查, 类型变量
from typing_extensions 导入 参数规范

导入 火炬

from . 导入 配置


如果 类型检查:
    from ._缓存 导入 缓存信息


全部 = [
    "编译",
    配置,
    假设常量结果,
    重置,
    "允许在图中",
    "在图中替换",
    "列出后端",
    "禁用",
    设置站姿,
    标记步骤开始,
    包装 NumPy,
    正在编译,
    "是否 Dynamo 正在编译",
    "是否正在导出",
    "保存缓存工件",
    "加载缓存工件",
]


_P = 参数规范("_P")
_R = 类型变量("_R")


[文档]def 编译(*args, **kwargs):
"""
查看 :func:`torch.compile` 了解此函数的参数详情。
"""
return torch.compile(*args, **kwargs)


[文档]def reset() -> None:
""
此函数清除所有编译缓存并恢复系统到初始状态。
建议在执行如 `torch.compile(...)` 等操作后调用此函数。
确保在执行另一个无关的编译操作前系统处于干净状态。
"""
导入 torch._dynamo

torch._dynamo.reset()


定义 允许在图中(fn):
    ""
告诉编译器前端（Dynamo）跳过函数的符号内省
直接在遇到时写入图，而不是先写下来。

如果你正在使用 :func:`torch.compile`（后端为 "inductor"（默认）），或者
使用 :func:`torch.export.export`，并尝试在整个跟踪过程中黑盒化一个 Python 函数，
请不要使用此 API。
因此，请创建一个自定义操作符（请参阅 `PyTorch 自定义操作页面
`_）

..警告：

如果你是典型的 torch.compile 用户（例如，你正在将 torch.compile 应用到
（为了使模型运行更快），你可能不想使用这个函数。
func:`allow_in_graph` 是一个容易误用的功能，因为它跳过了编译器前端。
（Dynamo）负责进行安全检查（图断开、处理闭包等）。错误使用会导致难以调试的静默错误。
等）。错误使用将导致难以调试的静默错误。
错误问题。

给定一个没有 `allow_in_graph` 装饰器的 Python 函数，常规执行
torch.compile 跟踪函数。`:func:`allow_in_graph` 改变
它，使得前端不会在函数内部跟踪，但编译器
后端仍然追踪它。与自定义操作符相比，
在整个 torch.compile 堆栈中将函数视为黑盒。以下表格比较了这些机制。
表格比较了这些机制。

    +------------------------+-----------------------+--------------------------------+
| 机制              | 前端（Dynamo）     | 后端（AOTAutograd+Inductor） |
    +========================+=======================+================================+
| 无装饰器               | 跟踪内部               | 跟踪内部                       |
    +------------------------+-----------------------+--------------------------------+
| 允许在图中             | 不透明可调用函数       | 跟踪内部                       |
    +------------------------+-----------------------+--------------------------------+
| 自定义操作             | 不透明可调用函数       | 不透明可调用函数              |
    +------------------------+-----------------------+--------------------------------+

一个常见的用途是：func:`allow_in_graph()` 作为编译器的逃生门
前端：如果你知道该函数与下游组件的工作方式
编译栈（AOTAutograd 和 Inductor），但存在一个 Dynamo 错误阻止了它
符号性地正确内省函数（或者如果你的代码是用 C/C++编写的）
因此无法使用 Dynamo 进行内省），然后可以通过装饰器：func:`allow_in_graph` 来绕过 Dynamo。
使用：func:`allow_in_graph` 装饰器来绕过 Dynamo。

我们要求 ``fn`` 遵守以下限制。如果违反这些限制，则会导致未定义的行为：
结果将导致未定义行为：

- ``fn`` 的输入必须是 FX 图中的可代理类型。有效类型包括：
      Tensor/int/bool/float/None/List[Tensor?]/List[int?]/List[float?]
      Tuple[Tensor?, ...]/Tuple[int?, ...]/Tuple[float?, ...]/torch.dtype/torch.device
- ``fn`` 的输出必须是 FX 图中的可代理类型（参见上一条项目）
函数 ``fn`` 内部使用的所有张量必须直接作为输入传递给 ``fn``
（而不是作为捕获变量）。

参数：
fn：表示要包含在图中的函数的可调用对象。
如果 ``fn`` 是可调用对象的列表或元组，则递归地应用
将 `:func:`allow_in_graph()` 函数应用于每个函数，并返回一个新的列表。
包含修改后函数的元组。

示例::

        torch.compiler.allow_in_graph(my_custom_function)

        @torch.compile(...)
        def fn(x):
            x = torch.add(x, 1)
            x = my_custom_function(x)
            x = torch.add(x, 1)
            return x

        fn(...)

将捕获包含 `my_custom_function()` 的单个图。

    """
    导入 torch._dynamo

    返回 火炬._dynamo.允许在图中(fn)



定义 substitute_in_graph(
    original_fn: 可调用[_P, _R]
    *,
    可以常量折叠: 布尔值 = False,
    跳过签名检查: 布尔值 = False,
) -> 可调用[[可调用[_P, _R]], 可调用[_P, _R]]
    ""
注册一个 polyfill 处理程序，通常是一个 C 扩展的 C 函数，用于在图形中内联原始函数时替代原始函数。
在图形中内联原始函数时，通常用注册的 polyfill 处理程序来替代原始函数。

.. 注意::

当内联原始函数时，才会使用 polyfill 处理器。当直接调用原始函数时，不会使用 polyfill 处理器。
在急切模式下，装饰函数调用的是性能更好的 C 函数，而不是 polyfill 处理器。
polyfill 处理器是一个函数，当需要替代原始函数时，将在这个位置被调用。

polyfill 处理器是一个函数，当需要替代原始函数时，将在这个位置被调用。
将原始函数内联。polyfill 处理程序应具有相同的签名和相同的行为。
与原始函数相同的行为。

参数：
original_fn（可调用对象）：要注册 polyfill 处理程序的原始函数，通常是一个 C 函数。
。
can_constant_fold_through (bool, 可选): 是否允许 polyfill 处理器进行常量折叠
即，如果 polyfill 处理器是一个纯函数，并且其参数
是常量，则 polyfill 处理器的结果可以在编译期间进行常量折叠。
默认为 ``False``。
skip_signature_check (bool, 可选): 是否跳过原始函数和 polyfill 处理程序之间的签名检查。
原始函数和 polyfill 处理程序。默认为 ``False``。

返回值：
用于注册原始函数的 polyfill 处理程序的装饰器。

示例::

>>> 导入 operator
        >>> operator.indexOf([1, 2, 3, 4, 5], 3)
        2
        >>> torch.compile(operator.indexOf, fullgraph=True)([1, 2, 3, 4, 5], 3)
        ... # xdoctest: +SKIP("Long tracebacks")
        Traceback (most recent call last):
        ...
        torch._dynamo.exc.Unsupported: ...

        >>> @torch.compiler.substitute_in_graph(operator.indexOf)
        ... def indexOf(a, b, /):
        ...     for i, item in enumerate(a):
...如果项目是 b 或 item == b:
...则返回 i
...引发 ValueError("sequence.index(x): x 不在序列中")
...
        >>> torch.compile(operator.indexOf, fullgraph=True)([1, 2, 3, 4, 5], 3)
        2
    """
    导入 torch._dynamo

    返回 火炬._dynamo.substitute_in_graph(
        original_fn,
        可以常量折叠=可以常量折叠,
        跳过签名检查=跳过签名检查,
    )



[文档]def list_backends(exclude_tags=("debug", "experimental")) -> list[str]:
"""
返回可用于传递给 `torch.compile(..., backend="name")` 的有效字符串。

参数:
排除标签（可选）：表示要排除的标签的字符串元组。
"""
导入 torch._dynamo

返回 torch._dynamo.list_backends(exclude_tags)


[文档]def assume_constant_result(fn):
"""
此函数用于标记函数 `fn` 具有恒定结果。
这允许编译器优化掉您的函数。
返回相同的函数 `fn`

参数:
fn: 被标记为具有恒定结果的函数。

.. 警告::
`假设常量结果`可能导致安全性和稳健性问题，:func:`torch.compile`
不会尝试验证常量假设是否为真

```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)
```
import torch._dynamo

return torch._dynamo.assume_constant_result(fn)


[文档]def disable(fn=None, recursive=True):
"""
此函数提供了一个装饰器，用于禁用函数的编译。
它还提供了递归禁用被调用函数的选项。

Args:
fn（可选）：要禁用的函数。
recursive（可选）：一个布尔值，表示是否递归禁用。
"""
导入 torch._dynamo

返回 torch._dynamo.disable(fn, recursive)


定义 set_stance(
    姿态: 字符串 = 默认, *, 跳过不安全守卫评估=False, force_backend=无
):
    ""
设置编译器的当前状态。
可用作函数、上下文管理器或装饰器。
不要在 `torch.compile` 区域内使用此函数 - 否则将引发错误。

.. 代码块 :: python

        @torch.compile
        def foo(x):
            ...

        @torch.compiler.set_stance("force_eager")
        def bar():
            # will not be compiled
            foo(...)

        bar()

        with torch.compiler.set_stance("force_eager"):
# 将不会被编译
            foo(...)

        torch.compiler.set_stance("force_eager")
# 将不会被编译
        foo(...)
        torch.compiler.set_stance("default")

待编译
        foo(...)

参数：
立场：设置编译器的立场。有效值有：

- "default"：默认立场，用于正常编译。
- "force_eager": 忽略所有 `torch.compile` 指令。
- "eager_on_recompile": 需要重新编译时，积极运行代码。
如果存在针对输入有效的缓存编译代码，它仍然会被使用。
- "fail_on_recompile": 在重新编译函数时引发错误。

跳过_guard_eval_unsafe：仅运行区分守卫的标志
注意 - 此标志不安全，仅在您的设置中需要时使用
满足以下条件。

torch.compile 使用守卫系统来支持重新编译
选择在运行时运行的编译工件。这些守卫虽然高效，但会增加一些开销，这可能会影响性能，在需要优化最小化守卫处理时间的场景中。
这些开销可能会影响性能，在需要优化最小化守卫处理时间的场景中。
此 API 允许您禁用守卫评估，假设
此 API 允许您禁用守卫评估，假设
您已用足够多样的数据预热了编译好的模型
输入后的假设意味着，在预热阶段之后，没有
进一步重新编译将有必要。如果这个假设失败，
存在静默产生错误结果的风险（因此）
API 名称中的“unsafe”术语。

force_backend: 如果 `stance` 是 "default"，则可以使用此参数强制 `torch.compile`
使用特定的后端。否则，将引发错误。
    """
    导入 torch._dynamo

    返回 火炬._dynamo.set_stance(
        姿态,
        跳过不安全守卫评估=跳过不安全守卫评估,
        force_backend=force_backend,
    )



# 禁止在图中
set_stance._dynamo_forbidden = 真实  # 类型：忽略[已定义]


[文档]def cudagraph_mark_step_begin():
"""
表示即将开始新的推理或训练迭代。

CUDA 图将释放先前迭代的张量。每次调用 torch.compile 时，只要没有未调用的待处理反向操作，就会启动一个新的迭代。
如果该启发式方法错误，例如在以下示例中，请使用此 API 手动标记。

如果该启发式方法错误，例如在以下示例中，请使用此 API 手动标记。

.. 代码块 :: python

        @torch.compile(mode="reduce-overhead")
def rand_foo():
return torch.rand([4], device="cuda")

for _ in range(5):
torch.compiler.cudagraph_mark_step_begin()
rand_foo() + rand_foo()

更多详细信息，请参阅 `torch.compiler_cudagraph_trees `__
"""
从 torch._inductor 导入 cudagraph_trees

cudagraph_trees.mark_step_begin()


定义 wrap_numpy(fn):
    r装饰器，将一个从 ``np.ndarray`` 转换为 ``np.ndarray`` 的函数转换为函数
从 `torch.Tensor` 转换为 `torch.Tensor`。

它旨在与 `:func:`torch.compile` 配合使用，其中 `fullgraph=True`。它允许将 NumPy 函数编译成 PyTorch 函数。
这允许您在 CUDA 上运行 NumPy 代码或计算其梯度。
它可以编译 NumPy 函数，就像它是 PyTorch 函数一样。这使得您可以在 CUDA 上运行 NumPy 代码或计算其梯度。

.. 注意::

此装饰器在没有 :func:`torch.compile` 的情况下无法工作。

示例::

        >>> # xdoctest: +REQUIRES(env:TORCH_DOCTEST_CUDA)
>>> # 将 NumPy 函数编译为 Tensor -> Tensor 函数
        >>> @torch.compile(fullgraph=True)
        >>> @torch.compiler.wrap_numpy
        >>> def fn(a: np.ndarray):
        >>>     return np.sum(a * a)
>>> 执行 NumPy 函数使用 CUDA 上的张量并计算梯度
        >>> x = torch.arange(6, dtype=torch.float32, device="cuda", requires_grad=True)
        >>> out = fn(x)
        >>> out.backward()
>>> 打印(x.grad)
        tensor([ 0.,  2.,  4.,  6.,  8., 10.], device='cuda:0')
    """
    from torch._dynamo.external_utils 导入 wrap_numpy as 包装

    返回 包裹(fn)


_is_compiling_flag: 布尔值 = 假
_is_exporting_flag: 布尔值 = 假


[文档]def is_compiling() -> bool:
"""
表示一个图是否作为 torch.compile()或 torch.export()的一部分执行/跟踪。

注意，还有两个相关的标志最终应该被弃用：
* torch._dynamo.external_utils.is_compiling()
* torch._utils.is_compiling()

示例::

>>> def forward(self, x):
>>>     如果不是 torch.compiler.is_compiling():
>>>        pass # ...在编译/追踪图中不需要的逻辑...
>>>
>>>     # ...rest of the function...
"""
如果 torch.jit.is_scripting():
返回 False
else:
返回 _is_compiling_flag


[文档]def is_dynamo_compiling() -> bool:
"源代码"
指示一个图是否通过 TorchDynamo 进行追踪。

比 is_compiling()标志更严格，因为它只有在使用 TorchDynamo 时才会设置为 True。
TorchDynamo 被使用。

示例::

>>> def forward(self, x):
>>>     如果不是 torch.compiler.is_dynamo_compiling():
>>>        pass # ...在 TorchDynamo-traced 图中不需要的逻辑...
>>>
>>>     # ...rest of the function...
"""
返回 False


[文档]def is_exporting() -> bool:
"""
表示是否处于导出状态。

它比 is_compiling()标志更为严格，因为它只有在
torch.export 被使用。

示例::

>>> def forward(self, x):
>>>     如果不是 torch.compiler.is_exporting():
>>>        通过 # ...逻辑在导出中不需要...
>>>
>>>     # ...函数的其余部分...
"""
返回_is_exporting_flag 标志


定义 保存缓存工件() -> 可选[元组[字节, 缓存信息]]
    ""
序列化编译过程中创建的所有缓存工件

示例：

执行 torch.compile
调用 torch.compiler.save_cache_artifacts()
    """
    from ._缓存 导入 CacheArtifactManager, 缓存信息

    返回 CacheArtifactManager.序列化()


定义 加载缓存工件(序列化工件: 字节) -> 可选[缓存信息]
    ""
热加载之前通过序列化存储的缓存工件
保存缓存工件

示例：

# 从之前的调用
    artifacts = torch.compiler.save_cache_artifacts()

    torch.compiler.load_cache_artifacts(artifacts[0])
    """
    from ._缓存 导入 缓存工件管理器, 缓存信息

    返回 缓存工件管理器.反序列化(序列化工件)