Eseguire un workflow gestito

Versioni dei pacchetti

Il codice in questa pagina è stato sviluppato usando i seguenti requisiti. Ti consigliamo di usare queste versioni o successive.

qiskit-serverless~=0.38.0

Un job di workflow gestito funziona come un job batch che invii a uno scheduler HPC. run() invia il job e restituisce subito un oggetto Job. Usa l'oggetto Job per le attività che su un cluster sono coperte da squeue, dal file di output del job e da scancel. Il job continua a essere eseguito dopo che chiudi il notebook.

Gli esempi in questa pagina usano un piccolo workflow che legge un angolo, theta, e fallisce se l'angolo è fuori intervallo. Caricalo come descritto in Distribuisci il tuo primo workflow gestito.

# 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

Avviare un job

Ottieni il workflow tramite il suo titolo e chiama run() con gli argomenti che il workflow legge:

# 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() restituisce il controllo dopo che il workflow ha messo il job in coda. Salva l'ID del job in modo da poter trovare il job da una nuova sessione.

• Per scegliere quali argomenti accetta un workflow e controllarli prima dell'avvio del job, consulta Definire input e output.

• Per eseguire il job con una dimensione di calcolo diversa, passa function_size. Consulta Scegliere una dimensione per ogni esecuzione.

Seguire un job

Controllare lo stato

Subito dopo aver chiamato run(), lo stato è di solito QUEUED: run_workflow_remotely.ipynb

job.status()

'QUEUED'

Stato	Significato
QUEUED	Il job attende uno slot di calcolo.
INITIALIZING	Il workflow sta avviando un ambiente con il profilo di calcolo del job.
RUNNING	L'entrypoint del workflow è in esecuzione.
RUNNING: <stage>	L'entrypoint ha segnalato una fase con update_status(), ad esempio RUNNING: WAITING_FOR_QPU.
CANCELING	Hai annullato il job e il workflow lo sta fermando.
DONE	L'entrypoint è terminato con successo.
ERROR	L'entrypoint ha generato un errore o è terminato con un codice diverso da zero. Consulta Trovare la causa di un errore.
CANCELED	Hai annullato il job, oppure ha raggiunto il limite di 24 ore.

DONE, ERROR e CANCELED sono stati finali e job.in_terminal_state() restituisce True per essi. Un ciclo di controllo dello stato che attende un altro valore, come SUCCEEDED, non termina mai.

Attendere la fine del job

Per attendere la fine del job, chiama wait_for_final_state(). Restituisce lo stato finale e, con verbose=True, stampa ogni cambio di stato. Se imposti timeout in secondi, wait_for_final_state() solleva un TimeoutError quando il tempo scade prima della fine del job. Il job continua a essere eseguito:

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() controlla lo stato dopo 1, 2, 4, 8, 16 e 32 secondi, e poi ogni 60 secondi, quindi rileva la fine di un job breve in pochi secondi ed effettua poche richieste per un job lungo. Per controllare invece a intervalli fissi, passa poll_interval in secondi.

Leggere i log

Tutto ciò che il workflow scrive sullo standard output o sullo standard error finisce nei log del job. Puoi leggerli con job.logs() mentre il job è in esecuzione, prima che abbia un risultato:

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

• Il workflow carica le nuove righe di log circa ogni 15 secondi, e un'ultima volta quando il job termina o viene annullato. Subito dopo l'avvio di un job, job.logs() restituisce No logs yet.

• Il workflow conserva gli ultimi 50 MB di log per ogni job e aggiunge in cima una riga che indica che i log sono stati troncati.

Per decidere cosa registra il tuo workflow, consulta Scrivere i log. Per mostrare solo le righe che corrispondono a un'espressione regolare, passa include a job.filtered_logs(). Per nascondere righe, passa exclude.

Ottenere il risultato

job.result() attende la fine del job, poi restituisce il valore che il workflow ha passato a 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

