From 40f949740abbcad70caf8be6b49cadb15a8621a7 Mon Sep 17 00:00:00 2001 From: Vivek1106-04 Date: Mon, 9 Mar 2026 11:42:54 +0530 Subject: [PATCH] Add support for handling CircuitOperations recursively --- .../routing/line_initial_mapper.py | 4 +- .../transformers/routing/route_circuit_cqc.py | 113 +++++++++++++++++- .../routing/route_circuit_cqc_test.py | 65 ++++++++++ 3 files changed, 177 insertions(+), 5 deletions(-) diff --git a/cirq-core/cirq/transformers/routing/line_initial_mapper.py b/cirq-core/cirq/transformers/routing/line_initial_mapper.py index 044a66fc4aa..8e5cc9433e6 100644 --- a/cirq-core/cirq/transformers/routing/line_initial_mapper.py +++ b/cirq-core/cirq/transformers/routing/line_initial_mapper.py @@ -37,7 +37,7 @@ import networkx as nx -from cirq import protocols, value +from cirq import circuits, protocols, value from cirq.transformers.routing import initial_mapper if TYPE_CHECKING: @@ -107,6 +107,8 @@ def degree_lt_two(q: cirq.Qid): return any(circuit_graph[component_id[q]][i] == q for i in [-1, 0]) for op in circuit.all_operations(): + if isinstance(op.untagged, circuits.CircuitOperation): + continue if protocols.num_qubits(op) != 2: continue diff --git a/cirq-core/cirq/transformers/routing/route_circuit_cqc.py b/cirq-core/cirq/transformers/routing/route_circuit_cqc.py index 4970db81f45..e6ac5ca678e 100644 --- a/cirq-core/cirq/transformers/routing/route_circuit_cqc.py +++ b/cirq-core/cirq/transformers/routing/route_circuit_cqc.py @@ -121,6 +121,7 @@ def __call__( lookahead_radius: int = 8, tag_inserted_swaps: bool = False, initial_mapper: cirq.AbstractInitialMapper | None = None, + min_qubit_mapping_threshold: float = 0.5, context: cirq.TransformerContext | None = None, ) -> cirq.AbstractCircuit: """Transforms the given circuit to make it executable on the device. @@ -137,6 +138,10 @@ def __call__( initial_mapper: an initial mapping strategy (placement) of logical qubits in the circuit onto physical qubits on the device. If not provided, defaults to an instance of `cirq.LineInitialMapper`. + min_qubit_mapping_threshold: the minimum fraction (0.0 to 1.0) of qubits that should + have their initial mapping computed from outer (non-CircuitOperation) 2-qubit gates + before proceeding with routing. If there are not enough outer 2-qubit gates, + CircuitOperations will be partially unrolled to reach this threshold. context: transformer context storing common configurable options for transformers. Returns: @@ -152,6 +157,7 @@ def __call__( lookahead_radius=lookahead_radius, tag_inserted_swaps=tag_inserted_swaps, initial_mapper=initial_mapper, + min_qubit_mapping_threshold=min_qubit_mapping_threshold, context=context, ) return routed_circuit @@ -163,15 +169,15 @@ def route_circuit( lookahead_radius: int = 8, tag_inserted_swaps: bool = False, initial_mapper: cirq.AbstractInitialMapper | None = None, + min_qubit_mapping_threshold: float = 0.5, context: cirq.TransformerContext | None = None, ) -> tuple[cirq.AbstractCircuit, dict[cirq.Qid, cirq.Qid], dict[cirq.Qid, cirq.Qid]]: """Transforms the given circuit to make it executable on the device. This transformer assumes that all multi-qubit operations have been decomposed into 2-qubit operations and will raise an error if `circuit` a n-qubit operation where n > 2. If - `circuit` contains `cirq.CircuitOperation`s and `context.deep` is True then they are first - unrolled before proceeding. If `context.deep` is False or `context` is None then any - `cirq.CircuitOperation` that acts on more than 2-qubits will also raise an error. + `circuit` contains `cirq.CircuitOperation`s and `min_qubit_mapping_threshold` < 1.0, + they are handled using a recursive routing strategy instead of being fully unrolled. The algorithm tries to find the best swap at each timestep by ranking a set of candidate swaps against operations starting from the current timestep (say s) to the timestep at index @@ -191,6 +197,11 @@ def route_circuit( operations. initial_mapper: an initial mapping strategy (placement) of logical qubits in the circuit onto physical qubits on the device. + min_qubit_mapping_threshold: the minimum fraction (0.0 to 1.0) of qubits that should + have their initial mapping computed from outer (non-CircuitOperation) 2-qubit gates + before proceeding with routing. If there are not enough outer 2-qubit gates, + CircuitOperations will be partially unrolled to reach this threshold. A value of 1.0 + disables recursive routing and falls back to unrolling all CircuitOperations. context: transformer context storing common configurable options for transformers. Returns: @@ -206,7 +217,20 @@ def route_circuit( ValueError: if circuit has operations that act on 3 or more qubits, except measurements. """ - # 0. Handle CircuitOperations by unrolling them. + # 0. Handle CircuitOperations - use recursive routing if threshold < 1.0 + has_circuit_ops = self._has_circuit_operations(circuit) + use_recursive_routing = has_circuit_ops and min_qubit_mapping_threshold < 1.0 + + if use_recursive_routing: + return self._route_circuit_recursive( + circuit=circuit, + min_qubit_mapping_threshold=min_qubit_mapping_threshold, + lookahead_radius=lookahead_radius, + tag_inserted_swaps=tag_inserted_swaps, + initial_mapper=initial_mapper, + ) + + # Legacy behavior: unroll CircuitOperations if deep=True if context is not None and context.deep is True: circuit = transformer_primitives.unroll_circuit_op(circuit, deep=True) if any( @@ -563,6 +587,87 @@ def _cost( mm.apply_swap(*swap) return max_length, sum_length + def _has_circuit_operations(self, circuit: cirq.AbstractCircuit) -> bool: + """Check if the circuit contains any CircuitOperations.""" + return any( + isinstance(op.untagged, circuits.CircuitOperation) for op in circuit.all_operations() + ) + + def _get_ops_outside_circuit_ops( + self, circuit: cirq.AbstractCircuit + ) -> tuple[list[list[cirq.Operation]], list[list[cirq.Operation]]]: + """Get 2-qubit and single-qubit ops that are NOT inside CircuitOperations.""" + outer_circuit = circuits.Circuit() + for moment in circuit: + outer_moment = circuits.Moment( + op for op in moment if not isinstance(op.untagged, circuits.CircuitOperation) + ) + outer_circuit.append(outer_moment) + return self._get_one_and_two_qubit_ops_as_timesteps(outer_circuit) + + def _route_circuit_recursive( + self, + circuit: cirq.AbstractCircuit, + min_qubit_mapping_threshold: float, + lookahead_radius: int, + tag_inserted_swaps: bool, + initial_mapper: cirq.AbstractInitialMapper | None, + ) -> tuple[cirq.AbstractCircuit, dict[cirq.Qid, cirq.Qid], dict[cirq.Qid, cirq.Qid]]: + """Route a circuit containing CircuitOperations using recursive strategy.""" + if initial_mapper is None: + initial_mapper = line_initial_mapper.LineInitialMapper(self.device_graph) + + num_total_qubits = len(list(circuit.all_qubits())) + outer_two_qubit_ops, outer_single_qubit_ops = self._get_ops_outside_circuit_ops(circuit) + outer_qubits = {q for ops in outer_two_qubit_ops for op in ops for q in op.qubits} + + if len(outer_qubits) / num_total_qubits >= min_qubit_mapping_threshold: + outer_for_map = circuits.Circuit(op for ops in outer_two_qubit_ops for op in ops) + initial_mapping = initial_mapper.initial_mapping(outer_for_map) + else: + initial_mapping = initial_mapper.initial_mapping(circuit) + + mm = mapping_manager.MappingManager(self.device_graph, initial_mapping) + + circuit_ops = [ + (i, op, op.untagged) + for i, m in enumerate(circuit) + for op in m + if isinstance(op.untagged, circuits.CircuitOperation) + ] + + routed_ops, routing_swaps = self._route( + mm, + outer_two_qubit_ops, + outer_single_qubit_ops, + lookahead_radius, + tag_inserted_swaps=tag_inserted_swaps, + ) + + routed_circuit = circuits.Circuit(circuits.Circuit(m) for m in routed_ops) + + for _, _, circuit_op in circuit_ops: + inner = circuit_op.circuit.unfreeze(copy=True) + inner_routed, inner_init, _ = self.route_circuit( + inner, + lookahead_radius=lookahead_radius, + tag_inserted_swaps=tag_inserted_swaps, + initial_mapper=initial_mapper, + min_qubit_mapping_threshold=1.0, + ) + routed_circuit.append(circuits.Circuit(inner_routed).transform_qubits(inner_init)) + + if routing_swaps and nx.is_directed(self.device_graph): + routed_circuit = circuits.Circuit( + self._replace_swaps_with_directional_decomposition(routed_circuit, routing_swaps) + ) + + final_mapping = { + mm.int_to_logical_qid[k]: mm.int_to_physical_qid[v] + for k, v in enumerate(mm.logical_to_physical) + } + return routed_circuit, initial_mapping, final_mapping + def __eq__(self, other) -> bool: return nx.utils.graphs_equal(self.device_graph, other.device_graph) diff --git a/cirq-core/cirq/transformers/routing/route_circuit_cqc_test.py b/cirq-core/cirq/transformers/routing/route_circuit_cqc_test.py index bd938b3e071..b2ac83f4514 100644 --- a/cirq-core/cirq/transformers/routing/route_circuit_cqc_test.py +++ b/cirq-core/cirq/transformers/routing/route_circuit_cqc_test.py @@ -414,3 +414,68 @@ def test_repr() -> None: device_graph = device.metadata.nx_graph router = cirq.RouteCQC(device_graph) cirq.testing.assert_equivalent_repr(router, setup_code='import cirq\nimport networkx as nx') + + +@pytest.mark.parametrize( + "test_type, threshold, n_qubits", + [ + ("single_op", 0.1, 4), + ("single_op", 0.25, 4), + ("single_op", 0.5, 4), + ("single_op", 0.75, 4), + ("multiple_ops", 0.1, 3), + ("multiple_ops", 0.1, 4), + ("multiple_ops", 0.1, 5), + ("threshold_behavior", 0.25, 4), + ("threshold_behavior", 0.5, 4), + ("threshold_behavior", 0.75, 4), + ], +) +def test_circuit_operations_recursive_routing(test_type, threshold, n_qubits) -> None: + """Test recursive routing of circuits containing CircuitOperations.""" + device = cirq.testing.construct_grid_device(4, 4) + router = cirq.RouteCQC(device.metadata.nx_graph) + q = cirq.LineQubit.range(n_qubits) + + if test_type == "single_op": + inner_circuit = cirq.Circuit(cirq.CNOT(q[0], q[1]), cirq.CNOT(q[1], q[2])) + outer_circuit = cirq.Circuit( + cirq.CircuitOperation(inner_circuit.freeze()), cirq.CNOT(q[0], q[1]) + ) + elif test_type == "multiple_ops": + inner1 = cirq.Circuit(cirq.CNOT(q[0], q[1]), cirq.CZ(q[1], q[2])) + inner2 = cirq.Circuit(cirq.CNOT(q[-2], q[-1]), cirq.CZ(q[0], q[1])) + outer_circuit = cirq.Circuit( + cirq.CircuitOperation(inner1.freeze()), + cirq.CNOT(q[0], q[n_qubits // 2]), + cirq.CircuitOperation(inner2.freeze()), + ) + elif test_type == "threshold_behavior": + inner_circuit = cirq.Circuit( + cirq.CNOT(q[0], q[1]), cirq.CNOT(q[1], q[2]), cirq.CNOT(q[2], q[3]) + ) + outer_circuit = cirq.Circuit(cirq.H(q[0]), cirq.CircuitOperation(inner_circuit.freeze())) + + routed, _, _ = router.route_circuit(outer_circuit, min_qubit_mapping_threshold=threshold) + device.validate_circuit(routed) + assert len(list(routed.all_operations())) > 0 + + +def test_directed_device_recursive_routing() -> None: + # Use a directed ring (strongly connected) so LineInitialMapper works + device = cirq.testing.construct_ring_device(4, directed=True) + device_graph = device.metadata.nx_graph + router = cirq.RouteCQC(device_graph) + + q = cirq.LineQubit.range(3) + # Inner circuit with adjacent gates; outer circuit forces a swap + inner_circuit = cirq.Circuit(cirq.CNOT(q[0], q[1])) + outer_circuit = cirq.Circuit( + cirq.CNOT(q[0], q[2]), # non-adjacent: forces a swap + cirq.CircuitOperation(inner_circuit.freeze()), + ) + + routed, _, _ = router.route_circuit( + outer_circuit, min_qubit_mapping_threshold=0.5, tag_inserted_swaps=True + ) + assert len(list(routed.all_operations())) > 0