Entradas y salidas de Executor

Versiones de paquetes

El código de esta página fue desarrollado con los siguientes requisitos. Recomendamos usar estas versiones o más recientes.

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

La primitiva Executor forma parte del modelo de ejecución dirigida, que proporciona mayor flexibilidad al personalizar un flujo de trabajo de mitigación de errores.

Las entradas y salidas del primitivo Executor son muy diferentes a las de los primitivos Sampler y Estimator. Por ejemplo, en lugar de tomar una lista de PUBs como entrada, Executor toma un QuantumProgram, que contiene una lista de objetos QuantumProgramItem. Estas clases contenedoras te ofrecen más flexibilidad que un PUB, que es una estructura de datos de tupla simple.

La salida de Executor es un QuantumProgramResult, que es un iterable y contiene un elemento por cada QuantumProgramItem de entrada.

Entradas: Programas cuánticos

Como se mencionó anteriormente, la entrada a un primitivo Executor es un QuantumProgram, que es un iterable de objetos QuantumProgramItem. Estos objetos pueden ser de dos tipos:

• CircuitItem, que generalmente almacena un circuito y sus valores de parámetros (si los hay).
• SamplexItem, que generalmente almacena lo siguiente:
  • Un circuito plantilla
  • Un objeto samplex, que se usa para generar conjuntos aleatorios de parámetros en tiempo de ejecución (por ejemplo, para realizar twirling o inyectar ruido)
  • Argumentos para el samplex, que pueden incluir valores de parámetros para el circuito original

Cada uno de estos elementos representa una tarea distinta que Executor debe realizar.

Antes de comenzar

Algunos de los ejemplos de código en esta página usan samplex, que forma parte del paquete Samplomatic. Por lo tanto, antes de ejecutar esos bloques de código, debes instalar Samplomatic, como se muestra en el siguiente bloque de código. Para más información, consulta la documentación de Samplomatic.

pip install samplomatic

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

Ejemplo: Crear un QuantumProgram con dos tareas diferentes

Primero inicializa tu programa cuántico y luego agrégale elementos de programa con append_circuit_item o append_samplex_item (si hay un samplex presente), como se muestra en los siguientes ejemplos.

La siguiente celda inicializa un QuantumProgram y especifica que debe ejecutar 1024 shots para cada configuración de cada elemento en el programa.

nota

A diferencia de Sampler, un QuantumProgram toma solo un único valor de shot. Si quieres un valor de shot diferente, necesitas un QuantumProgram separado, que sería un trabajo separado.

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

Agregar un CircuitItem

A continuación, agrega el circuito objetivo, que fue transpilado según la arquitectura de conjunto de instrucciones (ISA) del backend, al QuantumProgram. Dado que este circuito tiene dos parámetros, también debemos proporcionar los valores de los parámetros (10 conjuntos en este ejemplo). Ejecutar este CircuitItem es la primera tarea que realizará el programa.

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

Agregar un SamplexItem

Los elementos de circuito se ejecutan sin ningún tipo de aleatorización. Por el contrario, los elementos de samplex te permiten especificar cómo aleatorizar su contenido. La siguiente celda usa la función generate_boxing_pass_manager() para agrupar los gates y las mediciones del circuito en cajas y agregar una anotación de twirling a cada caja. Luego genera un par de circuito plantilla y samplex usando la función build().

Ejecutar este SamplexItem es la segunda tarea que realizará el programa.

Consulta la documentación de la API de Samplomatic para obtener detalles completos sobre samplex y sus argumentos. Consulta la guía del Transpilador de Samplomatic para obtener información sobre el uso de la función 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()

Salidas

La salida de Executor es un QuantumProgramResult, que es un iterable. Contiene una entrada por cada QuantumProgramItem de entrada en el mismo orden que los elementos de entrada. Cada uno de estos elementos de salida es un diccionario donde las claves son cadenas que corresponden a los nombres de los registros clásicos en los circuits de entrada (entre otros), por lo que ya no necesitas memorizar estos nombres como lo hacías con la salida de Sampler. Los valores del diccionario son de tipo np.ndarray.

El resultado del ejemplo anterior contiene estos elementos:

Resultado de CircuitItem

El primer elemento contiene los resultados de ejecutar la primera tarea (un CircuitItem) en el programa. Contiene una sola clave, meas, que es el nombre del registro clásico en el circuito de entrada. El valor de esta clave se asigna a un np.ndarray de forma (parameter sets, shots, register bits), que es (10, 1024, 3) para el ejemplo anterior.

El siguiente código ilustra cómo acceder a esta información:

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

El eje situado más a la derecha de result_0 recorre los bits clásicos en orden little-endian. Por tanto, el índice 0 contiene el resultado del bit clásico 0, el bit menos significativo.

