Einen verwalteten Workflow ausführen

Paketversionen

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

qiskit-serverless~=0.38.0

Ein Job eines verwalteten Workflows funktioniert wie ein Batch-Job, den du an einen HPC-Scheduler übergibst. run() reicht den Job ein und gibt sofort ein Job-Objekt zurück. Verwende das Job-Objekt für die Aufgaben, die auf einem Cluster squeue, die Ausgabedatei des Jobs und scancel abdecken. Der Job läuft weiter, nachdem du dein Notebook geschlossen hast.

Die Beispiele auf dieser Seite verwenden einen kleinen Workflow, der einen Winkel, theta, liest und fehlschlägt, wenn der Winkel außerhalb des zulässigen Bereichs liegt. Lade ihn hoch, wie in Deinen ersten verwalteten Workflow bereitstellen beschrieben.

# 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

Einen Job starten

Rufe den Workflow über seinen Titel ab und rufe run() mit den Argumenten auf, die der Workflow liest:

# 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() kehrt zurück, nachdem der Workflow den Job in die Warteschlange gestellt hat. Speichere die Job-ID, damit du den Job aus einer neuen Session wiederfinden kannst.

• Wie du festlegst, welche Argumente ein Workflow akzeptiert, und sie prüfst, bevor der Job startet, siehe Ein- und Ausgaben definieren.

• Um den Job mit einer anderen Rechengröße auszuführen, übergib function_size. Siehe Für jeden Lauf eine Größe wählen.

Einem Job folgen

Den Status prüfen

Unmittelbar nach dem Aufruf von run() ist der Status normalerweise QUEUED: run_workflow_remotely.ipynb

job.status()

'QUEUED'

Status	Bedeutung
QUEUED	Der Job wartet auf einen Rechenplatz.
INITIALIZING	Der Workflow startet eine Umgebung mit dem Rechenprofil des Jobs.
RUNNING	Der Entrypoint des Workflows läuft.
RUNNING: <stage>	Der Entrypoint hat mit update_status() eine Phase gemeldet, zum Beispiel RUNNING: WAITING_FOR_QPU.
CANCELING	Du hast den Job abgebrochen, und der Workflow stoppt ihn gerade.
DONE	Der Entrypoint wurde erfolgreich beendet.
ERROR	Der Entrypoint hat einen Fehler ausgelöst oder wurde mit einem Exit-Code ungleich null beendet. Siehe Die Ursache eines Fehlers finden.
CANCELED	Du hast den Job abgebrochen, oder er hat das 24-Stunden-Limit erreicht.

DONE, ERROR und CANCELED sind Endzustände, und job.in_terminal_state() gibt für sie True zurück. Eine Statusschleife, die auf einen anderen Wert wartet, etwa SUCCEEDED, endet nie.

Auf das Ende des Jobs warten

Um zu warten, bis der Job endet, rufe wait_for_final_state() auf. Es gibt den endgültigen Status zurück, und mit verbose=True gibt es jede Statusänderung aus. Wenn du timeout in Sekunden festlegst, löst wait_for_final_state() einen TimeoutError aus, sobald die Zeit abläuft, bevor der Job endet. Der Job läuft weiter:

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() prüft den Status nach 1, 2, 4, 8, 16 und 32 Sekunden und danach alle 60 Sekunden. So erkennt es das Ende eines kurzen Jobs innerhalb von Sekunden und stellt bei einem langen Job nur wenige Anfragen. Um stattdessen in einem festen Intervall zu prüfen, übergib poll_interval in Sekunden.

Die Logs lesen

Alles, was der Workflow in die Standardausgabe oder in die Standardfehlerausgabe schreibt, landet in den Logs des Jobs. Du kannst sie mit job.logs() lesen, während der Job läuft, noch bevor er ein Ergebnis hat:

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

• Der Workflow lädt neue Logzeilen etwa alle 15 Sekunden hoch und ein letztes Mal, wenn der Job endet oder abgebrochen wird. Unmittelbar nach dem Start eines Jobs gibt job.logs() No logs yet. zurück.

• Der Workflow behält für jeden Job die neuesten 50 MB an Logs und fügt oben eine Zeile hinzu, die besagt, dass die Logs gekürzt wurden.

Was dein Workflow protokolliert, legst du wie unter Logs schreiben beschrieben fest. Um nur die Zeilen anzuzeigen, die zu einem regulären Ausdruck passen, übergib include an job.filtered_logs(). Um Zeilen auszublenden, übergib exclude.

Das Ergebnis abrufen

job.result() wartet, bis der Job endet, und gibt dann den Wert zurück, den der Workflow an save_result() übergeben hat: 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

