Exécuter un workflow géré

Versions des paquets

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

qiskit-serverless~=0.38.0

Un job de workflow géré fonctionne comme un job batch que tu soumets à un ordonnanceur HPC. run() soumet le job et renvoie immédiatement un objet Job. Utilise l'objet Job pour les tâches que couvrent squeue, le fichier de sortie du job et scancel sur un cluster. Le job continue de s'exécuter après que tu as fermé ton notebook.

Les exemples de cette page utilisent un petit workflow qui lit un angle, theta, et échoue si l'angle est hors limites. Téléverse-le comme décrit dans Déployer ton premier workflow géré.

# 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

Démarrer un job

Récupère le workflow par son titre et appelle run() avec les arguments que le workflow lit :

# 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() renvoie le résultat une fois que le workflow a mis le job en file d'attente. Enregistre l'ID du job afin de pouvoir retrouver le job depuis une nouvelle session.

• Pour choisir les arguments qu'un workflow accepte et les vérifier avant le démarrage du job, consulte Définir les entrées et les sorties.

• Pour exécuter le job avec une autre taille de calcul, passe function_size. Consulte Choisir une taille pour chaque exécution.

Suivre un job

Vérifier le statut

Juste après avoir appelé run(), le statut est généralement QUEUED : run_workflow_remotely.ipynb

job.status()

'QUEUED'

Statut	Signification
QUEUED	Le job attend un emplacement de calcul.
INITIALIZING	Le workflow démarre un environnement avec le profil de calcul du job.
RUNNING	Le point d'entrée du workflow s'exécute.
RUNNING: <stage>	Le point d'entrée a signalé une étape avec update_status(), par exemple RUNNING: WAITING_FOR_QPU.
CANCELING	Tu as annulé le job, et le workflow est en train de l'arrêter.
DONE	Le point d'entrée s'est terminé avec succès.
ERROR	Le point d'entrée a levé une erreur ou s'est terminé avec un code non nul. Consulte Trouver la cause d'une erreur.
CANCELED	Tu as annulé le job, ou il a atteint la limite de 24 heures.

DONE, ERROR et CANCELED sont des statuts finaux, et job.in_terminal_state() renvoie True pour eux. Une boucle de statut qui attend une autre valeur, comme SUCCEEDED, ne se termine jamais.

Attendre la fin du job

Pour attendre la fin du job, appelle wait_for_final_state(). Elle renvoie le statut final, et avec verbose=True elle affiche chaque changement de statut. Si tu définis timeout en secondes, wait_for_final_state() lève une TimeoutError lorsque le délai expire avant la fin du job. Le job continue de s'exécuter :

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() vérifie le statut après 1, 2, 4, 8, 16 et 32 secondes, puis toutes les 60 secondes, de sorte qu'elle détecte la fin d'un job court en quelques secondes et fait peu de requêtes pour un job long. Pour vérifier plutôt à intervalle fixe, passe poll_interval en secondes.

Lire les logs

Tout ce que le workflow écrit sur la sortie standard ou la sortie d'erreur standard va dans les logs du job. Tu peux les lire avec job.logs() pendant l'exécution du job, avant qu'il ait un résultat :

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

• Le workflow téléverse les nouvelles lignes de log environ toutes les 15 secondes, et une dernière fois quand le job se termine ou est annulé. Juste après le démarrage d'un job, job.logs() renvoie No logs yet.

• Le workflow conserve les 50 Mo de logs les plus récents pour chaque job, et ajoute en haut une ligne indiquant que les logs ont été tronqués.

Pour décider de ce que ton workflow journalise, consulte Écrire des logs. Pour n'afficher que les lignes qui correspondent à une expression régulière, passe include à job.filtered_logs(). Pour masquer des lignes, passe exclude.

Obtenir le résultat

job.result() attend la fin du job, puis renvoie la valeur que le workflow a transmise à 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

