Ejecutar un flujo de trabajo gestionado

Versiones de paquetes

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

qiskit-serverless~=0.38.0

Un trabajo de flujo de trabajo gestionado funciona como un trabajo por lotes que envías a un planificador de HPC. run() envía el trabajo y devuelve un objeto Job de inmediato. Usa el objeto Job para las tareas que en un clúster cubren squeue, el archivo de salida del trabajo y scancel. El trabajo sigue ejecutándose después de que cierres tu notebook.

Los ejemplos de esta página usan un flujo de trabajo pequeño que lee un ángulo, theta, y falla si el ángulo está fuera de rango. Cárgalo como se describe en Despliega tu primer flujo de trabajo gestionado.

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

_needed = {"qiskit_ibm_catalog": "qiskit-ibm-catalog", "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
import math

from qiskit_serverless import get_arguments, save_result

arguments = get_arguments()
theta = arguments["theta"]
if not 0 <= theta <= math.pi:
    raise ValueError(f"theta must be between 0 and pi, got {theta}")

print(f"Computing the expectation value for theta={theta}", flush=True)
save_result({"theta": theta, "expectation_value": math.cos(theta)})

Writing ./source_files/my_workflow.py

Iniciar un trabajo

Obtén el flujo de trabajo por su título y llama a run() con los argumentos que lee el flujo de trabajo:

# This cell is hidden from users. It uploads the workflow under a title that
# other test notebooks do not use, and makes load("my-workflow") return it.
# It also silences the account warning that job.cancel() logs.
import logging

from qiskit_ibm_catalog import ManagedWorkflow, ManagedWorkflowClient

logging.getLogger("qiskit_ibm_runtime").setLevel(logging.ERROR)

_test_title = "docs-run-managed-workflow"
ManagedWorkflowClient().upload(
    ManagedWorkflow(
        title=_test_title,
        entrypoint="my_workflow.py",
        working_dir="./source_files/",
    )
)
_load = ManagedWorkflowClient.load
ManagedWorkflowClient.load = lambda self, title, *a, **k: _load(
    self, _test_title if title == "my-workflow" else title, *a, **k
)

run_workflow_remotely.ipynb

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
my_workflow = client.load("my-workflow")

job = my_workflow.run(theta=0.5)
job_id = job.job_id

run() regresa después de que el flujo de trabajo pone el trabajo en cola. Guarda el ID del trabajo para poder encontrar el trabajo desde una sesión nueva.

• Para elegir qué argumentos acepta un flujo de trabajo y comprobarlos antes de que empiece el trabajo, consulta Definir entradas y salidas.

• Para ejecutar el trabajo con un tamaño de cómputo diferente, pasa function_size. Consulta Elegir un tamaño para cada ejecución.

Seguir un trabajo

Comprobar el estado

Inmediatamente después de llamar a run(), el estado suele ser QUEUED: run_workflow_remotely.ipynb

job.status()

'QUEUED'

Estado	Significado
QUEUED	El trabajo espera un espacio de cómputo.
INITIALIZING	El flujo de trabajo está iniciando un entorno con el perfil de cómputo del trabajo.
RUNNING	Se está ejecutando el punto de entrada del flujo de trabajo.
RUNNING: <stage>	El punto de entrada informó de una etapa con update_status(), como RUNNING: WAITING_FOR_QPU.
CANCELING	Cancelaste el trabajo y el flujo de trabajo lo está deteniendo.
DONE	El punto de entrada terminó correctamente.
ERROR	El punto de entrada generó un error o terminó con un código distinto de cero. Consulta Encontrar la causa de un error.
CANCELED	Cancelaste el trabajo, o alcanzó el límite de 24 horas.

DONE, ERROR y CANCELED son estados finales, y job.in_terminal_state() devuelve True para ellos. Un bucle de estado que espera otro valor, como SUCCEEDED, nunca termina.

Esperar a que termine el trabajo

Para esperar hasta que el trabajo termine, llama a wait_for_final_state(). Devuelve el estado final y, con verbose=True, imprime cada cambio de estado. Si estableces timeout en segundos, wait_for_final_state() genera un TimeoutError cuando se agota el tiempo antes de que termine el trabajo. El trabajo sigue ejecutándose:

run_workflow_remotely.ipynb

job.wait_for_final_state(timeout=600, verbose=True)

Waiting for job 3f2a1c0e-6b1d-4c0e-9a51-2f0c4d1e7a90
0:00:00  QUEUED
0:02:10  RUNNING
0:03:05  RUNNING: WAITING_FOR_QPU
0:41:44  DONE
'DONE'

wait_for_final_state() comprueba el estado después de 1, 2, 4, 8, 16 y 32 segundos, y luego cada 60 segundos, de modo que detecta el final de un trabajo corto en cuestión de segundos y realiza pocas solicitudes en un trabajo largo. Para comprobar a un intervalo fijo en su lugar, pasa poll_interval en segundos.

Leer los registros

Todo lo que el flujo de trabajo escribe en la salida estándar o en el error estándar va a los registros del trabajo. Puedes leerlos con job.logs() mientras el trabajo se ejecuta, antes de que tenga un resultado:

# This cell is hidden from users. It waits for the job to end, so that the
# logs are complete.
job.wait_for_final_state()

run_workflow_remotely.ipynb

print(job.logs())

Computing the expectation value for theta=0.5

# This cell is hidden from users. It checks the logs.
assert "Computing the expectation value for theta=0.5" in job.logs()

• El flujo de trabajo carga las líneas de registro nuevas aproximadamente cada 15 segundos, y una última vez cuando el trabajo termina o se cancela. Inmediatamente después de que un trabajo comienza, job.logs() devuelve No logs yet.

• El flujo de trabajo conserva los 50 MB más recientes de registros de cada trabajo y añade una línea al principio que indica que los registros se truncaron.

Para decidir qué registra tu flujo de trabajo, consulta Escribir registros. Para mostrar solo las líneas que coinciden con una expresión regular, pasa include a job.filtered_logs(). Para ocultar líneas, pasa exclude.

Obtener el resultado

job.result() espera hasta que el trabajo termina y luego devuelve el valor que el flujo de trabajo pasó a save_result(): run_workflow_remotely.ipynb

job.result()

{'theta': 0.5, 'expectation_value': 0.8775825618903728}

# This cell is hidden from users. It checks the result.
assert _ == {"theta": 0.5, "expectation_value": 0.8775825618903728}, _  # noqa: F821

De forma predeterminada, job.result() comprueba el estado cada 30 segundos y espera sin límite de tiempo. Pasa maxwait para limitar el número de comprobaciones. Por ejemplo, job.result(maxwait=40) espera unos 20 minutos. Para comprobar una sola vez sin esperar, pasa wait=False. Si job.result() deja de esperar antes de que el trabajo termine, devuelve {}.

Comprobar cuánto duró cada etapa

job.events() enumera cada cambio de estado y de etapa de un trabajo con su hora, para que puedas ver cuánto esperó el trabajo por el cómputo y cuánto duró cada etapa: run_workflow_remotely.ipynb

for event in reversed(job.events()):
    print(
        event.created,
        event.data.get("status") or event.data.get("sub_status"),
    )

2026-11-02T14:43:07.134377Z QUEUED
2026-11-02T14:43:24.474199Z PENDING
2026-11-02T14:46:07.309519Z RUNNING
2026-11-02T14:46:21.115038Z SUCCEEDED

# This cell is hidden from users. It checks the events.
_statuses = [e.data.get("status") for e in job.events()]
assert "QUEUED" in _statuses and "SUCCEEDED" in _statuses, _statuses

Los eventos usan los nombres de estado almacenados. PENDING es INITIALIZING, y SUCCEEDED es DONE. Para consultar el uso de QPU de una ejecución, consulta Encontrar los trabajos de QPU de una ejecución.

Encontrar la causa de un error

Cuando el flujo de trabajo genera una excepción, el trabajo termina con el estado ERROR. Por ejemplo, ejecuta el flujo de trabajo con un ángulo que rechaza: run_workflow_remotely.ipynb

failed_job = my_workflow.run(theta=-1.0)

# This cell is hidden from users. It waits for the job to end.
failed_job.wait_for_final_state()

run_workflow_remotely.ipynb

failed_job.status()

'ERROR'

Los registros muestran el traceback: run_workflow_remotely.ipynb

print(failed_job.logs())

Traceback (most recent call last):
  File "/function_user_data/my_workflow.py", line 8, in <module>
    raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

job.result() raises a QiskitServerlessException: run_workflow_remotely.ipynb

from qiskit_serverless.exception import QiskitServerlessException

try:
    failed_job.result()
except QiskitServerlessException as error:
    print(error)

raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

# This cell is hidden from users. It checks the error.
assert failed_job.status() == "ERROR", failed_job.status()
assert (
    "ValueError: theta must be between 0 and pi, got -1.0"
    in failed_job.logs()
)
assert failed_job.events(type="ERROR") == []

Si el flujo de trabajo informa de errores con send_error(), léelos con job.events(type="ERROR") y lee sus mensajes con job.error_message(). Para un trabajo sin errores informados, job.events(type="ERROR") devuelve una lista vacía y job.error_message() devuelve {}.

run_workflow_remotely.ipynb

for event in job.events(type="ERROR"):
    print(event.created, event.data.get("code"), event.data.get("message"))

Gestionar trabajos

Cancelar un trabajo

Cancela un trabajo con job.cancel(). El estado cambia a CANCELING mientras el flujo de trabajo detiene el trabajo y, después, a CANCELED: run_workflow_remotely.ipynb

job_to_cancel = my_workflow.run(theta=1.0)
job_to_cancel.cancel()

'Job has been stopped. No active runtime job ID associated with this serverless job ID.'

# This cell is hidden from users. It checks the cancellation.
assert (
    job_to_cancel.wait_for_final_state() == "CANCELED"
), job_to_cancel.status()

job.cancel() devuelve un mensaje sobre el trabajo y sobre sus trabajos de QPU vinculados, que también intenta cancelar. Para un trabajo en ejecución, el flujo de trabajo envía SIGTERM a tu punto de entrada. Para guardar resultados parciales, consulta Limpiar cuando se cancela un trabajo. Si el trabajo ya terminó, job.cancel() devuelve Job already in terminal state. y no cambia nada.

Encontrar un trabajo desde una sesión nueva

El objeto Job y el flujo de trabajo existen solo en tu sesión actual de Python. Después de reiniciar el kernel de un notebook, pasa el ID de trabajo guardado a client.job(), o enumera tus trabajos recientes con client.jobs(): run_workflow_remotely.ipynb

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
job = client.job(job_id)
my_workflow = client.load("my-workflow")

client.jobs(function=my_workflow, status="ERROR", limit=5)

[<Job | e6b1ffa1-cd26-437d-bd53-b265dd4ba952>,
 <Job | 11308f2a-8d7a-40a0-a970-7fdeca2b171b>,
 <Job | af597785-3470-469a-9051-607e0185892e>,
 <Job | a9051931-3bc2-4b80-9ce6-c39d28ce260c>,
 <Job | e329e8e3-e955-4188-9bfd-8f3050ee2d39>]

# This cell is hidden from users. It checks the job lookups.
assert job.result()["theta"] == 0.5
assert failed_job.job_id in [j.job_id for j in _]  # noqa: F821

client.jobs() devuelve tus 10 trabajos más recientes, los más nuevos primero. Fíltralos con function, status (de la tabla de estados) o created_after (una marca de tiempo ISO 8601), y pagínalos con limit y offset. Para enumerar los flujos de trabajo que cargaste, llama a client.list().

La tabla Workloads de IBM Quantum Platform también enumera tus trabajos, con su estado, perfil de cómputo y tiempos de ejecución.

Límite de tiempo

Un flujo de trabajo gestionado cancela un trabajo que se ejecuta durante más de 24 horas, y el trabajo termina con el estado CANCELED. Durante las ejecuciones largas, guarda puntos de control en el almacenamiento de archivos para no perder trabajo.

Solucionar problemas de trabajos

Problema	Causa
run() falla con Code: 429 y Active job limit reached.	Alcanzaste el límite de trabajos activos. Espera a que terminen algunos trabajos o cancela los que ya no necesites.
El trabajo permanece en QUEUED, y run() advirtió que your account, plan or subscription is blocked.	El trabajo no se inicia hasta que se levante el bloqueo de tu cuenta, plan o suscripción. Contacta con tu administrador.
job.logs() devuelve No logs yet.	El trabajo está en cola o acaba de empezar. Espera unos 15 segundos.
job.result() devuelve {}	El trabajo no ha terminado, se canceló o no guardó ningún resultado. Consulta Guardar resultados.
job.error_message() devuelve {}	El flujo de trabajo generó una excepción sin informar de un error. Lee el traceback con job.logs().
El trabajo está en CANCELED, pero no lo cancelaste	El trabajo alcanzó el límite de 24 horas.

Próximos pasos

Recomendaciones

• Escribe registros y gestiona errores para configurar los registros y errores que informa un trabajo.
• Ejecuta un barrido de parámetros para ejecutar el flujo de trabajo una vez por cada conjunto de entradas.