Einen Parameter-Sweep ausführen

Paketversionen

Der Code auf dieser Seite wurde mit den folgenden Anforderungen entwickelt. Wir empfehlen, diese Versionen oder neuere zu verwenden.

qiskit-serverless~=0.38.0

Ein Parameter-Sweep funktioniert wie ein Slurm-Job-Array. Rufe run() einmal für jeden Satz von Eingaben auf, und die Jobs laufen gleichzeitig, jeder auf einem eigenen Knoten. Verwende dasselbe Muster für jede Menge unabhängiger Eingaben, etwa einen Job für jedes Molekül oder jedes Backend. Jobs können keine Daten untereinander austauschen, teile die Arbeit also in unabhängige Eingaben auf.

Die Beispiele auf dieser Seite verwenden den Workflow aus Einen verwalteten Workflow ausführen, der einen Winkel, theta, liest:

# 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
)

Die Jobs starten

Rufe den Workflow über seinen Titel ab und rufe run() einmal für jede Eingabe auf. Speichere die Job-IDs in einer Datei: 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)

• Prüfe zuerst jede Eingabe. Wenn der Workflow ein Argumentschema hat, rufe vor der Schleife validate_arguments() für jede Eingabe auf.

• Speichere die Job-IDs in einer Datei. So kannst du die Ergebnisse später aus einer neuen Session sammeln.

• Gib den Dateien jedes Laufs einen eindeutigen Namen. Alle Läufe teilen sich den Dateispeicher des Workflows. Siehe Best Practice: Den Dateien jedes Laufs einen eindeutigen Namen geben.

• Bleibe unter dem Limit für aktive Jobs. Ein verwalteter Workflow stellt Jobs über dem Limit nicht in die Warteschlange. Wenn du es erreichst, starte daher weitere Jobs erst, nachdem frühere Jobs beendet sind.

Die Ergebnisse sammeln

Rufe jeden Job über seine ID ab und fange fehlgeschlagene Jobs ab, damit ein Fehler die Schleife nicht stoppt: 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)

Um die fehlgeschlagenen Eingaben erneut auszuführen, rufe run() erneut mit denselben Argumenten auf.

Limit für aktive Jobs

Du kannst bis zu 50 Jobs gleichzeitig haben, die in der Warteschlange stehen, initialisiert werden, laufen oder abgebrochen werden. Dieses Limit zählt alle deine Jobs, einschließlich der Jobs anderer Workflows. Wenn du dieses Limit überschreitest, löst run() eine QiskitServerlessException aus und erstellt den Job nicht und stellt ihn nicht in die Warteschlange:

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

Nächste Schritte

Empfehlungen

• Dateien über mehrere Ausführungen hinweg verwalten, um Eingabedateien zwischen Ausführungen zu teilen und den Dateien jeder Ausführung einen eindeutigen Namen zu geben.
• Ein Compute-Profil auswählen, um jeden Job des Sweeps in einer Größe auszuführen, die zu seiner Eingabe passt.