(WIP) Add Support of Primitives V2 and ISA circuits for GroverOptimizer and QRAO

.pylintdict

@@ -151,6 +151,7 @@ panchenko
 param
 params
 parikh
+passmanager
 pauli
 paulis
 peleato

qiskit_optimization/algorithms/grover_optimizer.py

@@ -20,7 +20,8 @@
 import numpy as np
 from qiskit import QuantumCircuit, QuantumRegister
 from qiskit.circuit.library import QuadraticForm
-from qiskit.primitives import BaseSampler
+from qiskit.passmanager import BasePassManager
+from qiskit.primitives import BaseSamplerV1, BaseSamplerV2, SamplerResult
 from qiskit_algorithms import AmplificationProblem
 from qiskit_algorithms.amplitude_amplifiers.grover import Grover
 from qiskit_algorithms.utils import algorithm_globals
@@ -49,7 +50,8 @@ def __init__(
             Union[QuadraticProgramConverter, List[QuadraticProgramConverter]]
         ] = None,
         penalty: Optional[float] = None,
-        sampler: Optional[BaseSampler] = None,
+        sampler: Optional[Union[BaseSamplerV1, BaseSamplerV2]] = None,
+        passmanager: Optional[BasePassManager] = None,
     ) -> None:
         """
         Args:
@@ -62,6 +64,7 @@ def __init__(
             penalty: The penalty factor used in the default
                 :class:`~qiskit_optimization.converters.QuadraticProgramToQubo` converter
             sampler: A Sampler to use for sampling the results of the circuits.
+            passmanager: A pass manager to use to transpile the circuits
 
         Raises:
             ValueError: If both a quantum instance and sampler are set.
@@ -73,6 +76,7 @@ def __init__(
         self._circuit_results = {}  # type: dict
         self._converters = self._prepare_converters(converters, penalty)
         self._sampler = sampler
+        self._passmanager = passmanager
 
     def get_compatibility_msg(self, problem: QuadraticProgram) -> str:
         """Checks whether a given problem can be solved with this optimizer.
@@ -294,19 +298,28 @@ def _measure(self, circuit: QuantumCircuit) -> str:
 
     def _get_prob_dist(self, qc: QuantumCircuit) -> Dict[str, float]:
         """Gets probabilities from a given backend."""
+        # Transpile the circuit
+        if self._passmanager:
+            qc = self._passmanager.run(qc)
+
         # Execute job and filter results.
         job = self._sampler.run([qc])
 
         try:
             result = job.result()
         except Exception as exc:
             raise QiskitOptimizationError("Sampler job failed.") from exc
-        quasi_dist = result.quasi_dists[0]
-        raw_prob_dist = {
-            k: v
-            for k, v in quasi_dist.binary_probabilities(qc.num_qubits).items()
-            if v >= self._MIN_PROBABILITY
-        }
+
+        if isinstance(result, SamplerResult):
+            # SamplerV1
+            prob_dist = result.quasi_dists[0].binary_probabilities(qc.num_qubits)
+        else:
+            # SamplerV2
+            counts = getattr(result[0].data, qc.cregs[0].name).get_counts()
+            shots = sum(counts.values())
+            prob_dist = {k: v / shots for k, v in counts.items()}
+
+        raw_prob_dist = {k: v for k, v in prob_dist.items() if v >= self._MIN_PROBABILITY}
         prob_dist = {k[::-1]: v for k, v in raw_prob_dist.items()}
         self._circuit_results = {i: v**0.5 for i, v in raw_prob_dist.items()}
         return prob_dist

qiskit_optimization/algorithms/qrao/encoding_commutation_verifier.py

@@ -1,6 +1,6 @@
 # This code is part of a Qiskit project.
 #
-# (C) Copyright IBM 2023.
+# (C) Copyright IBM 2023, 2024.
 #
 # This code is licensed under the Apache License, Version 2.0. You may
 # obtain a copy of this license in the LICENSE.txt file in the root directory
@@ -14,7 +14,8 @@
 
 from __future__ import annotations
 
-from qiskit.primitives import BaseEstimator
+from qiskit.passmanager import BasePassManager
+from qiskit.primitives import BaseEstimatorV1, BaseEstimatorV2
 
 from qiskit_optimization.exceptions import QiskitOptimizationError
 
@@ -24,14 +25,21 @@
 class EncodingCommutationVerifier:
     """Class for verifying that the relaxation commutes with the objective function."""
 
