شغّل سير عمل مُدار

إصدارات الحزم

تم تطوير الكود في هذه الصفحة باستخدام المتطلبات التالية. نوصي باستخدام هذه الإصدارات أو الأحدث.

qiskit-serverless~=0.38.0

تعمل مهمة سير العمل المُدار مثل مهمة دفعية ترسلها إلى مجدول HPC. تُرسل run() المهمة وتعيد كائن Job فورًا. استخدم كائن Job للمهام التي تغطيها squeue وملف مخرجات المهمة وscancel على العنقود. تستمر المهمة في العمل بعد إغلاق دفتر الملاحظات.

تستخدم الأمثلة في هذه الصفحة سير عمل صغيرًا يقرأ زاوية واحدة، theta، ويفشل إذا كانت الزاوية خارج النطاق. ارفعه كما هو موضح في انشر أول سير عمل مُدار لك.

# 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

ابدأ مهمة

احصل على سير العمل بعنوانه، واستدعِ run() بالوسائط التي يقرؤها سير العمل:

# 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() النتيجة بعد أن يضع سير العمل المهمة في الطابور. احفظ معرّف المهمة حتى تتمكن من العثور على المهمة من جلسة جديدة.

• لاختيار الوسائط التي يقبلها سير العمل والتحقق منها قبل بدء المهمة، راجع عرّف المدخلات والمخرجات.

• لتشغيل المهمة بحجم حوسبة مختلف، مرّر function_size. راجع اختر حجمًا لكل تشغيل.

تابع مهمة

تحقق من الحالة

مباشرة بعد استدعاء run()، تكون الحالة عادةً QUEUED: run_workflow_remotely.ipynb

job.status()

'QUEUED'

الحالة	المعنى
QUEUED	تنتظر المهمة خانة حوسبة.
INITIALIZING	يبدأ سير العمل بيئة بملف الحوسبة الخاص بالمهمة.
RUNNING	نقطة دخول سير العمل قيد التشغيل.
RUNNING: <stage>	أبلغت نقطة الدخول عن مرحلة باستخدام update_status()، مثل RUNNING: WAITING_FOR_QPU.
CANCELING	ألغيت المهمة، ويقوم سير العمل بإيقافها.
DONE	انتهت نقطة الدخول بنجاح.
ERROR	أطلقت نقطة الدخول خطأ أو انتهت برمز غير صفري. راجع اعثر على سبب الخطأ.
CANCELED	ألغيت المهمة، أو بلغت حد 24 ساعة.

الحالات DONE وERROR وCANCELED نهائية، وتعيد job.in_terminal_state() القيمة True لها. حلقة الحالة التي تنتظر قيمة أخرى، مثل SUCCEEDED، لا تنتهي أبدًا.

انتظر انتهاء المهمة

للانتظار حتى تنتهي المهمة، استدعِ wait_for_final_state(). تعيد الحالة النهائية، ومع verbose=True تطبع كل تغيّر في الحالة. إذا حددت timeout بالثواني، فإن wait_for_final_state() تطلق TimeoutError عندما ينتهي الوقت قبل انتهاء المهمة. تستمر المهمة في العمل:

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() من الحالة بعد 1 و2 و4 و8 و16 و32 ثانية، ثم كل 60 ثانية، لذلك ترصد نهاية المهمة القصيرة خلال ثوانٍ وتُجري طلبات قليلة للمهمة الطويلة. وللتحقق بفاصل زمني ثابت بدلًا من ذلك، مرّر poll_interval بالثواني.

اقرأ السجلات

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

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

• يرفع سير العمل أسطر السجل الجديدة كل 15 ثانية تقريبًا، ومرة أخيرة عند انتهاء المهمة أو إلغائها. مباشرة بعد بدء المهمة، تعيد job.logs() الرسالة No logs yet.

• يحتفظ سير العمل بأحدث 50 ميغابايت من السجلات لكل مهمة، ويضيف في الأعلى سطرًا يذكر أن السجلات قد اقتُطعت.

لتحديد ما يسجله سير العمل، راجع اكتب السجلات. ولإظهار الأسطر المطابقة لتعبير نمطي فقط، مرّر include إلى job.filtered_logs(). ولإخفاء أسطر، مرّر exclude.

احصل على النتيجة

تنتظر job.result() حتى تنتهي المهمة، ثم تعيد القيمة التي مرّرها سير العمل إلى 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

افتراضيًا، تتحقق job.result() من الحالة كل 30 ثانية وتنتظر بلا حد زمني. مرّر maxwait لتحديد عدد مرات التحقق. على سبيل المثال، job.result(maxwait=40) تنتظر نحو 20 دقيقة. وللتحقق مرة واحدة دون انتظار، مرّر wait=False. إذا توقفت job.result() عن الانتظار قبل انتهاء المهمة، فإنها تعيد {}.