Standardmäßig prüft job.result() den Status alle 30 Sekunden und wartet ohne Zeitlimit. Übergib maxwait, um die Anzahl der Prüfungen zu begrenzen. Zum Beispiel wartet job.result(maxwait=40) etwa 20 Minuten. Um einmal zu prüfen, ohne zu warten, übergib wait=False. Wenn job.result() zu warten aufhört, bevor der Job endet, gibt es {} zurück.

Prüfen, wie lange jede Phase gedauert hat

job.events() listet jede Status- und Phasenänderung eines Jobs mit ihrer Zeit auf, sodass du sehen kannst, wie lange der Job auf Rechenleistung gewartet hat und wie lange jede Phase dauerte: 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

Die Ereignisse verwenden die gespeicherten Statusnamen. PENDING entspricht INITIALIZING, und SUCCEEDED entspricht DONE. Zur QPU-Nutzung eines Laufs siehe Die QPU-Jobs eines Laufs finden.

Die Ursache eines Fehlers finden

Wenn der Workflow eine Exception auslöst, endet der Job mit dem Status ERROR. Führe den Workflow zum Beispiel mit einem Winkel aus, den er ablehnt: 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'

Die Logs zeigen den 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") == []

Wenn der Workflow Fehler mit send_error() meldet, lies sie mit job.events(type="ERROR") und ihre Meldungen mit job.error_message(). Bei einem Job ohne gemeldete Fehler gibt job.events(type="ERROR") eine leere Liste zurück, und job.error_message() gibt {} zurück.

run_workflow_remotely.ipynb

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

Jobs verwalten

Einen Job abbrechen

Breche einen Job mit job.cancel() ab. Der Status wechselt zu CANCELING, während der Workflow den Job stoppt, und danach zu 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() gibt eine Meldung über den Job und über seine verknüpften QPU-Jobs zurück, die es ebenfalls abzubrechen versucht. Bei einem laufenden Job sendet der Workflow SIGTERM an deinen Entrypoint. Wie du Teilergebnisse speicherst, siehe Aufräumen, wenn ein Job abgebrochen wird. Wenn der Job bereits beendet ist, gibt job.cancel() Job already in terminal state. zurück und ändert nichts.

Einen Job aus einer neuen Session wiederfinden

Das Job-Objekt und der Workflow existieren nur in deiner aktuellen Python-Session. Nachdem du einen Notebook-Kernel neu gestartet hast, übergib die gespeicherte Job-ID an client.job() oder liste deine letzten Jobs mit client.jobs() auf: 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() gibt deine 10 neuesten Jobs zurück, den neuesten zuerst. Filtere mit function, status (aus der Statustabelle) oder created_after (ein ISO-8601-Zeitstempel) und blättere mit limit und offset durch die Liste. Um die Workflows aufzulisten, die du hochgeladen hast, rufe client.list() auf.

Die Tabelle Workloads auf IBM Quantum Platform listet ebenfalls deine Jobs auf, mit ihrem Status, Rechenprofil und ihren Laufzeiten.

Zeitlimit

Ein verwalteter Workflow bricht einen Job ab, der länger als 24 Stunden läuft, und der Job endet mit dem Status CANCELED. Bei langen Läufen speichere Checkpoints im Dateispeicher, damit du keine Arbeit verlierst.

Probleme bei Jobs beheben

Problem	Ursache
run() schlägt mit Code: 429 und Active job limit reached. fehl.	Du hast das Limit für aktive Jobs erreicht. Warte, bis einige Jobs enden, oder breche Jobs ab, die du nicht mehr brauchst.
Der Job bleibt QUEUED, und run() hat gewarnt: your account, plan or subscription is blocked.	Der Job startet erst, wenn die Sperre deines Kontos, Plans oder Abonnements aufgehoben ist. Wende dich an deine Administratorin oder deinen Administrator.
job.logs() gibt No logs yet. zurück.	Der Job steht in der Warteschlange oder wurde gerade gestartet. Warte etwa 15 Sekunden.
job.result() gibt {} zurück.	Der Job ist nicht beendet, wurde abgebrochen oder hat kein Ergebnis gespeichert. Siehe Ergebnisse speichern.
job.error_message() gibt {} zurück.	Der Workflow hat eine Exception ausgelöst, ohne einen Fehler zu melden. Lies den Traceback mit job.logs().
Der Job ist CANCELED, aber du hast ihn nicht abgebrochen.	Der Job hat das 24-Stunden-Limit erreicht.

Nächste Schritte

Empfehlungen

• Logs schreiben und Fehler behandeln, um die Logs und Fehler einzurichten, die ein Job meldet.
• Einen Parameter-Sweep ausführen, um den Workflow einmal für jeden Satz von Eingaben auszuführen.