-    def __init__(self, encoding: QuantumRandomAccessEncoding, estimator: BaseEstimator):
+    def __init__(
+        self,
+        encoding: QuantumRandomAccessEncoding,
+        estimator: BaseEstimatorV1 | BaseEstimatorV2,
+        passmanager: BasePassManager | None = None,
+    ):
         """
         Args:
             encoding: The encoding to verify.
             estimator: The estimator to use for the verification.
+            passmanager: The pass manager to transpile the circuits
         """
         self._encoding = encoding
         self._estimator = estimator
+        self._passmanager = passmanager
 
     def __len__(self) -> int:
         return 2**self._encoding.num_vars
@@ -48,6 +56,8 @@ def __getitem__(self, i: int) -> tuple[str, float, float]:
         str_dvars = f"{i:0{encoding.num_vars}b}"
         dvars = [int(b) for b in str_dvars]
         encoded_bitstr_qc = encoding.state_preparation_circuit(dvars)
+        if self._passmanager:
+            encoded_bitstr_qc = self._passmanager.run(encoded_bitstr_qc)
 
         # Evaluate the original objective function
         problem = encoding.problem
@@ -58,13 +68,24 @@ def __getitem__(self, i: int) -> tuple[str, float, float]:
         encoded_op = encoding.qubit_op
         offset = encoding.offset
 
-        job = self._estimator.run([encoded_bitstr_qc], [encoded_op])
-
-        try:
-            encoded_obj_val = job.result().values[0] + offset
-        except Exception as exc:
-            raise QiskitOptimizationError(
-                "The primitive job to verify commutation failed!"
-            ) from exc
-
-        return (str_dvars, obj_val, encoded_obj_val)
+        if isinstance(self._estimator, BaseEstimatorV1):
+            job = self._estimator.run([encoded_bitstr_qc], [encoded_op])
+
+            try:
+                encoded_obj_val = job.result().values[0] + offset
+            except Exception as exc:
+                raise QiskitOptimizationError(
+                    "The primitive job to verify commutation failed!"
+                ) from exc
+        else:  # BaseEstimatorV2
+            job = self._estimator.run([(encoded_bitstr_qc, encoded_op)])
+
+            try:
+                result = job.result()
+                encoded_obj_val = result[0].data.evs.item() + offset
+            except Exception as exc:
+                raise QiskitOptimizationError(
+                    "The primitive job to verify commutation failed!"
+                ) from exc
+
+        return str_dvars, obj_val, encoded_obj_val

qiskit_optimization/algorithms/qrao/magic_rounding.py

@@ -17,7 +17,8 @@
 
 import numpy as np
 from qiskit import QuantumCircuit
-from qiskit.primitives import BaseSampler
+from qiskit.passmanager import BasePassManager
+from qiskit.primitives import BaseSamplerV1, BaseSamplerV2, SamplerResult
 from qiskit.quantum_info import SparsePauliOp
 from qiskit_algorithms.exceptions import AlgorithmError
 
@@ -58,9 +59,10 @@ class MagicRounding(RoundingScheme):
 
     def __init__(
         self,
-        sampler: BaseSampler,
+        sampler: BaseSamplerV1 | BaseSamplerV2,
         basis_sampling: str = "uniform",
         seed: int | None = None,
+        passmanager: BasePassManager | None = None,
     ):
         """
         Args:
@@ -76,6 +78,7 @@ def __init__(
                 sampling.
             seed: Seed for random number generator, which is used to sample the
                 magic bases.
+            passmanager: Pass manager to transpile the circuits
 
         Raises:
             ValueError: If ``basis_sampling`` is not ``"uniform"`` or ``"weighted"``.
@@ -89,13 +92,23 @@ def __init__(
         self._sampler = sampler
         self._rng = np.random.default_rng(seed)
         self._basis_sampling = basis_sampling
-        if self._sampler.options.get("shots") is None:
-            raise ValueError("Magic rounding requires a sampler configured with a number of shots.")
-        self._shots = sampler.options.shots
+        self._passmanager = passmanager
+        if isinstance(self._sampler, BaseSamplerV1):
+            if self._sampler.options.get("shots") is None:
+                raise ValueError(
+                    "Magic rounding requires a sampler configured with a number of shots."
+                )
+            self._shots = sampler.options.shots
+        else:  # BaseSamplerV2
+            if self._sampler.default_shots is None:
+                raise ValueError(
+                    "Magic rounding requires a sampler configured with a number of shots."
+                )
+            self._shots = self._sampler.default_shots
         super().__init__()
 
     @property
-    def sampler(self) -> BaseSampler:
+    def sampler(self) -> BaseSamplerV1 | BaseSamplerV2:
         """Returns the Sampler used to sample the magic bases."""
         return self._sampler
 
@@ -166,6 +179,8 @@ def _evaluate_magic_bases(
             QiskitOptimizationError: If some of the results from the primitive job are not collected.
         """
         circuits = self._make_circuits(circuit, bases, vars_per_qubit)
