Open Neural Network Exchange for AI model interoperability and machine learning frameworks
—
Complete reference implementation of ONNX operators for testing, validation, and educational purposes. This module provides a Python-based execution engine that implements all ONNX operators according to the specification.
Complete reference implementation for executing ONNX models with all standard operators.
class ReferenceEvaluator:
"""
Reference implementation for ONNX model execution.
Provides accurate, specification-compliant execution of ONNX models
primarily for testing and validation purposes.
"""
def __init__(self, model_proto, verbose=0, **kwargs):
"""
Initialize reference evaluator with ONNX model.
Parameters:
- model_proto: ModelProto to execute
- verbose: Verbosity level for debugging (0=silent, 1=basic, 2=detailed)
- **kwargs: Additional evaluator options
Raises:
RuntimeError: If model contains unsupported operators or is invalid
"""
def run(self, output_names, feed_inputs, attributes=None):
"""
Execute model and return specified outputs.
Parameters:
- output_names: List of output names to compute (None for all outputs)
- feed_inputs: Dictionary mapping input names to numpy arrays
- attributes: Optional dictionary of additional attributes
Returns:
list: List of output arrays corresponding to output_names
Raises:
RuntimeError: If execution fails due to invalid inputs or operators
ValueError: If input shapes or types are incompatible
"""import onnx
from onnx.reference import ReferenceEvaluator
import numpy as np
# Load an ONNX model
model = onnx.load_model("example_model.onnx")
# Create reference evaluator
evaluator = ReferenceEvaluator(model, verbose=1)
# Prepare input data
input_data = {
"input": np.random.randn(1, 3, 224, 224).astype(np.float32)
}
try:
# Execute model
outputs = evaluator.run(None, input_data) # None means all outputs
print(f"Model executed successfully!")
print(f"Number of outputs: {len(outputs)}")
for i, output in enumerate(outputs):
print(f"Output {i} shape: {output.shape}, dtype: {output.dtype}")
except Exception as e:
print(f"Execution failed: {e}")import onnx
from onnx.reference import ReferenceEvaluator
import numpy as np
def compare_backends(model_path, input_data):
"""Compare reference implementation with other backends."""
model = onnx.load_model(model_path)
# Reference implementation
ref_evaluator = ReferenceEvaluator(model)
ref_outputs = ref_evaluator.run(None, input_data)
print("Reference implementation results:")
for i, output in enumerate(ref_outputs):
print(f" Output {i}: mean={output.mean():.6f}, std={output.std():.6f}")
# You could compare with other backends here
# For example, if you have ONNX Runtime installed:
"""
import onnxruntime as ort
ort_session = ort.InferenceSession(model_path)
ort_outputs = ort_session.run(None, input_data)
print("ONNX Runtime results:")
for i, output in enumerate(ort_outputs):
print(f" Output {i}: mean={output.mean():.6f}, std={output.std():.6f}")
# Compare results
for i, (ref_out, ort_out) in enumerate(zip(ref_outputs, ort_outputs)):
max_diff = np.max(np.abs(ref_out - ort_out))
print(f"Output {i} max difference: {max_diff}")
"""
# Example usage
input_data = {"input": np.random.randn(1, 10).astype(np.float32)}
# compare_backends("simple_model.onnx", input_data)import onnx
from onnx.reference import ReferenceEvaluator
import numpy as np
def debug_model_execution(model_path, input_data):
"""Debug model execution with detailed logging."""
model = onnx.load_model(model_path)
# Create evaluator with maximum verbosity
evaluator = ReferenceEvaluator(model, verbose=2)
try:
print("Starting model execution with detailed logging...")
outputs = evaluator.run(None, input_data)
print("Execution completed successfully!")
return outputs
except Exception as e:
print(f"Execution failed at: {e}")
print("This can help identify:")
print("- Which operator caused the failure")
print("- Input/output shape mismatches")
print("- Type conversion issues")
print("- Unsupported operator attributes")
return None
# Example debugging session
def create_debug_model():
"""Create a simple model for debugging demonstration."""
from onnx import helper, TensorProto
# Create a model with potential issues
X = helper.make_tensor_value_info('X', TensorProto.FLOAT, [2, 3])
Y = helper.make_tensor_value_info('Y', TensorProto.FLOAT, [2, 3])
# Create nodes that might have issues
relu_node = helper.make_node('Relu', ['X'], ['relu_out'])
# Intentionally problematic reshape (wrong dimensions)
reshape_node = helper.make_node('Reshape', ['relu_out'], ['Y'],
shape=[6]) # This will cause shape mismatch
graph = helper.make_graph([relu_node, reshape_node], 'debug_model',
[X], [Y])
return helper.make_model(graph)
# Test debugging
debug_model = create_debug_model()
test_input = {"X": np.array([[1, -2, 3], [4, -5, 6]], dtype=np.float32)}
print("=== Debugging Model Execution ===")
try:
evaluator = ReferenceEvaluator(debug_model, verbose=1)
result = evaluator.run(None, test_input)
print("Unexpected success!")
except Exception as e:
print(f"Expected error caught: {e}")
print("This demonstrates how the reference evaluator helps identify issues")import onnx
from onnx.reference import ReferenceEvaluator
from onnx import helper, TensorProto
import numpy as np
def test_operator_reference(op_type, inputs, attributes=None, **kwargs):
"""Test reference implementation of a specific operator."""
# Create minimal model with just the operator
input_infos = []
input_names = []
for i, input_array in enumerate(inputs):
input_name = f"input_{i}"
input_names.append(input_name)
input_info = helper.make_tensor_value_info(
input_name, TensorProto.FLOAT, list(input_array.shape)
)
input_infos.append(input_info)
# Create output info (shape will be inferred)
output_info = helper.make_tensor_value_info('output', TensorProto.FLOAT, [])
# Create node
node_attrs = attributes or {}
node = helper.make_node(op_type, input_names, ['output'], **node_attrs)
# Create graph and model
graph = helper.make_graph([node], f'{op_type}_test',
input_infos, [output_info])
model = helper.make_model(graph)
# Test with reference evaluator
evaluator = ReferenceEvaluator(model, **kwargs)
# Prepare input dictionary
feed_dict = {f"input_{i}": inp for i, inp in enumerate(inputs)}
try:
outputs = evaluator.run(['output'], feed_dict)
return outputs[0]
except Exception as e:
print(f"Operator {op_type} test failed: {e}")
return None
# Test various operators
def run_operator_tests():
"""Run tests for various operators."""
print("=== Testing Reference Implementation Operators ===")
# Test Add operator
a = np.array([[1, 2], [3, 4]], dtype=np.float32)
b = np.array([[5, 6], [7, 8]], dtype=np.float32)
add_result = test_operator_reference('Add', [a, b])
if add_result is not None:
print(f"Add result:\n{add_result}")
print(f"Expected:\n{a + b}")
print(f"Correct: {np.allclose(add_result, a + b)}")
print()
# Test Relu operator
x = np.array([[-1, 2], [-3, 4]], dtype=np.float32)
relu_result = test_operator_reference('Relu', [x])
if relu_result is not None:
print(f"Relu result:\n{relu_result}")
print(f"Expected:\n{np.maximum(0, x)}")
print(f"Correct: {np.allclose(relu_result, np.maximum(0, x))}")
print()
# Test Conv operator (more complex)
input_tensor = np.random.randn(1, 1, 5, 5).astype(np.float32)
weight_tensor = np.random.randn(1, 1, 3, 3).astype(np.float32)
conv_result = test_operator_reference('Conv', [input_tensor, weight_tensor],
attributes={'kernel_shape': [3, 3],
'pads': [1, 1, 1, 1]})
if conv_result is not None:
print(f"Conv output shape: {conv_result.shape}")
print("Conv operation completed successfully")
print()
# Run the operator tests
run_operator_tests()Install with Tessl CLI
npx tessl i tessl/pypi-onnx