{
"cells": [
{
"cell_type": "markdown",
"id": "5664d52d",
"metadata": {},
"source": [
"---\n",
"title: \"Run quantum jobs from a workflow\"\n",
"description: \"Run QPU jobs from a managed workflow, report its stages, find the QPU jobs that it ran, and cancel them.\"\n",
"---\n",
"\n",
"# Run quantum jobs from a workflow\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "35ecf711",
"metadata": {
"tags": [
"version-info"
]
},
"source": [
"{/*\n",
" DO NOT EDIT THIS CELL!!!\n",
" This cell's content is generated automatically by a script. Anything you add\n",
" here will be removed next time the notebook is run. To add new content, create\n",
" a new cell before or after this one.\n",
" */}\n",
"\n",
"\n",
" \n",
" The code on this page was developed using the following requirements.\n",
" We recommend using these versions or newer.\n",
"\n",
" ```\n",
" qiskit[all]~=2.6.1\n",
" qiskit-ibm-runtime~=0.51.0\n",
" qiskit-serverless~=0.38.0\n",
" ```\n",
" \n",
"\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "4025e66e",
"metadata": {},
"source": [
"A quantum-classical workflow runs as one managed workflow job. Its classical stages run on your compute profile, and the primitive jobs that it sends to an IBM Quantum® QPU run on your instance and count against your usage. From that one job, you can report each stage, find the QPU jobs, and cancel them. To start a job and check its status and logs, see [Run a managed workflow](/docs/managed-workflows/run-managed-workflow).\n",
"\n",
"Before you begin, [deploy your first managed workflow](/docs/managed-workflows/deploy-first-managed-workflow).\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "4b2e216a",
"metadata": {},
"source": [
"## Create the Runtime service\n",
"\n",
"Create the Runtime service in your workflow code with `get_runtime_service()`, and use it in the same way that you use `QiskitRuntimeService` locally. The following workflow runs a Bell circuit with the Sampler primitive, reports each stage with `update_status()`, and returns the counts:\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": 2,
"id": "a0db5954",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Overwriting ./source_files/my_workflow.py\n"
]
}
],
"source": [
"%%writefile ./source_files/my_workflow.py\n",
"from qiskit import QuantumCircuit\n",
"from qiskit.transpiler import generate_preset_pass_manager\n",
"from qiskit_ibm_runtime.executor_sampler import Sampler\n",
"from qiskit_serverless import (\n",
" Job,\n",
" get_arguments,\n",
" get_runtime_service,\n",
" save_result,\n",
" update_status,\n",
")\n",
"\n",
"arguments = get_arguments()\n",
"\n",
"service = get_runtime_service()\n",
"backend = service.backend(arguments[\"backend_name\"])\n",
"\n",
"circuit = QuantumCircuit(2)\n",
"circuit.h(0)\n",
"circuit.cx(0, 1)\n",
"circuit.measure_all()\n",
"\n",
"update_status(Job.OPTIMIZING_HARDWARE)\n",
"pass_manager = generate_preset_pass_manager(optimization_level=1, backend=backend)\n",
"isa_circuit = pass_manager.run(circuit)\n",
"\n",
"update_status(Job.WAITING_QPU)\n",
"sampler = Sampler(mode=backend)\n",
"primitive_job = sampler.run([isa_circuit])\n",
"counts = primitive_job.result()[0].data.meas.get_counts()\n",
"\n",
"update_status(Job.POST_PROCESSING)\n",
"save_result({\"backend\": backend.name, \"counts\": counts})"
]
},