Ejecutar un barrido de parámetros

Versiones de paquetes

El código de esta página se desarrolló con los siguientes requisitos. Recomendamos usar estas versiones o posteriores.

qiskit-serverless~=0.38.0

Un barrido de parámetros funciona como un arreglo de trabajos de Slurm. Llama a run() una vez por cada conjunto de entradas y los trabajos se ejecutan al mismo tiempo, cada uno en su propio nodo. Usa el mismo patrón para cualquier lote de entradas independientes, como un trabajo por cada molécula o cada backend. Los trabajos no pueden enviarse datos entre sí, así que divide el trabajo en entradas independientes.

Los ejemplos de esta página usan el flujo de trabajo de Ejecutar un flujo de trabajo gestionado, que lee un ángulo, 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
)

Iniciar los trabajos

Obtén el flujo de trabajo por su título y llama a run() una vez por cada entrada. Guarda los ID de los trabajos en un archivo: 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)

• Comprueba primero cada entrada. Si el flujo de trabajo tiene un esquema de argumentos, llama a validate_arguments() con cada entrada antes del bucle.

• Guarda los ID de los trabajos en un archivo. Así podrás recopilar los resultados desde una sesión nueva.

• Da a los archivos de cada ejecución un nombre único. Todas las ejecuciones comparten el almacenamiento de archivos del flujo de trabajo. Consulta Práctica recomendada: dar a los archivos de cada ejecución un nombre único.

• Mantente por debajo del límite de trabajos activos. Un flujo de trabajo gestionado no pone en cola los trabajos que superan el límite, así que cuando lo alcances, inicia más trabajos solo después de que terminen los anteriores.

Recopilar los resultados

Obtén cada job por su ID y captura los jobs fallidos para que un fallo no detenga el bucle: 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)

Para volver a ejecutar las entradas que fallaron, llama a run() de nuevo con los mismos argumentos.

Límite de jobs activos

Puedes tener hasta 50 jobs en cola, inicializándose, en ejecución o cancelándose al mismo tiempo. Este límite cuenta todos tus jobs, incluidos los jobs de otros workflows. Si superas este límite, run() lanza una QiskitServerlessException y no crea ni pone en cola el job:

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

Próximos pasos

Recomendaciones

• Administra archivos entre ejecuciones para compartir archivos de entrada entre ejecuciones y dar un nombre único a los archivos de cada ejecución.
• Elige un perfil de cómputo para ejecutar cada job del barrido con un tamaño que se ajuste a su entrada.