捆綁程式 (Bundled Program) – 用於 ExecuTorch 模型驗證的工具¶
簡介¶
BundledProgram
是核心 ExecuTorch 程式的封裝器,旨在協助使用者將測試案例與他們部署的模型進行封裝。BundledProgram
不一定是程式的核心部分,也不是其執行所必需的,但對於各種其他使用案例特別重要,例如模型正確性評估,包括模型啟動過程中的端對端測試。
總體而言,這個程序可以分成兩個階段,而我們在每個階段都提供支援:
Emit 階段:將測試 I/O 案例與 ExecuTorch 程式捆綁在一起,並序列化為 flatbuffer。
Runtime 階段:在運行時訪問、執行和驗證捆綁的測試案例。
Emit 階段¶
此階段主要著重於建立 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
包含單次執行所需的輸入和預期輸出。
MethodTestCase
- executorch.devtools.bundled_program.config.MethodTestCase.__init__(self, inputs, expected_outputs=None)
用於驗證特定方法的單個測試案例
- 參數
inputs –
具有特定推論方法的 eager_model 單次執行所需的所有輸入。
值得一提的是,雖然捆綁程式和 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
的變數,包括 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]
是否符合給定程式的需求。具體來說:
List[MethodTestSuite]
中每個MethodTestSuite
的 method_names 也應該在程式中。請注意,無需為程式中的每個方法設定測試案例。每個測試案例的元資料應滿足相應推論方法輸入的要求。
步驟 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
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.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(
export_for_training(model, capture_input).module(),
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.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 程式指標¶
我們需要指向 ExecuTorch 程式的指標才能執行。為了統一加載和執行 BundledProgram
和 Program flatbuffer 的過程,我們建立了一個 API
get_program_data
警告
doxygenfunction: 無法在專案 "ExecuTorch" 的 doxygen xml 輸出中找到函數 "::executorch::bundled_program::get_program_data",目錄為:../build/xml/
以下是如何使用 get_program_data
API 的範例
// Assume that the user has read the contents of the file into file_data using
// whatever method works best for their application. The file could contain
// either BundledProgram data or Program data.
void* file_data = ...;
size_t file_data_len = ...;
// If file_data contains a BundledProgram, get_program_data() will return a
// pointer to the Program data embedded inside it. Otherwise it will return
// file_data, which already pointed to Program data.
const void* program_ptr;
size_t program_len;
status = executorch::bundled_program::get_program_data(
file_data, file_data_len, &program_ptr, &program_len);
ET_CHECK_MSG(
status == Error::Ok,
"get_program_data() failed with status 0x%" PRIx32,
status);
將綁定輸入加載到方法¶
為了在綁定的輸入上執行程式,我們需要將綁定的輸入加載到方法中。 在此,我們提供了一個名為 executorch::bundled_program::load_bundled_input
的 API
load_bundled_input
警告
doxygenfunction: 無法在專案 "ExecuTorch" 的 doxygen xml 輸出中找到函數 "::executorch::bundled_program::load_bundled_input",目錄為:../build/xml/
驗證方法的輸出。¶
我們呼叫 executorch::bundled_program::verify_method_outputs
以使用綁定的預期輸出驗證方法的輸出。 以下是此 API 的詳細資訊
verify_method_outputs
警告
doxygenfunction: 無法在專案 "ExecuTorch" 的 doxygen xml 輸出中找到函數 "::executorch::bundled_program::verify_method_outputs",目錄為:../build/xml/
執行階段範例¶
在此,我們提供一個關於如何逐步執行綁定程式的範例。 大部分程式碼取自 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 = executorch::bundled_program::load_bundled_input(
*method,
program_data.bundled_program_data(),
FLAGS_testset_idx);
ET_CHECK_MSG(
status == Error::Ok,
"load_bundled_input 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 = executorch::bundled_program::verify_method_outputs(
*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 模型的每個推論方法都有自己的輸入要求,例如輸入數量、每個輸入的 dtype 等。 如果測試輸入不符合要求,BundledProgram
會引發錯誤。
以下是測試輸入的 dtype 不符合模型要求的範例
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.