diff --git a/CHANGELOG.md b/CHANGELOG.md index bddaf7bea..3886aaa4f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,10 @@ releases may include breaking changes. ### Added +- ✨ Expand and compact the RL observation with normalized OpenQASM operation + frequencies and one-element `float32` qubit-count and depth arrays, and + include measurements in the shared ML feature schema ([#758]) + ([**@flowerthrower**]) - 👷 Enable testing on Python 3.14 ([#488]) ([**@denialhaag**]) - ✨ Add selectable `v2` and `v3` RL MDP strategies, make `v3` the default, and use the selected strategy in compilation traces and model artifact names @@ -89,6 +93,7 @@ for previous changelogs._ [#773]: https://github.com/munich-quantum-toolkit/predictor/pull/771 [#769]: https://github.com/munich-quantum-toolkit/predictor/pull/769 +[#758]: https://github.com/munich-quantum-toolkit/predictor/pull/758 [#755]: https://github.com/munich-quantum-toolkit/predictor/pull/755 [#731]: https://github.com/munich-quantum-toolkit/predictor/pull/731 [#714]: https://github.com/munich-quantum-toolkit/predictor/pull/714 diff --git a/UPGRADING.md b/UPGRADING.md index 4540e6dbd..fd0caf47f 100644 --- a/UPGRADING.md +++ b/UPGRADING.md @@ -6,6 +6,16 @@ of changes including minor and patch releases, please refer to the ## [Unreleased] +### RL observation features + +The RL observation now includes normalized frequencies for supported OpenQASM +gates and measurements. The `num_qubits` and `depth` entries are now one-element +`float32` arrays in `[0, 1]` instead of discrete integers. The qubit count is +linearly scaled by the target device's qubit count; the depth is `log1p`-scaled +and capped at 999,999. Existing RL models must be retrained, and code that +consumes `PredictorEnv` observations directly must handle the expanded schema +and array values. + ### End of support for Python 3.10 Starting with this release, MQT Predictor no longer supports Python 3.10. As a diff --git a/src/mqt/predictor/ml/helper.py b/src/mqt/predictor/ml/helper.py index d73a7d85c..4d87bcaea 100644 --- a/src/mqt/predictor/ml/helper.py +++ b/src/mqt/predictor/ml/helper.py @@ -15,7 +15,7 @@ from pathlib import Path from typing import TYPE_CHECKING -from mqt.predictor.utils import calc_supermarq_features +from mqt.predictor.utils import calc_supermarq_features, get_openqasm_gates if TYPE_CHECKING: import numpy as np @@ -50,55 +50,6 @@ def get_path_training_circuits_compiled() -> Path: return get_path_training_data() / "training_circuits_compiled" -def get_openqasm_gates() -> list[str]: - """Returns a list of all quantum gates within the openQASM 2.0 standard header.""" - # according to https://github.com/Qiskit/qiskit-terra/blob/main/qiskit/qasm/libs/qelib1.inc - return [ - "u3", - "u2", - "u1", - "cx", - "id", - "u0", - "u", - "p", - "x", - "y", - "z", - "h", - "s", - "sdg", - "t", - "tdg", - "rx", - "ry", - "rz", - "sx", - "sxdg", - "cz", - "cy", - "swap", - "ch", - "ccx", - "cswap", - "crx", - "cry", - "crz", - "cu1", - "cp", - "cu3", - "csx", - "cu", - "rxx", - "rzz", - "rccx", - "rc3x", - "c3x", - "c3sqrtx", - "c4x", - ] - - def dict_to_featurevector(gate_dict: dict[str, int]) -> dict[str, int]: """Calculates and returns the feature vector of a given quantum circuit gate dictionary.""" res_dct = dict.fromkeys(get_openqasm_gates(), 0) diff --git a/src/mqt/predictor/rl/helper.py b/src/mqt/predictor/rl/helper.py index f2d8c01e8..25812b453 100644 --- a/src/mqt/predictor/rl/helper.py +++ b/src/mqt/predictor/rl/helper.py @@ -11,13 +11,14 @@ from __future__ import annotations import logging +from math import log1p from pathlib import Path from typing import TYPE_CHECKING import numpy as np from qiskit import QuantumCircuit -from mqt.predictor.utils import calc_supermarq_features +from mqt.predictor.utils import calc_supermarq_features, get_openqasm_gates if TYPE_CHECKING: from numpy.random import Generator @@ -28,6 +29,9 @@ logger = logging.getLogger("mqt-predictor") +MAX_CIRCUIT_DEPTH = 999_999 +OBSERVATION_OPERATIONS = tuple(get_openqasm_gates()) + def get_state_sample(max_qubits: int, path_training_circuits: Path, rng: Generator) -> tuple[QuantumCircuit, str]: """Returns a random quantum circuit from the training circuits folder. @@ -70,22 +74,25 @@ def get_state_sample(max_qubits: int, path_training_circuits: Path, rng: Generat return qc, str(file_list[random_index]) -def create_feature_dict(qc: QuantumCircuit) -> dict[str, int | NDArray[np.float32]]: - """Creates a feature dictionary for a given quantum circuit. - - Arguments: - qc: The quantum circuit for which the feature dictionary is created. - - Returns: - The feature dictionary for the given quantum circuit. - """ - feature_dict: dict[str, int | NDArray[np.float32]] = { - "num_qubits": qc.num_qubits, - "depth": qc.depth(), +def create_feature_dict(qc: QuantumCircuit, max_num_qubits: int) -> dict[str, NDArray[np.float32]]: + """Create a normalized feature dictionary for a quantum circuit.""" + operation_counts = dict(qc.count_ops()) + total_operations = sum(value for gate, value in operation_counts.items() if gate != "barrier") + normalized_num_qubits = min(qc.num_qubits, max_num_qubits) / max_num_qubits + normalized_depth = log1p(min(qc.depth(), MAX_CIRCUIT_DEPTH)) / log1p(MAX_CIRCUIT_DEPTH) + feature_dict = { + **{ + operation: np.array( + [operation_counts.get(operation, 0) / total_operations if total_operations else 0.0], + dtype=np.float32, + ) + for operation in OBSERVATION_OPERATIONS + }, + "num_qubits": np.array([normalized_num_qubits], dtype=np.float32), + "depth": np.array([normalized_depth], dtype=np.float32), } supermarq_features = calc_supermarq_features(qc) - # for all dict values, put them in a list each feature_dict["program_communication"] = np.array([supermarq_features.program_communication], dtype=np.float32) feature_dict["critical_depth"] = np.array([supermarq_features.critical_depth], dtype=np.float32) feature_dict["entanglement_ratio"] = np.array([supermarq_features.entanglement_ratio], dtype=np.float32) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 0e5958e7e..c433ccaa7 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -46,7 +46,12 @@ from mqt.predictor.rl.actions.bqskit_actions import is_bqskit_action_available, run_bqskit_action from mqt.predictor.rl.actions.qiskit_actions import is_qiskit_action_available, run_qiskit_action from mqt.predictor.rl.actions.tket_actions import is_tket_action_available, run_tket_action -from mqt.predictor.rl.helper import create_feature_dict, get_path_training_circuits, get_state_sample +from mqt.predictor.rl.helper import ( + OBSERVATION_OPERATIONS, + create_feature_dict, + get_path_training_circuits, + get_state_sample, +) from mqt.predictor.rl.tracer import CompilationTracer, FigureOfMeritMetric, FigureOfMeritMetrics logger = logging.getLogger("mqt-predictor") @@ -174,9 +179,13 @@ def __init__( self.has_parameterized_gates = False self.rng = np.random.default_rng(10) + operation_spaces = { + operation: Box(low=0, high=1, shape=(1,), dtype=np.float32) for operation in OBSERVATION_OPERATIONS + } spaces: dict[str, Space] = { - "num_qubits": Discrete(128), - "depth": Discrete(1000000), + "num_qubits": Box(low=0, high=1, shape=(1,), dtype=np.float32), + "depth": Box(low=0, high=1, shape=(1,), dtype=np.float32), + **operation_spaces, "program_communication": Box(low=0, high=1, shape=(1,), dtype=np.float32), "critical_depth": Box(low=0, high=1, shape=(1,), dtype=np.float32), "entanglement_ratio": Box(low=0, high=1, shape=(1,), dtype=np.float32), @@ -288,7 +297,7 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any action_duration = time.perf_counter() - start_time # Different passes may fail for various reasons (e.g., found no routing solution). self.error_occurred = True - obs = create_feature_dict(self.state) + obs = create_feature_dict(self.state, self.device.num_qubits) # Trace the error before aborting self._collect_tracer_data( @@ -328,7 +337,7 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any reward_val = 0 done = False - obs = create_feature_dict(self.state) + obs = create_feature_dict(self.state, self.device.num_qubits) # Trace+truncate if step limit is reached if not done and self.max_steps is not None and self.num_steps >= self.max_steps: @@ -438,7 +447,7 @@ def reset( self.num_qubits_uncompiled_circuit = self.state.num_qubits self.has_parameterized_gates = len(self.state.parameters) > 0 - obs = create_feature_dict(self.state) + obs = create_feature_dict(self.state, self.device.num_qubits) # Setup tracer for the new episode if self.tracer_output_path is not None: diff --git a/src/mqt/predictor/rl/tracer.py b/src/mqt/predictor/rl/tracer.py index a9c493491..868899fd6 100644 --- a/src/mqt/predictor/rl/tracer.py +++ b/src/mqt/predictor/rl/tracer.py @@ -209,7 +209,7 @@ def record_step( reward: float, current_qc: QuantumCircuit, figures_of_merit: FigureOfMeritMetrics, - features: dict[str, int | NDArray[np.float32]], + features: dict[str, NDArray[np.float32]], synthesized: bool, laid_out: bool, routed: bool, @@ -337,8 +337,6 @@ def _extract_device_metadata(device: Target) -> DeviceMetadata: ) @staticmethod - def _extract_float(val: int | NDArray[np.float32]) -> float: - """Safely extracts a float from a scalar or a 1D NumPy array to satisfy linter requirements.""" - if isinstance(val, int): - return float(val) + def _extract_float(val: NDArray[np.float32]) -> float: + """Extract a float from a one-element feature array.""" return float(val.item()) diff --git a/src/mqt/predictor/utils.py b/src/mqt/predictor/utils.py index f3433113a..7eebf6bb7 100644 --- a/src/mqt/predictor/utils.py +++ b/src/mqt/predictor/utils.py @@ -144,3 +144,53 @@ def calc_supermarq_features( parallelism, liveness, ) + + +def get_openqasm_gates() -> list[str]: + """Return the canonical operation names from the OpenQASM standard libraries.""" + return [ + "u3", + "u2", + "u1", + "cx", + "id", + "u0", + "u", + "p", + "x", + "y", + "z", + "h", + "s", + "sdg", + "t", + "tdg", + "rx", + "ry", + "rz", + "sx", + "sxdg", + "cz", + "cy", + "swap", + "ch", + "ccx", + "cswap", + "crx", + "cry", + "crz", + "cu1", + "cp", + "cu3", + "csx", + "cu", + "rxx", + "rzz", + "rccx", + "rc3x", + "c3x", + "c3sqrtx", + "c4x", + "measure", + "r", + ] diff --git a/tests/compilation/test_helper_rl.py b/tests/compilation/test_helper_rl.py index b99ad0fca..5acc1f75e 100644 --- a/tests/compilation/test_helper_rl.py +++ b/tests/compilation/test_helper_rl.py @@ -29,6 +29,7 @@ from mqt.predictor.rl.actions.bqskit_actions import bqskit_to_qiskit, get_bqskit_native_gates from mqt.predictor.rl.actions.qiskit_actions import postprocess_vf2postlayout from mqt.predictor.rl.helper import create_feature_dict, get_path_trained_model, get_path_training_circuits +from mqt.predictor.utils import get_openqasm_gates if TYPE_CHECKING: from collections.abc import Callable @@ -39,10 +40,31 @@ def test_create_feature_dict() -> None: """Test the creation of a feature dictionary.""" - qc = get_benchmark("dj", BenchmarkLevel.ALG, 5) - features = create_feature_dict(qc) + qc = get_benchmark("dj", BenchmarkLevel.ALG, 3) + device = get_device("ibm_eagle_127") + features = create_feature_dict(qc, device.num_qubits) + for feature in features.values(): - assert isinstance(feature, np.ndarray | int) + assert isinstance(feature, np.ndarray) + assert feature.dtype == np.float32 + + expected_features = dict.fromkeys(get_openqasm_gates(), 0.0) + expected_features.update({ + "x": 1 / 9, + "h": 5 / 9, + "measure": 2 / 9, + "num_qubits": 3 / device.num_qubits, + "depth": np.log1p(5) / np.log1p(999_999), + "program_communication": 0.0, + "critical_depth": 0.0, + "entanglement_ratio": 0.0, + "parallelism": 1 / 5, + "liveness": 11 / 15, + }) + + assert features.keys() == expected_features.keys() + for name, expected in expected_features.items(): + np.testing.assert_allclose(features[name], [expected]) def test_get_path_trained_model() -> None: diff --git a/tests/compilation/test_predictor_rl.py b/tests/compilation/test_predictor_rl.py index 48b3648d0..c057431d6 100644 --- a/tests/compilation/test_predictor_rl.py +++ b/tests/compilation/test_predictor_rl.py @@ -48,7 +48,10 @@ def test_predictor_env_reset_from_string() -> None: qc = get_benchmark("dj", BenchmarkLevel.ALG, 3) with qasm_path.open("w", encoding="utf-8") as f: dump(qc, f) - assert predictor.env.reset(qc=qasm_path)[0] == create_feature_dict(qc) + observation, _ = predictor.env.reset(qc=qasm_path) + + assert observation == create_feature_dict(qc, device.num_qubits) + assert observation["num_qubits"][0] == pytest.approx(qc.num_qubits / device.num_qubits) def test_predictor_env_esp_error() -> None: diff --git a/tests/compilation/test_tracer.py b/tests/compilation/test_tracer.py index 6911a8657..772eb1b8e 100644 --- a/tests/compilation/test_tracer.py +++ b/tests/compilation/test_tracer.py @@ -77,8 +77,8 @@ def test_compilation_tracer_generates_valid_json(tmp_path: Path) -> None: qc_terminal.rz(0.5, 2) # 2. Mock the feature dictionaries that the RL environment normally passes - features_baseline = create_feature_dict(qc_baseline) - features_terminal = create_feature_dict(qc_terminal) + features_baseline = create_feature_dict(qc_baseline, device.num_qubits) + features_terminal = create_feature_dict(qc_terminal, device.num_qubits) # 3. Use the actual `record_step` API instead of manually building CompilationSteps tracer.record_step( diff --git a/tests/device_selection/test_helper_ml.py b/tests/device_selection/test_helper_ml.py index 87f69f009..7df00be2b 100644 --- a/tests/device_selection/test_helper_ml.py +++ b/tests/device_selection/test_helper_ml.py @@ -10,6 +10,7 @@ from __future__ import annotations +import numpy as np from mqt.bench import BenchmarkLevel, get_benchmark from mqt.predictor.ml.helper import ( @@ -25,7 +26,12 @@ def test_create_feature_vector() -> None: """Test the creation of a feature dictionary.""" qc = get_benchmark("dj", BenchmarkLevel.ALG, 3) feature_vector = create_feature_vector(qc) - assert feature_vector is not None + + expected_operations = dict.fromkeys(get_openqasm_gates(), 0.0) + expected_operations.update({"x": 1.0, "h": 5.0, "measure": 2.0}) + expected_features = [*expected_operations.values(), 3.0, 5.0, 0.0, 0.0, 0.0, 1 / 5, 11 / 15] + + np.testing.assert_allclose(feature_vector, expected_features) def test_get_openqasm_gates() -> None: