Magpatakbo ng managed workflow

Mga bersyon ng package

Ang code sa pahinang ito ay binuo gamit ang mga sumusunod na requirement. Inirerekomenda naming gamitin ang mga bersyong ito o mas bago.

qiskit-serverless~=0.38.0

Ang isang managed workflow job ay gumagana tulad ng batch job na isinusumite mo sa isang HPC scheduler. Isinusumite ng run() ang job at agad na nagbabalik ng Job object. Gamitin ang Job object para sa mga gawaing sakop ng squeue, output file ng job, at scancel sa isang cluster. Patuloy na tumatakbo ang job kahit isara mo ang notebook mo.

Ang mga halimbawa sa pahinang ito ay gumagamit ng maliit na workflow na nagbabasa ng isang anggulo, theta, at pumapalya kapag wala sa saklaw ang anggulo. I-upload ito ayon sa paliwanag sa Deploy your first managed 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

Magsimula ng job

Kunin ang workflow gamit ang title nito, at tawagin ang run() kasama ang mga argument na binabasa ng 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

Nagbabalik ang run() pagkatapos i-queue ng workflow ang job. I-save ang job ID para mahanap mo ang job mula sa bagong session.

• Para piliin kung anong mga argument ang tinatanggap ng workflow at suriin ang mga ito bago magsimula ang job, tingnan ang Define inputs and outputs.

• Para patakbuhin ang job sa ibang compute size, ipasa ang function_size. Tingnan ang Choose a size for each run.

Subaybayan ang job

Tingnan ang status

Agad pagkatapos mong tawagin ang run(), kadalasang QUEUED ang status: run_workflow_remotely.ipynb

job.status()

'QUEUED'

Status	Kahulugan
QUEUED	Naghihintay ang job ng compute slot.
INITIALIZING	Nagsisimula ang workflow ng environment gamit ang compute profile ng job.
RUNNING	Tumatakbo ang entrypoint ng workflow.
RUNNING: <stage>	Nag-ulat ang entrypoint ng stage gamit ang update_status(), tulad ng RUNNING: WAITING_FOR_QPU.
CANCELING	Kinansela mo ang job, at itinitigil ito ng workflow.
DONE	Matagumpay na nag-exit ang entrypoint.
ERROR	Nag-raise ng error ang entrypoint o nag-exit na may nonzero code. Tingnan ang Find the cause of an error.
CANCELED	Kinansela mo ang job, o umabot ito sa 24-hour limit.

Final na ang DONE, ERROR, at CANCELED, at nagbabalik ang job.in_terminal_state() ng True para sa mga ito. Ang status loop na naghihintay ng ibang value, tulad ng SUCCEEDED, ay hindi kailanman matatapos.

Hintayin ang pagtatapos ng job

Para hintayin hanggang matapos ang job, tawagin ang wait_for_final_state(). Ibinabalik nito ang final status, at kapag verbose=True ay ipi-print nito ang bawat pagbabago ng status. Kung magtatakda ka ng timeout sa segundo, mag-raise ang wait_for_final_state() ng TimeoutError kapag naubos ang oras bago matapos ang job. Patuloy na tumatakbo ang 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'

Sinusuri ng wait_for_final_state() ang status pagkalipas ng 1, 2, 4, 8, 16, at 32 segundo, at pagkatapos ay bawat 60 segundo, kaya nakikita nito ang pagtatapos ng maikling job sa loob ng ilang segundo at kakaunti ang request para sa mahabang job. Para mag-check sa takdang interval, ipasa ang poll_interval sa segundo.

Basahin ang logs

Anumang isulat ng workflow sa standard output o standard error ay napupunta sa logs ng job. Mababasa mo ang mga ito gamit ang job.logs() habang tumatakbo ang job, bago pa ito magkaroon ng resulta:

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

• Nag-a-upload ang workflow ng mga bagong log line mga bawat 15 segundo, at huling beses kapag natapos o nakansela ang job. Agad pagkatapos magsimula ng job, nagbabalik ang job.logs() ng No logs yet.

• Itinatago ng workflow ang pinakabagong 50 MB ng logs para sa bawat job, at nagdaragdag ng linya sa itaas na nagsasabing na-truncate ang logs.

Para magpasya kung ano ang ila-log ng workflow mo, tingnan ang Write logs. Para ipakita lang ang mga linyang tumutugma sa isang regular expression, ipasa ang include sa job.filtered_logs(). Para itago ang mga linya, ipasa ang exclude.

Kunin ang resulta

Naghihintay ang job.result() hanggang matapos ang job, saka ibinabalik ang value na ipinasa ng workflow sa 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

