捆绑程序 - ExecuTorch 模型验证工具¶

介绍¶

BundledProgram 是围绕核心 ExecuTorch 程序的包装器，旨在帮助用户将测试用例与他们部署的模型包装在一起。 BundledProgram 不一定是程序的核心部分，也不需要执行，但在其他各种用例中尤其重要，例如模型正确性评估，包括模型启动过程中的端到端测试。

总的来说，该过程可以分为两个阶段，并且在每个阶段中，我们都支持

**发射阶段**：将测试 I/O 用例与 ExecuTorch 程序捆绑在一起，序列化为 flatbuffer。
**运行时阶段**：在运行时访问、执行和验证捆绑的测试用例。

发射阶段¶

此阶段主要侧重于创建 BundledProgram 并将其作为 flatbuffer 文件转储到磁盘。主要程序如下

创建模型并发射其 ExecuTorch 程序。
构建一个 List[MethodTestSuite] 来记录需要捆绑的所有测试用例。
使用发射的模型和 List[MethodTestSuite] 生成 BundledProgram。
序列化 BundledProgram 并将其转储到磁盘。

步骤 1：创建模型并发射其 ExecuTorch 程序。¶

可以使用 ExecuTorch API 从用户的模型发射 ExecuTorch 程序。请参阅生成示例 ExecuTorch 程序或导出到 ExecuTorch 教程。

步骤 2：构建 `List[MethodTestSuite]` 来保存测试信息¶

在 BundledProgram 中，我们创建了两个新类， MethodTestCase 和 MethodTestSuite，用于保存 ExecuTorch 程序验证的基本信息。

MethodTestCase 表示单个测试用例。每个 MethodTestCase 包含单个执行的输入和预期输出。

MethodTestSuite 包含单个方法的所有测试信息，包括表示方法名称的字符串以及用于所有测试用例的 List[MethodTestCase]

由于每个模型可能有多种推理方法，因此我们需要生成 List[MethodTestSuite] 来保存所有基本信息。

步骤 3：生成 `BundledProgram`¶

我们在 executorch/sdk/bundled_program/core.py 下提供 BundledProgram 类来捆绑类似于 ExecutorchProgram 的变量，包括 ExecutorchProgram、MultiMethodExecutorchProgram 或 ExecutorchProgramManager，以及 List[MethodTestSuite]

BundledProgram 的构造函数将在内部进行完整性检查，以查看给定的 List[MethodTestSuite] 是否与给定程序的要求匹配。具体来说

List[MethodTestSuite] 中每个 MethodTestSuite 的方法名称也应该在 program 中。请注意，无需为程序中的每个方法设置测试用例。
每个测试用例的元数据应满足相应推理方法输入的要求。

步骤 4：将 `BundledProgram` 序列化为 Flatbuffer。¶

为了将 BundledProgram 序列化以使运行时 API 使用它，我们提供了两个 API，这两个 API 都位于 executorch/sdk/bundled_program/serialize/__init__.py 下。

发出示例¶

这是一个流程，重点介绍了如何在给定 PyTorch 模型和我们想要测试的代表性输入以及模型的输入的情况下，生成 BundledProgram。

import torch

from executorch.exir import to_edge
from executorch.sdk import BundledProgram

from executorch.sdk.bundled_program.config import MethodTestCase, MethodTestSuite
from executorch.sdk.bundled_program.serialize import (
    serialize_from_bundled_program_to_flatbuffer,
)
from torch._export import capture_pre_autograd_graph
from torch.export import export


# Step 1: ExecuTorch Program Export
class SampleModel(torch.nn.Module):
    """An example model with multi-methods. Each method has multiple input and single output"""

    def __init__(self) -> None:
        super().__init__()
        self.a: torch.Tensor = 3 * torch.ones(2, 2, dtype=torch.int32)
        self.b: torch.Tensor = 2 * torch.ones(2, 2, dtype=torch.int32)

    def forward(self, x: torch.Tensor, q: torch.Tensor) -> torch.Tensor:
        z = x.clone()
        torch.mul(self.a, x, out=z)
        y = x.clone()
        torch.add(z, self.b, out=y)
        torch.add(y, q, out=y)
        return y


# Inference method name of SampleModel we want to bundle testcases to.
# Notices that we do not need to bundle testcases for every inference methods.
method_name = "forward"
model = SampleModel()

# Inputs for graph capture.
capture_input = (
    (torch.rand(2, 2) - 0.5).to(dtype=torch.int32),
    (torch.rand(2, 2) - 0.5).to(dtype=torch.int32),
)

