伝播型ノイズ吸収（PNA）による期待値の改善

from qiskit import QuantumCircuit
from qiskit.quantum_info import Pauli, SparsePauliOp
from qiskit.transpiler import generate_preset_pass_manager, PassManager
from qiskit_ibm_runtime import (
    QiskitRuntimeService,
    QuantumProgram,
    Executor,
    NoiseLearnerV3,
)
from qiskit_addon_utils.exp_vals.measurement_bases import (
    get_measurement_bases,
)
from qiskit_addon_utils.exp_vals.expectation_values import (
    executor_expectation_values,
)
from qiskit_addon_utils.noise_management import trex_factors
from qiskit_addon_utils.noise_management.post_selection import PostSelector
from qiskit_addon_utils.noise_management.post_selection.transpiler.passes import (
    AddPostSelectionMeasures,
    AddSpectatorMeasures,
)
from qiskit_addon_pna import generate_noise_mitigating_observable
import samplomatic
from samplomatic.transpiler import generate_boxing_pass_manager
from samplomatic.annotations import InjectNoise
from samplomatic.utils import get_annotation, find_unique_box_instructions
import numpy as np
import matplotlib.pyplot as plt


# Selects a connected chain of low-error qubits on the target backend.
# The mirrored kicked-Ising circuit is a 1D chain, so we only need a line
# of connected physical qubits; the helper walks the backend's coupling map
# and grows a chain along the lowest-error two-qubit edges, so it works for
# any backend rather than relying on a hardcoded layout.
def find_qubit_chain(backend, length):
    """Find a connected chain of ``length`` physical qubits on ``backend``.

    The chain is grown greedily along the lowest-error two-qubit edges, so it
    favors better-performing qubits. Because the mirrored kicked-Ising circuit
    is a 1D chain, a connected line is all we need.
    """
    target = backend.target

    # Identify the native two-qubit gate and build a per-edge error lookup.
    two_qubit_gate = next(
        name
        for name in target.operation_names
        if target[name]
        and all(q is not None and len(q) == 2 for q in target[name])
    )
    edge_error = {
        frozenset(qargs): (
            1.0 if props is None or props.error is None else props.error
        )
        for qargs, props in target[two_qubit_gate].items()
    }

    graph = backend.coupling_map.graph.to_undirected(multigraph=False)
    neighbors = {n: list(graph.neighbors(n)) for n in graph.node_indices()}

    def first_chain_from(start):
        path, visited = [start], {start}

        def grow():
            if len(path) == length:
                return True
            node = path[-1]
            order = sorted(
                neighbors[node],
                key=lambda m: edge_error.get(frozenset((node, m)), 1.0),
            )
            for nxt in order:
                if nxt not in visited:
                    visited.add(nxt)
                    path.append(nxt)
                    if grow():
                        return True
                    path.pop()
                    visited.remove(nxt)
            return False

        return path if grow() else None

    def chain_cost(path):
        return sum(
            edge_error.get(frozenset((path[i], path[i + 1])), 1.0)
            for i in range(len(path) - 1)
        )

    # Try low-degree qubits first (the natural ends of long chains) and keep
    # the lowest-error chain found.
    best_path, best_cost = None, float("inf")
    for start in sorted(neighbors, key=lambda n: len(neighbors[n])):
        chain = first_chain_from(start)
        if chain is not None and (cost := chain_cost(chain)) < best_cost:
            best_path, best_cost = chain, cost

    if best_path is None:
        raise ValueError(
            f"Could not find a connected chain of {length} qubits "
            f"on '{backend.name}'."
        )
    return best_path

num_qubits = 30
num_trotter_steps = 10
rx_angle = np.pi / 8

# Avg single-site magnetization
id_pauli = Pauli("I" * num_qubits)
observable = (
    SparsePauliOp([id_pauli.dot(Pauli("Z"), [i]) for i in range(num_qubits)])
    / num_qubits
)

# Implement Trotterized kicked-Ising model
circuit = QuantumCircuit(num_qubits)
for _step in range(num_trotter_steps):
    circuit.rx(rx_angle, range(num_qubits))
    for first_qubit in (1, 2):
        for idx in range(first_qubit, num_qubits, 2):
            # equivalent to Rzz(-pi/2):
            circuit.sdg([idx - 1, idx])
            circuit.cz(idx - 1, idx)
