Rulează un flux de lucru gestionat

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-serverless~=0.38.0

Un job de flux de lucru gestionat funcționează ca un job batch pe care îl trimiți unui planificator HPC. run() trimite jobul și returnează imediat un obiect Job. Folosește obiectul Job pentru sarcinile pe care într-un cluster le acoperă squeue, fișierul de ieșire al jobului și scancel. Jobul continuă să ruleze după ce închizi notebook-ul.

Exemplele de pe această pagină folosesc un flux de lucru mic care citește un unghi, theta, și eșuează dacă unghiul este în afara intervalului. Încarcă-l așa cum este descris în Implementează primul tău flux de lucru gestionat.

# 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

Pornește un job

Obține fluxul de lucru după titlul său și apelează run() cu argumentele pe care le citește fluxul de lucru:

# 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() returnează după ce fluxul de lucru pune jobul în coadă. Salvează ID-ul jobului pentru a putea găsi jobul dintr-o sesiune nouă.

• Pentru a alege ce argumente acceptă un flux de lucru și a le verifica înainte de pornirea jobului, consultă Definește intrările și ieșirile.

• Pentru a rula jobul cu o altă dimensiune de calcul, transmite function_size. Consultă Alege o dimensiune pentru fiecare rulare.

Urmărește un job

Verifică starea

Imediat după ce apelezi run(), starea este de obicei QUEUED: run_workflow_remotely.ipynb

job.status()

'QUEUED'

Stare	Semnificație
QUEUED	Jobul așteaptă un slot de calcul.
INITIALIZING	Fluxul de lucru pornește un mediu cu profilul de calcul al jobului.
RUNNING	Punctul de intrare al fluxului de lucru rulează.
RUNNING: <stage>	Punctul de intrare a raportat o etapă cu update_status(), cum ar fi RUNNING: WAITING_FOR_QPU.
CANCELING	Ai anulat jobul, iar fluxul de lucru îl oprește.
DONE	Punctul de intrare s-a încheiat cu succes.
ERROR	Punctul de intrare a generat o eroare sau s-a încheiat cu un cod diferit de zero. Consultă Găsește cauza unei erori.
CANCELED	Ai anulat jobul sau acesta a atins limita de 24 de ore.

DONE, ERROR și CANCELED sunt finale, iar job.in_terminal_state() returnează True pentru ele. O buclă de stare care așteaptă o altă valoare, cum ar fi SUCCEEDED, nu se termină niciodată.

Așteaptă încheierea jobului

Pentru a aștepta până când jobul se încheie, apelează wait_for_final_state(). Returnează starea finală, iar cu verbose=True afișează fiecare schimbare de stare. Dacă setezi timeout în secunde, wait_for_final_state() ridică un TimeoutError atunci când timpul expiră înainte ca jobul să se încheie. Jobul continuă să ruleze:

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() verifică starea după 1, 2, 4, 8, 16 și 32 de secunde, apoi la fiecare 60 de secunde, astfel încât observă sfârșitul unui job scurt în câteva secunde și face puține cereri pentru un job lung. Pentru a verifica în schimb la un interval fix, transmite poll_interval în secunde.

Citește jurnalele

Tot ce scrie fluxul de lucru la ieșirea standard sau la eroarea standard ajunge în jurnalele jobului. Le poți citi cu job.logs() cât timp jobul rulează, înainte să aibă un rezultat:

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

• Fluxul de lucru încarcă linii noi de jurnal aproximativ la fiecare 15 secunde și încă o dată la final, când jobul se încheie sau este anulat. Imediat după pornirea unui job, job.logs() returnează No logs yet.

• Fluxul de lucru păstrează ultimii 50 MB de jurnale pentru fiecare job și adaugă sus o linie care spune că jurnalele au fost trunchiate.

Pentru a decide ce înregistrează fluxul tău de lucru în jurnale, consultă Scrie jurnale. Pentru a afișa doar liniile care se potrivesc cu o expresie regulată, transmite include către job.filtered_logs(). Pentru a ascunde linii, transmite exclude.

Obține rezultatul

job.result() așteaptă până când jobul se încheie, apoi returnează valoarea pe care fluxul de lucru a transmis-o către 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

