Eseguire uno sweep di parametri

Versioni dei pacchetti

Il codice in questa pagina è stato sviluppato usando i seguenti requisiti. Ti consigliamo di usare queste versioni o successive.

qiskit-serverless~=0.38.0

Uno sweep di parametri funziona come un job array di Slurm. Chiama run() una volta per ogni insieme di input e i job vengono eseguiti contemporaneamente, ciascuno sul proprio nodo. Usa lo stesso schema per qualsiasi gruppo di input indipendenti, come un job per ogni molecola o per ogni backend. I job non possono scambiarsi dati, quindi suddividi il lavoro in input indipendenti.

Gli esempi in questa pagina usano il workflow di Eseguire un workflow gestito, che legge un angolo, 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
)

Avviare i job

Ottieni il workflow tramite il suo titolo e chiama run() una volta per ogni input. Salva gli ID dei job in un 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)

• Controlla prima ogni input. Se il workflow ha uno schema degli argomenti, chiama validate_arguments() su ogni input prima del ciclo.

• Salva gli ID dei job in un file. Potrai poi raccogliere i risultati da una nuova sessione.

• Assegna un nome univoco ai file di ogni esecuzione. Tutte le esecuzioni condividono l'archiviazione file del workflow. Consulta Best practice: assegna un nome univoco ai file di ogni esecuzione.

• Resta entro il limite di job attivi. Un workflow gestito non mette in coda i job oltre il limite, quindi quando lo raggiungi, avvia altri job solo dopo che quelli precedenti sono terminati.

Raccogliere i risultati

Ottieni ogni job tramite il suo ID e intercetta i job falliti in modo che un errore non interrompa il ciclo: 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)

Per rieseguire gli input che sono falliti, chiama di nuovo run() con gli stessi argomenti.

Limite di job attivi

Puoi avere fino a 50 job in coda, in inizializzazione, in esecuzione o in annullamento contemporaneamente. Questo limite conta tutti i tuoi job, inclusi quelli di altri workflow. Se superi questo limite, run() solleva una QiskitServerlessException e non crea né mette in coda il job:

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

Passaggi successivi

Raccomandazioni

• Gestire i file tra le esecuzioni per condividere i file di input tra le esecuzioni e assegnare un nome univoco ai file di ciascuna.
• Scegliere un profilo di calcolo per eseguire ogni job dello sweep con una dimensione adatta al suo input.