Lewati ke konten utama
Proyek ini masih dalam tahap beta. Menemukan bug? Buat isu GitHub. Punya pertanyaan atau ide? Tanya di Discussions — kami sangat menghargai masukan kamu!
Bagaimana kualitas terjemahan di halaman ini?

Jalankan managed workflow

Versi paket

Kode di halaman ini dikembangkan dengan persyaratan berikut. Kami menyarankan memakai versi ini atau yang lebih baru.

qiskit-serverless~=0.38.0

Job managed workflow bekerja seperti batch job yang kamu kirim ke scheduler HPC. run() mengirim job dan langsung mengembalikan objek Job. Pakai objek Job untuk tugas-tugas yang di cluster ditangani oleh squeue, file output job, dan scancel. Job tetap berjalan setelah kamu menutup notebook.

Contoh di halaman ini memakai workflow kecil yang membaca satu sudut, theta, dan gagal kalau sudutnya di luar rentang. Unggah workflow itu seperti dijelaskan di Deploy managed workflow pertamamu.

# 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

Mulai job​

Ambil workflow berdasarkan judulnya, lalu panggil run() dengan argumen yang dibaca 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() selesai setelah workflow memasukkan job ke antrean. Simpan ID job supaya kamu bisa menemukan job dari sesi baru.

Pantau job​

Periksa status​

Tepat setelah kamu memanggil run(), statusnya biasanya QUEUED: run_workflow_remotely.ipynb

job.status()
'QUEUED'
StatusArti
QUEUEDJob menunggu slot compute.
INITIALIZINGWorkflow sedang memulai lingkungan dengan compute profile milik job.
RUNNINGEntrypoint workflow sedang berjalan.
RUNNING: <stage>Entrypoint melaporkan sebuah stage dengan update_status(), misalnya RUNNING: WAITING_FOR_QPU.
CANCELINGKamu membatalkan job, dan workflow sedang menghentikannya.
DONEEntrypoint selesai dengan sukses.
ERROREntrypoint memunculkan error atau keluar dengan kode bukan nol. Lihat Cari penyebab error.
CANCELEDKamu membatalkan job, atau job mencapai batas 24 jam.

DONE, ERROR, dan CANCELED adalah status akhir, dan job.in_terminal_state() mengembalikan True untuk ketiganya. Loop status yang menunggu nilai lain, seperti SUCCEEDED, tidak akan pernah berhenti.

Tunggu sampai job selesai​

Untuk menunggu sampai job selesai, panggil wait_for_final_state(). Fungsi ini mengembalikan status akhir, dan dengan verbose=True ia mencetak setiap perubahan status. Kalau kamu mengatur timeout dalam detik, wait_for_final_state() memunculkan TimeoutError saat waktu habis sebelum job selesai. Job tetap berjalan:

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() memeriksa status setelah 1, 2, 4, 8, 16, dan 32 detik, lalu setiap 60 detik, jadi ia mengetahui selesainya job singkat dalam hitungan detik dan hanya membuat sedikit request untuk job yang lama. Kalau mau memeriksa dengan interval tetap, berikan poll_interval dalam detik.

Baca log​

Apa pun yang ditulis workflow ke standard output atau standard error masuk ke log job. Kamu bisa membacanya dengan job.logs() saat job berjalan, sebelum ada hasil:

# 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 mengunggah baris log baru kira-kira setiap 15 detik, dan sekali lagi saat job selesai atau dibatalkan. Tepat setelah job dimulai, job.logs() mengembalikan No logs yet.

  • Workflow menyimpan 50 MB log terbaru untuk setiap job, dan menambahkan satu baris di bagian atas yang menyatakan bahwa log sudah dipotong.

Untuk menentukan apa yang dicatat workflow-mu, lihat Tulis log. Untuk hanya menampilkan baris yang cocok dengan regular expression, berikan include ke job.filtered_logs(). Untuk menyembunyikan baris, berikan exclude.

Ambil hasil​

job.result() menunggu sampai job selesai, lalu mengembalikan nilai yang diberikan workflow ke 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

