使用本地 Python 代码创建混合作业安装其他 Python 软件包和源代码将数据保存并加载到混合作业实例中混合作业修饰器最佳实践

将本地代码作为混合作业运行

Amazon Braket Hybrid Jobs 提供了混合量子经典算法的完全托管编排，将 Amazon EC2 计算资源与 Amazon Braket 量子处理单元（QPU）访问权限相结合。在混合作业中创建的量子任务优先于单个量子任务，因此您的算法不会因量子任务队列的波动而中断。每个 QPU 都维护一个单独的混合作业队列，确保在任何给定时间只能运行一个混合作业。

本节内容：

使用本地 Python 代码创建混合作业
安装其他 Python 软件包和源代码
将数据保存并加载到混合作业实例中
混合作业修饰器最佳实践

使用本地 Python 代码创建混合作业

您可以将本地 Python 代码作为 Amazon Braket Hybrid Jobs 运行。您可以使用 @hybrid_job 装饰器为代码添加注释来做到这一点，如以下代码示例中所示。对于自定义环境，您可以选择使用 Amazon Elastic Registry（ECR）中的自定义容器。

注意

默认情况下，仅支持 Python 3.10。

您可以使用 @hybrid_job 装饰器为函数添加注释。Braket 将装饰器内部的代码转换为 Braket 混合作业算法脚本。然后，混合作业在 Amazon EC2 实例上调用装饰器内部的函数。您可以使用 job.state() 或 Braket 控制台监控作业进度。以下代码示例显示如何在 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}

您可以像普通 Python 函数一样通过调用函数来创建混合作业。但是，装饰器函数返回的是混合任务句柄而不是函数的结果。要在结果完成后检索结果，请使用 job.result()。


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

@hybrid_job 装饰器中的 device 参数指定混合作业可以优先访问的设备，在本例中为 SV1 模拟器。要获得 QPU 优先级，必须确保函数中使用的设备 ARN 与装饰器中指定的设备 ARN 相匹配。为方便起见，您可以使用辅助函数 get_job_device_arn() 来捕获 @hybrid_job 中声明的设备 ARN。

注意

自从在 Amazon EC2 上创建容器化环境以来，每个混合任务至少有一分钟的启动时间。因此，对于非常短的工作负载，例如单个电路或一批电路，使用量子任务可能就足够了。

超参数

该 run_hybrid_job() 函数采用参数 num_tasks 来控制创建的量子任务的数量。混合作业会自动将其捕获为超参数。

注意

超参数在 Braket 控制台上显示为字符串，限制为 2500 个字符。

指标和日志记录

在 run_hybrid_job() 函数中，使用 log_metrics 记录来自迭代算法的指标。相关指标会自动绘制在 Braket 控制台页面的“混合作业”选项卡下。在混合作业运行期间，您可以使用 Braket 成本跟踪器近乎实时地跟踪量子任务成本。上面的示例使用指标名称“概率”来记录结果类型的第一个概率。

检索结果

混合作业完成后，您可以使用 job.result() 检索混合作业的结果。Braket 会自动捕获返回语句中的任何对象。请注意，该函数返回的对象必须是一个元组，每个元素都是可序列化的。例如，以下代码显示了一个正在工作和一个失败的示例。


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

作业名称

默认情况下，此混合作业的名称是根据函数名称推断出来的。您还可以指定长度最多为 50 个字符的名称。例如，在以下代码中，作业名称为“my-job-name”。


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

本地模式

通过向装饰器添加参数 local=True 来创建本地作业。这样可以在本地计算环境（如笔记本）的容器化环境中运行混合作业。本地作业没有优先排队等候量子任务。对于多节点或 MPI 等高级案例，本地作业可以访问所需的 Braket 环境变量。以下代码使用设备作为 SV1 模拟器创建本地混合作业。


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

支持所有其他混合作业选项。有关选项列表，请参阅 braket.jobs.quantum_job_creation 模块。

安装其他 Python 软件包和源代码

您可以自定义运行时环境以使用首选 Python 包。您可以使用 requirements.txt 文件、软件包名称列表或自带容器（BYOC）。例如，requirements.txt 文件可能包含其他要安装的软件包。


qiskit 
pennylane >= 0.31
mitiq == 0.29

要使用 requirements.txt 文件自定义运行时系统环境，请参阅以下代码示例。


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

您还可以按如下方式以 Python 列表的形式提供软件包名称。


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

