הרצת workflow מנוהל

גרסאות חבילות

הקוד בדף הזה פותח עם הדרישות הבאות. מומלץ להשתמש בגרסאות האלה או בחדשות יותר.

qiskit-serverless~=0.38.0

job של workflow מנוהל עובד כמו job אצווה ששולחים למתזמן HPC. run() שולח את ה-job ומחזיר מיד אובייקט Job. השתמשו באובייקט Job למשימות ש-squeue, קובץ הפלט של ה-job ו-scancel מכסים באשכול. ה-job ממשיך לרוץ גם אחרי שסוגרים את המחברת.

הדוגמאות בדף הזה משתמשות ב-workflow קטן שקורא זווית אחת, theta, ונכשל אם הזווית מחוץ לטווח. העלו אותו כמתואר בפריסת ה-workflow המנוהל הראשון שלכם.

# 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

הפעלת job

קבלו את ה-workflow לפי הכותרת שלו, וקראו ל-run() עם הארגומנטים ש-workflow קורא:

# 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() חוזר אחרי שה-workflow מכניס את ה-job לתור. שמרו את מזהה ה-job כדי שתוכלו למצוא את ה-job מ-session חדש.

• כדי לבחור אילו ארגומנטים workflow מקבל ולבדוק אותם לפני שה-job מתחיל, ראו הגדרת קלטים ופלטים.

• כדי להריץ את ה-job בגודל חישוב אחר, העבירו function_size. ראו בחירת גודל לכל הרצה.

מעקב אחרי job

בדיקת הסטטוס

מיד אחרי שקוראים ל-run(), הסטטוס הוא בדרך כלל QUEUED: run_workflow_remotely.ipynb

job.status()

'QUEUED'

סטטוס	משמעות
QUEUED	ה-job מחכה לחריץ חישוב.
INITIALIZING	ה-workflow מפעיל סביבה עם פרופיל החישוב של ה-job.
RUNNING	נקודת הכניסה של ה-workflow רצה.
RUNNING: <stage>	נקודת הכניסה דיווחה על שלב באמצעות update_status(), למשל RUNNING: WAITING_FOR_QPU.
CANCELING	ביטלתם את ה-job, וה-workflow עוצר אותו.
DONE	נקודת הכניסה הסתיימה בהצלחה.
ERROR	נקודת הכניסה העלתה שגיאה או הסתיימה בקוד שאינו אפס. ראו מציאת הסיבה לשגיאה.
CANCELED	ביטלתם את ה-job, או שהוא הגיע למגבלת 24 השעות.

DONE, ERROR ו-CANCELED הם סופיים, ו-job.in_terminal_state() מחזיר True עבורם. לולאת סטטוס שמחכה לערך אחר, כמו SUCCEEDED, אף פעם לא מסתיימת.

המתנה לסיום ה-job

כדי לחכות עד שה-job מסתיים, קראו ל-wait_for_final_state(). הוא מחזיר את הסטטוס הסופי, ועם verbose=True הוא מדפיס כל שינוי סטטוס. אם מגדירים timeout בשניות, wait_for_final_state() מעלה TimeoutError כשהזמן נגמר לפני שה-job מסתיים. ה-job ממשיך לרוץ:

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() בודק את הסטטוס אחרי 1, 2, 4, 8, 16 ו-32 שניות, ואז כל 60 שניות, כך שהוא מזהה את סיומו של job קצר תוך שניות ושולח מעט בקשות עבור job ארוך. כדי לבדוק במקום זאת במרווח קבוע, העבירו poll_interval בשניות.

קריאת הלוגים

כל מה שה-workflow כותב לפלט הסטנדרטי או לשגיאה הסטנדרטית נכנס ללוגים של ה-job. אפשר לקרוא אותם עם job.logs() בזמן שה-job רץ, עוד לפני שיש לו תוצאה:

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

• ה-workflow מעלה שורות לוג חדשות בערך כל 15 שניות, ופעם אחרונה כשה-job מסתיים או מבוטל. מיד אחרי שה-job מתחיל, job.logs() מחזיר No logs yet.

• ה-workflow שומר את 50 המגה-בייט החדשים ביותר של לוגים לכל job, ומוסיף בראש שורה שאומרת שהלוגים נקטעו.

כדי להחליט מה ה-workflow שלכם רושם ללוגים, ראו כתיבת לוגים. כדי להציג רק שורות שמתאימות לביטוי רגולרי, העבירו include ל-job.filtered_logs(). כדי להסתיר שורות, העבירו exclude.

קבלת התוצאה

job.result() מחכה עד שה-job מסתיים, ואז מחזיר את הערך ש-workflow העביר ל-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

