Adds observable evaluator with primitives.

qiskit/algorithms/__init__.py

@@ -262,6 +262,7 @@
    :toctree: ../stubs/
 
    eval_observables
+   estimate_observables
 
 Utility classes
 ---------------
@@ -319,6 +320,7 @@
 )
 from .exceptions import AlgorithmError
 from .aux_ops_evaluator import eval_observables
+from .observables_evaluator import estimate_observables
 from .evolvers.trotterization import TrotterQRTE
 from .evolvers.variational.var_qite import VarQITE
 from .evolvers.variational.var_qrte import VarQRTE
@@ -382,4 +384,5 @@
     "IterativePhaseEstimation",
     "AlgorithmError",
     "eval_observables",
+    "estimate_observables",
 ]

qiskit/algorithms/aux_ops_evaluator.py

@@ -26,10 +26,18 @@
 from qiskit.providers import Backend
 from qiskit.quantum_info import Statevector
 from qiskit.utils import QuantumInstance
+from qiskit.utils.deprecation import deprecate_function
 
 from .list_or_dict import ListOrDict
 
 
+@deprecate_function(
+    "The eval_observables function has been superseded by the "
+    "qiskit.algorithms.observables_evaluator.estimate_observables function. "
+    "This function will be deprecated in a future release and subsequently "
+    "removed after that.",
+    category=PendingDeprecationWarning,
+)
 def eval_observables(
     quantum_instance: Union[QuantumInstance, Backend],
     quantum_state: Union[
@@ -42,10 +50,16 @@ def eval_observables(
     threshold: float = 1e-12,
 ) -> ListOrDict[Tuple[complex, complex]]:
     """
-    Accepts a list or a dictionary of operators and calculates their expectation values - means
+    Pending deprecation: Accepts a list or a dictionary of operators and calculates
+    their expectation values - means
     and standard deviations. They are calculated with respect to a quantum state provided. A user
     can optionally provide a threshold value which filters mean values falling below the threshold.
 
+    This function has been superseded by the
+    :func:`qiskit.algorithms.observables_evaluator.eval_observables` function.
+    It will be deprecated in a future release and subsequently
+    removed after that.
+
     Args:
         quantum_instance: A quantum instance used for calculations.
         quantum_state: An unparametrized quantum circuit representing a quantum state that

qiskit/algorithms/observables_evaluator.py

@@ -0,0 +1,157 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2021, 2022.
+#
+# 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
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+"""Evaluator of observables for algorithms."""
+from __future__ import annotations
+
+import numpy as np
+
+from qiskit import QuantumCircuit
+from qiskit.opflow import PauliSumOp
+from .exceptions import AlgorithmError
+from .list_or_dict import ListOrDict
+from ..primitives import EstimatorResult, BaseEstimator
+from ..quantum_info.operators.base_operator import BaseOperator
+
+
+def estimate_observables(
+    estimator: BaseEstimator,
+    quantum_state: QuantumCircuit,
+    observables: ListOrDict[BaseOperator | PauliSumOp],
+    threshold: float = 1e-12,
+) -> ListOrDict[tuple[complex, tuple[complex, int]]]:
+    """
+    Accepts a sequence of operators and calculates their expectation values - means
+    and standard deviations. They are calculated with respect to a quantum state provided. A user
+    can optionally provide a threshold value which filters mean values falling below the threshold.
+
+    Args:
+        estimator: An estimator primitive used for calculations.
+        quantum_state: An unparametrized quantum circuit representing a quantum state that
+            expectation values are computed against.
+        observables: A list or a dictionary of operators whose expectation values are to be
+            calculated.
+        threshold: A threshold value that defines which mean values should be neglected (helpful for
+            ignoring numerical instabilities close to 0).
+
+    Returns:
+        A list or a dictionary of tuples (mean, (variance, shots)).
+
+    Raises:
+        ValueError: If a ``quantum_state`` with free parameters is provided.
+        AlgorithmError: If a primitive job is not successful.
+    """
+
+    if (
+        isinstance(quantum_state, QuantumCircuit)  # State cannot be parametrized
+        and len(quantum_state.parameters) > 0
+    ):
+        raise ValueError(
+            "A parametrized representation of a quantum_state was provided. It is not "
+            "allowed - it cannot have free parameters."
+        )
+    if isinstance(observables, dict):
+        observables_list = list(observables.values())
+    else:
+        observables_list = observables
+
+    observables_list = _handle_zero_ops(observables_list)
+    quantum_state = [quantum_state] * len(observables)
+    try:
+        estimator_job = estimator.run(quantum_state, observables_list)
+        expectation_values = estimator_job.result().values
+    except Exception as exc:
+        raise AlgorithmError("The primitive job failed!") from exc
+
+    variance_and_shots = _prep_variance_and_shots(estimator_job, len(expectation_values))
+
+    # Discard values below threshold
+    observables_means = expectation_values * (np.abs(expectation_values) > threshold)
+    # zip means and standard deviations into tuples
+    observables_results = list(zip(observables_means, variance_and_shots))
+
+    return _prepare_result(observables_results, observables)
+
+
+def _handle_zero_ops(
+    observables_list: list[BaseOperator | PauliSumOp],
+) -> list[BaseOperator | PauliSumOp]:
+    """Replaces all occurrence of operators equal to 0 in the list with an equivalent ``PauliSumOp``
+    operator."""
+    if observables_list:
+        zero_op = PauliSumOp.from_list([("I" * observables_list[0].num_qubits, 0)])
+        for ind, observable in enumerate(observables_list):
+            if observable == 0:
+                observables_list[ind] = zero_op
+    return observables_list
+
+
+def _prepare_result(
+    observables_results: list[tuple[complex, tuple[complex, int]]],
+    observables: ListOrDict[BaseOperator | PauliSumOp],
+) -> ListOrDict[tuple[complex, tuple[complex, int]]]:
+    """
+    Prepares a list of tuples of eigenvalues and (variance, shots) tuples from
+    ``observables_results`` and ``observables``.
+
+    Args:
+        observables_results: A list of tuples (mean, (variance, shots)).
+        observables: A list or a dictionary of operators whose expectation values are to be
+            calculated.
+
+    Returns:
+        A list or a dictionary of tuples (mean, (variance, shots)).
+    """
+
+    if isinstance(observables, list):
+        # by construction, all None values will be overwritten
+        observables_eigenvalues = [None] * len(observables)
+        key_value_iterator = enumerate(observables_results)
+    else:
+        observables_eigenvalues = {}
+        key_value_iterator = zip(observables.keys(), observables_results)
+
+    for key, value in key_value_iterator:
+        observables_eigenvalues[key] = value
+    return observables_eigenvalues
+
+
+def _prep_variance_and_shots(
+    estimator_result: EstimatorResult,
+    results_length: int,
+) -> list[tuple[complex, int]]:
+    """
+    Prepares a list of tuples with variances and shots from results provided by expectation values
+    calculations. If there is no variance or shots data available from a primitive, the values will
+    be set to ``0``.
+
+    Args:
+        estimator_result: An estimator result.
+        results_length: Number of expectation values calculated.
+
+    Returns:
+        A list of tuples of the form (variance, shots).
+    """
+    if not estimator_result.metadata:
+        return [(0, 0)] * results_length
+
+    results = []
+    for metadata in estimator_result.metadata:
+        variance, shots = 0.0, 0
+        if metadata:
+            if "variance" in metadata.keys():
+                variance = metadata["variance"]
+            if "shots" in metadata.keys():
+                shots = metadata["shots"]
+
+        results.append((variance, shots))
+
+    return results

releasenotes/notes/observable-eval-primitives-e1fd989e15c7760c.yaml

@@ -0,0 +1,12 @@
+---
+features:
+  - |
+    Added :meth:`qiskit.algorithms.observables_evaluator.eval_observables` with
+    :class:`qiskit.primitives.BaseEstimator` as ``init`` parameter. It will soon replace
+    :meth:`qiskit.algorithms.aux_ops_evaluator.eval_observables`.
+deprecations:
+  - |
+    Using :meth:`qiskit.algorithms.aux_ops_evaluator.eval_observables` will now issue a
+    ``PendingDeprecationWarning``. This method will be deprecated in a future release and
+    subsequently removed after that. This is being replaced by the new
+    :meth:`qiskit.algorithms.observables_evaluator.eval_observables` primitive-enabled method.

test/python/algorithms/test_observables_evaluator.py

@@ -0,0 +1,157 @@
+# This code is part of Qiskit.
+#
+# (C) Copyright IBM 2022.
+#
+# 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
+# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
+#
+# Any modifications or derivative works of this code must retain this
+# copyright notice, and modified files need to carry a notice indicating
+# that they have been altered from the originals.
+"""Tests evaluator of auxiliary operators for algorithms."""
+from __future__ import annotations
+import unittest
+from typing import Tuple
+
+from test.python.algorithms import QiskitAlgorithmsTestCase
+import numpy as np
+from ddt import ddt, data
+
+from qiskit.algorithms.list_or_dict import ListOrDict
+from qiskit.quantum_info.operators.base_operator import BaseOperator
+from qiskit.algorithms import estimate_observables
+from qiskit.primitives import Estimator
+from qiskit.quantum_info import Statevector, SparsePauliOp
+from qiskit import QuantumCircuit
+from qiskit.circuit.library import EfficientSU2
+from qiskit.opflow import PauliSumOp
+from qiskit.utils import algorithm_globals
+
+
+@ddt
+class TestObservablesEvaluator(QiskitAlgorithmsTestCase):
+    """Tests evaluator of auxiliary operators for algorithms."""
+
+    def setUp(self):
+        super().setUp()
+        self.seed = 50
+        algorithm_globals.random_seed = self.seed
+
+        self.threshold = 1e-8
+
+    def get_exact_expectation(
+        self, ansatz: QuantumCircuit, observables: ListOrDict[BaseOperator | PauliSumOp]
+    ):
+        """
+        Calculates the exact expectation to be used as an expected result for unit tests.
+        """
+        if isinstance(observables, dict):
+            observables_list = list(observables.values())
+        else:
+            observables_list = observables
+        # the exact value is a list of (mean, (variance, shots)) where we expect 0 variance and
+        # 0 shots
+        exact = [
+            (Statevector(ansatz).expectation_value(observable), (0, 0))
+            for observable in observables_list
+        ]
+
+        if isinstance(observables, dict):
+            return dict(zip(observables.keys(), exact))
+
+        return exact
+
+    def _run_test(
+        self,
+        expected_result: ListOrDict[Tuple[complex, complex]],
+        quantum_state: QuantumCircuit,
+        decimal: int,
+        observables: ListOrDict[BaseOperator | PauliSumOp],
+        estimator: Estimator,
+    ):
+        result = estimate_observables(estimator, quantum_state, observables, self.threshold)
+
+        if isinstance(observables, dict):
+            np.testing.assert_equal(list(result.keys()), list(expected_result.keys()))
+            means = [element[0] for element in result.values()]
+            expected_means = [element[0] for element in expected_result.values()]
+            np.testing.assert_array_almost_equal(means, expected_means, decimal=decimal)
+
+            vars_and_shots = [element[1] for element in result.values()]
+            expected_vars_and_shots = [element[1] for element in expected_result.values()]
+            np.testing.assert_array_equal(vars_and_shots, expected_vars_and_shots)
+        else:
+            means = [element[0] for element in result]
+            expected_means = [element[0] for element in expected_result]
+            np.testing.assert_array_almost_equal(means, expected_means, decimal=decimal)
+
+            vars_and_shots = [element[1] for element in result]
+            expected_vars_and_shots = [element[1] for element in expected_result]
+            np.testing.assert_array_equal(vars_and_shots, expected_vars_and_shots)
+
+    @data(
+        [
+            PauliSumOp.from_list([("II", 0.5), ("ZZ", 0.5), ("YY", 0.5), ("XX", -0.5)]),
+            PauliSumOp.from_list([("II", 2.0)]),
+        ],
+        [
+            PauliSumOp.from_list([("ZZ", 2.0)]),
+        ],
+        {
+            "op1": PauliSumOp.from_list([("II", 2.0)]),
+            "op2": PauliSumOp.from_list([("II", 0.5), ("ZZ", 0.5), ("YY", 0.5), ("XX", -0.5)]),
+        },
+        {
+            "op1": PauliSumOp.from_list([("ZZ", 2.0)]),
+        },
+        [],
+        {},
+    )
+    def test_estimate_observables(self, observables: ListOrDict[BaseOperator | PauliSumOp]):
+        """Tests evaluator of auxiliary operators for algorithms."""
+
+        ansatz = EfficientSU2(2)
+        parameters = np.array(
+            [1.2, 4.2, 1.4, 2.0, 1.2, 4.2, 1.4, 2.0, 1.2, 4.2, 1.4, 2.0, 1.2, 4.2, 1.4, 2.0],
+            dtype=float,
+        )
+
+        bound_ansatz = ansatz.bind_parameters(parameters)
+        states = bound_ansatz
+        expected_result = self.get_exact_expectation(bound_ansatz, observables)
+        estimator = Estimator()
+        decimal = 6
+        self._run_test(
+            expected_result,
+            states,
+            decimal,
+            observables,
+            estimator,
+        )
+
+    def test_estimate_observables_zero_op(self):
+        """Tests if a zero operator is handled correctly."""
+        ansatz = EfficientSU2(2)
+        parameters = np.array(
+            [1.2, 4.2, 1.4, 2.0, 1.2, 4.2, 1.4, 2.0, 1.2, 4.2, 1.4, 2.0, 1.2, 4.2, 1.4, 2.0],
+            dtype=float,
+        )
+
+        bound_ansatz = ansatz.bind_parameters(parameters)
+        state = bound_ansatz
+        estimator = Estimator()
+        observables = [SparsePauliOp(["XX", "YY"]), 0]
+        result = estimate_observables(estimator, state, observables, self.threshold)
+        expected_result = [(0.015607318055509564, (0, 0)), (0.0, (0, 0))]
+        means = [element[0] for element in result]
+        expected_means = [element[0] for element in expected_result]
+        np.testing.assert_array_almost_equal(means, expected_means, decimal=0.01)
+
+        vars_and_shots = [element[1] for element in result]
+        expected_vars_and_shots = [element[1] for element in expected_result]
+        np.testing.assert_array_equal(vars_and_shots, expected_vars_and_shots)
+
+
+if __name__ == "__main__":
+    unittest.main()