torch._inductor 源代码

# mypy: 允许未类型化定义
from 未来 导入 注释

导入 io
导入 记录日志
导入 操作系统
from 打字 导入 任何, IO, 可选, 类型检查, 联合

导入 torch._inductor 配置
导入 torch.fx


如果 类型检查:
    from torch._inductor 工具 导入 输入类型
    from torch.export 导入 导出程序
    from torch 的类型 导入 FileLike

全部 = [
    "编译",
    "沉浸式翻译选项",
    "列表选项",
    标记步骤开始,
]


日志 = 记录.获取日志记录器(__name__)


定义 编译(
    gm: 火炬.fx.GraphModule,
    示例输入: 列表[输入类型]
    选项: 可选[字典[str, 任何]] = 无,
):
    ""
使用 TorchInductor 编译给定的 FX 图。这允许编译
没有使用 TorchDynamo 捕获的 FX 图。

参数：
gm: 要编译的 FX 图。
example_inputs: 张量输入的列表。
选项：可选的配置选项字典。请参阅 `torch._inductor.config`。

返回值：
可调用，与 gm 具有相同的行为但速度更快。
    """
    from .compile_fx 导入 编译_fx

    返回 编译_fx(gm, 示例输入, 配置修补=选项)


定义 aoti_编译和打包(
    导出的程序: 导出程序,
    _已弃用未使用参数=无,
    _已弃用未使用关键字参数=无,
    *,
    包路径: 可选[文件类] = 无,
    电感器配置: 可选[字典[str, 任何]] = 无,
) -> str:
    ""
使用 AOTInductor 编译导出的程序，并将其打包成.pt2 文件
通过 input package_path 指定的工件。要加载该包，您可以调用
``torch._inductor.aoti_load_package(package_path)``。

以下是一个示例用法：

.. 代码块 :: python

        ep = torch.export.export(M(), ...)
        aoti_file = torch._inductor.aoti_compile_and_package(
            ep, package_path="my_package.pt2"
        )
        compiled_model = torch._inductor.aoti_load_package("my_package.pt2")

要编译并保存多个模型到一个单独的 ``.pt2`` 文件中，你可以这样做：
以下是一个例子：

.. 代码块 :: python

        ep1 = torch.export.export(M1(), ...)
        aoti_file1 = torch._inductor.aot_compile(
ep1, ..., 选项={"aot_inductor.package": True}
        )
        ep2 = torch.export.export(M2(), ...)
        aoti_file2 = torch._inductor.aot_compile(
ep2, ..., 选项={"aot_inductor.package": True}
        )

从 torch._inductor.package 导入 package_aoti, load_package

        package_aoti("my_package.pt2", {"model1": aoti_file1, "model2": aoti_file2})

        compiled_model1 = load_package("my_package.pt2", "model1")
compiled_model2 = 加载包("my_package.pt2", "model2")

参数：
exported_program: 通过 torch.export 调用的导出程序
package_path: 生成.pt2 工件的可选指定路径
inductor_configs: 控制电感器的可选配置字典

返回值：
生成物路径
    """
    from torch.export 导入 导出程序

    from .debug 导入 aot 电感器最小化包装器

    如果 不 isinstance(导出的程序, 导出程序):
        抛出 值错误(仅支持 ExportedProgram)

    如果 导出的程序.示例输入 是 无:
        抛出 运行时错误(
            "必须设置 exported_program.example_inputs。"
            "用于 AOTInductor 编译。"
        )

    如果 _已弃用未使用参数 是 不 无 或者 已弃用未使用的 kwargs 是 不 无:
        日志.警告(
            "您不再需要指定 aoti_compile_and_package 的 args/kwargs，"
            "因为我们可以从导出的 program.example_inputs 中获取这些信息。"
        )

    断言 (
        软件包路径 是 无
        或者 (
            isinstance(软件包路径, (输入/输出.IO 基础, IO))
            和 软件包路径.可写的()
            和 包路径.可寻址的()
        )
        或者 (
            isinstance(包路径, (str, os.PathLike))
            和 os.文件系统路径(软件包路径).以...结尾(".pt2")
        )
    ), (
        f"期望软件包路径是一个以 .pt2 结尾的文件、为空或是一个缓冲区。然而得到了"{软件包路径}"
    )

    电感器配置 = 电感器配置 或者 {}
    电感器配置["aot_inductor.package"] = 真实

    如果 电感器配置.获取("aot_inductor 的输出路径"):
        抛出 运行时错误(
            "请传递一个包路径给 aot_inductor_compile()，而不是设置 aot_inductor.output_path 配置。"
            "设置 aot_inductor.output_path 配置。"
        )

    aoti_compile_and_package_inner 的包装器。
    返回 aot_inductor_minifier_wrapper(
        _aoti_compile_and_package_inner,
        导出的程序,
        打包路径=包路径,
        电感器配置=电感器配置,
    )



