Exécuter un balayage de paramètres

Versions des paquets

Le code de cette page a été développé avec les prérequis suivants. Nous recommandons d'utiliser ces versions ou des versions plus récentes.

qiskit-serverless~=0.38.0

Un balayage de paramètres fonctionne comme un tableau de jobs Slurm. Appelle run() une fois pour chaque jeu d'entrées, et les jobs s'exécutent en même temps, chacun sur son propre nœud. Utilise le même schéma pour tout lot d'entrées indépendantes, comme un job par molécule ou par backend. Les jobs ne peuvent pas s'envoyer de données entre eux, donc découpe le travail en entrées indépendantes.

Les exemples de cette page utilisent le workflow de Exécuter un workflow géré, qui lit un angle, 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
)

Démarrer les jobs

Récupère le workflow par son titre et appelle run() une fois pour chaque entrée. Enregistre les ID des jobs dans un fichier : 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)

• Vérifie d'abord chaque entrée. Si le workflow possède un schéma d'arguments, appelle validate_arguments() sur chaque entrée avant la boucle.

• Enregistre les ID de jobs dans un fichier. Tu pourras ensuite récupérer les résultats depuis une nouvelle session.

• Donne un nom unique aux fichiers de chaque exécution. Toutes les exécutions partagent le stockage de fichiers du workflow. Voir Bonne pratique : donner un nom unique aux fichiers de chaque exécution.

• Reste sous la limite de jobs actifs. Un workflow géré ne met pas en file d'attente les jobs au-delà de la limite ; quand tu l'atteins, ne lance de nouveaux jobs qu'après la fin de jobs précédents.

Récupérer les résultats

Récupère chaque job par son ID et intercepte les jobs en échec afin qu'un échec n'arrête pas la boucle : 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)

Pour relancer les entrées qui ont échoué, appelle à nouveau run() avec les mêmes arguments.

Limite de jobs actifs

Tu peux avoir jusqu'à 50 jobs en file d'attente, en cours d'initialisation, en cours d'exécution ou en cours d'annulation en même temps. Cette limite compte tous tes jobs, y compris ceux d'autres workflows. Si tu dépasses cette limite, run() lève une QiskitServerlessException et ne crée ni ne met en file d'attente le job :

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

Étapes suivantes

Recommandations

• Gérer les fichiers entre les exécutions pour partager des fichiers d'entrée entre les exécutions et donner un nom unique aux fichiers de chaque exécution.
• Choisir un profil de calcul pour exécuter chaque job du balayage sur une taille adaptée à son entrée.