
{
"cell_type": "markdown",
"id": "15e69809",
"metadata": {},
"source": [
"`get_runtime_service()` returns a `QiskitRuntimeService` that is set up for your managed workflow job, and it provides the following:\n",
"\n",
"* **Credentials.** The Runtime service uses the API key, instance, and channel that you used to call `run()`, so the workflow code needs no credentials.\n",
"* **Tracking.** A workflow records every primitive job that runs through this Runtime service under the managed workflow job, including Sampler, Estimator, and Executor jobs, and jobs in a session or batch. You can list them with `job.runtime_jobs()`.\n",
"* **Cancellation.** When you call `job.cancel()`, the managed workflow also cancels the workflow's sessions, batches, and primitive jobs that are still running.\n",
"\n",
"A managed workflow does not track or cancel the jobs of a service that you create with `QiskitRuntimeService()`, although the jobs still run. If the workflow must use another service, cancel its jobs in a [`SIGTERM` handler](/docs/managed-workflows/managed-workflow-logs-and-errors#clean-up-when-a-job-is-canceled).\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "b962b96b",
"metadata": {},
"source": [
"## Run primitives in a session or batch\n",
"\n",
"A managed workflow also tracks primitive jobs that run in a session or batch. Create the session or batch from a backend that the Runtime service returned, and create the primitives inside the `with` block. `Session` works in the same way:\n",
"\n",
"\n",
"\n",
"```python\n",
"from qiskit_ibm_runtime import Batch\n",
"\n",
"with Batch(backend=backend) as batch:\n",
" sampler = Sampler(mode=batch)\n",
" primitive_jobs = [sampler.run([isa_circuit]) for _ in range(2)]\n",
" results = [primitive_job.result() for primitive_job in primitive_jobs]\n",
"```\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "9ad60010",
"metadata": {},
"source": [
"## Report progress with update\\_status()\n",
"\n",
"The workflow in [Create the Runtime service](#create-the-runtime-service) calls `update_status()` at the start of each stage, so that the job status shows where the workflow is, such as `RUNNING: WAITING_FOR_QPU`. Pass one of the following `Job` constants:\n",
"\n",
"| Call in the workflow | Job status |\n",
"| ---------------------------------------- | ---------------------------------- |\n",
"| `update_status(Job.MAPPING)` | `RUNNING: MAPPING` |\n",
"| `update_status(Job.OPTIMIZING_HARDWARE)` | `RUNNING: OPTIMIZING_FOR_HARDWARE` |\n",
"| `update_status(Job.WAITING_QPU)` | `RUNNING: WAITING_FOR_QPU` |\n",
"| `update_status(Job.EXECUTING_QPU)` | `RUNNING: EXECUTING_QPU` |\n",
"| `update_status(Job.POST_PROCESSING)` | `RUNNING: POST_PROCESSING` |\n",
"\n",
"`update_status()` returns `True` if the workflow changed the status. It returns `False` and writes a warning to the logs if the value is not one of these constants, if the job is no longer active, or if the code runs outside a job. It does not raise an error, so a typo does not stop the workflow.\n",
"\n",
"For progress within a stage, such as the iterations of an optimizer, [write log lines](/docs/managed-workflows/managed-workflow-logs-and-errors#best-practice-decide-what-to-log).\n",
"\n",
"{/* REVIEW: each update_status() call writes a SUB_STATUS_CHANGE event, and the 24-hour timeout is measured from the job's latest event (update_fleets_jobs_statuses.py _timed_out). send_error() events count too. This looks like a bug; do not document it until the intended behavior is confirmed. */}\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "cd6f7e33",
"metadata": {},
"source": [
"## Find the QPU jobs of a run\n",
"\n",
"In your local notebook, choose a QPU that your instance can access. `client.backends()` lists them, and `client.least_busy()` returns the one with the fewest pending jobs:\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": 3,
"id": "4ed48660",
"metadata": {},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"['ibm_boston', 'ibm_fez', 'ibm_kingston', 'ibm_marrakesh', 'ibm_phoenix', 'ibm_pittsburgh']\n"
]
}
],
"source": [
"from qiskit_ibm_catalog import ManagedWorkflowClient\n",
"\n",
"client = ManagedWorkflowClient()\n",
"print([backend.name for backend in client.backends(simulator=False)])\n",
"\n",
"backend_name = client.least_busy(operational=True, simulator=False).name"
]
},
{
"cell_type": "markdown",
"id": "b7c1e2d4",
"metadata": {},
"source": [
"Upload the workflow, and pass the name of the backend when you run it:\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": 4,
"id": "a7295000",
"metadata": {},
"outputs": [],
"source": [
"from qiskit_ibm_catalog import ManagedWorkflow\n",
"\n",
"my_workflow = client.upload(\n",
" ManagedWorkflow(\n",
" title=\"my-workflow\",\n",
" entrypoint=\"my_workflow.py\",\n",
" working_dir=\"./source_files/\",\n",
" )\n",
")\n",
"\n",
"job = my_workflow.run(backend_name=backend_name)"
]
},
{
"cell_type": "markdown",
"id": "58d3ef8d",
"metadata": {},
"source": [
"After the job ends, get the IDs of the workflow's QPU jobs from the managed workflow job:\n",
"\n"
]
},
{
"cell_type": "code",
"execution_count": 6,
"id": "9abaf965",
"metadata": {},
"outputs": [
{
"data": {
"text/plain": [
"['dbl6lst8ft8s738secf0']"
]
},
"execution_count": 6,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"runtime_job_ids = job.runtime_jobs()\n",
"runtime_job_ids"
]
},