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

介绍¶

BundledProgram 是核心 ExecuTorch 程序的包装器，旨在帮助用户使用他们部署的模型包装测试用例。BundledProgram 不一定是程序的核心部分，也不是其执行所必需的，但对于各种其他用例（例如模型正确性评估，包括模型启动过程中的 e2e 测试）尤为重要。

总体而言，该程序可以分为两个阶段，在每个阶段我们都支持：

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

Emit 阶段¶

此阶段主要侧重于创建 a 并将其作为 flatbuffer 文件转储到磁盘上。主要步骤如下：BundledProgram

创建一个模型并发出其 ExecuTorch 程序。
构造一个来记录所有需要捆绑的信息。BundledConfig
使用发出的模型和生成。BundledProgramBundledConfig
序列化并将其转储到磁盘。BundledProgram

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

ExecuTorch 程序可以通过 ExecuTorch API 从用户的模型发出。按照生成示例 ExecuTorch 程序或导出到 ExecuTorch 教程进行操作。

第 2 步：构造`BundledConfig`¶

BundledConfig是一个类，其中包含要捆绑用于模型验证的所有信息。这是要创建的构造函数 API ：executorch/bundled_program/config.pyBundledConfig

第 3 步：生成`BundledProgram`¶

我们通过将发出的 ExecuTorch 程序与 bundled_config 捆绑在一起来生成 API 下提供 API：create_bundled_programexecutorch/bundled_program/core.pyBundledProgram

create_bundled_program将在内部进行 sannity 检查，以查看给定的 BundledConfig 是否符合给定的 Program 的要求。具体说来：

我们为其创建 BundledConfig 的方法的名称也应该在 program 中。请注意，无需为 Program 中的每个方法设置测试用例。
每个测试用例的元数据应满足相应推理方法输入的要求。

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

为了序列化以使运行时 API 使用它，我们提供了两个 API，它们都位于 .BundledProgramexecutorch/bundled_program/serialize/__init__.py

发出示例¶

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

import torch
from torch.export import export

from executorch.bundled_program.config import BundledConfig
from executorch.bundled_program.core import create_bundled_program
from executorch.bundled_program.serialize import serialize_from_bundled_program_to_flatbuffer

from executorch.exir import to_edge


# 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 encode(
        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

    def decode(
        self, x: torch.Tensor, q: torch.Tensor
    ) -> torch.Tensor:
        y = x * q
        torch.add(y, self.b, out=y)
        return y

# Inference method names of SampleModel we want to bundle testcases to.
# Notices that we do not need to bundle testcases for every inference methods.
method_names = ["encode", "decode"]
model = SampleModel()

capture_inputs = {
    m_name: (
        (torch.rand(2, 2) - 0.5).to(dtype=torch.int32),
        (torch.rand(2, 2) - 0.5).to(dtype=torch.int32),
    )
    for m_name in method_names
}

# Find each method of model needs to be traced my its name, export its FX Graph.
method_graphs = {
    m_name: export(getattr(model, m_name), capture_inputs[m_name])
    for m_name in method_names
}

# Emit the traced methods into ET Program.
program = to_edge(method_graphs).to_executorch().executorch_program

# Step 2: Construct BundledConfig

# number of input sets to be verified
n_input = 10

# Input sets to be verified for each inference methods.
inputs = [
    # The below list is all inputs for a single inference method.
    [
        # 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)
    ]
    for _ in method_names
]

# Expected outputs align with inputs.
expected_outputs = [
    [[getattr(model, m_name)(*x)] for x in inputs[i]]
    for i, m_name in enumerate(method_names)
]

# Create BundledConfig
bundled_config = BundledConfig(
    method_names, inputs, expected_outputs
)


# Step 3: Generate BundledProgram

bundled_program = create_bundled_program(program, bundled_config)

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

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

from executorch.bundled_program.serialize import deserialize_from_flatbuffer_to_bundled_program
save_path = "bundled_program.bp"
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 来处理它的关键部分。

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

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

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

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::util::GetProgramData(
    buff_ptr.get(), buff_len, &program_ptr, &program_len);
ET_CHECK_MSG(
    status == Error::Ok,
    "GetProgramData() failed with status 0x%" PRIx32,
    status);

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

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

验证方法的输出。¶

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

运行时示例¶

这里我们提供了一个有关如何逐步运行捆绑程序的示例。大部分代码是从 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::util::LoadBundledInput(
        *method,
        program_data.bundled_program_data(),
        &bundled_input_allocator,
        method_name,
        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::util::VerifyResultWithBundledExpectedOutput(
        *method,
        program_data.bundled_program_data(),
        &bundled_input_allocator,
        method_name,
        FLAGS_testset_idx,
        FLAGS_rtol,
        FLAGS_atol);
ET_CHECK_MSG(
    status == Error::Ok,
    "Bundle verification failed with status 0x%" PRIx32,
    status);

常见错误¶

如果与 .以下是两种常见情况：BundledConfigProgram

测试输入与模型的要求不匹配。¶

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

以下是测试输入的 dtype 不满足模型要求的示例：

import torch
from torch.export import export

from executorch.bundled_program.config import BundledConfig
from executorch.bundled_program.core import create_bundled_program

from executorch.exir import to_edge


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)
print(model(inputs))

