Lumaktaw sa pangunahing nilalaman

Pamahalaan ang mga compute at data resources ng Qiskit Serverless

Mga bersyon ng package

Ang code sa pahinang ito ay ginawa gamit ang mga sumusunod na requirements. Inirerekomenda naming gamitin ang mga bersyong ito o mas bago pa.

qiskit[all]~=2.0.0
qiskit-ibm-runtime~=0.37.0
qiskit-serverless~=0.22.0

Gamit ang Qiskit Serverless, maaari mong pamahalaan ang compute at data sa buong Qiskit pattern mo, kasama na ang mga CPU, QPU, at iba pang compute accelerators.

Magtakda ng detalyadong mga status​

Maraming yugto ang mga serverless workload sa isang workflow. Sa default, ang mga sumusunod na status ay makikita gamit ang job.status():

  • QUEUED: ang workload ay nakapila para sa mga classical resources
  • INITIALIZING: ini-set up ang workload
  • RUNNING: kasalukuyang tumatakbo ang workload sa mga classical resources
  • DONE: matagumpay na natapos ang workload

Maaari ka ring magtakda ng mga custom na status na nagbibigay ng mas detalyadong paglalarawan sa partikular na yugto ng workflow, tulad ng sumusunod.

# Added by doQumentation — required packages for this notebook
!pip install -q qiskit qiskit-ibm-runtime qiskit-serverless
# This cell is hidden from users, it just creates a new folder
from pathlib import Path

Path("./source_files").mkdir(exist_ok=True)
%%writefile ./source_files/status_example.py

from qiskit_serverless import update_status, Job

## If your function has a mapping stage, particularly application functions, you can set the status to "RUNNING: MAPPING" as follows:
update_status(Job.MAPPING)

## While handling transpilation, error suppression, and so forth, you can set the status to "RUNNING: OPTIMIZING_FOR_HARDWARE":
update_status(Job.OPTIMIZING_HARDWARE)

## After you submit jobs to Qiskit Runtime, the underlying quantum job will be queued. You can set status to "RUNNING: WAITING_FOR_QPU":
update_status(Job.WAITING_QPU)

## When the Qiskit Runtime job starts running on the QPU, set the following status "RUNNING: EXECUTING_QPU":
update_status(Job.EXECUTING_QPU)

## Once QPU is completed and post-processing has begun, set the status "RUNNING: POST_PROCESSING":
update_status(Job.POST_PROCESSING)
Writing ./source_files/status_example.py

Pagkatapos matagumpay na makumpleto ang workload na ito (gamit ang save_result()), awtomatikong ia-update ang status na ito sa DONE.

Mga parallel na workflow​

Para sa mga classical na gawain na maaaring i-parallelize, gamitin ang @distribute_task decorator para tukuyin ang mga kinakailangang compute na kailangan para maisagawa ang isang gawain. Simulan sa pag-alala sa halimbawa ng transpile_remote.py mula sa paksa ng Isulat ang iyong unang Qiskit Serverless na programa gamit ang sumusunod na code.

Ang sumusunod na code ay nangangailangan na na-save mo na ang iyong mga credentials.

%%writefile ./source_files/transpile_remote.py

from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime import QiskitRuntimeService
from qiskit_serverless import distribute_task

service = QiskitRuntimeService()

@distribute_task(target={"cpu": 1})
def transpile_remote(circuit, optimization_level, backend):
    """Transpiles an abstract circuit (or list of circuits) into an ISA circuit for a given backend."""
    pass_manager = generate_preset_pass_manager(
        optimization_level=optimization_level,
        backend=service.backend(backend)
    )
    isa_circuit = pass_manager.run(circuit)
    return isa_circuit
Writing ./source_files/transpile_remote.py

Sa halimbawang ito, dinekorahan mo ang function na transpile_remote() ng @distribute_task(target={"cpu": 1}). Kapag pinatakbo, lumilikha ito ng asynchronous na parallel worker task na may iisang CPU core, at nagbabalik ng reference para subaybayan ang worker. Para makuha ang resulta, ipasa ang reference sa function na get(). Magagamit natin ito para magpatakbo ng maraming parallel na gawain:

%%writefile --append ./source_files/transpile_remote.py

from time import time
from qiskit_serverless import get, get_arguments, save_result, update_status, Job

# Get arguments
arguments = get_arguments()
circuit = arguments.get("circuit")
optimization_level = arguments.get("optimization_level")
backend = arguments.get("backend")
Appending to ./source_files/transpile_remote.py
%%writefile --append ./source_files/transpile_remote.py
# Start distributed transpilation

update_status(Job.OPTIMIZING_HARDWARE)

start_time = time()
transpile_worker_references = [
    transpile_remote(circuit, optimization_level, backend)
    for circuit in arguments.get("circuit_list")
]

transpiled_circuits = get(transpile_worker_references)
end_time = time()
Appending to ./source_files/transpile_remote.py
%%writefile --append ./source_files/transpile_remote.py
# Save result, with metadata

result = {
    "circuits": transpiled_circuits,
    "metadata": {
        "resource_usage": {
            "RUNNING: OPTIMIZING_FOR_HARDWARE": {
                "CPU_TIME": end_time - start_time,
                "QPU_TIME": 0,
            },
        }
    },
}

save_result(result)
Appending to ./source_files/transpile_remote.py
# This cell is hidden from users.
# It uploads the serverless program and checks it runs.

