Tulis log dan tangani error

Atur kode workflow-mu untuk menulis log, melaporkan error ke notebook lokalmu, dan menyimpan hasil parsial saat job dibatalkan. Untuk membaca output ini dari notebook lokalmu, lihat Jalankan managed workflow.

Di workflow	Di notebook lokalmu
Tulis dengan logger dari modul `logging`, atau panggil `print()`.	`job.logs()` mengembalikan baris-barisnya.
Munculkan exception, atau keluar dengan kode bukan nol.	`job.status()` mengembalikan `ERROR`, dan `job.logs()` menampilkan traceback.
Panggil `send_error()`.	`job.error_message()` dan `job.events(type="ERROR")` mengembalikan kode, pesan, dan detail.
Panggil `save_result()` di handler `SIGTERM`.	`job.result()` mengembalikan hasil parsial dari job yang dibatalkan.

Tulis log

Apa pun yang ditulis workflow ke standard output atau standard error masuk ke log job. Baca log dengan job.logs() saat job berjalan atau setelah selesai.

Siapkan logger

Gunakan modul logging dari pustaka standar Python. logging.basicConfig() menulis ke standard error, yang sampai ke log tanpa tertunda. Format pada contoh berikut menambahkan timestamp ke setiap baris:

source_files/my_workflow.py

import logging

logging.basicConfig(
    level=logging.WARNING,
    format="%(asctime)s %(levelname)s %(name)s: %(message)s",
)
logger = logging.getLogger("my_workflow")
logger.setLevel(logging.INFO)

logger.info("Transpiled %d circuits: depth %d -> %d", len(circuits), 1250, 730)

2026-11-01 21:57:35,195 INFO my_workflow: Transpiled 12 circuits: depth 1250 -> 730

level=logging.WARNING menyembunyikan pesan INFO dari paket lain, dan logger.setLevel(logging.INFO) mempertahankan pesan INFO untuk logger milikmu sendiri. qiskit_ibm_runtime mengatur levelnya sendiri dengan environment variable QISKIT_IBM_RUNTIME_LOG_LEVEL.

print() juga sampai ke log, tetapi standard output di-buffer, jadi baris yang dicetak mungkin baru muncul setelah job selesai. Kalau kamu memakai print(), berikan flush=True.

Atur level log untuk setiap run

Untuk men-debug satu run tanpa mengubah kode, berikan level log sebagai argumen:

run_workflow_remotely.ipynb

job = my_workflow.run(circuits=circuits, log_level="DEBUG")

Agar ini berfungsi, atur level logger-mu dari argumen:

source_files/my_workflow.py

from qiskit_serverless import get_arguments

arguments = get_arguments()
logger.setLevel(arguments.get("log_level", "INFO"))

Kalau workflow punya schema argumen, deklarasikan log_level di dalamnya, misalnya sebagai {"enum": ["DEBUG", "INFO", "WARNING"]}.

Praktik terbaik: tentukan apa yang dicatat

Catat cukup banyak agar sebuah run bisa direkonstruksi dari lognya:

Di awal, catat apa yang diterima run. Catat nama backend, jumlah circuit, dan pengaturan lain yang berubah antar run.
Catat awal dan akhir setiap tahap. Pakai pesan tetap yang bisa kamu cari.
Catat iterasi pada interval tetap. Misalnya, catat baris seperti iteration 10/200, energy=-1.1000 setiap 10 iterasi, bukan setiap iterasi.
Catat exception dengan logger.exception() di dalam blok except. Barisnya menyertakan traceback lengkap, bahkan kalau workflow pulih dari error tersebut.
Matikan progress bar. Log menyimpan setiap refresh progress bar, seperti dari tqdm, sebagai baris terpisah. Berikan disable=True ke tqdm, atau catat pada interval tertentu saja.
Jaga log tetap kecil. Jangan catat data besar, seperti setiap shot. Managed workflow hanya menyimpan 50 MB log terbaru untuk setiap job.
Gunakan kata "error" dan "exception" hanya dalam pesan tentang kegagalan. Kalau job yang gagal tidak melaporkan errornya, exception yang dimunculkan job.result() hanya berisi baris log yang memuat kata-kata ini.

Praktik terbaik: laporkan error dengan send_error()

Saat workflow memunculkan exception, job berakhir dengan status ERROR, dan traceback masuk ke log. Managed workflow tidak menyimpan error tersebut, bahkan untuk ServerlessError. job.error_message() mengembalikan {}, dan exception yang dimunculkan job.result() hanya berisi baris log yang memuat error atau exception. Pemeriksaan tipe dan rentang argumen sebaiknya ada di schema argumen, yang menolak run yang salah sebelum job dimulai.

Laporkan setiap error yang menghentikan workflow dengan send_error(), supaya notebook lokalmu mendapat error singkat dengan kode dan detail:

run_workflow_remotely.ipynb

print(job.error_message())

| Message: The circuit has more qubits than the backend.
| Code: 4500
| Exception: InputError
| Details:
|   - circuit_qubits: 10
|   - backend_qubits: 5

