Run quantum jobs from a workflow

• Package versions

  The code on this page was developed using the following requirements. We recommend using these versions or newer.

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

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.

Before you begin, deploy your first managed workflow.

Create the Runtime service

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:

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})

get_runtime_service() returns a QiskitRuntimeService that is set up for your managed workflow job, and it provides the following:

• Credentials. The Runtime service uses the API key, instance, and channel that you used to call run(), so the workflow code needs no credentials.
• 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().
• Cancellation. When you call job.cancel(), the managed workflow also cancels the workflow's sessions, batches, and primitive jobs that are still running.

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.

Run primitives in a session or batch

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:

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]

Report progress with update_status()

The workflow in 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:

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.

For progress within a stage, such as the iterations of an optimizer, write log lines.

Find the QPU jobs of a run

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:

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

Output:

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

Upload the workflow, and pass the name of the backend when you run it:

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)

After the job ends, get the IDs of the workflow's QPU jobs from the managed workflow job:

runtime_job_ids = job.runtime_jobs()
runtime_job_ids

Output:

['dbl6lst8ft8s738secf0']

Each ID is a regular primitive job ID. To read a job's status and results, see Retrieve job results at a later time.

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:

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

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

The job object exists only in the notebook that started the run. In a new notebook, get the job again with client.job("<your-job-id>"), then call runtime_jobs() on it. See Find a job from a new session.

For the QPU usage and run times of each QPU job, see 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.

Cancel linked QPU jobs

When you cancel a managed workflow job with job.cancel(), the workflow also tries to cancel the QPU work that the workflow started:

• Sessions and batches. The workflow cancels each of the workflow's sessions and batches, which also cancels their queued jobs.
• Other primitive jobs. The workflow cancels each of the workflow's other primitive jobs that are still queued or running.

job.cancel() returns a message for each QPU job and session, so you can check what was canceled:

print(job.cancel())

For a workflow that ran in a batch, the message looks like the following:

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

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().

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().

Troubleshoot QPU jobs in a workflow

Next steps