From 8a86438d1f7b55fa1b3200a2218c8f152c620b5f Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Thu, 22 Jan 2026 10:17:17 +0100 Subject: [PATCH 01/22] =?UTF-8?q?=F0=9F=90=9B=20gracefully=20handle=20VF2L?= =?UTF-8?q?ayout=20no-solution-found?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 5 +++-- tests/compilation/test_integration_further_SDKs.py | 2 +- 2 files changed, 4 insertions(+), 3 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index eb509da66..231d16711 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -353,7 +353,8 @@ def _handle_qiskit_layout_postprocessing( if post_layout: altered_qc, _ = postprocess_vf2postlayout(altered_qc, post_layout, self.layout) elif action.name == "VF2Layout": - assert pm.property_set["VF2Layout_stop_reason"] == VF2LayoutStopReason.SOLUTION_FOUND + if pm.property_set["VF2Layout_stop_reason"] != VF2LayoutStopReason.SOLUTION_FOUND: + return self.state assert pm.property_set["layout"] else: assert pm.property_set["layout"] @@ -379,7 +380,7 @@ def _apply_tket_action(self, action: Action, action_index: int) -> QuantumCircui qbs = tket_qc.qubits tket_qc.rename_units({qbs[i]: Qubit("q", i) for i in range(len(qbs))}) - altered_qc = tk_to_qiskit(tket_qc) + altered_qc = tk_to_qiskit(tket_qc, perm_warning=False) if action_index in self.actions_routing_indices: assert self.layout is not None diff --git a/tests/compilation/test_integration_further_SDKs.py b/tests/compilation/test_integration_further_SDKs.py index 4cfa6d966..ce27c8dfd 100644 --- a/tests/compilation/test_integration_further_SDKs.py +++ b/tests/compilation/test_integration_further_SDKs.py @@ -218,7 +218,7 @@ def test_tket_routing(available_actions_dict: dict[str, Action]) -> None: qubit_map = {qbs[i]: Qubit("q", i) for i in range(len(qbs))} tket_qc.rename_units(qubit_map) # type: ignore[arg-type] - mapped_qc = tk_to_qiskit(tket_qc) + mapped_qc = tk_to_qiskit(tket_qc, perm_warning=False) final_layout = final_layout_pytket_to_qiskit(tket_qc, mapped_qc) From f34a9ae70dd6bcc8528670fed7941edca1239b1f Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Thu, 22 Jan 2026 10:21:07 +0100 Subject: [PATCH 02/22] disable progress bar --- src/mqt/predictor/rl/predictor.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/mqt/predictor/rl/predictor.py b/src/mqt/predictor/rl/predictor.py index bfe3aeb90..0329d26c5 100644 --- a/src/mqt/predictor/rl/predictor.py +++ b/src/mqt/predictor/rl/predictor.py @@ -104,11 +104,12 @@ def train_model( batch_size = 10 progress_bar = False else: + set_random_seed(0) # default PPO values n_steps = 2048 n_epochs = 10 batch_size = 64 - progress_bar = True + progress_bar = False logger.debug("Start training for: " + self.figure_of_merit + " on " + self.device_name) model = MaskablePPO( From 7a893757f586fc3e300b0bed9384cc17ca07c60b Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Thu, 22 Jan 2026 10:52:50 +0100 Subject: [PATCH 03/22] =?UTF-8?q?=F0=9F=90=9B=20only=20consider=20valid=20?= =?UTF-8?q?circuits=20for=20terminal=20action?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 13 ++++++++++++- 1 file changed, 12 insertions(+), 1 deletion(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 231d16711..76db980a7 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -211,6 +211,15 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any obs = create_feature_dict(self.state) return obs, reward_val, done, False, {} + def _is_valid_circuit(self) -> bool: + for instruction in self.state.data: + if instruction.operation.name == "barrier": + continue + qubit_indices = tuple(self.state.find_bit(q).index for q in instruction.qubits) + if not self.device.instruction_supported(operation_name=instruction.operation.name, qargs=qubit_indices): + return False + return True + def calculate_reward(self) -> float: """Calculates and returns the reward for the current state.""" if self.reward_function == "expected_fidelity": @@ -441,7 +450,9 @@ def determine_valid_actions_for_state(self) -> list[int]: mapped = check_mapping.property_set["is_swap_mapped"] if mapped and self.layout is not None: # The circuit is correctly mapped. - return [self.action_terminate_index, *self.actions_opt_indices] + if self._is_valid_circuit(): # The circuit respects native gates (icl. directionality). + return [self.action_terminate_index, *self.actions_opt_indices] + return self.actions_opt_indices if self.layout is not None: # The circuit is not yet mapped but a layout is set. return self.actions_routing_indices From af50584a99c3d872cc7d6ed544d61f8e99a9891d Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Thu, 22 Jan 2026 14:26:10 +0100 Subject: [PATCH 04/22] =?UTF-8?q?=F0=9F=9A=A7=20debug=20predictor?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 241 ++++++++++++++++++++------- 1 file changed, 179 insertions(+), 62 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 76db980a7..e977cb0b6 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -24,6 +24,7 @@ from pathlib import Path from bqskit import Circuit + from pytket.circuit import Node from qiskit.transpiler import Target from mqt.predictor.reward import figure_of_merit @@ -42,8 +43,15 @@ from pytket.extensions.qiskit import qiskit_to_tk, tk_to_qiskit from qiskit import QuantumCircuit from qiskit.passmanager.flow_controllers import DoWhileController -from qiskit.transpiler import CouplingMap, PassManager, TranspileLayout -from qiskit.transpiler.passes import CheckMap, GatesInBasis +from qiskit.transpiler import CouplingMap, Layout, PassManager, TranspileLayout +from qiskit.transpiler.passes import ( + ApplyLayout, + CheckGateDirection, + EnlargeWithAncilla, + FullAncillaAllocation, + GatesInBasis, + SetLayout, +) from qiskit.transpiler.passes.layout.vf2_layout import VF2LayoutStopReason from mqt.predictor.hellinger import get_hellinger_model_path @@ -54,8 +62,17 @@ estimated_success_probability, expected_fidelity, ) -from mqt.predictor.rl.actions import CompilationOrigin, DeviceDependentAction, PassType, get_actions_by_pass_type -from mqt.predictor.rl.helper import create_feature_dict, get_path_training_circuits, get_state_sample +from mqt.predictor.rl.actions import ( + CompilationOrigin, + DeviceDependentAction, + PassType, + get_actions_by_pass_type, +) +from mqt.predictor.rl.helper import ( + create_feature_dict, + get_path_training_circuits, + get_state_sample, +) from mqt.predictor.rl.parsing import ( final_layout_bqskit_to_qiskit, final_layout_pytket_to_qiskit, @@ -95,6 +112,7 @@ def __init__( self.actions_mapping_indices = [] self.actions_opt_indices = [] self.actions_final_optimization_indices = [] + self.actions_structure_preserving_indices = [] # Actions that preserves the mapping and native gates self.used_actions: list[str] = [] self.device = device @@ -111,6 +129,12 @@ def __init__( self.action_set[index] = elem self.actions_synthesis_indices.append(index) index += 1 + for elem in action_dict[PassType.OPT]: + self.action_set[index] = elem + self.actions_opt_indices.append(index) + if getattr(elem, "preserve_layout", False): + self.actions_structure_preserving_indices.append(index) + index += 1 for elem in action_dict[PassType.LAYOUT]: self.action_set[index] = elem self.actions_layout_indices.append(index) @@ -119,10 +143,6 @@ def __init__( self.action_set[index] = elem self.actions_routing_indices.append(index) index += 1 - for elem in action_dict[PassType.OPT]: - self.action_set[index] = elem - self.actions_opt_indices.append(index) - index += 1 for elem in action_dict[PassType.MAPPING]: self.action_set[index] = elem self.actions_mapping_indices.append(index) @@ -164,6 +184,10 @@ def __init__( } self.observation_space = Dict(spaces) self.filename = "" + self.max_iter = 20 + self.node_err: dict[Node, float] | None = None + self.edge_err: dict[tuple[Node, Node], float] | None = None + self.readout_err: dict[Node, float] | None = None def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any, Any]]: """Executes the given action and returns the new state, the reward, whether the episode is done, whether the episode is truncated and additional information. @@ -203,34 +227,48 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any reward_val = 0 done = False - # in case the Qiskit.QuantumCircuit has unitary or u gates in it, decompose them (because otherwise qiskit will throw an error when applying the BasisTranslator + # in case the Qiskit.QuantumCircuit has unitary or u gates or clifford in it, decompose them (because otherwise qiskit will throw an error when applying the BasisTranslator) if self.state.count_ops().get("unitary"): self.state = self.state.decompose(gates_to_decompose="unitary") + elif self.state.count_ops().get("clifford"): + self.state = self.state.decompose(gates_to_decompose="clifford") self.state._layout = self.layout # noqa: SLF001 - obs = create_feature_dict(self.state) - return obs, reward_val, done, False, {} - def _is_valid_circuit(self) -> bool: + return create_feature_dict(self.state), reward_val, done, False, {} + + def _verify_native_mapped(self) -> tuple[bool, bool]: + """Check if the current circuit is fully valid for the device.""" + is_native, is_mapped = True, True for instruction in self.state.data: if instruction.operation.name == "barrier": continue qubit_indices = tuple(self.state.find_bit(q).index for q in instruction.qubits) - if not self.device.instruction_supported(operation_name=instruction.operation.name, qargs=qubit_indices): - return False - return True + # Is this operation generally in the device's basis gates? + if instruction.operation.name not in self.device.operation_names: + logger.debug(f"Instruction {instruction.operation.name} is not in the device's basis gates.") + is_native = False - def calculate_reward(self) -> float: - """Calculates and returns the reward for the current state.""" + # Is this operation supported on these specific qubits? + if not self.device.instruction_supported(operation_name=instruction.operation.name, qargs=qubit_indices): + logger.debug( + f"Instruction {instruction.operation.name} on qubits {qubit_indices} is not supported by the device." + ) + is_mapped = False + return is_native, is_mapped + + def calculate_reward(self, qc: QuantumCircuit | None = None) -> float: + """Calculates and returns the reward for either the current state or a quantum circuit (if provided).""" + circuit = self.state if qc is None else qc if self.reward_function == "expected_fidelity": - return expected_fidelity(self.state, self.device) + return expected_fidelity(circuit, self.device) if self.reward_function == "estimated_success_probability": - return estimated_success_probability(self.state, self.device) + return estimated_success_probability(circuit, self.device) if self.reward_function == "estimated_hellinger_distance": - return estimated_hellinger_distance(self.state, self.device, self.hellinger_model) + return estimated_hellinger_distance(circuit, self.device, self.hellinger_model) if self.reward_function == "critical_depth": - return crit_depth(self.state) - assert_never(self.state) + return crit_depth(circuit) + assert_never(circuit) def render(self) -> None: """Renders the current state.""" @@ -329,53 +367,71 @@ def apply_action(self, action_index: int) -> QuantumCircuit | None: raise ValueError(msg) def _apply_qiskit_action(self, action: Action, action_index: int) -> QuantumCircuit: - if action.name == "QiskitO3" and isinstance(action, DeviceDependentAction): + pm_property_set: dict[str, Any] | None = None + if action.name in ["QiskitO3", "Opt2qBlocks_preserve"] and isinstance(action, DeviceDependentAction): passes = action.transpile_pass( self.device.operation_names, CouplingMap(self.device.build_coupling_map()) if self.layout else None, ) - pm = PassManager([DoWhileController(passes, do_while=action.do_while)]) + if action.name == "QiskitO3": + pm = PassManager([DoWhileController(passes, do_while=action.do_while)]) + else: + pm = PassManager(passes) + altered_qc = pm.run(self.state) + pm_property_set = dict(pm.property_set) if hasattr(pm, "property_set") else None else: transpile_pass = ( action.transpile_pass(self.device) if callable(action.transpile_pass) else action.transpile_pass ) pm = PassManager(transpile_pass) - - altered_qc = pm.run(self.state) + altered_qc = pm.run(self.state) + pm_property_set = dict(pm.property_set) if hasattr(pm, "property_set") else None if action_index in ( self.actions_layout_indices + self.actions_mapping_indices + self.actions_final_optimization_indices ): - altered_qc = self._handle_qiskit_layout_postprocessing(action, pm, altered_qc) - - elif action_index in self.actions_routing_indices and self.layout: - self.layout.final_layout = pm.property_set["final_layout"] + altered_qc = self._handle_qiskit_layout_postprocessing(action, pm_property_set, altered_qc) + elif ( + action_index in self.actions_routing_indices + and self.layout is not None + and pm_property_set is not None + and "final_layout" in pm_property_set + ): + self.layout.final_layout = pm_property_set["final_layout"] return altered_qc def _handle_qiskit_layout_postprocessing( - self, action: Action, pm: PassManager, altered_qc: QuantumCircuit + self, + action: Action, + pm_property_set: dict[str, Any] | None, + altered_qc: QuantumCircuit, ) -> QuantumCircuit: + if not pm_property_set: + return altered_qc if action.name == "VF2PostLayout": - assert pm.property_set["VF2PostLayout_stop_reason"] is not None - post_layout = pm.property_set["post_layout"] + assert pm_property_set["VF2PostLayout_stop_reason"] is not None + post_layout = pm_property_set.get("post_layout") if post_layout: altered_qc, _ = postprocess_vf2postlayout(altered_qc, post_layout, self.layout) elif action.name == "VF2Layout": - if pm.property_set["VF2Layout_stop_reason"] != VF2LayoutStopReason.SOLUTION_FOUND: - return self.state - assert pm.property_set["layout"] + if pm_property_set["VF2Layout_stop_reason"] == VF2LayoutStopReason.SOLUTION_FOUND: + assert pm_property_set["layout"] else: - assert pm.property_set["layout"] + assert pm_property_set["layout"] - if pm.property_set["layout"]: + layout = pm_property_set.get("layout") + if layout: self.layout = TranspileLayout( - initial_layout=pm.property_set["layout"], - input_qubit_mapping=pm.property_set["original_qubit_indices"], - final_layout=pm.property_set["final_layout"], + initial_layout=layout, + input_qubit_mapping=pm_property_set.get("original_qubit_indices"), + final_layout=pm_property_set.get("final_layout"), _output_qubit_list=altered_qc.qubits, _input_qubit_count=self.num_qubits_uncompiled_circuit, ) + + if self.layout is not None and pm_property_set.get("final_layout"): + self.layout.final_layout = pm_property_set["final_layout"] return altered_qc def _apply_tket_action(self, action: Action, action_index: int) -> QuantumCircuit: @@ -384,12 +440,45 @@ def _apply_tket_action(self, action: Action, action_index: int) -> QuantumCircui action.transpile_pass(self.device) if callable(action.transpile_pass) else action.transpile_pass ) assert isinstance(transpile_pass, list) - for p in transpile_pass: - p.apply(tket_qc) + # Map TKET placement into a Qiskit layout + if action_index in self.actions_layout_indices: + try: + placement = transpile_pass[0].get_placement_map(tket_qc) + except Exception as e: + logger.warning("Placement failed (%s): %s. Falling back to original circuit.", action.name, e) + return tk_to_qiskit(tket_qc, replace_implicit_swaps=True) + else: + qc_tmp = tk_to_qiskit(tket_qc, replace_implicit_swaps=True) + + qiskit_mapping = { + qc_tmp.qubits[i]: placement[list(placement.keys())[i]].index[0] for i in range(len(placement)) + } + layout = Layout(qiskit_mapping) + + pm = PassManager([ + SetLayout(layout), + FullAncillaAllocation(coupling_map=CouplingMap(self.device.build_coupling_map())), + EnlargeWithAncilla(), + ApplyLayout(), + ]) + altered_qc = pm.run(qc_tmp) + + self.layout = TranspileLayout( + initial_layout=pm.property_set.get("layout"), + input_qubit_mapping=pm.property_set["original_qubit_indices"], + final_layout=pm.property_set["final_layout"], + _output_qubit_list=altered_qc.qubits, + _input_qubit_count=self.num_qubits_uncompiled_circuit, + ) + return altered_qc + + else: + for p in transpile_pass: + p.apply(tket_qc) qbs = tket_qc.qubits tket_qc.rename_units({qbs[i]: Qubit("q", i) for i in range(len(qbs))}) - altered_qc = tk_to_qiskit(tket_qc, perm_warning=False) + altered_qc = tk_to_qiskit(tket_qc, replace_implicit_swaps=True) if action_index in self.actions_routing_indices: assert self.layout is not None @@ -434,28 +523,56 @@ def _apply_bqskit_action(self, action: Action, action_index: int) -> QuantumCirc return bqskit_to_qiskit(bqskit_compiled_qc) def determine_valid_actions_for_state(self) -> list[int]: - """Determines and returns the valid actions for the current state.""" + """Determine the valid actions for the current circuit state.""" + # Check if circuit uses only native gates check_nat_gates = GatesInBasis(basis_gates=self.device.operation_names) check_nat_gates(self.state) only_nat_gates = check_nat_gates.property_set["all_gates_in_basis"] - if not only_nat_gates: - actions = self.actions_synthesis_indices + self.actions_opt_indices - if self.layout is not None: - actions += self.actions_routing_indices - return actions - - check_mapping = CheckMap(coupling_map=self.device.build_coupling_map()) - check_mapping(self.state) - mapped = check_mapping.property_set["is_swap_mapped"] + # Check if circuit is mapped to the device coupling map + # Note this does not validate directionality of the connectivity between qubits. + # If you need to check gates are implemented in a native direction for a target use the :class:`~.CheckGateDirection` pass instead. + # check_mapping = CheckMap(coupling_map=CouplingMap(self.device.build_coupling_map())) + # check_mapping(self.state) + # mapped = check_mapping.property_set["is_swap_mapped"] + check_direction = CheckGateDirection(coupling_map=CouplingMap(self.device.build_coupling_map())) + check_direction(self.state) + mapped = check_direction.property_set["all_gates_in_basis_and_direction"] + + verified_native, verified_mapped = self._verify_native_mapped() + + if verified_mapped != mapped: + logger.warning( + f"Discrepancy in mapping verification: CheckMap says {mapped}, _verify_native_mapped says {verified_mapped}." + ) + mapped = verified_mapped + if verified_native != only_nat_gates: + logger.warning( + f"Discrepancy in native gate verification: GatesInBasis says {only_nat_gates}, _verify_native_mapped says {verified_native}." + ) + only_nat_gates = verified_native - if mapped and self.layout is not None: # The circuit is correctly mapped. - if self._is_valid_circuit(): # The circuit respects native gates (icl. directionality). - return [self.action_terminate_index, *self.actions_opt_indices] - return self.actions_opt_indices + if not only_nat_gates: + if not mapped: + # Non-native + unmapped → synthesis or optimization + return self.actions_synthesis_indices + self.actions_opt_indices + # Non-native + mapped → synthesis or structure-preserving (native gates & mapping) + return self.actions_synthesis_indices + self.actions_structure_preserving_indices - if self.layout is not None: # The circuit is not yet mapped but a layout is set. + if mapped: + if self.layout is not None: + # Native + fully mapped & layout assigned → terminate or fine-tune + return [ + self.action_terminate_index, + *self.actions_structure_preserving_indices, + *self.actions_final_optimization_indices, + ] + # Mapped but no explicit layout yet → explore layout/mapping improvements + return ( + self.actions_structure_preserving_indices + self.actions_layout_indices + self.actions_mapping_indices + ) + if self.layout is not None: + # Layout chosen but not mapped → must do routing return self.actions_routing_indices - - # No layout applied yet - return self.actions_mapping_indices + self.actions_layout_indices + self.actions_opt_indices + # Not mapped and without layout → explore layout/mapping/optimizations + return self.actions_opt_indices + self.actions_layout_indices + self.actions_mapping_indices From 1e3d3d4eddfa7e1e42f1831cbf5d5ed18241a27b Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Thu, 22 Jan 2026 14:46:41 +0100 Subject: [PATCH 05/22] =?UTF-8?q?=F0=9F=90=9Bhandle=20layout=20fail=20more?= =?UTF-8?q?=20gracefully?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index e977cb0b6..70c0e8e3d 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -413,10 +413,19 @@ def _handle_qiskit_layout_postprocessing( assert pm_property_set["VF2PostLayout_stop_reason"] is not None post_layout = pm_property_set.get("post_layout") if post_layout: - altered_qc, _ = postprocess_vf2postlayout(altered_qc, post_layout, self.layout) + try: + altered_qc, _ = postprocess_vf2postlayout(altered_qc, post_layout, self.layout) + except Exception as e: + logger.warning( + "VF2PostLayout postprocessing failed. Contintinuing with previous circuit. Error: %s", e + ) + return self.state elif action.name == "VF2Layout": if pm_property_set["VF2Layout_stop_reason"] == VF2LayoutStopReason.SOLUTION_FOUND: assert pm_property_set["layout"] + else: + logger.warning("VF2Layout did not find a solution. Continuing with previous circuit.") + return self.state else: assert pm_property_set["layout"] From 25c8a4a6af80cc3aa1e5939934b2ea1f12453e46 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Thu, 22 Jan 2026 15:20:38 +0100 Subject: [PATCH 06/22] =?UTF-8?q?=F0=9F=9A=A7=20restructure=20state=20mach?= =?UTF-8?q?ine?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 130 ++++++++++++--------------- 1 file changed, 58 insertions(+), 72 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 70c0e8e3d..a372da375 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -46,7 +46,6 @@ from qiskit.transpiler import CouplingMap, Layout, PassManager, TranspileLayout from qiskit.transpiler.passes import ( ApplyLayout, - CheckGateDirection, EnlargeWithAncilla, FullAncillaAllocation, GatesInBasis, @@ -237,26 +236,6 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any return create_feature_dict(self.state), reward_val, done, False, {} - def _verify_native_mapped(self) -> tuple[bool, bool]: - """Check if the current circuit is fully valid for the device.""" - is_native, is_mapped = True, True - for instruction in self.state.data: - if instruction.operation.name == "barrier": - continue - qubit_indices = tuple(self.state.find_bit(q).index for q in instruction.qubits) - # Is this operation generally in the device's basis gates? - if instruction.operation.name not in self.device.operation_names: - logger.debug(f"Instruction {instruction.operation.name} is not in the device's basis gates.") - is_native = False - - # Is this operation supported on these specific qubits? - if not self.device.instruction_supported(operation_name=instruction.operation.name, qargs=qubit_indices): - logger.debug( - f"Instruction {instruction.operation.name} on qubits {qubit_indices} is not supported by the device." - ) - is_mapped = False - return is_native, is_mapped - def calculate_reward(self, qc: QuantumCircuit | None = None) -> float: """Calculates and returns the reward for either the current state or a quantum circuit (if provided).""" circuit = self.state if qc is None else qc @@ -531,57 +510,64 @@ def _apply_bqskit_action(self, action: Action, action_index: int) -> QuantumCirc return bqskit_to_qiskit(bqskit_compiled_qc) + def is_circuit_mapped(self, circuit: QuantumCircuit, coupling_map: CouplingMap) -> bool: + """Check if a circuit is fully routed/mapped to the device, including directionality. + + A circuit is considered mapped if all two-qubit gates are on qubits + allowed by the coupling map and follow the allowed direction. + + Args: + circuit: QuantumCircuit to check. + coupling_map: CouplingMap of the target device. + + Returns: + True if fully mapped and directed, False otherwise. + """ + # Create a set of directed edges for fast lookup + directed_edges = set(coupling_map.get_edges()) + + for instr in circuit.data: + qubits = [q._index for q in instr.qubits] + if len(qubits) == 2: + q0, q1 = qubits + # If this two-qubit gate is not allowed in the device coupling map, return False + if (q0, q1) not in directed_edges: + return False + return True + def determine_valid_actions_for_state(self) -> list[int]: - """Determine the valid actions for the current circuit state.""" - # Check if circuit uses only native gates + """Determine valid actions based on circuit state: synthesized, layouted, routed.""" check_nat_gates = GatesInBasis(basis_gates=self.device.operation_names) check_nat_gates(self.state) - only_nat_gates = check_nat_gates.property_set["all_gates_in_basis"] - - # Check if circuit is mapped to the device coupling map - # Note this does not validate directionality of the connectivity between qubits. - # If you need to check gates are implemented in a native direction for a target use the :class:`~.CheckGateDirection` pass instead. - # check_mapping = CheckMap(coupling_map=CouplingMap(self.device.build_coupling_map())) - # check_mapping(self.state) - # mapped = check_mapping.property_set["is_swap_mapped"] - check_direction = CheckGateDirection(coupling_map=CouplingMap(self.device.build_coupling_map())) - check_direction(self.state) - mapped = check_direction.property_set["all_gates_in_basis_and_direction"] - - verified_native, verified_mapped = self._verify_native_mapped() - - if verified_mapped != mapped: - logger.warning( - f"Discrepancy in mapping verification: CheckMap says {mapped}, _verify_native_mapped says {verified_mapped}." - ) - mapped = verified_mapped - if verified_native != only_nat_gates: - logger.warning( - f"Discrepancy in native gate verification: GatesInBasis says {only_nat_gates}, _verify_native_mapped says {verified_native}." - ) - only_nat_gates = verified_native - - if not only_nat_gates: - if not mapped: - # Non-native + unmapped → synthesis or optimization - return self.actions_synthesis_indices + self.actions_opt_indices - # Non-native + mapped → synthesis or structure-preserving (native gates & mapping) - return self.actions_synthesis_indices + self.actions_structure_preserving_indices - - if mapped: - if self.layout is not None: - # Native + fully mapped & layout assigned → terminate or fine-tune - return [ - self.action_terminate_index, - *self.actions_structure_preserving_indices, - *self.actions_final_optimization_indices, - ] - # Mapped but no explicit layout yet → explore layout/mapping improvements - return ( - self.actions_structure_preserving_indices + self.actions_layout_indices + self.actions_mapping_indices - ) - if self.layout is not None: - # Layout chosen but not mapped → must do routing - return self.actions_routing_indices - # Not mapped and without layout → explore layout/mapping/optimizations - return self.actions_opt_indices + self.actions_layout_indices + self.actions_mapping_indices + synthesized = check_nat_gates.property_set["all_gates_in_basis"] + layouted = self.layout is not None + routed = self.is_circuit_mapped(self.state, CouplingMap(self.device.build_coupling_map())) + + # Final state + if synthesized and layouted and routed: + return [ + self.action_terminate_index, + *self.actions_structure_preserving_indices, + *self.actions_final_optimization_indices, + ] + + actions = [] + + # Synthesis is possible if not fully synthesized + if not synthesized: + actions.extend(self.actions_synthesis_indices) + + # Layout can be explored if not layouted + if not layouted: + actions.extend(self.actions_layout_indices) + actions.extend(self.actions_mapping_indices) + + # Routing is only valid if layout is assigned + if layouted and not routed: + actions.extend(self.actions_routing_indices) + actions.extend(self.actions_mapping_indices) + + # General optimizations can happen in any non-final state + actions.extend(self.actions_opt_indices) + + return actions From 7eb5138341ebaa5f30f60c3f1b064f1a23c9ed24 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Thu, 22 Jan 2026 17:18:07 +0100 Subject: [PATCH 07/22] =?UTF-8?q?=F0=9F=9A=A7=20update=20state=20machine?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 83 +++++++++++++++++++--------- 1 file changed, 56 insertions(+), 27 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index a372da375..255494662 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -108,7 +108,7 @@ def __init__( self.actions_synthesis_indices = [] self.actions_layout_indices = [] self.actions_routing_indices = [] - self.actions_mapping_indices = [] + self.actions_mapping_and_routing_indices = [] self.actions_opt_indices = [] self.actions_final_optimization_indices = [] self.actions_structure_preserving_indices = [] # Actions that preserves the mapping and native gates @@ -144,7 +144,7 @@ def __init__( index += 1 for elem in action_dict[PassType.MAPPING]: self.action_set[index] = elem - self.actions_mapping_indices.append(index) + self.actions_mapping_and_routing_indices.append(index) index += 1 for elem in action_dict[PassType.FINAL_OPT]: self.action_set[index] = elem @@ -367,7 +367,9 @@ def _apply_qiskit_action(self, action: Action, action_index: int) -> QuantumCirc pm_property_set = dict(pm.property_set) if hasattr(pm, "property_set") else None if action_index in ( - self.actions_layout_indices + self.actions_mapping_indices + self.actions_final_optimization_indices + self.actions_layout_indices + + self.actions_mapping_and_routing_indices + + self.actions_final_optimization_indices ): altered_qc = self._handle_qiskit_layout_postprocessing(action, pm_property_set, altered_qc) elif ( @@ -495,7 +497,7 @@ def _apply_bqskit_action(self, action: Action, action_index: int) -> QuantumCirc elif action_index in self.actions_synthesis_indices: factory = cast("Callable[[Target], Callable[[Circuit], Circuit]]", action.transpile_pass) bqskit_compiled_qc = factory(self.device)(bqskit_qc) - elif action_index in self.actions_mapping_indices: + elif action_index in self.actions_mapping_and_routing_indices: factory = cast( "Callable[[Target], Callable[[Circuit], tuple[Circuit, list[int], list[int]]]]", action.transpile_pass, @@ -510,7 +512,21 @@ def _apply_bqskit_action(self, action: Action, action_index: int) -> QuantumCirc return bqskit_to_qiskit(bqskit_compiled_qc) - def is_circuit_mapped(self, circuit: QuantumCircuit, coupling_map: CouplingMap) -> bool: + def is_circuit_mapped(self, circuit: QuantumCircuit, layout: TranspileLayout | Layout) -> bool: + """True if every logical qubit in the circuit has a physical assignment.""" + if isinstance(layout, TranspileLayout): + # Use final_layout if available; otherwise fallback to initial_layout + layout = layout.final_layout or layout.initial_layout + + v2p = layout.get_virtual_bits() + for instr in circuit.data: + for q in instr.qubits: + if q not in v2p: + # Logical qubit not assigned + return False + return True + + def is_circuit_routed(self, circuit: QuantumCircuit, coupling_map: CouplingMap) -> bool: """Check if a circuit is fully routed/mapped to the device, including directionality. A circuit is considered mapped if all two-qubit gates are on qubits @@ -527,7 +543,7 @@ def is_circuit_mapped(self, circuit: QuantumCircuit, coupling_map: CouplingMap) directed_edges = set(coupling_map.get_edges()) for instr in circuit.data: - qubits = [q._index for q in instr.qubits] + qubits = [q._index for q in instr.qubits] # noqa: SLF001 if len(qubits) == 2: q0, q1 = qubits # If this two-qubit gate is not allowed in the device coupling map, return False @@ -536,38 +552,51 @@ def is_circuit_mapped(self, circuit: QuantumCircuit, coupling_map: CouplingMap) return True def determine_valid_actions_for_state(self) -> list[int]: - """Determine valid actions based on circuit state: synthesized, layouted, routed.""" + """Determine valid actions based on circuit state: synthesized, mapped, routed.""" check_nat_gates = GatesInBasis(basis_gates=self.device.operation_names) check_nat_gates(self.state) synthesized = check_nat_gates.property_set["all_gates_in_basis"] - layouted = self.layout is not None - routed = self.is_circuit_mapped(self.state, CouplingMap(self.device.build_coupling_map())) - - # Final state - if synthesized and layouted and routed: - return [ - self.action_terminate_index, - *self.actions_structure_preserving_indices, - *self.actions_final_optimization_indices, - ] + mapped = self.is_circuit_mapped(self.state, self.layout) if self.layout else False + # Routing is only allowed after mapping + routed = self.is_circuit_routed(self.state, CouplingMap(self.device.build_coupling_map())) if mapped else False actions = [] - # Synthesis is possible if not fully synthesized - if not synthesized: + # Initial state + if not synthesized and not mapped and not routed: actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_mapping_and_routing_indices) + actions.extend(self.actions_layout_indices) + actions.extend(self.actions_opt_indices) - # Layout can be explored if not layouted - if not layouted: + # LEFT TOP + if synthesized and not mapped and not routed: + actions.extend(self.actions_mapping_and_routing_indices) actions.extend(self.actions_layout_indices) - actions.extend(self.actions_mapping_indices) + actions.extend(self.actions_opt_indices) + + # LEFT BOTTOM + if not synthesized and mapped and not routed: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_routing_indices) + actions.extend(self.actions_opt_indices) - # Routing is only valid if layout is assigned - if layouted and not routed: + # RIGHT TOP + if synthesized and mapped and not routed: actions.extend(self.actions_routing_indices) - actions.extend(self.actions_mapping_indices) + actions.extend(self.actions_opt_indices) + + # RIGHT BOTTOM + if not synthesized and mapped and routed: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_opt_indices) - # General optimizations can happen in any non-final state - actions.extend(self.actions_opt_indices) + # Final state + if synthesized and mapped and routed: + return [ + self.action_terminate_index, + *self.actions_structure_preserving_indices, + *self.actions_final_optimization_indices, + ] return actions From 37653fb4eeeb0a9c5094709cda5d37b428ac5648 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Thu, 22 Jan 2026 17:32:51 +0100 Subject: [PATCH 08/22] =?UTF-8?q?=F0=9F=8E=A8=20add=20strict=20policy?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 37 +++++++++++++++++++--------- 1 file changed, 25 insertions(+), 12 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 255494662..74cbe0c3a 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -561,33 +561,47 @@ def determine_valid_actions_for_state(self) -> list[int]: routed = self.is_circuit_routed(self.state, CouplingMap(self.device.build_coupling_map())) if mapped else False actions = [] + flexible = False + strict = True # Initial state if not synthesized and not mapped and not routed: - actions.extend(self.actions_synthesis_indices) - actions.extend(self.actions_mapping_and_routing_indices) - actions.extend(self.actions_layout_indices) - actions.extend(self.actions_opt_indices) + if flexible: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_mapping_and_routing_indices) + actions.extend(self.actions_layout_indices) + actions.extend(self.actions_opt_indices) + if strict: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_opt_indices) # LEFT TOP if synthesized and not mapped and not routed: - actions.extend(self.actions_mapping_and_routing_indices) - actions.extend(self.actions_layout_indices) - actions.extend(self.actions_opt_indices) + if flexible: + actions.extend(self.actions_mapping_and_routing_indices) + actions.extend(self.actions_layout_indices) + actions.extend(self.actions_opt_indices) + if strict: + actions.extend(self.actions_mapping_and_routing_indices) + actions.extend(self.actions_layout_indices) + actions.extend(self.actions_structure_preserving_indices) # LEFT BOTTOM - if not synthesized and mapped and not routed: + if not synthesized and mapped and not routed and flexible: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) # RIGHT TOP if synthesized and mapped and not routed: - actions.extend(self.actions_routing_indices) - actions.extend(self.actions_opt_indices) + if flexible: + actions.extend(self.actions_routing_indices) + actions.extend(self.actions_opt_indices) + if strict: + actions.extend(self.actions_routing_indices) # RIGHT BOTTOM - if not synthesized and mapped and routed: + if not synthesized and mapped and routed and flexible: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) @@ -598,5 +612,4 @@ def determine_valid_actions_for_state(self) -> list[int]: *self.actions_structure_preserving_indices, *self.actions_final_optimization_indices, ] - return actions From 8802fbdba5297a2da8d23ec9ac3a0589053ea7cc Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Thu, 22 Jan 2026 18:07:32 +0100 Subject: [PATCH 09/22] =?UTF-8?q?=F0=9F=9A=A7=20add=20og=20paper=20strateg?= =?UTF-8?q?y?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 53 ++++++++++++++++++++-------- 1 file changed, 38 insertions(+), 15 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 74cbe0c3a..a9178c3c4 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -561,8 +561,10 @@ def determine_valid_actions_for_state(self) -> list[int]: routed = self.is_circuit_routed(self.state, CouplingMap(self.device.build_coupling_map())) if mapped else False actions = [] - flexible = False - strict = True + + flexible = True # no restrictions + strict = False # masters thesis + og = False # original paper # Initial state if not synthesized and not mapped and not routed: @@ -574,8 +576,10 @@ def determine_valid_actions_for_state(self) -> list[int]: if strict: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) + if og: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_opt_indices) - # LEFT TOP if synthesized and not mapped and not routed: if flexible: actions.extend(self.actions_mapping_and_routing_indices) @@ -585,31 +589,50 @@ def determine_valid_actions_for_state(self) -> list[int]: actions.extend(self.actions_mapping_and_routing_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_structure_preserving_indices) + if og: + actions.extend(self.actions_mapping_and_routing_indices) + actions.extend(self.actions_layout_indices) + actions.extend(self.actions_opt_indices) - # LEFT BOTTOM - if not synthesized and mapped and not routed and flexible: - actions.extend(self.actions_synthesis_indices) - actions.extend(self.actions_routing_indices) - actions.extend(self.actions_opt_indices) + if not synthesized and mapped and not routed: + if flexible: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_routing_indices) + actions.extend(self.actions_opt_indices) + if strict: # not depicted in thesis + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_routing_indices) + actions.extend(self.actions_opt_indices) + if og: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_routing_indices) + actions.extend(self.actions_opt_indices) - # RIGHT TOP if synthesized and mapped and not routed: if flexible: actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) if strict: actions.extend(self.actions_routing_indices) + if og: + actions.extend(self.actions_routing_indices) - # RIGHT BOTTOM + # NEW if not synthesized and mapped and routed and flexible: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) # Final state if synthesized and mapped and routed: - return [ - self.action_terminate_index, - *self.actions_structure_preserving_indices, - *self.actions_final_optimization_indices, - ] + if flexible: + actions.extend([self.action_terminate_index]) + actions.extend(self.actions_opt_indices) + if strict: + actions.extend([self.action_terminate_index]) + actions.extend(self.actions_structure_preserving_indices) + actions.extend(self.actions_final_optimization_indices) + if og: + actions.extend([self.action_terminate_index]) + actions.extend(self.actions_opt_indices) + return actions From 6db8ee75044dfbaac7c5964e72694ff3c1632978 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Mon, 9 Feb 2026 13:14:18 +0100 Subject: [PATCH 10/22] =?UTF-8?q?=F0=9F=8E=A8=20fix=20og=20policy?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 19 ++++++++++++++----- 1 file changed, 14 insertions(+), 5 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index a9178c3c4..52a9bdde2 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -594,12 +594,13 @@ def determine_valid_actions_for_state(self) -> list[int]: actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) + # Not depicted in paper/thesis, but covered in implementation because optimization can destroy the native gate set. if not synthesized and mapped and not routed: if flexible: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) - if strict: # not depicted in thesis + if strict: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) @@ -608,6 +609,7 @@ def determine_valid_actions_for_state(self) -> list[int]: actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) + # Not depicted in paper, but possible due to mapping-only passes if synthesized and mapped and not routed: if flexible: actions.extend(self.actions_routing_indices) @@ -617,10 +619,17 @@ def determine_valid_actions_for_state(self) -> list[int]: if og: actions.extend(self.actions_routing_indices) - # NEW - if not synthesized and mapped and routed and flexible: - actions.extend(self.actions_synthesis_indices) - actions.extend(self.actions_opt_indices) + # Not depicted in paper/thesis, but possible since routing can insert non-native SWAPs + if not synthesized and mapped and routed: + if flexible: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_opt_indices) + if strict: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_structure_preserving_indices) + if og: + actions.extend(self.actions_synthesis_indices) + actions.extend(self.actions_opt_indices) # Final state if synthesized and mapped and routed: From dcc38102a376cdba29de721c264081234e0215fc Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Tue, 17 Feb 2026 10:24:13 +0100 Subject: [PATCH 11/22] =?UTF-8?q?=E2=8F=AA=20remove=20thesis=20changes?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictor.py | 2 +- src/mqt/predictor/rl/predictorenv.py | 55 +++++++--------------------- 2 files changed, 14 insertions(+), 43 deletions(-) diff --git a/src/mqt/predictor/rl/predictor.py b/src/mqt/predictor/rl/predictor.py index 0329d26c5..e04aedf87 100644 --- a/src/mqt/predictor/rl/predictor.py +++ b/src/mqt/predictor/rl/predictor.py @@ -109,7 +109,7 @@ def train_model( n_steps = 2048 n_epochs = 10 batch_size = 64 - progress_bar = False + progress_bar = True logger.debug("Start training for: " + self.figure_of_merit + " on " + self.device_name) model = MaskablePPO( diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 52a9bdde2..70bdfb944 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -24,7 +24,6 @@ from pathlib import Path from bqskit import Circuit - from pytket.circuit import Node from qiskit.transpiler import Target from mqt.predictor.reward import figure_of_merit @@ -111,7 +110,6 @@ def __init__( self.actions_mapping_and_routing_indices = [] self.actions_opt_indices = [] self.actions_final_optimization_indices = [] - self.actions_structure_preserving_indices = [] # Actions that preserves the mapping and native gates self.used_actions: list[str] = [] self.device = device @@ -128,12 +126,6 @@ def __init__( self.action_set[index] = elem self.actions_synthesis_indices.append(index) index += 1 - for elem in action_dict[PassType.OPT]: - self.action_set[index] = elem - self.actions_opt_indices.append(index) - if getattr(elem, "preserve_layout", False): - self.actions_structure_preserving_indices.append(index) - index += 1 for elem in action_dict[PassType.LAYOUT]: self.action_set[index] = elem self.actions_layout_indices.append(index) @@ -142,6 +134,10 @@ def __init__( self.action_set[index] = elem self.actions_routing_indices.append(index) index += 1 + for elem in action_dict[PassType.OPT]: + self.action_set[index] = elem + self.actions_opt_indices.append(index) + index += 1 for elem in action_dict[PassType.MAPPING]: self.action_set[index] = elem self.actions_mapping_and_routing_indices.append(index) @@ -183,10 +179,6 @@ def __init__( } self.observation_space = Dict(spaces) self.filename = "" - self.max_iter = 20 - self.node_err: dict[Node, float] | None = None - self.edge_err: dict[tuple[Node, Node], float] | None = None - self.readout_err: dict[Node, float] | None = None def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any, Any]]: """Executes the given action and returns the new state, the reward, whether the episode is done, whether the episode is truncated and additional information. @@ -226,7 +218,7 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any reward_val = 0 done = False - # in case the Qiskit.QuantumCircuit has unitary or u gates or clifford in it, decompose them (because otherwise qiskit will throw an error when applying the BasisTranslator) + # In case the Qiskit.QuantumCircuit has unitary or u gates or clifford in it, decompose them (because otherwise qiskit will throw an error when applying the BasisTranslator) if self.state.count_ops().get("unitary"): self.state = self.state.decompose(gates_to_decompose="unitary") elif self.state.count_ops().get("clifford"): @@ -436,9 +428,9 @@ def _apply_tket_action(self, action: Action, action_index: int) -> QuantumCircui placement = transpile_pass[0].get_placement_map(tket_qc) except Exception as e: logger.warning("Placement failed (%s): %s. Falling back to original circuit.", action.name, e) - return tk_to_qiskit(tket_qc, replace_implicit_swaps=True) + return tk_to_qiskit(tket_qc, replace_implicit_swaps=True, perm_warning=False) else: - qc_tmp = tk_to_qiskit(tket_qc, replace_implicit_swaps=True) + qc_tmp = tk_to_qiskit(tket_qc, replace_implicit_swaps=True, perm_warning=False) qiskit_mapping = { qc_tmp.qubits[i]: placement[list(placement.keys())[i]].index[0] for i in range(len(placement)) @@ -468,7 +460,7 @@ def _apply_tket_action(self, action: Action, action_index: int) -> QuantumCircui qbs = tket_qc.qubits tket_qc.rename_units({qbs[i]: Qubit("q", i) for i in range(len(qbs))}) - altered_qc = tk_to_qiskit(tket_qc, replace_implicit_swaps=True) + altered_qc = tk_to_qiskit(tket_qc, replace_implicit_swaps=True, perm_warning=False) if action_index in self.actions_routing_indices: assert self.layout is not None @@ -562,9 +554,8 @@ def determine_valid_actions_for_state(self) -> list[int]: actions = [] - flexible = True # no restrictions - strict = False # masters thesis - og = False # original paper + flexible = True # No restrictions + og = False # Original paper # Initial state if not synthesized and not mapped and not routed: @@ -573,9 +564,6 @@ def determine_valid_actions_for_state(self) -> list[int]: actions.extend(self.actions_mapping_and_routing_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) - if strict: - actions.extend(self.actions_synthesis_indices) - actions.extend(self.actions_opt_indices) if og: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) @@ -585,48 +573,35 @@ def determine_valid_actions_for_state(self) -> list[int]: actions.extend(self.actions_mapping_and_routing_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) - if strict: - actions.extend(self.actions_mapping_and_routing_indices) - actions.extend(self.actions_layout_indices) - actions.extend(self.actions_structure_preserving_indices) if og: actions.extend(self.actions_mapping_and_routing_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) - # Not depicted in paper/thesis, but covered in implementation because optimization can destroy the native gate set. + # Not *depicted* in paper; necessary because optimization can destroy the native gate set if not synthesized and mapped and not routed: if flexible: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) - if strict: - actions.extend(self.actions_synthesis_indices) - actions.extend(self.actions_routing_indices) - actions.extend(self.actions_opt_indices) if og: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) - # Not depicted in paper, but possible due to mapping-only passes + # Not *depicted* in paper; necessary because of mapping-only passes if synthesized and mapped and not routed: if flexible: actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) - if strict: - actions.extend(self.actions_routing_indices) if og: actions.extend(self.actions_routing_indices) - # Not depicted in paper/thesis, but possible since routing can insert non-native SWAPs + # Not *depicted* in paper; necessary because routing can insert non-native SWAPs if not synthesized and mapped and routed: if flexible: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) - if strict: - actions.extend(self.actions_synthesis_indices) - actions.extend(self.actions_structure_preserving_indices) if og: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) @@ -636,10 +611,6 @@ def determine_valid_actions_for_state(self) -> list[int]: if flexible: actions.extend([self.action_terminate_index]) actions.extend(self.actions_opt_indices) - if strict: - actions.extend([self.action_terminate_index]) - actions.extend(self.actions_structure_preserving_indices) - actions.extend(self.actions_final_optimization_indices) if og: actions.extend([self.action_terminate_index]) actions.extend(self.actions_opt_indices) From 450d2dcab5251a186d322d58c0f745aca80e66d4 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Tue, 17 Feb 2026 14:58:28 +0100 Subject: [PATCH 12/22] =?UTF-8?q?=E2=8F=AA=20revert=20thesis=20updates?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/actions.py | 3 +- src/mqt/predictor/rl/predictorenv.py | 226 +++++++++------------------ 2 files changed, 76 insertions(+), 153 deletions(-) diff --git a/src/mqt/predictor/rl/actions.py b/src/mqt/predictor/rl/actions.py index 7b4432f5a..9efcc76d7 100644 --- a/src/mqt/predictor/rl/actions.py +++ b/src/mqt/predictor/rl/actions.py @@ -86,6 +86,7 @@ from bqskit import Circuit from pytket._tket.passes import BasePass as tket_BasePass + from qiskit.passmanager import PropertySet from qiskit.transpiler.basepasses import BasePass as qiskit_BasePass @@ -143,7 +144,7 @@ class DeviceDependentAction(Action): Callable[..., tuple[Any, ...] | Circuit], ] ) - do_while: Callable[[dict[str, Circuit]], bool] | None = None + do_while: Callable[[PropertySet], bool] | None = None # Registry of actions diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 70bdfb944..d8f3de857 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -11,20 +11,17 @@ from __future__ import annotations import logging -import sys from typing import TYPE_CHECKING, Any -if sys.version_info >= (3, 11) and TYPE_CHECKING: # pragma: no cover - from typing import assert_never -else: - from typing_extensions import assert_never +from pytket._tket.passes import BasePass as TketBasePass # noqa: PLC2701 if TYPE_CHECKING: from collections.abc import Callable from pathlib import Path from bqskit import Circuit - from qiskit.transpiler import Target + from qiskit.passmanager.base_tasks import Task + from qiskit.transpiler import Layout, Target from mqt.predictor.reward import figure_of_merit from mqt.predictor.rl.actions import Action @@ -42,14 +39,8 @@ from pytket.extensions.qiskit import qiskit_to_tk, tk_to_qiskit from qiskit import QuantumCircuit from qiskit.passmanager.flow_controllers import DoWhileController -from qiskit.transpiler import CouplingMap, Layout, PassManager, TranspileLayout -from qiskit.transpiler.passes import ( - ApplyLayout, - EnlargeWithAncilla, - FullAncillaAllocation, - GatesInBasis, - SetLayout, -) +from qiskit.transpiler import CouplingMap, PassManager, TranspileLayout +from qiskit.transpiler.passes import GatesInBasis from qiskit.transpiler.passes.layout.vf2_layout import VF2LayoutStopReason from mqt.predictor.hellinger import get_hellinger_model_path @@ -60,18 +51,10 @@ estimated_success_probability, expected_fidelity, ) -from mqt.predictor.rl.actions import ( - CompilationOrigin, - DeviceDependentAction, - PassType, - get_actions_by_pass_type, -) -from mqt.predictor.rl.helper import ( - create_feature_dict, - get_path_training_circuits, - get_state_sample, -) +from mqt.predictor.rl.actions import CompilationOrigin, DeviceDependentAction, PassType, get_actions_by_pass_type +from mqt.predictor.rl.helper import create_feature_dict, get_path_training_circuits, get_state_sample from mqt.predictor.rl.parsing import ( + PreProcessTKETRoutingAfterQiskitLayout, final_layout_bqskit_to_qiskit, final_layout_pytket_to_qiskit, postprocess_vf2postlayout, @@ -80,7 +63,7 @@ logger = logging.getLogger("mqt-predictor") -class PredictorEnv(Env): # type: ignore[misc] +class PredictorEnv(Env): """Predictor environment for reinforcement learning.""" def __init__( @@ -107,7 +90,7 @@ def __init__( self.actions_synthesis_indices = [] self.actions_layout_indices = [] self.actions_routing_indices = [] - self.actions_mapping_and_routing_indices = [] + self.actions_mapping_indices = [] self.actions_opt_indices = [] self.actions_final_optimization_indices = [] self.used_actions: list[str] = [] @@ -140,7 +123,7 @@ def __init__( index += 1 for elem in action_dict[PassType.MAPPING]: self.action_set[index] = elem - self.actions_mapping_and_routing_indices.append(index) + self.actions_mapping_indices.append(index) index += 1 for elem in action_dict[PassType.FINAL_OPT]: self.action_set[index] = elem @@ -218,28 +201,26 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any reward_val = 0 done = False - # In case the Qiskit.QuantumCircuit has unitary or u gates or clifford in it, decompose them (because otherwise qiskit will throw an error when applying the BasisTranslator) - if self.state.count_ops().get("unitary"): + # in case the Qiskit.QuantumCircuit has unitary or u gates in it, decompose them (because otherwise qiskit will throw an error when applying the BasisTranslator + if self.state.count_ops().get("unitary"): # ty: ignore[invalid-argument-type] self.state = self.state.decompose(gates_to_decompose="unitary") - elif self.state.count_ops().get("clifford"): - self.state = self.state.decompose(gates_to_decompose="clifford") self.state._layout = self.layout # noqa: SLF001 + obs = create_feature_dict(self.state) + return obs, reward_val, done, False, {} - return create_feature_dict(self.state), reward_val, done, False, {} - - def calculate_reward(self, qc: QuantumCircuit | None = None) -> float: - """Calculates and returns the reward for either the current state or a quantum circuit (if provided).""" - circuit = self.state if qc is None else qc + def calculate_reward(self) -> float: + """Calculates and returns the reward for the current state.""" if self.reward_function == "expected_fidelity": - return expected_fidelity(circuit, self.device) + return expected_fidelity(self.state, self.device) if self.reward_function == "estimated_success_probability": - return estimated_success_probability(circuit, self.device) + return estimated_success_probability(self.state, self.device) if self.reward_function == "estimated_hellinger_distance": - return estimated_hellinger_distance(circuit, self.device, self.hellinger_model) + return estimated_hellinger_distance(self.state, self.device, self.hellinger_model) if self.reward_function == "critical_depth": - return crit_depth(circuit) - assert_never(circuit) + return crit_depth(self.state) + msg = f"No implementation for reward function {self.reward_function}." + raise NotImplementedError(msg) def render(self) -> None: """Renders the current state.""" @@ -250,7 +231,7 @@ def reset( qc: Path | str | QuantumCircuit | None = None, seed: int | None = None, options: dict[str, Any] | None = None, # noqa: ARG002 - ) -> tuple[QuantumCircuit, dict[str, Any]]: + ) -> tuple[dict[str, Any], dict[str, Any]]: """Resets the environment to the given state or a random state. Arguments: @@ -265,7 +246,7 @@ def reset( if isinstance(qc, QuantumCircuit): self.state = qc elif qc: - self.state = QuantumCircuit.from_qasm_file(str(qc)) + self.state = QuantumCircuit.from_qasm_file(qc) # ty: ignore[invalid-argument-type] else: self.state, self.filename = get_state_sample(self.device.num_qubits, self.path_training_circuits, self.rng) @@ -338,129 +319,68 @@ def apply_action(self, action_index: int) -> QuantumCircuit | None: raise ValueError(msg) def _apply_qiskit_action(self, action: Action, action_index: int) -> QuantumCircuit: - pm_property_set: dict[str, Any] | None = None - if action.name in ["QiskitO3", "Opt2qBlocks_preserve"] and isinstance(action, DeviceDependentAction): - passes = action.transpile_pass( + if action.name == "QiskitO3" and isinstance(action, DeviceDependentAction): + assert callable(action.transpile_pass) + passes_ = action.transpile_pass( self.device.operation_names, CouplingMap(self.device.build_coupling_map()) if self.layout else None, ) - if action.name == "QiskitO3": - pm = PassManager([DoWhileController(passes, do_while=action.do_while)]) - else: - pm = PassManager(passes) - altered_qc = pm.run(self.state) - pm_property_set = dict(pm.property_set) if hasattr(pm, "property_set") else None + passes = cast("list[Task]", passes_) + assert action.do_while is not None + pm = PassManager([DoWhileController(passes, do_while=action.do_while)]) else: - transpile_pass = ( - action.transpile_pass(self.device) if callable(action.transpile_pass) else action.transpile_pass - ) - pm = PassManager(transpile_pass) + passes_ = action.transpile_pass(self.device) if callable(action.transpile_pass) else action.transpile_pass + passes = cast("list[Task]", passes_) + pm = PassManager(passes) + altered_qc = pm.run(self.state) - pm_property_set = dict(pm.property_set) if hasattr(pm, "property_set") else None if action_index in ( - self.actions_layout_indices - + self.actions_mapping_and_routing_indices - + self.actions_final_optimization_indices - ): - altered_qc = self._handle_qiskit_layout_postprocessing(action, pm_property_set, altered_qc) - elif ( - action_index in self.actions_routing_indices - and self.layout is not None - and pm_property_set is not None - and "final_layout" in pm_property_set + self.actions_layout_indices + self.actions_mapping_indices + self.actions_final_optimization_indices ): - self.layout.final_layout = pm_property_set["final_layout"] + altered_qc = self._handle_qiskit_layout_postprocessing(action, pm, altered_qc) + + elif action_index in self.actions_routing_indices and self.layout: + self.layout.final_layout = pm.property_set["final_layout"] return altered_qc def _handle_qiskit_layout_postprocessing( - self, - action: Action, - pm_property_set: dict[str, Any] | None, - altered_qc: QuantumCircuit, + self, action: Action, pm: PassManager, altered_qc: QuantumCircuit ) -> QuantumCircuit: - if not pm_property_set: - return altered_qc if action.name == "VF2PostLayout": - assert pm_property_set["VF2PostLayout_stop_reason"] is not None - post_layout = pm_property_set.get("post_layout") + assert pm.property_set["VF2PostLayout_stop_reason"] is not None + post_layout = pm.property_set["post_layout"] if post_layout: - try: - altered_qc, _ = postprocess_vf2postlayout(altered_qc, post_layout, self.layout) - except Exception as e: - logger.warning( - "VF2PostLayout postprocessing failed. Contintinuing with previous circuit. Error: %s", e - ) - return self.state + assert self.layout is not None + altered_qc, _ = postprocess_vf2postlayout(altered_qc, post_layout, self.layout) elif action.name == "VF2Layout": - if pm_property_set["VF2Layout_stop_reason"] == VF2LayoutStopReason.SOLUTION_FOUND: - assert pm_property_set["layout"] - else: - logger.warning("VF2Layout did not find a solution. Continuing with previous circuit.") - return self.state + assert pm.property_set["VF2Layout_stop_reason"] == VF2LayoutStopReason.SOLUTION_FOUND + assert pm.property_set["layout"] else: - assert pm_property_set["layout"] + assert pm.property_set["layout"] - layout = pm_property_set.get("layout") - if layout: + if pm.property_set["layout"]: self.layout = TranspileLayout( - initial_layout=layout, - input_qubit_mapping=pm_property_set.get("original_qubit_indices"), - final_layout=pm_property_set.get("final_layout"), + initial_layout=pm.property_set["layout"], + input_qubit_mapping=pm.property_set["original_qubit_indices"], + final_layout=pm.property_set["final_layout"], _output_qubit_list=altered_qc.qubits, _input_qubit_count=self.num_qubits_uncompiled_circuit, ) - - if self.layout is not None and pm_property_set.get("final_layout"): - self.layout.final_layout = pm_property_set["final_layout"] return altered_qc def _apply_tket_action(self, action: Action, action_index: int) -> QuantumCircuit: tket_qc = qiskit_to_tk(self.state, preserve_param_uuid=True) - transpile_pass = ( - action.transpile_pass(self.device) if callable(action.transpile_pass) else action.transpile_pass - ) - assert isinstance(transpile_pass, list) - # Map TKET placement into a Qiskit layout - if action_index in self.actions_layout_indices: - try: - placement = transpile_pass[0].get_placement_map(tket_qc) - except Exception as e: - logger.warning("Placement failed (%s): %s. Falling back to original circuit.", action.name, e) - return tk_to_qiskit(tket_qc, replace_implicit_swaps=True, perm_warning=False) - else: - qc_tmp = tk_to_qiskit(tket_qc, replace_implicit_swaps=True, perm_warning=False) - - qiskit_mapping = { - qc_tmp.qubits[i]: placement[list(placement.keys())[i]].index[0] for i in range(len(placement)) - } - layout = Layout(qiskit_mapping) - - pm = PassManager([ - SetLayout(layout), - FullAncillaAllocation(coupling_map=CouplingMap(self.device.build_coupling_map())), - EnlargeWithAncilla(), - ApplyLayout(), - ]) - altered_qc = pm.run(qc_tmp) - - self.layout = TranspileLayout( - initial_layout=pm.property_set.get("layout"), - input_qubit_mapping=pm.property_set["original_qubit_indices"], - final_layout=pm.property_set["final_layout"], - _output_qubit_list=altered_qc.qubits, - _input_qubit_count=self.num_qubits_uncompiled_circuit, - ) - return altered_qc - - else: - for p in transpile_pass: - p.apply(tket_qc) + passes = action.transpile_pass(self.device) if callable(action.transpile_pass) else action.transpile_pass + assert isinstance(passes, list) + for pass_ in passes: + assert isinstance(pass_, TketBasePass | PreProcessTKETRoutingAfterQiskitLayout) + pass_.apply(tket_qc) qbs = tket_qc.qubits tket_qc.rename_units({qbs[i]: Qubit("q", i) for i in range(len(qbs))}) - altered_qc = tk_to_qiskit(tket_qc, replace_implicit_swaps=True, perm_warning=False) + altered_qc = tk_to_qiskit(tket_qc) if action_index in self.actions_routing_indices: assert self.layout is not None @@ -489,7 +409,7 @@ def _apply_bqskit_action(self, action: Action, action_index: int) -> QuantumCirc elif action_index in self.actions_synthesis_indices: factory = cast("Callable[[Target], Callable[[Circuit], Circuit]]", action.transpile_pass) bqskit_compiled_qc = factory(self.device)(bqskit_qc) - elif action_index in self.actions_mapping_and_routing_indices: + elif action_index in self.actions_mapping_indices: factory = cast( "Callable[[Target], Callable[[Circuit], tuple[Circuit, list[int], list[int]]]]", action.transpile_pass, @@ -504,7 +424,7 @@ def _apply_bqskit_action(self, action: Action, action_index: int) -> QuantumCirc return bqskit_to_qiskit(bqskit_compiled_qc) - def is_circuit_mapped(self, circuit: QuantumCircuit, layout: TranspileLayout | Layout) -> bool: + def is_circuit_laid_out(self, circuit: QuantumCircuit, layout: TranspileLayout | Layout) -> bool: """True if every logical qubit in the circuit has a physical assignment.""" if isinstance(layout, TranspileLayout): # Use final_layout if available; otherwise fallback to initial_layout @@ -548,9 +468,11 @@ def determine_valid_actions_for_state(self) -> list[int]: check_nat_gates = GatesInBasis(basis_gates=self.device.operation_names) check_nat_gates(self.state) synthesized = check_nat_gates.property_set["all_gates_in_basis"] - mapped = self.is_circuit_mapped(self.state, self.layout) if self.layout else False - # Routing is only allowed after mapping - routed = self.is_circuit_routed(self.state, CouplingMap(self.device.build_coupling_map())) if mapped else False + laid_out = self.is_circuit_laid_out(self.state, self.layout) if self.layout else False + # Routing is only allowed after layout + routed = ( + self.is_circuit_routed(self.state, CouplingMap(self.device.build_coupling_map())) if laid_out else False + ) actions = [] @@ -558,28 +480,28 @@ def determine_valid_actions_for_state(self) -> list[int]: og = False # Original paper # Initial state - if not synthesized and not mapped and not routed: + if not synthesized and not laid_out and not routed: if flexible: actions.extend(self.actions_synthesis_indices) - actions.extend(self.actions_mapping_and_routing_indices) + actions.extend(self.actions_mapping_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) if og: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) - if synthesized and not mapped and not routed: + if synthesized and not laid_out and not routed: if flexible: - actions.extend(self.actions_mapping_and_routing_indices) + actions.extend(self.actions_mapping_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) if og: - actions.extend(self.actions_mapping_and_routing_indices) + actions.extend(self.actions_mapping_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) # Not *depicted* in paper; necessary because optimization can destroy the native gate set - if not synthesized and mapped and not routed: + if not synthesized and laid_out and not routed: if flexible: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_routing_indices) @@ -590,7 +512,7 @@ def determine_valid_actions_for_state(self) -> list[int]: actions.extend(self.actions_opt_indices) # Not *depicted* in paper; necessary because of mapping-only passes - if synthesized and mapped and not routed: + if synthesized and laid_out and not routed: if flexible: actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) @@ -598,7 +520,7 @@ def determine_valid_actions_for_state(self) -> list[int]: actions.extend(self.actions_routing_indices) # Not *depicted* in paper; necessary because routing can insert non-native SWAPs - if not synthesized and mapped and routed: + if not synthesized and laid_out and routed: if flexible: actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) @@ -607,7 +529,7 @@ def determine_valid_actions_for_state(self) -> list[int]: actions.extend(self.actions_opt_indices) # Final state - if synthesized and mapped and routed: + if synthesized and laid_out and routed: if flexible: actions.extend([self.action_terminate_index]) actions.extend(self.actions_opt_indices) From e69cf5142e5bdace4a0b4ad7294d5a7f48f558b1 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Tue, 17 Feb 2026 15:27:30 +0100 Subject: [PATCH 13/22] =?UTF-8?q?=E2=8F=AA=20use=20og=20strategy?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index d8f3de857..8396c7a2c 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -333,7 +333,7 @@ def _apply_qiskit_action(self, action: Action, action_index: int) -> QuantumCirc passes = cast("list[Task]", passes_) pm = PassManager(passes) - altered_qc = pm.run(self.state) + altered_qc = pm.run(self.state) if action_index in ( self.actions_layout_indices + self.actions_mapping_indices + self.actions_final_optimization_indices @@ -476,8 +476,8 @@ def determine_valid_actions_for_state(self) -> list[int]: actions = [] - flexible = True # No restrictions - og = False # Original paper + og = True # Original (restricted) MDP + flexible = False # General MDP # Initial state if not synthesized and not laid_out and not routed: From 934e46c10215f9c92e78e71527e3802e0267d999 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Tue, 17 Feb 2026 15:56:24 +0100 Subject: [PATCH 14/22] =?UTF-8?q?=F0=9F=90=9B=20fix=20no-layout=20found=20?= =?UTF-8?q?bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 8396c7a2c..1ce8efa14 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -355,8 +355,12 @@ def _handle_qiskit_layout_postprocessing( assert self.layout is not None altered_qc, _ = postprocess_vf2postlayout(altered_qc, post_layout, self.layout) elif action.name == "VF2Layout": - assert pm.property_set["VF2Layout_stop_reason"] == VF2LayoutStopReason.SOLUTION_FOUND - assert pm.property_set["layout"] + if pm.property_set["VF2Layout_stop_reason"] != VF2LayoutStopReason.SOLUTION_FOUND: + logger.warning( + "VF2Layout pass did not find a solution. Reason: " + str(pm.property_set["VF2Layout_stop_reason"]) + ) + else: + assert pm.property_set["layout"] else: assert pm.property_set["layout"] From 8b27982c60614c4cdbd470698a2181e76eff7c08 Mon Sep 17 00:00:00 2001 From: Patrick Hopf <81010725+flowerthrower@users.noreply.github.com> Date: Mon, 23 Feb 2026 22:06:21 +0100 Subject: [PATCH 15/22] Update src/mqt/predictor/rl/predictorenv.py Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> Signed-off-by: Patrick Hopf <81010725+flowerthrower@users.noreply.github.com> --- src/mqt/predictor/rl/predictorenv.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 01cad99d4..2e1244f8d 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -359,7 +359,8 @@ def _handle_qiskit_layout_postprocessing( elif action.name == "VF2Layout": if pm.property_set["VF2Layout_stop_reason"] != VF2LayoutStopReason.SOLUTION_FOUND: logger.warning( - "VF2Layout pass did not find a solution. Reason: " + str(pm.property_set["VF2Layout_stop_reason"]) + "VF2Layout pass did not find a solution. Reason: %s", + pm.property_set["VF2Layout_stop_reason"], ) else: assert pm.property_set["layout"] From 729ade70619c0737279435a8485d17d1c6791d9d Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Fri, 27 Feb 2026 14:58:48 +0100 Subject: [PATCH 16/22] =?UTF-8?q?=F0=9F=8E=A8=20docstring?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 2e1244f8d..ab31a976c 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -448,9 +448,9 @@ def is_circuit_laid_out(self, circuit: QuantumCircuit, layout: TranspileLayout | return True def is_circuit_routed(self, circuit: QuantumCircuit, coupling_map: CouplingMap) -> bool: - """Check if a circuit is fully routed/mapped to the device, including directionality. + """Check if a circuit is fully routed to the device, including directionality. - A circuit is considered mapped if all two-qubit gates are on qubits + A circuit is considered routed if all two-qubit gates are on qubits allowed by the coupling map and follow the allowed direction. Args: @@ -458,7 +458,7 @@ def is_circuit_routed(self, circuit: QuantumCircuit, coupling_map: CouplingMap) coupling_map: CouplingMap of the target device. Returns: - True if fully mapped and directed, False otherwise. + True if fully routed, False otherwise. """ # Create a set of directed edges for fast lookup directed_edges = set(coupling_map.get_edges()) From de36b57056fc5846118a2783205d0b42baaa93f4 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Tue, 3 Mar 2026 10:06:15 +0100 Subject: [PATCH 17/22] =?UTF-8?q?=F0=9F=90=9B=20fix=20routing=20check?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictor.py | 6 ++-- src/mqt/predictor/rl/predictorenv.py | 28 ++++++++++++------- .../test_integration_further_SDKs.py | 2 +- 3 files changed, 22 insertions(+), 14 deletions(-) diff --git a/src/mqt/predictor/rl/predictor.py b/src/mqt/predictor/rl/predictor.py index 2f355c123..1f75b1901 100644 --- a/src/mqt/predictor/rl/predictor.py +++ b/src/mqt/predictor/rl/predictor.py @@ -99,9 +99,9 @@ def train_model( """ if test: set_random_seed(0) # for reproducibility - n_steps = 10 - n_epochs = 1 - batch_size = 10 + n_steps = 32 + n_epochs = 2 + batch_size = 8 progress_bar = False else: set_random_seed(0) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index ab31a976c..d66df948b 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -390,7 +390,7 @@ def _apply_tket_action(self, action: Action, action_index: int) -> QuantumCircui qbs = tket_qc.qubits tket_qc.rename_units({qbs[i]: Qubit("q", i) for i in range(len(qbs))}) - altered_qc = tk_to_qiskit(tket_qc) + altered_qc = tk_to_qiskit(tket_qc, replace_implicit_swaps=True) if action_index in self.actions_routing_indices: assert self.layout is not None @@ -447,27 +447,33 @@ def is_circuit_laid_out(self, circuit: QuantumCircuit, layout: TranspileLayout | return False return True - def is_circuit_routed(self, circuit: QuantumCircuit, coupling_map: CouplingMap) -> bool: + def is_circuit_routed(self, circuit: QuantumCircuit, coupling_map: CouplingMap, layout: TranspileLayout) -> bool: """Check if a circuit is fully routed to the device, including directionality. - A circuit is considered routed if all two-qubit gates are on qubits - allowed by the coupling map and follow the allowed direction. + A circuit is considered routed if all two-qubit gates are on qubit pairs + that exist as directed edges in the device coupling map, using the physical + qubit indices resolved via the layout (virtual → physical mapping). Args: circuit: QuantumCircuit to check. coupling_map: CouplingMap of the target device. + layout: The transpile layout mapping virtual qubits to physical qubits. Returns: True if fully routed, False otherwise. """ - # Create a set of directed edges for fast lookup directed_edges = set(coupling_map.get_edges()) + # Resolve virtual → physical using the layout. + # q._index is register-local and must NOT be used here; v2p[q] gives the + # correct physical qubit index as seen by the device. + resolved_layout = layout.final_layout or layout.initial_layout + v2p = resolved_layout.get_virtual_bits() + for instr in circuit.data: - qubits = [q._index for q in instr.qubits] # noqa: SLF001 - if len(qubits) == 2: - q0, q1 = qubits - # If this two-qubit gate is not allowed in the device coupling map, return False + if len(instr.qubits) == 2: + q0 = v2p[instr.qubits[0]] + q1 = v2p[instr.qubits[1]] if (q0, q1) not in directed_edges: return False return True @@ -480,7 +486,9 @@ def determine_valid_actions_for_state(self) -> list[int]: laid_out = self.is_circuit_laid_out(self.state, self.layout) if self.layout else False # Routing is only allowed after layout routed = ( - self.is_circuit_routed(self.state, CouplingMap(self.device.build_coupling_map())) if laid_out else False + self.is_circuit_routed(self.state, CouplingMap(self.device.build_coupling_map()), self.layout) + if laid_out + else False ) actions = [] diff --git a/tests/compilation/test_integration_further_SDKs.py b/tests/compilation/test_integration_further_SDKs.py index bb49f971c..150363ea3 100644 --- a/tests/compilation/test_integration_further_SDKs.py +++ b/tests/compilation/test_integration_further_SDKs.py @@ -241,7 +241,7 @@ def test_tket_routing(available_actions_dict: dict[PassType, list[Action]]) -> N qubit_map = {qbs[i]: Qubit("q", i) for i in range(len(qbs))} tket_qc.rename_units(qubit_map) - mapped_qc = tk_to_qiskit(tket_qc, perm_warning=False) + mapped_qc = tk_to_qiskit(tket_qc, replace_implicit_swaps=True) final_layout = final_layout_pytket_to_qiskit(tket_qc, mapped_qc) From 7c745fea6b5f8e13569ac570bc7c6bb40885db84 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Wed, 4 Mar 2026 09:33:42 +0100 Subject: [PATCH 18/22] =?UTF-8?q?=F0=9F=90=9B=20use=20find=20qubit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/reward.py | 45 +++++-------------- src/mqt/predictor/rl/predictorenv.py | 60 +++++++++++++++----------- tests/compilation/test_predictor_rl.py | 2 +- 3 files changed, 48 insertions(+), 59 deletions(-) diff --git a/src/mqt/predictor/reward.py b/src/mqt/predictor/reward.py index 7c1ce1bba..a8cd76952 100644 --- a/src/mqt/predictor/reward.py +++ b/src/mqt/predictor/reward.py @@ -23,7 +23,6 @@ if TYPE_CHECKING: from qiskit import QuantumCircuit - from qiskit.circuit import QuantumRegister, Qubit from qiskit.transpiler import Target from sklearn.ensemble import RandomForestRegressor @@ -62,44 +61,22 @@ def expected_fidelity(qc: QuantumCircuit, device: Target, precision: int = 10) - if gate_type != "barrier": assert len(qargs) in [1, 2] - first_qubit_idx = calc_qubit_index(qargs, qc.qregs, 0) + first_qubit_idx = qc.find_bit(qargs[0]).index if len(qargs) == 1: specific_fidelity = 1 - device[gate_type][first_qubit_idx,].error else: - second_qubit_idx = calc_qubit_index(qargs, qc.qregs, 1) - specific_fidelity = 1 - device[gate_type][first_qubit_idx, second_qubit_idx].error - + second_qubit_idx = qc.find_bit(qargs[1]).index + try: + specific_fidelity = 1 - device[gate_type][first_qubit_idx, second_qubit_idx].error + except KeyError: + msg = f"Error rate for gate {gate_type} on qubits {first_qubit_idx} and {second_qubit_idx} not found in device properties." + raise KeyError(msg) from None res *= specific_fidelity return float(np.round(res, precision).item()) -def calc_qubit_index(qargs: list[Qubit], qregs: list[QuantumRegister], index: int) -> int: - """Calculates the global qubit index for a given quantum circuit and qubit index. - - Arguments: - qargs: The qubits of the quantum circuit. - qregs: The quantum registers of the quantum circuit. - index: The index of the qubit in the qargs list. - - Returns: - The global qubit index of the given qubit in the quantum circuit. - - Raises: - ValueError: If the qubit index is not found in the quantum registers. - """ - offset = 0 - for reg in qregs: - if qargs[index] not in reg: - offset += reg.size - else: - qubit_index: int = offset + reg.index(qargs[index]) - return qubit_index - error_msg = f"Global qubit index for local qubit {index} index not found." - raise ValueError(error_msg) - - def estimated_success_probability(qc: QuantumCircuit, device: Target, precision: int = 10) -> float: """Calculates the estimated success probability of a given quantum circuit on a given device. @@ -125,7 +102,7 @@ def estimated_success_probability(qc: QuantumCircuit, device: Target, precision: if gate_type == "barrier" or gate_type == "id": continue assert len(qargs) in (1, 2) - first_qubit_idx = calc_qubit_index(qargs, qc.qregs, 0) + first_qubit_idx = qc.find_bit(qargs[0]).index active_qubits.add(first_qubit_idx) if len(qargs) == 1: # single-qubit gate @@ -140,7 +117,7 @@ def estimated_success_probability(qc: QuantumCircuit, device: Target, precision: )) exec_time_per_qubit[first_qubit_idx] += duration else: # multi-qubit gate - second_qubit_idx = calc_qubit_index(qargs, qc.qregs, 1) + second_qubit_idx = qc.find_bit(qargs[1]).index active_qubits.add(second_qubit_idx) duration = device[gate_type][first_qubit_idx, second_qubit_idx].duration op_times.append((gate_type, [first_qubit_idx, second_qubit_idx], duration, "s")) @@ -191,7 +168,7 @@ def estimated_success_probability(qc: QuantumCircuit, device: Target, precision: continue assert len(qargs) in (1, 2) - first_qubit_idx = calc_qubit_index(qargs, qc.qregs, 0) + first_qubit_idx = scheduled_circ.find_bit(qargs[0]).index if len(qargs) == 1: if gate_type == "measure": @@ -213,7 +190,7 @@ def estimated_success_probability(qc: QuantumCircuit, device: Target, precision: continue res *= 1 - device[gate_type][first_qubit_idx,].error else: - second_qubit_idx = calc_qubit_index(qargs, qc.qregs, 1) + second_qubit_idx = scheduled_circ.find_bit(qargs[1]).index res *= 1 - device[gate_type][first_qubit_idx, second_qubit_idx].error if qiskit_version >= "2.0.0": diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index d66df948b..47d0c0a35 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -40,7 +40,6 @@ from qiskit import QuantumCircuit from qiskit.passmanager.flow_controllers import DoWhileController from qiskit.transpiler import CouplingMap, PassManager, TranspileLayout -from qiskit.transpiler.passes import GatesInBasis from qiskit.transpiler.passes.layout.vf2_layout import VF2LayoutStopReason from mqt.predictor.hellinger import get_hellinger_model_path @@ -195,16 +194,15 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any raise RuntimeError(msg) if action == self.action_terminate_index: + if action not in self.valid_actions: + msg = "Terminate action is not valid but was chosen." + raise RuntimeError(msg) reward_val = self.calculate_reward() done = True else: reward_val = 0 done = False - # in case the Qiskit.QuantumCircuit has unitary or u gates in it, decompose them (because otherwise qiskit will throw an error when applying the BasisTranslator - if self.state.count_ops().get("unitary"): # ty: ignore[invalid-argument-type] - self.state = self.state.decompose(gates_to_decompose="unitary") - self.state._layout = self.layout # noqa: SLF001 obs = create_feature_dict(self.state) return obs, reward_val, done, False, {} @@ -345,6 +343,11 @@ def _apply_qiskit_action(self, action: Action, action_index: int) -> QuantumCirc elif action_index in self.actions_routing_indices and self.layout: self.layout.final_layout = pm.property_set["final_layout"] + # BasisTranslator errors on unitary gates; decompose them immediately so + # the circuit is always in a consistent state after a Qiskit action. + if altered_qc.count_ops().get("unitary"): # ty: ignore[invalid-argument-type] + altered_qc = altered_qc.decompose(gates_to_decompose="unitary") + return altered_qc def _handle_qiskit_layout_postprocessing( @@ -447,48 +450,57 @@ def is_circuit_laid_out(self, circuit: QuantumCircuit, layout: TranspileLayout | return False return True - def is_circuit_routed(self, circuit: QuantumCircuit, coupling_map: CouplingMap, layout: TranspileLayout) -> bool: + def is_circuit_synthesized(self, circuit: QuantumCircuit) -> bool: + """Check if the circuit uses only native gates of the device. + + Verifies that every gate name in the circuit is present in + ``device.operation_names``, equivalent to the ``GatesInBasis`` pass. + + Args: + circuit: QuantumCircuit to check. + + Returns: + True if all gates are native to the device. + """ + native_names = set(self.device.operation_names) + return all( + instr.operation.name in native_names or instr.operation.name in ("barrier", "measure") + for instr in circuit.data + ) + + def is_circuit_routed(self, circuit: QuantumCircuit, coupling_map: CouplingMap) -> bool: """Check if a circuit is fully routed to the device, including directionality. A circuit is considered routed if all two-qubit gates are on qubit pairs - that exist as directed edges in the device coupling map, using the physical - qubit indices resolved via the layout (virtual → physical mapping). + that exist as directed edges in the device coupling map. + + After a layout pass the circuit's qubits are already physical qubits, so + ``circuit.find_bit(q).index`` gives the physical index directly — + consistent with how ``reward.py`` looks up gate calibrations. Args: circuit: QuantumCircuit to check. coupling_map: CouplingMap of the target device. - layout: The transpile layout mapping virtual qubits to physical qubits. Returns: True if fully routed, False otherwise. """ directed_edges = set(coupling_map.get_edges()) - - # Resolve virtual → physical using the layout. - # q._index is register-local and must NOT be used here; v2p[q] gives the - # correct physical qubit index as seen by the device. - resolved_layout = layout.final_layout or layout.initial_layout - v2p = resolved_layout.get_virtual_bits() - for instr in circuit.data: if len(instr.qubits) == 2: - q0 = v2p[instr.qubits[0]] - q1 = v2p[instr.qubits[1]] + q0 = circuit.find_bit(instr.qubits[0]).index + q1 = circuit.find_bit(instr.qubits[1]).index if (q0, q1) not in directed_edges: return False return True def determine_valid_actions_for_state(self) -> list[int]: """Determine valid actions based on circuit state: synthesized, mapped, routed.""" - check_nat_gates = GatesInBasis(basis_gates=self.device.operation_names) - check_nat_gates(self.state) - synthesized = check_nat_gates.property_set["all_gates_in_basis"] + synthesized = self.is_circuit_synthesized(self.state) laid_out = self.is_circuit_laid_out(self.state, self.layout) if self.layout else False # Routing is only allowed after layout routed = ( - self.is_circuit_routed(self.state, CouplingMap(self.device.build_coupling_map()), self.layout) - if laid_out - else False + self.is_circuit_routed(self.state, CouplingMap(self.device.build_coupling_map())) if laid_out else False ) actions = [] diff --git a/tests/compilation/test_predictor_rl.py b/tests/compilation/test_predictor_rl.py index 5aa086d3c..798da1da2 100644 --- a/tests/compilation/test_predictor_rl.py +++ b/tests/compilation/test_predictor_rl.py @@ -85,7 +85,7 @@ def test_qcompile_with_newly_trained_models() -> None: rl_compile(qc, device=device, figure_of_merit=figure_of_merit) predictor.train_model( - timesteps=100, + timesteps=1000, test=True, ) From 0d46fa9de06e40db89e9bdbb8adf98376dc83a69 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Wed, 4 Mar 2026 19:56:19 +0100 Subject: [PATCH 19/22] =?UTF-8?q?=F0=9F=90=9B=20fix=20no=20valid=20action?= =?UTF-8?q?=20bug?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 47d0c0a35..12c22690f 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -188,22 +188,21 @@ def step(self, action: int) -> tuple[dict[str, Any], float, bool, bool, dict[Any self.state: QuantumCircuit = altered_qc self.num_steps += 1 + self.state._layout = self.layout # noqa: SLF001 + self.valid_actions = self.determine_valid_actions_for_state() if len(self.valid_actions) == 0: msg = "No valid actions left." raise RuntimeError(msg) if action == self.action_terminate_index: - if action not in self.valid_actions: - msg = "Terminate action is not valid but was chosen." - raise RuntimeError(msg) + assert action in self.valid_actions, "Terminate action is not valid but was chosen." reward_val = self.calculate_reward() done = True else: reward_val = 0 done = False - self.state._layout = self.layout # noqa: SLF001 obs = create_feature_dict(self.state) return obs, reward_val, done, False, {} @@ -266,10 +265,14 @@ def action_masks(self) -> list[bool]: """Returns a list of valid actions for the current state.""" action_mask = [action in self.valid_actions for action in self.action_set] - # it is not clear how tket will handle the layout, so we remove all actions that are from "origin"=="tket" if a layout is set + # TKET layout/optimization actions must not run after a Qiskit layout has been set + # (it is not clear how tket will handle the layout). TKET routing actions are + # designed to work after a Qiskit layout via PreProcessTKETRoutingAfterQiskitLayout. if self.layout is not None: action_mask = [ - action_mask[i] and self.action_set[i].origin != CompilationOrigin.TKET for i in range(len(action_mask)) + action_mask[i] + and (self.action_set[i].origin != CompilationOrigin.TKET or i in self.actions_routing_indices) + for i in range(len(action_mask)) ] if self.has_parameterized_gates or self.layout is not None: @@ -340,7 +343,9 @@ def _apply_qiskit_action(self, action: Action, action_index: int) -> QuantumCirc ): altered_qc = self._handle_qiskit_layout_postprocessing(action, pm, altered_qc) - elif action_index in self.actions_routing_indices and self.layout: + elif ( + action_index in self.actions_routing_indices and self.layout and pm.property_set["final_layout"] is not None + ): self.layout.final_layout = pm.property_set["final_layout"] # BasisTranslator errors on unitary gates; decompose them immediately so From 7ecea7710fd112c2173d253e9cf5cb9fa102632a Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Tue, 10 Mar 2026 17:50:59 +0100 Subject: [PATCH 20/22] =?UTF-8?q?=F0=9F=8E=A8=20add=20mdp=20attribute?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 34 +++++++++++++++------------- 1 file changed, 18 insertions(+), 16 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index 12c22690f..c4ad59a83 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -68,6 +68,7 @@ class PredictorEnv(Env): def __init__( self, device: Target, + mdp: str = "paper", reward_function: figure_of_merit = "expected_fidelity", path_training_circuits: Path | None = None, ) -> None: @@ -75,6 +76,7 @@ def __init__( Arguments: device: The target device to be used for compilation. + mdp: The MDP transition policy. "paper" (default) enforces a strict, linear pipeline (synthesis -> (layout->routing) / mapping), while "flexible" allows for a cyclical approach where actions can be interleaved or reversed. reward_function: The figure of merit to be used for the reward function. Defaults to "expected_fidelity". path_training_circuits: The path to the training circuits folder. Defaults to None, which uses the default path. @@ -95,6 +97,9 @@ def __init__( self.used_actions: list[str] = [] self.device = device + logger.info("MDP: " + mdp) + self.mdp = mdp + # check for uni-directional coupling map coupling_set = {tuple(pair) for pair in self.device.build_coupling_map()} if any((b, a) not in coupling_set for (a, b) in coupling_set): @@ -510,64 +515,61 @@ def determine_valid_actions_for_state(self) -> list[int]: actions = [] - og = True # Original (restricted) MDP - flexible = False # General MDP - # Initial state if not synthesized and not laid_out and not routed: - if flexible: + if self.mdp == "flexible": actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_mapping_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) - if og: + if self.mdp == "paper": actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) if synthesized and not laid_out and not routed: - if flexible: + if self.mdp == "flexible": actions.extend(self.actions_mapping_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) - if og: + if self.mdp == "paper": actions.extend(self.actions_mapping_indices) actions.extend(self.actions_layout_indices) actions.extend(self.actions_opt_indices) # Not *depicted* in paper; necessary because optimization can destroy the native gate set if not synthesized and laid_out and not routed: - if flexible: + if self.mdp == "flexible": actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) - if og: + if self.mdp == "paper": actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) - # Not *depicted* in paper; necessary because of mapping-only passes + # Not *depicted* in paper; necessary because of layout-only passes if synthesized and laid_out and not routed: - if flexible: + if self.mdp == "flexible": actions.extend(self.actions_routing_indices) actions.extend(self.actions_opt_indices) - if og: + if self.mdp == "paper": actions.extend(self.actions_routing_indices) # Not *depicted* in paper; necessary because routing can insert non-native SWAPs if not synthesized and laid_out and routed: - if flexible: + if self.mdp == "flexible": actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) - if og: + if self.mdp == "paper": actions.extend(self.actions_synthesis_indices) actions.extend(self.actions_opt_indices) # Final state if synthesized and laid_out and routed: - if flexible: + if self.mdp == "flexible": actions.extend([self.action_terminate_index]) actions.extend(self.actions_opt_indices) - if og: + if self.mdp == "paper": actions.extend([self.action_terminate_index]) actions.extend(self.actions_opt_indices) From 1d26bc79e7315dfb45fd48903fd08b063a0325af Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Wed, 11 Mar 2026 10:56:53 +0100 Subject: [PATCH 21/22] =?UTF-8?q?=F0=9F=90=9B=20update=20reset=20with=20fl?= =?UTF-8?q?exible=20policy?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index c4ad59a83..f823053a4 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -258,7 +258,15 @@ def reset( self.layout = None - self.valid_actions = self.actions_opt_indices + self.actions_synthesis_indices + if self.mdp == "flexible": + self.valid_actions = ( + self.actions_synthesis_indices + + self.actions_mapping_indices + + self.actions_layout_indices + + self.actions_opt_indices + ) + else: + self.valid_actions = self.actions_synthesis_indices + self.actions_opt_indices self.error_occurred = False From 07e73387de13182d8d42d238007e5e73daec2667 Mon Sep 17 00:00:00 2001 From: flowerthrower Date: Wed, 11 Mar 2026 11:05:18 +0100 Subject: [PATCH 22/22] =?UTF-8?q?=F0=9F=90=9B=20check=20cirucit=20qubits?= =?UTF-8?q?=20layout=20not=20instructions?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mqt/predictor/rl/predictorenv.py | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/src/mqt/predictor/rl/predictorenv.py b/src/mqt/predictor/rl/predictorenv.py index f823053a4..c560b9e99 100644 --- a/src/mqt/predictor/rl/predictorenv.py +++ b/src/mqt/predictor/rl/predictorenv.py @@ -461,12 +461,7 @@ def is_circuit_laid_out(self, circuit: QuantumCircuit, layout: TranspileLayout | layout = layout.final_layout or layout.initial_layout v2p = layout.get_virtual_bits() - for instr in circuit.data: - for q in instr.qubits: - if q not in v2p: - # Logical qubit not assigned - return False - return True + return all(q in v2p for q in circuit.qubits) def is_circuit_synthesized(self, circuit: QuantumCircuit) -> bool: """Check if the circuit uses only native gates of the device.