マネージドワークフローを実行する

パッケージのバージョン

このページのコードは、次の要件を使用して開発されました。 これらのバージョン、またはそれ以降のバージョンの使用をお勧めします。

qiskit-serverless~=0.38.0

マネージドワークフローのジョブは、HPC スケジューラーに投入するバッチジョブのように動作します。run() はジョブを投入し、すぐに Job オブジェクトを返します。クラスターで squeue、ジョブの出力ファイル、scancel が担うタスクには、Job オブジェクトを使用します。ノートブックを閉じても、ジョブは実行を続けます。

このページの例では、1つの角度 theta を読み取り、その角度が範囲外の場合は失敗する小さなワークフローを使用します。最初のマネージドワークフローをデプロイするの説明に従ってアップロードしてください。

# Added by doQumentation — installs the packages this notebook needs if they are missing
import importlib.util

_needed = {"qiskit_ibm_catalog": "qiskit-ibm-catalog", "qiskit_serverless": "qiskit-serverless"}
_missing = [pip for module, pip in _needed.items()
            if importlib.util.find_spec(module) is None]
# One at a time, so a package that fails to install does not block the others
for _pip in _missing:
    %pip install -q {_pip}
if not _missing:
    print("\u2713 All required packages are installed")

# This cell is hidden from users. It creates the working directory.
from pathlib import Path

Path("./source_files").mkdir(exist_ok=True)

%%writefile ./source_files/my_workflow.py
import math

from qiskit_serverless import get_arguments, save_result

arguments = get_arguments()
theta = arguments["theta"]
if not 0 <= theta <= math.pi:
    raise ValueError(f"theta must be between 0 and pi, got {theta}")

print(f"Computing the expectation value for theta={theta}", flush=True)
save_result({"theta": theta, "expectation_value": math.cos(theta)})

Writing ./source_files/my_workflow.py

ジョブを開始する

タイトルでワークフローを取得し、ワークフローが読み取る引数を指定して run() を呼び出します。

# This cell is hidden from users. It uploads the workflow under a title that
# other test notebooks do not use, and makes load("my-workflow") return it.
# It also silences the account warning that job.cancel() logs.
import logging

from qiskit_ibm_catalog import ManagedWorkflow, ManagedWorkflowClient

logging.getLogger("qiskit_ibm_runtime").setLevel(logging.ERROR)

_test_title = "docs-run-managed-workflow"
ManagedWorkflowClient().upload(
    ManagedWorkflow(
        title=_test_title,
        entrypoint="my_workflow.py",
        working_dir="./source_files/",
    )
)
_load = ManagedWorkflowClient.load
ManagedWorkflowClient.load = lambda self, title, *a, **k: _load(
    self, _test_title if title == "my-workflow" else title, *a, **k
)

run_workflow_remotely.ipynb

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
my_workflow = client.load("my-workflow")

job = my_workflow.run(theta=0.5)
job_id = job.job_id

run() は、ワークフローがジョブをキューに入れた後に戻ります。新しいセッションからジョブを見つけることができるように、ジョブ ID を保存してください。

• ワークフローが受け入れる引数を選び、ジョブの開始前にそれらを検証する方法については、入力と出力を定義するを参照してください。

• 別のコンピュートサイズでジョブを実行するには、function_size を渡します。実行ごとにサイズを選択するを参照してください。

ジョブを追跡する

ステータスを確認する

run() を呼び出した直後は、通常、ステータスは QUEUED です。 run_workflow_remotely.ipynb

job.status()

'QUEUED'

ステータス	意味
QUEUED	ジョブはコンピュートスロットを待っています。
INITIALIZING	ワークフローが、ジョブのコンピュートプロファイルで環境を起動しています。
RUNNING	ワークフローのエントリポイントが実行中です。
RUNNING: <stage>	エントリポイントが update_status() でステージを報告しました。例: RUNNING: WAITING_FOR_QPU。
CANCELING	ジョブをキャンセルし、ワークフローがそれを停止しています。
DONE	エントリポイントが正常に終了しました。
ERROR	エントリポイントがエラーを発生させたか、ゼロ以外のコードで終了しました。エラーの原因を調べるを参照してください。
CANCELED	ジョブをキャンセルしたか、24時間の上限に達しました。

DONE、ERROR、CANCELED は最終状態であり、これらに対して job.in_terminal_state() は True を返します。SUCCEEDED など別の値を待つステータスループは、終了しません。

ジョブの終了を待つ

