Ejecución del código local como un trabajo híbrido

Los trabajos híbridos de Amazon Braket proporcionan una orquestación totalmente administrada de algoritmos híbridos cuánticos-clásicos, combinando los recursos de computación de Amazon EC2 con el acceso a la unidad de procesamiento cuántico (QPU) de Amazon Braket. Las tareas cuánticas creadas en un trabajo híbrido tienen prioridad en la cola sobre las tareas cuánticas individuales, de modo que sus algoritmos no se verán interrumpidos por las fluctuaciones en la cola de tareas cuánticas. Cada QPU mantiene una cola de trabajos híbridos independiente, lo que garantiza que solo se pueda ejecutar un trabajo híbrido en un momento dado.

Creación de un trabajo híbrido a partir de código de Python local

Puede ejecutar su código de Python local como un trabajo híbrido de Amazon Braket. Puede hacerlo anotando su código con un decorador @hybrid_job, como se muestra en el siguiente ejemplo de código. Para los entornos personalizados, puede optar por utilizar un contenedor personalizado de Amazon Elastic Container Registry (ECR).

Puede usar el decorador @hybrid_job para anotar una función. Braket transforma el código dentro del decorador en un script de algoritmo de trabajo híbrido de Braket. A continuación, el trabajo híbrido invoca la función dentro del decorador en una instancia de Amazon EC2. Puede supervisar el progreso del trabajo con job.state() o con la consola de Braket. El siguiente ejemplo de código muestra cómo ejecutar una secuencia de cinco estados en el State Vector Simulator (SV1) device.

from braket.aws import AwsDevice
from braket.circuits import Circuit, FreeParameter, Observable
from braket.devices import Devices
from braket.jobs.hybrid_job import hybrid_job
from braket.jobs.metrics import log_metric

device_arn = Devices.Amazon.SV1


@hybrid_job(device=device_arn)  # Choose priority device
def run_hybrid_job(num_tasks=1):
    device = AwsDevice(device_arn)  # Declare AwsDevice within the hybrid job

    # Create a parametric circuit
    circ = Circuit()
    circ.rx(0, FreeParameter("theta"))
    circ.cnot(0, 1)
    circ.expectation(observable=Observable.X(), target=0)

    theta = 0.0  # Initial parameter

    for i in range(num_tasks):
        task = device.run(circ, shots=100, inputs={"theta": theta})  # Input parameters
        exp_val = task.result().values[0]

        theta += exp_val  # Modify the parameter (possibly gradient descent)

        log_metric(metric_name="exp_val", value=exp_val, iteration_number=i)

    return {"final_theta": theta, "final_exp_val": exp_val}

El trabajo híbrido se crea invocando la función como lo haría con las funciones normales de Python. Sin embargo, la función de decorador devuelve el identificador del trabajo híbrido en lugar del resultado de la función. Para recuperar los resultados una vez que se haya completado, utilice job.result().

job = run_hybrid_job(num_tasks=1)
result = job.result() 

El argumento del dispositivo en el decorador @hybrid_job especifica el dispositivo al que el trabajo híbrido tiene acceso prioritario, en este caso, el simulador SV1. Para obtener la prioridad de la QPU, debe asegurarse de que el ARN del dispositivo utilizado en la función coincida con el especificado en el decorador. Para mayor comodidad, puede utilizar la función auxiliar get_job_device_arn() para capturar el ARN del dispositivo declarado en @hybrid_job.

Hiperparámetros

La función run_hybrid_job() utiliza el argumento num_tasks para controlar el número de tareas cuánticas creadas. El trabajo híbrido captura esto automáticamente como un hiperparámetro.

Métricas y registro

En la función run_hybrid_job(), las métricas de los algoritmos iterativos se registran con log_metrics. Las métricas se representan automáticamente en la página de la consola de Braket, en la pestaña de trabajos híbridos. Puede utilizar las métricas para realizar un seguimiento de los costos de las tareas cuánticas prácticamente en tiempo real durante la ejecución de un trabajo híbrido con el Rastreador de costos de Braket. En el ejemplo anterior, se utiliza el nombre de métrica «probabilidad», que registra la primera probabilidad del tipo de resultado.

Recuperación de resultados

Una vez finalizado el trabajo híbrido, se utiliza job.result() para recuperar los resultados del trabajo híbrido. Braket captura automáticamente todos los objetos de la declaración de devolución. Tenga en cuenta que los objetos devueltos por la función deben ser una tupla y cada elemento debe ser serializable. Por ejemplo, el código siguiente muestra un ejemplo de funcionamiento correcto y otro incorrecto.

import numpy as np


# Working example
@hybrid_job(device=Devices.Amazon.SV1)
def passing():
    np_array = np.random.rand(5)
    return np_array  # Serializable