כברירת מחדל, job.result() בודק את הסטטוס כל 30 שניות וממתין ללא מגבלת זמן. העבירו maxwait כדי להגביל את מספר הבדיקות. לדוגמה, job.result(maxwait=40) ממתין בערך 20 דקות. כדי לבדוק פעם אחת בלי להמתין, העבירו wait=False. אם job.result() מפסיק להמתין לפני שה-job מסתיים, הוא מחזיר {}.

בדיקה כמה זמן לקח כל שלב

job.events() מציג כל שינוי סטטוס ושלב של job יחד עם הזמן שלו, כך שאפשר לראות כמה זמן ה-job חיכה לחישוב וכמה זמן לקח כל שלב: 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

האירועים משתמשים בשמות הסטטוס השמורים. PENDING הוא INITIALIZING, ו-SUCCEEDED הוא DONE. לשימוש ב-QPU של הרצה, ראו מציאת jobs של QPU בהרצה.

מציאת הסיבה לשגיאה

כש-workflow מעלה חריגה, ה-job מסתיים בסטטוס ERROR. לדוגמה, הריצו את ה-workflow עם זווית שהוא דוחה: 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'

הלוגים מציגים את ה-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") == []

אם ה-workflow מדווח על שגיאות עם send_error(), קראו אותן עם job.events(type="ERROR"), וקראו את ההודעות שלהן עם job.error_message(). עבור job בלי שגיאות מדווחות, job.events(type="ERROR") מחזיר רשימה ריקה, ו-job.error_message() מחזיר {}.

run_workflow_remotely.ipynb

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

ניהול jobs

ביטול job

בטלו job עם job.cancel(). הסטטוס משתנה ל-CANCELING בזמן שה-workflow עוצר את ה-job, ואז ל-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() מחזיר הודעה על ה-job ועל jobs של QPU מקושרים שלו, שגם אותם הוא מנסה לבטל. עבור job רץ, ה-workflow שולח SIGTERM לנקודת הכניסה שלכם. כדי לשמור תוצאות חלקיות, ראו ניקוי כשה-job מבוטל. אם ה-job כבר הסתיים, job.cancel() מחזיר Job already in terminal state. ולא משנה דבר.

מציאת job מ-session חדש

אובייקט Job וה-workflow קיימים רק ב-session הנוכחי של Python. אחרי שמפעילים מחדש kernel של מחברת, העבירו את מזהה ה-job השמור ל-client.job(), או הציגו את ה-jobs האחרונים שלכם עם 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() מחזיר את 10 ה-jobs האחרונים שלכם, החדש ביותר ראשון. סננו אותו עם function, status (מטבלת הסטטוסים) או created_after (חותמת זמן בפורמט ISO 8601), ועברו בין עמודים עם limit ו-offset. כדי להציג את ה-workflows שהעלתם, קראו ל-client.list().

גם הטבלה Workloads ב-IBM Quantum Platform מציגה את ה-jobs שלכם, עם הסטטוס, פרופיל החישוב וזמני ההרצה.

מגבלת זמן

workflow מנוהל מבטל job שרץ יותר מ-24 שעות, וה-job מסתיים בסטטוס CANCELED. בהרצות ארוכות, שמרו נקודות ביקורת באחסון קבצים כדי לא לאבד עבודה.

פתרון בעיות ב-jobs

בעיה	סיבה
run() נכשל עם Code: 429 ו-Active job limit reached.	הגעתם למגבלת ה-jobs הפעילים. המתינו שכמה jobs יסתיימו, או בטלו jobs שאתם לא צריכים יותר.
ה-job נשאר QUEUED, ו-run() הזהיר ש-your account, plan or subscription is blocked.	ה-job לא מתחיל עד שהחסימה על החשבון, התוכנית או המנוי שלכם מוסרת. פנו למנהל המערכת.
job.logs() מחזיר No logs yet.	ה-job בתור או התחיל הרגע. המתינו בערך 15 שניות.
job.result() מחזיר {}	ה-job לא הסתיים, בוטל, או שלא שמר תוצאה. ראו שמירת תוצאות.
job.error_message() מחזיר {}	ה-workflow העלה חריגה בלי לדווח על שגיאה. קראו את ה-traceback עם job.logs().
ה-job במצב CANCELED, אבל לא ביטלתם אותו	ה-job הגיע למגבלת 24 השעות.

השלבים הבאים

המלצות

• כתיבת לוגים וטיפול בשגיאות כדי להגדיר את הלוגים והשגיאות ש-job מדווח עליהם.
• הרצת סריקת פרמטרים כדי להריץ את ה-workflow פעם אחת לכל קבוצת קלטים.