Support Dict[str, OperatorBase] for aux_operators (fix #6772)

qiskit/algorithms/eigen_solvers/eigen_solver.py

@@ -13,12 +13,16 @@
 """The Eigensolver interface"""
 
 from abc import ABC, abstractmethod
-from typing import List, Optional
+from typing import Dict, Optional, Any, List, Union, TypeVar
 
 import numpy as np
 from qiskit.opflow import OperatorBase
 from ..algorithm_result import AlgorithmResult
 
+# Introduced new type to maintain readability.
+T = TypeVar('T')
+ListOrDict = Union[List[Optional[T]], Dict[Any, T]]
+
 
 class Eigensolver(ABC):
     """The Eigensolver Interface.
@@ -30,7 +34,7 @@ class Eigensolver(ABC):
 
     @abstractmethod
     def compute_eigenvalues(
-        self, operator: OperatorBase, aux_operators: Optional[List[Optional[OperatorBase]]] = None
+        self, operator: OperatorBase, aux_operators: Optional[ListOrDict[OperatorBase]] = None
     ) -> "EigensolverResult":
         """
         Computes eigenvalues. Operator and aux_operators can be supplied here and

qiskit/algorithms/eigen_solvers/numpy_eigen_solver.py

@@ -13,14 +13,14 @@
 """The Eigensolver algorithm."""
 
 import logging
-from typing import List, Optional, Union, Tuple, Callable
+from typing import List, Optional, Union, Tuple, Callable, Dict, Any, TypeVar
 
 import numpy as np
 from scipy import sparse as scisparse
 
 from qiskit.opflow import OperatorBase, I, StateFn, ListOp
 from qiskit.utils.validation import validate_min
-from .eigen_solver import Eigensolver, EigensolverResult
+from .eigen_solver import Eigensolver, EigensolverResult, ListOrDict
 from ..exceptions import AlgorithmError
 
 logger = logging.getLogger(__name__)
@@ -46,7 +46,7 @@ def __init__(
         self,
         k: int = 1,
         filter_criterion: Callable[
-            [Union[List, np.ndarray], float, Optional[List[float]]], bool
+            [Union[List, np.ndarray], float, Optional[ListOrDict[float]]], bool
         ] = None,
     ) -> None:
         """
@@ -84,15 +84,15 @@ def k(self, k: int) -> None:
     @property
     def filter_criterion(
         self,
-    ) -> Optional[Callable[[Union[List, np.ndarray], float, Optional[List[float]]], bool]]:
+    ) -> Optional[Callable[[Union[List, np.ndarray], float, Optional[ListOrDict[float]]], bool]]:
         """returns the filter criterion if set"""
         return self._filter_criterion
 
     @filter_criterion.setter
     def filter_criterion(
         self,
         filter_criterion: Optional[
-            Callable[[Union[List, np.ndarray], float, Optional[List[float]]], bool]
+            Callable[[Union[List, np.ndarray], float, Optional[ListOrDict[float]]], bool]
         ],
     ) -> None:
         """set the filter criterion"""
@@ -141,7 +141,7 @@ def _get_ground_state_energy(self, operator: OperatorBase) -> None:
             self._solve(operator)
 
     def _get_energies(
-        self, operator: OperatorBase, aux_operators: Optional[List[OperatorBase]]
+        self, operator: OperatorBase, aux_operators: Optional[ListOrDict[OperatorBase]]
     ) -> None:
         if self._ret.eigenvalues is None or self._ret.eigenstates is None:
             self._solve(operator)
@@ -156,12 +156,19 @@ def _get_energies(
 
     @staticmethod
     def _eval_aux_operators(
-        aux_operators: List[OperatorBase], wavefn, threshold: float = 1e-12
-    ) -> np.ndarray:
-        values = []  # type: List[Tuple[float, int]]
-        for operator in aux_operators:
+        aux_operators: ListOrDict[OperatorBase], wavefn, threshold: float = 1e-12
+    ) -> ListOrDict[Tuple[float, int]]:
+
+        # If aux_operators is a list, it can contain None operators for which None values are returned.
+        # If aux_operators is a dict, the None operators have been dropped in compute_eigenvalues.
+        if isinstance(aux_operators, list):
+            values = [None] * len(aux_operators)  # type: ListOrDict[Tuple[float, int]]
+            key_op_iterator = enumerate(aux_operators)
+        else:
+            values = {}
+            key_op_iterator = aux_operators.items()
+        for key, operator in key_op_iterator:
             if operator is None:
-                values.append(None)
                 continue
             value = 0.0
             if operator.coeff != 0:
@@ -174,22 +181,24 @@ def _eval_aux_operators(
                 else:
                     value = StateFn(operator, is_measurement=True).eval(wavefn)
                 value = value.real if abs(value.real) > threshold else 0.0
-            values.append((value, 0))
-        return np.array(values, dtype=object)
+            values[key] = (value, 0)
+        return values
 
     def compute_eigenvalues(
-        self, operator: OperatorBase, aux_operators: Optional[List[Optional[OperatorBase]]] = None
+        self, operator: OperatorBase, aux_operators: Optional[ListOrDict[OperatorBase]] = None
     ) -> EigensolverResult:
         super().compute_eigenvalues(operator, aux_operators)
 
         if operator is None:
             raise AlgorithmError("Operator was never provided")
 
         self._check_set_k(operator)
-        if aux_operators:
+        if isinstance(aux_operators, list) and len(aux_operators) > 0:
             zero_op = I.tensorpower(operator.num_qubits) * 0.0
             # For some reason Chemistry passes aux_ops with 0 qubits and paulis sometimes.
             aux_operators = [zero_op if op == 0 else op for op in aux_operators]
+        elif isinstance(aux_operators, dict) and len(aux_operators) > 0:
+            aux_operators = {key: op for key, op in aux_operators.items() if not op}
         else:
             aux_operators = None
 

qiskit/algorithms/minimum_eigen_solvers/minimum_eigen_solver.py

@@ -13,12 +13,16 @@
 """The Minimum Eigensolver interface"""
 
 from abc import ABC, abstractmethod
-from typing import List, Optional
+from typing import Dict, Optional, Any, List, Union, TypeVar
 
 import numpy as np
 from qiskit.opflow import OperatorBase
 from ..algorithm_result import AlgorithmResult
 
+# Introduced new type to maintain readability.
+T = TypeVar('T')
+ListOrDict = Union[List[Optional[T]], Dict[Any, T]]
+
 
 class MinimumEigensolver(ABC):
     """The Minimum Eigensolver Interface.
@@ -30,7 +34,7 @@ class MinimumEigensolver(ABC):
 
     @abstractmethod
     def compute_minimum_eigenvalue(
-        self, operator: OperatorBase, aux_operators: Optional[List[Optional[OperatorBase]]] = None
+        self, operator: OperatorBase, aux_operators: Optional[ListOrDict[OperatorBase]] = None
     ) -> "MinimumEigensolverResult":
         """
         Computes minimum eigenvalue. Operator and aux_operators can be supplied here and

qiskit/algorithms/minimum_eigen_solvers/numpy_minimum_eigen_solver.py

@@ -12,13 +12,13 @@
 
 """The Numpy Minimum Eigensolver algorithm."""
 
-from typing import List, Optional, Union, Callable
+from typing import List, Optional, Union, Callable, Dict
 import logging
 import numpy as np
 
 from qiskit.opflow import OperatorBase
 from ..eigen_solvers.numpy_eigen_solver import NumPyEigensolver
-from .minimum_eigen_solver import MinimumEigensolver, MinimumEigensolverResult
+from .minimum_eigen_solver import MinimumEigensolver, MinimumEigensolverResult, ListOrDict
 
 logger = logging.getLogger(__name__)
 
@@ -31,7 +31,7 @@ class NumPyMinimumEigensolver(MinimumEigensolver):
     def __init__(
         self,
         filter_criterion: Callable[
-            [Union[List, np.ndarray], float, Optional[List[float]]], bool
+            [Union[List, np.ndarray], float, Optional[ListOrDict[float]]], bool
         ] = None,
     ) -> None:
         """
@@ -49,15 +49,15 @@ def __init__(
     @property
     def filter_criterion(
         self,
-    ) -> Optional[Callable[[Union[List, np.ndarray], float, Optional[List[float]]], bool]]:
+    ) -> Optional[Callable[[Union[List, np.ndarray], float, Optional[ListOrDict[float]]], bool]]:
         """returns the filter criterion if set"""
         return self._ces.filter_criterion
 
     @filter_criterion.setter
     def filter_criterion(
         self,
         filter_criterion: Optional[
-            Callable[[Union[List, np.ndarray], float, Optional[List[float]]], bool]
+            Callable[[Union[List, np.ndarray], float, Optional[ListOrDict[float]]], bool]
         ],
     ) -> None:
         """set the filter criterion"""
@@ -68,7 +68,7 @@ def supports_aux_operators(cls) -> bool:
         return NumPyEigensolver.supports_aux_operators()
 
     def compute_minimum_eigenvalue(
-        self, operator: OperatorBase, aux_operators: Optional[List[Optional[OperatorBase]]] = None
+        self, operator: OperatorBase, aux_operators: Optional[ListOrDict[OperatorBase]] = None
     ) -> MinimumEigensolverResult:
         super().compute_minimum_eigenvalue(operator, aux_operators)
         result_ces = self._ces.compute_eigenvalues(operator, aux_operators)

qiskit/algorithms/minimum_eigen_solvers/vqe.py

@@ -42,7 +42,7 @@
 from qiskit.utils import QuantumInstance, algorithm_globals
 from ..optimizers import Optimizer, SLSQP
 from ..variational_algorithm import VariationalAlgorithm, VariationalResult
-from .minimum_eigen_solver import MinimumEigensolver, MinimumEigensolverResult
+from .minimum_eigen_solver import MinimumEigensolver, MinimumEigensolverResult, ListOrDict
 from ..exceptions import AlgorithmError
 
 logger = logging.getLogger(__name__)
@@ -411,32 +411,49 @@ def supports_aux_operators(cls) -> bool:
     def _eval_aux_ops(
         self,
         parameters: np.ndarray,
-        aux_operators: List[OperatorBase],
+        aux_operators: ListOrDict[OperatorBase],
         expectation: ExpectationBase,
         threshold: float = 1e-12,
     ) -> np.ndarray:
         # Create new CircuitSampler to avoid breaking existing one's caches.
         sampler = CircuitSampler(self.quantum_instance)
 
-        aux_op_meas = expectation.convert(StateFn(ListOp(aux_operators), is_measurement=True))
+        if isinstance(aux_operators, dict):
+            list_op = ListOp(list(aux_operators.values()))
+        else:
+            list_op = ListOp(aux_operators)
+
+        aux_op_meas = expectation.convert(StateFn(list_op, is_measurement=True))
         aux_op_expect = aux_op_meas.compose(CircuitStateFn(self.ansatz.bind_parameters(parameters)))
         values = np.real(sampler.convert(aux_op_expect).eval())
 
         # Discard values below threshold
         aux_op_results = values * (np.abs(values) > threshold)
-        # Deal with the aux_op behavior where there can be Nones or Zero qubit Paulis in the list
-        _aux_op_nones = [op is None for op in aux_operators]
-        aux_operator_eigenvalues = [
-            None if is_none else [result]
-            for (is_none, result) in zip(_aux_op_nones, aux_op_results)
-        ]
+
+        # Return None eigenvalues for None operators if aux_operators is a list.
+        # If aux_operators is a dict, the None operators should have been dropped in compute_minimum_eigenvalue
+        if isinstance(aux_operators, list):
+            aux_operator_eigenvalues = [None] * len(aux_operators)
+            key_value_iterator = enumerate(aux_op_results)
+        else:
+            aux_operator_eigenvalues = {}
+            key_value_iterator = zip(aux_operators.keys(), aux_op_results)
+
+        for key, value in key_value_iterator:
+            if aux_operators[key] is not None:
+                aux_operator_eigenvalues[key] = value
+        # # Deal with the aux_op behavior where there can be Nones or Zero qubit Paulis in the list
+        # aux_operator_eigenvalues = {
+        #     key: None if aux_operators[key] is None else result
+        #     for key, result in aux_op_results.items()
+        # }
         # As this has mixed types, since it can included None, it needs to explicitly pass object
         # data type to avoid numpy 1.19 warning message about implicit conversion being deprecated
         aux_operator_eigenvalues = np.array([aux_operator_eigenvalues], dtype=object)
         return aux_operator_eigenvalues
 
     def compute_minimum_eigenvalue(
-        self, operator: OperatorBase, aux_operators: Optional[List[Optional[OperatorBase]]] = None
+        self, operator: OperatorBase, aux_operators: Optional[ListOrDict[OperatorBase]] = None
     ) -> MinimumEigensolverResult:
         super().compute_minimum_eigenvalue(operator, aux_operators)
 
@@ -454,19 +471,23 @@ def compute_minimum_eigenvalue(
         initial_point = _validate_initial_point(self.initial_point, self.ansatz)
 
         bounds = _validate_bounds(self.ansatz)
-
         # We need to handle the array entries being Optional i.e. having value None
         if aux_operators:
             zero_op = I.tensorpower(operator.num_qubits) * 0.0
-            converted = []
-            for op in aux_operators:
-                if op is None:
-                    converted.append(zero_op)
-                else:
-                    converted.append(op)
 
-            # For some reason Chemistry passes aux_ops with 0 qubits and paulis sometimes.
-            aux_operators = [zero_op if op == 0 else op for op in converted]
+            # Convert the None operators when aux_operators is a list. Drop them is aux_operators is a dict.
+            if isinstance(aux_operators, list):
+                key_op_iterator = enumerate(aux_operators)
+                converted = [zero_op] * len(aux_operators)
+            else:
+                key_op_iterator = aux_operators.items()
+                converted = {}
+            for key, op in key_op_iterator:
+                if op is not None:
+                    converted[key] = op
+
+            aux_operators = converted
+
         else:
             aux_operators = None
 
@@ -517,7 +538,7 @@ def compute_minimum_eigenvalue(
 
         if aux_operators is not None:
             aux_values = self._eval_aux_ops(opt_params, aux_operators, expectation=expectation)
-            result.aux_operator_eigenvalues = aux_values[0]
+            result.aux_operator_eigenvalues = aux_values
 
         return result