# # Failing example
# @hybrid_job(device=Devices.Amazon.SV1)
# def failing():
#     return MyObject() # Not serializable

Nombre del trabajo

De forma predeterminada, el nombre de este trabajo híbrido se deduce del nombre de la función. Puede especificar un nombre personalizado que tenga un máximo de 50 caracteres. Por ejemplo, en el código siguiente, el nombre del trabajo es «my-job-name».

@hybrid_job(device=Devices.Amazon.SV1, job_name="my-job-name")
def function():
    pass

Modo local

Los trabajos locales se crean añadiendo el argumento local=True al decorador. Esto ejecuta el trabajo híbrido en un entorno en contenedor en su entorno de computación local, como un portátil. Los trabajos locales no tienen prioridad en las colas para las tareas cuánticas. En casos avanzados, como los de varios nodos o MPI, los trabajos locales pueden tener acceso a las variables de entorno de Braket requeridas. El siguiente código crea un trabajo híbrido local con el dispositivo como el simulador SV1.

@hybrid_job(device=Devices.Amazon.SV1, local=True)
def run_hybrid_job(num_tasks=1):
    return ...

Todas las demás opciones de trabajo híbrido son compatibles. Para ver una lista de opciones, consulte el módulo braket.jobs.quantum_job_creation.

Instalación de paquetes y código fuente de Python adicionales

Puede personalizar su entorno de tiempo de ejecución para usar sus paquetes de Python preferidos. Puede usar un archivo requirements.txt, una lista de nombres de paquetes o utilizar su propio contenedor (BYOC). Por ejemplo, el archivo requirements.txt puede incluir otros paquetes para instalar.

qiskit 
pennylane >= 0.31
mitiq == 0.29

Para personalizar un entorno de tiempo de ejecución utilizando un archivo requirements.txt, consulte el siguiente ejemplo de código.

@hybrid_job(device=Devices.Amazon.SV1, dependencies="requirements.txt")
def run_hybrid_job(num_tasks=1):
    return ...

Como alternativa, puede proporcionar los nombres de los paquetes como una lista de Python de la siguiente manera.

@hybrid_job(device=Devices.Amazon.SV1, dependencies=["qiskit", "pennylane>=0.31", "mitiq==0.29"])
def run_hybrid_job(num_tasks=1):
    return ...

El código fuente adicional se puede especificar como una lista de módulos o como un solo módulo, como en el siguiente ejemplo de código.

@hybrid_job(device=Devices.Amazon.SV1, include_modules=["my_module1", "my_module2"])
def run_hybrid_job(num_tasks=1):
    return ...

Guardar y cargar datos en una instancia de trabajo híbrido

Especificación de los datos de entrenamiento de entrada

Al crear un trabajo híbrido, puede proporcionar conjuntos de datos de entrenamiento de entrada especificando un bucket de Amazon Simple Storage Service (Amazon S3). También puede especificar una ruta local y, a continuación, Braket carga automáticamente los datos en Amazon S3 en s3://<default_bucket_name>/jobs/<job_name>/<timestamp>/data/<channel_name>. Si especifica una ruta local, el nombre del canal predeterminado será «entrada». El siguiente código muestra un archivo numpy de la ruta local data/file.npy.

import numpy as np


@hybrid_job(device=Devices.Amazon.SV1, input_data="data/file.npy")
def run_hybrid_job(num_tasks=1):
    data = np.load("data/file.npy")
    return ...

Para S3, debe usar la función auxiliar get_input_data_dir().

import numpy as np
from braket.jobs import get_input_data_dir

s3_path = "s3://amazon-braket-us-east-1-123456789012/job-data/file.npy"


@hybrid_job(device=None, input_data=s3_path)
def job_s3_input():
    np.load(get_input_data_dir() + "/file.npy")


@hybrid_job(device=None, input_data={"channel": s3_path})
def job_s3_input_channel():
    np.load(get_input_data_dir("channel") + "/file.npy")

Puede especificar varios orígenes de datos de entrada proporcionando un diccionario de valores de canal y URI de S3 o rutas locales.

import numpy as np
from braket.jobs import get_input_data_dir

input_data = {
    "input": "data/file.npy",
    "input_2": "s3://amzn-s3-demo-bucket/data.json"
}


@hybrid_job(device=None, input_data=input_data)
def multiple_input_job():
    np.load(get_input_data_dir("input") + "/file.npy")
    np.load(get_input_data_dir("input_2") + "/data.json")

Guardar los resultados en S3

Para guardar los resultados no incluidos en la declaración de devolución de la función decorada, debe añadir el directorio correcto a todas las operaciones de escritura de archivos. En el siguiente ejemplo, se muestra cómo guardar una matriz numpy y una figura de matplotlib.

