Urus sumber pengkomputeran dan data Qiskit Serverless

Versi pakej

Kod pada halaman ini dibangunkan menggunakan keperluan berikut. Kami mengesyorkan penggunaan versi ini atau yang lebih baharu.

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

Dengan Qiskit Serverless, anda boleh mengurus pengkomputeran dan data merentas corak Qiskit anda, termasuk CPU, QPU, dan pempesat pengkomputeran lain.

Tetapkan status terperinci

Beban kerja Serverless mempunyai beberapa peringkat merentasi aliran kerja. Secara lalai, status berikut boleh dilihat dengan job.status():

• QUEUED: beban kerja sedang dalam barisan untuk sumber klasik
• INITIALIZING: beban kerja sedang disediakan
• RUNNING: beban kerja sedang berjalan pada sumber klasik
• DONE: beban kerja telah selesai dengan jayanya

Anda juga boleh menetapkan status tersuai yang menerangkan peringkat aliran kerja tertentu dengan lebih lanjut, seperti berikut.

# 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

Selepas beban kerja ini selesai dengan jayanya (dengan save_result()), status ini akan dikemaskini secara automatik ke DONE.

Aliran kerja selari

Untuk tugas klasik yang boleh diselariakn, gunakan penghias @distribute_task untuk menentukan keperluan pengkomputeran yang diperlukan bagi melaksanakan tugas. Mulakan dengan mengingat semula contoh transpile_remote.py daripada topik Tulis program Qiskit Serverless pertama anda dengan kod berikut.

Kod berikut memerlukan anda telah menyimpan kelayakan anda.

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

Dalam contoh ini, anda telah menghias fungsi transpile_remote() dengan @distribute_task(target={"cpu": 1}). Apabila dijalankan, ini mencipta tugas pekerja selari tak segerak dengan satu teras CPU, dan memulangkan rujukan untuk menjejak pekerja. Untuk mendapatkan keputusan, hantar rujukan ke fungsi get(). Kita boleh menggunakannya untuk menjalankan beberapa tugas selari:

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

Terokai konfigurasi tugas yang berbeza

Anda boleh memperuntukkan CPU, GPU, dan memori secara fleksibel untuk tugas anda melalui @distribute_task(). Untuk Qiskit Serverless di IBM Quantum® Platform, setiap program dilengkapi dengan 16 teras CPU dan 32 GB RAM, yang boleh diperuntukkan secara dinamik mengikut keperluan.

Teras CPU boleh diperuntukkan sebagai teras CPU penuh, atau bahkan peruntukan pecahan, seperti yang ditunjukkan berikut.

Memori diperuntukkan dalam bilangan bait. Ingat bahawa terdapat 1024 bait dalam satu kilobait, 1024 kilobait dalam satu megabait, dan 1024 megabait dalam satu gigabait. Untuk memperuntukkan memori sebanyak 2 GB bagi pekerja anda, anda perlu memperuntukkan "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

Urus data merentas program anda

Qiskit Serverless membolehkan anda mengurus fail dalam direktori /data merentas semua program anda. Ini termasuk beberapa had:

• Hanya fail tar dan h5 yang disokong pada masa ini
• Ini hanya storan /data rata, dan tidak boleh mempunyai subdirektori /data/folder/

Berikut menunjukkan cara memuat naik fail. Pastikan anda telah mengesahkan ke Qiskit Serverless dengan akaun IBM Quantum anda (lihat Sebarkan ke IBM Quantum Platform untuk arahan).

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"}'

Seterusnya, anda boleh menyenaraikan semua fail dalam direktori data anda. Data ini boleh diakses oleh semua program.

serverless.files(function=transpile_remote_demo)

['classifier_name.pkl.tar', 'output.json.tar', 'transpile_demo.tar']

Ini boleh dilakukan daripada program dengan menggunakan file_download() untuk memuat turun fail ke persekitaran program, dan menyahzipkan 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()

Pada ketika ini, program anda boleh berinteraksi dengan fail, sama seperti eksperimen tempatan. file_upload(), file_download(), dan file_delete() boleh dipanggil dari eksperimen tempatan anda, atau program yang dimuat naik anda, untuk pengurusan data yang konsisten dan fleksibel.

Langkah seterusnya

Cadangan

• Lihat contoh penuh yang memindahkan kod sedia ada ke Qiskit Serverless.
• Baca kertas kerja di mana penyelidik menggunakan Qiskit Serverless dan pengkomputeran super berpusatkan kuantum untuk meneroka kimia kuantum.