Resultado de SamplexItem

El segundo elemento contiene los resultados de ejecutar la segunda tarea (un SamplexItem) en el programa. Este elemento contiene varias claves. La clave meas, que es el nombre del registro clásico del circuito de entrada, se asigna al arreglo de resultados de ese registro. Este arreglo tiene la forma (randomizations, parameter sets, shots, classical bits), o (28, 10, 1024, 3) en este ejemplo.

Además, la salida contiene una clave measurement_flips.meas, que contiene las correcciones de inversión de bits necesarias para deshacer el twirling de medición para el registro meas. Esta forma de salida será (28, 10, 1, 3) para nuestro ejemplo porque solo se requiere un shot para realizar la inversión de bit.

Ten en cuenta que el eje situado más a la derecha de result_1 recorre los bits clásicos en orden 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)

Ver los execution spans

Una sola llamada a Executor.run() envía un solo trabajo. Sin embargo, en el lado del servidor, el servicio no ejecuta necesariamente todas las configuraciones del programa cuántico al mismo tiempo. En cambio, agrupa los circuitos enlazados de los elementos del programa en fragmentos (chunks) y ejecuta un fragmento a la vez en la QPU. No necesitas dividir tu programa ni enviar trabajos adicionales; la fragmentación ocurre automáticamente dentro del único trabajo. El atributo metadata del resultado contiene información de tiempos sobre estos fragmentos en result.metadata.chunk_timing. Estas marcas de tiempo son los instantes reales de inicio y fin de cada fragmento en la QPU. Puedes usar esta información para ver cuándo se recopilaron datos concretos en la QPU, para entender cómo distribuyó Executor el trabajo o para diagnosticar problemas de rendimiento.

chunk_timing es una lista que contiene un ChunkSpan por cada fragmento ejecutado. Cada ChunkSpan tiene los siguientes atributos:

• start y stop: Las marcas de tiempo UTC en que el fragmento comenzó y terminó de ejecutarse en la QPU.

• parts: Una lista de objetos ChunkPart que describen qué elementos del programa contribuyeron al fragmento. Cada ChunkPart contiene lo siguiente:

  • idx_item: El índice del elemento del programa, en el orden en que se añadieron los elementos al QuantumProgram.

  • size: El número de configuraciones (elementos de la forma del elemento) de ese elemento que se ejecutaron en este fragmento. Por ejemplo, el SamplexItem del ejemplo anterior tiene forma (28, 10), por lo que tiene 280 configuraciones en total.

El siguiente código imprime la duración de cada fragmento y el contenido de cada fragmento para el trabajo enviado anteriormente:

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 propiedad QuantumProgramResult.timing envuelve la misma lista en un objeto ChunkTiming, que además informa de datos globales sobre el conjunto de fragmentos:

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

Visualizar los execution spans

Usa la función draw_chunk_timings para visualizar los tiempos de los fragmentos en un gráfico de barras interactivo. El eje x es el tiempo, y cada fragmento se dibuja como un segmento horizontal que abarca sus instantes de inicio y fin. El eje y muestra el número acumulado de configuraciones completadas (o el porcentaje de la carga de trabajo total, si estableces normalize_y=True). Pasa el cursor sobre un segmento para ver las marcas de tiempo, la duración y las partes de ese fragmento.

La función devuelve una figura de plotly, que puedes mostrar directamente, guardar en un archivo o ambas cosas. Acepta la lista chunk_timing o un objeto ChunkTiming. Puedes pasar varios de ellos, por ejemplo de distintos trabajos, para dibujarlos en el mismo gráfico. Establece common_start=True para alinearlos en un instante de inicio común y poder compararlos.

Se requiere Plotly

draw_chunk_timings requiere plotly. Puedes instalarlo ejecutando pip install plotly.

Huecos entre fragmentos

La visualización puede mostrar huecos entre las ejecuciones de los fragmentos. Todos los trabajos enviados a una QPU real pasan por cierto preprocesamiento clásico, como la traducción a pulsos que pueden enviarse al hardware. Como este preprocesamiento puede hacerse en paralelo, el servicio saca de la cola e inicia varios trabajos simultáneamente. Sin embargo, la QPU solo puede procesar una unidad de trabajo a la vez. Por lo tanto, si un trabajo genera varios fragmentos, sus fragmentos pueden intercalarse con los de los trabajos de otros usuarios, y cada fragmento puede tener que esperar su turno en la QPU. Por eso pueden aparecer huecos entre fragmentos.

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

Próximos pasos

Recomendaciones

• Explora ejemplos que usan Executor.
• Aprende sobre el modelo de ejecución dirigida.
• Comprende la difusión de Executor.