ワークフローから量子ジョブを実行する
パッケージのバージョン
このページのコードは、次の要件で開発されました。 これらのバージョン、またはそれ以降のバージョンの使用をお勧めします。
qiskit[all]~=2.6.1
qiskit-ibm-runtime~=0.51.0
qiskit-serverless~=0.38.0
量子・古典ワークフローは、1つのマネージド・ワークフロー・ジョブとして実行されます。古典的なステージはコンピュート・プロファイル上で実行され、IBM Quantum® QPUに送られるプリミティブ・ジョブはインスタンス上で実行され、使用量に計上されます。この1つのジョブから、各ステージを報告し、QPUジョブを見つけ、それらをキャンセルできます。ジョブを開始し、そのステータスとログを確認するには、マネージド・ワークフローを実行するを参照してください。
始める前に、最初のマネージド・ワークフローをデプロイしてください。
Runtimeサービスを作成する
ワークフローのコードで get_runtime_service() を使ってRuntimeサービスを作成し、ローカルで QiskitRuntimeService を使うのと同じように使用します。次のワークフローは、Sampler プリミティブでBell回路を実行し、update_status() で各ステージを報告し、カウントを返します。
# Added by doQumentation — installs the packages this notebook needs if they are missing
import importlib.util
_needed = {"qiskit": "qiskit", "qiskit_ibm_catalog": "qiskit-ibm-catalog", "qiskit_ibm_runtime": "qiskit-ibm-runtime", "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
from qiskit import QuantumCircuit
from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime.executor_sampler import Sampler
from qiskit_serverless import (
Job,
get_arguments,
get_runtime_service,
save_result,
update_status,
)
arguments = get_arguments()
service = get_runtime_service()
backend = service.backend(arguments["backend_name"])
circuit = QuantumCircuit(2)
circuit.h(0)
circuit.cx(0, 1)
circuit.measure_all()
update_status(Job.OPTIMIZING_HARDWARE)
pass_manager = generate_preset_pass_manager(optimization_level=1, backend=backend)
isa_circuit = pass_manager.run(circuit)
update_status(Job.WAITING_QPU)
sampler = Sampler(mode=backend)
primitive_job = sampler.run([isa_circuit])
counts = primitive_job.result()[0].data.meas.get_counts()
update_status(Job.POST_PROCESSING)
save_result({"backend": backend.name, "counts": counts})
Overwriting ./source_files/my_workflow.py
get_runtime_service() は、マネージド・ワークフロー・ジョブ用に設定された QiskitRuntimeService を返し、次の機能を提供します。
-
認証情報。 Runtimeサービスは、
run()の呼び出しに使用したAPIキー、インスタンス、チャネルを使用するため、ワークフローのコードに認証情報は不要です。 -
トラッキング。 ワークフローは、このRuntimeサービスを通じて実行されるすべてのプリミティブ・ジョブを、マネージド・ワークフロー・ジョブの下に記録します。これには、Sampler、Estimator、Executorのジョブや、セッションまたはバッチ内のジョブが含まれます。
job.runtime_jobs()でそれらを一覧表示できます。 -
キャンセル。
job.cancel()を呼び出すと、マネージド・ワークフローは、ワークフローのセッション、バッチ、および実行中のプリミティブ・ジョブもキャンセルします。
マネージド・ワークフローは、QiskitRuntimeService() で作成したサービスのジョブを追跡もキャンセルもしません。ただし、それらのジョブは引き続き実行されます。ワークフローで別のサービスを使う必要がある場合は、SIGTERM ハンドラーでそのジョブをキャンセルしてください。
セッションまたはバッチでプリミティブを実行する
マネージド・ワークフローは、セッションまたはバッチで実行されるプリミティブ・ジョブも追跡します。Runtimeサービスが返したバックエンドからセッションまたはバッチを作成し、with ブロックの内側でプリミティブを作成してください。Session も同じように動作します。
source_files/my_workflow.py
from qiskit_ibm_runtime import Batch
with Batch(backend=backend) as batch:
sampler = Sampler(mode=batch)
primitive_jobs = [sampler.run([isa_circuit]) for _ in range(2)]
results = [primitive_job.result() for primitive_job in primitive_jobs]
update_status() で進捗を報告する
Runtimeサービスを作成するのワークフローは、各ステージの開始時に update_status() を呼び出します。これにより、ジョブのステータスに RUNNING: WAITING_FOR_QPU のようにワークフローの現在位置が表示されます。次の Job 定数のいずれかを渡してください。
| ワークフロー内の呼び出し | ジョブのステータス |
|---|---|
update_status(Job.MAPPING) | RUNNING: MAPPING |
update_status(Job.OPTIMIZING_HARDWARE) | RUNNING: OPTIMIZING_FOR_HARDWARE |
update_status(Job.WAITING_QPU) | RUNNING: WAITING_FOR_QPU |
update_status(Job.EXECUTING_QPU) | RUNNING: EXECUTING_QPU |
update_status(Job.POST_PROCESSING) | RUNNING: POST_PROCESSING |
update_status() は、ワークフローがステータスを変更した場合に True を返します。値がこれらの定数のいずれでもない場合、ジョブがすでにアクティブでない場合、またはコードがジョブの外で実行された場合は、False を返し、ログに警告を書き込みます。エラーは発生しないため、入力ミスがあってもワークフローは停止しません。
オプティマイザーの反復など、ステージ内の進捗については、ログ行を書き込んでください。
実行のQPUジョブを見つける
ローカルのノートブックで、インスタンスがアクセスできるQPUを選びます。client.backends() はそれらを一覧表示し、client.least_busy() は保留中のジョブが最も少ないものを返します。
from qiskit_ibm_catalog import ManagedWorkflowClient
client = ManagedWorkflowClient()
print([backend.name for backend in client.backends(simulator=False)])
backend_name = client.least_busy(operational=True, simulator=False).name
['ibm_boston', 'ibm_fez', 'ibm_kingston', 'ibm_marrakesh', 'ibm_phoenix', 'ibm_pittsburgh']
ワークフローをアップロードし、実行時にバックエンドの名前を渡します。
from qiskit_ibm_catalog import ManagedWorkflow
my_workflow = client.upload(
ManagedWorkflow(
title="my-workflow",
entrypoint="my_workflow.py",
working_dir="./source_files/",
)
)
job = my_workflow.run(backend_name=backend_name)
# This cell is hidden from users. It waits for the job to finish.
assert job.wait_for_final_state() == "DONE", job.logs()
ジョブが終了したら、マネージド・ワークフロー・ジョブからワークフローのQPUジョブのIDを取得します。
runtime_job_ids = job.runtime_jobs()
runtime_job_ids
['dbl6lst8ft8s738secf0']
# This cell is hidden from users. It checks the tracked job.
assert len(runtime_job_ids) == 1, runtime_job_ids
各IDは、通常のプリミティブ・ジョブIDです。ジョブのステータスと結果を読み取るには、後でジョブ結果を取得するを参照してください。
ワークフローがセッションまたはバッチでプリミティブを実行した場合は、runtime_sessions() でセッションIDとバッチIDを一覧表示し、そのうちの1つを runtime_jobs() に渡すと、その中のジョブだけを取得できます。
run_workflow_remotely.ipynb
for session_id in job.runtime_sessions():
print(session_id, job.runtime_jobs(session_id))
dc101077-d085-40fb-8fbe-78852e355994 ['dbjqp48nceqs73aki1k0', 'dbjqp4l6p51s73clb91g']
job オブジェクトは、実行を開始したノートブック内にのみ存在します。新しいノートブックでは、client.job("<your-job-id>") でジョブを再度取得し、それに対して runtime_jobs() を呼び出します。新しいセッションからジョブを見つけるを参照してください。
各QPUジョブのQPU使用量と実行時間については、ジョブのメトリクスを表示するを参照してください。現在の期間におけるインスタンス全体の使用量については、client.usage() を呼び出します。マネージド・ワークフロー・ジョブの各ステージにかかった時間を確認するには、各ステージにかかった時間を確認するを参照してください。
関連するQPUジョブをキャンセルする
job.cancel() でマネージド・ワークフロー・ジョブをキャンセルすると、ワークフローは、自身が開始したQPUの処理もキャンセルしようとします。
-
セッションとバッチ。 ワークフローは、自身の各セッションと各バッチをキャンセルし、それによってキューに入っているジョブもキャンセルされます。
-
その他のプリミティブ・ジョブ。 ワークフローは、自身のその他のプリミティブ・ジョブのうち、キューに入っているものまたは実行中のものをそれぞれキャンセルします。
job.cancel() は、QPUジョブとセッションごとにメッセージを返すため、何がキャンセルされたかを確認できます。
run_workflow_remotely.ipynb
print(job.cancel())
バッチで実行されたワークフローの場合、メッセージは次のようになります。
Job is stopping. Canceled runtime session: eb8874c9-a15e-45f8-9a90-a9fb35267f9d and associated runtime jobs.
クライアントがローカルの認証情報から QiskitRuntimeService を作成できない場合、キャンセルされるのはマネージド・ワークフロー・ジョブのみです。job.runtime_jobs() から取得したIDを使って、QPUジョブを自分でキャンセルしてください。
ワークフローが自らジョブを停止した場合、たとえば24時間の制限に達した場合は、そのQPUジョブはキャンセルされません。job.runtime_jobs() から取得したIDを使ってキャンセルしてください。
ワークフロー内のQPUジョブのトラブルシューティング
| 問題 | 原因 |
|---|---|
get_runtime_service() が KeyError: 'QISKIT_IBM_CHANNEL' を発生させる。 | コードが、マネージド・ワークフロー・ジョブではなくローカルのノートブックで実行されています。ローカルでは QiskitRuntimeService() を使ってください。 |
update_status() が False を返し、ログに 'sub_status' not provided or is not valid と表示される。 | 値が Job 定数のいずれでもないか、ジョブがすでにアクティブではありません。 |
job.runtime_jobs() が [] を返す。 | ワークフローが QiskitRuntimeService() でサービスを作成したか、まだプリミティブを実行していません。 |
| セッションのジョブが追跡されない。 | セッションが、別のサービスのバックエンドから作成されました。バックエンドは get_runtime_service() から取得してください。 |
ワークフローのログに Could not associate runtime job with serverless job と表示される。 | ワークフローがプリミティブ・ジョブを記録できませんでした。プリミティブ・ジョブは引き続き実行されますが、追跡されません。 |
job.cancel() が No QiskitRuntimeService can be associated to the given token and instance. を表示する。 | クライアントがローカルの認証情報から QiskitRuntimeService を作成できなかったため、マネージド・ワークフロー・ジョブのみをキャンセルしました。 |
client.job() が Job [...] not found で失敗する。 | ジョブIDが間違っているか、ジョブが別のユーザーのものです。 |
次のステップ
- マネージド・ワークフローを実行すると、ジョブのステータス、結果、ログを確認し、エラーの原因を見つけられます。