اكتب السجلات وتعامل مع الأخطاء

جهّز كود سير العمل ليكتب السجلات، ويبلّغ عن الأخطاء إلى دفتر الملاحظات المحلي، ويحفظ العمل الجزئي عند إلغاء وظيفة. لقراءة هذه المخرجات من دفترك المحلي، راجع تشغيل سير عمل مُدار.

في سير العمل	في دفتر الملاحظات المحلي
اكتب باستخدام logger من وحدة logging، أو استدعِ print().	job.logs() يرجع الأسطر.
ارفع استثناءً، أو اخرج بكود غير صفري.	job.status() يرجع ERROR، وjob.logs() يعرض الـ traceback.
استدعِ send_error().	job.error_message() وjob.events(type="ERROR") يرجعان الكود والرسالة والتفاصيل.
استدعِ save_result() في معالج SIGTERM.	job.result() يرجع النتيجة الجزئية لوظيفة ملغاة.

اكتب السجلات

أي شيء يكتبه سير العمل إلى المخرج القياسي أو خطأ المخرج القياسي يذهب إلى سجلات الوظيفة. اقرأ السجلات باستخدام job.logs() أثناء تشغيل الوظيفة أو بعد انتهائها.

جهّز logger

استخدم وحدة logging من مكتبة Python القياسية. logging.basicConfig() يكتب إلى خطأ المخرج القياسي، الذي يصل إلى السجلات بدون تأخير. الصيغة في المثال التالي تضيف طابعًا زمنيًا لكل سطر:

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() يصل أيضًا إلى السجلات، لكن المخرج القياسي مخزَّن مؤقتًا، فقد لا تظهر الأسطر المطبوعة حتى تنتهي الوظيفة. إذا استخدمت print()، مرّر flush=True.

اضبط مستوى السجل لكل تشغيل

لتصحيح تشغيل واحد بدون تغيير الكود، مرّر مستوى السجل كوسيط:

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"))

إذا كان لسير العمل مخطط وسائط، صرّح بـ log_level فيه، مثلًا كـ {"enum": ["DEBUG", "INFO", "WARNING"]}.

أفضل ممارسة: قرر ما الذي تسجّله

سجّل ما يكفي لإعادة بناء التشغيل من سجلاته:

• في البداية، سجّل ما استلمه التشغيل. سجّل اسم الـ backend وعدد الدارات والإعدادات الأخرى التي تتغير بين التشغيلات.

• سجّل بداية كل مرحلة ونهايتها. استخدم رسائل ثابتة تقدر تبحث عنها.

• سجّل التكرارات على فترات ثابتة. مثلًا، سجّل سطرًا مثل iteration 10/200, energy=-1.1000 كل 10 تكرارات، بدلًا من كل تكرار.

• سجّل الاستثناءات باستخدام logger.exception() داخل كتلة except. السطر يتضمن الـ traceback كاملًا، حتى لو تعافى سير العمل من الخطأ.

• أوقف أشرطة التقدم. السجلات تخزّن كل تحديث لشريط التقدم، مثل شريط من tqdm، كسطر منفصل. مرّر disable=True إلى tqdm، أو سجّل على فترات بدلًا من ذلك.

• أبقِ السجلات صغيرة. لا تسجّل بيانات كبيرة، مثل كل shot. سير العمل المُدار يحتفظ فقط بأحدث 50 ميغابايت من السجلات لكل وظيفة.

• استخدم كلمتي "error" و"exception" فقط في الرسائل عن الإخفاقات. إذا لم تكن وظيفة فاشلة قد أبلغت عن خطئها، فإن الاستثناء الذي يرفعه job.result() يحتوي فقط على أسطر السجل التي تتضمن هاتين الكلمتين.

أفضل ممارسة: أبلغ عن الأخطاء باستخدام send_error()

