Uruchamianie zarządzanego przepływu pracy

Wersje pakietów

Kod na tej stronie został opracowany z użyciem następujących wymagań. Zalecamy używanie tych wersji lub nowszych.

qiskit-serverless~=0.38.0

Zadanie zarządzanego przepływu pracy działa jak zadanie wsadowe przesyłane do harmonogramu HPC. run() przesyła zadanie i od razu zwraca obiekt Job. Użyj obiektu Job do zadań, które w klastrze obsługują squeue, plik wyjściowy zadania i scancel. Zadanie działa dalej po zamknięciu notatnika.

Przykłady na tej stronie używają niewielkiego przepływu pracy, który odczytuje jeden kąt, theta, i kończy się błędem, gdy kąt jest poza zakresem. Prześlij go zgodnie z opisem w Wdrażanie pierwszego zarządzanego przepływu pracy.

# 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

Rozpocznij zadanie

Pobierz przepływ pracy według jego tytułu i wywołaj run() z argumentami, które przepływ odczytuje:

# 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() zwraca wynik po tym, jak przepływ pracy umieści zadanie w kolejce. Zapisz identyfikator zadania, aby móc znaleźć zadanie z nowej sesji.

Aby wybrać, jakie argumenty akceptuje przepływ pracy, i sprawdzić je przed rozpoczęciem zadania, zobacz Definiowanie wejść i wyjść.
Aby uruchomić zadanie z innym rozmiarem zasobów obliczeniowych, przekaż function_size. Zobacz Wybór rozmiaru dla każdego uruchomienia.

Śledzenie zadania

Sprawdź status

Tuż po wywołaniu run() status to zwykle QUEUED: run_workflow_remotely.ipynb

job.status()

'QUEUED'

Status	Znaczenie
`QUEUED`	Zadanie czeka na zasoby obliczeniowe.
`INITIALIZING`	Przepływ pracy uruchamia środowisko z profilem obliczeniowym zadania.
`RUNNING`	Punkt wejścia przepływu pracy jest uruchomiony.
`RUNNING: <stage>`	Punkt wejścia zgłosił etap za pomocą `update_status()`, na przykład `RUNNING: WAITING_FOR_QPU`.
`CANCELING`	Anulowano zadanie, a przepływ pracy je zatrzymuje.
`DONE`	Punkt wejścia zakończył się pomyślnie.
`ERROR`	Punkt wejścia zgłosił błąd lub zakończył się niezerowym kodem. Zobacz Znajdowanie przyczyny błędu.
`CANCELED`	Zadanie zostało anulowane przez Ciebie lub osiągnęło limit 24 godzin.

DONE, ERROR i CANCELED są stanami końcowymi, a job.in_terminal_state() zwraca dla nich True. Pętla sprawdzająca status, która czeka na inną wartość, na przykład SUCCEEDED, nigdy się nie kończy.

Czekaj na zakończenie zadania

Aby poczekać do zakończenia zadania, wywołaj wait_for_final_state(). Zwraca ono końcowy status, a z verbose=True wypisuje każdą zmianę statusu. Jeśli ustawisz timeout w sekundach, wait_for_final_state() zgłasza TimeoutError, gdy czas upłynie przed zakończeniem zadania. Zadanie działa dalej:

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() sprawdza status po 1, 2, 4, 8, 16 i 32 sekundach, a potem co 60 sekund, dzięki czemu wykrywa koniec krótkiego zadania w ciągu sekund i wysyła niewiele żądań dla długiego zadania. Aby zamiast tego sprawdzać w stałych odstępach, przekaż poll_interval w sekundach.

Odczytaj logi

Wszystko, co przepływ pracy zapisuje na standardowe wyjście lub standardowe wyjście błędów, trafia do logów zadania. Możesz je odczytać za pomocą job.logs() w trakcie działania zadania, zanim będzie miało wynik:

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

Przepływ pracy przesyła nowe linie logów mniej więcej co 15 sekund oraz po raz ostatni, gdy zadanie się kończy lub zostaje anulowane. Tuż po rozpoczęciu zadania job.logs() zwraca No logs yet.
Przepływ pracy przechowuje najnowsze 50 MB logów dla każdego zadania i dodaje na górze linię informującą, że logi zostały obcięte.

Aby zdecydować, co loguje Twój przepływ pracy, zobacz Zapisywanie logów. Aby pokazać tylko linie pasujące do wyrażenia regularnego, przekaż include do job.filtered_logs(). Aby ukryć linie, przekaż exclude.

Pobierz wynik

job.result() czeka do zakończenia zadania, a następnie zwraca wartość, którą przepływ pracy przekazał do 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

Domyślnie job.result() sprawdza status co 30 sekund i czeka bez limitu czasu. Przekaż maxwait, aby ograniczyć liczbę sprawdzeń. Na przykład job.result(maxwait=40) czeka około 20 minut. Aby sprawdzić jednorazowo bez czekania, przekaż wait=False. Jeśli job.result() przestanie czekać przed zakończeniem zadania, zwraca {}.

Sprawdź, jak długo trwał każdy etap

