เขียน logs และจัดการข้อผิดพลาด

ตั้งค่าโค้ดเวิร์กโฟลว์ของคุณให้เขียน logs รายงานข้อผิดพลาดไปยังโน้ตบุ๊กในเครื่อง และบันทึกงานบางส่วนเมื่องานถูกยกเลิก หากต้องการอ่านเอาต์พุตเหล่านี้จากโน้ตบุ๊กในเครื่อง ดู รันเวิร์กโฟลว์แบบ managed

ในเวิร์กโฟลว์	ในโน้ตบุ๊กในเครื่องของคุณ
เขียนด้วย logger จากโมดูล logging หรือเรียก print()	job.logs() คืนบรรทัดเหล่านั้น
raise exception หรือออกด้วยรหัสที่ไม่ใช่ศูนย์	job.status() คืน ERROR และ job.logs() แสดง traceback
เรียก send_error()	job.error_message() และ job.events(type="ERROR") คืนรหัส ข้อความ และรายละเอียด
เรียก save_result() ใน handler ของ SIGTERM	job.result() คืนผลลัพธ์บางส่วนของงานที่ถูกยกเลิก

เขียน logs

ทุกอย่างที่เวิร์กโฟลว์เขียนลง standard output หรือ standard error จะไปที่ logs ของงาน อ่าน logs ด้วย job.logs() ระหว่างที่งานรันหรือหลังงานจบ

ตั้งค่า logger

ใช้โมดูล logging จากไลบรารีมาตรฐานของ Python logging.basicConfig() เขียนลง standard error ซึ่งไปถึง logs โดยไม่ล่าช้า รูปแบบในตัวอย่างต่อไปนี้เพิ่ม timestamp ให้แต่ละบรรทัด:

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 ซ่อนข้อความ INFO จากแพ็กเกจอื่น และ logger.setLevel(logging.INFO) คงข้อความ INFO ไว้สำหรับ logger ของคุณเอง qiskit_ibm_runtime ตั้งระดับของตัวเองด้วยตัวแปรสภาพแวดล้อม QISKIT_IBM_RUNTIME_LOG_LEVEL

print() ก็ไปถึง logs เช่นกัน แต่ standard output มีการบัฟเฟอร์ ดังนั้นบรรทัดที่พิมพ์อาจไม่ปรากฏจนกว่างานจะจบ ถ้าใช้ print() ให้ส่ง flush=True

ตั้งระดับ log สำหรับแต่ละการรัน

หากต้องการดีบักการรันหนึ่งครั้งโดยไม่เปลี่ยนโค้ด ให้ส่งระดับ log เป็นอาร์กิวเมนต์:

run_workflow_remotely.ipynb

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

เพื่อให้ใช้งานได้ ให้ตั้งระดับของ logger จากอาร์กิวเมนต์:

source_files/my_workflow.py

from qiskit_serverless import get_arguments

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

ถ้าเวิร์กโฟลว์มีschema ของอาร์กิวเมนต์ ให้ประกาศ log_level ไว้ใน schema เช่น {"enum": ["DEBUG", "INFO", "WARNING"]}

แนวปฏิบัติที่ดี: ตัดสินใจว่าจะ log อะไร

log ให้พอที่จะสร้างการรันขึ้นมาใหม่จาก logs ได้:

• ตอนเริ่มต้น ให้ log สิ่งที่การรันได้รับ log ชื่อ backend จำนวน circuit และการตั้งค่าอื่นที่เปลี่ยนไปในแต่ละการรัน

• log จุดเริ่มต้นและจุดสิ้นสุดของแต่ละขั้นตอน ใช้ข้อความคงที่ที่คุณค้นหาได้

• log การวนซ้ำเป็นช่วงคงที่ ตัวอย่างเช่น log บรรทัดอย่าง iteration 10/200, energy=-1.1000 ทุก 10 รอบ แทนที่จะ log ทุกรอบ

• log exception ด้วย logger.exception() ภายในบล็อก except บรรทัดนั้นมี traceback ทั้งหมด แม้เวิร์กโฟลว์จะกู้คืนจากข้อผิดพลาดได้

• ปิด progress bar logs เก็บการรีเฟรชแต่ละครั้งของ progress bar เช่นของ tqdm เป็นบรรทัดแยกกัน ให้ส่ง disable=True ให้ tqdm หรือ log เป็นช่วงแทน

• ทำให้ logs มีขนาดเล็ก อย่า log ข้อมูลขนาดใหญ่ เช่นทุก shot เวิร์กโฟลว์แบบ managed เก็บ logs ล่าสุดเพียง 50 MB ต่อหนึ่งงาน

• ใช้คำว่า "error" และ "exception" เฉพาะในข้อความเกี่ยวกับความล้มเหลว ถ้างานที่ล้มเหลวไม่ได้รายงานข้อผิดพลาด exception ที่ job.result() raise จะมีเฉพาะบรรทัด log ที่มีคำเหล่านี้

แนวปฏิบัติที่ดี: รายงานข้อผิดพลาดด้วย send_error()