ジョブが終了するまで待つには、wait_for_final_state() を呼び出します。これは最終ステータスを返し、verbose=True を指定するとステータスが変わるたびに出力します。timeout を秒単位で設定すると、ジョブが終了する前に時間切れになった場合、wait_for_final_state() は TimeoutError を発生させます。ジョブは実行を続けます。

run_workflow_remotely.ipynb

job.wait_for_final_state(timeout=600, verbose=True)

Waiting for job 3f2a1c0e-6b1d-4c0e-9a51-2f0c4d1e7a90
0:00:00  QUEUED
0:02:10  RUNNING
0:03:05  RUNNING: WAITING_FOR_QPU
0:41:44  DONE
'DONE'

wait_for_final_state() は、1、2、4、8、16、32秒後にステータスを確認し、その後は60秒ごとに確認します。そのため、短いジョブの終了は数秒以内に検知でき、長いジョブではリクエスト数が少なくて済みます。代わりに一定の間隔で確認するには、poll_interval を秒単位で渡します。

ログを読む

ワークフローが標準出力または標準エラーに書き込んだものはすべて、ジョブのログに記録されます。ジョブの実行中、結果が出る前でも、job.logs() でそれらを読み取ることができます。

# This cell is hidden from users. It waits for the job to end, so that the
# logs are complete.
job.wait_for_final_state()

run_workflow_remotely.ipynb

print(job.logs())

Computing the expectation value for theta=0.5

# This cell is hidden from users. It checks the logs.
assert "Computing the expectation value for theta=0.5" in job.logs()

• ワークフローは、約15秒ごとに新しいログ行をアップロードし、ジョブの終了時またはキャンセル時にも最後にもう一度アップロードします。ジョブの開始直後は、job.logs() は No logs yet. を返します。

• ワークフローは、ジョブごとに最新の50 MBのログを保持し、ログが切り詰められたことを示す行を先頭に追加します。

ワークフローでどのようなログを出力するかを決めるには、ログを書き込むを参照してください。正規表現に一致する行だけを表示するには、job.filtered_logs() に include を渡します。行を非表示にするには、exclude を渡します。

結果を取得する

job.result() は、ジョブが終了するまで待ってから、ワークフローが save_result() に渡した値を返します。 run_workflow_remotely.ipynb

job.result()

{'theta': 0.5, 'expectation_value': 0.8775825618903728}

# This cell is hidden from users. It checks the result.
assert _ == {"theta": 0.5, "expectation_value": 0.8775825618903728}, _  # noqa: F821

デフォルトでは、job.result() は30秒ごとにステータスを確認し、時間制限なしで待ちます。確認回数を制限するには maxwait を渡します。たとえば、job.result(maxwait=40) は約20分待ちます。待たずに1回だけ確認するには、wait=False を渡します。ジョブが終了する前に job.result() が待機をやめた場合は、{} を返します。

各ステージにかかった時間を確認する

job.events() は、ジョブのすべてのステータスとステージの変更を時刻付きで一覧表示するため、ジョブがコンピュートを待った時間と、各ステージにかかった時間を確認できます。 run_workflow_remotely.ipynb

for event in reversed(job.events()):
    print(
        event.created,
        event.data.get("status") or event.data.get("sub_status"),
    )

2026-11-02T14:43:07.134377Z QUEUED
2026-11-02T14:43:24.474199Z PENDING
2026-11-02T14:46:07.309519Z RUNNING
2026-11-02T14:46:21.115038Z SUCCEEDED

# This cell is hidden from users. It checks the events.
_statuses = [e.data.get("status") for e in job.events()]
assert "QUEUED" in _statuses and "SUCCEEDED" in _statuses, _statuses

イベントには、保存されているステータス名が使われます。PENDING は INITIALIZING を、SUCCEEDED は DONE を指します。実行の QPU 使用量については、実行の QPU ジョブを見つけるを参照してください。

エラーの原因を調べる

ワークフローが例外を発生させると、ジョブは ERROR ステータスで終了します。たとえば、ワークフローが拒否する角度を指定して実行します。 run_workflow_remotely.ipynb

failed_job = my_workflow.run(theta=-1.0)

# This cell is hidden from users. It waits for the job to end.
failed_job.wait_for_final_state()

run_workflow_remotely.ipynb

failed_job.status()

'ERROR'

ログにはトレースバックが表示されます。 run_workflow_remotely.ipynb

print(failed_job.logs())

Traceback (most recent call last):
  File "/function_user_data/my_workflow.py", line 8, in <module>
    raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

job.result() raises a QiskitServerlessException: run_workflow_remotely.ipynb

from qiskit_serverless.exception import QiskitServerlessException

try:
    failed_job.result()
except QiskitServerlessException as error:
    print(error)

raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

# This cell is hidden from users. It checks the error.
assert failed_job.status() == "ERROR", failed_job.status()
assert (
    "ValueError: theta must be between 0 and pi, got -1.0"
    in failed_job.logs()
)
assert failed_job.events(type="ERROR") == []

ワークフローが send_error() でエラーを報告する場合は、job.events(type="ERROR") でそれらを読み取り、job.error_message() でそのメッセージを読み取ります。報告されたエラーがないジョブでは、job.events(type="ERROR") は空のリストを返し、job.error_message() は {} を返します。

run_workflow_remotely.ipynb

for event in job.events(type="ERROR"):
    print(event.created, event.data.get("code"), event.data.get("message"))

ジョブを管理する

ジョブをキャンセルする

job.cancel() でジョブをキャンセルします。ワークフローがジョブを停止している間、ステータスは CANCELING に変わり、その後 CANCELED に変わります。 run_workflow_remotely.ipynb

job_to_cancel = my_workflow.run(theta=1.0)
job_to_cancel.cancel()

'Job has been stopped. No active runtime job ID associated with this serverless job ID.'

# This cell is hidden from users. It checks the cancellation.
assert (
    job_to_cancel.wait_for_final_state() == "CANCELED"
), job_to_cancel.status()

job.cancel() は、ジョブと、同時にキャンセルを試みる関連付けられた QPU ジョブに関するメッセージを返します。実行中のジョブの場合、ワークフローはエントリポイントに SIGTERM を送信します。部分的な結果を保存するには、ジョブがキャンセルされたときにクリーンアップするを参照してください。ジョブがすでに終了している場合、job.cancel() は Job already in terminal state. を返し、何も変更しません。

新しいセッションからジョブを見つける

Job オブジェクトとワークフローは、現在の Python セッション内にのみ存在します。ノートブックのカーネルを再起動した後は、保存したジョブ ID を client.job() に渡すか、client.jobs() で最近のジョブを一覧表示します。 run_workflow_remotely.ipynb

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
job = client.job(job_id)
my_workflow = client.load("my-workflow")

client.jobs(function=my_workflow, status="ERROR", limit=5)

[<Job | e6b1ffa1-cd26-437d-bd53-b265dd4ba952>,
 <Job | 11308f2a-8d7a-40a0-a970-7fdeca2b171b>,
 <Job | af597785-3470-469a-9051-607e0185892e>,
 <Job | a9051931-3bc2-4b80-9ce6-c39d28ce260c>,
 <Job | e329e8e3-e955-4188-9bfd-8f3050ee2d39>]

# This cell is hidden from users. It checks the job lookups.
assert job.result()["theta"] == 0.5
assert failed_job.job_id in [j.job_id for j in _]  # noqa: F821

client.jobs() は、最新のジョブ10件を新しい順に返します。function、status（ステータスの表の値）、または created_after（ISO 8601 のタイムスタンプ）で絞り込み、limit と offset でページ送りできます。アップロードしたワークフローを一覧表示するには、client.list() を呼び出します。

IBM Quantum Platform の Workloads テーブルにも、ジョブがステータス、コンピュートプロファイル、実行時間とともに一覧表示されます。

時間制限

マネージドワークフローは、24時間を超えて実行されるジョブをキャンセルし、そのジョブは CANCELED ステータスで終了します。長時間の実行中は、作業が失われないように、チェックポイントをファイルストレージに保存してください。

ジョブのトラブルシューティング

問題	原因
run() が Code: 429 と Active job limit reached. で失敗する	アクティブジョブの上限に達しました。いくつかのジョブが終了するのを待つか、不要なジョブをキャンセルしてください。
ジョブが QUEUED のままで、run() が your account, plan or subscription is blocked という警告を出した。	アカウント、プラン、またはサブスクリプションのブロックが解除されるまで、ジョブは開始されません。管理者に連絡してください。
job.logs() が No logs yet. を返す	ジョブがキューに入っているか、開始直後です。約15秒待ってください。
job.result() が {} を返す	ジョブが終了していないか、キャンセルされたか、結果が保存されていません。結果を保存するを参照してください。
job.error_message() が {} を返す	ワークフローがエラーを報告せずに例外を発生させました。job.logs() でトレースバックを読んでください。
ジョブが CANCELED だが、自分でキャンセルしていない	ジョブが24時間の上限に達しました。

次のステップ

推奨事項

• ログを書き込み、エラーを処理することで、ジョブが報告するログとエラーを設定します。
• パラメータースイープを実行することで、入力の組ごとにワークフローを1回ずつ実行します。