Rulează joburi cuantice dintr-un workflow

Versiuni de pachete

Codul de pe această pagină a fost dezvoltat folosind următoarele cerințe. Recomandăm să folosești aceste versiuni sau unele mai noi.

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

Un workflow cuantic-clasic rulează ca un singur job de workflow gestionat. Etapele sale clasice rulează pe profilul tău de calcul, iar joburile primitive pe care le trimite către un QPU IBM Quantum® rulează pe instanța ta și se contorizează în utilizarea ta. Din acel singur job poți raporta fiecare etapă, găsi joburile QPU și le poți anula. Pentru a porni un job și a-i verifica starea și logurile, vezi Rulează un workflow gestionat.

Înainte de a începe, implementează primul tău workflow gestionat.

Creează serviciul Runtime

Creează serviciul Runtime în codul workflow-ului tău cu get_runtime_service() și folosește-l în același mod în care folosești QiskitRuntimeService local. Următorul workflow rulează un circuit Bell cu primitiva Sampler, raportează fiecare etapă cu update_status() și returnează numărătorile:

# 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() returnează un QiskitRuntimeService configurat pentru jobul tău de workflow gestionat și oferă următoarele:

• Credențiale. Serviciul Runtime folosește cheia API, instanța și canalul pe care le-ai folosit pentru a apela run(), deci codul workflow-ului nu are nevoie de credențiale.

• Urmărire. Un workflow înregistrează fiecare job primitiv care rulează prin acest serviciu Runtime sub jobul de workflow gestionat, inclusiv joburile Sampler, Estimator și Executor, precum și joburile dintr-o sesiune sau un batch. Le poți lista cu job.runtime_jobs().

• Anulare. Când apelezi job.cancel(), workflow-ul gestionat anulează și sesiunile, batch-urile și joburile primitive ale workflow-ului care încă rulează.

Un workflow gestionat nu urmărește și nu anulează joburile unui serviciu pe care îl creezi cu QiskitRuntimeService(), deși joburile rulează în continuare. Dacă workflow-ul trebuie să folosească un alt serviciu, anulează joburile acestuia într-un handler SIGTERM.

Rulează primitive într-o sesiune sau un batch

Un workflow gestionat urmărește și joburile primitive care rulează într-o sesiune sau un batch. Creează sesiunea sau batch-ul dintr-un backend returnat de serviciul Runtime și creează primitivele în interiorul blocului with. Session funcționează în același mod:

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]

Raportează progresul cu update_status()

Workflow-ul din Creează serviciul Runtime apelează update_status() la începutul fiecărei etape, astfel încât starea jobului arată unde se află workflow-ul, de exemplu RUNNING: WAITING_FOR_QPU. Transmite una dintre următoarele constante Job:

Apel în workflow	Starea jobului
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() returnează True dacă workflow-ul a schimbat starea. Returnează False și scrie un avertisment în loguri dacă valoarea nu este una dintre aceste constante, dacă jobul nu mai este activ sau dacă codul rulează în afara unui job. Nu ridică o eroare, deci o greșeală de tastare nu oprește workflow-ul.

Pentru progresul în interiorul unei etape, cum ar fi iterațiile unui optimizator, scrie linii de log.

Găsește joburile QPU ale unei rulări

În notebookul tău local, alege un QPU la care instanța ta are acces. client.backends() le listează, iar client.least_busy() îl returnează pe cel cu cele mai puține joburi în așteptare:

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

Încarcă workflow-ul și transmite numele backend-ului când îl rulezi:

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

După ce jobul se termină, obține ID-urile joburilor QPU ale workflow-ului din jobul de workflow gestionat:

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

Fiecare ID este un ID obișnuit de job primitiv. Pentru a citi starea și rezultatele unui job, vezi Preia rezultatele unui job mai târziu. Dacă workflow-ul a rulat primitive într-o sesiune sau un batch, listează ID-urile sesiunilor și ale batch-urilor cu runtime_sessions() și transmite unul către runtime_jobs() pentru a obține doar joburile din acesta:

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

Obiectul job există doar în notebookul care a pornit rularea. Într-un notebook nou, obține din nou jobul cu client.job("<your-job-id>"), apoi apelează runtime_jobs() pe el. Vezi Găsește un job dintr-o sesiune nouă.

Pentru utilizarea QPU și timpii de rulare ai fiecărui job QPU, vezi Vizualizează metricile jobului. Pentru utilizarea întregii tale instanțe în perioada curentă, apelează client.usage(). Pentru a vedea cât a durat fiecare etapă a jobului de workflow gestionat, vezi Verifică cât a durat fiecare etapă.

Anulează joburile QPU asociate

Când anulezi un job de workflow gestionat cu job.cancel(), workflow-ul încearcă să anuleze și munca QPU pe care a pornit-o:

• Sesiuni și batch-uri. Workflow-ul anulează fiecare sesiune și fiecare batch al workflow-ului, ceea ce anulează și joburile lor din coadă.

• Alte joburi primitive. Workflow-ul anulează fiecare dintre celelalte joburi primitive ale workflow-ului care sunt încă în coadă sau rulează.

job.cancel() returnează un mesaj pentru fiecare job QPU și sesiune, astfel încât poți verifica ce a fost anulat:

run_workflow_remotely.ipynb

print(job.cancel())

Pentru un workflow care a rulat într-un batch, mesajul arată astfel:

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

Dacă clientul nu poate crea un QiskitRuntimeService din credențialele tale locale, anulează doar jobul de workflow gestionat. Anulează tu joburile QPU cu ID-urile lor din job.runtime_jobs().

Când un workflow oprește el însuși un job, de exemplu pentru că a atins limita de 24 de ore, nu își anulează joburile QPU. Anulează-le cu ID-urile lor din job.runtime_jobs().

Depanează joburile QPU dintr-un workflow

Problemă	Cauză
get_runtime_service() ridică KeyError: 'QISKIT_IBM_CHANNEL'.	Codul rulează în notebookul tău local, nu într-un job de workflow gestionat. Folosește QiskitRuntimeService() local.
update_status() returnează False, iar logurile arată 'sub_status' not provided or is not valid.	Valoarea nu este una dintre constantele Job sau jobul nu mai este activ.
job.runtime_jobs() returnează [].	Workflow-ul și-a creat serviciul cu QiskitRuntimeService() sau nu a rulat încă nicio primitivă.
Joburile unei sesiuni nu sunt urmărite.	Sesiunea a fost creată dintr-un backend al unui alt serviciu. Obține backend-ul din get_runtime_service().
Logurile workflow-ului arată Could not associate runtime job with serverless job.	Workflow-ul nu a putut înregistra un job primitiv. Jobul primitiv rulează în continuare, dar nu este urmărit.
job.cancel() arată No QiskitRuntimeService can be associated to the given token and instance.	Clientul nu a putut crea un QiskitRuntimeService din credențialele tale locale, deci a anulat doar jobul de workflow gestionat.
client.job() eșuează cu Job [...] not found.	ID-ul jobului este greșit sau jobul aparține altui utilizator.

Pașii următori

Recomandări

• Rulează un workflow gestionat pentru a verifica starea, rezultatul și logurile unui job și pentru a găsi cauza unei erori.