Logs schreiben und Fehler behandeln
Richte den Code deines Workflows so ein, dass er Logs schreibt, Fehler an dein lokales Notebook meldet und Teilergebnisse speichert, wenn ein Job abgebrochen wird. Wie du diese Ausgaben aus deinem lokalen Notebook liest, siehst du unter Einen verwalteten Workflow ausführen.
| Im Workflow | In deinem lokalen Notebook |
|---|---|
Schreibe mit einem Logger aus dem Modul logging oder rufe print() auf. | job.logs() gibt die Zeilen zurück. |
| Löse eine Exception aus oder beende mit einem Code ungleich null. | job.status() gibt ERROR zurück, und job.logs() zeigt den Traceback. |
Rufe send_error() auf. | job.error_message() und job.events(type="ERROR") geben Code, Meldung und Details zurück. |
Rufe save_result() in einem SIGTERM-Handler auf. | job.result() gibt das Teilergebnis eines abgebrochenen Jobs zurück. |
Logs schreiben
Alles, was der Workflow in die Standardausgabe oder den Standardfehler schreibt, landet in den Logs des Jobs. Lies die Logs mit job.logs(), während der Job läuft oder nachdem er beendet ist.
Einen Logger einrichten
Verwende das Modul logging aus der Python-Standardbibliothek. logging.basicConfig() schreibt in den Standardfehler, der ohne Verzögerung in den Logs ankommt. Das Format im folgenden Beispiel fügt jeder Zeile einen Zeitstempel hinzu:
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 blendet INFO-Meldungen anderer Pakete aus, und logger.setLevel(logging.INFO) behält INFO-Meldungen für deinen eigenen Logger bei. qiskit_ibm_runtime legt sein eigenes Level mit der Umgebungsvariable QISKIT_IBM_RUNTIME_LOG_LEVEL fest.
print() gelangt ebenfalls in die Logs, aber die Standardausgabe ist gepuffert, sodass ausgegebene Zeilen möglicherweise erst erscheinen, wenn der Job beendet ist. Wenn du print() verwendest, übergib flush=True.
Das Log-Level für jeden Lauf festlegen
Um einen Lauf zu debuggen, ohne den Code zu ändern, übergib das Log-Level als Argument:
run_workflow_remotely.ipynb
job = my_workflow.run(circuits=circuits, log_level="DEBUG")
Damit das funktioniert, setze das Level deines Loggers aus dem Argument:
source_files/my_workflow.py
from qiskit_serverless import get_arguments
arguments = get_arguments()
logger.setLevel(arguments.get("log_level", "INFO"))
Wenn der Workflow ein Argumentschema hat, deklariere darin log_level, zum Beispiel als {"enum": ["DEBUG", "INFO", "WARNING"]}.
Best Practice: entscheiden, was protokolliert wird
Protokolliere genug, um einen Lauf aus seinen Logs zu rekonstruieren:
-
Protokolliere zu Beginn, was der Lauf erhalten hat. Protokolliere den Backend-Namen, die Anzahl der Circuits und die anderen Einstellungen, die sich zwischen Läufen ändern.
-
Protokolliere Anfang und Ende jeder Phase. Verwende feste Meldungen, nach denen du suchen kannst.
-
Protokolliere Iterationen in einem festen Intervall. Protokolliere zum Beispiel alle 10 Iterationen eine Zeile wie
iteration 10/200, energy=-1.1000, statt bei jeder Iteration. -
Protokolliere Exceptions mit
logger.exception()innerhalb einesexcept-Blocks. Die Zeile enthält den vollständigen Traceback, selbst wenn sich der Workflow von dem Fehler erholt. -
Schalte Fortschrittsbalken aus. Die Logs speichern jede Aktualisierung eines Fortschrittsbalkens, etwa von
tqdm, als eigene Zeile. Übergibdisable=Trueantqdmoder protokolliere stattdessen in einem Intervall. -
Halte die Logs klein. Protokolliere keine großen Datenmengen, etwa jeden Shot. Ein verwalteter Workflow behält für jeden Job nur die neuesten 50 MB an Logs.
-
Verwende die Wörter "error" und "exception" nur in Meldungen über Fehlschläge. Wenn ein fehlgeschlagener Job seinen Fehler nicht gemeldet hat, enthält die Exception, die
job.result()auslöst, nur die Logzeilen, die diese Wörter enthalten.
Best Practice: Fehler mit send_error() melden
Wenn der Workflow eine Exception auslöst, endet der Job mit dem Status ERROR, und der Traceback landet in den Logs. Ein verwalteter Workflow speichert den Fehler nicht, auch nicht bei einem ServerlessError. job.error_message() gibt {} zurück, und die Exception, die job.result() auslöst, enthält nur die Logzeilen, die error oder exception enthalten. Prüfungen von Argumenttypen und -bereichen gehören in ein Argumentschema, das einen fehlerhaften Lauf vor dem Start des Jobs zurückweist.
Melde jeden Fehler, der den Workflow stoppt, mit send_error(), damit dein lokales Notebook einen kurzen Fehler mit Code und Details erhält:
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öst eine QiskitServerlessException mit demselben Text aus.
Damit das funktioniert, löse ServerlessError dort aus, wo der Fehler auftritt, und umschließe den Code deines Entrypoints mit einem try-Block, der jeden Fehler mit send_error() sendet. Ein Workflow fügt diesen Handler zu Catalog Functions hinzu, aber nicht zu verwalteten Workflows:
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
-
Löse den Fehler erneut aus. Das abschließende
raisebehält den Traceback in den Logs bei und beendet den Job mit dem StatusERROR. -
Bilde eine Unterklasse von
ServerlessError, um eine Gruppe von Fehlern zu benennen. Der Handler sendet den Namen der Klasse, etwaInputError, als Feldexception. -
Auch andere Exceptions erreichen den Handler. Sie erhalten den Code
1, und ihrrepr()wird zur Meldung, etwaValueError('bad shots').
Ein Problem melden und fortfahren
send_error() stoppt den Workflow nicht und ändert seinen Status nicht, sodass du auch Probleme melden kannst, von denen sich der Workflow erholt, etwa jeden Circuit, den er überspringt:
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") gibt jeden Fehler zurück, den der Workflow gemeldet hat, einschließlich der Fehler eines Jobs, der mit dem Status DONE endete. Siehe Die Ursache eines Fehlers finden.
-
Übergib Zahlen und Strings in
args.error_message()zeigt nur das erste Element einer Liste an. -
Prüfe den Rückgabewert. Wenn ein Workflow den Fehler nicht speichern kann, zum Beispiel weil der Code außerhalb eines Jobs läuft, schreibt
send_error()den Fehler in die Logs und gibtFalsezurück.
Aufräumen, wenn ein Job abgebrochen wird
Wenn du einen Job abbrichst, sendet der Workflow SIGTERM an deinen Entrypoint, lädt die endgültigen Logs hoch und fährt die Umgebung herunter. Ohne einen Handler stoppt Python sofort, und finally-Blöcke werden nicht ausgeführt. Der Workflow versucht außerdem, die QPU-Jobs und Sessions abzubrechen, die der Workflow mit get_runtime_service() gestartet hat. Siehe Verknüpfte QPU-Jobs abbrechen.
Um Teilergebnisse zu behalten, registriere einen Handler für SIGTERM. Dein lokales Notebook erhält dann das Teilergebnis eines abgebrochenen Jobs:
run_workflow_remotely.ipynb
job.cancel()
print(job.result()) # {'status': 'canceled', 'partial_results': [...]}
Um das zu aktivieren, füge den Handler am Anfang deines Entrypoints hinzu:
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() im Handler löst SystemExit im gerade laufenden Code aus, sodass finally-Blöcke ausgeführt werden, bevor das Skript endet. Der Job endet trotzdem mit dem Status CANCELED.
Halte den Handler kurz, weil der Workflow die Umgebung kurz nach SIGTERM herunterfährt. Speichere bei langen Läufen zusätzlich Checkpoints, damit ein späterer Lauf fortgesetzt werden kann, nachdem ein Job fehlgeschlagen ist oder das 24-Stunden-Limit erreicht hat. Siehe Best Practice: Checkpoints für lange Jobs.
Nächste Schritte
- Führe einen verwalteten Workflow aus, um einen Job zu starten und seinen Status, seine Logs und sein Ergebnis zu lesen.
- Verwalte Dateien über Läufe hinweg, um Checkpoints zu speichern, von denen ein späterer Lauf fortfahren kann.
- Wähle ein Compute-Profil, um deinen Workflow mit mehr CPU-, Speicher- oder GPU-Ressourcen auszuführen.