Zapisuj logi i obsługuj błędy

Przygotuj kod przepływu pracy tak, aby zapisywał logi, zgłaszał błędy do lokalnego notatnika i zapisywał częściową pracę po anulowaniu zadania. Aby odczytać te dane wyjściowe z lokalnego notatnika, zobacz Uruchom zarządzany przepływ pracy.

W przepływie pracy	W lokalnym notatniku
Zapisuj za pomocą loggera z modułu `logging` lub wywołaj `print()`.	`job.logs()` zwraca te wiersze.
Zgłoś wyjątek lub zakończ z niezerowym kodem.	`job.status()` zwraca `ERROR`, a `job.logs()` pokazuje ślad stosu.
Wywołaj `send_error()`.	`job.error_message()` i `job.events(type="ERROR")` zwracają kod, komunikat i szczegóły.
Wywołaj `save_result()` w obsłudze `SIGTERM`.	`job.result()` zwraca częściowy wynik anulowanego zadania.

Zapisuj logi

Wszystko, co przepływ pracy zapisuje na standardowe wyjście lub standardowe wyjście błędów, trafia do logów zadania. Odczytaj logi za pomocą job.logs() podczas działania zadania lub po jego zakończeniu.

Skonfiguruj logger

Użyj modułu logging z biblioteki standardowej Pythona. logging.basicConfig() zapisuje na standardowe wyjście błędów, które trafia do logów bez opóźnienia. Format w poniższym przykładzie dodaje znacznik czasu do każdego wiersza:

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 ukrywa komunikaty INFO z innych pakietów, a logger.setLevel(logging.INFO) zachowuje komunikaty INFO dla twojego własnego loggera. qiskit_ibm_runtime ustawia własny poziom za pomocą zmiennej środowiskowej QISKIT_IBM_RUNTIME_LOG_LEVEL.

print() również trafia do logów, ale standardowe wyjście jest buforowane, więc wypisane wiersze mogą pojawić się dopiero po zakończeniu zadania. Jeśli używasz print(), przekaż flush=True.

Ustaw poziom logowania dla każdego uruchomienia

Aby zdebugować jedno uruchomienie bez zmiany kodu, przekaż poziom logowania jako argument:

run_workflow_remotely.ipynb

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

Aby to umożliwić, ustaw poziom swojego loggera na podstawie argumentu:

source_files/my_workflow.py

from qiskit_serverless import get_arguments

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

Jeśli przepływ pracy ma schemat argumentów, zadeklaruj w nim log_level, na przykład jako {"enum": ["DEBUG", "INFO", "WARNING"]}.

Najlepsza praktyka: zdecyduj, co logować

Loguj tyle, aby dało się odtworzyć uruchomienie na podstawie logów:

Na początku zapisz, co otrzymało uruchomienie. Zapisz nazwę backendu, liczbę obwodów i pozostałe ustawienia, które zmieniają się między uruchomieniami.
Loguj początek i koniec każdego etapu. Używaj stałych komunikatów, które można wyszukać.
Loguj iteracje w stałych odstępach. Na przykład zapisuj wiersz taki jak iteration 10/200, energy=-1.1000 co 10 iteracji zamiast co każdą iterację.
Loguj wyjątki za pomocą logger.exception() wewnątrz bloku except. Wiersz zawiera pełny ślad stosu, nawet jeśli przepływ pracy odzyska sprawność po błędzie.
Wyłącz paski postępu. Logi zapisują każde odświeżenie paska postępu, na przykład z tqdm, jako osobny wiersz. Przekaż disable=True do tqdm lub loguj w odstępach.
Utrzymuj małe logi. Nie loguj dużych danych, takich jak każdy shot. Zarządzany przepływ pracy przechowuje tylko najnowsze 50 MB logów dla każdego zadania.
Używaj słów „error” i „exception” tylko w komunikatach o awariach. Jeśli nieudane zadanie nie zgłosiło swojego błędu, wyjątek zgłaszany przez job.result() zawiera tylko wiersze logów zawierające te słowa.

Najlepsza praktyka: zgłaszaj błędy za pomocą send_error()

Gdy przepływ pracy zgłasza wyjątek, zadanie kończy się statusem ERROR, a ślad stosu trafia do logów. Zarządzany przepływ pracy nie przechowuje błędu, nawet w przypadku ServerlessError. job.error_message() zwraca {}, a wyjątek zgłaszany przez job.result() zawiera tylko wiersze logów zawierające error lub exception. Kontrole typów i zakresów argumentów należą do schematu argumentów, który odrzuca błędne uruchomienie przed startem zadania.

