Ejecutar trabajos cuánticos desde un workflow

Versiones de los paquetes

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

qiskit[all]~=2.6.1
qiskit-ibm-runtime~=0.51.0
qiskit-serverless~=0.38.0

Un workflow cuántico-clásico se ejecuta como un único trabajo de workflow gestionado. Sus etapas clásicas se ejecutan en tu perfil de cómputo, y los trabajos de primitivas que envía a una QPU de IBM Quantum® se ejecutan en tu instancia y cuentan para tu uso. Desde ese único trabajo, puedes informar de cada etapa, encontrar los trabajos de QPU y cancelarlos. Para iniciar un trabajo y consultar su estado y sus registros, consulta Ejecutar un workflow gestionado.

Antes de empezar, despliega tu primer workflow gestionado.

Crear el servicio Runtime

Crea el servicio Runtime en el código de tu workflow con get_runtime_service() y úsalo del mismo modo que usas QiskitRuntimeService en local. El siguiente workflow ejecuta un circuito de Bell con la primitiva Sampler, informa de cada etapa con update_status() y devuelve los conteos:

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

_needed = {"qiskit": "qiskit", "qiskit_ibm_catalog": "qiskit-ibm-catalog", "qiskit_ibm_runtime": "qiskit-ibm-runtime", "qiskit_serverless": "qiskit-serverless"}
_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")

# This cell is hidden from users. It creates the working directory.
from pathlib import Path

Path("./source_files").mkdir(exist_ok=True)

%%writefile ./source_files/my_workflow.py
from qiskit import QuantumCircuit
from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime.executor_sampler import Sampler
from qiskit_serverless import (
    Job,
    get_arguments,
    get_runtime_service,
    save_result,
    update_status,
)

arguments = get_arguments()

service = get_runtime_service()
backend = service.backend(arguments["backend_name"])

circuit = QuantumCircuit(2)
circuit.h(0)
circuit.cx(0, 1)
circuit.measure_all()

update_status(Job.OPTIMIZING_HARDWARE)
pass_manager = generate_preset_pass_manager(optimization_level=1, backend=backend)
isa_circuit = pass_manager.run(circuit)

update_status(Job.WAITING_QPU)
sampler = Sampler(mode=backend)
primitive_job = sampler.run([isa_circuit])
counts = primitive_job.result()[0].data.meas.get_counts()

update_status(Job.POST_PROCESSING)
save_result({"backend": backend.name, "counts": counts})

Overwriting ./source_files/my_workflow.py

get_runtime_service() devuelve un QiskitRuntimeService configurado para tu trabajo de workflow gestionado y ofrece lo siguiente:

• Credenciales. El servicio Runtime usa la clave de API, la instancia y el canal que usaste para llamar a run(), así que el código del workflow no necesita credenciales.

• Seguimiento. Un workflow registra, bajo el trabajo de workflow gestionado, cada trabajo de primitiva que se ejecuta a través de este servicio Runtime, incluidos los trabajos de Sampler, Estimator y Executor, y los trabajos en una sesión o un batch. Puedes listarlos con job.runtime_jobs().

• Cancelación. Cuando llamas a job.cancel(), el workflow gestionado también cancela las sesiones, los batches y los trabajos de primitivas del workflow que aún se están ejecutando.

Un workflow gestionado no hace seguimiento ni cancela los trabajos de un servicio que creas con QiskitRuntimeService(), aunque los trabajos se sigan ejecutando. Si el workflow debe usar otro servicio, cancela sus trabajos en un controlador de SIGTERM.

Ejecutar primitivas en una sesión o un batch

Un workflow gestionado también hace seguimiento de los trabajos de primitivas que se ejecutan en una sesión o un batch. Crea la sesión o el batch a partir de un backend que devolvió el servicio Runtime y crea las primitivas dentro del bloque with. Session funciona del mismo modo:

source_files/my_workflow.py

from qiskit_ibm_runtime import Batch

with Batch(backend=backend) as batch:
    sampler = Sampler(mode=batch)
    primitive_jobs = [sampler.run([isa_circuit]) for _ in range(2)]
    results = [primitive_job.result() for primitive_job in primitive_jobs]

Informar del progreso con update_status()

El workflow de Crear el servicio Runtime llama a update_status() al inicio de cada etapa, de modo que el estado del trabajo muestra en qué punto está el workflow, por ejemplo RUNNING: WAITING_FOR_QPU. Pasa una de las siguientes constantes de Job:

Llamada en el workflow	Estado del trabajo
update_status(Job.MAPPING)	RUNNING: MAPPING
update_status(Job.OPTIMIZING_HARDWARE)	RUNNING: OPTIMIZING_FOR_HARDWARE
update_status(Job.WAITING_QPU)	RUNNING: WAITING_FOR_QPU
update_status(Job.EXECUTING_QPU)	RUNNING: EXECUTING_QPU
update_status(Job.POST_PROCESSING)	RUNNING: POST_PROCESSING

update_status() devuelve True si el workflow cambió el estado. Devuelve False y escribe una advertencia en los registros si el valor no es una de estas constantes, si el trabajo ya no está activo o si el código se ejecuta fuera de un trabajo. No lanza un error, así que un error tipográfico no detiene el workflow.

Para el progreso dentro de una etapa, como las iteraciones de un optimizador, escribe líneas de registro.

