Lewati ke konten utama

Kelola sumber daya komputasi dan data Qiskit Serverless

Versi paket

Kode di halaman ini dikembangkan menggunakan persyaratan berikut. Kami sarankan menggunakan versi ini atau yang lebih baru.

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

Dengan Qiskit Serverless, kamu bisa mengelola komputasi dan data di seluruh pola Qiskit, termasuk CPU, QPU, dan akselerator komputasi lainnya.

Atur status terperinci​

Beban kerja Serverless memiliki beberapa tahap dalam sebuah alur kerja. Secara default, status berikut bisa dilihat dengan job.status():

  • QUEUED: beban kerja sedang dalam antrean untuk sumber daya klasik
  • INITIALIZING: beban kerja sedang disiapkan
  • RUNNING: beban kerja sedang berjalan di sumber daya klasik
  • DONE: beban kerja telah berhasil diselesaikan

Kamu juga bisa mengatur status kustom yang lebih mendeskripsikan tahap alur kerja tertentu, 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

Setelah beban kerja selesai dengan sukses (menggunakan save_result()), status ini akan otomatis diperbarui menjadi DONE.

Alur kerja paralel​

Untuk tugas-tugas klasik yang bisa diparalelkan, gunakan dekorator @distribute_task untuk mendefinisikan kebutuhan komputasi yang diperlukan untuk menjalankan sebuah tugas. Mulai dengan mengingat kembali contoh transpile_remote.py dari topik Tulis program Qiskit Serverless pertamamu dengan kode berikut.

Kode berikut mengharuskan kamu sudah menyimpan kredensial kamu.

%%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, kamu mendekorasi fungsi transpile_remote() dengan @distribute_task(target={"cpu": 1}). Saat dijalankan, ini membuat tugas worker paralel asinkron dengan satu inti CPU, dan mengembalikan referensi untuk melacak worker tersebut. Untuk mengambil hasilnya, teruskan referensi ke fungsi get(). Kita bisa menggunakannya untuk menjalankan beberapa tugas paralel:

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

Eksplorasi konfigurasi tugas yang berbeda​

Kamu bisa mengalokasikan CPU, GPU, dan memori untuk tugas-tugasmu secara fleksibel melalui @distribute_task(). Untuk Qiskit Serverless di IBM Quantum® Platform, setiap program dilengkapi dengan 16 inti CPU dan 32 GB RAM, yang bisa dialokasikan secara dinamis sesuai kebutuhan.

Inti CPU bisa dialokasikan sebagai inti CPU penuh, atau bahkan alokasi pecahan, seperti yang ditunjukkan berikut ini.

Memori dialokasikan dalam jumlah byte. Ingat bahwa ada 1024 byte dalam satu kilobyte, 1024 kilobyte dalam satu megabyte, dan 1024 megabyte dalam satu gigabyte. Untuk mengalokasikan 2 GB memori bagi worker kamu, kamu perlu mengalokasikan "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

Kelola data di seluruh program kamu​

Qiskit Serverless memungkinkan kamu mengelola file di direktori /data di seluruh program kamu. Ini mencakup beberapa batasan:

  • Hanya file tar dan h5 yang didukung saat ini
  • Ini hanya penyimpanan /data yang datar, dan tidak bisa memiliki subdirektori /data/folder/

Berikut ini menunjukkan cara mengunggah file. Pastikan kamu sudah mengautentikasi ke Qiskit Serverless dengan akun IBM Quantum kamu (lihat Deploy ke IBM Quantum Platform untuk instruksinya).

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

Selanjutnya, kamu bisa melihat daftar semua file di direktori data kamu. Data ini bisa diakses oleh semua program.

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

Hal ini bisa dilakukan dari sebuah program dengan menggunakan file_download() untuk mengunduh file ke lingkungan program, dan mengekstrak 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 tahap ini, program kamu bisa berinteraksi dengan file, seperti layaknya eksperimen lokal. file_upload(), file_download(), dan file_delete() bisa dipanggil dari eksperimen lokal kamu, atau dari program yang sudah diunggah, untuk manajemen data yang konsisten dan fleksibel.

Langkah selanjutnya​

Rekomendasi
Source: IBM Quantum docs — updated 27 Apr 2026
English version on doQumentation — updated 7 Mei 2026
This translation based on the English version of 11 Mar 2026