+        if self._passmanager:
+            circuits = self._passmanager.run(circuits)
         # Execute each of the rotated circuits and collect the results
         # Batch the circuits into jobs where each group has the same number of
         # shots, so that you can wait for the queue as few times as possible if
@@ -182,14 +197,27 @@ def _evaluate_magic_bases(
             circuit_indices_by_shots[shots].append(i)
 
         for shots, indices in sorted(circuit_indices_by_shots.items(), reverse=True):
+            circuits_ = [circuits[i] for i in indices]
             try:
-                job = self._sampler.run([circuits[i] for i in indices], shots=shots)
+                job = self._sampler.run(circuits_, shots=shots)
                 result = job.result()
             except Exception as exc:
                 raise AlgorithmError(
                     "The primitive job to evaluate the magic state failed."
                 ) from exc
-            counts_list = [dist.binary_probabilities() for dist in result.quasi_dists]
+
+            if isinstance(result, SamplerResult):
+                counts_list = [dist.binary_probabilities() for dist in result.quasi_dists]
+            else:
+                counts_list = [
+                    getattr(res.data, circ.cregs[0].name).get_counts()
+                    for res, circ in zip(result, circuits_)
+                ]
+                counts_list = [
+                    {k: v / sum(counts.values()) for k, v in counts.items()}
+                    for counts in counts_list
+                ]
+
             if len(counts_list) != len(indices):
                 raise QiskitOptimizationError(
                     "Internal error: The number of circuits and the results from the primitive job "

test/algorithms/qrao/test_magic_rounding.py

@@ -15,8 +15,9 @@
 from test.optimization_test_case import QiskitOptimizationTestCase
 
 import numpy as np
+from ddt import data, ddt
 from qiskit.circuit import QuantumCircuit
-from qiskit.primitives import Sampler
+from qiskit.primitives import Sampler, StatevectorSampler
 from qiskit_algorithms import NumPyMinimumEigensolver
 
 from qiskit_optimization.algorithms import OptimizationResultStatus, SolutionSample
@@ -30,6 +31,7 @@
 from qiskit_optimization.problems import QuadraticProgram
 
 
+@ddt
 class TestMagicRounding(QiskitOptimizationTestCase):
     """MagicRounding tests."""
 
@@ -41,10 +43,15 @@ def setUp(self):
         self.problem.binary_var("y")
         self.problem.binary_var("z")
         self.problem.minimize(linear={"x": 1, "y": 2, "z": 3})
+        self._sampler = {
+            "v1": Sampler(options={"shots": 10000, "seed": 42}),
+            "v2": StatevectorSampler(default_shots=10000, seed=42),
+        }
 
-    def test_magic_rounding_constructor(self):
+    @data("v1", "v2")
+    def test_magic_rounding_constructor(self, version):
         """Test constructor"""
-        sampler = Sampler(options={"shots": 10000, "seed": 42})
+        sampler = self._sampler[version]
         # test default
         magic_rounding = MagicRounding(sampler)
         self.assertEqual(magic_rounding.sampler, sampler)
@@ -61,9 +68,10 @@ def test_magic_rounding_constructor(self):
         with self.assertRaises(ValueError):
             MagicRounding(sampler, basis_sampling="invalid")
 
-    def test_magic_rounding_round_uniform_1_1_qrac(self):
+    @data("v1", "v2")
+    def test_magic_rounding_round_uniform_1_1_qrac(self, version):
         """Test round method with uniform basis sampling for max_vars_per_qubit=1"""
-        sampler = Sampler(options={"shots": 10000, "seed": 42})
+        sampler = self._sampler[version]
         encoding = QuantumRandomAccessEncoding(max_vars_per_qubit=1)
         encoding.encode(self.problem)
         np_solver = NumPyMinimumEigensolver()
@@ -91,9 +99,10 @@ def test_magic_rounding_round_uniform_1_1_qrac(self):
             [1, 1, 1],
         )
 
-    def test_magic_rounding_round_weighted_1_1_qrac(self):
+    @data("v1", "v2")
+    def test_magic_rounding_round_weighted_1_1_qrac(self, version):
         """Test round method with uniform basis sampling for max_vars_per_qubit=1"""
-        sampler = Sampler(options={"shots": 10000, "seed": 42})
+        sampler = self._sampler[version]
         encoding = QuantumRandomAccessEncoding(max_vars_per_qubit=1)
         encoding.encode(self.problem)
         np_solver = NumPyMinimumEigensolver()
@@ -121,9 +130,10 @@ def test_magic_rounding_round_weighted_1_1_qrac(self):
             [1, 1, 1],
         )
 