其他源代码可以指定为模块列表，也可以指定为单个模块，如以下代码示例所示。


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

将数据保存并加载到混合作业实例中

指定输入训练数据

创建混合作业时，您可以通过指定 Amazon Simple Storage Service (Amazon S3) 存储桶，提供输入训练数据集。您也可以指定本地路径。然后，Braket 会自动将数据上传到 Amazon S3，URL 为 s3://<default_bucket_name>/jobs/<job_name>/<timestamp>/data/<channel_name>。如果您指定本地路径，则频道名称默认为“input”。以下代码显示了来自本地路径 data/file.npy 的 numpy 文件。


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 ...

对于 S3，必须使用 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")

您可以通过提供通道值和 S3 URI 或本地路径的字典来指定多个输入数据来源。


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

注意

当输入数据很大 (>1GB) 时，需要等待很长时间才能创建作业。这是由于本地输入数据首次上传到 S3 存储桶，然后将 S3 路径添加到任务请求中。最后，工作请求将提交给 Braket 服务。

将结果保存到 S3

要保存未包含在装饰函数的返回语句中的结果，必须将正确的目录附加到所有文件写入操作中。以下示例显示了如何保存一个 numpy 数组和 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 ...

所有结果都压缩到名为 model.tar.gz 的文件中。您可以使用 Python 函数 job.result() 下载结果，也可以从 Braket 管理控制台的混合作业页面导航到结果文件夹。

保存并从检查点恢复

对于长时间运行的混合作业，建议定期保存算法的中间状态。您可以使用内置的 save_job_checkpoint() 辅助函数，也可以将文件保存到 AMZN_BRAKET_JOB_RESULTS_DIR 路径中。后者可通过辅助函数 get_job_results_dir() 获得。

以下是使用混合作业装饰器保存和加载检查点的最小工作示例：


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

在第一个混合作业中，使用包含我们要保存的数据的字典调用 save_job_checkpoint()。默认情况下，每个值都必须可序列化为文本。对于检查点更为复杂的 Python 对象，如 numpy 数组，您可以设置 data_format = PersistedJobDataFormat.PICKLED_V4。此代码在名为 <jobname>.json 的子文件夹下的混合作业构件中创建并覆盖具有默认名称的检查点文件。

要创建一个新的混合作业以从检查点继续，我们需要传递 copy_checkpoints_from_job=job_arn 到 job_arn 作为前一个作业的混合作业 ARN 的位置。然后，我们使用 load_job_checkpoint(job_name) 从检查点加载。

混合作业修饰器最佳实践

接受异步性

使用装饰器注释创建的混合作业是异步的，它们将在经典资源和量子资源可用后运行。您可以使用 Braket Management Console 或 Amazon CloudWatch 监控算法的进度。当您提交算法以供运行时，Braket 会在可扩展的容器化环境中运行您的算法，并在算法完成后检索结果。

运行迭代变分算法

混合作业为您提供了运行迭代量子经典算法的工具。对于纯粹的量子问题，请使用量子任务或一批量子任务。优先访问某些 QPU 最有利于长时间运行的变分算法，这些算法需要对 QPU 进行多次迭代调用，中间有经典处理。

使用本地模式进行调试

在 QPU 上运行混合作业之前，建议先在模拟器 SV1 上运行以确认其按预期运行。对于小规模测试，可以在本地模式下运行，以实现快速迭代和调试。

使用自带容器（BYOC）提升可重复性

将您的软件及其依赖项封装在容器化环境中，从而创建可重现的实验。通过将所有代码、依赖项和设置打包到容器中，可以防止潜在的冲突和版本控制问题。

多实例分布式模拟器

要运行大量电路，可以考虑使用内置的 MPI 支持，在单个混合作业中的多个实例上运行本地模拟器。有关更多信息，请参阅嵌入式模拟器。

使用参数电路

您通过混合作业提交的参数电路会使用参数化编译在特定 QPU 上自动编译，以改善算法的运行时。

定期检查点

对于长时间运行的混合作业，建议定期保存算法的中间状态。

有关更多示例、使用案例和最佳实践，请参阅 Amazon Braket 示例 GitHub。

Javascript 在您的浏览器中被禁用或不可用。

要使用 Amazon Web Services 文档，必须启用 Javascript。请参阅浏览器的帮助页面以了解相关说明。

文档惯例

创建混合作业

将 API 和 Hybrid Jobs 配合使用