def test_serverless_job(title, entrypoint):
    # Import in function to stop them interfering with user-facing code
    from qiskit.circuit.random import random_circuit
    from qiskit_serverless import IBMServerlessClient, QiskitFunction
    import time
    import uuid

    title += "_" + uuid.uuid4().hex[:8]
    serverless = IBMServerlessClient()
    transpile_remote_demo = QiskitFunction(
        title=title,
        entrypoint=entrypoint,
        working_dir="./source_files/",
    )
    serverless.upload(transpile_remote_demo)
    job = serverless.get(title).run(
        circuit=random_circuit(3, 3),
        circuit_list=[random_circuit(3, 3) for _ in range(3)],
        backend="ibm_torino",
        optimization_level=1,
    )
    for retry in range(25):
        time.sleep(5)
        status = job.status()
        if status == "DONE":
            print("Job completed successfully")
            return
        if status not in [
            "QUEUED",
            "INITIALIZING",
            "RUNNING",
            "RUNNING: OPTIMIZING_FOR_HARDWARE",
            "DONE",
        ]:
            raise Exception(
                f"Unexpected job status '{status}'.\nHere's the logs:\n"
                + job.logs()
            )
        print(f"Waiting for job (status '{status}')")
    raise Exception("Job did not complete in time")

test_serverless_job(
    title="transpile_remote_serverless_test", entrypoint="transpile_remote.py"
)
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'RUNNING')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Job completed successfully

I-explore ang iba't ibang configuration ng gawain​

Maaari kang mag-allocate ng CPU, GPU, at memory para sa iyong mga gawain nang flexible gamit ang @distribute_task(). Para sa Qiskit Serverless sa IBM Quantum® Platform, ang bawat programa ay may kasamang 16 CPU cores at 32 GB RAM, na maaaring i-allocate nang dynamic ayon sa pangangailangan.

Ang mga CPU core ay maaaring i-allocate bilang buong CPU cores, o kahit fractional na allocation, tulad ng ipinapakita sa sumusunod.

Ang memory ay ini-allocate sa bilang ng bytes. Tandaan na mayroon 1024 bytes sa isang kilobyte, 1024 kilobytes sa isang megabyte, at 1024 megabytes sa isang gigabyte. Para mag-allocate ng 2 GB ng memory para sa iyong worker, kailangan mong mag-allocate ng "mem": 2 * 1024 * 1024 * 1024.

%%writefile --append ./source_files/transpile_remote.py

@distribute_task(target={
    "cpu": 16,
    "mem": 2 * 1024 * 1024 * 1024
})
def transpile_remote(circuit, optimization_level, backend):
    return None
Appending to ./source_files/transpile_remote.py
# This cell is hidden from users.
# It checks the distributed program works.
test_serverless_job(
    title="transpile_remote_serverless_test", entrypoint="transpile_remote.py"
)
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'QUEUED')
Waiting for job (status 'RUNNING')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Waiting for job (status 'RUNNING: OPTIMIZING_FOR_HARDWARE')
Job completed successfully

Pamahalaan ang data sa buong programa mo​

Pinapayagan ka ng Qiskit Serverless na pamahalaan ang mga file sa direktoryo ng /data sa lahat ng iyong programa. Kasama dito ang ilang mga limitasyon:

  • Ang tar at h5 na mga file lamang ang sinusuportahan sa ngayon
  • Flat na /data storage lamang ito, at hindi maaaring magkaroon ng /data/folder/ na mga subdirectory

Ang sumusunod ay nagpapakita kung paano mag-upload ng mga file. Tiyaking na-authenticate ka na sa Qiskit Serverless gamit ang iyong IBM Quantum account (tingnan ang I-deploy sa IBM Quantum Platform para sa mga tagubilin).

import tarfile
from qiskit_serverless import IBMServerlessClient

# Create a tar
filename = "transpile_demo.tar"
file = tarfile.open(filename, "w")
file.add("./source_files/transpile_remote.py")
file.close()

# Get a reference to a QiskitFunction
serverless = IBMServerlessClient()
transpile_remote_demo = next(
    program
    for program in serverless.list()
    if program.title == "transpile_remote_serverless"
)

# Upload the tar to Serverless data directory
serverless.file_upload(file=filename, function=transpile_remote_demo)
'{"message":"/usr/src/app/media/5e1f442128cdf60018496a04/transpile_demo.tar"}'

Susunod, maaari mong ilista ang lahat ng mga file sa iyong direktoryo ng data. Ang data na ito ay accessible sa lahat ng programa.

serverless.files(function=transpile_remote_demo)
['classifier_name.pkl.tar', 'output.json.tar', 'transpile_demo.tar']

Maaari itong gawin mula sa isang programa sa pamamagitan ng paggamit ng file_download() para i-download ang file sa kapaligiran ng programa, at pag-uncompress ng tar.

%%writefile ./source_files/extract_tarfile.py

import tarfile
from qiskit_serverless import IBMServerlessClient

serverless = IBMServerlessClient(token="<YOUR_API_KEY>") # Use the 44-character API_KEY you created and saved from the IBM Quantum Platform Home dashboard
files = serverless.files()
demo_file = files[0]
downloaded_tar = serverless.file_download(demo_file)

with tarfile.open(downloaded_tar, 'r') as tar:
    tar.extractall()

Sa puntong ito, maaari nang makipag-ugnayan ang iyong programa sa mga file, tulad ng sa isang lokal na eksperimento. Ang file_upload(), file_download(), at file_delete() ay maaaring tawagan mula sa iyong lokal na eksperimento, o ang iyong na-upload na programa, para sa pare-pareho at flexible na pamamahala ng data.

Mga susunod na hakbang​

Mga rekomendasyon