เมื่อเวิร์กโฟลว์ raise exception งานจะจบด้วยสถานะ ERROR และ traceback จะไปที่ logs เวิร์กโฟลว์แบบ managed ไม่เก็บข้อผิดพลาด แม้เป็น ServerlessError ก็ตาม job.error_message() คืน {} และ exception ที่ job.result() raise มีเฉพาะบรรทัด log ที่มี error หรือ exception การตรวจสอบชนิดและช่วงของอาร์กิวเมนต์ควรอยู่ในschema ของอาร์กิวเมนต์ ซึ่งปฏิเสธการรันที่ไม่ถูกต้องก่อนที่งานจะเริ่ม

รายงานทุกข้อผิดพลาดที่ทำให้เวิร์กโฟลว์หยุดด้วย send_error() เพื่อให้โน้ตบุ๊กในเครื่องของคุณได้รับข้อผิดพลาดสั้น ๆ พร้อมรหัสและรายละเอียด:

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() raise QiskitServerlessException พร้อมข้อความเดียวกัน

เพื่อให้ใช้งานได้ ให้ raise ServerlessError ตรงจุดที่เกิดข้อผิดพลาด และครอบโค้ดของ entrypoint ไว้ในบล็อก try ที่ส่งทุกข้อผิดพลาดด้วย send_error() เวิร์กโฟลว์เพิ่ม handler นี้ให้ Catalog Functions แต่ไม่เพิ่มให้เวิร์กโฟลว์แบบ managed:

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

• raise ข้อผิดพลาดอีกครั้ง raise ตัวสุดท้ายเก็บ traceback ไว้ใน logs และทำให้งานจบด้วยสถานะ ERROR

• สร้าง subclass ของ ServerlessError เพื่อตั้งชื่อกลุ่มข้อผิดพลาด handler ส่งชื่อของคลาส เช่น InputError เป็นฟิลด์ exception

• exception อื่นก็ไปถึง handler เช่นกัน exception เหล่านั้นได้รหัส 1 และ repr() ของมันกลายเป็นข้อความ เช่น ValueError('bad shots')

รายงานปัญหาแล้วทำงานต่อ

send_error() ไม่หยุด workflow และไม่เปลี่ยนสถานะของ workflow ดังนั้นคุณจึงรายงานปัญหาที่ workflow กู้คืนได้ด้วย เช่น แต่ละ circuit ที่ workflow ข้ามไป:

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") ส่งคืนทุก error ที่ workflow รายงาน รวมถึง error ของ job ที่จบด้วยสถานะ DONE ดู Find the cause of an error

• ส่งตัวเลขและ string ใน args error_message() จะแสดงเฉพาะรายการแรกของ list เท่านั้น

• ตรวจสอบค่าที่ส่งคืน หาก workflow เก็บ error ไม่ได้ เช่น เพราะโค้ดรันอยู่นอก job send_error() จะเขียน error ลงใน logs และส่งคืน False

ล้างข้อมูลเมื่อ job ถูกยกเลิก

เมื่อคุณยกเลิก job workflow จะส่ง SIGTERM ไปยัง entrypoint ของคุณ อัปโหลด logs สุดท้าย และปิด environment หากไม่มี handler Python จะหยุดทันที และบล็อก finally จะไม่ทำงาน workflow ยังพยายามยกเลิก QPU job และ session ที่ workflow เริ่มด้วย get_runtime_service() ด้วย ดู Cancel linked QPU jobs

หากต้องการเก็บงานที่ทำไปบางส่วน ให้ลงทะเบียน handler สำหรับ SIGTERM จากนั้น notebook บนเครื่องของคุณจะได้ผลลัพธ์บางส่วนจาก job ที่ถูกยกเลิก:

run_workflow_remotely.ipynb

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

หากต้องการเปิดใช้งานสิ่งนี้ ให้เพิ่ม handler ที่จุดเริ่มต้นของ entrypoint:

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() ใน handler จะ raise SystemExit ในโค้ดที่กำลังรันอยู่ ดังนั้นบล็อก finally จะทำงานก่อนที่สคริปต์จะจบ job ยังคงจบด้วยสถานะ CANCELED

ให้ handler สั้นเข้าไว้ เพราะ workflow จะปิด environment ไม่นานหลังจาก SIGTERM สำหรับการรันที่ยาวนาน ควรบันทึก checkpoint ด้วย เพื่อให้การรันครั้งถัดไปทำต่อได้หลังจาก job ล้มเหลวหรือถึงขีดจำกัด 24 ชั่วโมง ดู Best practice: checkpoint long jobs

ขั้นตอนถัดไป

คำแนะนำ

• Run a managed workflow เพื่อเริ่ม job และอ่านสถานะ logs และผลลัพธ์
• Manage files across runs เพื่อบันทึก checkpoint ที่การรันครั้งถัดไปทำต่อได้
• Choose a compute profile เพื่อรัน workflow ของคุณด้วย CPU หน่วยความจำ หรือทรัพยากร GPU ที่มากขึ้น