add deepcopy_data

optimum/intel/openvino/quantization.py

@@ -74,6 +74,7 @@
     PREDEFINED_SD_DATASETS,
     PREDEFINED_SPEECH_TO_TEXT_DATASETS,
     PREDEFINED_VISUAL_LM_DATASETS,
+    deepcopy_data,
 )
 
 
@@ -131,7 +132,7 @@ def __init__(
 
     def collect_inputs(self, inputs):
         if not self.apply_caching or not isinstance(inputs, dict):
-            self.collected_inputs.append(copy.deepcopy(inputs))
+            self.collected_inputs.append(deepcopy_data(inputs))
             return
 
         copied_inputs = {}
@@ -146,7 +147,7 @@ def collect_inputs(self, inputs):
             # Avoid data copying if tensor contains data encountered earlier
             self.tensor_cache.setdefault(k, {})
             if data_hash not in self.tensor_cache[k]:
-                self.tensor_cache[k][data_hash] = copy.deepcopy(v)
+                self.tensor_cache[k][data_hash] = deepcopy_data(v)
             copied_inputs[k] = self.tensor_cache[k][data_hash]
         self.collected_inputs.append(copied_inputs)
 

optimum/intel/openvino/utils.py

@@ -19,21 +19,22 @@
 import stat
 import warnings
 import weakref
+from copy import deepcopy
 from glob import glob
 from pathlib import Path
 from tempfile import TemporaryDirectory as OrigTemporaryDirectory
 from tempfile import mkdtemp
-from typing import Tuple, Type, Union
+from typing import Tuple, Type, Union, Any
 
 import numpy as np
 import torch
 from huggingface_hub import model_info
-from openvino.runtime import Core, Model, properties
+from openvino.runtime import Core, Model, properties, Tensor
 from openvino.runtime import Type as OVType
 from packaging.version import Version
 from transformers import AutoTokenizer, CLIPTokenizer, PreTrainedTokenizer, PreTrainedTokenizerFast
 from transformers.onnx.utils import ParameterFormat, compute_serialized_parameters_size
-
+import openvino
 from optimum.intel.utils.import_utils import is_torch_version
 
 
@@ -586,3 +587,20 @@ def check_scale_available(model: Union[Model, str, Path]):
     if runtime_options is None:
         return False
     return runtime_options.find("ACTIVATIONS_SCALE_FACTOR") is not None
+
+
+def deepcopy_data(inputs: Any) -> Any:
+    if isinstance(inputs, dict):
+        new_inputs = {}
+        for k, v in inputs.items():
+            new_inputs[deepcopy_data(k)] = deepcopy_data(v)
+    elif isinstance(inputs, list):
+        new_inputs = [deepcopy_data(elem) for elem in inputs]
+    elif isinstance(inputs, tuple):
+        new_inputs = tuple(deepcopy_data(elem) for elem in inputs)
+    elif isinstance(inputs, openvino.Tensor):
+        new_inputs = openvino.Tensor(np.zeros(inputs.shape, dtype=inputs.element_type.to_dtype()))
+        new_inputs.copy_from(inputs)
+    else:
+        new_inputs = deepcopy(inputs)
+    return new_inputs

tests/openvino/test_quantization.py

@@ -42,6 +42,9 @@
 from transformers.testing_utils import slow
 from transformers.utils.quantization_config import QuantizationMethod
 
+from optimum.intel.openvino.utils import deepcopy_data
+
+
 from optimum.intel import (
     OVConfig,
     OVFluxPipeline,
@@ -1354,6 +1357,34 @@ def test_calibration_data_uniqueness(self, model_name, apply_caching):
             # Without caching, encoder hidden states tensors will be unique for each collected input
             self.assertGreater(len(data_id_per_key["encoder_hidden_states"]), 2)
 
+    def test_deepcopy_data(self):
+        data = {
+            "a": torch.tensor([1, 2, 3]),
+            "b": np.array([1, 2, 3]),
+            "c": 1,
+            "d": "string",
+            "e": {"a": torch.tensor([1, 2, 3]), "b": np.array([1, 2, 3])},
+            "f": [ov.Tensor(np.ones((1, 2, 3))), ov.Tensor(np.ones((1, 2, 3)))],
+        }
+        copied_data = deepcopy_data(data)
+        assert copied_data["a"] is not data["a"]
+        assert copied_data["b"] is not data["b"]
+        assert copied_data["e"]["a"] is not data["e"]["a"]
+        assert copied_data["e"]["b"] is not data["e"]["b"]
+        assert copied_data["f"][0] is not data["f"][0]
+        assert copied_data["f"][1] is not data["f"][1]
+
+        assert torch.equal(copied_data["a"], data["a"])
+        assert np.array_equal(copied_data["b"], data["b"])
+        assert copied_data["c"] == data["c"]
+        assert copied_data["d"] == data["d"]
+        assert torch.equal(copied_data["e"]["a"], data["e"]["a"])
+        assert np.array_equal(copied_data["e"]["b"], data["e"]["b"])
+        assert np.array_equal(copied_data["f"][0].data, data["f"][0].data)
+        assert np.array_equal(copied_data["f"][1].data, data["f"][1].data)
+
+        assert copied_data is not data
+
 
 def check_optimization_not_applicable_to_optimized_model(model, quantization_config):
     quantizer = OVQuantizer(model)