Bundled Program – 用于 ExecuTorch 模型验证的工具¶
介绍¶
BundledProgram
是核心 ExecuTorch program 的包装器,旨在帮助用户将测试用例与他们部署的模型捆绑在一起。BundledProgram
不一定是程序的核心部分,也无需用于程序执行,但对于各种其他用例(例如模型正确性评估,包括模型启用过程中的端到端测试)尤其重要。
总的来说,该过程可以分为两个阶段,并且在每个阶段我们都支持
生成阶段: 将测试 I/O 用例与 ExecuTorch program 捆绑在一起,序列化为 flatbuffer。
运行时阶段: 在运行时访问、执行和验证捆绑的测试用例。
生成阶段¶
该阶段主要侧重于 BundledProgram
的创建并将其作为 flatbuffer 文件转储到磁盘。主要过程如下
创建模型并生成其 ExecuTorch program。
构造
List[MethodTestSuite]
以记录所有需要捆绑的测试用例。使用已生成的模型和
List[MethodTestSuite]
生成BundledProgram
。序列化
BundledProgram
并将其转储到磁盘。
步骤 1:创建模型并生成其 ExecuTorch Program。¶
ExecuTorch Program 可以使用 ExecuTorch API 从用户模型生成。请参阅生成和生成示例 ExecuTorch 程序或导出到 ExecuTorch 教程。
步骤 2:构造 List[MethodTestSuite]
以保存测试信息¶
在 BundledProgram
中,我们创建了两个新类,MethodTestCase
和 MethodTestSuite
,以保存 ExecuTorch program 验证的基本信息。
MethodTestCase
表示单个测试用例。每个 MethodTestCase
包含单次执行的输入和预期输出。
MethodTestCase
- executorch.devtools.bundled_program.config.MethodTestCase.__init__(self, inputs, expected_outputs=None)
用于验证特定方法的单个测试用例
- 参数
inputs –
eager_model 使用特定推理方法进行一次执行所需的所有输入。
值得一提的是,虽然 bundled program 和 ET 运行时 API 都支持设置 torch.tensor 类型以外的输入,但方法中实际更新的只有 torch.tensor 类型的输入,其余输入只会检查它们是否与方法中的默认值匹配。
expected_outputs – 用于验证给定输入的预期输出。如果用户只想使用测试用例进行性能分析,则可以为 None。
- 返回
self
MethodTestSuite
包含单个方法的所有测试信息,包括表示方法名称的字符串以及所有测试用例的 List[MethodTestCase]
MethodTestSuite
- executorch.devtools.bundled_program.config.MethodTestSuite(method_name, test_cases)[source]
与方法验证相关的所有测试信息
- executorch.devtools.bundled_program.config.method_name
待验证方法的名称。
- executorch.devtools.bundled_program.config.test_cases
用于验证方法的所有测试用例。
由于每个模型可能包含多个推理方法,我们需要生成 List[MethodTestSuite]
来保存所有基本信息。
步骤 3:生成 BundledProgram
¶
我们在 executorch/devtools/bundled_program/core.py
下提供了 BundledProgram
类,用于将 ExecutorchProgram
-like 变量(包括 ExecutorchProgram
、MultiMethodExecutorchProgram
或 ExecutorchProgramManager
)与 List[MethodTestSuite]
捆绑在一起
BundledProgram
- executorch.devtools.bundled_program.core.BundledProgram.__init__(self, executorch_program, method_test_suites, pte_file_path=None)
通过将给定的程序和 method_test_suites 捆绑在一起创建 BundledProgram。
- 参数
executorch_program – 要捆绑的程序。
method_test_suites – 要捆绑的特定方法的测试用例。
pte_file_path – 如果未提供 executorch_program,则用于反序列化程序的 .pte 文件路径。
BundledProgram
的构造函数将在内部进行健全性检查,以查看给定的 List[MethodTestSuite]
是否符合给定 Program 的要求。具体来说
List[MethodTestSuite]
中每个MethodTestSuite
的 method_names 也应该在 program 中。请注意,无需为 Program 中的每个方法设置测试用例。每个测试用例的元数据应满足相应推理方法输入的要求。
步骤 4:将 BundledProgram
序列化为 Flatbuffer。¶
为了序列化 BundledProgram
以供运行时 API 使用,我们提供了两个 API,均在 executorch/devtools/bundled_program/serialize/__init__.py
下。
序列化和反序列化
- executorch.devtools.bundled_program.serialize.serialize_from_bundled_program_to_flatbuffer(bundled_program)[source]
将 BundledProgram 序列化为 FlatBuffer 二进制格式。
- 参数
bundled_program (BundledProgram) – 要序列化的 BundledProgram 变量。
- 返回
序列化后的 FlatBuffer 二进制数据(字节格式)。
生成示例¶
下面是一个流程,重点说明如何从 PyTorch 模型及其要测试的代表性输入生成 BundledProgram
。
import torch
from executorch.exir import to_edge_transform_and_lower
from executorch.devtools import BundledProgram
from executorch.devtools.bundled_program.config import MethodTestCase, MethodTestSuite
from executorch.devtools.bundled_program.serialize import (
serialize_from_bundled_program_to_flatbuffer,
)
from torch.export import export, export_for_training
# 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.register_buffer('a', 3 * torch.ones(2, 2, dtype=torch.int32))
self.register_buffer('b', 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(
export_for_training(model, capture_input).module(),
capture_input,
)
# Emit the traced method into ET Program.
et_program = to_edge_transform_and_lower(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.devtools.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 Program 指针¶
我们需要 ExecuTorch program 的指针来执行。为了统一加载和执行 BundledProgram
和 Program flatbuffer 的过程,我们为此创建了一个 API executorch::bundled_program::get_program_data
。请参阅此 API 的示例用法。
加载捆绑的输入到方法¶
要使用捆绑的输入执行程序,我们需要将捆绑的输入加载到方法中。这里我们提供了一个名为 executorch::bundled_program::load_bundled_input
的 API。请参阅此 API 的示例用法。
验证方法的输出。¶
我们调用 executorch::bundled_program::verify_method_outputs
来验证方法的输出是否与捆绑的预期输出一致。请参阅此 API 的示例用法。
运行时示例¶
请查看我们用于 bundled program 的示例运行器。您可以运行这些命令来测试您在上一步中生成的 BundledProgram 二进制文件(.bpte
)
cd executorch
./examples/devtools/build_example_runner.sh
./cmake-out/examples/devtools/example_runner --bundled_program_path {your-bpte-file} --output_verification
运行上述代码片段后,预期不会看到任何输出。
有关运行器应如何构建的详细示例,请参阅我们的示例运行器。
常见错误¶
如果 List[MethodTestSuites]
与 Program
不匹配,将抛出错误。以下是两种常见情况
测试输入与模型要求不匹配。¶
PyTorch 模型的每个推理方法对其输入都有自己的要求,例如输入数量、每个输入的数据类型等。如果测试输入不符合要求,BundledProgram
将抛出错误。
以下是测试输入的数据类型不符合模型要求的示例
import torch
from executorch.exir import to_edge
from executorch.devtools import BundledProgram
from executorch.devtools.bundled_program.config import MethodTestCase, MethodTestSuite
from torch.export import export, export_for_training
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(
export_for_training(model, inputs).module(),
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/devtools/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/devtools/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.devtools import BundledProgram
from executorch.devtools.bundled_program.config import MethodTestCase, MethodTestSuite
from torch.export import export, export_for_training
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(
export_for_training(model, inputs).module(),
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)
错误信息
All method names in bundled config should be found in program.execution_plan, but {'MISSING_METHOD_NAME'} does not include.
---------------------------------------------------------------------------
AssertionError Traceback (most recent call last)
Cell In[3], line 73
70 method_test_suites[0].method_name = "MISSING_METHOD_NAME"
72 # Generate BundledProgram
---> 73 bundled_program = create_bundled_program(program, method_test_suites)
File /executorch/devtools/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/devtools/bundled_program/core.py:141, in assert_valid_bundle(program, method_test_suites)
138 method_name_of_program = {e.name for e in program.execution_plan}
139 method_name_of_test_suites = {t.method_name for t in method_test_suites}
--> 141 assert method_name_of_test_suites.issubset(
142 method_name_of_program
143 ), f"All method names in bundled config should be found in program.execution_plan, \
144 but {str(method_name_of_test_suites - method_name_of_program)} does not include."
146 # check if method_tesdt_suites has been sorted in ascending alphabetical order of method name.
147 for test_suite_id in range(1, len(method_test_suites)):
AssertionError: All method names in bundled config should be found in program.execution_plan, but {'MISSING_METHOD_NAME'} does not include.