🎨 Enhance RL feature vector

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

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

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)

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)

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:

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())

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",
+    ]

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:

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:

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(

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: