Evaluate quantum kernel matrix elements for identical samples (#432)

* bug fix

* add reno

* format reno

* format reno

* format reno

* format reno

* format reno

* fix tests

* fix style

* fix mypy

* code review

* Update qiskit_machine_learning/kernels/quantum_kernel.py

Co-authored-by: ElePT <57907331+ElePT@users.noreply.github.com>

* Update releasenotes/notes/fix-quantum-kernel-duplicates-b75d6ed8a0f37f60.yaml

Co-authored-by: ElePT <57907331+ElePT@users.noreply.github.com>

* Update qiskit_machine_learning/kernels/quantum_kernel.py

Co-authored-by: ElePT <57907331+ElePT@users.noreply.github.com>

* Update qiskit_machine_learning/kernels/quantum_kernel.py

Co-authored-by: ElePT <57907331+ElePT@users.noreply.github.com>

Co-authored-by: ElePT <57907331+ElePT@users.noreply.github.com>

qiskit_machine_learning/kernels/quantum_kernel.py

@@ -11,8 +11,9 @@
 # that they have been altered from the originals.
 
 """Quantum Kernel Algorithm"""
+from __future__ import annotations
 
-from typing import Optional, Union, Sequence, Mapping, List
+from typing import Sequence, Mapping, List
 import copy
 import numbers
 
@@ -55,11 +56,12 @@ class QuantumKernel:
     @deprecate_arguments("0.5.0", {"user_parameters": "training_parameters"})
     def __init__(
         self,
-        feature_map: Optional[QuantumCircuit] = None,
+        feature_map: QuantumCircuit | None = None,
         enforce_psd: bool = True,
         batch_size: int = 900,
-        quantum_instance: Optional[Union[QuantumInstance, Backend]] = None,
-        training_parameters: Optional[Union[ParameterVector, Sequence[Parameter]]] = None,
+        quantum_instance: QuantumInstance | Backend | None = None,
+        training_parameters: ParameterVector | Sequence[Parameter] | None = None,
+        evaluate_duplicates: str = "off_diagonal",
     ) -> None:
         """
         Args:
@@ -72,6 +74,21 @@ def __init__(
             training_parameters: Iterable containing ``Parameter`` objects which correspond to
                  quantum gates on the feature map circuit which may be tuned. If users intend to
                  tune feature map parameters to find optimal values, this field should be set.
+            evaluate_duplicates: Defines a strategy how kernel matrix elements are evaluated if
+               duplicate samples are found. Possible values are:
+
+                    - ``all`` means that all kernel matrix elements are evaluated, even the diagonal
+                        ones when training. This may introduce additional noise in the matrix.
+                    - ``off_diagonal`` when training the matrix diagonal is set to `1`, the rest
+                        elements are fully evaluated, e.g., for two identical samples in the
+                        dataset. When inferring, all elements are evaluated. This is the default
+                        value.
+                    - ``none`` when training the diagonal is set to `1` and if two identical samples
+                        are found in the dataset the corresponding matrix element is set to `1`.
+                        When inferring, matrix elements for identical samples are set to `1`.
+
+        Raises:
+            ValueError: When unsupported value is passed to `evaluate_duplicates`.
         """
         # Class fields
         self._feature_map = None
@@ -81,6 +98,12 @@ def __init__(
         self._enforce_psd = enforce_psd
         self._batch_size = batch_size
         self._quantum_instance = quantum_instance
+        eval_duplicates = evaluate_duplicates.lower()
+        if eval_duplicates not in ("all", "off_diagonal", "none"):
+            raise ValueError(
+                f"Unsupported value passed as evaluate_duplicates: {evaluate_duplicates}"
+            )
+        self._evaluate_duplicates = eval_duplicates
 
         # Setters
         self.feature_map = feature_map if feature_map is not None else ZZFeatureMap(2)
@@ -116,30 +139,28 @@ def quantum_instance(self) -> QuantumInstance:
         return self._quantum_instance
 
     @quantum_instance.setter
-    def quantum_instance(self, quantum_instance: Union[Backend, QuantumInstance]) -> None:
+    def quantum_instance(self, quantum_instance: Backend | QuantumInstance) -> None:
         """Set quantum instance"""
         if isinstance(quantum_instance, Backend):
             self._quantum_instance = QuantumInstance(quantum_instance)
         else:
             self._quantum_instance = quantum_instance
 
     @property
-    def training_parameters(self) -> Optional[Union[ParameterVector, Sequence[Parameter]]]:
+    def training_parameters(self) -> ParameterVector | Sequence[Parameter] | None:
         """Return the vector of training parameters."""
         return copy.copy(self._training_parameters)
 
     @training_parameters.setter
-    def training_parameters(
-        self, training_params: Union[ParameterVector, Sequence[Parameter]]
-    ) -> None:
+    def training_parameters(self, training_params: ParameterVector | Sequence[Parameter]) -> None:
         """Set the training parameters"""
         self._training_parameter_binds = {
             training_param: training_param for training_param in training_params
         }
         self._training_parameters = copy.deepcopy(training_params)
 
     def assign_training_parameters(
-        self, values: Union[Mapping[Parameter, ParameterValueType], Sequence[ParameterValueType]]
+        self, values: Mapping[Parameter, ParameterValueType] | Sequence[ParameterValueType]
     ) -> None:
         """
         Assign training parameters in the ``QuantumKernel`` feature map.
@@ -253,12 +274,12 @@ def assign_training_parameters(
         )
 
     @property
-    def training_parameter_binds(self) -> Optional[Mapping[Parameter, float]]:
+    def training_parameter_binds(self) -> Mapping[Parameter, float] | None:
         """Return a copy of the current training parameter mappings for the feature map circuit."""
         return copy.deepcopy(self._training_parameter_binds)
 
     def bind_training_parameters(
-        self, values: Union[Mapping[Parameter, ParameterValueType], Sequence[ParameterValueType]]
+        self, values: Mapping[Parameter, ParameterValueType] | Sequence[ParameterValueType]
     ) -> None:
         """
         Alternate function signature for ``assign_training_parameters``
@@ -280,19 +301,19 @@ def get_unbound_training_parameters(self) -> List[Parameter]:
 
     @property  # type: ignore
     @deprecate_property("0.5.0", new_name="training_parameters")
-    def user_parameters(self) -> Optional[Union[ParameterVector, Sequence[Parameter]]]:
+    def user_parameters(self) -> ParameterVector | Sequence[Parameter] | None:
         """[Deprecated property]Return the vector of training parameters."""
         return self.training_parameters
 
     @user_parameters.setter  # type: ignore
     @deprecate_property("0.5.0", new_name="training_parameters")
-    def user_parameters(self, training_params: Union[ParameterVector, Sequence[Parameter]]) -> None:
+    def user_parameters(self, training_params: ParameterVector | Sequence[Parameter]) -> None:
         """[Deprecated property setter]Set the training parameters"""
         self.training_parameters = training_params
 
     @deprecate_method("0.5.0", new_name="assign_training_parameters")
     def assign_user_parameters(
-        self, values: Union[Mapping[Parameter, ParameterValueType], Sequence[ParameterValueType]]
+        self, values: Mapping[Parameter, ParameterValueType] | Sequence[ParameterValueType]
     ) -> None:
         """
         [Deprecated method]Assign training parameters in the ``QuantumKernel`` feature map.
@@ -303,7 +324,7 @@ def assign_user_parameters(
 
     @property  # type: ignore
     @deprecate_property("0.5.0", new_name="training_parameter_binds")
-    def user_param_binds(self) -> Optional[Mapping[Parameter, float]]:
+    def user_param_binds(self) -> Mapping[Parameter, float] | None:
         """
         [Deprecated property]Return a copy of the current training parameter mappings
         for the feature map circuit.
@@ -312,7 +333,7 @@ def user_param_binds(self) -> Optional[Mapping[Parameter, float]]:
 
     @deprecate_method("0.5.0", new_name="bind_training_parameters")
     def bind_user_parameters(
-        self, values: Union[Mapping[Parameter, ParameterValueType], Sequence[ParameterValueType]]
+        self, values: Mapping[Parameter, ParameterValueType] | Sequence[ParameterValueType]
     ) -> None:
         """
         [Deprecated method]Alternate function signature for ``assign_training_parameters``
@@ -509,13 +530,14 @@ def evaluate(self, x_vec: np.ndarray, y_vec: np.ndarray = None) -> np.ndarray:
         # initialize kernel matrix
         kernel = np.zeros((x_vec.shape[0], y_vec.shape[0]))
 
-        # set diagonal to 1 if symmetric
-        if is_symmetric:
-            np.fill_diagonal(kernel, 1)
-
         # get indices to calculate
         if is_symmetric:
-            mus, nus = np.triu_indices(x_vec.shape[0], k=1)  # remove diagonal
+            if self._evaluate_duplicates == "all":
+                mus, nus = np.triu_indices(x_vec.shape[0])
+            else:
+                # exclude diagonal and fill it with ones
+                mus, nus = np.triu_indices(x_vec.shape[0], k=1)
+                np.fill_diagonal(kernel, 1)
         else:
             mus, nus = np.indices((x_vec.shape[0], y_vec.shape[0]))
             mus = np.asarray(mus.flat)
@@ -559,15 +581,21 @@ def evaluate(self, x_vec: np.ndarray, y_vec: np.ndarray = None) -> np.ndarray:
                     statevectors.append(results.get_statevector(j))
 
             offset = 0 if is_symmetric else len(x_vec)
-            matrix_elements = [
-                self._compute_overlap(idx, statevectors, is_statevector_sim, measurement_basis)
-                for idx in list(zip(mus, nus + offset))
-            ]
+            for (i, j) in zip(mus, nus):
+                x_i = x_vec[i]
+                y_j = y_vec[j]
+
+                # fill in ones for identical samples
+                if np.all(x_i == y_j) and self._evaluate_duplicates == "none":
+                    kernel_value = 1.0
+                else:
+                    kernel_value = self._compute_overlap(
+                        [i, j + offset], statevectors, is_statevector_sim, measurement_basis
+                    )
 
-            for i, j, value in zip(mus, nus, matrix_elements):
-                kernel[i, j] = value
+                kernel[i, j] = kernel_value
                 if is_symmetric:
-                    kernel[j, i] = kernel[i, j]
+                    kernel[j, i] = kernel_value
 
         else:  # not using state vector simulator
             feature_map_params_x = ParameterVector("par_x", self._feature_map.num_parameters)
@@ -590,7 +618,14 @@ def evaluate(self, x_vec: np.ndarray, y_vec: np.ndarray = None) -> np.ndarray:
                     j = nus[sub_idx]
                     x_i = x_vec[i]
                     y_j = y_vec[j]
-                    if not np.all(x_i == y_j):
+
+                    # fill in ones for identical samples
+                    if np.all(x_i == y_j) and self._evaluate_duplicates == "none":
+                        kernel[i, j] = 1
+                        if is_symmetric:
+                            kernel[j, i] = 1
+                    else:
+                        # otherwise evaluate the element
                         to_be_computed_data_pair.append((x_i, y_j))
                         to_be_computed_index.append((i, j))
 

releasenotes/notes/fix-quantum-kernel-duplicates-b75d6ed8a0f37f60.yaml

@@ -0,0 +1,23 @@
+---
+features:
+  - |
+    Introduced a new parameter `evaluate_duplicates` in
+    :class:`~qiskit_machine_learning.kernels.QuantumKernel`. This parameter defines a strategy how
+    kernel matrix elements are evaluated if duplicate samples are found.
+    Possible values are:
+
+      - ``all`` means that all kernel matrix elements are evaluated, even the diagonal ones when
+          training. This may introduce additional noise in the matrix.
+      - ``off_diagonal`` when training the matrix diagonal is set to `1`, the rest elements are
+          fully evaluated, e.g., for two identical samples in the dataset. When inferring, all
+          elements are evaluated. This is the default value.
+      - ``none`` when training the diagonal is set to `1` and if two identical samples are found
+          in the dataset the corresponding matrix element is set to `1`. When inferring, matrix
+          elements for identical samples are set to `1`.
+fixes:
+  - |
+    Fixed quantum kernel evaluation when duplicate samples are found in the dataset. Originally,
+    kernel matrix elements were not evaluated for identical samples in the dataset and such elements
+    were set wrongly to zero. Now we introduced a new parameter `evaluate_duplicates` that ensures
+    that elements of the kernel matrix are evaluated correctly. See the feature section for more
+    details.

test/kernels/test_qkernel.py

@@ -19,10 +19,10 @@
 
 import numpy as np
 import qiskit
-from ddt import data, ddt
+from ddt import data, ddt, idata, unpack
 from qiskit import BasicAer, QuantumCircuit
 from qiskit.circuit import Parameter
-from qiskit.circuit.library import ZZFeatureMap
+from qiskit.circuit.library import ZZFeatureMap, ZFeatureMap
 from qiskit.transpiler import PassManagerConfig
 from qiskit.transpiler.preset_passmanagers import level_1_pass_manager
 from qiskit.utils import QuantumInstance, algorithm_globals, optionals
@@ -706,5 +706,92 @@ def test_qasm_batching(self):
         self.assertEqual(sum(self.circuit_counts), num_circuits)
 
 
+@ddt
+class TestQuantumKernelEvaluateDuplicates(QiskitMachineLearningTestCase):
+    """Test QuantumKernel for duplicate evaluation."""
+
+    def count_circuits(self, func):
+        """Wrapper to record the number of circuits passed to QuantumInstance.execute.
+
+        Args:
+            func (Callable): execute function to be wrapped
+
+        Returns:
+            Callable: function wrapper
+        """
+
+        @functools.wraps(func)
+        def wrapper(*args, **kwds):
+            self.circuit_counts += len(args[0])
+            return func(*args, **kwds)
+
+        return wrapper
+
+    def setUp(self):
+        super().setUp()
+        algorithm_globals.random_seed = 10598
+        self.circuit_counts = 0
+
+        self.qasm_simulator = QuantumInstance(
+            BasicAer.get_backend("qasm_simulator"),
+            seed_simulator=algorithm_globals.random_seed,
+            seed_transpiler=algorithm_globals.random_seed,
+        )
+
+        # monkey patch the qasm simulator
+        self.qasm_simulator.execute = self.count_circuits(self.qasm_simulator.execute)
+
+        self.feature_map = ZFeatureMap(feature_dimension=2, reps=1)
+
+        self.properties = {
+            "no_dups": np.array([[1, 2], [2, 3], [3, 4]]),
+            "dups": np.array([[1, 2], [1, 2], [3, 4]]),
+            "y_vec": np.array([[0, 1], [1, 2]]),
+        }
+
+    @idata(
+        [
+            ("no_dups", "all", 6),
+            ("no_dups", "off_diagonal", 3),
+            ("no_dups", "none", 3),
+            ("dups", "all", 6),
+            ("dups", "off_diagonal", 3),
+            ("dups", "none", 2),
+        ]
+    )
+    @unpack
+    def test_evaluate_duplicates(self, dataset_name, evaluate_duplicates, expected_num_circuits):
+        """Tests symmetric quantum kernel evaluation with duplicate samples."""
+        self.circuit_counts = 0
+        qkernel = QuantumKernel(
+            feature_map=self.feature_map,
+            evaluate_duplicates=evaluate_duplicates,
+            quantum_instance=self.qasm_simulator,
+        )
+        qkernel.evaluate(self.properties.get(dataset_name))
+        self.assertEqual(self.circuit_counts, expected_num_circuits)
+
+    @idata(
+        [
+            ("no_dups", "all", 6),
+            ("no_dups", "off_diagonal", 6),
+            ("no_dups", "none", 5),
+        ]
+    )
+    @unpack
+    def test_evaluate_duplicates_not_symmetric(
+        self, dataset_name, evaluate_duplicates, expected_num_circuits
+    ):
+        """Tests non-symmetric quantum kernel evaluation with duplicate samples."""
+        self.circuit_counts = 0
+        qkernel = QuantumKernel(
+            feature_map=self.feature_map,
+            evaluate_duplicates=evaluate_duplicates,
+            quantum_instance=self.qasm_simulator,
+        )
+        qkernel.evaluate(self.properties.get(dataset_name), self.properties.get("y_vec"))
+        self.assertEqual(self.circuit_counts, expected_num_circuits)
+
+
 if __name__ == "__main__":
     unittest.main()