Implicit, job.result() verifică starea la fiecare 30 de secunde și așteaptă fără limită de timp. Transmite maxwait pentru a limita numărul de verificări. De exemplu, job.result(maxwait=40) așteaptă aproximativ 20 de minute. Pentru a verifica o singură dată, fără a aștepta, transmite wait=False. Dacă job.result() nu mai așteaptă înainte ca jobul să se încheie, returnează {}.

Verifică cât a durat fiecare etapă

job.events() listează fiecare schimbare de stare și de etapă a unui job împreună cu momentul ei, astfel încât poți vedea cât a așteptat jobul după resurse de calcul și cât a durat fiecare etapă: 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

Evenimentele folosesc numele stocate ale stărilor. PENDING este INITIALIZING, iar SUCCEEDED este DONE. Pentru utilizarea QPU a unei rulări, consultă Găsește joburile QPU ale unei rulări.

Găsește cauza unei erori

Când fluxul de lucru ridică o excepție, jobul se încheie cu starea ERROR. De exemplu, rulează fluxul de lucru cu un unghi pe care îl respinge: 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'

Jurnalele arată traceback-ul: 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") == []

Dacă fluxul de lucru raportează erori cu send_error(), citește-le cu job.events(type="ERROR"), iar mesajele lor cu job.error_message(). Pentru un job fără erori raportate, job.events(type="ERROR") returnează o listă goală, iar job.error_message() returnează {}.

run_workflow_remotely.ipynb

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

Gestionează joburile

Anulează un job

Anulează un job cu job.cancel(). Starea se schimbă în CANCELING cât timp fluxul de lucru oprește jobul, apoi în 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() returnează un mesaj despre job și despre joburile QPU asociate, pe care încearcă să le anuleze și pe ele. Pentru un job care rulează, fluxul de lucru trimite SIGTERM punctului tău de intrare. Pentru a salva rezultate parțiale, consultă Curăță când un job este anulat. Dacă jobul s-a încheiat deja, job.cancel() returnează Job already in terminal state. și nu schimbă nimic.

Găsește un job dintr-o sesiune nouă

Obiectul Job și fluxul de lucru există doar în sesiunea ta Python curentă. După ce repornești kernelul notebook-ului, transmite ID-ul salvat al jobului către client.job() sau listează joburile tale recente cu 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() returnează cele mai recente 10 joburi ale tale, începând cu cel mai nou. Filtrează-le cu function, status (din tabelul de stări) sau created_after (un timestamp ISO 8601) și parcurge-le pe pagini cu limit și offset. Pentru a lista fluxurile de lucru pe care le-ai încărcat, apelează client.list().

Tabelul Workloads din IBM Quantum Platform listează și el joburile tale, cu starea, profilul de calcul și timpii de rulare.

Limită de timp

Un flux de lucru gestionat anulează un job care rulează mai mult de 24 de ore, iar jobul se încheie cu starea CANCELED. În timpul rulărilor lungi, salvează puncte de control în stocarea de fișiere, astfel încât să nu pierzi munca.

Depanează joburile

Problemă	Cauză
run() eșuează cu Code: 429 și Active job limit reached.	Ai atins limita de joburi active. Așteaptă să se încheie unele joburi sau anulează joburile de care nu mai ai nevoie.
Jobul rămâne QUEUED, iar run() a avertizat că your account, plan or subscription is blocked.	Jobul nu pornește până când blocarea contului, planului sau abonamentului tău nu este ridicată. Contactează administratorul.
job.logs() returnează No logs yet.	Jobul este în coadă sau tocmai a pornit. Așteaptă aproximativ 15 secunde.
job.result() returnează {}	Jobul nu s-a încheiat, a fost anulat sau nu a salvat niciun rezultat. Consultă Salvează rezultatele.
job.error_message() returnează {}	Fluxul de lucru a ridicat o excepție fără a raporta o eroare. Citește traceback-ul cu job.logs().
Jobul este CANCELED, dar nu l-ai anulat tu	Jobul a atins limita de 24 de ore.

Pașii următori

Recomandări

• Scrie jurnale și gestionează erorile pentru a configura jurnalele și erorile pe care le raportează un job.
• Rulează o baleiere de parametri pentru a rula fluxul de lucru o dată pentru fiecare set de intrări.