Qiskit Serverlessへのコードの移植
ヒント
Qiskit Serverlessはアップグレード中であり、その機能は急速に変化しています。 この開発フェーズ中は、Qiskit Serverless GitHub ページでリリースノートと最新ドキュメントをご確認ください。
以下の例では、既存のコードをQiskit Serverlessを活用するように移植する方法を示します。
備考
以下のコードは、認証情報が保存済みであることを前提としています。まだ保存していない場合は、IBM Cloudアカウントのセットアップの手順に従って、APIキーで認証を行ってください。
実験のアップデート
- ローカル実験
- Serverless
from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime import QiskitRuntimeService
from qiskit.circuit.random import random_circuit
qc_random = [(random_circuit(20, 20, measure=True)) for _ in range(30)]
optimization_level = 3
service = QiskitRuntimeService(channel="ibm_quantum_platform")
backend = service.get_backend(backend_name)
pass_manager = generate_preset_pass_manager(
optimization_level=optimization_level, backend=backend
)
# @distribute_task(target={"cpu": 1})
def transpile_parallel(circuit, pass_manager):
"""Distributed transpilation for an abstract circuit into an ISA circuit for a given backend."""
isa_circuit = pass_manager.run(circuit)
return isa_circuit
transpiled_circuits = [
transpile_parallel(circuit, pass_manager)
for circuit in circuits
]
print(transpiled_circuits)
# transpile_remote.py
from qiskit.transpiler import generate_preset_pass_manager
from qiskit_serverless import get_arguments, save_result, distribute_task, get
from qiskit_ibm_runtime import QiskitRuntimeService
# Get program arguments
arguments = get_arguments()
circuits = arguments.get("circuits")
backend_name = arguments.get("backend_name")
optimization_level = arguments.get("optimization_level")
pass_manager = generate_preset_pass_manager(
optimization_level=optimization_level, backend=backend_name
)
# Distribute task across workers
@distribute_task(target={"cpu": 1})
def transpile_parallel(circuit, pass_manager):
"""Distributed transpilation for an abstract circuit into an ISA circuit for a given backend."""
isa_circuit = pass_manager.run(circuit)
return isa_circuit
try:
# Get backend
service = QiskitRuntimeService()
backend = service.get_backend(backend_name)
# run distributed tasks as async function
# we get task references as a return type
sample_task_references = [
transpile_parallel(circuit, pass_manager)
for circuit in circuits
]
# now we need to collect results from task references
results = get(sample_task_references)
# Return results
save_result({
"transpiled_circuits": results
})
except Exception as e:
# Exception handling
import traceback
print(traceback.format_exc())
Qiskit Serverlessへのアップロード
APIキーでの認証方法については、Qiskit Functions入門ページの手順に従ってください。
from qiskit_ibm_catalog import QiskitServerless, QiskitFunction
# Authenticate to the remote cluster and submit the pattern for remote execution.
serverless = QiskitServerless()
transpile_remote_demo = QiskitFunction(
title="transpile_remote_serverless",
entrypoint="transpile_remote.py",
working_dir="./source_files/",
)
serverless.upload(transpile_remote_demo)
出力
'transpile_remote_serverless'
Qiskit Serverlessでのリモート実行
from qiskit.circuit.random import random_circuit
from qiskit_ibm_runtime import QiskitRuntimeService
# Setup inputs
qc_random = [(random_circuit(20, 20, measure=True)) for _ in range(30)]
backend = "ibm_brisbane"
optimization_level = 3
# Running program
transpile_remote_serverless = serverless.load('transpile_remote_serverless')
job = transpile_remote_serverless.run(
circuits=qc_random,
backend=backend,
optimization_level=optimization_level
)
job.job_id
出力
'727e921d-512d-4b7d-af97-fe29e93ce7ea'
次のステップ
推奨事項
- Qiskit Serverlessと量子中心のスーパーコンピューティングを使用した研究者による量子化学の探求に関する論文を読んでください。