Écrire des logs et gérer les erreurs

Configure le code de ton workflow pour écrire des logs, signaler des erreurs à ton notebook local et enregistrer le travail partiel lorsqu'un job est annulé. Pour lire ces sorties depuis ton notebook local, consulte Exécuter un workflow géré.

Dans le workflow	Dans ton notebook local
Écris avec un logger du module `logging`, ou appelle `print()`.	`job.logs()` renvoie les lignes.
Lève une exception, ou quitte avec un code non nul.	`job.status()` renvoie `ERROR`, et `job.logs()` affiche la trace d'appels.
Appelle `send_error()`.	`job.error_message()` et `job.events(type="ERROR")` renvoient le code, le message et les détails.
Appelle `save_result()` dans un gestionnaire `SIGTERM`.	`job.result()` renvoie le résultat partiel d'un job annulé.

Écrire des logs

Tout ce que le workflow écrit sur la sortie standard ou l'erreur standard va dans les logs du job. Lis les logs avec job.logs() pendant l'exécution du job ou après sa fin.

Configurer un logger

Utilise le module logging de la bibliothèque standard Python. logging.basicConfig() écrit sur l'erreur standard, qui atteint les logs sans délai. Le format de l'exemple suivant ajoute un horodatage à chaque ligne :

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 masque les messages INFO des autres paquets, et logger.setLevel(logging.INFO) conserve les messages INFO pour ton propre logger. qiskit_ibm_runtime définit son propre niveau avec la variable d'environnement QISKIT_IBM_RUNTIME_LOG_LEVEL.

print() atteint aussi les logs, mais la sortie standard est mise en tampon, donc les lignes affichées peuvent n'apparaître qu'à la fin du job. Si tu utilises print(), passe flush=True.

Définir le niveau de log pour chaque exécution

Pour déboguer une exécution sans modifier le code, passe le niveau de log en argument :

run_workflow_remotely.ipynb

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

Pour activer cela, définis le niveau de ton logger à partir de l'argument :

source_files/my_workflow.py

from qiskit_serverless import get_arguments

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

Si le workflow a un schéma d'arguments, déclare-y log_level, par exemple sous la forme {"enum": ["DEBUG", "INFO", "WARNING"]}.

Bonne pratique : décider quoi journaliser

Journalise assez pour reconstituer une exécution à partir de ses logs :

Au début, journalise ce que l'exécution a reçu. Journalise le nom du backend, le nombre de circuits et les autres paramètres qui changent d'une exécution à l'autre.
Journalise le début et la fin de chaque étape. Utilise des messages fixes que tu peux rechercher.
Journalise les itérations à intervalle fixe. Par exemple, journalise une ligne comme iteration 10/200, energy=-1.1000 toutes les 10 itérations, plutôt qu'à chaque itération.
Journalise les exceptions avec logger.exception() dans un bloc except. La ligne inclut la trace d'appels complète, même si le workflow se rétablit après l'erreur.
Désactive les barres de progression. Les logs stockent chaque rafraîchissement d'une barre de progression, comme celle de tqdm, sous forme de ligne séparée. Passe disable=True à tqdm, ou journalise plutôt à intervalle régulier.
Garde les logs petits. Ne journalise pas de grandes quantités de données, comme chaque shot. Un workflow géré ne conserve que les 50 Mo de logs les plus récents pour chaque job.
Utilise les mots « error » et « exception » uniquement dans les messages concernant des échecs. Si un job en échec n'a pas signalé son erreur, l'exception que job.result() lève ne contient que les lignes de log qui comprennent ces mots.

Bonne pratique : signaler les erreurs avec send_error()

Lorsque le workflow lève une exception, le job se termine avec le statut ERROR, et la trace d'appels va dans les logs. Un workflow géré ne stocke pas l'erreur, même pour une ServerlessError. job.error_message() renvoie {}, et l'exception que job.result() lève ne contient que les lignes de log qui comprennent error ou exception. Les vérifications des types et des plages d'arguments relèvent d'un schéma d'arguments, qui rejette une mauvaise exécution avant le démarrage du job.

