パラメータースイープを実行する
パッケージのバージョン
このページのコードは、次の要件を使用して開発されました。 これらのバージョン、またはそれ以降のバージョンの使用をお勧めします。
qiskit-serverless~=0.38.0
パラメータースイープは、Slurm のジョブ配列のように動作します。入力の組ごとに run() を1回呼び出すと、ジョブは同時に実行され、それぞれが独自のノードで動作します。分子ごと、バックエンドごとのジョブなど、独立した入力のバッチであれば、同じパターンを使えます。ジョブ同士でデータを送り合うことはできないため、作業を独立した入力に分割してください。
このページの例では、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
# 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.
import logging
from qiskit_ibm_catalog import ManagedWorkflow, ManagedWorkflowClient
logging.getLogger("qiskit_ibm_runtime").setLevel(logging.ERROR)
_test_title = "docs-run-parameter-sweep"
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() を1回呼び出します。ジョブ ID をファイルに保存します。
run_workflow_remotely.ipynb
import json
from qiskit_ibm_catalog import ManagedWorkflowClient
client = ManagedWorkflowClient()
my_workflow = client.load("my-workflow")
thetas = [0.1, 0.2, 0.3]
job_ids = {
f"theta-{theta}": my_workflow.run(theta=theta).job_id for theta in thetas
}
with open("job_ids.json", "w") as f:
json.dump(job_ids, f)
-
最初にすべての入力を確認する。 ワークフローに引数スキーマがある場合は、ループの前に各入力に対して
validate_arguments()を呼び出します。 -
ジョブ ID をファイルに保存する。 そうすれば、新しいセッションから結果を収集できます。
-
各実行のファイルに一意の名前を付ける。 すべての実行は、ワークフローのファイルストレージを共有します。ベストプラクティス: 各実行のファイルに一意の名前を付けるを参照してください。
-
アクティブジョブの上限を超えないようにする。 マネージドワークフローは、上限を超えたジョブをキューに入れないため、上限に達したときは、先に開始したジョブが終了してから、さらにジョブを開始してください。
結果を収集する
ID で各ジョブを取得し、1つの失敗でループが止まらないように、失敗したジョブをキャッチします。
run_workflow_remotely.ipynb
from qiskit_serverless.exception import QiskitServerlessException
with open("job_ids.json") as f:
job_ids = json.load(f)
results, errors = {}, {}
for run_name, sweep_job_id in job_ids.items():
try:
results[run_name] = client.job(sweep_job_id).result()
except QiskitServerlessException as error:
errors[run_name] = str(error)
print(f"{len(results)} jobs finished, {len(errors)} failed")
3 jobs finished, 0 failed
# This cell is hidden from users. It checks the sweep results.
assert len(results) == 3 and not errors, (results, errors)
assert all(results[f"theta-{t}"]["theta"] == t for t in thetas)
失敗した入力を再実行するには、同じ引数で run() をもう一度呼び出します。
アクティブジョブの上限
キュー待ち、初期化中、実行中、またはキャンセル中のジョブは、同時に最大50件まで持つことができます。この上限は、他のワークフローのジョブを含め、すべてのジョブを対象に数えられます。この上限を超えると、run() は QiskitServerlessException を発生させ、ジョブを作成もキューへの追加もしません。
| Message: Http bad request.
| Code: 429
| Details: Active job limit reached. The maximum allowed is 50.
次のステップ
- 実行間でファイルを管理することで、実行間で入力ファイルを共有し、各実行のファイルに一意の名前を付けます。
- コンピュートプロファイルを選択することで、スイープの各ジョブを、その入力に合ったサイズで実行します。