Quanten-Jobs aus einem Workflow ausführen

Paketversionen

Der Code auf dieser Seite wurde mit den folgenden Anforderungen entwickelt. Wir empfehlen, diese Versionen oder neuere zu verwenden.

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

Ein Quanten-klassischer Workflow läuft als ein verwalteter Workflow-Job. Seine klassischen Phasen laufen auf deinem Compute-Profil, und die Primitive-Jobs, die er an eine IBM Quantum® QPU sendet, laufen auf deiner Instanz und werden auf deine Nutzung angerechnet. Von diesem einen Job aus kannst du jede Phase melden, die QPU-Jobs finden und sie abbrechen. Wie du einen Job startest und seinen Status und seine Logs prüfst, erfährst du unter Einen verwalteten Workflow ausführen.

Bevor du beginnst, stelle deinen ersten verwalteten Workflow bereit.

Den Runtime-Service erstellen

Erstelle den Runtime-Service in deinem Workflow-Code mit get_runtime_service() und verwende ihn so, wie du lokal QiskitRuntimeService verwendest. Der folgende Workflow führt einen Bell-Circuit mit dem Sampler-Primitive aus, meldet jede Phase mit update_status() und gibt die Counts zurück:

# 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() gibt einen QiskitRuntimeService zurück, der für deinen verwalteten Workflow-Job eingerichtet ist, und bietet Folgendes:

• Zugangsdaten. Der Runtime-Service verwendet den API-Schlüssel, die Instanz und den Kanal, mit denen du run() aufgerufen hast, sodass der Workflow-Code keine Zugangsdaten braucht.

• Nachverfolgung. Ein Workflow zeichnet jeden Primitive-Job, der über diesen Runtime-Service läuft, unter dem verwalteten Workflow-Job auf, einschließlich Sampler-, Estimator- und Executor-Jobs sowie Jobs in einer Session oder einem Batch. Du kannst sie mit job.runtime_jobs() auflisten.

• Abbruch. Wenn du job.cancel() aufrufst, bricht der verwaltete Workflow auch die Sessions, Batches und Primitive-Jobs des Workflows ab, die noch laufen.

Ein verwalteter Workflow verfolgt die Jobs eines Service, den du mit QiskitRuntimeService() erstellst, weder nach noch bricht er sie ab, obwohl die Jobs weiterlaufen. Wenn der Workflow einen anderen Service verwenden muss, brich dessen Jobs in einem SIGTERM handler ab.

Primitives in einer Session oder einem Batch ausführen

Ein verwalteter Workflow verfolgt auch Primitive-Jobs nach, die in einer Session oder einem Batch laufen. Erstelle die Session oder den Batch aus einem Backend, das der Runtime-Service zurückgegeben hat, und erstelle die Primitives innerhalb des with-Blocks. Session funktioniert auf die gleiche Weise:

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]

Fortschritt mit update_status() melden

Der Workflow unter Den Runtime-Service erstellen ruft update_status() zu Beginn jeder Phase auf, damit der Job-Status zeigt, wo der Workflow steht, etwa RUNNING: WAITING_FOR_QPU. Übergib eine der folgenden Job-Konstanten:

Aufruf im Workflow	Job-Status
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() gibt True zurück, wenn der Workflow den Status geändert hat. Es gibt False zurück und schreibt eine Warnung in die Logs, wenn der Wert keine dieser Konstanten ist, wenn der Job nicht mehr aktiv ist oder wenn der Code außerhalb eines Jobs läuft. Es löst keinen Fehler aus, sodass ein Tippfehler den Workflow nicht stoppt.

Für den Fortschritt innerhalb einer Phase, etwa die Iterationen eines Optimierers, schreibe Log-Zeilen.

Die QPU-Jobs eines Laufs finden

Wähle in deinem lokalen Notebook eine QPU, auf die deine Instanz zugreifen kann. client.backends() listet sie auf, und client.least_busy() gibt diejenige mit den wenigsten wartenden Jobs zurück:

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

Lade den Workflow hoch und übergib beim Ausführen den Namen des Backends:

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

Hole nach dem Ende des Jobs die IDs der QPU-Jobs des Workflows aus dem verwalteten Workflow-Job:

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

