# ------------------------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # Licensed under the MIT License. # -------------------------------------------------------------------------- # This file is modified from https://github.com/microsoft/onnxconverter-common/blob/master/onnxconverter_common/float16.py # Modifications: # (1) Update default value of min_positive_val and max_finite_val # (2) keep_io_types can be list of names # (3) convert initializers if needed to preserve precision # (4) add force_fp16_initializers option # (5) handle Resize and GroupNorm with mixed float inputs # (6) allow convert_float_to_float16 to accept model path import itertools import logging import os import tempfile from typing import Dict import numpy as np import onnx from onnx import AttributeProto, GraphProto, ModelProto, NodeProto, TensorProto, helper, numpy_helper from onnx.shape_inference import infer_shapes, infer_shapes_path from packaging import version logger = logging.getLogger(__name__) def _npfloat16_to_int(np_list): """ Convert numpy float16 to python int. :param np_list: numpy float16 list :return int_list: python int list """ return [int(bin(_.view("H"))[2:].zfill(16), 2) for _ in np_list] def convert_np_to_float16(np_array, min_positive_val=5.96e-08, max_finite_val=65504.0): """ Convert float32 numpy array to float16 without changing sign or finiteness. Positive values less than min_positive_val are mapped to min_positive_val. Positive finite values greater than max_finite_val are mapped to max_finite_val. Similar for negative values. NaN, 0, inf, and -inf are unchanged. """ def between(a, b, c): return np.logical_and(a < b, b < c) if np_array[np.where(np_array > 0)].shape[0] > 0: positive_max = np_array[np.where(np_array > 0)].max() positive_min = np_array[np.where(np_array > 0)].min() if positive_max >= max_finite_val: logger.debug(f"the float32 number {positive_max} will be truncated to {max_finite_val}") if positive_min <= min_positive_val: logger.debug(f"the float32 number {positive_min} will be truncated to {min_positive_val}") if np_array[np.where(np_array < 0)].shape[0] > 0: negative_max = np_array[np.where(np_array < 0)].max() negative_min = np_array[np.where(np_array < 0)].min() if negative_min <= -max_finite_val: logger.debug(f"the float32 number {negative_min} will be truncated to {-max_finite_val}") if negative_max >= -min_positive_val: logger.debug(f"the float32 number {negative_max} will be truncated to {-min_positive_val}") np_array = np.where(between(0, np_array, min_positive_val), min_positive_val, np_array) np_array = np.where(between(-min_positive_val, np_array, 0), -min_positive_val, np_array) np_array = np.where(between(max_finite_val, np_array, float("inf")), max_finite_val, np_array) np_array = np.where(between(float("-inf"), np_array, -max_finite_val), -max_finite_val, np_array) return np.float16(np_array) def convert_tensor_float_to_float16(tensor, min_positive_val=5.96e-08, max_finite_val=65504.0): """Convert tensor float to float16. Args: tensor (TensorProto): the tensor to convert. min_positive_val (float, optional): minimal positive value. Defaults to 1e-7. max_finite_val (float, optional): maximal finite value. Defaults to 1e4. Raises: ValueError: input type is not TensorProto. Returns: TensorProto: the converted tensor. """ if not isinstance(tensor, TensorProto): raise ValueError(f"Expected input type is an ONNX TensorProto but got {type(tensor)}") if tensor.data_type == TensorProto.FLOAT: tensor.data_type = TensorProto.FLOAT16 # convert float_data (float type) to float16 and write to int32_data if tensor.float_data: float16_data = convert_np_to_float16(np.array(tensor.float_data), min_positive_val, max_finite_val) int_list = _npfloat16_to_int(float16_data) tensor.int32_data[:] = int_list tensor.float_data[:] = [] # convert raw_data (bytes type) if tensor.raw_data: # convert n.raw_data to float float32_list = np.frombuffer(tensor.raw_data, dtype="float32") # convert float to float16 float16_list = convert_np_to_float16(float32_list, min_positive_val, max_finite_val) # convert float16 to bytes and write back to raw_data tensor.raw_data = float16_list.tobytes() return tensor def make_value_info_from_tensor(tensor): shape = numpy_helper.to_array(tensor).shape return helper.make_tensor_value_info(tensor.name, tensor.data_type, shape) DEFAULT_OP_BLOCK_LIST = [ "ArrayFeatureExtractor", "Binarizer", "CastMap", "CategoryMapper", "DictVectorizer", "FeatureVectorizer", "Imputer", "LabelEncoder", "LinearClassifier", "LinearRegressor", "Normalizer", "OneHotEncoder", "RandomUniformLike", "SVMClassifier", "SVMRegressor", "Scaler", "TreeEnsembleClassifier", "TreeEnsembleRegressor", "ZipMap", "NonMaxSuppression", "TopK", "RoiAlign", "Range", "CumSum", "Min", "Max", "Upsample", ] # Some operators has data type fixed as float for some inputs. Key is op_type, value is list of input indices # Note that DirectML allows float16 gamma and beta in GroupNorm. Use force_fp16_inputs parameter could overwrite this. ALWAYS_FLOAT_INPUTS = {"Resize": [2], "GroupNorm": [1, 2], "SkipGroupNorm": [1, 2]} class InitializerTracker: """Class for keeping track of initializer.""" def __init__(self, initializer: TensorProto): self.initializer = initializer self.fp32_nodes = [] self.fp16_nodes = [] def add_node(self, node: NodeProto, is_node_blocked): if is_node_blocked: self.fp32_nodes.append(node) else: self.fp16_nodes.append(node) def convert_float_to_float16( model, min_positive_val=5.96e-08, max_finite_val=65504.0, keep_io_types=False, disable_shape_infer=False, op_block_list=None, node_block_list=None, force_fp16_initializers=False, force_fp16_inputs=None, use_bfloat16_as_blocked_nodes_dtype=False, ): """Convert tensor float type in the input ONNX model to tensor float16. Args: model (ModelProto or str): The ONNX model or path of the model to convert. min_positive_val (float, optional): minimal positive value. Defaults to 5.96e-08. max_finite_val (float, optional): maximal finite value of float16. Defaults to 65504. keep_io_types (Union[bool, List[str]], optional): It could be boolean or a list of float32 input/output names. If True, model inputs/outputs should be left as float32. Defaults to False. disable_shape_infer (bool, optional): Skips running onnx shape/type inference. Useful if shape inference has been done. Defaults to False. op_block_list (List[str], optional): List of op types to leave as float32. Defaults to None, which will use `float16.DEFAULT_OP_BLOCK_LIST`. node_block_list (List[str], optional): List of node names to leave as float32. Defaults to None. force_fp16_initializers(bool): force converting all float initializers to float16. Default to false, which will convert only the one needed to avoid precision loss. force_fp16_inputs(Dict[str, List[int]]): Force the conversion of the inputs of some operators to float16, even if this script's preference it to keep them in float32. Raises: ValueError: input type is not ModelProto. Returns: ModelProto: converted model. """ assert ( min_positive_val >= 5.96e-08 ), "invalid min_positive_val. smallest positive float16 value: subnormal 5.96e-08, and normalized 6.104e-05" assert max_finite_val <= float(np.finfo(np.float16).max), "invalid max_finite_val. largest float16 value: 65504" force_fp16_inputs_dict = {} if force_fp16_inputs is None else force_fp16_inputs if isinstance(model, str): model_path = model if version.parse(onnx.__version__) >= version.parse("1.8.0") and not disable_shape_infer: # shape_infer_model_path should be in the same folder of model_path with tempfile.NamedTemporaryFile(dir=os.path.dirname(model_path)) as tmpfile: shape_infer_model_path = tmpfile.name # infer_shapes_path can be used for model >2GB, and infer_shapes cannot. infer_shapes_path(model_path, shape_infer_model_path) model = onnx.load(shape_infer_model_path) disable_shape_infer = True else: model = onnx.load(model_path) if not isinstance(model, ModelProto): raise ValueError(f"Expected an ONNX ModelProto but got {type(model)}") func_infer_shape = None if not disable_shape_infer and version.parse(onnx.__version__) >= version.parse("1.2.0"): try: func_infer_shape = infer_shapes finally: pass # create blocklists if op_block_list is None: op_block_list = DEFAULT_OP_BLOCK_LIST if node_block_list is None: node_block_list = [] op_block_list = set(op_block_list) node_block_list = set(node_block_list) logger.debug( f"fp16 parameters: min_positive_val={min_positive_val} max_finite_val={max_finite_val} keep_io_types={keep_io_types} disable_shape_infer={disable_shape_infer} op_block_list={op_block_list} node_block_list={node_block_list} force_fp16_initializers={force_fp16_initializers}" ) # create a queue for BFS queue = [] value_info_list = [] node_list = [] # Some operators (Like Resize or GroupNorm) have data type fixed as float for some input. # When it is converted to float16, there are mixed types: some inputs are float32 and some are float16. # This list keeps track of such nodes that are not in block list. mixed_float_type_node_list = [] # type inference on input model if func_infer_shape is not None: model = func_infer_shape(model) queue.append(model) name_mapping = {} graph_io_to_skip = set() io_casts = set() fp32_inputs = [n.name for n in model.graph.input if n.type.tensor_type.elem_type == TensorProto.FLOAT] fp32_outputs = [n.name for n in model.graph.output if n.type.tensor_type.elem_type == TensorProto.FLOAT] if isinstance(keep_io_types, list): fp32_inputs = [n for n in fp32_inputs if n in keep_io_types] fp32_outputs = [n for n in fp32_outputs if n in keep_io_types] elif not keep_io_types: fp32_inputs = [] fp32_outputs = [] for i, n in enumerate(model.graph.input): if n.name in fp32_inputs: output_name = "graph_input_cast_" + str(i) name_mapping[n.name] = output_name graph_io_to_skip.add(n.name) node_name = "graph_input_cast" + str(i) new_value_info = model.graph.value_info.add() new_value_info.CopyFrom(n) new_value_info.name = output_name new_value_info.type.tensor_type.elem_type = TensorProto.FLOAT16 # add Cast node (from tensor(float) to tensor(float16) after graph input new_node = [helper.make_node("Cast", [n.name], [output_name], to=TensorProto.FLOAT16, name=node_name)] model.graph.node.extend(new_node) value_info_list.append(new_value_info) io_casts.add(node_name) for i, n in enumerate(model.graph.output): if n.name in fp32_outputs: input_name = "graph_output_cast_" + str(i) name_mapping[n.name] = input_name graph_io_to_skip.add(n.name) node_name = "graph_output_cast" + str(i) # add Cast node (from tensor(float16) to tensor(float) before graph output new_value_info = model.graph.value_info.add() new_value_info.CopyFrom(n) new_value_info.name = input_name new_value_info.type.tensor_type.elem_type = TensorProto.FLOAT16 new_node = [helper.make_node("Cast", [input_name], [n.name], to=1, name=node_name)] model.graph.node.extend(new_node) value_info_list.append(new_value_info) io_casts.add(node_name) fp32_initializers: Dict[str, InitializerTracker] = {} while queue: next_level = [] for q in queue: # if q is model, push q.graph (GraphProto) if isinstance(q, ModelProto): next_level.append(q.graph) # if q is model.graph, push q.node.attribute (AttributeProto) if isinstance(q, GraphProto): for n in q.initializer: # TensorProto type if n.data_type == TensorProto.FLOAT: assert n.name not in fp32_initializers fp32_initializers[n.name] = InitializerTracker(n) for n in q.node: # if n is in the block list (doesn't support float16), no conversion for the node, # and save the node for further processing if n.name in io_casts: continue for i in range(len(n.input)): if n.input[i] in name_mapping: n.input[i] = name_mapping[n.input[i]] for i in range(len(n.output)): if n.output[i] in name_mapping: n.output[i] = name_mapping[n.output[i]] is_node_blocked = n.op_type in op_block_list or n.name in node_block_list for i, input_name in enumerate(n.input): if input_name in fp32_initializers: # For Resize/GroupNorm, only the first input can be float16 use_fp32_weight = is_node_blocked or ( i in ALWAYS_FLOAT_INPUTS.get(n.op_type, []) and i not in force_fp16_inputs_dict.get(n.op_type, []) ) fp32_initializers[input_name].add_node(n, use_fp32_weight) if is_node_blocked: node_list.append(n) else: if n.op_type == "Cast": for attr in n.attribute: if attr.name == "to" and attr.i == TensorProto.FLOAT: attr.i = TensorProto.FLOAT16 break if n.op_type in [ "EyeLike", "Multinomial", "RandomNormal", "RandomNormalLike", "RandomUniform", "RandomUniformLike", "SequenceEmpty", "Bernoulli", ]: has_dtype = False for attr in n.attribute: if attr.name == "dtype": has_dtype = True if attr.i == TensorProto.FLOAT: attr.i = TensorProto.FLOAT16 # The dtype attribute is optional and default is FLOAT in the following operators # so we need add dtype attribute to specify the data type float16 if (n.op_type in ["RandomNormal", "RandomUniform", "SequenceEmpty"]) and not has_dtype: n.attribute.extend([helper.make_attribute("dtype", TensorProto.FLOAT16)]) # For Resize/GroupNorm, attribute data type cannot be changed if n.op_type not in ALWAYS_FLOAT_INPUTS or n.op_type in force_fp16_inputs_dict: for attr in n.attribute: next_level.append(attr) # noqa: PERF402 else: mixed_float_type_node_list.append(n) # if q is model.graph.node.attribute, push q.g and q.graphs (GraphProto) # and process node.attribute.t and node.attribute.tensors (TensorProto) if isinstance(q, AttributeProto): next_level.append(q.g) for n in q.graphs: next_level.append(n) # noqa: PERF402 q.t.CopyFrom(convert_tensor_float_to_float16(q.t, min_positive_val, max_finite_val)) for n in q.tensors: n = convert_tensor_float_to_float16(n, min_positive_val, max_finite_val) # noqa: PLW2901 # if q is graph, process input, output and value_info (ValueInfoProto) if isinstance(q, GraphProto): # Note that float initializers tracked by fp32_initializers will be processed later. # for all ValueInfoProto with tensor(float) type in input, output and value_info, convert them to # tensor(float16) except map and seq(map). And save them in value_info_list for further processing for n in itertools.chain(q.input, q.output, q.value_info): if n.type.tensor_type.elem_type == TensorProto.FLOAT: if n.name not in graph_io_to_skip: n.type.tensor_type.elem_type = TensorProto.FLOAT16 value_info_list.append(n) if n.type.HasField("sequence_type"): if n.type.sequence_type.elem_type.tensor_type.elem_type == TensorProto.FLOAT: if n.name not in graph_io_to_skip: n.type.sequence_type.elem_type.tensor_type.elem_type = TensorProto.FLOAT16 value_info_list.append(n) queue = next_level for value in fp32_initializers.values(): # By default, to avoid precision loss, do not convert an initializer to fp16 when it is used only by fp32 nodes. if force_fp16_initializers or value.fp16_nodes: value.initializer = convert_tensor_float_to_float16(value.initializer, min_positive_val, max_finite_val) value_info_list.append(make_value_info_from_tensor(value.initializer)) if value.fp32_nodes and not force_fp16_initializers: logger.info( f"initializer is used by both fp32 and fp16 nodes. Consider add these nodes to block list:{value.fp16_nodes}" ) # Some operators have data type fixed as float for some input. Add a float16 to float cast for those inputs. for node in mixed_float_type_node_list: for i, input_name in enumerate(node.input): if i not in ALWAYS_FLOAT_INPUTS[node.op_type] or i in force_fp16_inputs_dict.get(node.op_type, []): continue for value_info in value_info_list: if input_name == value_info.name: # create new value_info for current node's new input name new_value_info = model.graph.value_info.add() new_value_info.CopyFrom(value_info) output_name = node.name + "_input_cast_" + str(i) new_value_info.name = output_name new_value_info.type.tensor_type.elem_type = TensorProto.FLOAT # add Cast node (from tensor(float16) to tensor(float) before current node node_name = node.name + "_input_cast" + str(i) new_node = [helper.make_node("Cast", [input_name], [output_name], to=1, name=node_name)] model.graph.node.extend(new_node) # change current node's input name node.input[i] = output_name break accuracy_type = TensorProto.BFLOAT16 if use_bfloat16_as_blocked_nodes_dtype else TensorProto.FLOAT # process the nodes in block list that doesn't support tensor(float16) for node in node_list: # if input's name is in the value_info_list meaning input is tensor(float16) type, # insert a float16 to float Cast node before the node, # change current node's input name and create new value_info for the new name for i in range(len(node.input)): input_name = node.input[i] for value_info in value_info_list: if input_name == value_info.name: # create new value_info for current node's new input name new_value_info = model.graph.value_info.add() new_value_info.CopyFrom(value_info) output_name = node.name + "_input_cast_" + str(i) new_value_info.name = output_name new_value_info.type.tensor_type.elem_type = accuracy_type # add Cast node (from tensor(float16) to tensor(float) before current node node_name = node.name + "_input_cast" + str(i) new_node = [helper.make_node("Cast", [input_name], [output_name], to=accuracy_type, name=node_name)] model.graph.node.extend(new_node) # change current node's input name node.input[i] = output_name break # if output's name is in the value_info_list meaning output is tensor(float16) type, insert a float to # float16 Cast node after the node, change current node's output name and create new value_info for the new name for i in range(len(node.output)): output = node.output[i] for value_info in value_info_list: if output == value_info.name: # create new value_info for current node's new output new_value_info = model.graph.value_info.add() new_value_info.CopyFrom(value_info) input_name = node.name + "_output_cast_" + str(i) new_value_info.name = input_name new_value_info.type.tensor_type.elem_type = accuracy_type # add Cast node (from tensor(float) to tensor(float16) after current node node_name = node.name + "_output_cast" + str(i) new_node = [helper.make_node("Cast", [input_name], [output], to=10, name=node_name)] model.graph.node.extend(new_node) # change current node's input name node.output[i] = input_name break return model def float_to_float16_max_diff(tensor, min_positive_val=5.96e-08, max_finite_val=65504.0): """Measure the maximum absolute difference after converting a float tensor to float16.""" if not isinstance(tensor, TensorProto): raise ValueError(f"Expected input type is an ONNX TensorProto but got {type(tensor)}") if tensor.data_type != TensorProto.FLOAT: raise ValueError("Expected tensor data type is float.") float32_data = None if tensor.float_data: float32_data = np.array(tensor.float_data) if tensor.raw_data: float32_data = np.frombuffer(tensor.raw_data, dtype="float32") if float32_data is None: raise RuntimeError("external data not loaded!") float16_data = convert_np_to_float16(float32_data, min_positive_val, max_finite_val) return np.amax(np.abs(float32_data - np.float32(float16_data)))