
{
"cell_type": "markdown",
"id": "a8608d91",
"metadata": {},
"source": [
"Each ID is a regular primitive job ID. To read a job's status and results, see [Retrieve job results at a later time](/docs/ibm-quantum-compute/monitor-job#retrieve-job-results-at-a-later-time).\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "b6b1b429",
"metadata": {},
"source": [
"If the workflow ran primitives in a session or batch, list the session and batch IDs with `runtime_sessions()`, and pass one to `runtime_jobs()` to get only the jobs in it:\n",
"\n",
"\n",
"\n",
"```python\n",
"for session_id in job.runtime_sessions():\n",
" print(session_id, job.runtime_jobs(session_id))\n",
"```\n",
"\n",
"```text\n",
"dc101077-d085-40fb-8fbe-78852e355994 ['dbjqp48nceqs73aki1k0', 'dbjqp4l6p51s73clb91g']\n",
"```\n",
"\n",
"The `job` object exists only in the notebook that started the run. In a new notebook, get the job again with `client.job(\"\")`, then call `runtime_jobs()` on it. See [Find a job from a new session](/docs/managed-workflows/run-managed-workflow#find-a-job-from-a-new-session).\n",
"\n",
"For the QPU usage and run times of each QPU job, see [View job metrics](/docs/ibm-quantum-compute/monitor-job#view-job-metrics). For the usage of your whole instance in the current period, call `client.usage()`. To see how long each stage of the managed workflow job took, see [Check how long each stage took](/docs/managed-workflows/run-managed-workflow#check-how-long-each-stage-took).\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "a92f4fdc",
"metadata": {},
"source": [
"## Cancel linked QPU jobs\n",
"\n",
"When you cancel a managed workflow job with `job.cancel()`, the workflow also tries to cancel the QPU work that the workflow started:\n",
"\n",
"* **Sessions and batches.** The workflow cancels each of the workflow's sessions and batches, which also cancels their queued jobs.\n",
"* **Other primitive jobs.** The workflow cancels each of the workflow's other primitive jobs that are still queued or running.\n",
"\n",
"`job.cancel()` returns a message for each QPU job and session, so you can check what was canceled:\n",
"\n",
"\n",
"\n",
"```python\n",
"print(job.cancel())\n",
"```\n",
"\n",
"For a workflow that ran in a batch, the message looks like the following:\n",
"\n",
"```text\n",
"Job is stopping. Canceled runtime session: eb8874c9-a15e-45f8-9a90-a9fb35267f9d and associated runtime jobs.\n",
"```\n",
"\n",
"If the client cannot create a `QiskitRuntimeService` from your local credentials, it cancels only the managed workflow job. Cancel the QPU jobs yourself with their IDs from `job.runtime_jobs()`.\n",
"\n",
"When a workflow stops a job itself, for example because it reached the 24-hour limit, it does not cancel its QPU jobs. Cancel them with their IDs from `job.runtime_jobs()`.\n",
"\n",
"{/* REVIEW: the scheduler's timeout and blocked-account stops (update_fleets_jobs_statuses.py stop_job_if_timeout, FleetsJobCanceller) cancel only the fleet. Only the user stop endpoint (StopJobUseCase) cancels runtime jobs. Confirm whether this is intended before publishing. */}\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "6c74dbdb",
"metadata": {},
"source": [
"## Troubleshoot QPU jobs in a workflow\n",
"\n",
"| Problem | Cause |\n",
"| ------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------- |\n",
"| `get_runtime_service()` raises `KeyError: 'QISKIT_IBM_CHANNEL'`. | The code runs in your local notebook, not in a managed workflow job. Use `QiskitRuntimeService()` locally. |\n",
"| `update_status()` returns `False`, and the logs show `'sub_status' not provided or is not valid`. | The value is not one of the `Job` constants, or the job is no longer active. |\n",
"| `job.runtime_jobs()` returns `[]`. | The workflow created its service with `QiskitRuntimeService()`, or it has not run a primitive yet. |\n",
"| A session's jobs are not tracked. | The session was created from a backend of a different service. Get the backend from `get_runtime_service()`. |\n",
"| The workflow logs show `Could not associate runtime job with serverless job`. | The workflow could not record a primitive job. The primitive job still runs, but it is not tracked. |\n",
"| `job.cancel()` shows `No QiskitRuntimeService can be associated to the given token and instance.` | The client could not create a `QiskitRuntimeService` from your local credentials, so it canceled only the managed workflow job. |\n",
"| `client.job()` fails with `Job [...] not found`. | The job ID is wrong, or the job belongs to another user. |\n",
"\n"
]
},
{
"cell_type": "markdown",
"id": "6f7a3b00",
"metadata": {},
"source": [
"## Next steps\n",
"\n",
"\n",
" * [Run a managed workflow](/docs/managed-workflows/run-managed-workflow) to check the status, result, and logs of a job, and find the cause of an error.\n",
"\n",
"\n"
]
},
{
"cell_type": "markdown",
"metadata": {},
"id": "a1b8767d",
"source": "© IBM Corp., 2017-2026"
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3"
}
},
"nbformat": 4,
"nbformat_minor": 5
}