# Append the inverse circuit to complete the mirroring
circuit.compose(circuit.inverse(), inplace=True)
circuit.measure_active()
circuit.draw("mpl", fold=-1)

# Initialize IBM Quantum Compute Service using your saved credentials
service = QiskitRuntimeService()

backend = service.least_busy(
    operational=True, simulator=False, min_num_qubits=num_qubits
)
# Re-fetch with fractional gates enabled (least_busy does not forward this)
# Fractional gates are enabled so the non-Clifford Rx rotations are supported natively.
backend = service.backend(backend.name, use_fractional_gates=True)
print(f"Selected backend: {backend.name}")

# Find a connected, low-error chain of qubits on the chosen backend
layout = find_qubit_chain(backend, num_qubits)

# Transpile the circuit for the target backend, pinning the chain as the layout.
# optimization_level=0 preserves the mirrored two-qubit-gate layers that the
# boxing and noise-learning steps rely on.
pm = generate_preset_pass_manager(
    backend=backend, optimization_level=0, initial_layout=layout
)
isa_circuit = pm.run(circuit)
isa_observable = observable.apply_layout(isa_circuit.layout)
isa_circuit.draw("mpl", fold=-1)

# Box up circuit with Twirl and InjectNoise annotations
pm = generate_boxing_pass_manager(
    enable_gates=True,
    enable_measures=True,
    measure_annotations="all",
    twirling_strategy="active",
    inject_noise_targets="gates",
    inject_noise_strategy="uniform_modification",
)
boxed_circuit = pm.run(isa_circuit)

draw_circ = QuantumCircuit(boxed_circuit.num_qubits)
draw_circ.append(boxed_circuit.data[0], qargs=boxed_circuit.data[0].qubits)
draw_circ.append(boxed_circuit.data[1], qargs=boxed_circuit.data[1].qubits)
draw_circ.draw("mpl", fold=-1, scale=0.3, idle_wires=False)

# Build template circuit and samplex for later use with the "Executor"
template_circuit, samplex = samplomatic.build(boxed_circuit)

# Add post-selection instructions to the template circuit
post_selection_pm = PassManager(
    [
        AddSpectatorMeasures(backend.coupling_map),
        AddPostSelectionMeasures(x_pulse_type="rx"),
    ]
)
template_circuit = post_selection_pm.run(template_circuit)

draw_circ = template_circuit.copy_empty_like()
draw_circ.data = template_circuit.data[:324]
draw_circ.draw("mpl", fold=-1, scale=0.3, idle_wires=False)

# Noise learning parameters
num_randomizations_nl = 64
shots_per_randomization_nl = 128

# Match the deepest noise-learning depth to the depth of the circuit being
# mitigated. ``layer_pair_depths`` are measured in layer pairs (a layer + its
# inverse), so the deepest value is half the circuits's two-qubit-layer depth.
# Learning to this depth improves the quality of the mitigation.

depth_2q = isa_circuit.depth(lambda instr: instr.operation.num_qubits == 2)
max_layer_pair_depth = depth_2q // 2

# Use a fixed schedule of learning depths, but drop any that exceed the circuit's
# depth and always cap the deepest value at ``max_layer_pair_depth`` so we never
# learn deeper than the circuit being mitigated
candidate_depths = [1, 2, 4, 8, 12, 16, 24, 32, 40, 48]
layer_pair_depths = sorted(
    {d for d in candidate_depths if d < max_layer_pair_depth}
    | {max_layer_pair_depth}
)

# Find the unique instructions (layers) from the boxed-up circuit
unique_2q_layers_and_meas = find_unique_box_instructions(
    boxed_circuit, normalize_annotations=None, undress_boxes=True
)

# Configure and run the noise learner on the unique layers.
# Options can be passed directly as a dictionary.
noise_learner_options = {
    "num_randomizations": num_randomizations_nl,
    "shots_per_randomization": shots_per_randomization_nl,
    "layer_pair_depths": layer_pair_depths,
    "post_selection": {
        "enable": True,
        "strategy": "edge",
        "x_pulse_type": "rx",
    },
    "environment": {"job_tags": ["TUT_PNA"]},
}