Secara default, job.result() memeriksa status setiap 30 detik dan menunggu tanpa batas waktu. Berikan maxwait untuk membatasi jumlah pemeriksaan. Misalnya, job.result(maxwait=40) menunggu sekitar 20 menit. Untuk memeriksa sekali saja tanpa menunggu, berikan wait=False. Kalau job.result() berhenti menunggu sebelum job selesai, ia mengembalikan {}.

Periksa berapa lama setiap stage berjalan​

job.events() mencantumkan setiap perubahan status dan stage dari sebuah job beserta waktunya, jadi kamu bisa melihat berapa lama job menunggu compute dan berapa lama setiap stage berjalan: 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

Event memakai nama status yang tersimpan. PENDING adalah INITIALIZING, dan SUCCEEDED adalah DONE. Untuk penggunaan QPU dari sebuah run, lihat Temukan job QPU dari sebuah run.

Cari penyebab error​

Saat workflow memunculkan exception, job berakhir dengan status ERROR. Misalnya, jalankan workflow dengan sudut yang ditolaknya: 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'

Log menunjukkan traceback-nya: 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") == []

Kalau workflow melaporkan error dengan send_error(), baca dengan job.events(type="ERROR"), dan baca pesannya dengan job.error_message(). Untuk job tanpa error yang dilaporkan, job.events(type="ERROR") mengembalikan list kosong, dan job.error_message() mengembalikan {}.

run_workflow_remotely.ipynb

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

Kelola job​

Batalkan job​

Batalkan job dengan job.cancel(). Status berubah menjadi CANCELING saat workflow menghentikan job, lalu menjadi 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() mengembalikan pesan tentang job tersebut dan tentang job QPU yang terhubung, yang juga coba ia batalkan. Untuk job yang sedang berjalan, workflow mengirim SIGTERM ke entrypoint-mu. Untuk menyimpan hasil parsial, lihat Bersihkan saat job dibatalkan. Kalau job sudah selesai, job.cancel() mengembalikan Job already in terminal state. dan tidak mengubah apa pun.

Temukan job dari sesi baru​

Objek Job dan workflow hanya ada di sesi Python-mu saat ini. Setelah kamu me-restart kernel notebook, berikan ID job yang tersimpan ke client.job(), atau daftar job terbaru kamu dengan 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() mengembalikan 10 job terbaru kamu, dari yang paling baru. Saring dengan function, status (dari tabel status), atau created_after (timestamp ISO 8601), dan telusuri halamannya dengan limit dan offset. Untuk mendaftar workflow yang sudah kamu unggah, panggil client.list().

Tabel Workloads di IBM Quantum Platform juga mencantumkan job kamu, lengkap dengan status, compute profile, dan waktu run-nya.

Batas waktu​

Managed workflow membatalkan job yang berjalan lebih dari 24 jam, dan job berakhir dengan status CANCELED. Selama run yang panjang, simpan checkpoint ke file storage supaya kamu tidak kehilangan pekerjaan.

Pemecahan masalah job​

MasalahPenyebab
run() gagal dengan Code: 429 dan Active job limit reached.Kamu mencapai batas job aktif. Tunggu beberapa job selesai, atau batalkan job yang tidak lagi kamu butuhkan.
Job tetap QUEUED, dan run() memberi peringatan your account, plan or subscription is blocked.Job tidak akan dimulai sampai blokir pada akun, paket, atau langganan kamu dicabut. Hubungi administrator kamu.
job.logs() mengembalikan No logs yet.Job sedang antre atau baru dimulai. Tunggu sekitar 15 detik.
job.result() mengembalikan {}Job belum selesai, dibatalkan, atau tidak menyimpan hasil. Lihat Simpan hasil.
job.error_message() mengembalikan {}Workflow memunculkan exception tanpa melaporkan error. Baca traceback dengan job.logs().
Job berstatus CANCELED, padahal kamu tidak membatalkannyaJob mencapai batas 24 jam.

Langkah berikutnya​

Rekomendasi
Sumber: Dokumentasi IBM Quantum — diperbarui 6 Nov 2026
Versi bahasa Inggris di doQumentation — diperbarui 6 Nov 2026
Masalah situs atau terjemahan?Masalah konten? Edit di IBM Quantum docs