عندما يرفع سير العمل استثناءً، تنتهي الوظيفة بالحالة ERROR، ويذهب الـ traceback إلى السجلات. سير العمل المُدار لا يخزّن الخطأ، حتى لو كان ServerlessError. job.error_message() يرجع {}، والاستثناء الذي يرفعه job.result() يحتوي فقط على أسطر السجل التي تتضمن error أو exception. فحوصات أنواع الوسائط ونطاقاتها مكانها مخطط الوسائط، الذي يرفض أي تشغيل سيئ قبل بدء الوظيفة.

أبلغ عن كل خطأ يوقف سير العمل باستخدام 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() يرفع QiskitServerlessException بنفس النص.

لتفعيل ذلك، ارفع ServerlessError في المكان الذي يحدث فيه الخطأ، وغلّف كود نقطة الدخول في كتلة try ترسل أي خطأ باستخدام send_error(). سير العمل يضيف هذا المعالج إلى Catalog Functions، لكن ليس إلى سير العمل المُدار:

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 الأخيرة تُبقي الـ traceback في السجلات وتنهي الوظيفة بالحالة ERROR.

• اشتق صنفًا فرعيًا من ServerlessError لتسمية مجموعة أخطاء. المعالج يرسل اسم الصنف، مثل InputError، كحقل exception.

• الاستثناءات الأخرى تصل أيضًا إلى المعالج. تحصل على الكود 1، ويصبح repr() الخاص بها هو الرسالة، مثل ValueError('bad shots').

الإبلاغ عن مشكلة ومواصلة التنفيذ

لا تُوقف send_error() سير العمل ولا تغيّر حالته، لذا يمكنك أيضاً الإبلاغ عن المشكلات التي يتعافى منها سير العمل، مثل كل دارة يتخطاها:

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") كل خطأ أبلغ عنه سير العمل، بما في ذلك أخطاء المهمة التي انتهت بالحالة DONE. راجع العثور على سبب الخطأ.

• مرّر الأرقام والسلاسل النصية في args. تعرض error_message() العنصر الأول فقط من القائمة.

• تحقق من القيمة المُرجعة. إذا تعذّر على سير العمل تخزين الخطأ، مثلاً لأن الكود يعمل خارج مهمة، فإن send_error() تكتب الخطأ في السجلات وتُرجع False.

التنظيف عند إلغاء مهمة

عندما تلغي مهمة، يرسل سير العمل SIGTERM إلى نقطة الدخول الخاصة بك، ويرفع السجلات النهائية، ويُغلق البيئة. وبدون معالج، يتوقف Python فوراً ولا تُنفَّذ كتل finally. كما يحاول سير العمل إلغاء مهام وجلسات QPU التي بدأها سير العمل باستخدام get_runtime_service(). راجع إلغاء مهام QPU المرتبطة.

للاحتفاظ بالعمل الجزئي، سجّل معالجاً لـ SIGTERM. عندها يحصل دفتر الملاحظات المحلي على النتيجة الجزئية من المهمة الملغاة:

run_workflow_remotely.ipynb

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

لتفعيل ذلك، أضف المعالج في بداية نقطة الدخول:

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() في المعالج الاستثناء SystemExit في الكود الذي يعمل، لذا تُنفَّذ كتل finally قبل انتهاء السكربت. وما زالت المهمة تنتهي بالحالة CANCELED.

اجعل المعالج قصيراً، لأن سير العمل يُغلق البيئة بعد SIGTERM بوقت قصير. وفي التشغيلات الطويلة، احفظ أيضاً نقاط تحقق (checkpoints) حتى يتمكن تشغيل لاحق من الاستئناف بعد فشل المهمة أو بلوغها حد 24 ساعة. راجع أفضل الممارسات: نقاط التحقق للمهام الطويلة.

الخطوات التالية

التوصيات

• شغّل سير عمل مُدار لبدء مهمة وقراءة حالتها وسجلاتها ونتيجتها.
• أدِر الملفات عبر التشغيلات لحفظ نقاط التحقق التي يمكن لتشغيل لاحق الاستئناف منها.
• اختر ملف حوسبة لتشغيل سير عملك بمزيد من موارد CPU أو الذاكرة أو GPU.