Input e output di Executor

Versioni dei pacchetti

Il codice in questa pagina è stato sviluppato con i seguenti requisiti. Ti consigliamo di utilizzare queste versioni o versioni più recenti.

qiskit[all]~=2.6.1
qiskit-ibm-runtime~=0.50.0
samplomatic~=0.21.0

La primitiva Executor fa parte del modello di esecuzione guidato, che fornisce maggiore flessibilità nella personalizzazione di un flusso di lavoro di mitigazione degli errori.

Gli input e gli output della primitiva Executor sono molto diversi da quelli delle primitive Sampler e Estimator. Ad esempio, invece di prendere un elenco di PUB come input, Executor prende un QuantumProgram, che contiene un elenco di oggetti QuantumProgramItem. Queste classi contenitore ti danno più flessibilità rispetto a un PUB, che è una semplice struttura dati a tupla.

L'output di Executor è un QuantumProgramResult, che è iterabile e contiene un elemento per ogni QuantumProgramItem di input.

Input: Programmi quantistici

Come indicato in precedenza, l'input a una primitiva Executor è un QuantumProgram, che è un iterabile di oggetti QuantumProgramItem. Questi oggetti possono essere di due tipi:

• CircuitItem, che tipicamente memorizza un circuito e i suoi valori dei parametri (se presenti).
• SamplexItem, che tipicamente memorizza quanto segue:
  • Un circuito template
  • Un oggetto samplex, che viene usato per generare set randomizzati di parametri a runtime (ad esempio per eseguire il twirling o iniettare rumore)
  • Argomenti per il samplex, che potrebbero includere valori dei parametri per il circuito originale

Ognuno di questi elementi rappresenta un compito diverso che Executor deve eseguire.

Prima di iniziare

Alcuni degli esempi di codice in questa pagina utilizzano samplex, che fa parte del pacchetto Samplomatic. Pertanto, prima di eseguire quei blocchi di codice, devi installare Samplomatic, come mostrato nel seguente blocco di codice. Per ulteriori informazioni, consulta la documentazione di Samplomatic.

pip install samplomatic

# For visualization support, include the visualization dependencies.
# pip install samplomatic[vis]

Esempio: Crea un QuantumProgram con due compiti diversi

Prima inizializza il tuo programma quantistico, poi aggiungi elementi del programma usando append_circuit_item o append_samplex_item (se è presente un samplex), come mostrato negli esempi seguenti.

La seguente cella inizializza un QuantumProgram e specifica che deve eseguire 1024 shot per ogni configurazione di ogni elemento nel programma.

nota

A differenza di Sampler, un QuantumProgram accetta solo un singolo valore di shot. Se vuoi un valore di shot diverso, hai bisogno di un QuantumProgram separato, che sarebbe un job separato.

# Added by doQumentation — installs the packages this notebook needs if they are missing
import importlib.util

_needed = {"numpy": "numpy", "qiskit": "qiskit", "qiskit_ibm_runtime": "qiskit-ibm-runtime", "samplomatic": "samplomatic"}
_missing = [pip for module, pip in _needed.items()
            if importlib.util.find_spec(module) is None]
# One at a time, so a package that fails to install does not block the others
for _pip in _missing:
    %pip install -q {_pip}
if not _missing:
    print("\u2713 All required packages are installed")

from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime.quantum_program import QuantumProgram
from qiskit_ibm_runtime import Executor, QiskitRuntimeService
from qiskit.circuit import Parameter, QuantumCircuit
import numpy as np
from samplomatic import build
from samplomatic.transpiler import generate_boxing_pass_manager

# Initialize an empty program
program = QuantumProgram(shots=1024)

# Initialize and transpile a 3-qubit quantum circuit with 2 parameters.
circuit = QuantumCircuit(3)
circuit.h(0)
circuit.cx(0, 1)
circuit.cx(1, 2)
circuit.rz(Parameter("theta"), 0)
circuit.rz(Parameter("phi"), 1)

# `measure_all` adds a 3-bit classical register named "meas"
circuit.measure_all()

# Choose the least busy backend
service = QiskitRuntimeService()
backend = service.least_busy(operational=True, simulator=False)

# Generate a preset pass manager
# This will be used to convert the abstract circuit to an
# equivalent Instruction Set Architecture (ISA) circuit.
preset_pass_manager = generate_preset_pass_manager(
    backend=backend, optimization_level=0
)

# Transpile the circuit
isa_circuit = preset_pass_manager.run(circuit)

Aggiungi un CircuitItem

Poi, aggiungi il circuito target, transpilato secondo l'instruction set architecture (ISA) del backend, al QuantumProgram. Poiché questo circuito ha due parametri, dobbiamo anche fornire i valori dei parametri (10 set in questo esempio). L'esecuzione di questo CircuitItem è il primo compito che il programma eseguirà.

# Append the transpiled circuit and an array
# containing 10 sets of parameter values to the program
program.append_circuit_item(
    isa_circuit,
    circuit_arguments=np.random.rand(
        10, 2
    ),  # 10 sets of parameter values and 2 parameters
)

