Entrées et sorties Executor

Versions des packages

Le code de cette page a été développé avec les versions suivantes. Nous recommandons d'utiliser ces versions ou des versions plus récentes.

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

La primitive Executor fait partie du modèle d'exécution dirigée, qui offre plus de flexibilité lors de la personnalisation d'un flux de travail d'atténuation d'erreurs.

Les entrées et les sorties de la primitive Executor sont très différentes de celles des primitives Sampler et Estimator. Par exemple, au lieu de prendre une liste de PUBs comme entrée, Executor prend un QuantumProgram, qui contient une liste d'objets QuantumProgramItem. Ces classes conteneurs te donnent plus de flexibilité qu'un PUB, qui est une simple structure de données tuple.

La sortie d'Executor est un QuantumProgramResult, qui est itérable et contient un élément pour chaque QuantumProgramItem d'entrée.

Entrées : Programmes quantiques

Comme indiqué précédemment, l'entrée d'une primitive Executor est un QuantumProgram, qui est un itérable d'objets QuantumProgramItem. Ces objets peuvent être de deux types :

• CircuitItem, qui stocke généralement un circuit et ses valeurs de paramètres (le cas échéant).
• SamplexItem, qui stocke généralement les éléments suivants :
  • Un circuit modèle
  • Un objet samplex, qui est utilisé pour générer des ensembles de paramètres randomisés au moment de l'exécution (par exemple pour effectuer le twirling ou injecter du bruit)
  • Arguments pour le samplex, qui pourraient inclure des valeurs de paramètres pour le circuit original

Chacun de ces éléments représente une tâche différente qu'Executor doit effectuer.

Avant de commencer

Certains des exemples de code sur cette page utilisent samplex, qui fait partie du package Samplomatic. Par conséquent, avant d'exécuter ces blocs de code, tu dois installer Samplomatic, comme indiqué dans le bloc de code suivant. Pour plus d'informations, consulte la documentation Samplomatic.

pip install samplomatic

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

Exemple : Créer un QuantumProgram avec deux tâches différentes

D'abord, initialise ton programme quantique, puis ajoute des éléments de programme en utilisant soit append_circuit_item soit append_samplex_item (si un samplex est présent), comme indiqué dans les exemples suivants.

La cellule suivante initialise un QuantumProgram et spécifie qu'il doit exécuter 1024 shots pour chaque configuration de chaque élément dans le programme.

remarque

Contrairement à Sampler, un QuantumProgram ne prend qu'une seule valeur de shot. Si tu veux une valeur de shot différente, tu as besoin d'un QuantumProgram séparé, ce qui serait un job séparé.

# 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)

Ajouter un CircuitItem

Ensuite, ajoute le circuit cible, qui a été transpilé selon l'architecture d'ensemble d'instructions (ISA) du backend, au QuantumProgram. Puisque ce circuit a deux paramètres, nous devons également fournir les valeurs des paramètres (10 ensembles dans cet exemple). L'exécution de ce CircuitItem est la première tâche que le programme effectuera.

# 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
)

Ajouter un SamplexItem

Les éléments de circuit sont exécutés sans aucune randomisation. En revanche, les éléments samplex te permettent de spécifier comment randomiser leur contenu. La cellule suivante utilise la fonction generate_boxing_pass_manager() pour regrouper les gates et les mesures du circuit dans des boîtes et ajouter une annotation de twirling à chaque boîte. Elle génère ensuite un circuit modèle et une paire samplex en utilisant la fonction build().

L'exécution de ce SamplexItem est la deuxième tâche que le programme effectuera.

Consulte la documentation de l'API Samplomatic pour tous les détails sur samplex et ses arguments. Consulte le guide Transpiler de Samplomatic pour des informations sur l'utilisation de la fonction 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()

Sorties

La sortie d'Executor est un QuantumProgramResult, qui est itérable. Il contient une entrée par QuantumProgramItem d'entrée dans le même ordre que les éléments d'entrée. Chacun de ces éléments de sortie est un dictionnaire où les clés sont des chaînes qui correspondent aux noms des registres classiques dans les circuits d'entrée (entre autres), tu n'as donc plus besoin de mémoriser ces noms comme tu le faisais avec la sortie Sampler. Les valeurs du dictionnaire sont de type np.ndarray.

Le résultat de l'exemple précédent contient ces éléments :

Résultat CircuitItem

Le premier élément contient les résultats de l'exécution de la première tâche (un CircuitItem) dans le programme. Il contient une seule clé, meas, qui est le nom du registre classique dans le circuit d'entrée. La valeur de cette clé correspond à un np.ndarray de forme (parameter sets, shots, register bits), soit (10, 1024, 3) pour l'exemple ci-dessus.

Le code suivant illustre comment accéder à ces informations :