Jede ID ist eine reguläre Primitive-Job-ID. Wie du den Status und die Ergebnisse eines Jobs liest, erfährst du unter Jobergebnisse zu einem späteren Zeitpunkt abrufen. Wenn der Workflow Primitives in einer Session oder einem Batch ausgeführt hat, liste die Session- und Batch-IDs mit runtime_sessions() auf und übergib eine davon an runtime_jobs(), um nur die darin enthaltenen Jobs zu erhalten:

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

Das job-Objekt existiert nur in dem Notebook, das den Lauf gestartet hat. Hole den Job in einem neuen Notebook erneut mit client.job("<your-job-id>") und rufe dann runtime_jobs() darauf auf. Siehe Einen Job aus einer neuen Session finden.

Die QPU-Nutzung und die Laufzeiten jedes QPU-Jobs findest du unter Job-Metriken anzeigen. Die Nutzung deiner gesamten Instanz im aktuellen Zeitraum erhältst du mit client.usage(). Wie lange jede Phase des verwalteten Workflow-Jobs gedauert hat, erfährst du unter Prüfen, wie lange jede Phase gedauert hat.

Verknüpfte QPU-Jobs abbrechen

Wenn du einen verwalteten Workflow-Job mit job.cancel() abbrichst, versucht der Workflow auch, die QPU-Arbeit abzubrechen, die er gestartet hat:

• Sessions und Batches. Der Workflow bricht jede seiner Sessions und Batches ab, wodurch auch deren Jobs in der Warteschlange abgebrochen werden.

• Andere Primitive-Jobs. Der Workflow bricht jeden seiner anderen Primitive-Jobs ab, die noch in der Warteschlange stehen oder laufen.

job.cancel() gibt für jeden QPU-Job und jede Session eine Meldung zurück, sodass du prüfen kannst, was abgebrochen wurde:

run_workflow_remotely.ipynb

print(job.cancel())

Bei einem Workflow, der in einem Batch lief, sieht die Meldung wie folgt aus:

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

Wenn der Client aus deinen lokalen Zugangsdaten keinen QiskitRuntimeService erstellen kann, bricht er nur den verwalteten Workflow-Job ab. Brich die QPU-Jobs selbst mit ihren IDs aus job.runtime_jobs() ab.

Wenn ein Workflow einen Job selbst stoppt, zum Beispiel weil er das 24-Stunden-Limit erreicht hat, bricht er seine QPU-Jobs nicht ab. Brich sie mit ihren IDs aus job.runtime_jobs() ab.

QPU-Jobs in einem Workflow beheben

Problem	Ursache
get_runtime_service() löst KeyError: 'QISKIT_IBM_CHANNEL' aus.	Der Code läuft in deinem lokalen Notebook, nicht in einem verwalteten Workflow-Job. Verwende lokal QiskitRuntimeService().
update_status() gibt False zurück, und die Logs zeigen 'sub_status' not provided or is not valid.	Der Wert ist keine der Job-Konstanten, oder der Job ist nicht mehr aktiv.
job.runtime_jobs() gibt [] zurück.	Der Workflow hat seinen Service mit QiskitRuntimeService() erstellt, oder er hat noch kein Primitive ausgeführt.
Die Jobs einer Session werden nicht nachverfolgt.	Die Session wurde aus einem Backend eines anderen Service erstellt. Hole das Backend aus get_runtime_service().
Die Workflow-Logs zeigen Could not associate runtime job with serverless job.	Der Workflow konnte einen Primitive-Job nicht aufzeichnen. Der Primitive-Job läuft trotzdem, wird aber nicht nachverfolgt.
job.cancel() zeigt No QiskitRuntimeService can be associated to the given token and instance.	Der Client konnte aus deinen lokalen Zugangsdaten keinen QiskitRuntimeService erstellen und hat daher nur den verwalteten Workflow-Job abgebrochen.
client.job() schlägt mit Job [...] not found fehl.	Die Job-ID ist falsch, oder der Job gehört einem anderen Benutzer.

Nächste Schritte

Empfehlungen

• Führe einen verwalteten Workflow aus, um Status, Ergebnis und Logs eines Jobs zu prüfen und die Ursache eines Fehlers zu finden.