Scrie loguri și gestionează erorile

Configurează codul workflow-ului tău să scrie loguri, să raporteze erori către notebookul tău local și să salveze munca parțială când un job este anulat. Pentru a citi aceste ieșiri din notebookul tău local, vezi Rulează un workflow gestionat.

În workflow	În notebookul tău local
Scrie cu un logger din modulul logging sau apelează print().	job.logs() returnează liniile.
Ridică o excepție sau ieși cu un cod diferit de zero.	job.status() returnează ERROR, iar job.logs() arată traceback-ul.
Apelează send_error().	job.error_message() și job.events(type="ERROR") returnează codul, mesajul și detaliile.
Apelează save_result() într-un handler SIGTERM.	job.result() returnează rezultatul parțial al unui job anulat.

Scrie loguri

Tot ce scrie workflow-ul la ieșirea standard sau la eroarea standard ajunge în logurile jobului. Citește logurile cu job.logs() cât timp rulează jobul sau după ce se încheie.

Configurează un logger

Folosește modulul logging din biblioteca standard Python. logging.basicConfig() scrie la eroarea standard, care ajunge în loguri fără întârziere. Formatul din exemplul următor adaugă o marcă de timp fiecărei linii:

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 ascunde mesajele INFO din alte pachete, iar logger.setLevel(logging.INFO) păstrează mesajele INFO pentru propriul tău logger. qiskit_ibm_runtime își setează propriul nivel cu variabila de mediu QISKIT_IBM_RUNTIME_LOG_LEVEL.

print() ajunge și el în loguri, dar ieșirea standard este bufferizată, deci liniile afișate ar putea apărea abia când se încheie jobul. Dacă folosești print(), transmite flush=True.

Setează nivelul de logare pentru fiecare rulare

Pentru a depana o rulare fără a schimba codul, transmite nivelul de logare ca argument:

run_workflow_remotely.ipynb

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

Pentru a activa acest lucru, setează nivelul loggerului tău din argument:

source_files/my_workflow.py

from qiskit_serverless import get_arguments

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

Dacă workflow-ul are o schemă de argumente, declară în ea log_level, de exemplu ca {"enum": ["DEBUG", "INFO", "WARNING"]}.

Bună practică: decide ce să loghezi

Loghează suficient cât să reconstitui o rulare din loguri:

• La început, loghează ce a primit rularea. Loghează numele backendului, numărul de circuite și celelalte setări care se schimbă între rulări.

• Loghează începutul și sfârșitul fiecărei etape. Folosește mesaje fixe pe care le poți căuta.

• Loghează iterațiile la un interval fix. De exemplu, loghează o linie precum iteration 10/200, energy=-1.1000 la fiecare 10 iterații, în loc de fiecare iterație.

• Loghează excepțiile cu logger.exception() într-un bloc except. Linia include traceback-ul complet, chiar dacă workflow-ul își revine după eroare.

• Dezactivează barele de progres. Logurile stochează fiecare reîmprospătare a unei bare de progres, precum una din tqdm, ca linie separată. Transmite disable=True către tqdm sau loghează la interval.

• Păstrează logurile mici. Nu loga date mari, precum fiecare shot. Un workflow gestionat păstrează doar cei mai recenți 50 MB de loguri pentru fiecare job.

• Folosește cuvintele „error” și „exception” doar în mesaje despre eșecuri. Dacă un job eșuat nu și-a raportat eroarea, excepția pe care o ridică job.result() conține doar liniile de log care includ aceste cuvinte.

Bună practică: raportează erorile cu send_error()

Când workflow-ul ridică o excepție, jobul se încheie cu statusul ERROR, iar traceback-ul ajunge în loguri. Un workflow gestionat nu stochează eroarea, nici măcar pentru o ServerlessError. job.error_message() returnează {}, iar excepția pe care o ridică job.result() conține doar liniile de log care includ error sau exception. Verificările tipurilor și intervalelor argumentelor aparțin unei scheme de argumente, care respinge o rulare greșită înainte de pornirea jobului.

Raportează cu send_error() fiecare eroare care oprește workflow-ul, astfel încât notebookul tău local să primească o eroare scurtă cu un cod și detalii:

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() ridică o QiskitServerlessException cu același text.

Pentru a activa acest lucru, ridică ServerlessError unde apare eroarea și învelește codul punctului tău de intrare într-un bloc try care trimite orice eroare cu send_error(). Un workflow adaugă acest handler la Catalog Functions, dar nu la workflow-urile gestionate:

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

• Ridică eroarea din nou. raise final păstrează traceback-ul în loguri și încheie jobul cu statusul ERROR.

• Creează o subclasă a ServerlessError pentru a denumi un grup de erori. Handlerul trimite numele clasei, precum InputError, ca și câmp exception.

• Alte excepții ajung și ele la handler. Primesc codul 1, iar repr() al lor devine mesajul, precum ValueError('bad shots').

Raportează o problemă și continuă

send_error() nu oprește workflow-ul și nu îi schimbă starea, așa că poți raporta și probleme din care workflow-ul își revine, cum ar fi fiecare circuit pe care îl omite:

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") returnează fiecare eroare raportată de workflow, inclusiv erorile unui job care s-a încheiat cu starea DONE. Vezi Găsește cauza unei erori.

• Transmite numere și șiruri de caractere în args. error_message() afișează doar primul element al unei liste.

• Verifică valoarea returnată. Dacă un workflow nu poate stoca eroarea, de exemplu pentru că), codul rulează în afara unui job, send_error() scrie eroarea în jurnale și returnează False.

Curăță resursele când un job este anulat

Când anulezi un job, workflow-ul trimite SIGTERM către entrypoint-ul tău, încarcă jurnalele finale și oprește mediul. Fără un handler, Python se oprește imediat, iar blocurile finally nu rulează. Workflow-ul încearcă, de asemenea, să anuleze job-urile QPU și sesiunile pe care workflow-ul le-a pornit cu get_runtime_service(). Vezi Anulează job-urile QPU asociate.

Pentru a păstra munca parțială, înregistrează un handler pentru SIGTERM. Notebook-ul tău local primește astfel rezultatul parțial al unui job anulat:

run_workflow_remotely.ipynb

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

Pentru a activa acest lucru, adaugă handler-ul la începutul entrypoint-ului tău:

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() în handler ridică SystemExit în codul care rulează, așa că blocurile finally rulează înainte de încheierea scriptului. Job-ul se încheie în continuare cu starea CANCELED.

Păstrează handler-ul scurt, deoarece workflow-ul oprește mediul la scurt timp după SIGTERM. Pentru rulările lungi, salvează și puncte de control, astfel încât o rulare ulterioară să poată relua după ce un job eșuează sau atinge limita de 24 de ore. Vezi Bune practici: creează puncte de control pentru job-urile lungi.

Pașii următori

Recomandări

• Rulează un workflow gestionat pentru a porni un job și a-i citi starea, jurnalele și rezultatul.
• Gestionează fișierele între rulări pentru a salva puncte de control de la care o rulare ulterioară poate relua.
• Alege un profil de calcul pentru a rula workflow-ul cu mai mult CPU, memorie sau resurse GPU.