تحقق من المدة التي استغرقتها كل مرحلة

تسرد job.events() كل تغيّر في حالة المهمة ومراحلها مع وقته، لتتمكن من معرفة المدة التي انتظرتها المهمة للحصول على الحوسبة والمدة التي استغرقتها كل مرحلة: 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

تستخدم الأحداث أسماء الحالات المخزّنة. PENDING هي INITIALIZING، وSUCCEEDED هي DONE. لمعرفة استخدام QPU في تشغيل ما، راجع اعثر على مهام QPU الخاصة بتشغيل.

اعثر على سبب الخطأ

عندما يطلق سير العمل استثناءً، تنتهي المهمة بالحالة ERROR. على سبيل المثال، شغّل سير العمل بزاوية يرفضها: 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'

تعرض السجلات تتبّع الاستدعاءات: 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") == []

إذا كان سير العمل يبلّغ عن الأخطاء باستخدام send_error()، فاقرأها باستخدام job.events(type="ERROR")، واقرأ رسائلها باستخدام job.error_message(). وبالنسبة لمهمة بلا أخطاء مُبلَّغ عنها، تعيد job.events(type="ERROR") قائمة فارغة، وتعيد job.error_message() القيمة {}.

run_workflow_remotely.ipynb

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

أدِر المهام

ألغِ مهمة

ألغِ مهمة باستخدام job.cancel(). تتغير الحالة إلى CANCELING بينما يوقف سير العمل المهمة، ثم إلى 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() رسالة عن المهمة وعن مهام QPU المرتبطة بها، والتي تحاول إلغاءها أيضًا. وبالنسبة لمهمة قيد التشغيل، يرسل سير العمل SIGTERM إلى نقطة الدخول الخاصة بك. لحفظ النتائج الجزئية، راجع نظّف عند إلغاء مهمة. وإذا كانت المهمة قد انتهت بالفعل، تعيد job.cancel() الرسالة Job already in terminal state. ولا تغيّر شيئًا.

اعثر على مهمة من جلسة جديدة

كائن Job وسير العمل موجودان فقط في جلسة Python الحالية. بعد إعادة تشغيل نواة دفتر الملاحظات، مرّر معرّف المهمة المحفوظ إلى client.job()، أو اسرد مهامك الأخيرة باستخدام 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() أحدث 10 مهام لديك، الأحدث أولًا. رشّحها باستخدام function أو status (من جدول الحالات) أو created_after (طابع زمني بصيغة ISO 8601)، وتنقّل بين صفحاتها باستخدام limit وoffset. ولسرد سير العمل التي رفعتها، استدعِ client.list().

يسرد جدول Workloads في IBM Quantum Platform مهامك أيضًا، مع حالتها وملف الحوسبة وأوقات التشغيل.

الحد الزمني

يلغي سير العمل المُدار أي مهمة تعمل أكثر من 24 ساعة، وتنتهي المهمة بالحالة CANCELED. أثناء التشغيلات الطويلة، احفظ نقاط التحقق في تخزين الملفات حتى لا تفقد عملك.

استكشاف أخطاء المهام وإصلاحها

المشكلة	السبب
تفشل run() مع Code: 429 وActive job limit reached.	وصلت إلى حد المهام النشطة. انتظر حتى تنتهي بعض المهام، أو ألغِ المهام التي لم تعد بحاجة إليها.
تبقى المهمة QUEUED، وحذّرت run() من your account, plan or subscription is blocked.	لن تبدأ المهمة حتى يُرفع الحظر عن حسابك أو خطتك أو اشتراكك. تواصل مع المسؤول لديك.
تعيد job.logs() الرسالة No logs yet.	المهمة في الطابور أو بدأت للتو. انتظر نحو 15 ثانية.
تعيد job.result() القيمة {}	لم تنتهِ المهمة، أو أُلغيت، أو لم تحفظ نتيجة. راجع احفظ النتائج.
تعيد job.error_message() القيمة {}	أطلق سير العمل استثناءً دون الإبلاغ عن خطأ. اقرأ تتبّع الاستدعاءات باستخدام job.logs().
المهمة CANCELED، لكنك لم تلغها	بلغت المهمة حد 24 ساعة.

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

التوصيات

• اكتب السجلات وتعامل مع الأخطاء لإعداد السجلات والأخطاء التي تبلّغ عنها المهمة.
• شغّل مسحًا للمعاملات لتشغيل سير العمل مرة واحدة لكل مجموعة مدخلات.