定义 _aoti 编译与打包内部(
    gm: 火炬.神经网络.模块,
    # flat_example_inputs: 列表[任意类型],
    参数: 元组[任何]
    关键字参数: 可选[字典[str, 任何]] = 无,
    *,
    load_and_run: 布尔值 = False,
    check_accuracy: 可选[str] = 无,
    包路径: 可选[联合[str, 输入/输出.BytesIO]] = 无,
    传感器配置: 可选[字典[str, 任何]] = 无,
):
    ""
请参阅 aoti_compile_and_package 的文档字符串。

如果`load_and_run`为 True，此函数将加载编译后的模型并运行。
这是为了让压缩器检查编译模型的正确性。

如果设置了`check_accuracy`，则此函数将检查编译模型与 gm 的准确性。如果设置了`check_accuracy`，则 kwargs 必须为 None。
"strict_accuracy"表示“每次我们看到任何东西时，我们都会进行压缩”。
“严格准确性”意味着“每次我们看到任何东西时，我们都会进行压缩”。
与“diverges”不同，“accuracy”更为保守，只有当存在有意义的 fp64 diverges 时才会进行压缩
当存在有意义的 fp64 diverges 时
    """

    如果 check_accuracy:
        断言 kwargs 是 无 或者 长度(关键字参数) == 0, (
            "在检查准确性时，输入必须已经被展平，且 kwargs 为 None"
        )

    from 包件 导入 包件_奥蒂

    断言 isinstance(gm, 火炬.fx.GraphModule)

    kwargs = kwargs 或者 {}

    奥蒂文件 = aot 编译(gm, 参数, 关键字参数, 选项=电感器配置)
    断言 isinstance(aoti 文件, 列表)

    如果 包路径 是 无:
        路径 = [
            os.路径.分割扩展名(文件)]0]
            对于 文件 在 源文件
            如果 os.路径.splitext(文件)]1] == .so
        ]
        如果 长度(路径) == 0:
            路径 = [
                os.路径.splitext(文件)]0]
                对于 文件 在 aoti_files
                如果 os.路径.splitext(文件)]1] == ".cpp"
            ]
        包路径 = 路径[0] + ".pt2"

    res = 包标题(软件包路径, 源文件)
    断言 res == 软件包路径

    如果 加载并运行 或者 检查准确性:
        编译模型 = 加载包(包路径)
        如果 检查准确性:
            from torch._dynamo.debug_utils 导入 准确度错误, 相同的两个模型

            这可能看起来是倒置的，但实际上并不是。strict_accuracy 意味着“我们将
            # 在看到任何偏离的情况时进行最小化”，而准确性则更为
            # 保守，只有当存在有意义的 fp64
            # 偏差时才会进行最小化
            不严格准确度 = 检查准确度 == 准确度
            如果 不 相同的两个模型(
                gm,
                编译模型,
                参数,
                仅前向=True,
                需要双精度浮点数=不严格准确度,
                忽略非浮点数=不严格准确度,
            ):
                抛出 准确度错误("检测到准确性差")
        否则:
            编译模型(*参数, **关键字参数)

    返回 包路径


[文档]def aoti_load_package(path: 文件类型, run_single_threaded: bool = False) -> 任意:  # 类型: 忽略[type-arg]
```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)
```
加载 PT2 包中的模型。

如果多个模型打包进 PT2，这将加载默认
模型.要加载特定模型，可以直接调用 load API

.. 代码块 :: python

从 torch._inductor.package 导入 load_package

compiled_model1 = load_package("my_package.pt2", "model1")
compiled_model2 = load_package("my_package.pt2", "model2")

Args:
路径：.pt2 包的路径
run_single_threaded (bool)：是否以单线程模式运行模型
thread synchronization logic：是否使用线程同步逻辑。这有助于避免与
CUDAGraphs.
"""
从 torch._inductor.package 导入 load_package