Par défaut, job.result() vérifie le statut toutes les 30 secondes et attend sans limite de temps. Passe maxwait pour limiter le nombre de vérifications. Par exemple, job.result(maxwait=40) attend environ 20 minutes. Pour vérifier une seule fois sans attendre, passe wait=False. Si job.result() cesse d'attendre avant la fin du job, il renvoie {}.

Vérifier la durée de chaque étape

job.events() liste chaque changement de statut et d'étape d'un job avec son horodatage, ce qui te permet de voir combien de temps le job a attendu le calcul et combien de temps chaque étape a duré : 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

Les événements utilisent les noms de statut stockés. PENDING correspond à INITIALIZING, et SUCCEEDED correspond à DONE. Pour l'utilisation de QPU d'une exécution, consulte Trouver les jobs QPU d'une exécution.

Trouver la cause d'une erreur

Quand le workflow lève une exception, le job se termine avec le statut ERROR. Par exemple, exécute le workflow avec un angle qu'il rejette : 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'

Les logs affichent la trace d'appels : 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 le workflow signale les erreurs avec send_error(), lis-les avec job.events(type="ERROR"), et lis leurs messages avec job.error_message(). Pour un job sans erreur signalée, job.events(type="ERROR") renvoie une liste vide, et job.error_message() renvoie {}.

run_workflow_remotely.ipynb

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

Gérer les jobs

Annuler un job

Annule un job avec job.cancel(). Le statut passe à CANCELING pendant que le workflow arrête le job, puis à 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() renvoie un message sur le job et sur ses jobs QPU liés, qu'il tente aussi d'annuler. Pour un job en cours d'exécution, le workflow envoie SIGTERM à ton point d'entrée. Pour enregistrer des résultats partiels, consulte Nettoyer lorsqu'un job est annulé. Si le job est déjà terminé, job.cancel() renvoie Job already in terminal state. et ne change rien.

Retrouver un job depuis une nouvelle session

L'objet Job et le workflow n'existent que dans ta session Python actuelle. Après le redémarrage d'un noyau de notebook, passe l'ID de job enregistré à client.job(), ou liste tes jobs récents avec 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() renvoie tes 10 jobs les plus récents, du plus récent au plus ancien. Filtre-les avec function, status (d'après le tableau des statuts) ou created_after (un horodatage ISO 8601), et parcours-les avec limit et offset. Pour lister les workflows que tu as téléversés, appelle client.list().

Le tableau Workloads sur IBM Quantum Platform liste aussi tes jobs, avec leur statut, leur profil de calcul et leurs durées d'exécution.

Limite de temps

Un workflow géré annule un job qui s'exécute pendant plus de 24 heures, et le job se termine avec le statut CANCELED. Pendant les longues exécutions, enregistre des points de contrôle dans le stockage de fichiers pour ne pas perdre ton travail.

Dépanner les jobs

Problème	Cause
run() échoue avec Code: 429 et Active job limit reached.	Tu as atteint la limite de jobs actifs. Attends la fin de certains jobs, ou annule les jobs dont tu n'as plus besoin.
Le job reste QUEUED, et run() a averti que your account, plan or subscription is blocked.	Le job ne démarre pas tant que le blocage de ton compte, plan ou abonnement n'est pas levé. Contacte ton administrateur.
job.logs() renvoie No logs yet.	Le job est en file d'attente ou vient de démarrer. Attends environ 15 secondes.
job.result() renvoie {}	Le job n'est pas terminé, a été annulé ou n'a enregistré aucun résultat. Consulte Enregistrer les résultats.
job.error_message() renvoie {}	Le workflow a levé une exception sans signaler d'erreur. Lis la trace d'appels avec job.logs().
Le job est CANCELED, mais tu ne l'as pas annulé	Le job a atteint la limite de 24 heures.

Étapes suivantes

Recommandations

• Écrire des logs et gérer les erreurs pour configurer les logs et les erreurs qu'un job signale.
• Exécuter un balayage de paramètres pour exécuter le workflow une fois pour chaque jeu d'entrées.