-    def test_magic_rounding_round_uniform_2_1_qrac(self):
+    @data("v1", "v2")
+    def test_magic_rounding_round_uniform_2_1_qrac(self, version):
         """Test round method with uniform basis sampling for max_vars_per_qubit=2"""
-        sampler = Sampler(options={"shots": 10000, "seed": 42})
+        sampler = self._sampler[version]
         encoding = QuantumRandomAccessEncoding(max_vars_per_qubit=2)
         encoding.encode(self.problem)
         np_solver = NumPyMinimumEigensolver()
@@ -142,7 +152,7 @@ def test_magic_rounding_round_uniform_2_1_qrac(self):
         ]
         for i, basis_counts in enumerate(rounding_result.basis_counts):
             for key, value in basis_counts.items():
-                self.assertAlmostEqual(value, expected_basis_counts[i][key], delta=50)
+                self.assertAlmostEqual(value, expected_basis_counts[i][key], delta=70)
         samples = rounding_result.samples
         samples.sort(key=lambda sample: np.array2string(sample.x))
         expected_samples = [
@@ -163,9 +173,10 @@ def test_magic_rounding_round_uniform_2_1_qrac(self):
             [0.44721359549995743, 0.8944271909999162, 1],
         )
 
-    def test_magic_rounding_round_weighted_2_1_qrac(self):
+    @data("v1", "v2")
+    def test_magic_rounding_round_weighted_2_1_qrac(self, version):
         """Test round method with weighted basis sampling for max_vars_per_qubit=2"""
-        sampler = Sampler(options={"shots": 10000, "seed": 42})
+        sampler = self._sampler[version]
         encoding = QuantumRandomAccessEncoding(max_vars_per_qubit=2)
         encoding.encode(self.problem)
         np_solver = NumPyMinimumEigensolver()
@@ -182,7 +193,7 @@ def test_magic_rounding_round_weighted_2_1_qrac(self):
         ]
         for i, basis_counts in enumerate(rounding_result.basis_counts):
             for key, value in basis_counts.items():
-                self.assertAlmostEqual(value, expected_basis_counts[i][key], delta=50)
+                self.assertAlmostEqual(value, expected_basis_counts[i][key], delta=70)
         samples = rounding_result.samples
         samples.sort(key=lambda sample: np.array2string(sample.x))
         expected_samples = [
@@ -203,9 +214,10 @@ def test_magic_rounding_round_weighted_2_1_qrac(self):
             [0.44721359549995743, 0.8944271909999162, 1],
         )
 
-    def test_magic_rounding_round_uniform_3_1_qrac(self):
+    @data("v1", "v2")
+    def test_magic_rounding_round_uniform_3_1_qrac(self, version):
         """Test round method with uniform basis sampling for max_vars_per_qubit=3"""
-        sampler = Sampler(options={"shots": 10000, "seed": 42})
+        sampler = self._sampler[version]
         encoding = QuantumRandomAccessEncoding(max_vars_per_qubit=3)
         encoding.encode(self.problem)
         np_solver = NumPyMinimumEigensolver()
@@ -245,14 +257,15 @@ def test_magic_rounding_round_uniform_3_1_qrac(self):
             [0.2672612419124245, 0.5345224838248487, 0.8017837257372733],
         )
 
-    def test_magic_rounding_round_weighted_3_1_qrac(self):
+    @data("v1", "v2")
+    def test_magic_rounding_round_weighted_3_1_qrac(self, version):
         """Test round method with weighted basis sampling for max_vars_per_qubit=3"""
         encoding = QuantumRandomAccessEncoding(max_vars_per_qubit=3)
         encoding.encode(self.problem)
         np_solver = NumPyMinimumEigensolver()
         qrao = QuantumRandomAccessOptimizer(min_eigen_solver=np_solver)
         _, rounding_context = qrao.solve_relaxed(encoding=encoding)
-        sampler = Sampler(options={"shots": 10000, "seed": 42})
+        sampler = self._sampler[version]
         magic_rounding = MagicRounding(sampler, basis_sampling="weighted", seed=42)
         rounding_result = magic_rounding.round(rounding_context)
         self.assertIsInstance(rounding_result, RoundingResult)