torch.onnx._internal.exporter._onnx_program 的源代码

# mypy: 允许未类型化定义
# mypy: disable-error-code="attr-defined,name-defined"
来自 未来 导入 注释


__all__ = [ONNX 程序]

导入 contextlib
导入 复制
导入 gc
导入 记录日志
导入 操作系统
导入 tempfile
导入 文本换行
导入 警告
来自 打字 导入 任意, 可调用, 类型检查

导入 火炬
来自 torch.onnx._internal._lazy_import 导入 onnx, onnxscript_apis, onnxscript_ir 作为 ir
来自 torch.onnx._internal.exporter 导入 _dynamic_shapes, _ir_passes
来自 torch.utils 导入 _pytree


# 不要在此模块的全局范围内从 torch.onnx._internal 导入模块
# 因为 ONNXProgram 是对外公开的 API

如果 类型检查:
    来自 collections.abc 导入 序列

    导入 onnxruntime 作为 ort

大型模型阈值 = 1536 * 1024 * 1024  # 1536MB

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


def ort 会话初始化器(模型: 字符串 | 字节) -> ort.推理会话:
    使用指定的模型初始化 ONNX Runtime 推理会话。
    导入 onnxruntime 作为 ort

    会话选项 = ort.SessionOptions()
    会话选项.日志严重级别 = 3  # 3: 错误
    可能的提供者 = (
        CUDA 执行提供者,
        CPU 执行提供者,
    )
    可用提供者 = 集合(ort.获取可用提供商())
    提供商 = [
        提供商 为 提供商 在 可能的提供商 如果 提供商 在 可用提供商
    ]
    返回 ort.推理会话(
        模型, 提供者=提供者, 会话选项=会话选项
    )


def _初始化器计数器大小(图: ir.图) -> int:
    计算初始化器总大小（以字节为单位。）
    返回 总和(
        v.常量值.字节数
        为 v 在 图.初始化器.值()
        如果 v.常量值 是 not None
    )


@contextlib.contextmanager
def _设置图输出(
    图: ir.图,
    输出: 列表[ir.值]
):
    临时设置图的输出。

参数：
图：要设置输出的图。
设置的输出
```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)
```
    原始输出 = 图.输出.复制()
    图.输出.清晰()
    图.输出.扩展(输出)
    try:
        产生
    最后:
        图.输出.清晰()
        图.输出.扩展(原始输出)


def _创建值映射(图: ir.图) -> 字典[字符串, ir.值]：
    返回一个将名称映射到图中值的字典。

该映射不包括子图中的值。

参数：
图：提取映射的图。

返回：
一个将名称映射到值的字典。
```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)
```
    values = {}
    值.更新(图.初始化器)
    # 值的名称可以是 None 或 ""，我们需要排除这些。
    为 输入 在 图.输入:
        如果 not 输入.名称:
            继续
        值[输入.名称] = 输入
    为 节点 在 图:
        为 value 在 节点.输出:
            如果 not 值.名称:
                继续
            值[值.名称] = value
    返回 values


类 ONNX 程序:
    表示可使用 torch 张量调用的 ONNX 程序的类。

    def 初始化(
        自身, 模型: ir.模型, 导出的程序: 火炬.导出.导出程序 | None
    ):
        初始化 ONNX 程序，使用指定的模型和导出的程序。
参数：
模型：ONNX 模型。
导出的程序：生成 ONNX 模型的导出程序。可选。
```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)
```
        自身.模型: ir.模型 = 模型
        自身.导出程序 = 导出程序
        自身._推理会话: ort.推理会话 | None = None
        自身._临时目录: 临时文件.临时目录 | None = None
        导出程序所使用的策略
        自身._捕获策略: 字符串 | None = None

    def __repr__(自身) -> 字符串:
        返回 f"""\
ONNXProgram(
模型=
{textwrap.缩进(字符串(自身.模型), 输入文本为空，请提供需要翻译的文本 * 8)}
    ,
导出程序=
{textwrap.缩进(字符串(自身.导出的程序), 输入文本为空，请提供需要翻译的文本 * 8)}
)
""

    def __调用__(自身, *参数, **kwargs) -> 序列[火炬.张量]：
        使用与提供给 GraphModule 相同的参数运行 ONNX 模型。
        导入 onnxruntime 作为 ort

        展平参数 = _处理参数(参数, kwargs)

        如果 自身._推理会话 是 无:
            自身.初始化推理会话()

        断言 自身._推理会话 是 not None

        # 我们不期望输入非张量
        ort 输入 = {
            k.名称: v.numpy(强制=True)
            为 k, v 在 压缩(自身.模型.图.输入, 展平参数)
        }
        运行选项 = ort.运行选项()
        运行选项.日志严重级别 = 3  # 3: 错误
        记录器.调试(正在运行推理会话%s参数。, 长度(ort 输入))
        输出 = 自身._推理会话.run(无, ort 输入, 运行选项=运行选项)
        记录器.调试(推理会话运行完成。)
        # TODO(justinchuby): 可能需要输出所需的复杂张量
        返回 元组(火炬.从 NumPy 转换(输出) 为 输出 在 输出)

[文档]    def 计算值(
self, value_names: Sequence[str], 参数=(), 关键字参数=None
) -> Sequence[torch.Tensor]:
"""计算 ONNX 模型中指定名称的值。

