From 333c2a466225085bcfc1ea88136e3b3ed58457d5 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" Date: Wed, 5 Aug 2026 10:59:37 +0800 Subject: [PATCH] fix(fp16): handle initializer-backed graph outputs --- src/winml/modelkit/quant/fp16.py | 275 ++++++++++++++++++-- tests/unit/optim/test_fp16.py | 432 ++++++++++++++++++++++++++++++- 2 files changed, 683 insertions(+), 24 deletions(-) diff --git a/src/winml/modelkit/quant/fp16.py b/src/winml/modelkit/quant/fp16.py index 0294b8076..c324c808b 100644 --- a/src/winml/modelkit/quant/fp16.py +++ b/src/winml/modelkit/quant/fp16.py @@ -4,24 +4,253 @@ # -------------------------------------------------------------------------- """FP16 conversion utility for ONNX models. -Provides a single entry point for FP32→FP16 model conversion, used by +Provides a single entry point for FP32->FP16 model conversion, used by the quantizer's ``mode="fp16"`` path. """ from __future__ import annotations import logging -from typing import TYPE_CHECKING +from dataclasses import dataclass +from math import prod +from typing import TYPE_CHECKING, cast from google.protobuf.message import EncodeError if TYPE_CHECKING: - from onnx import ModelProto + from onnx import GraphProto, ModelProto, TensorProto logger = logging.getLogger(__name__) +@dataclass(frozen=True) +class _InitializerOutput: + """A safe top-level graph output supplied directly by an initializer.""" + + name: str + output_index: int + + +def _tensor_data_is_loaded(initializer: TensorProto) -> bool: + """Whether a FLOAT tensor carries resident data rather than only a sidecar ref.""" + if initializer.raw_data or initializer.float_data: + return True + return prod(initializer.dims) == 0 + + +def _all_tensor_names(model: ModelProto) -> set[str]: + """Collect tensor names across scopes relevant to ORT's global name maps.""" + names: set[str] = set() + for graph in [model.graph, *_iter_nested_graphs(model)]: + names.update(value.name for value in graph.input) + names.update(value.name for value in graph.output) + names.update(value.name for value in graph.value_info) + names.update(initializer.name for initializer in graph.initializer) + names.update(sparse.values.name for sparse in graph.sparse_initializer) + names.update(name for node in graph.node for name in (*node.input, *node.output) if name) + return names + + +def _all_node_names(model: ModelProto) -> set[str]: + """Collect node names across scopes because ORT reserves generated names globally.""" + return { + node.name + for graph in [model.graph, *_iter_nested_graphs(model)] + for node in graph.node + if node.name + } + + +def _iter_nested_graphs(model: ModelProto) -> list[GraphProto]: + """Return nested attribute graphs without relying on tensor-name scope.""" + from onnx import AttributeProto + + nested: list[GraphProto] = [] + pending = [model.graph] + while pending: + graph = pending.pop() + for node in graph.node: + for attribute in node.attribute: + if attribute.type == AttributeProto.GRAPH: + nested.append(attribute.g) + pending.append(attribute.g) + elif attribute.type == AttributeProto.GRAPHS: + nested.extend(attribute.graphs) + pending.extend(attribute.graphs) + return nested + + +def _reject_nested_initializer_outputs(model: ModelProto) -> None: + """Reject nested initializer outputs whose lexical semantics are ambiguous.""" + from onnx import TensorProto + + for graph in _iter_nested_graphs(model): + produced = {name for node in graph.node for name in node.output if name} + output_names = {output.name for output in graph.output} + matches = [ + initializer.name + for initializer in graph.initializer + if initializer.name in output_names + and initializer.name not in produced + and initializer.data_type == TensorProto.FLOAT + ] + if matches: + msg = ( + "FP16 conversion cannot safely normalize initializer-backed outputs " + f"inside nested graphs: {', '.join(matches)}." + ) + raise RuntimeError(msg) + + +def _all_node_inputs(model: ModelProto) -> set[str]: + """Collect node input names across top-level and nested graphs.""" + inputs = {name for node in model.graph.node for name in node.input if name} + for graph in _iter_nested_graphs(model): + inputs.update(name for node in graph.node for name in node.input if name) + return inputs + + +def _capture_safe_initializer_outputs( + model: ModelProto, + *, + keep_io_types: bool, +) -> list[_InitializerOutput]: + """Capture safe direct initializer outputs or fail before ORT mutates the model. + + This fix intentionally covers only top-level, non-overridable, no-consumer + dense FLOAT outputs. Shared, nested, or overridable initializer outputs need + lexical-scope and mixed-precision semantics that ORT's converter does not + model safely; fail explicitly instead of publishing an invalid graph. + """ + from onnx import TensorProto + from onnx.external_data_helper import uses_external_data + + if not hasattr(model.graph, "input") or not hasattr(model.graph, "output"): + return [] + + _reject_nested_initializer_outputs(model) + produced = {name for node in model.graph.node for name in node.output if name} + consumed = _all_node_inputs(model) + graph_inputs = {value.name for value in model.graph.input} + initializers = {initializer.name: initializer for initializer in model.graph.initializer} + tensor_names = _all_tensor_names(model) + node_names = _all_node_names(model) + + captured: list[_InitializerOutput] = [] + for output_index, output in enumerate(model.graph.output): + initializer = initializers.get(output.name) + if ( + initializer is None + or output.name in produced + or initializer.data_type != TensorProto.FLOAT + ): + continue + + if output.name in graph_inputs: + msg = ( + f"Initializer-backed output '{output.name}' is also a graph input; " + "FP16 conversion cannot preserve overridable-initializer semantics." + ) + raise RuntimeError(msg) + if output.name in consumed: + msg = ( + f"Initializer-backed output '{output.name}' has internal consumers; " + "FP16 conversion cannot safely infer their mixed-precision semantics." + ) + raise RuntimeError(msg) + if uses_external_data(initializer) and not _tensor_data_is_loaded(initializer): + msg = ( + f"Initializer-backed output '{output.name}' uses unloaded external data; " + "load external weights before FP16 conversion." + ) + raise RuntimeError(msg) + + if keep_io_types: + generated_tensor = f"graph_output_cast_{output_index}" + generated_node = f"graph_output_cast{output_index}" + collisions = [] + if generated_tensor in tensor_names: + collisions.append(generated_tensor) + if generated_node in node_names: + collisions.append(generated_node) + if collisions: + msg = ( + f"FP16 conversion cannot safely allocate ORT graph-output Cast names for " + f"'{output.name}'; existing names collide: {', '.join(collisions)}." + ) + raise RuntimeError(msg) + + captured.append(_InitializerOutput(output.name, output_index)) + return captured + + +def _convert_output_initializer_to_fp16(model: ModelProto, name: str) -> None: + """Convert a captured output's resident FLOAT initializer in place.""" + from onnx import TensorProto + from onnx.external_data_helper import uses_external_data + from onnxruntime.transformers.float16 import convert_tensor_float_to_float16 + + initializer = next(value for value in model.graph.initializer if value.name == name) + if uses_external_data(initializer): + del initializer.external_data[:] + initializer.data_location = TensorProto.DEFAULT + initializer.CopyFrom(cast("TensorProto", convert_tensor_float_to_float16(initializer))) + + +def _internalize_output_initializer(model: ModelProto, name: str) -> None: + """Drop stale external metadata after resident bytes were loaded.""" + from onnx import TensorProto + from onnx.external_data_helper import uses_external_data + + initializer = next(value for value in model.graph.initializer if value.name == name) + if uses_external_data(initializer): + del initializer.external_data[:] + initializer.data_location = TensorProto.DEFAULT + + +def _remove_orphan_output_casts( + model: ModelProto, + captured: list[_InitializerOutput], +) -> None: + """Remove ORT output Casts whose inputs cannot have producers by construction.""" + from onnx import TensorProto + + if not captured: + return + + remove_indices: list[int] = [] + orphan_inputs: set[str] = set() + for item in captured: + generated_tensor = f"graph_output_cast_{item.output_index}" + generated_node = f"graph_output_cast{item.output_index}" + matches = [ + (index, node) + for index, node in enumerate(model.graph.node) + if node.name == generated_node + and node.op_type == "Cast" + and list(node.input) == [generated_tensor] + and list(node.output) == [item.name] + and any( + attribute.name == "to" and attribute.i == TensorProto.FLOAT + for attribute in node.attribute + ) + ] + if len(matches) != 1: + msg = f"Expected one ORT graph-output Cast for initializer-backed output '{item.name}'." + raise RuntimeError(msg) + remove_indices.append(matches[0][0]) + orphan_inputs.add(generated_tensor) + + for index in sorted(remove_indices, reverse=True): + del model.graph.node[index] + for item in captured: + _internalize_output_initializer(model, item.name) + retained = [value for value in model.graph.value_info if value.name not in orphan_inputs] + del model.graph.value_info[:] + model.graph.value_info.extend(retained) + + def convert_to_fp16( model: ModelProto, *, @@ -31,33 +260,30 @@ def convert_to_fp16( """Convert an ONNX model from FP32 to FP16 precision. Uses onnxruntime.transformers.float16.convert_float_to_float16 internally. - No new dependencies — ORT is already a project dependency. + The successful conversion mutates and returns ``model`` as before. - Note: ORT's converter mutates the model in-place and returns the same object. - - Args: - model: Input ONNX ModelProto (will be mutated in-place by ORT). - keep_io_types: If True, preserve FP32 model inputs/outputs by inserting - Cast nodes at boundaries. Recommended for CPU-safe inference. - op_block_list: Op types to keep in FP32 (e.g., ["LayerNorm", "Softmax"]). - When None, ORT uses its DEFAULT_OP_BLOCK_LIST which includes ops - known to be numerically unsafe in FP16 (e.g., TopK, CumSum, etc.). - - Returns: - The converted model (same object as input due to ORT in-place mutation). + ORT assumes each graph output has a node producer. For a safe top-level + output supplied only by a dense FLOAT initializer, keep-I/O conversion adds + a Cast with no producer; remove that exact Cast. Pure-FP16 conversion changes + the output declaration but not its initializer, so convert that initializer + explicitly. """ from onnx import TensorProto from onnxruntime.transformers.float16 import convert_float_to_float16 - # Skip if model is already FP16 (check floating-point initializer dtypes) + _reject_nested_initializer_outputs(model) + fp32_types = {TensorProto.FLOAT, TensorProto.DOUBLE, TensorProto.BFLOAT16} initializers = model.graph.initializer if initializers: - float_inits = [t for t in initializers if t.data_type in fp32_types | {TensorProto.FLOAT16}] - if float_inits and all(t.data_type == TensorProto.FLOAT16 for t in float_inits): - logger.info("Model is already FP16 — skipping conversion.") + floating = [ + value for value in initializers if value.data_type in fp32_types | {TensorProto.FLOAT16} + ] + if floating and all(value.data_type == TensorProto.FLOAT16 for value in floating): + logger.info("Model is already FP16 - skipping conversion.") return model + captured = _capture_safe_initializer_outputs(model, keep_io_types=keep_io_types) original_nodes = len(model.graph.node) logger.info("Converting model to FP16...") @@ -85,9 +311,12 @@ def convert_to_fp16( op_block_list=op_block_list, ) - # ORT's converter appends Cast nodes at the end of the node list (for - # keep_io_types), which breaks topological ordering. Re-sort the graph - # using ORT's own topological sort utility. + if keep_io_types: + _remove_orphan_output_casts(converted, captured) + else: + for item in captured: + _convert_output_initializer_to_fp16(converted, item.name) + if keep_io_types: from onnxruntime.transformers.onnx_model import OnnxModel diff --git a/tests/unit/optim/test_fp16.py b/tests/unit/optim/test_fp16.py index 2e63ab193..031232b2b 100644 --- a/tests/unit/optim/test_fp16.py +++ b/tests/unit/optim/test_fp16.py @@ -16,7 +16,8 @@ from __future__ import annotations import numpy as np -from onnx import ModelProto, TensorProto, helper, numpy_helper +import onnxruntime as ort +from onnx import GraphProto, ModelProto, TensorProto, checker, helper, numpy_helper, shape_inference from winml.modelkit.quant.fp16 import convert_to_fp16 @@ -47,6 +48,165 @@ def _build_multi_op_fp32_model() -> ModelProto: return helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) +def _build_initializer_backed_output_model() -> ModelProto: + """Build a graph whose output is supplied directly by an initializer.""" + out = helper.make_tensor_value_info("constant_output", TensorProto.FLOAT, [1, 2]) + value = numpy_helper.from_array( + np.array([[1.0001, 2.0003]], dtype=np.float32), "constant_output" + ) + graph = helper.make_graph([], "initializer_output", [], [out], [value]) + return helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + + +def _build_shared_initializer_output_model() -> ModelProto: + """Build a graph where an initializer is both an output and a node input.""" + x = helper.make_tensor_value_info("x", TensorProto.FLOAT, [1, 2]) + shared = helper.make_tensor_value_info("shared", TensorProto.FLOAT, [1, 2]) + y = helper.make_tensor_value_info("y", TensorProto.FLOAT, [1, 2]) + value = numpy_helper.from_array(np.array([[1.0, 2.0]], dtype=np.float32), "shared") + add = helper.make_node("Add", ["x", "shared"], ["y"], name="add") + graph = helper.make_graph([add], "shared_initializer", [x], [shared, y], [value]) + return helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + + +def _build_nested_initializer_output_model() -> ModelProto: + """Build an If whose branch outputs are supplied by initializers.""" + condition = helper.make_tensor_value_info("condition", TensorProto.BOOL, []) + output = helper.make_tensor_value_info("output", TensorProto.FLOAT, [1]) + + def _branch(name: str, output_name: str, value: float): + branch_output = helper.make_tensor_value_info(output_name, TensorProto.FLOAT, [1]) + initializer = numpy_helper.from_array(np.array([value], dtype=np.float32), output_name) + return helper.make_graph([], name, [], [branch_output], [initializer]) + + node = helper.make_node( + "If", + ["condition"], + ["output"], + name="if", + then_branch=_branch("then", "then_output", 1.0), + else_branch=_branch("else", "else_output", 2.0), + ) + graph = helper.make_graph([node], "nested_initializer", [condition], [output]) + return helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + + +def _build_lexically_captured_initializer_output_model() -> ModelProto: + """Build an output initializer captured only by nested If branches.""" + condition = helper.make_tensor_value_info("condition", TensorProto.BOOL, []) + shared = helper.make_tensor_value_info("shared", TensorProto.FLOAT, [1]) + output = helper.make_tensor_value_info("output", TensorProto.FLOAT, [1]) + initializer = numpy_helper.from_array(np.array([1.0], dtype=np.float32), "shared") + + def _branch(name: str) -> GraphProto: + branch_output = helper.make_tensor_value_info(f"{name}_output", TensorProto.FLOAT, [1]) + identity = helper.make_node( + "Identity", ["shared"], [f"{name}_output"], name=f"{name}_identity" + ) + return helper.make_graph([identity], name, [], [branch_output]) + + node = helper.make_node( + "If", + ["condition"], + ["output"], + name="if", + then_branch=_branch("then"), + else_branch=_branch("else"), + ) + graph = helper.make_graph( + [node], "lexical_capture", [condition], [shared, output], [initializer] + ) + return helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + + +def _build_initializer_output_name_collision_model() -> ModelProto: + """Build a legal graph with a name that collides with ORT's generated alias.""" + existing = helper.make_tensor_value_info("graph_output_cast_0", TensorProto.FLOAT, [1]) + constant_output = helper.make_tensor_value_info("constant_output", TensorProto.FLOAT, [1]) + result = helper.make_tensor_value_info("result", TensorProto.FLOAT, [1]) + initializer = numpy_helper.from_array(np.array([1.0], dtype=np.float32), "constant_output") + identity = helper.make_node("Identity", ["graph_output_cast_0"], ["result"], name="identity") + graph = helper.make_graph( + [identity], "name_collision", [existing], [constant_output, result], [initializer] + ) + return helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + + +def _build_initializer_output_node_name_collision_model() -> ModelProto: + """Build a graph with a user node named like ORT's output Cast node.""" + model = _build_initializer_backed_output_model() + x = helper.make_tensor_value_info("x", TensorProto.FLOAT, [1]) + y = helper.make_tensor_value_info("y", TensorProto.FLOAT, [1]) + model.graph.input.append(x) + model.graph.output.append(y) + model.graph.node.append(helper.make_node("Identity", ["x"], ["y"], name="graph_output_cast0")) + return model + + +def _build_nested_node_name_collision_model() -> ModelProto: + """Build a nested user node named like ORT's top-level output Cast.""" + model = _build_initializer_backed_output_model() + condition = helper.make_tensor_value_info("condition", TensorProto.BOOL, []) + nested_output = helper.make_tensor_value_info("nested_output", TensorProto.FLOAT, [1]) + + def _branch(name: str, value: float) -> GraphProto: + branch_output = helper.make_tensor_value_info(f"{name}_output", TensorProto.FLOAT, [1]) + initializer = numpy_helper.from_array(np.array([value], dtype=np.float32), f"{name}_value") + identity = helper.make_node( + "Identity", + [f"{name}_value"], + [f"{name}_output"], + name="graph_output_cast0" if name == "then" else f"{name}_identity", + ) + return helper.make_graph([identity], name, [], [branch_output], [initializer]) + + node = helper.make_node( + "If", + ["condition"], + ["nested_output"], + name="if", + then_branch=_branch("then", 1.0), + else_branch=_branch("else", 2.0), + ) + model.graph.input.append(condition) + model.graph.output.append(nested_output) + model.graph.node.append(node) + return model + + +def _build_nested_shadowed_initializer_output_model() -> ModelProto: + """Build legal nested outputs that shadow an outer initializer name.""" + condition = helper.make_tensor_value_info("condition", TensorProto.BOOL, []) + output = helper.make_tensor_value_info("output", TensorProto.FLOAT, [1]) + outer = numpy_helper.from_array(np.array([9.0], dtype=np.float32), "same") + + def _branch(name: str, value: float) -> GraphProto: + branch_output = helper.make_tensor_value_info("same", TensorProto.FLOAT, [1]) + initializer = numpy_helper.from_array(np.array([value], dtype=np.float32), "same") + return helper.make_graph([], name, [], [branch_output], [initializer]) + + node = helper.make_node( + "If", + ["condition"], + ["output"], + name="if", + then_branch=_branch("then", 1.0), + else_branch=_branch("else", 2.0), + ) + graph = helper.make_graph([node], "shadow", [condition], [output], [outer]) + return helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + + +def _build_blocked_initializer_consumer_model() -> ModelProto: + """Build an output initializer consumed only by an FP32-blocked node.""" + shared = helper.make_tensor_value_info("shared", TensorProto.FLOAT, [1]) + copied = helper.make_tensor_value_info("copied", TensorProto.FLOAT, [1]) + initializer = numpy_helper.from_array(np.array([1.0001], dtype=np.float32), "shared") + identity = helper.make_node("Identity", ["shared"], ["copied"], name="identity") + graph = helper.make_graph([identity], "blocked_consumer", [], [shared, copied], [initializer]) + return helper.make_model(graph, opset_imports=[helper.make_opsetid("", 17)]) + + # ============================================================================= # CONVERT_TO_FP16 TESTS # ============================================================================= @@ -83,6 +243,276 @@ def test_keep_io_types_false_converts_io(self) -> None: for outp in result.graph.output: assert outp.type.tensor_type.elem_type == TensorProto.FLOAT16 + def test_initializer_backed_output_stays_fp32_when_io_types_are_kept(self) -> None: + """An initializer graph output remains valid FP32 when preserving I/O.""" + model = _build_initializer_backed_output_model() + + result = convert_to_fp16(model, keep_io_types=True) + + output = result.graph.output[0] + assert output.type.tensor_type.elem_type == TensorProto.FLOAT + assert any( + initializer.data_type == TensorProto.FLOAT for initializer in result.graph.initializer + ) + checker.check_model(result) + shape_inference.infer_shapes(result, strict_mode=True) + session = ort.InferenceSession( + result.SerializeToString(), providers=["CPUExecutionProvider"] + ) + np.testing.assert_array_equal( + session.run(None, {})[0], + np.array([[1.0001, 2.0003]], dtype=np.float32), + ) + + def test_initializer_backed_output_converts_data_when_io_types_are_not_kept(self) -> None: + """A pure-FP16 output converts its backing initializer as well as its type.""" + model = _build_initializer_backed_output_model() + + result = convert_to_fp16(model, keep_io_types=False, op_block_list=[]) + + output = result.graph.output[0] + initializer = result.graph.initializer[0] + assert output.type.tensor_type.elem_type == TensorProto.FLOAT16 + assert initializer.data_type == TensorProto.FLOAT16 + shape_inference.infer_shapes(result, strict_mode=True) + + def test_multiple_initializer_backed_outputs_are_all_converted(self) -> None: + """Every initializer-backed output is repaired independently.""" + model = _build_initializer_backed_output_model() + second_output = helper.make_tensor_value_info("second_output", TensorProto.FLOAT, [1, 2]) + second_value = numpy_helper.from_array( + np.array([[3.0, 4.0]], dtype=np.float32), "second_output" + ) + model.graph.output.append(second_output) + model.graph.initializer.append(second_value) + + result = convert_to_fp16(model, keep_io_types=False, op_block_list=[]) + + assert all( + output.type.tensor_type.elem_type == TensorProto.FLOAT16 + for output in result.graph.output + ) + assert all( + initializer.data_type == TensorProto.FLOAT16 for initializer in result.graph.initializer + ) + shape_inference.infer_shapes(result, strict_mode=True) + + def test_multiple_initializer_outputs_keep_exact_fp32_values(self) -> None: + """Removing several orphan Casts preserves every FP32 model output.""" + model = _build_initializer_backed_output_model() + second_output = helper.make_tensor_value_info("second_output", TensorProto.FLOAT, [1, 2]) + second_value = numpy_helper.from_array( + np.array([[3.0005, 4.0007]], dtype=np.float32), "second_output" + ) + model.graph.output.append(second_output) + model.graph.initializer.append(second_value) + + result = convert_to_fp16(model, keep_io_types=True) + + checker.check_model(result) + session = ort.InferenceSession( + result.SerializeToString(), providers=["CPUExecutionProvider"] + ) + first, second = session.run(None, {}) + np.testing.assert_array_equal(first, np.array([[1.0001, 2.0003]], dtype=np.float32)) + np.testing.assert_array_equal(second, np.array([[3.0005, 4.0007]], dtype=np.float32)) + + def test_overridable_initializer_output_is_rejected_before_mutation(self) -> None: + """A graph-input initializer keeps its caller override semantics.""" + model = _build_initializer_backed_output_model() + model.graph.input.append( + helper.make_tensor_value_info("constant_output", TensorProto.FLOAT, [1, 2]) + ) + original = model.SerializeToString() + + with np.testing.assert_raises_regex(RuntimeError, "also a graph input"): + convert_to_fp16(model, keep_io_types=True) + + assert model.SerializeToString() == original + + def test_collision_uses_original_mixed_output_index(self) -> None: + """ORT Cast collision checks count preceding non-FLOAT outputs.""" + model = _build_initializer_backed_output_model() + int_output = helper.make_tensor_value_info("int_output", TensorProto.INT64, [1]) + int_value = numpy_helper.from_array(np.array([1], dtype=np.int64), "int_output") + model.graph.output.insert(0, int_output) + model.graph.initializer.append(int_value) + model.graph.input.append( + helper.make_tensor_value_info("graph_output_cast_1", TensorProto.FLOAT, [1]) + ) + original = model.SerializeToString() + + with np.testing.assert_raises_regex(RuntimeError, "graph_output_cast_1"): + convert_to_fp16(model, keep_io_types=True) + + assert model.SerializeToString() == original + + def test_unloaded_external_initializer_output_is_rejected(self) -> None: + """Conversion refuses external backing data that was not loaded.""" + model = _build_initializer_backed_output_model() + initializer = model.graph.initializer[0] + initializer.ClearField("raw_data") + initializer.data_location = TensorProto.EXTERNAL + location = initializer.external_data.add() + location.key = "location" + location.value = "weights.data" + + original_output_type = model.graph.output[0].type.tensor_type.elem_type + original_initializer_type = initializer.data_type + + with np.testing.assert_raises_regex( + RuntimeError, + "load external weights before FP16 conversion", + ): + convert_to_fp16(model, keep_io_types=False, op_block_list=[]) + + assert model.graph.output[0].type.tensor_type.elem_type == original_output_type + assert model.graph.initializer[0].data_type == original_initializer_type + + def test_loaded_external_initializer_output_is_converted(self) -> None: + """Resident tensor bytes are valid even if external metadata remains.""" + model = _build_initializer_backed_output_model() + initializer = model.graph.initializer[0] + initializer.data_location = TensorProto.EXTERNAL + location = initializer.external_data.add() + location.key = "location" + location.value = "weights.data" + + result = convert_to_fp16(model, keep_io_types=False, op_block_list=[]) + + assert result.graph.output[0].type.tensor_type.elem_type == TensorProto.FLOAT16 + assert result.graph.initializer[0].data_type == TensorProto.FLOAT16 + + def test_loaded_external_initializer_output_is_internalized_when_io_is_kept(self) -> None: + """Resident output data no longer points to a stale sidecar after repair.""" + model = _build_initializer_backed_output_model() + initializer = model.graph.initializer[0] + initializer.data_location = TensorProto.EXTERNAL + location = initializer.external_data.add() + location.key = "location" + location.value = "weights.data" + + result = convert_to_fp16(model, keep_io_types=True) + + repaired = result.graph.initializer[0] + assert repaired.data_location == TensorProto.DEFAULT + assert not repaired.external_data + session = ort.InferenceSession( + result.SerializeToString(), providers=["CPUExecutionProvider"] + ) + np.testing.assert_array_equal( + session.run(None, {})[0], + np.array([[1.0001, 2.0003]], dtype=np.float32), + ) + + def test_shared_initializer_output_is_rejected_before_mutation(self) -> None: + """Shared initializer-output semantics are rejected before mutation.""" + model = _build_shared_initializer_output_model() + original = model.SerializeToString() + + with np.testing.assert_raises_regex(RuntimeError, "has internal consumers"): + convert_to_fp16(model, keep_io_types=True, op_block_list=[]) + assert model.SerializeToString() == original + + def test_nested_initializer_outputs_are_rejected_before_mutation(self) -> None: + """Nested initializer-output semantics are rejected before mutation.""" + model = _build_nested_initializer_output_model() + original = model.SerializeToString() + + with np.testing.assert_raises_regex(RuntimeError, "inside nested graphs"): + convert_to_fp16(model, keep_io_types=True, op_block_list=[]) + assert model.SerializeToString() == original + + def test_lexical_nested_consumers_are_rejected_before_mutation(self) -> None: + """Lexically shared output initializers are rejected before mutation.""" + model = _build_lexically_captured_initializer_output_model() + original = model.SerializeToString() + + with np.testing.assert_raises_regex(RuntimeError, "has internal consumers"): + convert_to_fp16(model, keep_io_types=True, op_block_list=[]) + assert model.SerializeToString() == original + + def test_generated_tensor_name_collision_is_rejected_before_mutation(self) -> None: + """Repair allocates a fresh alias instead of duplicating an existing name.""" + model = _build_initializer_output_name_collision_model() + + original = model.SerializeToString() + with np.testing.assert_raises_regex(RuntimeError, "existing names collide"): + convert_to_fp16(model, keep_io_types=True, op_block_list=[]) + assert model.SerializeToString() == original + + def test_generated_cast_node_name_collision_is_rejected_before_mutation(self) -> None: + """A user node occupying ORT's deterministic Cast name fails safely.""" + model = _build_initializer_output_node_name_collision_model() + original = model.SerializeToString() + + with np.testing.assert_raises_regex(RuntimeError, "existing names collide"): + convert_to_fp16(model, keep_io_types=True, op_block_list=[]) + + assert model.SerializeToString() == original + + def test_nested_generated_node_name_collision_is_rejected_before_mutation(self) -> None: + """Nested nodes also participate in ORT's global generated-name set.""" + model = _build_nested_node_name_collision_model() + original = model.SerializeToString() + + with np.testing.assert_raises_regex(RuntimeError, "existing names collide"): + convert_to_fp16(model, keep_io_types=True, op_block_list=[]) + + assert model.SerializeToString() == original + + def test_nested_initializer_output_can_shadow_an_outer_initializer(self) -> None: + """Nested shadowing is rejected rather than handed to ORT's global map.""" + model = _build_nested_shadowed_initializer_output_model() + original = model.SerializeToString() + + with np.testing.assert_raises_regex(RuntimeError, "inside nested graphs"): + convert_to_fp16(model, keep_io_types=False, op_block_list=[]) + assert model.SerializeToString() == original + + def test_nested_fp32_initializer_prevents_already_fp16_shortcut(self) -> None: + """Nested FP32 outputs are rejected before the already-FP16 shortcut.""" + model = _build_nested_initializer_output_model() + model.graph.initializer.append( + numpy_helper.from_array(np.array([1.0], dtype=np.float16), "top_level_fp16") + ) + + original = model.SerializeToString() + with np.testing.assert_raises_regex(RuntimeError, "inside nested graphs"): + convert_to_fp16(model, keep_io_types=False, op_block_list=[]) + assert model.SerializeToString() == original + + def test_blocked_fp32_consumer_does_not_round_trip_through_fp16(self) -> None: + """Blocked consumers are rejected instead of silently losing precision.""" + model = _build_blocked_initializer_consumer_model() + original = model.SerializeToString() + + with np.testing.assert_raises_regex(RuntimeError, "has internal consumers"): + convert_to_fp16(model, keep_io_types=True, op_block_list=["Identity"]) + assert model.SerializeToString() == original + + def test_initializer_output_repair_preserves_unrelated_casts(self) -> None: + """Repair removes only ORT's orphan output Cast, not user graph Casts.""" + model = _build_initializer_backed_output_model() + x = helper.make_tensor_value_info("x", TensorProto.FLOAT, [1]) + cast_output = helper.make_tensor_value_info("cast_output", TensorProto.INT32, [1]) + model.graph.input.append(x) + model.graph.output.append(cast_output) + model.graph.node.append( + helper.make_node( + "Cast", + ["x"], + ["cast_output"], + name="user_cast", + to=TensorProto.INT32, + ) + ) + + result = convert_to_fp16(model, keep_io_types=True) + + checker.check_model(result) + assert any(node.name == "user_cast" for node in result.graph.node) + def test_preserves_model_structure(self) -> None: """FP16 conversion preserves graph structure (node count diff ≤ 2).""" model = _build_multi_op_fp32_model()