워크플로에서 양자 작업 실행

패키지 버전

이 페이지의 코드는 다음 요구 사항을 사용하여 개발되었습니다. 이 버전 또는 그 이상의 버전을 사용하시길 권장합니다.

qiskit[all]~=2.6.1
qiskit-ibm-runtime~=0.51.0
qiskit-serverless~=0.38.0

양자-고전 워크플로는 하나의 관리형 워크플로 작업으로 실행됩니다. 고전 단계는 사용자의 컴퓨트 프로필에서 실행되고, IBM Quantum® QPU로 보내는 primitive 작업은 사용자의 인스턴스에서 실행되며 사용량에 포함됩니다. 이 하나의 작업에서 각 단계를 보고하고, QPU 작업을 찾고, 취소할 수 있습니다. 작업을 시작하고 상태와 로그를 확인하려면 관리형 워크플로 실행을 참조하세요.

시작하기 전에 첫 번째 관리형 워크플로를 배포하세요.

Runtime 서비스 생성

워크플로 코드에서 get_runtime_service()로 Runtime 서비스를 생성하고, 로컬에서 QiskitRuntimeService를 사용하는 것과 같은 방식으로 사용하세요. 다음 워크플로는 Sampler primitive로 Bell 회로를 실행하고, update_status()로 각 단계를 보고하며, 카운트를 반환합니다.

# Added by doQumentation — installs the packages this notebook needs if they are missing
import importlib.util

_needed = {"qiskit": "qiskit", "qiskit_ibm_catalog": "qiskit-ibm-catalog", "qiskit_ibm_runtime": "qiskit-ibm-runtime", "qiskit_serverless": "qiskit-serverless"}
_missing = [pip for module, pip in _needed.items()
            if importlib.util.find_spec(module) is None]
# One at a time, so a package that fails to install does not block the others
for _pip in _missing:
    %pip install -q {_pip}
if not _missing:
    print("\u2713 All required packages are installed")

# This cell is hidden from users. It creates the working directory.
from pathlib import Path

Path("./source_files").mkdir(exist_ok=True)

%%writefile ./source_files/my_workflow.py
from qiskit import QuantumCircuit
from qiskit.transpiler import generate_preset_pass_manager
from qiskit_ibm_runtime.executor_sampler import Sampler
from qiskit_serverless import (
    Job,
    get_arguments,
    get_runtime_service,
    save_result,
    update_status,
)

arguments = get_arguments()

service = get_runtime_service()
backend = service.backend(arguments["backend_name"])

circuit = QuantumCircuit(2)
circuit.h(0)
circuit.cx(0, 1)
circuit.measure_all()

update_status(Job.OPTIMIZING_HARDWARE)
pass_manager = generate_preset_pass_manager(optimization_level=1, backend=backend)
isa_circuit = pass_manager.run(circuit)

update_status(Job.WAITING_QPU)
sampler = Sampler(mode=backend)
primitive_job = sampler.run([isa_circuit])
counts = primitive_job.result()[0].data.meas.get_counts()

update_status(Job.POST_PROCESSING)
save_result({"backend": backend.name, "counts": counts})

Overwriting ./source_files/my_workflow.py

get_runtime_service()는 관리형 워크플로 작업을 위해 설정된 QiskitRuntimeService를 반환하며, 다음을 제공합니다.

• 자격 증명. Runtime 서비스는 run()을 호출할 때 사용한 API 키, 인스턴스, 채널을 사용하므로 워크플로 코드에는 자격 증명이 필요하지 않습니다.

• 추적. 워크플로는 이 Runtime 서비스를 통해 실행되는 모든 primitive 작업(Sampler, Estimator, Executor 작업, 그리고 세션 또는 배치 내의 작업 포함)을 관리형 워크플로 작업 아래에 기록합니다. job.runtime_jobs()로 목록을 볼 수 있습니다.

• 취소. job.cancel()을 호출하면 관리형 워크플로는 아직 실행 중인 워크플로의 세션, 배치, primitive 작업도 함께 취소합니다.