job.events() wyświetla każdą zmianę statusu i etapu zadania wraz z czasem, dzięki czemu widać, jak długo zadanie czekało na zasoby obliczeniowe i jak długo trwał każdy etap: 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

Zdarzenia używają zapisanych nazw statusów. PENDING oznacza INITIALIZING, a SUCCEEDED oznacza DONE. Informacje o użyciu QPU w danym uruchomieniu znajdziesz w Znajdowanie zadań QPU uruchomienia.

Znajdowanie przyczyny błędu

Gdy przepływ pracy zgłasza wyjątek, zadanie kończy się ze statusem ERROR. Na przykład uruchom przepływ pracy z kątem, który go odrzuca: 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'

Logi pokazują ślad stosu: 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") == []

Jeśli przepływ pracy zgłasza błędy za pomocą send_error(), odczytaj je za pomocą job.events(type="ERROR"), a ich komunikaty za pomocą job.error_message(). Dla zadania bez zgłoszonych błędów job.events(type="ERROR") zwraca pustą listę, a job.error_message() zwraca {}.

run_workflow_remotely.ipynb

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

Zarządzanie zadaniami

Anuluj zadanie

Anuluj zadanie za pomocą job.cancel(). Status zmienia się na CANCELING, gdy przepływ pracy zatrzymuje zadanie, a następnie na 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() zwraca komunikat o zadaniu i o jego powiązanych zadaniach QPU, które również próbuje anulować. Dla uruchomionego zadania przepływ pracy wysyła SIGTERM do Twojego punktu wejścia. Aby zapisać wyniki częściowe, zobacz Sprzątanie po anulowaniu zadania. Jeśli zadanie już się zakończyło, job.cancel() zwraca Job already in terminal state. i nic nie zmienia.

Znajdź zadanie z nowej sesji

Obiekt Job i przepływ pracy istnieją tylko w Twojej bieżącej sesji Pythona. Po ponownym uruchomieniu jądra notatnika przekaż zapisany identyfikator zadania do client.job() lub wyświetl swoje ostatnie zadania za pomocą 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() zwraca Twoje 10 ostatnich zadań, od najnowszego. Filtruj je za pomocą function, status (z tabeli statusów) lub created_after (znacznik czasu ISO 8601) i przeglądaj strony za pomocą limit i offset. Aby wyświetlić przesłane przepływy pracy, wywołaj client.list().

Tabela Workloads w IBM Quantum Platform również wyświetla Twoje zadania wraz ze statusem, profilem obliczeniowym i czasami wykonania.

Limit czasu

Zarządzany przepływ pracy anuluje zadanie, które działa dłużej niż 24 godziny, a zadanie kończy się ze statusem CANCELED. Podczas długich uruchomień zapisuj punkty kontrolne w magazynie plików, aby nie stracić pracy.

Rozwiązywanie problemów z zadaniami

Problem	Przyczyna
`run()` kończy się błędem `Code: 429` i `Active job limit reached.`	Osiągnięto limit aktywnych zadań. Poczekaj na zakończenie części zadań lub anuluj zadania, których już nie potrzebujesz.
Zadanie pozostaje w stanie `QUEUED`, a `run()` ostrzegło, że `your account, plan or subscription is blocked`.	Zadanie nie rozpocznie się, dopóki blokada konta, planu lub subskrypcji nie zostanie zdjęta. Skontaktuj się z administratorem.
`job.logs()` zwraca `No logs yet.`	Zadanie jest w kolejce lub właśnie się rozpoczęło. Poczekaj około 15 sekund.
`job.result()` zwraca `{}`	Zadanie jeszcze się nie zakończyło, zostało anulowane lub nie zapisało wyniku. Zobacz Zapisywanie wyników.
`job.error_message()` zwraca `{}`	Przepływ pracy zgłosił wyjątek bez zgłoszenia błędu. Odczytaj ślad stosu za pomocą `job.logs()`.
Zadanie ma status `CANCELED`, ale nie zostało anulowane przez Ciebie	Zadanie osiągnęło limit 24 godzin.

Kolejne kroki

Zalecenia

Zapisuj logi i obsługuj błędy, aby skonfigurować logi i błędy zgłaszane przez zadanie.
Uruchom przeszukiwanie parametrów, aby uruchomić przepływ pracy jednokrotnie dla każdego zestawu danych wejściowych.

Rozpocznij zadanie​

Śledzenie zadania​

Sprawdź status​

Czekaj na zakończenie zadania​

Odczytaj logi​

Pobierz wynik​

Sprawdź, jak długo trwał każdy etap​

Znajdowanie przyczyny błędu​

Zarządzanie zadaniami​

Anuluj zadanie​

Znajdź zadanie z nowej sesji​

Limit czasu​

Rozwiązywanie problemów z zadaniami​

Kolejne kroki​

Rozpocznij zadanie

Śledzenie zadania

Sprawdź status

Czekaj na zakończenie zadania

Odczytaj logi

Pobierz wynik

Sprawdź, jak długo trwał każdy etap

Znajdowanie przyczyny błędu

Zarządzanie zadaniami

Anuluj zadanie

Znajdź zadanie z nowej sesji

Limit czasu

Rozwiązywanie problemów z zadaniami

Kolejne kroki