From a2e4ac1c0d033951620ec8711674a69a759e415b Mon Sep 17 00:00:00 2001 From: stan Date: Tue, 22 Sep 2026 14:21:55 +0800 Subject: [PATCH] fix(Grover): retain sampled solutions equal to the incumbent --- .../pyqpanda_alg/Grover/Grover_core.py | 8 +- test/QAlgBase/Test_grover_run.py | 174 ++++++++++-------- 2 files changed, 105 insertions(+), 77 deletions(-) diff --git a/pyqpanda-algorithm/pyqpanda_alg/Grover/Grover_core.py b/pyqpanda-algorithm/pyqpanda_alg/Grover/Grover_core.py index 0d2c2b3c..0d9ae6db 100644 --- a/pyqpanda-algorithm/pyqpanda_alg/Grover/Grover_core.py +++ b/pyqpanda-algorithm/pyqpanda_alg/Grover/Grover_core.py @@ -498,7 +498,11 @@ def run(self, continue_times: int = 3, n_value_function=None, value_function=Non Returns minimum_indexes, minimum_res : ( ``list[list[int]]``, ``float``)\n - The optimization result including the solution array and the optimal value. + Measured solutions attaining the incumbent value, in variable order + (least significant bit first), without duplicates. Equal-value samples + are retained even when the initial value is already optimal. The list + is empty if no sampled state attains or improves the initial value. + This finite search does not certify global optimality or enumerate all ties. Examples An example for minimization of quadratic binary function: x0 * x1 + x0 - x1. @@ -565,7 +569,7 @@ def run(self, continue_times: int = 3, n_value_function=None, value_function=Non if outcome not in indexes_measured: indexes_measured.append(outcome) if v == 0: - if outcome not in indexes_measured: + if outcome not in minimum_indexes: minimum_indexes.append(outcome) if process_show: print('minimum Key Again: ', outcome) diff --git a/test/QAlgBase/Test_grover_run.py b/test/QAlgBase/Test_grover_run.py index f2e94e70..b5808eaf 100644 --- a/test/QAlgBase/Test_grover_run.py +++ b/test/QAlgBase/Test_grover_run.py @@ -1,75 +1,99 @@ -# import pytest -# import numpy as np -# from pyqpanda_alg.Grover import GroverAdaptiveSearch -# from pyqpanda3.core import QCircuit, U1 -# from pyqpanda_alg.plugin import hadamard_circuit, QFT -# -# -# class Test_grover_run: -# -# def flip_oracle_function(self, q_index_value, current_min): -# q_index = q_index_value[:2] -# q_value = q_index_value[2:] -# n_value = len(q_value) -# factor = np.pi * 2 ** (1 - n_value) -# cal_cir = QCircuit() -# cal_cir << hadamard_circuit(q_value) -# for i, q_i in enumerate(q_value): -# cal_cir << U1(q_i, factor * 2 ** i).control(q_index) -# cal_cir << U1(q_i, factor * 2 ** i).control(q_index[0]) -# cal_cir << U1(q_i, -factor * 2 ** i).control(q_index[1]) -# cal_cir << U1(q_i, factor * 2 ** i * (-current_min)) -# cal_cir << QFT(q_value).dagger() -# return cal_cir -# -# def n_value_function_basic(self, current_min): -# n_value = 2 if current_min == 0 else 3 -# return n_value -# -# def value_function_basic(self, var_array): -# var_array = list(map(int, var_array))[::-1] # 转换为二进制并反转顺序 -# x0, x1 = var_array[0], var_array[1] -# value = x0 * x1 + x0 - x1 -# return value -# -# def calculate_optimal_solution_brute_force(self, value_function, n_bits=2): -# best_value = float('inf') -# best_solution = None -# -# # 遍历所有可能的二进制组合 -# for i in range(2 ** n_bits): -# binary_str = format(i, f'0{n_bits}b') -# current_value = value_function(binary_str) -# -# if current_value < best_value: -# best_value = current_value -# best_solution = binary_str -# -# return best_solution, best_value -# -# def test_run_basic_parameters(self): -# """测试run接口基本参数""" -# demo_search = GroverAdaptiveSearch( -# init_value=0, -# n_index=2, -# oracle_circuit=self.flip_oracle_function -# ) -# -# theoretical_opt_solution, theoretical_opt_value = self.calculate_optimal_solution_brute_force( -# self.value_function_basic -# ) -# -# res = demo_search.run( -# continue_times=5, -# n_value_function=self.n_value_function_basic, -# value_function=self.value_function_basic, -# process_show=True -# ) -# -# optimal_solution, optimal_value = res[0], res[1] -# print(f"✓ run接口返回有效结果: 最优解='{optimal_solution}', 函数值={optimal_value}") -# -# -# if __name__ == "__main__": -# # 运行测试 -# pytest.main([__file__, "-v", "-s"]) \ No newline at end of file +"""Deterministic CPU regressions for adaptive search result retention.""" + +import numpy as np +import pytest +from pyqpanda3.core import QCircuit, X +from pyqpanda_alg.Grover import GroverAdaptiveSearch, mark_data_reflection + + +@pytest.mark.parametrize("rotation_change", ["increase", "random"]) +@pytest.mark.parametrize( + "values, initial, samples, expected_keys, expected_value", + [ + ([2, 3, 4, 5], 2, [0], {0}, 2), + ([5, 2, 2, 9], 2, [1, 2], {1, 2}, 2), + ([2, 2, 3, 4], 9, [0, 1], {0, 1}, 2), + ([5, 5, 2, 2], 5, [0, 1, 2, 3], {2, 3}, 2), + ([-3, -3, 1, 2], -3, [0, 1], {0, 1}, -3), + ([1, 4, 5, 6], 1, [1], set(), 1), + ([2, 4, 5, 6], 9, [0, 1], {0}, 2), + ([0, 0, 0, 0], 0, [0, 1, 2, 3], {0, 1, 2, 3}, 0), + ], + ids=[ + "initial-already-optimal", + "equal-initial-minima", + "improve-then-tie", + "discard-old-minima-after-improvement", + "negative-initial-minima", + "no-sampled-minimum", + "worse-samples-excluded", + "constant-objective", + ], +) +def test_search_retains_measured_minima( + rotation_change: str, + values: list[int], + initial: int, + samples: list[int], + expected_keys: set[int], + expected_value: int, +) -> None: + """Only measured states attaining the incumbent belong in the result. + + Prepare a chosen computational basis state using the public init callback. + A diagonal threshold oracle and Grover reflection preserve that basis state + up to phase. Thus CPUQVM's real measurements are deterministic; no simulator + or result is mocked. Advance preparation only after the value callback sees + a measurement, so forward and inverse preparations within an iteration agree. + """ + observed: list[int] = [] + + def prepare(qubits: list[int], threshold: float) -> QCircuit: + key = samples[min(len(observed), len(samples) - 1)] + circuit = QCircuit() + for bit in range(2): + if key & (1 << bit): + circuit << X(qubits[bit]) + return circuit + + def oracle(qubits: list[int], threshold: float) -> QCircuit: + marked = [ + format(key, "02b") for key, value in enumerate(values) if value < threshold + ] + if not marked: + return QCircuit() + return mark_data_reflection(qubits[:2], marked) + + def measured_value(bits: str) -> int: + key = int(bits, 2) + observed.append(key) + return values[key] + + search = GroverAdaptiveSearch( + init_value=initial, + n_index=2, + init_circuit=prepare, + oracle_circuit=oracle, + ) + random_state = np.random.get_state() + try: + np.random.seed(7) + solutions, value = search.run( + continue_times=4, + n_value_function=lambda threshold: 1, + value_function=measured_value, + rotation_change=rotation_change, + ) + finally: + np.random.set_state(random_state) + + assert observed[: len(samples)] == samples + assert value == expected_value + # Public solutions are in variable order (least significant bit first). + keys = [sum(bit << index for index, bit in enumerate(bits)) for bits in solutions] + assert set(keys) == expected_keys + assert len(keys) == len(set(keys)), "Repeated samples must not duplicate solutions" + assert all( + len(bits) == 2 and all(bit in (0, 1) for bit in bits) for bits in solutions + ) + assert all(key in observed and values[key] == value for key in keys)