Skill v1.0.1
currentAutomated scan100/1001 files
version: "1.0.1" name: pass-writing description: Comprehensive guidance for creating transformation passes in the ONNX IR project. It encapsulates best practices, conventions, and patterns derived from existing passes in onnx_ir/passes/common/.
Overview
The ONNX IR pass infrastructure is designed for graph construction, analysis, and transformation. Passes are composable units that transform ONNX models in a well-defined way.
Key Concepts
- Pass: A transformation that takes an
ir.Modeland returns aPassResultcontaining the transformed model and a boolean indicating if modifications were made - InPlacePass: A pass that modifies the input model directly and returns it (most efficient)
- FunctionalPass: A pass that returns a new model without modifying the input
- PassManager/Sequential: Composes multiple passes to run in sequence
Pass Infrastructure
Base Classes
All passes inherit from one of these base classes defined in onnx_ir.passes:
InPlacePass
class MyPass(ir.passes.InPlacePass):"""Most common pass type - modifies model in place."""def call(self, model: ir.Model) -> ir.passes.PassResult:modified = False# Transform the modelreturn ir.passes.PassResult(model, modified=modified)
Use when: You want efficient in-place mutation (recommended for most passes)
Properties:
in_place = True(automatically set)changes_input = True(automatically set)
FunctionalPass
class MyPass(ir.passes.FunctionalPass):"""Pure functional pass - does not modify input."""def call(self, model: ir.Model) -> ir.passes.PassResult:# Must return a different model objectnew_model = model.clone()# Transform new_modelreturn ir.passes.PassResult(new_model, modified=True)
Use when: You need to preserve the original model unchanged
Properties:
in_place = False(automatically set)changes_input = False(automatically set)
Pass Lifecycle
- Preconditions (
requiresmethod): Check input model validity (optional) - Transformation (
callmethod): Apply the transformation - Postconditions (
ensuresmethod): Validate output model (optional)
class MyPass(ir.passes.InPlacePass):def requires(self, model: ir.Model) -> None:"""Validate preconditions. Raise PreconditionError if violated."""# Example: Ensure specific opset versionif model.graph.opset_imports.get("", 0) < 13:raise ir.passes.PreconditionError("Requires opset >= 13")def call(self, model: ir.Model) -> ir.passes.PassResult:"""Main transformation logic."""modified = False# ... transformation code ...return ir.passes.PassResult(model, modified=modified)def ensures(self, model: ir.Model) -> None:"""Validate postconditions. Raise PostconditionError if violated."""# Example: Check model validitypass
Best Practices for Pass Implementation
1. Graph Traversal
Traverse All Nodes (Including Subgraphs)
import onnx_ir as ir# Use RecursiveGraphIterator to process all nodes including subgraphsfor node in ir.traversal.RecursiveGraphIterator(model.graph):# Process nodeif node.op_type == "Identity":# ... handle identity node ...pass
Process Functions
# Don't forget to process functions in the modelfor function in model.functions.values():for node in ir.traversal.RecursiveGraphIterator(function):# Process node in functionpass
Simple Graph Iteration
# For non-recursive iteration of the main graphfor node in model.graph:# Process only direct nodes (no subgraphs)pass# Reverse iteration (useful for removal)for node in reversed(model.graph):# Process in reverse topological orderpass
2. Node Manipulation
Safely Remove Nodes
# Always use safe=True to ensure proper cleanupgraph.remove(node, safe=True)
Create New Nodes
# Create a node with the ir.node() helpernew_node = ir.node("Identity",inputs=[input_value],outputs=[ir.Value(name="output_name",type=ir.TensorType(ir.DataType.FLOAT),shape=ir.Shape([1, 3, 224, 224]),)],)# Insert the node at a specific positiongraph.insert_before(reference_node, new_node)graph.insert_after(reference_node, new_node)# Or append to the endgraph.append(new_node)
Modify Node Attributes
# Access attributes as a dictionaryif "training_mode" in node.attributes:node.attributes.pop("training_mode")# Add new attributesnode.attributes["new_attr"] = ir.Attr("new_attr", ir.AttributeType.STRING, "value")
3. Value Manipulation
Replace All Uses of a Value
import onnx_ir.convenience as convenience# Replace all uses of old_value with new_valueconvenience.replace_all_uses_with(old_value,new_value,replace_graph_outputs=True # Also replace in graph outputs if present)# Replace multiple values at onceconvenience.replace_all_uses_with([old_value1, old_value2],[new_value1, new_value2],)
Check Value Usage
# Check if a value is usedif output_value.uses():# Value has consumerspass# Check if value is a graph outputif value.is_graph_output():# Special handling for outputspass# Check if value is a graph inputif value.is_graph_input():# Special handling for inputspass
Merge Value Information
# When eliminating nodes, preserve shape/type informationdef merge_shapes(shape1: ir.Shape | None, shape2: ir.Shape | None) -> ir.Shape | None:if shape1 is None:return shape2if shape2 is None:return shape1# More sophisticated merging logic...return shape1# Copy shape and type informationinput_value.shape = merge_shapes(input_value.shape, output_value.shape)if input_value.type is None:input_value.type = output_value.type
4. Initializers and Constants
Work with Initializers
# Access initializers by nameinitializers = graph.initializersif "weight" in initializers:weight_initializer = initializers["weight"]# Register a new initializernew_initializer = ir.Value(name="new_weight",shape=ir.Shape([3, 3, 64, 64]),type=ir.TensorType(ir.DataType.FLOAT),const_value=tensor_data,)graph.register_initializer(new_initializer)# Remove unused initializersgraph_outputs = frozenset(graph.outputs)graph_inputs = frozenset(graph.inputs)for init in list(initializers.values()):if not (init.uses() or init in graph_outputs or init in graph_inputs):assert init.name is not Nonedel initializers[init.name]
Lift Constants to Initializers
# Check if node is a Constantif node.op_type == "Constant" and node.domain in ("", "onnx.ai"):# Get the tensor from the value attributeattr_value = node.attributes.get("value")if attr_value:tensor = attr_value.as_tensor()# Create initializerinitializer = ir.Value(name=node.outputs[0].name,shape=tensor.shape,type=ir.TensorType(tensor.dtype),const_value=tensor,)graph.register_initializer(initializer)# Replace uses and remove nodenode.outputs[0].replace_all_uses_with(initializer)graph.remove(node, safe=True)
5. Logging and Debugging
import logginglogger = logging.getLogger(__name__)class MyPass(ir.passes.InPlacePass):def call(self, model: ir.Model) -> ir.passes.PassResult:count = 0for node in model.graph:# Use debug for detailed informationlogger.debug("Processing node: %s", node)# Use info for important changeslogger.info("Removed node: %s", node.name)count += 1# Summarize at the endif count:logger.info("MyPass removed %s nodes", count)return ir.passes.PassResult(model, modified=bool(count))
6. Handling Subgraphs
# Process attributes that may contain subgraphsfor attr in node.attributes.values():if not isinstance(attr, ir.Attr):continueif attr.type == ir.AttributeType.GRAPH:subgraph = attr.as_graph()# Process the subgraph recursivelymodified |= self._process_graph(subgraph)elif attr.type == ir.AttributeType.GRAPHS:for subgraph in attr.as_graphs():# Process each subgraphmodified |= self._process_graph(subgraph)
7. Opset Version Handling
# Get opset version for the graphonnx_opset_version = model.graph.opset_imports.get("", None)# Check if a specific opset is availableif onnx_opset_version is not None and onnx_opset_version >= 13:# Use features from opset 13+pass# Get schema information for a nodeimport onnxtry:op_schema = onnx.defs.get_schema(node.op_type,onnx_opset_version,domain=node.domain)# Use schema informationexcept Exception:logger.info("Failed to get schema for %s", node)
Common Pass Patterns
Pattern 1: Node Elimination
Eliminate nodes that match certain criteria (e.g., Identity, unused nodes).
class NodeEliminationPass(ir.passes.InPlacePass):def call(self, model: ir.Model) -> ir.passes.PassResult:modified = Falsefor node in ir.traversal.RecursiveGraphIterator(model.graph):if self._should_eliminate(node):if self._try_eliminate_node(node):modified = Truereturn ir.passes.PassResult(model, modified=modified)def _should_eliminate(self, node: ir.Node) -> bool:"""Check if node should be eliminated."""return node.op_type == "Identity" and node.domain == ""def _try_eliminate_node(self, node: ir.Node) -> bool:"""Try to eliminate node. Returns True if successful."""# Validate node structureif len(node.inputs) != 1 or len(node.outputs) != 1:return Falseinput_value = node.inputs[0]output_value = node.outputs[0]if input_value is None:return False# Replace usesir.convenience.replace_all_uses_with(output_value, input_value, replace_graph_outputs=True)# Remove nodeassert node.graph is not Nonenode.graph.remove(node, safe=True)return True
Pattern 2: Dead Code Elimination
Remove unused nodes and values.
class DeadCodeEliminationPass(ir.passes.InPlacePass):def call(self, model: ir.Model) -> ir.passes.PassResult:count = self._remove_unused_nodes(model.graph)for function in model.functions.values():count += self._remove_unused_nodes(function)return ir.passes.PassResult(model, modified=bool(count))def _remove_unused_nodes(self, graph_like: ir.Graph | ir.Function) -> int:"""Remove nodes that produce no used outputs."""graph_outputs = frozenset(graph_like.outputs)count = 0# Iterate in reverse to handle dependenciesfor node in reversed(graph_like):removable = Truefor output in node.outputs:if output in graph_outputs or output.uses():removable = Falsebreakif removable:graph_like.remove(node, safe=True)count += 1return count
Pattern 3: Common Subexpression Elimination
Eliminate duplicate computations.
class CSEPass(ir.passes.InPlacePass):def call(self, model: ir.Model) -> ir.passes.PassResult:modified = self._eliminate_cse(model.graph)return ir.passes.PassResult(model, modified=modified)def _eliminate_cse(self, graph: ir.Graph) -> bool:modified = False# Map from (op_identifier, inputs, attributes) to nodeexisting_nodes: dict[tuple, ir.Node] = {}for node in graph:# Skip non-deterministic opsif self._is_non_deterministic(node):continue# Create a hashable key for the nodenode_key = (node.op_identifier(),tuple(id(inp) for inp in node.inputs),tuple(sorted(node.attributes.items())),)if node_key in existing_nodes:# Found duplicate - replace with existingexisting_node = existing_nodes[node_key]ir.convenience.replace_all_uses_with(node.outputs,existing_node.outputs)graph.remove(node, safe=True)modified = Trueelse:existing_nodes[node_key] = nodereturn modifieddef _is_non_deterministic(self, node: ir.Node) -> bool:"""Check if node is non-deterministic."""non_deterministic_ops = frozenset({"RandomUniform", "RandomNormal","RandomUniformLike", "RandomNormalLike","Multinomial"})return node.op_type in non_deterministic_ops and node.domain == ""
Pattern 4: Graph Normalization
Ensure graph is in a canonical form (e.g., topological sort, name fixing).
class TopologicalSortPass(ir.passes.InPlacePass):"""Sort nodes in topological order."""def call(self, model: ir.Model) -> ir.passes.PassResult:original_nodes = list(model.graph)model.graph.sort() # Built-in methodsorted_nodes = list(model.graph)# Check if order changedmodified = Falsefor node, new_node in zip(original_nodes, sorted_nodes):if node is not new_node:modified = Truebreak# Also sort functionsfor function in model.functions.values():function.sort()return ir.passes.PassResult(model, modified=modified)
Pattern 5: Attribute/Metadata Manipulation
Modify node attributes or clear metadata.
class ClearMetadataPass(ir.passes.InPlacePass):"""Clear metadata and doc strings from the model."""def call(self, model: ir.Model) -> ir.passes.PassResult:modified = False# Clear model metadataif model.doc_string or model.metadata_props:model.doc_string = ""model.metadata_props.clear()modified = True# Clear graph metadatamodified |= self._clear_graph_metadata(model.graph)# Clear function metadatafor function in model.functions.values():modified |= self._clear_graph_metadata(function)return ir.passes.PassResult(model, modified=modified)def _clear_graph_metadata(self, graph_like: ir.Graph | ir.Function) -> bool:modified = Falseif graph_like.doc_string:graph_like.doc_string = ""modified = Truefor node in ir.traversal.RecursiveGraphIterator(graph_like):if node.doc_string or node.metadata_props:node.doc_string = ""node.metadata_props.clear()modified = Truereturn modified
Testing Your Pass
Basic Test Structure
import onnx_ir as irdef test_my_pass():# Create a test modelmodel = create_test_model()# Apply the passpass_instance = MyPass()result = pass_instance(model)# Verify the resultassert result.modified == Trueassert len(result.model.graph) == expected_node_count# Verify specific transformations# ...
Common Pitfalls to Avoid
- Modifying while iterating: ONNX IR's iterators are robust and support modification during iteration
```python # Forward iteration with removal is safe in onnx_ir for node in graph: if should_remove(node): graph.remove(node, safe=True)
# Reversed iteration is useful for dependency order for node in reversed(graph): if should_remove(node): graph.remove(node, safe=True) ```
Note: Unlike standard Python iterators, onnx_ir's graph iterators are specifically designed to handle modifications during iteration. Choose forward or reverse iteration based on your algorithm's needs, not safety concerns.
- Forgetting subgraphs: Always use
RecursiveGraphIteratoror manually process subgraphs
- Not checking for None inputs: Nodes can have optional None inputs
``python for input_value in node.inputs: if input_value is not None: # Always check # Process input pass ``
- Modifying graph outputs incorrectly: Be careful when replacing graph output values
``python # Update graph outputs properly if output_value.is_graph_output(): # Find and update in graph.outputs for idx, graph_output in enumerate(graph.outputs): if graph_output is output_value: graph.outputs[idx] = new_value ``
- Not handling edge cases: Check for empty inputs/outputs, graph boundaries
- Forgetting to process functions: Many passes should also process
model.functions
Performance Considerations
- Use in-place passes when possible (most efficient)
- Minimize graph traversals: Combine multiple checks in one traversal
- Use frozenset for lookups: When checking membership in graph inputs/outputs
``python graph_outputs = frozenset(graph.outputs) if value in graph_outputs: # O(1) lookup pass ``
- Batch operations: Remove multiple nodes in one traversal rather than multiple passes
- Early exit: Return early if no modifications are needed
Pass Composition
Sequential Pass Execution
# Combine multiple passespasses = ir.passes.Sequential(RemoveUnusedNodesPass(),IdentityEliminationPass(),TopologicalSortPass(),)result = passes(model)
Pass Manager with Iteration
# Run passes multiple times until convergencepasses = ir.passes.PassManager([CommonSubexpressionEliminationPass(),RemoveUnusedNodesPass(),],steps=5, # Maximum iterationsearly_stop=True, # Stop if no changes)result = passes(model)
Security and Error Handling
- Validate inputs: Check node structure, input/output counts
- Handle exceptions gracefully: Catch and log errors in schema lookups
- Use safe removal: Always call
graph.remove(node, safe=True) - Avoid hardcoding assumptions: Check opset versions, schema availability
- Log warnings: When skipping nodes due to unexpected conditions
try:schema = onnx.defs.get_schema(node.op_type, opset_version, domain=node.domain)except Exception:logger.warning("Could not get schema for %s, skipping", node, exc_info=True)continue
Import Conventions
# Standard imports for pass filesfrom __future__ import annotationsimport loggingimport onnx_ir as irlogger = logging.getLogger(__name__)
Summary
When creating a new pass:
- Choose the right base class:
InPlacePass(most common) orFunctionalPass - Implement `call` method: Main transformation logic
- Return `PassResult`: With model and modified flag
- Use `RecursiveGraphIterator`: To process all nodes including subgraphs
- Process functions: Don't forget
model.functions - Use convenience functions:
replace_all_uses_with, etc. - Log appropriately: Use debug/info logging levels
- Handle edge cases: None inputs, graph boundaries, subgraphs
- Test thoroughly: Write tests for various scenarios
- Document clearly: Explain what the pass does and any parameters
References
- Pass Infrastructure:
src/onnx_ir/passes/_pass_infra.py - Example Passes:
src/onnx_ir/passes/common/ - Convenience Functions:
src/onnx_ir/convenience.py - Traversal Utilities:
src/onnx_ir/traversal.py