從本地 Python 代碼創建一個混合任務安裝額外的 Python 軟件包和源代碼將資料儲存並載入混合式作業執行個體混合工作裝飾器的最佳實踐

以混合工作的形式執行您的本機程式碼

Amazon Braket 混合任務提供混合量子經典演算法的全受管協調流程，將 Amazon EC2 運算資源與 Amazon Braket 量子處理單元 (QPU) 存取相結合。在混合任務中創建的量子任務優先於單個量子任務，因此您的算法不會被量子任務隊列中的波動打斷。每個 QPU 都會維護個別的混合式作業佇列，以確保在任何指定時間只能執行一個混合式作業。

在本節中：

從本地 Python 代碼創建一個混合任務
安裝額外的 Python 軟件包和源代碼
將資料儲存並載入混合式作業執行個體
混合工作裝飾器的最佳實踐

從本地 Python 代碼創建一個混合任務

您可以將本地 Python 代碼 Job 為亞馬遜編碼混合任務運行。您可以透過使用@hybrid_job裝飾器註解程式碼來執行此操作，如下列程式碼範例所示。對於自訂環境，您可以選擇使用 Amazon 彈性容器登錄 (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裝飾器中的設備參數指定了混合作業優先訪問的設備-在本例中為SV1模擬器。要獲得 QPU 優先級，您必須確保函數中使用的設備 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()以用來擷取混合式工作結果。return 語句中的任何對象都會被 Braket 自動捕獲。請注意，函數返回的對象必須是一個元組，每個元素都是可序列化的。例如，下面的代碼顯示了一個工作和一個失敗的例子。


@hybrid_job(device=Devices.Amazon.SV1)
def passing():
    np_array = np.random.rand(5)
    return np_array  # serializable
    
@hybrid_job(device=Devices.Amazon.SV1)
def failing():
    return MyObject() # not serializable

Job 名稱

根據預設，這個混合工作的名稱是從函數名稱推斷出來的。您也可以指定最多 50 個字元的自訂名稱。例如，在下面的代碼中，作業名稱是「我的工作名稱」。


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

支援所有其他混合式工作選項。有關選項的列表，請參閱文件。

安裝額外的 Python 軟件包和源代碼

您可以自訂執行階段環境以使用偏好的 Python 套件。您可以使用requirements.txt檔案、套件名稱清單，或使用自己的容器 (BYOC)。若要使用requirements.txt檔案自訂執行階段環境，請參閱下列程式碼範例。


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

例如，requirements.txt檔案可能包含要安裝的其他套件。


qiskit 
pennylane >= 0.31
mitiq == 0.29

或者，您可以提供包名稱作為 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 的網址。s3://<default_bucket_name>/jobs/<job_name>/<timestamp>/data/<channel_name>如果您指定本機路徑，通道名稱預設為「input」。下面的代碼顯示了來自本地路徑的 numpy 文件。data/file.npy


@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()輔助功能。


s3_path = "s3://amazon-braket-us-west-1-961591465522/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 或本機路徑的字典來指定多個輸入資料來源。


input_data = {
    "input": "data/file.npy",
    "input_2": "s3://my-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

要保存未包含在裝飾函數的 return 語句中的結果，您必須將正確的目錄附加到所有文件寫入操作中。下面的例子中，顯示了保存一個 numpy 數組和一個矩陣圖。


@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。

您必須啟用 Javascript，才能使用 AWS 文件。請參閱您的瀏覽器說明頁以取得說明。

文件慣用形式

何時使用 Amazon Braket 混合任務

使用 Amazon Braket 混合式任務執行混合任務