Evaluate quantum kernel matrix elements for identical samples

qiskit_machine_learning/kernels/quantum_kernel.py

@@ -11,6 +11,7 @@
 # that they have been altered from the originals.
 
 """Quantum Kernel Algorithm"""
+from __future__ import annotations
 
 from typing import Optional, Union, Sequence, Mapping, List
 import copy
@@ -60,6 +61,7 @@ def __init__(
         batch_size: int = 900,
         quantum_instance: Optional[Union[QuantumInstance, Backend]] = None,
         training_parameters: Optional[Union[ParameterVector, Sequence[Parameter]]] = None,
+        evaluate_duplicates: str | None = "non_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
+                identical 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.
+                    - ``non_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 = str(evaluate_duplicates).lower()
+        if eval_duplicates not in ("all", "non_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)
@@ -509,13 +532,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 +583,24 @@ 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 +623,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 identical 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.
+      - ``non_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
@@ -708,5 +708,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", "non_diagonal", 3),
+            ("no_dups", "none", 3),
+            ("dups", "all", 6),
+            ("dups", "non_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", "non_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()