# Export method's FX Graph.
method_graph = export(
    capture_pre_autograd_graph(model, capture_input),
    capture_input,
)


# Emit the traced method into ET Program.
et_program = to_edge(method_graph).to_executorch()

# Step 2: Construct MethodTestSuite for Each Method

# Prepare the Test Inputs.

# Number of input sets to be verified
n_input = 10

# Input sets to be verified.
inputs = [
    # Each list below is a individual input set.
    # The number of inputs, dtype and size of each input follow Program's spec.
    [
        (torch.rand(2, 2) - 0.5).to(dtype=torch.int32),
        (torch.rand(2, 2) - 0.5).to(dtype=torch.int32),
    ]
    for _ in range(n_input)
]

# Generate Test Suites
method_test_suites = [
    MethodTestSuite(
        method_name=method_name,
        test_cases=[
            MethodTestCase(
                inputs=input,
                expected_outputs=(getattr(model, method_name)(*input), ),
            )
            for input in inputs
        ],
    ),
]

# Step 3: Generate BundledProgram
bundled_program = BundledProgram(et_program, method_test_suites)

# Step 4: Serialize BundledProgram to flatbuffer.
serialized_bundled_program = serialize_from_bundled_program_to_flatbuffer(
    bundled_program
)
save_path = "bundled_program.bpte"
with open(save_path, "wb") as f:
    f.write(serialized_bundled_program)

如果需要，我们还可以从 flatbuffer 文件重新生成 BundledProgram

from executorch.sdk.bundled_program.serialize import deserialize_from_flatbuffer_to_bundled_program
save_path = "bundled_program.bpte"
with open(save_path, "rb") as f:
    serialized_bundled_program = f.read()

regenerate_bundled_program = deserialize_from_flatbuffer_to_bundled_program(serialized_bundled_program)

运行时阶段¶

此阶段主要侧重于使用捆绑的输入执行模型，并将模型的输出与捆绑的预期输出进行比较。我们提供了多个 API 来处理它的关键部分。

从 `BundledProgram` 缓冲区获取 ExecuTorch 程序指针¶

我们需要 ExecuTorch 程序的指针来执行。为了统一加载和执行 BundledProgram 和 Program flatbuffer 的过程，我们创建了一个 API

以下是如何使用 GetProgramData API 的示例

std::shared_ptr<char> buff_ptr;
size_t buff_len;

// FILE_PATH here can be either BundledProgram or Program flatbuffer file.
Error status = torch::executor::util::read_file_content(
    FILE_PATH, &buff_ptr, &buff_len);
ET_CHECK_MSG(
    status == Error::Ok,
    "read_file_content() failed with status 0x%" PRIx32,
    status);

const void* program_ptr;
size_t program_len;
status = torch::executor::bundled_program::GetProgramData(
    buff_ptr.get(), buff_len, &program_ptr, &program_len);
ET_CHECK_MSG(
    status == Error::Ok,
    "GetProgramData() failed with status 0x%" PRIx32,
    status);

将捆绑的输入加载到方法¶

要在捆绑的输入上执行程序，我们需要将捆绑的输入加载到方法中。这里我们提供了一个名为 torch::executor::bundled_program::LoadBundledInput 的 API

验证方法的输出。¶

我们调用 torch::executor::bundled_program::VerifyResultWithBundledExpectedOutput 来使用捆绑的预期输出验证方法的输出。以下是此 API 的详细信息

运行时示例¶

这里我们提供一个关于如何逐步运行捆绑程序的示例。大多数代码都借用自 executor_runner，如果您需要更多信息和上下文，请查看该文件

// method_name is the name for the method we want to test
// memory_manager is the executor::MemoryManager variable for executor memory allocation.
// program is the ExecuTorch program.
Result<Method> method = program->load_method(method_name, &memory_manager);

ET_CHECK_MSG(
    method.ok(),
    "load_method() failed with status 0x%" PRIx32,
    method.error());

// Load testset_idx-th input in the buffer to plan
status = torch::executor::bundled_program::LoadBundledInput(
        *method,
        program_data.bundled_program_data(),
        FLAGS_testset_idx);
ET_CHECK_MSG(
    status == Error::Ok,
    "LoadBundledInput failed with status 0x%" PRIx32,
    status);

// Execute the plan
status = method->execute();
ET_CHECK_MSG(
    status == Error::Ok,
    "method->execute() failed with status 0x%" PRIx32,
    status);