# Find each method of model needs to be traced my its name, export its FX Graph.
method_graphs = {
    m_name: export(getattr(model, m_name), (inputs, ))
    for m_name in method_names
}

# Emit the traced methods into ET Program.
program = to_edge(method_graphs).to_executorch().executorch_program


# number of input sets to be verified
n_input = 10

# All Input sets to be verified.
inputs = [
    [
        # 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)
    ]
]

# Expected outputs align with inputs.
expected_outpus = [
    [[model(*x)] for x in inputs[0]]
]

bundled_config = BundledConfig(method_names, inputs, expected_outpus)

bundled_program = create_bundled_program(program, bundled_config)

引发的错误

The input tensor tensor([[ 0,  3],
        [-3, -3]], dtype=torch.int32) dtype shall be torch.float32, but now is torch.int32
---------------------------------------------------------------------------
AssertionError                            Traceback (most recent call last)
expected_outpus = [
   [[model(*x)] for x in inputs[0]]
]
bundled_config = BundledConfig(method_names, inputs, expected_outpus)
---> 63 bundled_program = create_bundled_program(program, bundled_config)
File /executorch/bundled_program/core.py:270, in create_bundled_program(program, bundled_config)
def create_bundled_program(
   program: Program,
   bundled_config: BundledConfig,
) -> BundledProgram:
   """Create BundledProgram by bundling the given program and bundled_config together.
    264
   Args:
       program: The program to be bundled.
       bundled_config: The config to be bundled.
   """
--> 270     assert_valid_bundle(program, bundled_config)
   execution_plan_tests: List[BundledExecutionPlanTest] = []
   # Emit data and metadata of bundled tensor
File /executorch/bundled_program/core.py:224, in assert_valid_bundle(program, bundled_config)
# type of tensor input should match execution plan
if type(cur_plan_test_inputs[j]) == torch.Tensor:
   # pyre-fixme[16]: Undefined attribute [16]: Item `bool` of `typing.Union[bool, float, int, torch._tensor.Tensor]`
   # has no attribute `dtype`.
--> 224     assert cur_plan_test_inputs[j].dtype == get_input_dtype(
       program, program_plan_id, j
   ), "The input tensor {} dtype shall be {}, but now is {}".format(
       cur_plan_test_inputs[j],
       get_input_dtype(program, program_plan_id, j),
       cur_plan_test_inputs[j].dtype,
   )
elif type(cur_plan_test_inputs[j]) in (
   int,
   bool,
   float,
):
   assert type(cur_plan_test_inputs[j]) == get_input_type(
       program, program_plan_id, j
   ), "The input primitive dtype shall be {}, but now is {}".format(
       get_input_type(program, program_plan_id, j),
       type(cur_plan_test_inputs[j]),
   )
AssertionError: The input tensor tensor([[ 0,  3],
        [-3, -3]], dtype=torch.int32) dtype shall be torch.float32, but now is torch.int32

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

另一个常见错误是 Model 中不存在的方法名称。将引发错误并显示不存在的方法名称：BundledConfigBundledProgram

import torch
from torch.export import export

from executorch.bundled_program.config import BundledConfig
from executorch.bundled_program.core import create_bundled_program

from executorch.exir import to_edge



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_graphs = {
    m_name: export(getattr(model, m_name), (inputs, ))
    for m_name in method_names
}

# Emit the traced methods into ET Program.
program = to_edge(method_graphs).to_executorch().executorch_program

# Number of input sets to be verified
n_input = 10

# All Input sets to be verified.
inputs = [
    [
        [
            torch.randint(-5, 5, (2, 2), dtype=torch.float),
        ]
        for _ in range(n_input)
    ]
]

# Expected outputs align with inputs.
expected_outpus = [
    [[model(*x)] for x in inputs[0]]
]


# NOTE: MISSING_METHOD_NAME is not an inference method in the above model.
wrong_method_names = ['MISSING_METHOD_NAME']

bundled_config = BundledConfig(wrong_method_names, inputs, expected_outpus)

bundled_program = create_bundled_program(program, bundled_config)

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

介绍¶

Emit 阶段¶

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

第 2 步：构造`BundledConfig`¶

第 3 步：生成`BundledProgram`¶

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

发出示例¶

运行时阶段¶

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

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

验证方法的输出。¶

运行时示例¶

常见错误¶

测试输入与模型的要求不匹配。¶

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

文档

教程

资源

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

介绍¶

Emit 阶段¶

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

第 2 步：构造BundledConfig¶

第 3 步：生成BundledProgram¶

第 4 步：序列化为 Flatbuffer。BundledProgram¶

发出示例¶

运行时阶段¶

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

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

验证方法的输出。¶

运行时示例¶

常见错误¶

测试输入与模型的要求不匹配。¶

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

文档

教程

资源

第 2 步：构造`BundledConfig`¶

第 3 步：生成`BundledProgram`¶

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

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

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