job.result() memunculkan QiskitServerlessException dengan teks yang sama.

Agar ini berfungsi, munculkan ServerlessError di tempat error terjadi, dan bungkus kode entrypoint-mu dalam blok try yang mengirim error apa pun dengan send_error(). Workflow menambahkan handler ini ke Catalog Functions, tetapi tidak ke managed workflow:

source_files/my_workflow.py

from qiskit_serverless import ServerlessError, send_error

class InputError(ServerlessError):
    """The arguments are valid, but the workflow cannot use them."""

def main():
    ...
    if num_qubits > backend.num_qubits:
        raise InputError(
            code="4500",
            message="The circuit has more qubits than the backend.",
            details={"circuit_qubits": num_qubits, "backend_qubits": backend.num_qubits},
        )
    ...

try:
    main()
except Exception as error:
    send_error(
        code=getattr(error, "code", "1"),
        message=getattr(error, "message", repr(error)),
        exception=type(error).__name__,
        args=getattr(error, "details", None),
    )
    raise

Munculkan error itu lagi. raise terakhir menjaga traceback di log dan mengakhiri job dengan status ERROR.
Buat subclass ServerlessError untuk menamai sekelompok error. Handler mengirim nama class, seperti InputError, sebagai field exception.
Exception lain juga sampai ke handler. Exception itu mendapat kode 1, dan repr()-nya menjadi pesan, seperti ValueError('bad shots').

Laporkan masalah dan lanjutkan

send_error() tidak menghentikan workflow atau mengubah statusnya, jadi kamu juga bisa melaporkan masalah yang berhasil dipulihkan workflow, seperti setiap circuit yang dilewati:

source_files/my_workflow.py

valid_circuits = []
for index, circuit in enumerate(circuits):
    if circuit.num_qubits > backend.num_qubits:
        send_error(
            code="4501",
            message="Skipped a circuit that is wider than the backend.",
            exception="InputError",
            args={"circuit_index": index},
        )
        continue
    valid_circuits.append(circuit)

job.events(type="ERROR") mengembalikan setiap error yang dilaporkan workflow, termasuk error dari job yang berakhir dengan status DONE. Lihat Cari penyebab error.

Berikan angka dan string di args. error_message() hanya menampilkan item pertama dari sebuah list.
Periksa nilai kembaliannya. Kalau workflow tidak bisa menyimpan error, misalnya karena kode berjalan di luar job, send_error() menulis error ke log dan mengembalikan False.

Bersihkan saat job dibatalkan

Saat kamu membatalkan job, workflow mengirim SIGTERM ke entrypoint-mu, mengunggah log terakhir, dan mematikan environment. Tanpa handler, Python langsung berhenti, dan blok finally tidak dijalankan. Workflow juga mencoba membatalkan job QPU dan session yang dimulai workflow dengan get_runtime_service(). Lihat Batalkan job QPU yang terhubung.

Untuk menyimpan hasil parsial, daftarkan handler untuk SIGTERM. Notebook lokalmu kemudian mendapat hasil parsial dari job yang dibatalkan:

run_workflow_remotely.ipynb

job.cancel()
print(job.result())  # {'status': 'canceled', 'partial_results': [...]}

Agar ini berfungsi, tambahkan handler di awal entrypoint-mu:

source_files/my_workflow.py

import signal
import sys

from qiskit_serverless import save_result

partial_results = []

def handle_sigterm(signum, frame):
    save_result({"status": "canceled", "partial_results": partial_results})
    sys.exit(0)

signal.signal(signal.SIGTERM, handle_sigterm)

sys.exit() di handler memunculkan SystemExit di kode yang sedang berjalan, jadi blok finally dijalankan sebelum skrip berakhir. Job tetap berakhir dengan status CANCELED.

Buat handler tetap singkat, karena workflow mematikan environment tak lama setelah SIGTERM. Untuk run yang panjang, simpan juga checkpoint, supaya run berikutnya bisa dilanjutkan setelah job gagal atau mencapai batas 24 jam. Lihat Praktik terbaik: beri checkpoint pada job panjang.

Langkah selanjutnya

Rekomendasi

Jalankan managed workflow untuk memulai job dan membaca status, log, serta hasilnya.
Kelola file antar-run untuk menyimpan checkpoint yang bisa dilanjutkan oleh run berikutnya.
Pilih compute profile untuk menjalankan workflow-mu dengan lebih banyak CPU, memori, atau sumber daya GPU.

Tulis log​

Siapkan logger​

Atur level log untuk setiap run​

Praktik terbaik: tentukan apa yang dicatat​

Praktik terbaik: laporkan error dengan send_error()​

Laporkan masalah dan lanjutkan​

Bersihkan saat job dibatalkan​

Langkah selanjutnya​

Tulis log

Siapkan logger

Atur level log untuk setiap run

Praktik terbaik: tentukan apa yang dicatat

Praktik terbaik: laporkan error dengan send_error()

Laporkan masalah dan lanjutkan

Bersihkan saat job dibatalkan

Langkah selanjutnya