Spuštění parametrického průchodu

Verze balíčků

Kód na této stránce byl vyvinut s použitím následujících požadavků. Doporučujeme používat tyto nebo novější verze.

qiskit-serverless~=0.38.0

Parametrický průchod funguje jako pole úloh Slurm. Zavolej run() jednou pro každou sadu vstupů a úlohy poběží současně, každá na vlastním uzlu. Stejný postup použij pro jakoukoli dávku nezávislých vstupů, například jednu úlohu pro každou molekulu nebo každý backend. Úlohy si nemohou posílat data, proto rozděl práci na nezávislé vstupy.

Příklady na této stránce používají pracovní postup z Spuštění spravovaného pracovního postupu, který načítá jeden úhel 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
)

Spuštění úloh

Získej workflow podle jeho názvu a pro každý vstup zavolej run() jednou. Ulož ID úloh do souboru: 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)

• Nejprve zkontroluj každý vstup. Pokud má workflow schéma argumentů, zavolej před cyklem pro každý vstup validate_arguments().

• Ulož ID úloh do souboru. Výsledky pak můžeš získat z nové relace.

• Dej souborům každého běhu jedinečný název. Všechny běhy sdílejí úložiště souborů workflow. Viz Doporučený postup: dej souborům každého běhu jedinečný název.

• Zůstaň pod limitem aktivních úloh. Spravovaný workflow neřadí úlohy nad limit do fronty, takže když limitu dosáhneš, spouštěj další úlohy až po skončení předchozích.

Získání výsledků

Získej každou úlohu podle jejího ID a zachytávej neúspěšné úlohy, aby jedno selhání nezastavilo cyklus: 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)

Chceš-li znovu spustit vstupy, které selhaly, zavolej run() znovu se stejnými argumenty.

Limit aktivních úloh

Současně můžeš mít až 50 úloh, které jsou ve frontě, inicializují se, běží nebo se ruší. Tento limit zahrnuje všechny tvoje úlohy, včetně úloh jiných workflow. Pokud limit překročíš, run() vyvolá QiskitServerlessException a úlohu nevytvoří ani nezařadí do fronty:

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

Další kroky

Doporučení

• Správa souborů napříč běhy ti umožní sdílet vstupní soubory mezi běhy a dát souborům každého běhu jedinečný název.
• Výběr výpočetního profilu ti umožní spouštět každou úlohu sweepu na velikosti, která odpovídá jejímu vstupu.