Per impostazione predefinita, job.result() controlla lo stato ogni 30 secondi e attende senza limite di tempo. Passa maxwait per limitare il numero di controlli. Ad esempio, job.result(maxwait=40) attende circa 20 minuti. Per controllare una sola volta senza attendere, passa wait=False. Se job.result() smette di attendere prima della fine del job, restituisce {}.

Controllare quanto è durata ogni fase

job.events() elenca ogni cambio di stato e di fase di un job con il suo orario, così puoi vedere quanto il job ha atteso il calcolo e quanto è durata ogni fase: 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

Gli eventi usano i nomi di stato memorizzati. PENDING corrisponde a INITIALIZING e SUCCEEDED corrisponde a DONE. Per l'utilizzo della QPU di un'esecuzione, consulta Trovare i job QPU di un'esecuzione.

Trovare la causa di un errore

Quando il workflow solleva un'eccezione, il job termina con lo stato ERROR. Ad esempio, esegui il workflow con un angolo che rifiuta: 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'

I log mostrano il traceback: 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") == []

Se il workflow segnala gli errori con send_error(), leggili con job.events(type="ERROR") e leggi i loro messaggi con job.error_message(). Per un job senza errori segnalati, job.events(type="ERROR") restituisce una lista vuota e job.error_message() restituisce {}.

run_workflow_remotely.ipynb

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

Gestire i job

Annullare un job

Annulla un job con job.cancel(). Lo stato passa a CANCELING mentre il workflow ferma il job, e poi a 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() restituisce un messaggio sul job e sui suoi job QPU collegati, che cerca a sua volta di annullare. Per un job in esecuzione, il workflow invia SIGTERM al tuo entrypoint. Per salvare i risultati parziali, consulta Ripulire quando un job viene annullato. Se il job è già terminato, job.cancel() restituisce Job already in terminal state. e non modifica nulla.

Trovare un job da una nuova sessione

L'oggetto Job e il workflow esistono solo nella tua sessione Python corrente. Dopo aver riavviato il kernel di un notebook, passa l'ID del job salvato a client.job(), oppure elenca i tuoi job recenti con 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() restituisce i tuoi 10 job più recenti, dal più nuovo. Filtralo con function, status (dalla tabella degli stati) o created_after (un timestamp ISO 8601) e scorri i risultati con limit e offset. Per elencare i workflow che hai caricato, chiama client.list().

Anche la tabella Workloads su IBM Quantum Platform elenca i tuoi job, con stato, profilo di calcolo e tempi di esecuzione.

Limite di tempo

Un workflow gestito annulla un job che viene eseguito per più di 24 ore, e il job termina con lo stato CANCELED. Durante le esecuzioni lunghe, salva i checkpoint nell'archiviazione file per non perdere il lavoro.

Risolvere i problemi dei job

Problema	Causa
run() fallisce con Code: 429 e Active job limit reached.	Hai raggiunto il limite di job attivi. Attendi che alcuni job terminino, oppure annulla i job che non ti servono più.
Il job resta QUEUED e run() ha avvisato che your account, plan or subscription is blocked.	Il job non parte finché il blocco sul tuo account, piano o abbonamento non viene rimosso. Contatta il tuo amministratore.
job.logs() restituisce No logs yet.	Il job è in coda o è appena partito. Attendi circa 15 secondi.
job.result() restituisce {}	Il job non è terminato, è stato annullato o non ha salvato alcun risultato. Consulta Salvare i risultati.
job.error_message() restituisce {}	Il workflow ha sollevato un'eccezione senza segnalare un errore. Leggi il traceback con job.logs().
Il job è CANCELED, ma non l'hai annullato tu	Il job ha raggiunto il limite di 24 ore.

Passaggi successivi

Raccomandazioni

• Scrivere i log e gestire gli errori per configurare i log e gli errori che un job segnala.
• Eseguire uno sweep di parametri per eseguire il workflow una volta per ogni insieme di input.