✨ Add intermediate rewards with paper semantics

CHANGELOG.md

@@ -34,6 +34,8 @@ releases may include breaking changes.
 
 ### Changed
 
+- ✨ Enable configurable intermediate rewards for RL training by default
+  ([#799]) ([**@flowerthrower**])
 - 🐛 Restore nondeterministic RL circuit sampling without a seed and make
   explicitly seeded sampling and randomized Qiskit actions reproducible ([#797])
   ([**@flowerthrower**])
@@ -116,6 +118,7 @@ for previous changelogs._
 [#796]: https://github.com/munich-quantum-toolkit/predictor/pull/796
 [#795]: https://github.com/munich-quantum-toolkit/predictor/pull/795
 [#794]: https://github.com/munich-quantum-toolkit/predictor/pull/794
+[#799]: https://github.com/munich-quantum-toolkit/predictor/pull/799
 [#758]: https://github.com/munich-quantum-toolkit/predictor/pull/758
 [#755]: https://github.com/munich-quantum-toolkit/predictor/pull/755
 [#731]: https://github.com/munich-quantum-toolkit/predictor/pull/731

UPGRADING.md

@@ -6,6 +6,15 @@ of changes including minor and patch releases, please refer to the
 
 ## [Unreleased]
 
+### Intermediate RL rewards
+
+`PredictorEnv` now enables intermediate rewards by default. For
+`expected_fidelity` and `estimated_success_probability`, comparable non-terminal
+steps are rewarded from changes in the exact or approximate figure of merit.
+Optimization actions with no measurable change receive `-0.001`. Set
+`intermediate_reward=False` to retain the previous terminal-only reward
+behavior; use `reward_scale` and `no_effect_penalty` to tune reward shaping.
+
 ### Expanded TKET action set
 
 The RL action space now includes TKET's `KAKDecomposition` optimization action

src/mqt/predictor/reward.py

@@ -15,13 +15,15 @@
 
 import numpy as np
 from joblib import load
-from qiskit import __version__ as qiskit_version
-from qiskit import transpile
+from qiskit.transpiler import InstructionDurations, PassManager
+from qiskit.transpiler.passes import ASAPScheduleAnalysis
 
 from mqt.predictor.hellinger import calc_device_specific_features, get_hellinger_model_path
 from mqt.predictor.utils import calc_supermarq_features
 
 if TYPE_CHECKING:
+    from collections.abc import Iterable
+
     from qiskit import QuantumCircuit
     from qiskit.transpiler import Target
     from sklearn.ensemble import RandomForestRegressor
@@ -91,124 +93,75 @@ def estimated_success_probability(qc: QuantumCircuit, device: Target, precision:
     Returns:
         The expected success probability of the given quantum circuit on the given device.
     """
-    exec_time_per_qubit = dict.fromkeys(range(device.num_qubits), 0.0)
-
-    op_times, active_qubits = [], set()
+    operation_times: list[tuple[str, Iterable[int] | None, float, str]] = []
     for instr in qc.data:
-        instruction = instr.operation
-        qargs = instr.qubits
-        gate_type = instruction.name
-
-        if gate_type == "barrier" or gate_type == "id":
+        gate_type = str(instr.operation.name)
+        if gate_type in {"barrier", "id"}:
             continue
-        assert len(qargs) in (1, 2)
-        first_qubit_idx = qc.find_bit(qargs[0]).index
-        active_qubits.add(first_qubit_idx)
-
-        if len(qargs) == 1:  # single-qubit gate
-            duration = device[gate_type][first_qubit_idx,].duration
-            op_times.append((
-                gate_type,
-                [
-                    first_qubit_idx,
-                ],
-                duration,
-                "s",
-            ))
-            exec_time_per_qubit[first_qubit_idx] += duration
-        else:  # multi-qubit gate
-            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"))
-            exec_time_per_qubit[first_qubit_idx] += duration
-            exec_time_per_qubit[second_qubit_idx] += duration
-
-    if qiskit_version < "2.0.0":
-        from qiskit.transpiler import (  # ruff:ignore[import-outside-top-level]
-            InstructionDurations,
-            Layout,
-            PassManager,
-            passes,
-        )
-        from qiskit.transpiler.passes import ApplyLayout, SetLayout  # ruff:ignore[import-outside-top-level]
-
-        if qc.qregs[0].name != "q":
-            # create a layout that maps the (tket) 'node' registers to the (qiskit) 'q' registers
-            layouts = [
-                SetLayout(Layout({node_qubit: i for i, node_qubit in enumerate(node_reg)})) for node_reg in qc.qregs
-            ]
-            # create a pass manager with the SetLayout and ApplyLayout passes
-            pm = PassManager(list(layouts))
-            pm.append(ApplyLayout())
-
-            # replace the 'node' register with the 'q' register in the circuit
-            qc = pm.run(qc)
-            assert qc.qregs[0].name == "q"
-
-        sched_pass = passes.ASAPScheduleAnalysis(InstructionDurations(op_times))
-        delay_pass = passes.PadDelay()
-        pm = PassManager([sched_pass, delay_pass])
-        scheduled_circ = pm.run(qc)
-
-    else:
-        scheduled_circ = transpile(
-            qc,
-            target=device,
-            scheduling_method="asap",
-            optimization_level=0,
-            initial_layout=None,
-            routing_method=None,
-            layout_method=None,
-        )
-        overall_estimated_duration = scheduled_circ.estimate_duration(target=device)
+        qubit_indices = [int(qc.find_bit(qubit).index) for qubit in instr.qubits]
+        properties = device[gate_type].get(tuple(qubit_indices))
+        if properties is None or properties.duration is None:
+            msg = f"Duration for gate {gate_type} on qubits {tuple(qubit_indices)} not found in device properties."
+            raise ValueError(msg)
+        operation_times.append((gate_type, qubit_indices, float(properties.duration), "s"))
+
+    durations = InstructionDurations(operation_times, dt=device.dt)
+    pass_manager = PassManager([ASAPScheduleAnalysis(durations=durations)])
+    pass_manager.run(qc)
+
+    time_unit = pass_manager.property_set["time_unit"]
+    execution_time_per_qubit = dict.fromkeys(range(device.num_qubits), 0.0)
+    last_end_per_qubit = dict.fromkeys(range(device.num_qubits), 0.0)
+    last_operation_per_qubit = dict.fromkeys(range(device.num_qubits), "")
+    circuit_duration = 0.0
+
+    for node, start_time in pass_manager.property_set["node_start_time"].items():
+        qubit_indices = [qc.find_bit(qubit).index for qubit in node.qargs]
+        duration = float(durations.get(node.name, qubit_indices, unit=time_unit))
+        end_time = float(start_time) + duration
+        circuit_duration = max(circuit_duration, end_time)
+        for qubit in qubit_indices:
+            execution_time_per_qubit[qubit] += duration
+            if end_time >= last_end_per_qubit[qubit]:
+                last_end_per_qubit[qubit] = end_time
+                last_operation_per_qubit[qubit] = node.name
 
     res = 1.0
-    for instr in scheduled_circ.data:
+    active_qubits = set()
+    for instr in qc.data:
         instruction = instr.operation
         qargs = instr.qubits
         gate_type = instruction.name
 
-        if gate_type == "barrier" or gate_type == "id":
+        if gate_type in {"barrier", "id"}:
             continue
 
         assert len(qargs) in (1, 2)
-        first_qubit_idx = scheduled_circ.find_bit(qargs[0]).index
+        qubit_indices = [qc.find_bit(qubit).index for qubit in qargs]
+        active_qubits.update(qubit_indices)
+        first_qubit_idx = qubit_indices[0]
 
         if len(qargs) == 1:
-            if gate_type == "measure":
-                res *= 1 - device[gate_type][first_qubit_idx,].error
-                continue
-            if gate_type == "delay":
-                if qiskit_version < "2.0.0":
-                    continue
-                # only consider active qubits
-                if first_qubit_idx not in active_qubits:
-                    continue
-
-                dt = device.dt  # instruction durations are stored in unit dt
-                res *= np.exp(
-                    -instruction.duration
-                    * dt
-                    / min(device.qubit_properties[first_qubit_idx].t1, device.qubit_properties[first_qubit_idx].t2)
-                )
-                continue
             res *= 1 - device[gate_type][first_qubit_idx,].error
         else:
-            second_qubit_idx = scheduled_circ.find_bit(qargs[1]).index
+            second_qubit_idx = qubit_indices[1]
             try:
                 res *= 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
 
-    if qiskit_version >= "2.0.0":
-        for i in range(device.num_qubits):
-            qubit_execution_time = exec_time_per_qubit[i]
-            if qubit_execution_time == 0:
-                continue
-            idle_time = overall_estimated_duration - qubit_execution_time
-            res *= np.exp(-idle_time / min(device.qubit_properties[i].t1, device.qubit_properties[i].t2))
+    assert device.qubit_properties is not None
+    for qubit in active_qubits:
+        properties = device.qubit_properties[qubit]
+        assert properties is not None
+        assert properties.t1 is not None
+        assert properties.t2 is not None
+        live_end = (
+            last_end_per_qubit[qubit] if last_operation_per_qubit[qubit] in {"measure", "reset"} else circuit_duration
+        )
+        idle_time = max(live_end - execution_time_per_qubit[qubit], 0.0)
+        res *= np.exp(-idle_time / min(properties.t1, properties.t2))
     return float(np.round(res, precision).item())