noise_learner = NoiseLearnerV3(backend, noise_learner_options)
noise_learner_job = noise_learner.run(unique_2q_layers_and_meas)
noise_learner_result = noise_learner_job.result()

hw_rates_1q = []
hw_rates_2q = []
for nlr in noise_learner_result[:2]:
    plm_list = nlr.to_pauli_lindblad_map().to_sparse_list()
    hw_rates_1q += [
        rate for (pstr, qubits, rate) in plm_list if len(pstr) == 1
    ]
    hw_rates_2q += [
        rate for (pstr, qubits, rate) in plm_list if len(pstr) == 2
    ]
hw_rates_1q = sorted(hw_rates_1q)
hw_rates_2q = sorted(hw_rates_2q)
median_1q = hw_rates_1q[len(hw_rates_1q) // 2]
median_2q = hw_rates_2q[len(hw_rates_2q) // 2]
fig, ax = plt.subplots(1, 1, figsize=(14, 5))
ax.scatter(
    (hw_rates_1q),
    [(i) / (len(hw_rates_1q) - 1) for i in range(len(hw_rates_1q))],
    color="red",
    label="1q rates",
)
ax.set_xscale("log")
ax.set_ylim(0, 1.1)
ax.vlines(median_1q, 0, 1, color="red")
ax.text(median_1q * 1.1, 0.1, f"{median_1q:.2e}")
ax.scatter(
    (hw_rates_2q),
    [(i) / (len(hw_rates_2q) - 1) for i in range(len(hw_rates_2q))],
    color="blue",
    label="2q rates",
)
ax.set_xscale("log")
ax.set_ylim(0, 1.1)
ax.vlines(median_2q, 0, 1, color="blue")
ax.text(median_2q * 1.1, 0.2, f"{median_2q:.2e}")
ax.set_title("Learned noise rates")
ax.set_xlabel("Noise rate")
ax.set_yticks([])
plt.legend()

# map inject noise refs to pauli lindblad maps
refs_to_noise_models = {}
for instruction, result in zip(
    unique_2q_layers_and_meas, noise_learner_result, strict=False
):
    if inject_noise_annot := get_annotation(
        instruction.operation, InjectNoise
    ):
        refs_to_noise_models[inject_noise_annot.ref] = (
            result.to_pauli_lindblad_map()
        )

# PNA parameters
num_processes = 8
max_err_terms = 10_000
max_obs_terms = 10_000
num_to_measure = num_qubits

obs_tilde_isa = generate_noise_mitigating_observable(
    boxed_circuit,
    isa_observable,
    refs_to_noise_models,
    max_err_terms=max_err_terms,
    max_obs_terms=max_obs_terms,
    num_processes=num_processes,
    print_progress=True,
    search_step=8,
)
p_2_v = {p: v for v, p in enumerate(layout)}
obs_tilde_virtual = SparsePauliOp.from_sparse_list(
    [
        (pstr, [p_2_v[p] for p in p_qubits], coeff)
        for (pstr, p_qubits, coeff) in obs_tilde_isa.to_sparse_list()
    ],
    num_qubits=num_qubits,
)
obs_tilde_virtual = obs_tilde_virtual[
    np.argsort(np.abs(obs_tilde_virtual.coeffs))[::-1]
][:num_to_measure]

obs_tilde_isa = obs_tilde_isa[np.argsort(np.abs(obs_tilde_isa.coeffs))][::-1]
plt.xscale("log")
plt.yscale("log")
plt.title(r"$\tilde{O}$ coeff magnitudes")
plt.ylabel("Magnitude")
plt.xlabel("Pauli term index")
plt.plot(np.abs(obs_tilde_isa.coeffs), ".")

meas_box = boxed_circuit.data[-1]
canonical_qubits = [
    idx
    for idx, qubit in enumerate(boxed_circuit.qubits)
    if qubit in meas_box.qubits
]
c_2_p = {
    c: p for c, p in enumerate(canonical_qubits)
}  # canonical -> physical
p_2_v = {p: v for v, p in enumerate(layout)}  # physical -> virtual
c_2_v = {c: p_2_v[p] for c, p in c_2_p.items()}  # canonical -> virtual
meas_bases, bases_reverser = get_measurement_bases(obs_tilde_virtual)
meas_bases_canonical = [
    np.array([base[c_2_v[c]] for c in range(num_qubits)], dtype=np.uint8)
    for base in meas_bases
]

# Control the # of shots during execution
shots_per_randomization_exec = 64
num_randomizations_exec = 6144

# Zero out the noise to prevent noise from being injected during execution.
# We only added InjectNoise annotations so PNA could associate the noise
# to layers in the circuit
samplex_inputs = {f"noise_scales.{ref}": 0.0 for ref in refs_to_noise_models}
samplex_inputs |= {"pauli_lindblad_maps": refs_to_noise_models}

# Specify the bases to measure. The samplex exposes one basis-change input per
# ChangeBasis-annotated box; here a single all-Z basis covers every term. We
# look up the basis-change interface name rather than hardcoding an index, since
# the name depends on the circuit's box structure.
bases_broadcastable = np.expand_dims(np.array(meas_bases_canonical), axis=1)
samplex_inputs |= {
    spec.name: bases_broadcastable
    for spec in samplex.inputs().get_specs(r"^basis_changes\.")
}

# Convert samplex_inputs into a dict to pass to QuantumProgram
samplex_arguments = (
    samplex.inputs().make_broadcastable().bind(**samplex_inputs)
)

# Instantiate the QuantumProgram with the specified parameters
program = QuantumProgram(shots=shots_per_randomization_exec)
program.append_samplex_item(
    circuit=template_circuit,
    samplex=samplex,
    samplex_arguments=samplex_arguments,
    shape=(num_randomizations_exec,),
)

# Execute (sample) the circuit
executor = Executor(backend)
job_exec = executor.run(program)
exec_results = job_exec.result()

# Computing the TREX factors
measurement_noise_map = noise_learner_result[2].to_pauli_lindblad_map()
trex_rescale_factors = trex_factors(measurement_noise_map, bases_reverser)

# Post-select the results
post_selector = PostSelector.from_circuit(
    circuit=template_circuit, coupling_map=backend.coupling_map
)

# Compute the ps mask for filtering results
mask = post_selector.compute_mask(exec_results[0], strategy="edge")

# Compute expvals using post selected results
results = executor_expectation_values(
    exec_results[0]["meas"],
    bases_reverser,
    meas_basis_axis=0,
    avg_axis=1,
    measurement_flips=exec_results[0]["measurement_flips.meas"],
    pauli_signs=exec_results[0].get("pauli_signs", None),
    postselect_mask=mask,
    rescale_factors=trex_rescale_factors,
)

bases_reverser_unmit = {Pauli("Z" * num_qubits): [observable]}
args = [
    (bases_reverser_unmit, None, None),
    (bases_reverser, None, None),
    (bases_reverser, None, trex_rescale_factors),
    (bases_reverser, mask, None),
    (bases_reverser, mask, trex_rescale_factors),
]

evs = []
for reverser, postsel_mask, factors in args:
    # Compute expvals using post selected results
    res_ps = executor_expectation_values(
        exec_results[0]["meas"],
        reverser,
        meas_basis_axis=0,
        avg_axis=1,
        measurement_flips=exec_results[0]["measurement_flips.meas"],
        pauli_signs=exec_results[0].get("pauli_signs", None),
        postselect_mask=postsel_mask,
        rescale_factors=factors,
    )
    res_ps = np.array(res_ps)
    evs.append(res_ps[:, 0][0])

experiments = ["PNA", "PNA+TREX", "PNA+PS", "PNA+PS+TREX"]
colors = ["#d9d9d9", "#b0b0b0", "#7f7f7f", "#4c4c4c"]
plt.bar(experiments, evs[1:], color=colors)
plt.axhline(y=1, color="green", linestyle="--", linewidth=2, label="Ideal")
plt.axhline(
    y=evs[0], color="red", linestyle="--", linewidth=2, label="Unmitigated"
)
plt.ylabel("Expectation value", fontsize=14)

plt.title(
    r"30q Mirrored Ising, 10 Trotter steps, $\theta_{rx}=\frac{\pi}{8}$",
    fontsize=14,
)
plt.legend(loc="upper left", bbox_to_anchor=(1.05, 1), borderaxespad=0.0)
plt.xticks(rotation=45)
plt.tight_layout()
plt.show()