Exécuter des jobs quantiques à partir d'un workflow

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[all]~=2.6.1
qiskit-ibm-runtime~=0.51.0
qiskit-serverless~=0.38.0

Un workflow quantique-classique s'exécute comme un seul job de workflow géré. Ses étapes classiques s'exécutent sur ton profil de calcul, et les jobs de primitives qu'il envoie à un QPU IBM Quantum® s'exécutent sur ton instance et comptent dans ton utilisation. À partir de ce job unique, tu peux signaler chaque étape, trouver les jobs QPU et les annuler. Pour lancer un job et vérifier son statut et ses journaux, consulte Exécuter un workflow géré.

Avant de commencer, déploie ton premier workflow géré.

Créer le service Runtime

Crée le service Runtime dans le code de ton workflow avec get_runtime_service(), et utilise-le de la même manière que tu utilises QiskitRuntimeService en local. Le workflow suivant exécute un circuit de Bell avec la primitive Sampler, signale chaque étape avec update_status() et renvoie les comptages :

# 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() renvoie un QiskitRuntimeService configuré pour ton job de workflow géré, et il fournit ce qui suit :

• Identifiants. Le service Runtime utilise la clé API, l'instance et le canal que tu as utilisés pour appeler run(), de sorte que le code du workflow n'a besoin d'aucun identifiant.

• Suivi. Un workflow enregistre chaque job de primitive exécuté via ce service Runtime sous le job de workflow géré, y compris les jobs Sampler, Estimator et Executor, ainsi que les jobs dans une session ou un batch. Tu peux les lister avec job.runtime_jobs().

• Annulation. Lorsque tu appelles job.cancel(), le workflow géré annule aussi les sessions, les batches et les jobs de primitives du workflow qui sont encore en cours d'exécution.

Un workflow géré ne suit pas et n'annule pas les jobs d'un service que tu crées avec QiskitRuntimeService(), bien que les jobs s'exécutent quand même. Si le workflow doit utiliser un autre service, annule ses jobs dans un gestionnaire SIGTERM.

Exécuter des primitives dans une session ou un batch

Un workflow géré suit aussi les jobs de primitives qui s'exécutent dans une session ou un batch. Crée la session ou le batch à partir d'un backend renvoyé par le service Runtime, et crée les primitives à l'intérieur du bloc with. Session fonctionne de la même manière :

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]

Signaler la progression avec update_status()

Le workflow de Créer le service Runtime appelle update_status() au début de chaque étape, afin que le statut du job indique où en est le workflow, par exemple RUNNING: WAITING_FOR_QPU. Passe l'une des constantes Job suivantes :

Appel dans le workflow	Statut du job
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() renvoie True si le workflow a changé le statut. Il renvoie False et écrit un avertissement dans les journaux si la valeur n'est pas l'une de ces constantes, si le job n'est plus actif ou si le code s'exécute en dehors d'un job. Il ne lève pas d'erreur, donc une faute de frappe n'arrête pas le workflow.

Pour la progression au sein d'une étape, par exemple les itérations d'un optimiseur, écris des lignes de journal.

Trouver les jobs QPU d'une exécution

Dans ton notebook local, choisis un QPU auquel ton instance peut accéder. client.backends() les liste, et client.least_busy() renvoie celui qui a le moins de jobs en attente :

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']

Téléverse le workflow, et passe le nom du backend lorsque tu l'exécutes :

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

Une fois le job terminé, récupère les ID des jobs QPU du workflow à partir du job de workflow géré :

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

Chaque ID est un ID de job de primitive ordinaire. Pour lire le statut et les résultats d'un job, consulte Récupérer les résultats d'un job plus tard. Si le workflow a exécuté des primitives dans une session ou un batch, liste les ID de session et de batch avec runtime_sessions(), et passes-en un à runtime_jobs() pour obtenir uniquement les jobs qu'il contient :

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']

L'objet job n'existe que dans le notebook qui a lancé l'exécution. Dans un nouveau notebook, récupère de nouveau le job avec client.job("<your-job-id>"), puis appelle runtime_jobs() dessus. Consulte Retrouver un job depuis une nouvelle session.

Pour l'utilisation du QPU et les temps d'exécution de chaque job QPU, consulte Afficher les métriques d'un job. Pour l'utilisation de l'ensemble de ton instance sur la période en cours, appelle client.usage(). Pour voir combien de temps a pris chaque étape du job de workflow géré, consulte Vérifier la durée de chaque étape.

Annuler les jobs QPU liés

Lorsque tu annules un job de workflow géré avec job.cancel(), le workflow essaie aussi d'annuler le travail QPU qu'il a lancé :

• Sessions et batches. Le workflow annule chacune de ses sessions et chacun de ses batches, ce qui annule aussi leurs jobs en file d'attente.

• Autres jobs de primitives. Le workflow annule chacun de ses autres jobs de primitives qui sont encore en file d'attente ou en cours d'exécution.

job.cancel() renvoie un message pour chaque job QPU et chaque session, ce qui te permet de vérifier ce qui a été annulé :

run_workflow_remotely.ipynb

print(job.cancel())

Pour un workflow exécuté dans un batch, le message ressemble à ce qui suit :

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

Si le client ne peut pas créer un QiskitRuntimeService à partir de tes identifiants locaux, il n'annule que le job de workflow géré. Annule toi-même les jobs QPU avec leurs ID obtenus via job.runtime_jobs().

Lorsqu'un workflow arrête lui-même un job, par exemple parce qu'il a atteint la limite de 24 heures, il n'annule pas ses jobs QPU. Annule-les avec leurs ID obtenus via job.runtime_jobs().

Dépanner les jobs QPU dans un workflow

Problème	Cause
get_runtime_service() lève KeyError: 'QISKIT_IBM_CHANNEL'.	Le code s'exécute dans ton notebook local, et non dans un job de workflow géré. Utilise QiskitRuntimeService() en local.
update_status() renvoie False, et les journaux affichent 'sub_status' not provided or is not valid.	La valeur n'est pas l'une des constantes Job, ou le job n'est plus actif.
job.runtime_jobs() renvoie [].	Le workflow a créé son service avec QiskitRuntimeService(), ou il n'a pas encore exécuté de primitive.
Les jobs d'une session ne sont pas suivis.	La session a été créée à partir d'un backend d'un autre service. Obtiens le backend avec get_runtime_service().
Les journaux du workflow affichent Could not associate runtime job with serverless job.	Le workflow n'a pas pu enregistrer un job de primitive. Le job de primitive s'exécute quand même, mais il n'est pas suivi.
job.cancel() affiche No QiskitRuntimeService can be associated to the given token and instance.	Le client n'a pas pu créer un QiskitRuntimeService à partir de tes identifiants locaux ; il n'a donc annulé que le job de workflow géré.
client.job() échoue avec Job [...] not found.	L'ID du job est incorrect, ou le job appartient à un autre utilisateur.

Étapes suivantes

Recommandations

• Exécuter un workflow géré pour vérifier le statut, le résultat et les journaux d'un job, et trouver la cause d'une erreur.