Signale avec send_error() chaque erreur qui arrête le workflow, afin que ton notebook local reçoive une erreur courte avec un code et des détails :

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() lève une QiskitServerlessException avec le même texte.

Pour activer cela, lève ServerlessError à l'endroit où l'erreur se produit, et entoure le code de ton point d'entrée d'un bloc try qui envoie toute erreur avec send_error(). Un workflow ajoute ce gestionnaire aux Catalog Functions, mais pas aux workflows gérés :

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

Relève l'erreur. Le raise final conserve la trace d'appels (traceback) dans les logs et termine le job avec le statut ERROR.
Crée une sous-classe de ServerlessError pour nommer un groupe d'erreurs. Le gestionnaire envoie le nom de la classe, par exemple InputError, dans le champ exception.
Les autres exceptions atteignent aussi le gestionnaire. Elles reçoivent le code 1, et leur repr() devient le message, par exemple ValueError('bad shots').

Signaler un problème et continuer

send_error() n'arrête pas le workflow et ne change pas son statut, tu peux donc aussi signaler des problèmes dont le workflow se remet, comme chaque circuit qu'il ignore :

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") renvoie toutes les erreurs que le workflow a signalées, y compris les erreurs d'un job qui s'est terminé avec le statut DONE. Consulte Trouver la cause d'une erreur.

Passe des nombres et des chaînes dans args. error_message() n'affiche que le premier élément d'une liste.
Vérifie la valeur de retour. Si un workflow ne peut pas stocker l'erreur, par exemple parce que le code s'exécute en dehors d'un job, send_error() écrit l'erreur dans les logs et renvoie False.

Nettoyer lorsqu'un job est annulé

Lorsque tu annules un job, le workflow envoie SIGTERM à ton point d'entrée, téléverse les logs finaux et arrête l'environnement. Sans gestionnaire, Python s'arrête immédiatement et les blocs finally ne s'exécutent pas. Le workflow essaie aussi d'annuler les jobs QPU et les sessions qu'il a démarrés avec get_runtime_service(). Consulte Annuler les jobs QPU liés.

Pour conserver le travail partiel, enregistre un gestionnaire pour SIGTERM. Ton notebook local obtient alors le résultat partiel d'un job annulé :

run_workflow_remotely.ipynb

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

Pour activer cela, ajoute le gestionnaire au début de ton point d'entrée :

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() dans le gestionnaire lève SystemExit dans le code en cours d'exécution, donc les blocs finally s'exécutent avant la fin du script. Le job se termine quand même avec le statut CANCELED.

Garde le gestionnaire court, car le workflow arrête l'environnement peu après SIGTERM. Pour les longues exécutions, enregistre aussi des points de contrôle, afin qu'une exécution ultérieure puisse reprendre après l'échec d'un job ou l'atteinte de la limite de 24 heures. Consulte Bonne pratique : placer des points de contrôle dans les longs jobs.

Prochaines étapes

Recommandations

Exécute un workflow géré pour démarrer un job et lire son statut, ses logs et son résultat.
Gère les fichiers entre les exécutions pour enregistrer des points de contrôle à partir desquels une exécution ultérieure peut reprendre.
Choisis un profil de calcul pour exécuter ton workflow avec plus de ressources CPU, de mémoire ou de GPU.

Écrire des logs​

Configurer un logger​

Définir le niveau de log pour chaque exécution​

Bonne pratique : décider quoi journaliser​

Bonne pratique : signaler les erreurs avec send_error()​

Signaler un problème et continuer​

Nettoyer lorsqu'un job est annulé​

Prochaines étapes​

Écrire des logs

Configurer un logger

Définir le niveau de log pour chaque exécution

Bonne pratique : décider quoi journaliser

Bonne pratique : signaler les erreurs avec send_error()

Signaler un problème et continuer

Nettoyer lorsqu'un job est annulé

Prochaines étapes