// Verify the result.
status = torch::executor::bundled_program::VerifyResultWithBundledExpectedOutput(
        *method,
        program_data.bundled_program_data(),
        FLAGS_testset_idx,
        FLAGS_rtol,
        FLAGS_atol);
ET_CHECK_MSG(
    status == Error::Ok,
    "Bundle verification failed with status 0x%" PRIx32,
    status);

常见错误¶

如果 List[MethodTestSuites] 与 Program 不匹配，将引发错误。以下两种情况很常见

测试输入不满足模型的要求。¶

PyTorch 模型的每个推理方法对输入都有自己的要求，例如输入的数量、每个输入的数据类型等。如果测试输入不满足要求，BundledProgram 将引发错误。

以下是测试输入的数据类型不满足模型要求的示例

import torch

from executorch.exir import to_edge
from executorch.sdk import BundledProgram

from executorch.sdk.bundled_program.config import MethodTestCase, MethodTestSuite
from torch.export import export


class Module(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.a = 3 * torch.ones(2, 2, dtype=torch.float)
        self.b = 2 * torch.ones(2, 2, dtype=torch.float)

    def forward(self, x):
        out_1 = torch.ones(2, 2, dtype=torch.float)
        out_2 = torch.ones(2, 2, dtype=torch.float)
        torch.mul(self.a, x, out=out_1)
        torch.add(out_1, self.b, out=out_2)
        return out_2


model = Module()
method_names = ["forward"]

inputs = (torch.ones(2, 2, dtype=torch.float), )

# Find each method of model needs to be traced my its name, export its FX Graph.
method_graph = export(
    capture_pre_autograd_graph(model, inputs),
    inputs,
)

# Emit the traced methods into ET Program.
et_program = to_edge(method_graph).to_executorch()

# number of input sets to be verified
n_input = 10

# Input sets to be verified for each inference methods.
# To simplify, here we create same inputs for all methods.
inputs = {
    # Inference method name corresponding to its test cases.
    m_name: [
        # NOTE: executorch program needs torch.float, but here is torch.int
        [
            torch.randint(-5, 5, (2, 2), dtype=torch.int),
        ]
        for _ in range(n_input)
    ]
    for m_name in method_names
}

# Generate Test Suites
method_test_suites = [
    MethodTestSuite(
        method_name=m_name,
        test_cases=[
            MethodTestCase(
                inputs=input,
                expected_outputs=(getattr(model, m_name)(*input),),
            )
            for input in inputs[m_name]
        ],
    )
    for m_name in method_names
]

# Generate BundledProgram

bundled_program = BundledProgram(et_program, method_test_suites)

引发的错误

The input tensor tensor([[-2,  0],
        [-2, -1]], dtype=torch.int32) dtype shall be torch.float32, but now is torch.int32
---------------------------------------------------------------------------
AssertionError                            Traceback (most recent call last)
Cell In[1], line 72
     56 method_test_suites = [
     57     MethodTestSuite(
     58         method_name=m_name,
   (...)
     67     for m_name in method_names
     68 ]
     70 # Step 3: Generate BundledProgram
---> 72 bundled_program = create_bundled_program(program, method_test_suites)
File /executorch/sdk/bundled_program/core.py:276, in create_bundled_program(program, method_test_suites)
    264 """Create bp_schema.BundledProgram by bundling the given program and method_test_suites together.
    265
    266 Args:
   (...)
    271     The `BundledProgram` variable contains given ExecuTorch program and test cases.
    272 """
    274 method_test_suites = sorted(method_test_suites, key=lambda x: x.method_name)
--> 276 assert_valid_bundle(program, method_test_suites)
    278 bundled_method_test_suites: List[bp_schema.BundledMethodTestSuite] = []
    280 # Emit data and metadata of bundled tensor
File /executorch/sdk/bundled_program/core.py:219, in assert_valid_bundle(program, method_test_suites)
    215 # type of tensor input should match execution plan
    216 if type(cur_plan_test_inputs[j]) == torch.Tensor:
    217     # pyre-fixme[16]: Undefined attribute [16]: Item `bool` of `typing.Union[bool, float, int, torch._tensor.Tensor]`
    218     # has no attribute `dtype`.
--> 219     assert cur_plan_test_inputs[j].dtype == get_input_dtype(
    220         program, program_plan_id, j
    221     ), "The input tensor {} dtype shall be {}, but now is {}".format(
    222         cur_plan_test_inputs[j],
    223         get_input_dtype(program, program_plan_id, j),
    224         cur_plan_test_inputs[j].dtype,
    225     )
    226 elif type(cur_plan_test_inputs[j]) in (
    227     int,
    228     bool,
    229     float,
    230 ):
    231     assert type(cur_plan_test_inputs[j]) == get_input_type(
    232         program, program_plan_id, j
    233     ), "The input primitive dtype shall be {}, but now is {}".format(
    234         get_input_type(program, program_plan_id, j),
    235         type(cur_plan_test_inputs[j]),
    236     )
AssertionError: The input tensor tensor([[-2,  0],
        [-2, -1]], dtype=torch.int32) dtype shall be torch.float32, but now is torch.int32

`BundleConfig` 中的方法名称不存在。¶

另一个常见错误是任何 MethodTestSuite 中的方法名称在模型中不存在。 BundledProgram 将引发错误并显示不存在的方法名称

import torch

from executorch.exir import to_edge
from executorch.sdk import BundledProgram

from executorch.sdk.bundled_program.config import MethodTestCase, MethodTestSuite
from torch.export import export


class Module(torch.nn.Module):
    def __init__(self):
        super().__init__()
        self.a = 3 * torch.ones(2, 2, dtype=torch.float)
        self.b = 2 * torch.ones(2, 2, dtype=torch.float)

    def forward(self, x):
        out_1 = torch.ones(2, 2, dtype=torch.float)
        out_2 = torch.ones(2, 2, dtype=torch.float)
        torch.mul(self.a, x, out=out_1)
        torch.add(out_1, self.b, out=out_2)
        return out_2


model = Module()
method_names = ["forward"]

inputs = (torch.ones(2, 2, dtype=torch.float),)

# Find each method of model needs to be traced my its name, export its FX Graph.
method_graph = export(
    capture_pre_autograd_graph(model, inputs),
    inputs,
)

# Emit the traced methods into ET Program.
et_program = to_edge(method_graph).to_executorch()

# number of input sets to be verified
n_input = 10

# Input sets to be verified for each inference methods.
# To simplify, here we create same inputs for all methods.
inputs = {
    # Inference method name corresponding to its test cases.
    m_name: [
        [
            torch.randint(-5, 5, (2, 2), dtype=torch.float),
        ]
        for _ in range(n_input)
    ]
    for m_name in method_names
}

# Generate Test Suites
method_test_suites = [
    MethodTestSuite(
        method_name=m_name,
        test_cases=[
            MethodTestCase(
                inputs=input,
                expected_outputs=(getattr(model, m_name)(*input),),
            )
            for input in inputs[m_name]
        ],
    )
    for m_name in method_names
]

# NOTE: MISSING_METHOD_NAME is not an inference method in the above model.
method_test_suites[0].method_name = "MISSING_METHOD_NAME"

# Generate BundledProgram
bundled_program = BundledProgram(et_program, method_test_suites)

捆绑程序 - ExecuTorch 模型验证工具¶

介绍¶

发射阶段¶

步骤 1：创建模型并发射其 ExecuTorch 程序。¶

步骤 2：构建 `List[MethodTestSuite]` 来保存测试信息¶

步骤 3：生成 `BundledProgram`¶

步骤 4：将 `BundledProgram` 序列化为 Flatbuffer。¶

发出示例¶

运行时阶段¶

从 `BundledProgram` 缓冲区获取 ExecuTorch 程序指针¶

将捆绑的输入加载到方法¶

验证方法的输出。¶

运行时示例¶

常见错误¶

测试输入不满足模型的要求。¶

`BundleConfig` 中的方法名称不存在。¶

文档

教程

资源

捆绑程序 - ExecuTorch 模型验证工具¶

介绍¶

发射阶段¶

步骤 1：创建模型并发射其 ExecuTorch 程序。¶

步骤 2：构建 List[MethodTestSuite] 来保存测试信息¶

步骤 3：生成 BundledProgram¶

步骤 4：将 BundledProgram 序列化为 Flatbuffer。¶

发出示例¶

运行时阶段¶

从 BundledProgram 缓冲区获取 ExecuTorch 程序指针¶

将捆绑的输入加载到方法¶

验证方法的输出。¶

运行时示例¶

常见错误¶

测试输入不满足模型的要求。¶

BundleConfig 中的方法名称不存在。¶

文档

教程

资源

步骤 2：构建 `List[MethodTestSuite]` 来保存测试信息¶

步骤 3：生成 `BundledProgram`¶

步骤 4：将 `BundledProgram` 序列化为 Flatbuffer。¶

从 `BundledProgram` 缓冲区获取 ExecuTorch 程序指针¶

`BundleConfig` 中的方法名称不存在。¶