Fix the aggregation argument of SamplingVQE

qiskit/algorithms/minimum_eigensolvers/diagonal_estimator.py

@@ -171,7 +171,7 @@ def aggregate(measurements):
             accumulated_percent = 0  # once alpha is reached, stop
             cvar = 0
             for probability, value in sorted_measurements:
-                cvar += value * max(probability, alpha - accumulated_percent)
+                cvar += value * min(probability, alpha - accumulated_percent)
                 accumulated_percent += probability
                 if accumulated_percent >= alpha:
                     break

qiskit/algorithms/minimum_eigensolvers/sampling_vqe.py

@@ -98,10 +98,12 @@ def my_minimizer(fun, x0, jac=None, bounds=None) -> OptimizerResult:
         optimizer (Optimizer | Minimizer): A classical optimizer to find the minimum energy. This
             can either be a Qiskit :class:`.Optimizer` or a callable implementing the
             :class:`.Minimizer` protocol.
-        aggregation (float | Callable[[list[float]], float] | None): A float or callable to specify
-            how the objective function evaluated on the basis states should be aggregated. If a
-            float, this specifies the :math:`\alpha \in [0,1]` parameter for a CVaR expectation
-            value [1].
+        aggregation (float | Callable[[list[tuple[float, complex]], float] | None):
+            A float or callable to specify how the objective function evaluated on the basis states
+            should be aggregated. If a float, this specifies the :math:`\alpha \in [0,1]` parameter
+            for a CVaR expectation value [1]. If a callable, it takes a list of basis state
+            measurements specified as  ``[(probability, objective_value)]`` and return an objective
+            value as float. If None, all an ordinary expectation value is calculated.
         callback (Callable[[int, np.ndarray, float, dict[str, Any]], None] | None): A callback that
             can access the intermediate data at each optimization step. These data are: the
             evaluation count, the optimizer parameters for the ansatz, the evaluated value, and the
@@ -292,7 +294,9 @@ def store_best_measurement(best):
                 ):
                     best_measurement["best"] = best_i
 
-        estimator = _DiagonalEstimator(sampler=self.sampler, callback=store_best_measurement)
+        estimator = _DiagonalEstimator(
+            sampler=self.sampler, callback=store_best_measurement, aggregation=self.aggregation
+        )
 
         def evaluate_energy(parameters):
             nonlocal eval_count

releasenotes/notes/fix-sampling-vqe-aggregation-107e3983147c57bc.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    Fixed a bug in :class:`.SamplingVQE` where the ``aggregation`` did not have an effect.
+    Now the aggregation function and CVaR expectation value can correctly be specified.

test/python/algorithms/minimum_eigensolvers/test_sampling_vqe.py

@@ -256,6 +256,29 @@ def store_intermediate_result(eval_count, parameters, mean, metadata):
         for params in history["parameters"]:
             self.assertTrue(all(isinstance(param, float) for param in params))
 
+    def test_aggregation(self):
+        """Test the aggregation works."""
+
+        # test a custom aggregration that just uses the best measurement
+        def best_measurement(measurements):
+            res = min(measurements, key=lambda meas: meas[1])[1]
+            return res
+
+        # test CVaR with alpha of 0.4 (i.e. 40% of the best measurements)
+        alpha = 0.4
+
+        ansatz = RealAmplitudes(1, reps=0)
+        ansatz.h(0)
+
+        for aggregation in [alpha, best_measurement]:
+            with self.subTest(aggregation=aggregation):
+                vqe = SamplingVQE(Sampler(), ansatz, _mock_optimizer, aggregation=best_measurement)
+                result = vqe.compute_minimum_eigenvalue(Pauli("Z"))
+
+                # evaluation at x0=0 samples -1 and 1 with 50% probability, and our aggregation
+                # takes the smallest value
+                self.assertAlmostEqual(result.optimal_value, -1)
+
 
 if __name__ == "__main__":
     unittest.main()