返回 load_package(path, run_single_threaded=run_single_threaded)


定义 aot 编译(
    gm: 火炬.fx.GraphModule,
    参数: 元组[任何]
    关键字参数: 可选[字典[str, 任何]] = 无,
    *,
    选项: 可选[字典[str, 任何]] = 无,
) -> 联合[str, 列表[str]]
    ""
使用 TorchInductor 将给定的 FX 图即时编译成共享库。

参数：
gm: 要编译的 FX 图。
args: 示例参数
kwargs: 示例关键字参数
options: 可选的配置选项字典。请参阅 `torch._inductor.config`。

返回值：
        Path to the generated shared library, or a list of files generated by
AOTI 如果 aot_inductor.package=True。
TODO：默认返回一个列表
    """
    from .编译_fx 导入 _aoti_flatten_inputs, 编译_fx_aot

    平坦示例输入, 选项 = _aoti 平坦输入(
        gm, 参数, 关键字参数, 选项=选项
    )

    返回 编译_fx_aot(
        gm,
        flat_example_inputs,  # type: ignore[arg-type]
        配置修补=选项,
    )


定义 列出模式选项(
    模式: 可选[str] = 无, 动态: 可选[bool] = 无
) -> 字典[str, 任何]
    r返回一个字典，描述了每个可用的优化
`torch.compile()` 传入的模式执行。

参数：
mode（str，可选）：返回优化模式的参数。
如果为 None，则返回所有模式的优化。
dynamic（bool，可选）：是否启用动态形状。

示例::
        >>> torch._inductor.list_mode_options()
    """

    模式选项: 字典[str, 字典[str, bool]] = {
        默认: {},
        # 启用 cudagraphs
        "减少开销": {
            triton.cuDAG 图: True,
        },
        启用最大自调优
        "max-autotune-no-cudagraphs": {
            "max_autotune": True,
            "坐标下降调优": True,
        },
        启用 max-autotune
        启用 cudagraphs
        max-autotune: {
            max_autotune: True,
            triton.cuDAG 图: True,
            "坐标下降调优": True,
        },
    }
    尝试:
        返回 模式选项[模式] 如果 模式 否则 模式选项
    除了 键错误 as e:
        抛出 运行时错误(
            f"未识别的模式="{模式}应该是以下之一：{“，”.连接(模式选项.键())}"
        ) from e


定义 列表选项() -> 列表[str]
    r返回一个描述优化和调试配置的字典
这些都是 `torch.compile()` 可用的选项。

这些选项在 `torch._inductor.config` 中有文档说明。

示例::

        >>> torch._inductor.list_options()
    """

    from torch._inductor 导入 配置

    当前配置: 字典[str, 任何] = 配置.获取配置副本()

    返回 列表(当前配置.键())


定义 cudagraph_mark_step_begin():
    表示即将开始新的推理或训练迭代。
    from .cudagraph_trees 导入 标记步骤开始

    标记步骤开始()