Aggiungi un SamplexItem

Gli elementi circuit vengono eseguiti senza alcun tipo di randomizzazione. Al contrario, gli elementi samplex ti permettono di specificare come randomizzare il loro contenuto. La cella successiva usa la funzione generate_boxing_pass_manager() per raggruppare i gate e le misurazioni del circuito in box e aggiungere un'annotazione di twirling a ogni box. Genera poi una coppia composta da un circuito template e un samplex usando la funzione build().

L'esecuzione di questo SamplexItem è il secondo compito che il programma eseguirà.

Consulta la documentazione dell'API di Samplomatic per i dettagli completi su samplex e i suoi argomenti. Consulta la guida al Transpiler di Samplomatic per informazioni sull'uso della funzione generate_boxing_pass_manager().

# Transpile the circuit, additionally grouping gates and measurements into annotated boxes
preset_pass_manager = generate_preset_pass_manager(
    backend=backend, optimization_level=0
)

# Use the boxing pass manager to group gates
# and measurements into boxes and add
# a`Twirl` annotation.
preset_pass_manager.post_scheduling = generate_boxing_pass_manager(
    # Add gate twirling
    enable_gates=True,
    # Add measurement twirling
    enable_measures=True,
)
boxed_circuit = preset_pass_manager.run(circuit)

# Build the template circuit and the samplex.  The template circuit has parametric gates
# without fixed values and the samplex randomly generates the parameter
# values on the server side at runtime to perform twirling.
template_circuit, samplex = build(boxed_circuit)

# Determine what arguments are required by the samplex.
# Input the arguments in samplex_arguments.
print(samplex.inputs())

TensorInterface(<
  - 'parameter_values' <float64[2]>: Input parameter values to use during sampling.
>)

# Append the template circuit and samplex as a samplex item
program.append_samplex_item(
    template_circuit,
    samplex=samplex,
    samplex_arguments={
        # the arguments required by the samplex.sample method
        "parameter_values": np.random.rand(10, 2),
    },
    shape=(28, 10),  # 28 randomizations and 10 sets of parameter values
)

# Initialize an Executor with the default options
executor = Executor(mode=backend)

# Submit the job
job = executor.run(program)

# Retrieve the result
result = job.result()

Output

L'output di Executor è un QuantumProgramResult, che è iterabile. Contiene una voce per ogni QuantumProgramItem di input nello stesso ordine degli elementi di input. Ognuno di questi elementi di output è un dizionario in cui le chiavi sono stringhe che corrispondono ai nomi dei registri classici nei circuiti di input (tra gli altri), quindi non è più necessario memorizzare questi nomi come si faceva con l'output di Sampler. I valori del dizionario sono di tipo np.ndarray.

Il risultato per l'esempio precedente contiene questi elementi:

Risultato di CircuitItem

Il primo elemento contiene i risultati dell'esecuzione del primo compito (un CircuitItem) nel programma. Contiene una singola chiave, meas, che è il nome del registro classico nel circuito di input. Il valore di questa chiave viene mappato su un np.ndarray di forma (parameter sets, shots, register bits), che è (10, 1024, 3) per l'esempio precedente.

Il seguente codice illustra come accedere a queste informazioni:

# Access the results of the classical register of task #0, a CircuitItem
result_0 = result[0]["meas"]
print(f"Result shape: {result_0.shape}")

Result shape: (10, 1024, 3)

L'asse più a destra di result_0 scorre sui bit classici in ordine little-endian. Quindi l'indice 0 contiene l'esito del bit classico 0, il bit meno significativo.

Risultato di SamplexItem

Il secondo elemento contiene i risultati dell'esecuzione del secondo compito (un SamplexItem) nel programma. Questo elemento contiene più chiavi. La chiave meas, che è il nome del registro classico del circuito di input, viene mappata sull'array di risultati di tale registro. Questo array ha la forma (randomizations, parameter sets, shots, classical bits), ovvero (28, 10, 1024, 3) in questo esempio.

Inoltre, l'output contiene una chiave measurement_flips.meas, che rappresenta le correzioni di bit-flip per annullare il twirling di misura per il registro meas. Questa forma di output sarà (28, 10, 1, 3) per il nostro esempio perché è richiesto un solo shot per eseguire il bit-flip.

Nota che l'asse più a destra di result_1 scorre sui bit classici in ordine little-endian.

# Access the results of the classical register of task #1
result_1 = result[1]["meas"]
print(f"Result shape: {result_1.shape}")

# Access the bit-flip corrections
flips_1 = result[1]["measurement_flips.meas"]
print(f"Bit-flip corrections shape: {flips_1.shape}")

# Undo the bit flips via classical XOR
unflipped_result_1 = result_1 ^ flips_1

Result shape: (28, 10, 1024, 3)
Bit-flip corrections shape: (28, 10, 1, 3)

Visualizzare gli execution span