# 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'axe le plus à droite de result_0 parcourt les bits classiques dans l'ordre little-endian. Ainsi, l'indice 0 contient le résultat du bit classique 0, le bit de poids faible.

Résultat SamplexItem

Le second élément contient les résultats de l'exécution de la deuxième tâche (un SamplexItem) dans le programme. Cet élément contient plusieurs clés. La clé meas, qui est le nom du registre classique du circuit d'entrée, correspond au tableau de résultats de ce registre. Ce tableau a la forme (randomizations, parameter sets, shots, classical bits), soit (28, 10, 1024, 3) dans cet exemple.

De plus, la sortie contient une clé measurement_flips.meas, qui correspond aux corrections de bit-flip permettant d'annuler le twirling de mesure pour le registre meas. La forme de cette sortie sera (28, 10, 1, 3) pour notre exemple, car un seul shot est nécessaire pour effectuer le bit-flip.

Note que l'axe le plus à droite de result_1 parcourt les bits classiques dans l'ordre 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)

Afficher les execution spans

Un seul appel à Executor.run() soumet un seul job. Côté serveur, cependant, le service n'exécute pas nécessairement toutes les configurations du programme quantique en même temps. Il regroupe plutôt les circuits liés des éléments du programme en chunks, et exécute un chunk à la fois sur le QPU. Tu n'as pas besoin de découper ton programme ni de soumettre des jobs supplémentaires ; le découpage en chunks se fait automatiquement au sein du même job. L'attribut metadata du résultat contient des informations de chronométrage sur ces chunks dans result.metadata.chunk_timing. Ces horodatages sont les heures réelles de début et de fin de chaque chunk sur le QPU. Tu peux utiliser ces informations pour voir quand des données particulières ont été collectées sur le QPU, pour comprendre comment Executor a réparti le travail, ou pour diagnostiquer des problèmes de performance.

chunk_timing est une liste qui contient un ChunkSpan par chunk exécuté. Chaque ChunkSpan a les attributs suivants :

• start et stop : Les horodatages UTC du début et de la fin de l'exécution du chunk sur le QPU.

• parts : Une liste d'objets ChunkPart qui décrivent quels éléments du programme ont contribué au chunk. Chaque ChunkPart contient les éléments suivants :

  • idx_item : L'indice de l'élément du programme, dans l'ordre où les éléments ont été ajoutés au QuantumProgram.

  • size : Le nombre de configurations (éléments de la forme de l'élément) de cet élément qui ont été exécutées dans ce chunk. Par exemple, le SamplexItem de l'exemple précédent a la forme (28, 10), il a donc 280 configurations au total.

Le code suivant affiche la durée de chaque chunk et le contenu de chaque chunk pour le job soumis précédemment :

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 propriété QuantumProgramResult.timing encapsule la même liste dans un objet ChunkTiming, qui fournit aussi des informations globales sur l'ensemble des chunks :

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

Visualiser les execution spans

Utilise la fonction draw_chunk_timings pour visualiser les chronométrages des chunks sur un graphique à barres interactif. L'axe des x représente le temps, et chaque chunk est dessiné comme un segment horizontal qui couvre ses heures de début et de fin. L'axe des y indique le nombre cumulé de configurations terminées (ou le pourcentage de la charge de travail totale, si tu définis normalize_y=True). Survole un segment pour voir les horodatages, la durée et les parties de ce chunk.

La fonction renvoie une figure plotly, que tu peux afficher directement, enregistrer dans un fichier, ou les deux. Elle accepte soit la liste chunk_timing, soit un objet ChunkTiming. Tu peux en passer plusieurs, par exemple issus de jobs différents, pour les tracer sur le même graphique. Définis common_start=True pour les aligner sur un instant de départ commun afin de les comparer.

Plotly requis

draw_chunk_timings nécessite plotly. Tu peux l'installer en exécutant pip install plotly.

Écarts entre les chunks

La visualisation peut montrer des écarts entre les exécutions des chunks. Tous les jobs soumis à un vrai QPU passent par un prétraitement classique, comme la traduction en impulsions pouvant être envoyées au matériel. Comme ce prétraitement peut se faire en parallèle, le service sort de la file et démarre plusieurs jobs simultanément. Le QPU, en revanche, ne peut traiter qu'une unité de travail à la fois. Par conséquent, si un job génère plusieurs chunks, ses chunks peuvent être entrelacés avec ceux des jobs d'autres utilisateurs, et chaque chunk peut devoir attendre son tour sur le QPU. C'est pourquoi des écarts peuvent apparaître entre les chunks.

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")

Étapes suivantes

Recommandations

• Explore des exemples qui utilisent Executor.
• Découvre le modèle d'exécution dirigée.
• Comprends la diffusion Executor.