관리형 워크플로 실행

패키지 버전

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

qiskit-serverless~=0.38.0

관리형 워크플로 작업은 HPC 스케줄러에 제출하는 배치 작업처럼 동작합니다. run()은 작업을 제출하고 즉시 Job 객체를 반환합니다. 클러스터에서 squeue, 작업의 출력 파일, scancel이 담당하는 작업에 Job 객체를 사용하세요. 노트북을 닫아도 작업은 계속 실행됩니다.

이 페이지의 예제는 각도 theta 하나를 읽고 각도가 범위를 벗어나면 실패하는 작은 워크플로를 사용합니다. 첫 관리형 워크플로 배포에 설명된 대로 업로드하세요.

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

_needed = {"qiskit_ibm_catalog": "qiskit-ibm-catalog", "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
import math

from qiskit_serverless import get_arguments, save_result

arguments = get_arguments()
theta = arguments["theta"]
if not 0 <= theta <= math.pi:
    raise ValueError(f"theta must be between 0 and pi, got {theta}")

print(f"Computing the expectation value for theta={theta}", flush=True)
save_result({"theta": theta, "expectation_value": math.cos(theta)})

Writing ./source_files/my_workflow.py

작업 시작

제목으로 워크플로를 가져온 다음, 워크플로가 읽는 인수와 함께 run()을 호출하세요.

# This cell is hidden from users. It uploads the workflow under a title that
# other test notebooks do not use, and makes load("my-workflow") return it.
# It also silences the account warning that job.cancel() logs.
import logging

from qiskit_ibm_catalog import ManagedWorkflow, ManagedWorkflowClient

logging.getLogger("qiskit_ibm_runtime").setLevel(logging.ERROR)

_test_title = "docs-run-managed-workflow"
ManagedWorkflowClient().upload(
    ManagedWorkflow(
        title=_test_title,
        entrypoint="my_workflow.py",
        working_dir="./source_files/",
    )
)
_load = ManagedWorkflowClient.load
ManagedWorkflowClient.load = lambda self, title, *a, **k: _load(
    self, _test_title if title == "my-workflow" else title, *a, **k
)

run_workflow_remotely.ipynb

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
my_workflow = client.load("my-workflow")

job = my_workflow.run(theta=0.5)
job_id = job.job_id

run()은 워크플로가 작업을 대기열에 넣은 후 반환됩니다. 새 세션에서 작업을 찾을 수 있도록 작업 ID를 저장하세요.

워크플로가 받는 인수를 선택하고 작업이 시작되기 전에 검증하려면 입력과 출력 정의를 참조하세요.
다른 컴퓨팅 크기로 작업을 실행하려면 function_size를 전달하세요. 실행마다 크기 선택을 참조하세요.

작업 추적

상태 확인

run()을 호출한 직후에는 상태가 보통 QUEUED입니다. run_workflow_remotely.ipynb

job.status()

'QUEUED'

상태	의미
`QUEUED`	작업이 컴퓨팅 슬롯을 기다리고 있습니다.
`INITIALIZING`	워크플로가 작업의 컴퓨팅 프로필로 환경을 시작하고 있습니다.
`RUNNING`	워크플로의 엔트리포인트가 실행 중입니다.
`RUNNING: <stage>`	엔트리포인트가 `update_status()`로 단계를 보고했습니다(예: `RUNNING: WAITING_FOR_QPU`).
`CANCELING`	작업을 취소했으며 워크플로가 작업을 중지하고 있습니다.
`DONE`	엔트리포인트가 성공적으로 종료되었습니다.
`ERROR`	엔트리포인트에서 오류가 발생했거나 0이 아닌 코드로 종료되었습니다. 오류 원인 찾기를 참조하세요.
`CANCELED`	작업을 취소했거나 24시간 제한에 도달했습니다.

DONE, ERROR, CANCELED는 최종 상태이며, 이 상태에서 job.in_terminal_state()는 True를 반환합니다. SUCCEEDED와 같은 다른 값을 기다리는 상태 루프는 끝나지 않습니다.

작업 종료 대기

작업이 끝날 때까지 기다리려면 wait_for_final_state()를 호출하세요. 이 함수는 최종 상태를 반환하며, verbose=True를 사용하면 상태가 바뀔 때마다 출력합니다. timeout을 초 단위로 설정하면 작업이 끝나기 전에 시간이 만료될 때 wait_for_final_state()가 TimeoutError를 발생시킵니다. 작업은 계속 실행됩니다.

run_workflow_remotely.ipynb

job.wait_for_final_state(timeout=600, verbose=True)

Waiting for job 3f2a1c0e-6b1d-4c0e-9a51-2f0c4d1e7a90
0:00:00  QUEUED
0:02:10  RUNNING
0:03:05  RUNNING: WAITING_FOR_QPU
0:41:44  DONE
'DONE'

wait_for_final_state()는 1, 2, 4, 8, 16, 32초 후에 상태를 확인하고 그 이후에는 60초마다 확인하므로, 짧은 작업의 종료는 몇 초 안에 파악하고 긴 작업에는 요청을 적게 보냅니다. 대신 고정된 간격으로 확인하려면 poll_interval을 초 단위로 전달하세요.

로그 읽기

워크플로가 표준 출력 또는 표준 오류에 기록하는 모든 내용은 작업의 로그로 들어갑니다. 작업이 실행되는 동안, 결과가 나오기 전에도 job.logs()로 읽을 수 있습니다.

# This cell is hidden from users. It waits for the job to end, so that the
# logs are complete.
job.wait_for_final_state()

run_workflow_remotely.ipynb

print(job.logs())

Computing the expectation value for theta=0.5

# This cell is hidden from users. It checks the logs.
assert "Computing the expectation value for theta=0.5" in job.logs()

워크플로는 약 15초마다, 그리고 작업이 끝나거나 취소될 때 마지막으로 한 번 새 로그 줄을 업로드합니다. 작업이 시작된 직후에는 job.logs()가 No logs yet.을 반환합니다.
워크플로는 작업마다 최신 로그 50MB를 보관하며, 로그가 잘렸음을 알리는 줄을 맨 위에 추가합니다.

워크플로가 기록할 내용을 결정하려면 로그 작성을 참조하세요. 정규식과 일치하는 줄만 표시하려면 job.filtered_logs()에 include를 전달하세요. 줄을 숨기려면 exclude를 전달하세요.

결과 가져오기

job.result()는 작업이 끝날 때까지 기다린 다음, 워크플로가 save_result()에 전달한 값을 반환합니다. run_workflow_remotely.ipynb

job.result()

{'theta': 0.5, 'expectation_value': 0.8775825618903728}

# This cell is hidden from users. It checks the result.
assert _ == {"theta": 0.5, "expectation_value": 0.8775825618903728}, _  # noqa: F821

기본적으로 job.result()는 30초마다 상태를 확인하며 시간 제한 없이 기다립니다. 확인 횟수를 제한하려면 maxwait를 전달하세요. 예를 들어 job.result(maxwait=40)은 약 20분 동안 기다립니다. 기다리지 않고 한 번만 확인하려면 wait=False를 전달하세요. 작업이 끝나기 전에 job.result()가 대기를 멈추면 {}를 반환합니다.

각 단계에 걸린 시간 확인

job.events()는 작업의 모든 상태 및 단계 변경을 시각과 함께 나열하므로, 작업이 컴퓨팅을 얼마나 기다렸고 각 단계에 얼마나 걸렸는지 확인할 수 있습니다. run_workflow_remotely.ipynb

for event in reversed(job.events()):
    print(
        event.created,
        event.data.get("status") or event.data.get("sub_status"),
    )

2026-11-02T14:43:07.134377Z QUEUED
2026-11-02T14:43:24.474199Z PENDING
2026-11-02T14:46:07.309519Z RUNNING
2026-11-02T14:46:21.115038Z SUCCEEDED

# This cell is hidden from users. It checks the events.
_statuses = [e.data.get("status") for e in job.events()]
assert "QUEUED" in _statuses and "SUCCEEDED" in _statuses, _statuses

이벤트는 저장된 상태 이름을 사용합니다. PENDING은 INITIALIZING이고, SUCCEEDED는 DONE입니다. 실행의 QPU 사용량은 실행의 QPU 작업 찾기를 참조하세요.

오류 원인 찾기

워크플로에서 예외가 발생하면 작업은 ERROR 상태로 끝납니다. 예를 들어 워크플로가 거부하는 각도로 실행해 보세요. run_workflow_remotely.ipynb

failed_job = my_workflow.run(theta=-1.0)

# This cell is hidden from users. It waits for the job to end.
failed_job.wait_for_final_state()

run_workflow_remotely.ipynb

failed_job.status()

'ERROR'

로그에 트레이스백이 표시됩니다. run_workflow_remotely.ipynb

print(failed_job.logs())

Traceback (most recent call last):
  File "/function_user_data/my_workflow.py", line 8, in <module>
    raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

job.result() raises a QiskitServerlessException: run_workflow_remotely.ipynb

from qiskit_serverless.exception import QiskitServerlessException

try:
    failed_job.result()
except QiskitServerlessException as error:
    print(error)

raise ValueError(f"theta must be between 0 and pi, got {theta}")
ValueError: theta must be between 0 and pi, got -1.0

# This cell is hidden from users. It checks the error.
assert failed_job.status() == "ERROR", failed_job.status()
assert (
    "ValueError: theta must be between 0 and pi, got -1.0"
    in failed_job.logs()
)
assert failed_job.events(type="ERROR") == []

워크플로가 send_error()로 오류를 보고하는 경우 job.events(type="ERROR")로 오류를 읽고, job.error_message()로 해당 메시지를 읽으세요. 보고된 오류가 없는 작업에서는 job.events(type="ERROR")가 빈 리스트를 반환하고 job.error_message()는 {}를 반환합니다.

run_workflow_remotely.ipynb

for event in job.events(type="ERROR"):
    print(event.created, event.data.get("code"), event.data.get("message"))

작업 관리

작업 취소

job.cancel()로 작업을 취소하세요. 워크플로가 작업을 중지하는 동안 상태는 CANCELING으로 바뀌었다가 CANCELED가 됩니다. run_workflow_remotely.ipynb

job_to_cancel = my_workflow.run(theta=1.0)
job_to_cancel.cancel()

'Job has been stopped. No active runtime job ID associated with this serverless job ID.'

# This cell is hidden from users. It checks the cancellation.
assert (
    job_to_cancel.wait_for_final_state() == "CANCELED"
), job_to_cancel.status()

job.cancel()은 작업과, 함께 취소를 시도하는 연결된 QPU 작업에 관한 메시지를 반환합니다. 실행 중인 작업의 경우 워크플로가 엔트리포인트에 SIGTERM을 보냅니다. 부분 결과를 저장하려면 작업이 취소될 때 정리하기를 참조하세요. 작업이 이미 끝났다면 job.cancel()은 Job already in terminal state.를 반환하고 아무것도 변경하지 않습니다.

새 세션에서 작업 찾기

Job 객체와 워크플로는 현재 Python 세션에서만 존재합니다. 노트북 커널을 다시 시작한 후에는 저장해 둔 작업 ID를 client.job()에 전달하거나, client.jobs()로 최근 작업을 나열하세요. run_workflow_remotely.ipynb

from qiskit_ibm_catalog import ManagedWorkflowClient

client = ManagedWorkflowClient()
job = client.job(job_id)
my_workflow = client.load("my-workflow")

client.jobs(function=my_workflow, status="ERROR", limit=5)

[<Job | e6b1ffa1-cd26-437d-bd53-b265dd4ba952>,
 <Job | 11308f2a-8d7a-40a0-a970-7fdeca2b171b>,
 <Job | af597785-3470-469a-9051-607e0185892e>,
 <Job | a9051931-3bc2-4b80-9ce6-c39d28ce260c>,
 <Job | e329e8e3-e955-4188-9bfd-8f3050ee2d39>]

# This cell is hidden from users. It checks the job lookups.
assert job.result()["theta"] == 0.5
assert failed_job.job_id in [j.job_id for j in _]  # noqa: F821

client.jobs()는 가장 최근 작업 10개를 최신순으로 반환합니다. function, status(상태 표 참조), created_after(ISO 8601 타임스탬프)로 필터링하고 limit 및 offset으로 페이지를 넘길 수 있습니다. 업로드한 워크플로를 나열하려면 client.list()를 호출하세요.

IBM Quantum Platform의 Workloads 표에도 작업이 상태, 컴퓨팅 프로필, 실행 시간과 함께 나열됩니다.

시간 제한

관리형 워크플로는 24시간을 초과하여 실행되는 작업을 취소하며, 해당 작업은 CANCELED 상태로 끝납니다. 긴 실행 중에는 작업이 손실되지 않도록 체크포인트를 파일 스토리지에 저장하세요.

작업 문제 해결

문제	원인
`run()`이 `Code: 429` 및 `Active job limit reached.`와 함께 실패합니다.	활성 작업 제한에 도달했습니다. 일부 작업이 끝나기를 기다리거나 더 이상 필요 없는 작업을 취소하세요.
작업이 `QUEUED` 상태로 유지되며 `run()`이 `your account, plan or subscription is blocked`라고 경고했습니다.	계정, 플랜 또는 구독의 차단이 해제될 때까지 작업이 시작되지 않습니다. 관리자에게 문의하세요.
`job.logs()`가 `No logs yet.`를 반환합니다.	작업이 대기 중이거나 막 시작되었습니다. 약 15초 기다리세요.
`job.result()`가 `{}`를 반환합니다.	작업이 끝나지 않았거나, 취소되었거나, 결과를 저장하지 않았습니다. 결과 저장을 참조하세요.
`job.error_message()`가 `{}`를 반환합니다.	워크플로가 오류를 보고하지 않고 예외를 발생시켰습니다. `job.logs()`로 트레이스백을 읽으세요.
작업이 `CANCELED` 상태인데 직접 취소하지 않았습니다.	작업이 24시간 제한에 도달했습니다.

다음 단계

권장 사항

로그 작성 및 오류 처리에서 작업이 보고하는 로그와 오류를 설정하세요.
파라미터 스윕 실행에서 입력 집합마다 워크플로를 한 번씩 실행하세요.

작업 시작​

작업 추적​

상태 확인​

작업 종료 대기​

로그 읽기​

결과 가져오기​

각 단계에 걸린 시간 확인​

오류 원인 찾기​

작업 관리​

작업 취소​

새 세션에서 작업 찾기​

시간 제한​

작업 문제 해결​

다음 단계​

작업 시작

작업 추적

상태 확인

작업 종료 대기

로그 읽기

결과 가져오기

각 단계에 걸린 시간 확인

오류 원인 찾기

작업 관리

작업 취소

새 세션에서 작업 찾기

시간 제한

작업 문제 해결

다음 단계