ログの書き込みとエラーの処理
ワークフローのコードを設定して、ログを書き込み、ローカルのノートブックにエラーを報告し、ジョブがキャンセルされたときに途中までの作業を保存します。これらの出力をローカルのノートブックから読み取るには、マネージド・ワークフローを実行するを参照してください。
| ワークフロー内 | ローカルのノートブック |
|---|---|
logging モジュールのロガーで書き込むか、print() を呼び出す。 | job.logs() がその行を返します。 |
| 例外を発生させるか、ゼロ以外のコードで終了する。 | job.status() は ERROR を返し、job.logs() はトレースバックを表示します。 |
send_error() を呼び出す。 | job.error_message() と job.events(type="ERROR") が、コード、メッセージ、詳細を返します。 |
SIGTERM ハンドラーで save_result() を呼び出す。 | job.result() は、キャンセルされたジョブの途中までの結果を返します。 |
ログの書き込み
ワークフローが標準出力または標準エラーに書き込んだ内容は、すべてジョブのログに送られます。ログは、ジョブの実行中または終了後に job.logs() で読み取ります。
ロガーの設定
Python 標準ライブラリの logging モジュールを使用します。logging.basicConfig() は標準エラーに書き込み、遅延なくログに届きます。次の例の形式では、各行にタイムスタンプが追加されます。
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 は他のパッケージの INFO メッセージを非表示にし、logger.setLevel(logging.INFO) は自分のロガーの INFO メッセージを保持します。qiskit_ibm_runtime は、環境変数 QISKIT_IBM_RUNTIME_LOG_LEVEL で独自のレベルを設定します。
print() もログに届きますが、標準出力はバッファリングされるため、出力した行はジョブが終了するまで表示されないことがあります。print() を使用する場合は、flush=True を渡してください。
実行ごとのログレベルの設定
コードを変更せずに1回の実行をデバッグするには、ログレベルを引数として渡します。
run_workflow_remotely.ipynb
job = my_workflow.run(circuits=circuits, log_level="DEBUG")
これを有効にするには、引数からロガーのレベルを設定します。
source_files/my_workflow.py
from qiskit_serverless import get_arguments
arguments = get_arguments()
logger.setLevel(arguments.get("log_level", "INFO"))
ワークフローに引数スキーマがある場合は、その中で log_level を宣言します。たとえば {"enum": ["DEBUG", "INFO", "WARNING"]} のようにします。
ベストプラクティス:ログに記録する内容を決める
ログから実行を再現できるだけの情報を記録します。
-
開始時に、実行が受け取った内容を記録します。 バックエンド名、回路の数、実行ごとに変わるその他の設定を記録します。
-
各ステージの開始と終了を記録します。 検索できる固定のメッセージを使用します。
-
反復は一定の間隔で記録します。 たとえば、すべての反復ではなく、10回の反復ごとに
iteration 10/200, energy=-1.1000のような行を記録します。 -
exceptブロック内でlogger.exception()を使って例外を記録します。 ワークフローがエラーから回復する場合でも、その行には完全なトレースバックが含まれます。 -
進行状況バーを無効にします。 ログには、
tqdmなどの進行状況バーの更新が、1回ごとに別の行として保存されます。tqdmにdisable=Trueを渡すか、代わりに一定間隔でログを記録してください。 -
ログを小さく保ちます。 すべてのショットなど、大きなデータは記録しないでください。マネージド・ワークフローは、ジョブごとに最新の 50 MB のログのみを保持します。
-
「error」と「exception」という語は、失敗に関するメッセージにのみ使用します。 失敗したジョブがエラーを報告しなかった場合、
job.result()が発生させる例外には、これらの語を含むログ行のみが含まれます。
ベストプラクティス:send_error() でエラーを報告する
ワークフローが例外を発生させると、ジョブは ERROR ステータスで終了し、トレースバックがログに送られます。マネージド・ワークフローは、ServerlessError の場合でも、エラーを保存しません。job.error_message() は {} を返し、job.result() が発生させる例外には、error または exception を含むログ行のみが含まれます。引数の型や範囲のチェックは、ジョブの開始前に不正な実行を拒否する引数スキーマで行うのが適切です。
ワークフローを停止させるすべてのエラーを send_error() で報告すると、ローカルのノートブックはコードと詳細を含む短いエラーを受け取れます。
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() は、同じテキストの QiskitServerlessException を発生させます。
これを有効にするには、エラーが発生した場所で ServerlessError を発生させ、エントリーポイントのコードを try ブロックで囲み、あらゆるエラーを send_error() で送信します。ワークフローはこのハンドラーを Catalog Functions には追加しますが、マネージド・ワークフローには追加しません。
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
-
エラーを再度発生させます。 最後の
raiseにより、トレースバックがログに残り、ジョブはERRORステータスで終了します。 -
ServerlessErrorをサブクラス化して、エラーのグループに名前を付けます。 ハンドラーは、InputErrorなどのクラス名をexceptionフィールドとして送信します。 -
他の例外もハンドラーに到達します。 それらにはコード
1が付き、repr()の結果がValueError('bad shots')のようにメッセージになります。
問題を報告して続行する
send_error() はワークフローを停止せず、ステータスも変更しません。そのため、スキップした各回路など、ワークフローが回復できる問題も報告できます。
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") は、DONE ステータスで終了したジョブのエラーも含め、ワークフローが報告したすべてのエラーを返します。エラーの原因を調べるを参照してください。
-
argsには数値と文字列を渡します。error_message()は、リストの最初の項目のみを表示します。 -
戻り値を確認してください。 コードがジョブの外で実行されているなどの理由で、ワークフローがエラーを保存できない場合、
send_error()はエラーをログに書き込み、Falseを返します。
ジョブがキャンセルされたときのクリーンアップ
ジョブをキャンセルすると、ワークフローはエントリーポイントに SIGTERM を送信し、最終ログをアップロードして、環境をシャットダウンします。ハンドラーがない場合、Python はすぐに停止し、finally ブロックは実行されません。ワークフローは、get_runtime_service() で開始した QPU ジョブとセッションのキャンセルも試みます。リンクされた QPU ジョブのキャンセルを参照してください。
途中までの作業を保持するには、SIGTERM のハンドラーを登録します。すると、ローカルのノートブックは、キャンセルされたジョブから途中までの結果を取得できます。
run_workflow_remotely.ipynb
job.cancel()
print(job.result()) # {'status': 'canceled', 'partial_results': [...]}
これを有効にするには、エントリーポイントの冒頭にハンドラーを追加します。
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() は、実行中のコードで SystemExit を発生させるため、スクリプトが終了する前に finally ブロックが実行されます。ジョブは引き続き CANCELED ステータスで終了します。
SIGTERM の後、ワークフローはまもなく環境をシャットダウンするため、ハンドラーは短くしてください。長時間の実行では、ジョブが失敗した場合や24時間の上限に達した場合に、後の実行で再開できるよう、チェックポイントも保存してください。ベストプラクティス:長時間のジョブにチェックポイントを設定するを参照してください。
次のステップ
- マネージド・ワークフローを実行すると、ジョブを開始し、そのステータス、ログ、結果を読み取れます。
- 実行をまたいでファイルを管理すると、後の実行で再開できるチェックポイントを保存できます。
- コンピュートプロファイルを選択すると、より多くの CPU、メモリー、GPU リソースでワークフローを実行できます。