import matplotlib.pyplot as plt
import numpy as np


@hybrid_job(device=Devices.Amazon.SV1)
def run_hybrid_job(num_tasks=1):
    result = np.random.rand(5)

    # Save a numpy array
    np.save("result.npy", result)

    # Save a matplotlib figure
    plt.plot(result)
    plt.savefig("fig.png")
    return ...

Todos los resultados se comprimen en un archivo denominado model.tar.gz. Puede descargar los resultados con la función job.result() de Python o yendo a la carpeta de resultados desde la página de trabajos híbridos de la consola de administración de Braket.

Guardar y reanudar desde los puntos de control

Para trabajos híbridos de larga duración, se recomienda guardar periódicamente el estado intermedio del algoritmo. Puede utilizar la función auxiliar save_job_checkpoint() integrada o guardar los archivos en la ruta AMZN_BRAKET_JOB_RESULTS_DIR. Esta última opción está disponible con la función auxiliar get_job_results_dir().

El siguiente es un ejemplo práctico mínimo para guardar y cargar puntos de control con un decorador de trabajo híbrido:

from braket.jobs import save_job_checkpoint, load_job_checkpoint, hybrid_job


@hybrid_job(device=None, wait_until_complete=True)
def function():
    save_job_checkpoint({"a": 1})


job = function()
job_name = job.name
job_arn = job.arn


@hybrid_job(device=None, wait_until_complete=True, copy_checkpoints_from_job=job_arn)
def continued_function():
    load_job_checkpoint(job_name)


continued_job = continued_function()

En el primer trabajo híbrido, save_job_checkpoint() se llama con un diccionario que contiene los datos que queremos guardar. De forma predeterminada, todos los valores deben poder serializarse como texto. Para comprobar objetos de Python más complejos, como matrices numpy, puede configurar data_format = PersistedJobDataFormat.PICKLED_V4. Este código crea y sobrescribe un archivo de puntos de control con el nombre predeterminado <jobname>.json en sus artefactos de trabajo híbrido, en una subcarpeta denominada «puntos de control».

Para crear un nuevo trabajo híbrido para continuar desde el punto de control, debemos pasar copy_checkpoints_from_job=job_arn dónde job_arn es el ARN del trabajo híbrido del trabajo anterior. A continuación, usamos load_job_checkpoint(job_name) para realizar la carga desde el punto de control.

Prácticas recomendadas para decoradores de trabajos híbridos

Uso de la asincronicidad

Los trabajos híbridos creados con la anotación de decorador son asíncronos: se ejecutan una vez que los recursos clásicos y cuánticos están disponibles. El progreso del algoritmo se supervisa mediante la Braket Management Console o Amazon CloudWatch. Cuando envía el algoritmo para su ejecución, Braket lo ejecuta en un entorno en contenedor escalable y los resultados se recuperan cuando se completa el algoritmo.

Ejecución de algoritmos variacionales iterativos

Los trabajos híbridos le proporcionan las herramientas necesarias para ejecutar algoritmos cuánticos-clásicos iterativos. Para problemas puramente cuánticos, utilice tareas cuánticas o un lote de tareas cuánticas. El acceso prioritario a determinadas QPU es más beneficioso para los algoritmos variacionales de larga duración que requieren múltiples llamadas iterativas a las QPU con el procesamiento clásico intermedio.

Depuración utilizando el modo local

Antes de ejecutar un trabajo híbrido en una QPU, se recomienda ejecutar primero el simulador SV1 para confirmar que se ejecuta según lo esperado. En el caso de pruebas a pequeña escala, puede ejecutarlas con el modo local para una iteración y depuración rápidas.

Mejora de la reproducibilidad utilizando su propio contenedor (BYOC)

Cree un experimento reproducible encapsulando su software y sus dependencias en un entorno en contenedor. Al empaquetar todo su código, dependencias y configuración en un contenedor, evitará posibles conflictos y problemas de control de versiones.

Simuladores distribuidos multiinstancia

Para ejecutar una gran cantidad de circuitos, considere la posibilidad de utilizar el soporte de MPI integrado para ejecutar simuladores locales en varias instancias en un solo trabajo híbrido. Para obtener más información, consulte simuladores integrados.

Uso de circuitos paramétricos

Los circuitos paramétricos que envíe a partir de un trabajo híbrido se compilan automáticamente en determinadas QPU mediante la compilación paramétrica para mejorar los tiempos de ejecución de los algoritmos.

Punto de control periódico

Para trabajos híbridos de larga duración, se recomienda guardar periódicamente el estado intermedio del algoritmo.

Para ver más ejemplos, casos de uso y mejores prácticas, consulte el GitHub de ejemplos de Amazon Braket.