Magsulat ng mga log at humawak ng mga error
I-set up ang code ng workflow mo para magsulat ng mga log, mag-ulat ng mga error sa local notebook mo, at mag-save ng bahagyang gawa kapag na-cancel ang job. Para basahin ang mga output na ito mula sa local notebook mo, tingnan ang Magpatakbo ng managed workflow.
| Sa workflow | Sa local notebook mo |
|---|---|
Magsulat gamit ang logger mula sa logging module, o tawagin ang print(). | Ibinabalik ng job.logs() ang mga linya. |
| Magtaas ng exception, o lumabas na may nonzero na code. | Ibinabalik ng job.status() ang ERROR, at ipinapakita ng job.logs() ang traceback. |
Tawagin ang send_error(). | Ibinabalik ng job.error_message() at job.events(type="ERROR") ang code, mensahe, at mga detalye. |
Tawagin ang save_result() sa isang SIGTERM handler. | Ibinabalik ng job.result() ang bahagyang resulta ng na-cancel na job. |
Magsulat ng mga log
Anumang isinusulat ng workflow sa standard output o standard error ay napupunta sa mga log ng job. Basahin ang mga log gamit ang job.logs() habang tumatakbo ang job o pagkatapos nitong matapos.
Mag-set up ng logger
Gamitin ang logging module mula sa Python standard library. Nagsusulat ang logging.basicConfig() sa standard error, na umaabot sa mga log nang walang pagkaantala. Ang format sa sumusunod na halimbawa ay nagdaragdag ng timestamp sa bawat linya:
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
Itinatago ng level=logging.WARNING ang mga mensaheng INFO mula sa ibang mga package, at pinananatili ng logger.setLevel(logging.INFO) ang mga mensaheng INFO para sa sarili mong logger. Nagtatakda ang qiskit_ibm_runtime ng sarili nitong level gamit ang environment variable na QISKIT_IBM_RUNTIME_LOG_LEVEL.
Umaabot din sa mga log ang print(), pero naka-buffer ang standard output, kaya ang mga naka-print na linya ay maaaring hindi lumabas hanggang matapos ang job. Kung gagamit ka ng print(), ipasa ang flush=True.
Itakda ang log level para sa bawat run
Para i-debug ang isang run nang hindi binabago ang code, ipasa ang log level bilang argument:
run_workflow_remotely.ipynb
job = my_workflow.run(circuits=circuits, log_level="DEBUG")
Para magawa ito, itakda ang level ng logger mo mula sa argument:
source_files/my_workflow.py
from qiskit_serverless import get_arguments
arguments = get_arguments()
logger.setLevel(arguments.get("log_level", "INFO"))
Kung may argument schema ang workflow, ideklara rito ang log_level, halimbawa bilang {"enum": ["DEBUG", "INFO", "WARNING"]}.
Best practice: magpasya kung ano ang ila-log
Mag-log ng sapat para muling buuin ang isang run mula sa mga log nito:
-
Sa simula, i-log kung ano ang natanggap ng run. I-log ang pangalan ng backend, ang bilang ng mga circuit, at ang iba pang setting na nagbabago sa bawat run.
-
I-log ang simula at wakas ng bawat yugto. Gumamit ng mga nakapirming mensahe na mahahanap mo.
-
I-log ang mga iteration sa nakapirming pagitan. Halimbawa, mag-log ng linya gaya ng
iteration 10/200, energy=-1.1000kada 10 iteration, sa halip na bawat iteration. -
I-log ang mga exception gamit ang
logger.exception()sa loob ngexceptblock. Kasama sa linya ang buong traceback, kahit nakabawi ang workflow mula sa error. -
Patayin ang mga progress bar. Iniimbak ng mga log ang bawat refresh ng progress bar, gaya ng mula sa
tqdm, bilang hiwalay na linya. Ipasa angdisable=Truesatqdm, o mag-log sa isang pagitan. -
Panatilihing maliit ang mga log. Huwag mag-log ng malaking data, gaya ng bawat shot. Ang managed workflow ay nagpapanatili lang ng pinakabagong 50 MB ng mga log para sa bawat job.
-
Gamitin ang mga salitang "error" at "exception" sa mga mensahe lang tungkol sa mga pagkabigo. Kung hindi iniulat ng bigong job ang error nito, ang exception na itinataas ng
job.result()ay naglalaman lang ng mga linya ng log na may mga salitang ito.
Best practice: mag-ulat ng mga error gamit ang send_error()
Kapag nagtaas ng exception ang workflow, nagtatapos ang job na may status na ERROR, at napupunta sa mga log ang traceback. Hindi iniimbak ng managed workflow ang error, kahit para sa ServerlessError. Nagbabalik ang job.error_message() ng {}, at ang exception na itinataas ng job.result() ay naglalaman lang ng mga linya ng log na may error o exception. Ang mga pagsusuri sa type at saklaw ng argument ay nnnararapat sa argument schema, na tumatanggi sa maling run bago magsimula ang job.
Iulat ang bawat error na nagpapahinto sa workflow gamit ang send_error(), para makakuha ang local notebook mo ng maikling error na may code at mga detalye:
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
Nagtataas ang job.result() ng QiskitServerlessException na may parehong teksto.
Para magawa ito, magtaas ng ServerlessError kung saan nangyayari ang error, at balutin ang code ng entrypoint mo sa isang try block na nagpapadala ng anumang error gamit ang send_error(). Nagdaragdag ang workflow ng handler na ito sa Catalog Functions, pero hindi sa mga managed workflow:
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
-
Itaas muli ang error. Pinananatili ng huling
raiseang traceback sa mga log at tinatapos ang job na may status naERROR. -
Mag-subclass ng
ServerlessErrorpara pangalanan ang isang grupo ng mga error. Ipinapadala ng handler ang pangalan ng class, gaya ngInputError, bilang field naexception. -
Umaabot din sa handler ang ibang mga exception. Nakakakuha sila ng code na
1, at angrepr()nila ang nagiging mensahe, gaya ngValueError('bad shots').
Mag-ulat ng problema at magpatuloy
Hindi pinipigilan ng send_error() ang workflow o binabago ang status nito, kaya puwede mo ring i-report ang mga problemang nare-recover ng workflow, tulad ng bawat circuit na nilalaktawan nito:
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)
Ibinabalik ng job.events(type="ERROR") ang bawat error na na-report ng workflow, kasama ang mga error ng job na natapos na may status na DONE. Tingnan ang Hanapin ang sanhi ng error.
-
Mag-pass ng mga numero at string sa
args. Ipinapakita lang ngerror_message()ang unang item ng isang list. -
Tingnan ang return value. Kung hindi ma-store ng workflow ang error, halimbawa dahil tumatakbo ang code sa labas ng job, isinusulat ng
send_error()ang error sa mga log at nagbabalik ngFalse.
Maglinis kapag na-cancel ang job
Kapag nag-cancel ka ng job, nagpapadala ang workflow ng SIGTERM sa entrypoint mo, ina-upload ang mga huling log, at isinasara ang environment. Kung walang handler, agad na titigil ang Python, at hindi tatakbo ang mga finally block. Sinusubukan din ng workflow na i-cancel ang mga QPU job at session na sinimulan ng workflow gamit ang get_runtime_service(). Tingnan ang I-cancel ang mga naka-link na QPU job.
Para mapanatili ang partial na gawa, mag-register ng handler para sa SIGTERM. Makakakuha na ang local notebook mo ng partial result mula sa na-cancel na job:
run_workflow_remotely.ipynb
job.cancel()
print(job.result()) # {'status': 'canceled', 'partial_results': [...]}
Para i-enable ito, idagdag ang handler sa simula ng entrypoint mo:
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)
Nagtataas ang sys.exit() sa handler ng SystemExit sa code na tumatakbo, kaya tatakbo ang mga finally block bago matapos ang script. Natatapos pa rin ang job na may status na CANCELED.
Panatilihing maikli ang handler, dahil isinasara ng workflow ang environment pagkatapos ng SIGTERM. Para sa mahahabang run, mag-save din ng mga checkpoint, para makapag-resume ang susunod na run kapag nag-fail ang job o umabot sa 24-hour na limit. Tingnan ang Best practice: mag-checkpoint ng mahahabang job.
Mga susunod na hakbang
- Magpatakbo ng managed workflow para magsimula ng job at basahin ang status, mga log, at result nito.
- Pamahalaan ang mga file sa iba't ibang run para mag-save ng mga checkpoint na puwedeng ipagpatuloy ng susunod na run.
- Pumili ng compute profile para patakbuhin ang workflow mo gamit ang mas maraming CPU, memory, o GPU resource.