此方法用于计算 ONNX 模型中指定名称的值。
返回的值以字典形式映射名称到张量。

参数：
value_names：要计算值的名称。

返回值：
名字到张量映射的字典。
"""
如果 kwargs 为 None：
kwargs = {}
self.release()
values = _create_value_mapping(self.model.graph)
for name in value_names:
if name not in values:
引发 ValueError(
"模型中未找到值 '{name}'。"
请提供有效的值名称。
                )
temporary_outputs = [values[name] for name in value_names]
with _set_graph_outputs(self.model.graph, temporary_outputs):
try:
result = self(*args, **kwargs)
finally:
self.release()
返回结果

    @property
    def model_proto(自身) -> onnx.模型原型:
        返回 ONNX 的``模型原型``对象。
        返回 ir.serde.序列化模型(自身.模型)

[文档]    def 优化(self) -> None:
优化 ONNX 模型。

该方法通过执行常量折叠优化 ONNX 模型
消除图中的冗余。优化是在原地进行的。
```python
# 输入文本
input_text = '"""'

# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
    # 假设的翻译结果
    return text

# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
self.model = onnxscript_apis.optimize(self.model)

    def 保存(
        自身,
        目的地: 字符串 | 操作系统.PathLike,
        *,
        包含初始化器: 布尔类型 = True,
        将初始化器作为输入保留: 布尔类型 = 错误,
        外部数据: 布尔类型 | None = 无,
    ):
        将 ONNX 模型保存到指定的目标位置。

当 `external_data` 为 `True` 或模型大小超过 2GB 时，
权重被保存为外部数据，存储在单独的文件中。

初始化器（模型权重）序列化行为：
* `include_initializers=True`，`keep_initializers_as_inputs=False`（默认）：
初始化器包含在保存的模型中。
``include_initializers=True``, ``keep_initializers_as_inputs=True``:
初始化器包含在保存的模型中，并保留为模型输入。
如果您想在推理期间覆盖模型权重，
请选择此选项。
``include_initializers=False``, ``keep_initializers_as_inputs=False``:
初始化器不包括在保存的模型中，也不会列出
作为模型输入。如果您想将初始化器附加到 ONNX 模型中，请选择此选项
在单独的后期处理步骤中。
``include_initializers=False``, ``keep_initializers_as_inputs=True``:
初始化器不包括在保存的模型中，但作为模型输入列出。
选择此选项，如果您想在推理时提供初始化器并希望最小化保存模型的尺寸。
请选择此选项，如果您想在推理时提供初始化器并希望最小化保存模型的尺寸。

参数：
保存 ONNX 模型的路径。
是否在保存的模型中包含初始化器。
是否将初始化器作为输入保留在保存的模型中。
如果为 True，初始化器将作为模型的输入添加，这意味着它们可以被覆盖。
通过提供初始化器作为模型输入。
external_data: 是否将权重保存为外部数据到单独的文件中。

抛出异常：
TypeError: 如果 `external_data` 为 `True` 且 `destination` 不是一个文件路径。
```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)
```
        original_initializers = 复制.复制(自身.模型.图.初始化器)
        原始输入 = 复制.复制(自身.模型.图.输入)

        调整模型以根据选项
        如果 not 包含初始化器:
            自身.模型.图.初始化器.清晰()
        如果 将初始化器作为输入保留:
            自身.模型.图.输入.扩展(原始初始化器.值())  # type: ignore[arg-type]

        # 将模型保存到磁盘
        如果 (
            外部数据
            或 初始化计数器大小(自身.模型.图) > 大型模型阈值
        ):
            onnxscript_apis.带外部数据保存模型(自身.模型, 目的地)
        否则:
            ir.保存(自身.模型, 目的地)

        撤销模型更改
        如果 not 包含初始化器:
            自身.模型.图.初始化器.更新(原始初始化器)
        如果 将初始化器作为输入保留:
            自身.模型.图.输入.清晰()
            自身.模型.图.输入.扩展(原始输入)


[文档]    def apply_weights(self, state_dict: dict[str, torch.Tensor]) -> None:
将指定状态字典中的权重应用到 ONNX 模型上。

使用此方法替换 FakeTensors 或其他权重。

Args:
state_dict: 包含要应用到 ONNX 模型上的权重的状态字典。
"""
从 torch.onnx._internal.exporter 导入 _core

for name, tensor in state_dict.items():
如果 name 在 self.model.graph.initializers 中:
self.model.graph.initializers[name].const_value = _core.TorchTensor(
tensor, name
                )
else:
warnings.warn(
f"权重 '{name}' 在模型中未找到。跳过应用。"
category=torch.onnx.errors.OnnxExporterWarning,
stacklevel=1,
                )

[文档]    初始化推理会话(
self,
initializer: 可调用[
[str | bytes], ort.InferenceSession
] = _ort_session_initializer,
) -> None:
"""初始化 ONNX Runtime 推理会话。

