Jalankan parameter sweep

Versi paket

Kode di halaman ini dikembangkan dengan persyaratan berikut. Kami menyarankan memakai versi ini atau yang lebih baru.

qiskit-serverless~=0.38.0

Parameter sweep bekerja seperti job array Slurm. Panggil run() sekali untuk setiap set input, dan job-job itu berjalan bersamaan, masing-masing di node-nya sendiri. Pakai pola yang sama untuk kumpulan input independen apa pun, misalnya satu job untuk setiap molekul atau setiap Backend. Job tidak bisa saling mengirim data, jadi bagi pekerjaan menjadi input yang independen.

Contoh di halaman ini memakai workflow dari Jalankan managed workflow, yang membaca satu sudut, 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
)

Mulai job

Ambil workflow berdasarkan judulnya, dan panggil run() sekali untuk setiap input. Simpan ID job ke file: 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)

• Periksa setiap input lebih dulu. Kalau workflow punya argument schema, panggil validate_arguments() pada setiap input sebelum loop.

• Simpan ID job ke file. Dengan begitu kamu bisa mengumpulkan hasilnya dari sesi baru.

• Beri nama unik untuk file setiap run. Semua run berbagi file storage milik workflow. Lihat Praktik terbaik: beri nama unik untuk file setiap run.

• Tetap di bawah batas job aktif. Managed workflow tidak mengantrekan job yang melebihi batas, jadi saat kamu mencapainya, mulai job lagi hanya setelah job sebelumnya selesai.

Kumpulkan hasil

Ambil setiap job berdasarkan ID-nya, dan tangkap job yang gagal supaya satu kegagalan tidak menghentikan loop: 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)

Untuk menjalankan ulang input yang gagal, panggil run() lagi dengan argumen yang sama.

Batas job aktif

Kamu bisa punya hingga 50 job yang sedang antre, diinisialisasi, berjalan, atau dibatalkan pada waktu yang sama. Batas ini menghitung semua job kamu, termasuk job dari workflow lain. Kalau melebihi batas ini, run() memunculkan QiskitServerlessException dan tidak membuat atau mengantrekan job:

| Message: Http bad request.
| Code: 429
| Details: Active job limit reached. The maximum allowed is 50.

Langkah berikutnya

Rekomendasi

• Kelola file lintas run untuk berbagi file input antar run dan memberi nama unik pada file setiap run.
• Pilih compute profile untuk menjalankan setiap job dalam sweep pada ukuran yang sesuai dengan inputnya.