Una singola chiamata a Executor.run() invia un singolo job. Lato server, tuttavia, il servizio non esegue necessariamente ogni configurazione del programma quantistico nello stesso momento. Raggruppa invece i circuiti associati degli elementi del programma in chunk ed esegue un chunk alla volta sulla QPU. Non devi dividere il tuo programma né inviare job aggiuntivi; la suddivisione in chunk avviene automaticamente all'interno del singolo job. L'attributo metadata del risultato contiene informazioni sui tempi di questi chunk in result.metadata.chunk_timing. Questi timestamp sono gli effettivi tempi di inizio e fine di ogni chunk sulla QPU. Puoi usare queste informazioni per vedere quando sono stati raccolti determinati dati sulla QPU, per capire come Executor ha distribuito il lavoro o per diagnosticare problemi di prestazioni.

chunk_timing è un elenco che contiene un ChunkSpan per ogni chunk eseguito. Ogni ChunkSpan ha i seguenti attributi:

• start e stop: I timestamp UTC in cui il chunk ha iniziato e terminato l'esecuzione sulla QPU.

• parts: Un elenco di oggetti ChunkPart che descrivono quali elementi del programma hanno contribuito al chunk. Ogni ChunkPart contiene quanto segue:

  • idx_item: L'indice dell'elemento del programma, nell'ordine in cui gli elementi sono stati aggiunti al QuantumProgram.

  • size: Il numero di configurazioni (elementi della forma dell'elemento) di quell'elemento che sono state eseguite in questo chunk. Ad esempio, il SamplexItem dell'esempio precedente ha forma (28, 10), quindi ha 280 configurazioni in totale.

Il codice seguente stampa la durata di ogni chunk e il contenuto di ogni chunk per il job inviato in precedenza:

for idx, span in enumerate(result.metadata.chunk_timing):
    duration = (span.stop - span.start).total_seconds()
    print(f"Chunk {idx}: {span.start} to {span.stop} ({duration:.3f} s)")
    for part in span.parts:
        print(f"  Item {part.idx_item}: {part.size} configurations")

Chunk 0: 2026-11-06 01:47:46.009718+00:00 to 2026-11-06 01:49:05.577720+00:00 (79.568 s)
  Item 0: 10 configurations
  Item 1: 280 configurations

La proprietà QuantumProgramResult.timing racchiude lo stesso elenco in un oggetto ChunkTiming, che riporta anche informazioni globali sull'insieme dei chunk:

timing = result.timing

print("Number of chunks:", len(timing))
print("   Start of the first chunk:", timing.start)
print("      End of the last chunk:", timing.stop)
print("        Total duration (s):", timing.duration)

Number of chunks: 1
   Start of the first chunk: 2026-11-06 01:47:46.009718+00:00
      End of the last chunk: 2026-11-06 01:49:05.577720+00:00
        Total duration (s): 79.568002

Visualizzare graficamente gli execution span

Usa la funzione draw_chunk_timings per visualizzare i tempi dei chunk in un grafico a barre interattivo. L'asse x è il tempo e ogni chunk è disegnato come un segmento orizzontale che si estende dal suo istante di inizio a quello di fine. L'asse y mostra il numero cumulativo di configurazioni completate (o la percentuale del carico di lavoro totale, se imposti normalize_y=True). Passa il cursore su un segmento per vedere i timestamp, la durata e le parti di quel chunk.

La funzione restituisce una figura plotly, che puoi mostrare direttamente, salvare su file, o entrambe le cose. Accetta l'elenco chunk_timing oppure un oggetto ChunkTiming. Puoi passarne più di uno, ad esempio da job diversi, per disegnarli nello stesso grafico. Imposta common_start=True per allinearli a un istante di inizio comune per il confronto.

Plotly richiesto

draw_chunk_timings richiede plotly. Puoi installarlo eseguendo pip install plotly.

Intervalli tra i chunk

La visualizzazione potrebbe mostrare intervalli tra l'esecuzione dei chunk. Tutti i job inviati a una QPU reale passano attraverso una preelaborazione classica, come la traduzione in impulsi che possono essere inviati all'hardware. Poiché questa preelaborazione può essere eseguita in parallelo, il servizio estrae dalla coda e avvia più job contemporaneamente. La QPU, tuttavia, può elaborare solo un'unità di lavoro alla volta. Pertanto, se un job genera più chunk, i suoi chunk potrebbero essere intercalati con quelli dei job di altri utenti e ogni chunk potrebbe dover attendere il proprio turno sulla QPU. Ecco perché possono comparire intervalli tra i chunk.

from qiskit_ibm_runtime.visualization import draw_chunk_timings

fig = draw_chunk_timings(result.timing, names="Executor job")

# Alternatively, call the `draw()` method of the ChunkTiming object
# fig = result.timing.draw()

# Uncomment the following line to display the figure
# fig.show(renderer="notebook")

# Save to a file
# fig.write_html("chunk_timings.html")

Passi successivi

Raccomandazioni

• Esplora gli esempi che usano Executor.
• Scopri il modello di esecuzione guidato.
• Comprendi il broadcasting di Executor.