参数：
初始化 ONNX Runtime 推理的函数
使用指定模型创建会话。默认情况下，它使用
func:`_ort_session_initializer` 函数。
"""
# TODO(justinchuby): 允许不同的推理选项
logger.debug("初始化推理会话。")
if (
byte_size := _count_initializer_size(self.model.graph)
) > _LARGE_MODEL_THRESHOLD:
logger.debug("模型初始化器大于 1.5GB (%s)。", 字节数)
# 将模型保存到临时文件中，如果太大
self._tempdir = tempfile.TemporaryDirectory(ignore_cleanup_errors=True)
model_path = os.path.join(self._tempdir.name, "model.onnx")
self.save(model_path, external_data=True)
model = model_path
else:
model = self.model_proto.SerializeToString()  # 忽略[赋值]

self._inference_session = initializer(model)
logger.debug("推理会话已初始化。")

[文档] def release(self) -> None:
释放推理会话。

您可以调用此方法来释放推理会话使用的资源。
```python
# 输入文本
input_text = '"""'

# 翻译函数（此处为示例，实际翻译功能需调用真实的翻译 API）
def translate_to_simplified_chinese(text):
    # 假设的翻译结果
    return text

# 输出翻译结果
translated_text = translate_to_simplified_chinese(input_text)
print(translated_text)
```
首先释放推理会话，以便可以删除模型文件
如果 self._inference_session 不是 None：
self._inference_session = None
gc.collect()
如果 self._tempdir 不是 None：
清理临时目录
将临时目录设置为 None

    def 重命名动态轴(
        自身,
        动态形状: 字典[字符串, 任意] | 元组[任意, ...] | 列表[任意]
    ) -> 无:
        "根据指定的 dynamic_axes 名称重命名模型中的动态轴。"
        重命名映射 = _dynamic_shapes.创建重命名映射(
            自身.模型.图.输入, dynamic_shapes
        )
        _ir_passes.重命名轴(自身.模型, 重命名映射)


def _处理参数(参数, kwargs) -> 元组[火炬.张量, ...]：
    """处理 ONNX 模型的输入参数。"""
    args = _展平输入(参数, kwargs)
    args = 从输入中移除 None(参数)
    args = 移除非张量(参数)
    args = _将复杂转换为真实表示(参数)
    返回 args


def _展平输入(model_args, 模型参数):
    展平参数, _ = _pytree.树扁平化((model_args, 模型参数))
    返回 展平的参数


def _从输入中移除 None(model_args):
    返回 元组(参数 为 参数 在 模型参数 如果 参数 是 not 无)


def _移除非张量(model_args):
    移除非张量输入参数。

Dynamo 不支持非张量输入参数（https://github.com/pytorch/pytorch/issues/99534）。

具体来说，它确实将输入放入了图，但没有任何节点消费它。
具体的值被嵌入到图中，作为目标节点的常量参数。元数据
在这种情况下建议应该重写模型代码，使其成为张量
输入值应该在运行时改变。我们可能需要进一步调查这个建议的可行性。
的可行性。

例如，

        def func(x, b=1.0):
            y = x + b
            z = y.relu()
            return (y, z)

        x = torch.randn(1, 1, 2, dtype=torch.float32)
        gm_fun, _ = dynamo.export(func, x, b=8.0, aten_graph=True, tracing_mode="real")

        # class GraphModule(torch.nn.Module):
        #     def forward(self, x, b):
        #         arg0: f32[1, 1, 2], arg1, = fx_pytree.tree_flatten_spec(([x, b], {}), self._in_spec)
#         # 文件：path/to/pytorch/test_constant_input.py:5, 代码：y = x + b
        #         add_tensor: f32[1, 1, 2] = torch.ops.aten.add.Tensor(arg0, 8.0);  arg0 = None

#         # 文件：path/to/pytorch/test_constant_input.py:6, 代码：z = y.relu()
        #         relu_default: f32[1, 1, 2] = torch.ops.aten.relu.default(add_tensor)
返回 pytree.tree_unflatten([add_tensor, relu_default], self._out_spec)

空 torch.fx.Node 输入导致与 PyTorch 不匹配的输入数量
在 ONNX 图中被忽略。因此，我们在这里删除了无用的输入。

```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)
```

    返回 元组(
        参数 为 参数 在 模型参数 如果 not isinstance(arg, (int, float, 布尔, 字符串))
    )


def 将复杂数据转换为实数表示(model_args):
    将复杂的数据类型张量转换为实数表示张量。

ONNX 不支持复杂 dtype 张量。因此，我们将复杂 dtype 张量
将真实表示张量（即具有额外维度的浮点类型张量）
表示复数的实部和虚部)。
```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)
```
    返回 元组(
        火炬.真实查看(arg.解决连词())
        如果 isinstance(arg, 火炬.张量) 和 arg.是复杂的()
        否则 参数
        为 参数 在 模型参数
    )