Rulează o baleiere de parametri

Versiuni de pachete

Codul de pe această pagină a fost dezvoltat folosind următoarele cerințe. Recomandăm să folosești aceste versiuni sau unele mai noi.

qiskit-serverless~=0.38.0

O baleiere de parametri funcționează ca un array de joburi Slurm. Apelează run() o dată pentru fiecare set de intrări, iar joburile rulează în același timp, fiecare pe propriul nod. Folosește același tipar pentru orice lot de intrări independente, cum ar fi un job pentru fiecare moleculă sau fiecare backend. Joburile nu pot trimite date între ele, așa că împarte munca în intrări independente.

Exemplele de pe această pagină folosesc fluxul de lucru din Rulează un flux de lucru gestionat, care citește un unghi, 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
)

Pornește joburile

Obține fluxul de lucru după titlul său și apelează run() o dată pentru fiecare intrare. Salvează ID-urile joburilor într-un fișier: 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)

• Verifică mai întâi fiecare intrare. Dacă fluxul de lucru are o schemă de argumente, apelează validate_arguments() pentru fiecare intrare înainte de buclă.

• Salvează ID-urile joburilor într-un fișier. Poți apoi colecta rezultatele dintr-o sesiune nouă.

• Dă fișierelor fiecărei rulări un nume unic. Toate rulările partajează stocarea de fișiere a fluxului de lucru. Consultă Bună practică: dă fișierelor fiecărei rulări un nume unic.

• Rămâi sub limita de joburi active. Un flux de lucru gestionat nu pune în coadă joburile peste limită, așa că, atunci când o atingi, pornește alte joburi doar după ce se încheie cele anterioare.

Colectează rezultatele

Obține fiecare job după ID-ul său și prinde job-urile eșuate, astfel încât o singură eroare să nu oprească bucla: 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)

Pentru a relua intrările care au eșuat, apelează din nou run() cu aceleași argumente.

Limita de job-uri active

Poți avea cel mult 50 de job-uri care sunt în coadă, în curs de inițializare, în execuție sau în curs de anulare în același timp. Această limită numără toate job-urile tale, inclusiv job-urile altor workflow-uri. Dacă depășești această limită, run() ridică o QiskitServerlessException și nu creează și nu pune în coadă job-ul:

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

Pașii următori

Recomandări

• Gestionează fișierele între rulări pentru a partaja fișierele de intrare între rulări și a da fișierelor fiecărei rulări un nume unic.
• Alege un profil de calcul pentru a rula fiecare job al sweep-ului la o dimensiune potrivită cu intrarea sa.