Scrivi log e gestisci gli errori

Imposta il codice del tuo workflow in modo che scriva log, segnali gli errori al tuo notebook locale e salvi il lavoro parziale quando un job viene annullato. Per leggere questi output dal tuo notebook locale, vedi Esegui un workflow gestito.

Nel workflow	Nel tuo notebook locale
Scrivi con un logger del modulo logging, oppure chiama print().	job.logs() restituisce le righe.
Solleva un'eccezione, oppure esci con un codice diverso da zero.	job.status() restituisce ERROR, e job.logs() mostra il traceback.
Chiama send_error().	job.error_message() e job.events(type="ERROR") restituiscono il codice, il messaggio e i dettagli.
Chiama save_result() in un gestore di SIGTERM.	job.result() restituisce il risultato parziale di un job annullato.

Scrivi log

Tutto ciò che il workflow scrive sullo standard output o sullo standard error finisce nei log del job. Leggi i log con job.logs() mentre il job è in esecuzione o dopo che è terminato.

Imposta un logger

Usa il modulo logging della libreria standard di Python. logging.basicConfig() scrive sullo standard error, che raggiunge i log senza ritardo. Il formato nel seguente esempio aggiunge un timestamp a ogni riga:

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 nasconde i messaggi INFO di altri pacchetti, e logger.setLevel(logging.INFO) mantiene i messaggi INFO per il tuo logger. qiskit_ibm_runtime imposta il proprio livello con la variabile d'ambiente QISKIT_IBM_RUNTIME_LOG_LEVEL.

Anche print() raggiunge i log, ma lo standard output è bufferizzato, quindi le righe stampate potrebbero non comparire fino alla fine del job. Se usi print(), passa flush=True.

Imposta il livello di log per ogni esecuzione

Per fare il debug di una singola esecuzione senza modificare il codice, passa il livello di log come argomento:

run_workflow_remotely.ipynb

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

Per abilitarlo, imposta il livello del tuo logger a partire dall'argomento:

source_files/my_workflow.py

from qiskit_serverless import get_arguments

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

Se il workflow ha uno schema degli argomenti, dichiara log_level al suo interno, ad esempio come {"enum": ["DEBUG", "INFO", "WARNING"]}.

Buona pratica: decidi cosa registrare

Registra abbastanza da ricostruire un'esecuzione dai suoi log:

• All'inizio, registra ciò che l'esecuzione ha ricevuto. Registra il nome del backend, il numero di circuiti e le altre impostazioni che cambiano tra le esecuzioni.

• Registra l'inizio e la fine di ogni fase. Usa messaggi fissi che puoi cercare.

• Registra le iterazioni a intervalli fissi. Ad esempio, registra una riga come iteration 10/200, energy=-1.1000 ogni 10 iterazioni, invece che a ogni iterazione.

• Registra le eccezioni con logger.exception() all'interno di un blocco except. La riga include il traceback completo, anche se il workflow si riprende dall'errore.

• Disattiva le barre di avanzamento. I log memorizzano ogni aggiornamento di una barra di avanzamento, come quella di tqdm, come riga separata. Passa disable=True a tqdm, oppure registra a intervalli.

• Mantieni i log piccoli. Non registrare dati di grandi dimensioni, come ogni shot. Un workflow gestito conserva solo gli ultimi 50 MB di log per ogni job.

• Usa le parole "error" ed "exception" solo nei messaggi sui fallimenti. Se un job fallito non ha segnalato il suo errore, l'eccezione sollevata da job.result() contiene solo le righe di log che includono queste parole.

Buona pratica: segnala gli errori con send_error()

Quando il workflow solleva un'eccezione, il job termina con lo stato ERROR e il traceback va nei log. Un workflow gestito non archivia l'errore, nemmeno per un ServerlessError. job.error_message() restituisce {}, e l'eccezione sollevata da job.result() contiene solo le righe di log che includono error o exception. I controlli sui tipi e sugli intervalli degli argomenti appartengono a uno schema degli argomenti, che rifiuta un'esecuzione errata prima dell'avvio del job.

Segnala con send_error() ogni errore che interrompe il workflow, in modo che il tuo notebook locale riceva un errore breve con un codice e dei dettagli:

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() solleva una QiskitServerlessException con lo stesso testo.

Per abilitarlo, solleva ServerlessError dove si verifica l'errore e racchiudi il codice del tuo entrypoint in un blocco try che invia qualsiasi errore con send_error(). Un workflow aggiunge questo gestore alle Catalog Functions, ma non ai workflow gestiti:

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

• Solleva di nuovo l'errore. L'ultimo raise mantiene il traceback nei log e termina il job con lo stato ERROR.

• Crea una sottoclasse di ServerlessError per dare un nome a un gruppo di errori. Il gestore invia il nome della classe, come InputError, nel campo exception.

• Anche altre eccezioni raggiungono il gestore. Ricevono il codice 1, e il loro repr() diventa il messaggio, come ValueError('bad shots').

Segnala un problema e continua

send_error() non interrompe il workflow né ne cambia lo stato, quindi puoi segnalare anche problemi da cui il workflow si riprende, come ogni circuito che salta:

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") restituisce ogni errore segnalato dal workflow, inclusi gli errori di un job terminato con lo stato DONE. Vedi Trova la causa di un errore.

• Passa numeri e stringhe in args. error_message() mostra solo il primo elemento di una lista.

• Controlla il valore restituito. Se un workflow non riesce ad archiviare l'errore, ad esempio perché il codice viene eseguito fuori da un job, send_error() scrive l'errore nei log e restituisce False.

Fai pulizia quando un job viene annullato

Quando annulli un job, il workflow invia SIGTERM al tuo entrypoint, carica i log finali e arresta l'ambiente. Senza un gestore, Python si ferma immediatamente e i blocchi finally non vengono eseguiti. Il workflow cerca anche di annullare i job e le sessioni QPU che il workflow ha avviato con get_runtime_service(). Vedi Annulla i job QPU collegati.

Per mantenere il lavoro parziale, registra un gestore per SIGTERM. Il tuo notebook locale riceve allora il risultato parziale da un job annullato:

run_workflow_remotely.ipynb

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

Per abilitarlo, aggiungi il gestore all'inizio del tuo 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() nel gestore solleva SystemExit nel codice in esecuzione, quindi i blocchi finally vengono eseguiti prima che lo script termini. Il job termina comunque con lo stato CANCELED.

Mantieni il gestore breve, perché il workflow arresta l'ambiente poco dopo SIGTERM. Per le esecuzioni lunghe, salva anche dei checkpoint, in modo che un'esecuzione successiva possa riprendere dopo che un job fallisce o raggiunge il limite di 24 ore. Consulta Best practice: checkpoint dei job lunghi.

Passaggi successivi

Raccomandazioni

• Esegui un workflow gestito per avviare un job e leggerne stato, log e risultato.
• Gestisci i file tra le esecuzioni per salvare checkpoint da cui un'esecuzione successiva può riprendere.
• Scegli un profilo di calcolo per eseguire il tuo workflow con più risorse di CPU, memoria o GPU.