관리형 워크플로는 QiskitRuntimeService()로 생성한 서비스의 작업을 추적하거나 취소하지 않지만, 해당 작업은 계속 실행됩니다. 워크플로에서 다른 서비스를 사용해야 한다면 SIGTERM handler에서 해당 작업을 취소하세요.

세션 또는 배치에서 primitive 실행

관리형 워크플로는 세션 또는 배치에서 실행되는 primitive 작업도 추적합니다. Runtime 서비스가 반환한 Backend로 세션 또는 배치를 생성하고, with 블록 안에서 primitive를 생성하세요. Session도 같은 방식으로 작동합니다.

source_files/my_workflow.py

from qiskit_ibm_runtime import Batch

with Batch(backend=backend) as batch:
    sampler = Sampler(mode=batch)
    primitive_jobs = [sampler.run([isa_circuit]) for _ in range(2)]
    results = [primitive_job.result() for primitive_job in primitive_jobs]

update_status()로 진행 상황 보고

Runtime 서비스 생성의 워크플로는 각 단계가 시작될 때 update_status()를 호출하여, RUNNING: WAITING_FOR_QPU와 같이 작업 상태에 워크플로의 현재 위치가 표시되도록 합니다. 다음 Job 상수 중 하나를 전달하세요.

워크플로의 호출	작업 상태
update_status(Job.MAPPING)	RUNNING: MAPPING
update_status(Job.OPTIMIZING_HARDWARE)	RUNNING: OPTIMIZING_FOR_HARDWARE
update_status(Job.WAITING_QPU)	RUNNING: WAITING_FOR_QPU
update_status(Job.EXECUTING_QPU)	RUNNING: EXECUTING_QPU
update_status(Job.POST_PROCESSING)	RUNNING: POST_PROCESSING

update_status()는 워크플로가 상태를 변경했으면 True를 반환합니다. 값이 이 상수 중 하나가 아니거나, 작업이 더 이상 활성 상태가 아니거나, 코드가 작업 밖에서 실행되면 False를 반환하고 로그에 경고를 씁니다. 오류를 발생시키지 않으므로 오타가 있어도 워크플로가 중단되지 않습니다.

옵티마이저의 반복처럼 한 단계 내의 진행 상황은 로그 줄을 작성하세요.

실행의 QPU 작업 찾기

로컬 노트북에서 인스턴스가 접근할 수 있는 QPU를 선택하세요. client.backends()는 해당 QPU를 나열하고, client.least_busy()는 대기 중인 작업이 가장 적은 QPU를 반환합니다.

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
print([backend.name for backend in client.backends(simulator=False)])

backend_name = client.least_busy(operational=True, simulator=False).name

['ibm_boston', 'ibm_fez', 'ibm_kingston', 'ibm_marrakesh', 'ibm_phoenix', 'ibm_pittsburgh']

워크플로를 업로드하고, 실행할 때 Backend 이름을 전달하세요.

from qiskit_ibm_catalog import ManagedWorkflow

my_workflow = client.upload(
    ManagedWorkflow(
        title="my-workflow",
        entrypoint="my_workflow.py",
        working_dir="./source_files/",
    )
)

job = my_workflow.run(backend_name=backend_name)

# This cell is hidden from users. It waits for the job to finish.
assert job.wait_for_final_state() == "DONE", job.logs()

작업이 끝난 후, 관리형 워크플로 작업에서 워크플로의 QPU 작업 ID를 가져옵니다.

runtime_job_ids = job.runtime_jobs()
runtime_job_ids

['dbl6lst8ft8s738secf0']

# This cell is hidden from users. It checks the tracked job.
assert len(runtime_job_ids) == 1, runtime_job_ids

각 ID는 일반 primitive 작업 ID입니다. 작업의 상태와 결과를 읽으려면 나중에 작업 결과 검색을 참조하세요. 워크플로가 세션 또는 배치에서 primitive를 실행했다면 runtime_sessions()로 세션 및 배치 ID를 나열하고, 그중 하나를 runtime_jobs()에 전달하여 해당 세션 또는 배치의 작업만 가져오세요.