Bilang default, sinusuri ng job.result() ang status bawat 30 segundo at naghihintay nang walang time limit. Ipasa ang maxwait para limitahan ang bilang ng mga check. Halimbawa, naghihintay ang job.result(maxwait=40) ng mga 20 minuto. Para mag-check nang isang beses nang hindi naghihintay, ipasa ang wait=False. Kung huminto sa paghihintay ang job.result() bago matapos ang job, nagbabalik ito ng {}.

Tingnan kung gaano katagal ang bawat stage

Inililista ng job.events() ang bawat pagbabago ng status at stage ng job kasama ang oras nito, para makita mo kung gaano katagal naghintay ang job para sa compute at gaano katagal ang bawat stage: 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

Ginagamit ng mga event ang mga nakaimbak na pangalan ng status. Ang PENDING ay INITIALIZING, at ang SUCCEEDED ay DONE. Para sa QPU usage ng isang run, tingnan ang Find the QPU jobs of a run.

Hanapin ang sanhi ng error

Kapag nag-raise ng exception ang workflow, nagtatapos ang job na may ERROR status. Halimbawa, patakbuhin ang workflow gamit ang anggulong tinatanggihan nito: 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'

Ipinapakita ng logs ang 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") == []

Kung nag-uulat ng mga error ang workflow gamit ang send_error(), basahin ang mga ito gamit ang job.events(type="ERROR"), at basahin ang mga mensahe nila gamit ang job.error_message(). Para sa job na walang naiulat na error, nagbabalik ang job.events(type="ERROR") ng walang laman na listahan, at nagbabalik ang job.error_message() ng {}.

run_workflow_remotely.ipynb

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

Pamahalaan ang mga job

Mag-cancel ng job

Mag-cancel ng job gamit ang job.cancel(). Magiging CANCELING ang status habang itinitigil ng workflow ang job, at pagkatapos ay 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()

Nagbabalik ang job.cancel() ng mensahe tungkol sa job at sa mga naka-link na QPU job nito, na sinusubukan din nitong i-cancel. Para sa tumatakbong job, nagpapadala ang workflow ng SIGTERM sa entrypoint mo. Para mag-save ng partial results, tingnan ang Clean up when a job is canceled. Kung tapos na ang job, nagbabalik ang job.cancel() ng Job already in terminal state. at walang binabago.

Maghanap ng job mula sa bagong session

Ang Job object at ang workflow ay umiiral lang sa kasalukuyan mong Python session. Pagkatapos mong i-restart ang notebook kernel, ipasa ang naka-save na job ID sa client.job(), o ilista ang mga kamakailan mong job gamit ang 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

Ibinabalik ng client.jobs() ang 10 pinakabagong job mo, pinakabago muna. I-filter ito gamit ang function, status (mula sa status table), o created_after (isang ISO 8601 timestamp), at mag-page through gamit ang limit at offset. Para ilista ang mga workflow na na-upload mo, tawagin ang client.list().

Inililista rin ng Workloads table sa IBM Quantum Platform ang mga job mo, kasama ang status, compute profile, at run time ng mga ito.

Time limit

Kinakansela ng managed workflow ang job na tumatakbo nang mahigit 24 oras, at nagtatapos ang job na may CANCELED status. Sa mahahabang run, mag-save ng mga checkpoint sa file storage para hindi mawala ang trabaho mo.

I-troubleshoot ang mga job

Problema	Sanhi
Pumapalya ang run() na may Code: 429 at Active job limit reached.	Naabot mo ang active job limit. Hintayin munang matapos ang ilang job, o i-cancel ang mga job na hindi mo na kailangan.
Nananatiling QUEUED ang job, at nagbabala ang run() na your account, plan or subscription is blocked.	Hindi magsisimula ang job hangga't hindi inaalis ang block sa account, plan, o subscription mo. Makipag-ugnayan sa administrator mo.
Nagbabalik ang job.logs() ng No logs yet.	Naka-queue ang job o kakasimula lang. Maghintay ng mga 15 segundo.
Nagbabalik ang job.result() ng {}	Hindi pa tapos ang job, nakansela ito, o walang na-save na resulta. Tingnan ang Save results.
Nagbabalik ang job.error_message() ng {}	Nag-raise ng exception ang workflow nang hindi nag-uulat ng error. Basahin ang traceback gamit ang job.logs().
CANCELED ang job, pero hindi mo ito kinansela	Umabot ang job sa 24-hour limit.

Mga susunod na hakbang

Mga rekomendasyon

• Write logs and handle errors para i-set up ang logs at mga error na iniuulat ng job.
• Run a parameter sweep para patakbuhin ang workflow nang isang beses para sa bawat set ng input.