Encontrar los trabajos de QPU de una ejecución

En tu notebook local, elige una QPU a la que pueda acceder tu instancia. client.backends() las lista y client.least_busy() devuelve la que tiene menos trabajos pendientes:

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
print([backend.name for backend in client.backends(simulator=False)])

backend_name = client.least_busy(operational=True, simulator=False).name

['ibm_boston', 'ibm_fez', 'ibm_kingston', 'ibm_marrakesh', 'ibm_phoenix', 'ibm_pittsburgh']

Sube el workflow y pasa el nombre del backend cuando lo ejecutes:

from qiskit_ibm_catalog import ManagedWorkflow

my_workflow = client.upload(
    ManagedWorkflow(
        title="my-workflow",
        entrypoint="my_workflow.py",
        working_dir="./source_files/",
    )
)

job = my_workflow.run(backend_name=backend_name)

# This cell is hidden from users. It waits for the job to finish.
assert job.wait_for_final_state() == "DONE", job.logs()

Después de que el trabajo termine, obtén los ID de los trabajos de QPU del workflow a partir del trabajo de workflow gestionado:

runtime_job_ids = job.runtime_jobs()
runtime_job_ids

['dbl6lst8ft8s738secf0']

# This cell is hidden from users. It checks the tracked job.
assert len(runtime_job_ids) == 1, runtime_job_ids

Cada ID es un ID normal de trabajo de primitiva. Para leer el estado y los resultados de un trabajo, consulta Recuperar los resultados de un trabajo más tarde. Si el workflow ejecutó primitivas en una sesión o un batch, lista los ID de sesión y de batch con runtime_sessions() y pasa uno a runtime_jobs() para obtener solo los trabajos que contiene:

run_workflow_remotely.ipynb

for session_id in job.runtime_sessions():
    print(session_id, job.runtime_jobs(session_id))

dc101077-d085-40fb-8fbe-78852e355994 ['dbjqp48nceqs73aki1k0', 'dbjqp4l6p51s73clb91g']

El objeto job existe solo en el notebook que inició la ejecución. En un notebook nuevo, vuelve a obtener el trabajo con client.job("<your-job-id>") y luego llama a runtime_jobs() sobre él. Consulta Encontrar un trabajo desde una sesión nueva.

Para el uso de QPU y los tiempos de ejecución de cada trabajo de QPU, consulta Ver las métricas del trabajo. Para el uso de toda tu instancia en el periodo actual, llama a client.usage(). Para ver cuánto tardó cada etapa del trabajo de workflow gestionado, consulta Comprobar cuánto tardó cada etapa.

Cancelar los trabajos de QPU vinculados

Cuando cancelas un trabajo de workflow gestionado con job.cancel(), el workflow también intenta cancelar el trabajo de QPU que inició:

• Sesiones y batches. El workflow cancela cada una de sus sesiones y batches, lo que también cancela sus trabajos en cola.

• Otros trabajos de primitivas. El workflow cancela cada uno de sus otros trabajos de primitivas que aún están en cola o en ejecución.

job.cancel() devuelve un mensaje por cada trabajo de QPU y sesión, para que puedas comprobar qué se canceló:

run_workflow_remotely.ipynb

print(job.cancel())

En un workflow que se ejecutó en un batch, el mensaje tiene el siguiente aspecto:

Job is stopping. Canceled runtime session: eb8874c9-a15e-45f8-9a90-a9fb35267f9d and associated runtime jobs.

Si el cliente no puede crear un QiskitRuntimeService a partir de tus credenciales locales, cancela solo el trabajo de workflow gestionado. Cancela tú mismo los trabajos de QPU con sus ID de job.runtime_jobs().

Cuando un workflow detiene un trabajo por sí mismo, por ejemplo porque alcanzó el límite de 24 horas, no cancela sus trabajos de QPU. Cancélalos con sus ID de job.runtime_jobs().

Solucionar problemas de trabajos de QPU en un workflow

Problema	Causa
get_runtime_service() lanza KeyError: 'QISKIT_IBM_CHANNEL'.	El código se ejecuta en tu notebook local, no en un trabajo de workflow gestionado. Usa QiskitRuntimeService() en local.
update_status() devuelve False y los registros muestran 'sub_status' not provided or is not valid.	El valor no es una de las constantes de Job, o el trabajo ya no está activo.
job.runtime_jobs() devuelve [].	El workflow creó su servicio con QiskitRuntimeService(), o todavía no ha ejecutado ninguna primitiva.
No se hace seguimiento de los trabajos de una sesión.	La sesión se creó a partir de un backend de otro servicio. Obtén el backend de get_runtime_service().
Los registros del workflow muestran Could not associate runtime job with serverless job.	El workflow no pudo registrar un trabajo de primitiva. El trabajo de primitiva se sigue ejecutando, pero no se le hace seguimiento.
job.cancel() muestra No QiskitRuntimeService can be associated to the given token and instance.	El cliente no pudo crear un QiskitRuntimeService a partir de tus credenciales locales, así que canceló solo el trabajo de workflow gestionado.
client.job() falla con Job [...] not found.	El ID del trabajo es incorrecto, o el trabajo pertenece a otro usuario.

Próximos pasos

Recomendaciones

• Ejecuta un workflow gestionado para comprobar el estado, el resultado y los registros de un trabajo, y encontrar la causa de un error.