0
# Backend Integration
1
2
Abstract interfaces for implementing ONNX model execution backends, enabling custom runtime integration and testing frameworks. This module provides the foundation for creating execution engines that can run ONNX models.
3
4
## Capabilities
5
6
### Backend Base Classes
7
8
Abstract base classes for implementing ONNX execution backends.
9
10
```python { .api }
11
class Backend:
12
"""
13
Abstract base class for ONNX execution backends.
14
15
Provides interface for preparing and running ONNX models
16
on various compute devices and runtimes.
17
"""
18
19
@classmethod
20
def is_compatible(cls, model, device="CPU", **kwargs):
21
"""
22
Check if backend can execute the given model.
23
24
Parameters:
25
- model: ModelProto to check compatibility for
26
- device: Target device ("CPU", "CUDA", etc.)
27
- **kwargs: Additional backend-specific options
28
29
Returns:
30
bool: True if backend can execute the model, False otherwise
31
"""
32
33
@classmethod
34
def prepare(cls, model, device="CPU", **kwargs):
35
"""
36
Prepare model for execution on this backend.
37
38
Parameters:
39
- model: ModelProto to prepare
40
- device: Target device for execution
41
- **kwargs: Backend-specific preparation options
42
43
Returns:
44
BackendRep: Prepared model representation ready for execution
45
46
Raises:
47
BackendIsNotSupposedToImplementIt: If backend doesn't support preparation
48
"""
49
50
@classmethod
51
def run_model(cls, model, inputs, device="CPU", **kwargs):
52
"""
53
Run model once with given inputs.
54
55
Parameters:
56
- model: ModelProto to execute
57
- inputs: Input data as list of numpy arrays
58
- device: Target device for execution
59
- **kwargs: Execution options
60
61
Returns:
62
namedtuple: Output values with names corresponding to model outputs
63
64
Raises:
65
BackendIsNotSupposedToImplementIt: If backend doesn't support direct execution
66
"""
67
68
@classmethod
69
def run_node(cls, node, inputs, device="CPU", outputs_info=None, **kwargs):
70
"""
71
Run a single node with given inputs.
72
73
Parameters:
74
- node: NodeProto to execute
75
- inputs: Input data as list of numpy arrays
76
- device: Target device for execution
77
- outputs_info: Optional output type information
78
- **kwargs: Execution options
79
80
Returns:
81
namedtuple: Output values
82
83
Raises:
84
BackendIsNotSupposedToImplementIt: If backend doesn't support node execution
85
"""
86
87
@classmethod
88
def supports_device(cls, device):
89
"""
90
Check if backend supports the specified device.
91
92
Parameters:
93
- device: Device identifier to check
94
95
Returns:
96
bool: True if device is supported, False otherwise
97
"""
98
99
class BackendRep:
100
"""
101
Backend representation of a prepared model.
102
103
Contains the model in a backend-specific format
104
optimized for repeated execution.
105
"""
106
107
def run(self, inputs, **kwargs):
108
"""
109
Execute the prepared model with given inputs.
110
111
Parameters:
112
- inputs: Input data as list of numpy arrays
113
- **kwargs: Execution options
114
115
Returns:
116
namedtuple: Output values with names corresponding to model outputs
117
"""
118
```
119
120
### Device Management
121
122
Classes and constants for device specification and management.
123
124
```python { .api }
125
class Device:
126
"""
127
Device specification for backend execution.
128
129
Encapsulates device type and device-specific configuration.
130
"""
131
132
def __init__(self, device_type, device_id=0):
133
"""
134
Initialize device specification.
135
136
Parameters:
137
- device_type: Type of device (CPU, CUDA, etc.)
138
- device_id: Device identifier for multi-device systems
139
"""
140
141
class DeviceType:
142
"""
143
Constants for standard device types.
144
"""
145
CPU = "CPU"
146
CUDA = "CUDA"
147
# Additional device types may be defined by specific backends
148
```
149
150
### Utility Functions
151
152
Helper functions for backend development and testing.
153
154
```python { .api }
155
def namedtupledict(typename, field_names, *args, **kwargs):
156
"""
157
Create a named tuple class with dictionary-like access.
158
159
Parameters:
160
- typename: Name for the named tuple class
161
- field_names: Field names for the tuple
162
- *args, **kwargs: Additional arguments for namedtuple creation
163
164
Returns:
165
type: Named tuple class with dict-like access methods
166
"""
167
```
168
169
## Usage Examples
170
171
### Implementing a Custom Backend
172
173
```python
174
import onnx
175
from onnx import backend
176
import numpy as np
177
from collections import namedtuple
178
179
class MyCustomBackend(backend.Backend):
180
"""Example implementation of a custom ONNX backend."""
181
182
@classmethod
183
def is_compatible(cls, model, device="CPU", **kwargs):
184
"""Check if we can run this model."""
185
186
# Only support CPU for this example
187
if device != "CPU":
188
return False
189
190
# Check if all operators are supported
191
supported_ops = {"Add", "Sub", "Mul", "Div", "Relu", "MatMul", "Conv"}
192
193
for node in model.graph.node:
194
if node.op_type not in supported_ops:
195
print(f"Unsupported operator: {node.op_type}")
196
return False
197
198
return True
199
200
@classmethod
201
def prepare(cls, model, device="CPU", **kwargs):
202
"""Prepare model for execution."""
203
204
if not cls.is_compatible(model, device, **kwargs):
205
raise RuntimeError("Model is not compatible with this backend")
206
207
# Create a backend representation
208
return MyBackendRep(model, device)
209
210
@classmethod
211
def run_model(cls, model, inputs, device="CPU", **kwargs):
212
"""Run model directly (without preparation)."""
213
214
# Prepare and run
215
rep = cls.prepare(model, device, **kwargs)
216
return rep.run(inputs, **kwargs)
217
218
@classmethod
219
def run_node(cls, node, inputs, device="CPU", **kwargs):
220
"""Run a single node."""
221
222
# Simple node execution logic
223
if node.op_type == "Add":
224
result = inputs[0] + inputs[1]
225
elif node.op_type == "Mul":
226
result = inputs[0] * inputs[1]
227
elif node.op_type == "Relu":
228
result = np.maximum(0, inputs[0])
229
else:
230
raise NotImplementedError(f"Node {node.op_type} not implemented")
231
232
# Return as named tuple
233
OutputTuple = namedtuple('Output', [f'output_{i}' for i in range(len(node.output))])
234
return OutputTuple(result)
235
236
@classmethod
237
def supports_device(cls, device):
238
"""Check device support."""
239
return device == "CPU"
240
241
class MyBackendRep(backend.BackendRep):
242
"""Backend representation for prepared models."""
243
244
def __init__(self, model, device):
245
self.model = model
246
self.device = device
247
self._prepare_model()
248
249
def _prepare_model(self):
250
"""Internal preparation logic."""
251
print(f"Preparing model '{self.model.graph.name}' for {self.device}")
252
# In a real backend, this would compile/optimize the model
253
254
def run(self, inputs, **kwargs):
255
"""Execute the prepared model."""
256
257
print(f"Executing model with {len(inputs)} inputs")
258
259
# Simple execution simulation
260
# In a real backend, this would use optimized execution
261
current_values = {}
262
263
# Set input values
264
for i, input_info in enumerate(self.model.graph.input):
265
current_values[input_info.name] = inputs[i]
266
267
# Set initializer values
268
for initializer in self.model.graph.initializer:
269
from onnx import numpy_helper
270
current_values[initializer.name] = numpy_helper.to_array(initializer)
271
272
# Execute nodes in order
273
for node in self.model.graph.node:
274
node_inputs = [current_values[name] for name in node.input]
275
node_result = MyCustomBackend.run_node(node, node_inputs)
276
277
# Store outputs
278
for i, output_name in enumerate(node.output):
279
if hasattr(node_result, f'output_{i}'):
280
current_values[output_name] = getattr(node_result, f'output_{i}')
281
else:
282
current_values[output_name] = node_result[i] if isinstance(node_result, (list, tuple)) else node_result
283
284
# Collect final outputs
285
outputs = []
286
output_names = []
287
for output_info in self.model.graph.output:
288
outputs.append(current_values[output_info.name])
289
output_names.append(output_info.name)
290
291
# Return as named tuple
292
OutputTuple = namedtuple('ModelOutput', output_names)
293
return OutputTuple(*outputs)
294
295
# Test the custom backend
296
def test_custom_backend():
297
"""Test the custom backend implementation."""
298
299
# Create a simple model for testing
300
from onnx import helper, TensorProto
301
302
X = helper.make_tensor_value_info('X', TensorProto.FLOAT, [2, 2])
303
Y = helper.make_tensor_value_info('Y', TensorProto.FLOAT, [2, 2])
304
305
# Create a simple Add node
306
add_node = helper.make_node('Add', ['X', 'X'], ['Y'])
307
308
graph = helper.make_graph([add_node], 'test_model', [X], [Y])
309
model = helper.make_model(graph)
310
311
# Test backend compatibility
312
backend_impl = MyCustomBackend()
313
314
if backend_impl.is_compatible(model):
315
print("✓ Model is compatible with custom backend")
316
317
# Test direct execution
318
test_input = np.array([[1, 2], [3, 4]], dtype=np.float32)
319
result = backend_impl.run_model(model, [test_input])
320
print(f"Direct execution result: {result.Y}")
321
322
# Test prepared execution
323
rep = backend_impl.prepare(model)
324
result2 = rep.run([test_input])
325
print(f"Prepared execution result: {result2.Y}")
326
327
print(f"Results match: {np.array_equal(result.Y, result2.Y)}")
328
329
else:
330
print("✗ Model is not compatible with custom backend")
331
332
# Run the test
333
test_custom_backend()
334
```
335
336
### Backend Testing Framework
337
338
```python
339
import onnx
340
from onnx import backend
341
import numpy as np
342
343
class BackendTester:
344
"""Framework for testing ONNX backend implementations."""
345
346
def __init__(self, backend_class):
347
self.backend = backend_class
348
349
def test_operator_support(self, test_cases):
350
"""Test backend support for various operators."""
351
352
results = {}
353
354
for op_name, test_model in test_cases.items():
355
try:
356
is_compatible = self.backend.is_compatible(test_model)
357
results[op_name] = {
358
'compatible': is_compatible,
359
'error': None
360
}
361
362
if is_compatible:
363
# Try to prepare the model
364
rep = self.backend.prepare(test_model)
365
results[op_name]['preparable'] = True
366
else:
367
results[op_name]['preparable'] = False
368
369
except Exception as e:
370
results[op_name] = {
371
'compatible': False,
372
'preparable': False,
373
'error': str(e)
374
}
375
376
return results
377
378
def test_device_support(self, devices):
379
"""Test backend device support."""
380
381
device_results = {}
382
for device in devices:
383
device_results[device] = self.backend.supports_device(device)
384
385
return device_results
386
387
def benchmark_execution(self, model, inputs, iterations=100):
388
"""Benchmark model execution performance."""
389
390
import time
391
392
# Test direct execution
393
start_time = time.time()
394
for _ in range(iterations):
395
result = self.backend.run_model(model, inputs)
396
direct_time = time.time() - start_time
397
398
# Test prepared execution
399
rep = self.backend.prepare(model)
400
start_time = time.time()
401
for _ in range(iterations):
402
result = rep.run(inputs)
403
prepared_time = time.time() - start_time
404
405
return {
406
'direct_execution_time': direct_time,
407
'prepared_execution_time': prepared_time,
408
'speedup': direct_time / prepared_time if prepared_time > 0 else float('inf'),
409
'iterations': iterations
410
}
411
412
# Example usage with custom backend
413
def test_backend_comprehensive():
414
"""Comprehensive backend testing example."""
415
416
# Create test models for different operators
417
from onnx import helper, TensorProto
418
419
test_cases = {}
420
421
# Add test
422
X = helper.make_tensor_value_info('X', TensorProto.FLOAT, [2, 2])
423
Y = helper.make_tensor_value_info('Y', TensorProto.FLOAT, [2, 2])
424
add_node = helper.make_node('Add', ['X', 'X'], ['Y'])
425
add_graph = helper.make_graph([add_node], 'add_test', [X], [Y])
426
test_cases['Add'] = helper.make_model(add_graph)
427
428
# ReLU test
429
relu_node = helper.make_node('Relu', ['X'], ['Y'])
430
relu_graph = helper.make_graph([relu_node], 'relu_test', [X], [Y])
431
test_cases['Relu'] = helper.make_model(relu_graph)
432
433
# Unsupported operator test
434
unsupported_node = helper.make_node('LSTM', ['X'], ['Y'])
435
unsupported_graph = helper.make_graph([unsupported_node], 'unsupported_test', [X], [Y])
436
test_cases['LSTM'] = helper.make_model(unsupported_graph)
437
438
# Run tests
439
tester = BackendTester(MyCustomBackend)
440
441
print("=== Operator Support Test ===")
442
op_results = tester.test_operator_support(test_cases)
443
for op, result in op_results.items():
444
status = "✓" if result['compatible'] else "✗"
445
print(f"{status} {op}: Compatible={result['compatible']}, Preparable={result.get('preparable', 'N/A')}")
446
if result.get('error'):
447
print(f" Error: {result['error']}")
448
449
print("\n=== Device Support Test ===")
450
device_results = tester.test_device_support(['CPU', 'CUDA', 'OpenCL'])
451
for device, supported in device_results.items():
452
status = "✓" if supported else "✗"
453
print(f"{status} {device}: {supported}")
454
455
print("\n=== Performance Benchmark ===")
456
test_input = np.array([[1, 2], [3, 4]], dtype=np.float32)
457
benchmark_results = tester.benchmark_execution(test_cases['Add'], [test_input], iterations=10)
458
459
print(f"Direct execution time: {benchmark_results['direct_execution_time']:.4f}s")
460
print(f"Prepared execution time: {benchmark_results['prepared_execution_time']:.4f}s")
461
print(f"Speedup: {benchmark_results['speedup']:.2f}x")
462
463
# Run comprehensive tests
464
test_backend_comprehensive()
465
```
466
467
### Integration with Testing Frameworks
468
469
```python
470
import onnx
471
from onnx import backend
472
473
def create_backend_test_suite(backend_class):
474
"""Create a test suite for backend validation."""
475
476
class BackendTestSuite:
477
def __init__(self):
478
self.backend = backend_class
479
480
def test_basic_compatibility(self):
481
"""Test basic backend functionality."""
482
from onnx import helper, TensorProto
483
484
# Create minimal model
485
X = helper.make_tensor_value_info('input', TensorProto.FLOAT, [1])
486
Y = helper.make_tensor_value_info('output', TensorProto.FLOAT, [1])
487
identity_node = helper.make_node('Identity', ['input'], ['output'])
488
graph = helper.make_graph([identity_node], 'identity', [X], [Y])
489
model = helper.make_model(graph)
490
491
# Test compatibility check
492
assert hasattr(self.backend, 'is_compatible'), "Backend must implement is_compatible"
493
494
# Test preparation
495
if self.backend.is_compatible(model):
496
rep = self.backend.prepare(model)
497
assert hasattr(rep, 'run'), "BackendRep must implement run method"
498
499
def test_device_support(self):
500
"""Test device support functionality."""
501
assert hasattr(self.backend, 'supports_device'), "Backend must implement supports_device"
502
503
# At minimum, should support CPU
504
assert self.backend.supports_device('CPU'), "Backend should support CPU"
505
506
def test_error_handling(self):
507
"""Test error handling for unsupported models."""
508
from onnx import helper, TensorProto
509
510
# Create model with unsupported operator
511
X = helper.make_tensor_value_info('input', TensorProto.FLOAT, [1])
512
Y = helper.make_tensor_value_info('output', TensorProto.FLOAT, [1])
513
514
# Use a hypothetical unsupported operator
515
unsupported_node = helper.make_node('UnsupportedOp', ['input'], ['output'])
516
graph = helper.make_graph([unsupported_node], 'unsupported', [X], [Y])
517
model = helper.make_model(graph)
518
519
# Should either return False for is_compatible or raise appropriate exception
520
try:
521
compatible = self.backend.is_compatible(model)
522
if compatible:
523
# If claiming compatibility, prepare should work
524
rep = self.backend.prepare(model)
525
except Exception as e:
526
# Acceptable if raises informative exception
527
assert len(str(e)) > 0, "Exception should have descriptive message"
528
529
return BackendTestSuite()
530
531
# Example usage
532
def validate_backend_implementation():
533
"""Validate that our backend implementation meets requirements."""
534
535
test_suite = create_backend_test_suite(MyCustomBackend)
536
537
try:
538
test_suite.test_basic_compatibility()
539
print("✓ Basic compatibility test passed")
540
except Exception as e:
541
print(f"✗ Basic compatibility test failed: {e}")
542
543
try:
544
test_suite.test_device_support()
545
print("✓ Device support test passed")
546
except Exception as e:
547
print(f"✗ Device support test failed: {e}")
548
549
try:
550
test_suite.test_error_handling()
551
print("✓ Error handling test passed")
552
except Exception as e:
553
print(f"✗ Error handling test failed: {e}")
554
555
# Validate our custom backend
556
validate_backend_implementation()
557
```