run_workflow_remotely.ipynb

for session_id in job.runtime_sessions():
    print(session_id, job.runtime_jobs(session_id))

dc101077-d085-40fb-8fbe-78852e355994 ['dbjqp48nceqs73aki1k0', 'dbjqp4l6p51s73clb91g']

job 객체는 실행을 시작한 노트북에만 존재합니다. 새 노트북에서는 client.job("<your-job-id>")로 작업을 다시 가져온 다음 runtime_jobs()를 호출하세요. 새 세션에서 작업 찾기를 참조하세요.

각 QPU 작업의 QPU 사용량과 실행 시간은 작업 지표 보기를 참조하세요. 현재 기간 동안 전체 인스턴스의 사용량은 client.usage()를 호출하세요. 관리형 워크플로 작업의 각 단계에 걸린 시간을 보려면 각 단계에 걸린 시간 확인을 참조하세요.

연결된 QPU 작업 취소

job.cancel()로 관리형 워크플로 작업을 취소하면, 워크플로는 자신이 시작한 QPU 작업도 취소하려고 시도합니다.

• 세션과 배치. 워크플로는 자신의 세션과 배치를 각각 취소하며, 이때 대기 중인 작업도 함께 취소됩니다.

• 기타 primitive 작업. 워크플로는 아직 대기 중이거나 실행 중인 자신의 다른 primitive 작업을 각각 취소합니다.

job.cancel()은 각 QPU 작업과 세션에 대한 메시지를 반환하므로, 무엇이 취소되었는지 확인할 수 있습니다.

run_workflow_remotely.ipynb

print(job.cancel())

배치에서 실행된 워크플로의 경우 메시지는 다음과 같습니다.

Job is stopping. Canceled runtime session: eb8874c9-a15e-45f8-9a90-a9fb35267f9d and associated runtime jobs.

클라이언트가 로컬 자격 증명으로 QiskitRuntimeService를 생성할 수 없으면 관리형 워크플로 작업만 취소합니다. job.runtime_jobs()의 ID로 QPU 작업을 직접 취소하세요.

24시간 제한에 도달하는 등의 이유로 워크플로가 스스로 작업을 중지하면 QPU 작업은 취소하지 않습니다. job.runtime_jobs()의 ID로 직접 취소하세요.

워크플로의 QPU 작업 문제 해결

문제	원인
get_runtime_service()가 KeyError: 'QISKIT_IBM_CHANNEL'을 발생시킵니다.	코드가 관리형 워크플로 작업이 아니라 로컬 노트북에서 실행되고 있습니다. 로컬에서는 QiskitRuntimeService()를 사용하세요.
update_status()가 False를 반환하고, 로그에 'sub_status' not provided or is not valid가 표시됩니다.	값이 Job 상수 중 하나가 아니거나 작업이 더 이상 활성 상태가 아닙니다.
job.runtime_jobs()가 []를 반환합니다.	워크플로가 QiskitRuntimeService()로 서비스를 생성했거나, 아직 primitive를 실행하지 않았습니다.
세션의 작업이 추적되지 않습니다.	세션이 다른 서비스의 Backend로 생성되었습니다. get_runtime_service()에서 Backend를 가져오세요.
워크플로 로그에 Could not associate runtime job with serverless job이 표시됩니다.	워크플로가 primitive 작업을 기록하지 못했습니다. primitive 작업은 계속 실행되지만 추적되지 않습니다.
job.cancel()이 No QiskitRuntimeService can be associated to the given token and instance.를 표시합니다.	클라이언트가 로컬 자격 증명으로 QiskitRuntimeService를 생성할 수 없어 관리형 워크플로 작업만 취소했습니다.
client.job()이 Job [...] not found와 함께 실패합니다.	작업 ID가 잘못되었거나 작업이 다른 사용자의 것입니다.

다음 단계

권장 사항

• 관리형 워크플로 실행으로 작업의 상태, 결과, 로그를 확인하고 오류의 원인을 찾습니다.