Zgłaszaj za pomocą send_error() każdy błąd, który zatrzymuje przepływ pracy, aby lokalny notatnik otrzymał krótki błąd z kodem i szczegółami:

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() zgłasza QiskitServerlessException z tym samym tekstem.

Aby to umożliwić, zgłoś ServerlessError tam, gdzie występuje błąd, i otocz kod punktu wejścia blokiem try, który wysyła każdy błąd za pomocą send_error(). Przepływ pracy dodaje tę obsługę do Catalog Functions, ale nie do zarządzanych przepływów pracy:

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

Zgłoś błąd ponownie. Końcowe raise zachowuje ślad stosu w logach i kończy zadanie ze statusem ERROR.
Utwórz podklasę ServerlessError, aby nazwać grupę błędów. Obsługa wysyła nazwę klasy, na przykład InputError, jako pole exception.
Inne wyjątki również trafiają do obsługi. Dostają kod 1, a ich repr() staje się komunikatem, na przykład ValueError('bad shots').

Zgłoś problem i kontynuuj

send_error() nie zatrzymuje przepływu pracy ani nie zmienia jego statusu, więc możesz też zgłaszać problemy, z których przepływ pracy się odzyskuje, na przykład każdy obwód, który pomija:

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") zwraca każdy błąd zgłoszony przez przepływ pracy, w tym błędy zadania, które zakończyło się statusem DONE. Zobacz Znajdowanie przyczyny błędu.

Przekazuj liczby i łańcuchy znaków w args. error_message() pokazuje tylko pierwszy element listy.
Sprawdź zwróconą wartość. Jeśli przepływ pracy nie może zapisać błędu, na przykład ponieważ kod działa poza zadaniem, send_error() zapisuje błąd w logach i zwraca False.

Sprzątanie po anulowaniu zadania

Gdy anulujesz zadanie, przepływ pracy wysyła SIGTERM do twojego punktu wejścia, przesyła końcowe logi i wyłącza środowisko. Bez procedury obsługi Python zatrzymuje się natychmiast, a bloki finally nie są wykonywane. Przepływ pracy próbuje też anulować zadania QPU i sesje, które uruchomił za pomocą get_runtime_service(). Zobacz Anulowanie powiązanych zadań QPU.

Aby zachować częściowe wyniki pracy, zarejestruj procedurę obsługi dla SIGTERM. Twój lokalny notebook otrzyma wtedy częściowy wynik z anulowanego zadania:

run_workflow_remotely.ipynb

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

Aby to włączyć, dodaj procedurę obsługi na początku punktu wejścia:

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() w procedurze obsługi zgłasza SystemExit w uruchomionym kodzie, więc bloki finally są wykonywane przed zakończeniem skryptu. Zadanie nadal kończy się statusem CANCELED.

Utrzymuj procedurę obsługi krótką, ponieważ przepływ pracy wyłącza środowisko wkrótce po SIGTERM. W przypadku długich uruchomień zapisuj też punkty kontrolne, aby kolejne uruchomienie mogło wznowić pracę po niepowodzeniu zadania lub osiągnięciu 24-godzinnego limitu. Zobacz Najlepsza praktyka: punkty kontrolne w długich zadaniach.

Kolejne kroki

Zalecenia

Uruchom zarządzany przepływ pracy, aby rozpocząć zadanie i odczytać jego status, logi i wynik.
Zarządzaj plikami między uruchomieniami, aby zapisywać punkty kontrolne, od których kolejne uruchomienie może wznowić pracę.
Wybierz profil obliczeniowy, aby uruchamiać przepływ pracy z większą liczbą zasobów CPU, pamięci lub GPU.

Zapisuj logi​

Skonfiguruj logger​

Ustaw poziom logowania dla każdego uruchomienia​

Najlepsza praktyka: zdecyduj, co logować​

Najlepsza praktyka: zgłaszaj błędy za pomocą send_error()​

Zgłoś problem i kontynuuj​

Sprzątanie po anulowaniu zadania​

Kolejne kroki​

Zapisuj logi

Skonfiguruj logger

Ustaw poziom logowania dla każdego uruchomienia

Najlepsza praktyka: zdecyduj, co logować

Najlepsza praktyka: zgłaszaj błędy za pomocą send_error()

Zgłoś problem i kontynuuj

Sprzątanie po anulowaniu zadania

Kolejne kroki