Upload a Qiskit Runtime program¶

Construct a runtime program¶

Below is a template of a runtime program. You can find the template file in the qiskit-ibm-runtime repository.

import json

from qiskit_ibm_runtime.program import UserMessenger, ProgramBackend


def program(backend: ProgramBackend, user_messenger: UserMessenger, **kwargs):
    """Function that does classical-quantum calculation."""
    # UserMessenger can be used to publish interim results.
    user_messenger.publish("This is an interim result.")
    return "final result"


def main(backend: ProgramBackend, user_messenger: UserMessenger, **kwargs):
    """This is the main entry point of a runtime program.

    The name of this method must not change. It also must have ``backend``
    and ``user_messenger`` as the first two positional arguments.

    Args:
        backend: Backend for the circuits to run on.
        user_messenger: Used to communicate with the program user.
        kwargs: User inputs.
    """
    # Massage the input if necessary.
    result = program(backend, user_messenger, **kwargs)
    # Final result can be directly returned
    return result

Each runtime program must have a main() function, which serves as the entry point to the program. This function must have backend and user_messenger as the first two positional arguments:

• backend is an instance of ProgramBackend and has a run() method that can be used to submit circuits.

• user_messenger is an instance of UserMessenger and has a publish() method that can be used to send interim and final results to the program user. This method takes a parameter final that indicates whether it’s a final result. However, it is recommended to return the final result directly from the main() function. Currently only final results are stored after a program execution finishes.

There are several runtime program source code in the program_source directory of the qiskit-ibm-runtime repository. program_source/hello_world/hello_world.py is one of them. It is a sample runtime program that submits random circuits for user-specified iterations:

"""A sample runtime program that submits random circuits for user-specified iterations."""

import random

from qiskit import transpile
from qiskit.circuit.random import random_circuit


def prepare_circuits(backend):
    """Generate a random circuit.

    Args:
        backend: Backend used for transpilation.

    Returns:
        Generated circuit.
    """
    circuit = random_circuit(
        num_qubits=5, depth=4, measure=True, seed=random.randint(0, 1000)
    )
    return transpile(circuit, backend)


def main(backend, user_messenger, **kwargs):
    """Main entry point of the program.

    Args:
        backend: Backend to submit the circuits to.
        user_messenger: Used to communicate with the program consumer.
        kwargs: User inputs.
    """
    iterations = kwargs.pop("iterations", 5)
    for it in range(iterations):
        qc = prepare_circuits(backend)
        result = backend.run(qc).result()
        user_messenger.publish({"iteration": it, "counts": result.get_counts()})

    return "Hello, World!"

Note: It is recommended that you remove all unnecessary code, such as imports, to minimize the risk of the program being considered hostile.

Data serialization¶

Runtime programs live in the cloud, and JSON is the standard way of passing data to and from cloud services. Therefore, when a user invokes a runtime program, the input parameters must first be serialized into the JSON format and then deserialized once received by the server. By default, this serialization and deserialization is done automatically using the RuntimeEncoder and RuntimeDecoder classes.

import json


class MyCustomClass:
    def __init__(self, foo, bar):
        self._foo = foo
        self._bar = bar

    def to_json(self):
        """Convert this instance to a JSON string."""
        return json.dumps({"foo": self._foo, "bar": self._bar})

    @classmethod
    def from_json(cls, json_str):
        """Return a MyCustomClass instance based on the input JSON string."""
        return cls(**json.loads(json_str))

program_inputs = {
    'my_obj': MyCustomClass("my foo", "my bar")
}

options = {"backend": "ibmq_qasm_simulator"}
job = service.run(program_id="some-program",
                           options=options,
                           inputs=program_inputs
                          )

def main(backend, user_messenger, **kwargs):
    """Main entry point of the program."""
    my_obj_str = kwargs.pop("my_obj")
    my_obj = MyCustomClass.from_json(my_obj_str)

def main(backend, user_messenger, **kwargs):
    """Main entry point of the program."""
    return MyCustomClass("this foo", "that bar")

output_str = job.result()
output = MyCustomClass.from_json(output_str)

from qiskit_ibm_runtime.program import ResultDecoder


class MyResultDecoder(ResultDecoder):
    @classmethod
    def decode(cls, data):
        data = super().decoded(data)  # Perform any preprocessing.
        return MyCustomClass.from_json(data)

output = job.result(decoder=MyResultDecoder)

Test your runtime program¶

You can test your runtime program using a local simulator or a real backend before uploading it. Simply import and invoke the main() function of your program and pass the following parameters:

• the backend instance you want to use

• a new UserMessenger instance

• program input parameters that are serialized and then deserialized using the correct encoder and decoder. While this may seem redundant, it is to ensure input parameters can be passed to your program properly once it’s uploaded to the cloud

The following example tests the hello-world program we saw earlier. It uses the qasm_simulator from Qiskit Aer as the test backend. It serializes and deserializes input data using RuntimeEncoder and RuntimeDecoder, which are the default en/decoders used by the runtime.

import sys

sys.path.insert(0, "..")  # Add source_program directory to the path

from program_source.hello_world import hello_world
from qiskit import Aer
from qiskit_ibm_runtime import RuntimeEncoder, RuntimeDecoder, UserMessenger

inputs = {"iterations": 3}

backend = Aer.get_backend("qasm_simulator")
user_messenger = UserMessenger()
serialized_inputs = json.dumps(inputs, cls=RuntimeEncoder)
deserialized_inputs = json.loads(serialized_inputs, cls=RuntimeDecoder)

hello_world.main(backend, user_messenger, **deserialized_inputs)

{"iteration": 0, "counts": {"00011": 10, "00010": 7, "01011": 527, "01010": 480}}
{"iteration": 1, "counts": {"00000": 884, "00110": 140}}
{"iteration": 2, "counts": {"00001": 506, "00011": 518}}

'Hello, World!'

Define program metadata¶

Program metadata helps users to understand how to use your program. It includes:

• name: Name of the program.

• max_execution_time: Maximum amount of time, in seconds, a program can run before being forcibly terminated.

• description: Describes the program.

• spec: Detailed information about the program, which includes the following attributes:

  • backend_requirements: Describes the backend attributes needed to run the program.

  • parameters: Describes the program input parameters as a JSON schema

  • return_values: Describes the return values as a JSON schema

  • interim_results: Describes the interim results as a JSON schema

When uploading a program, you must specify at least name, max_execution_time, and description. It is strongly encouraged to also specify parameters, return_values, and interim_results within spec if the program has them.

Below shows the metadata JSON file of the hello-world program as an example:

import os

hello_world_json = os.path.join(
    os.getcwd(), "../../program_source/hello_world/hello_world.json"
)

with open(hello_world_json, "r") as file:
    data = file.read()

print(data)

{
  "name": "hello-world",
  "description": "A sample runtime program.",
  "max_execution_time": 300,
  "spec": {
    "backend_requirements": {
      "min_num_qubits": 5
    },
    "parameters": {
      "$schema": "https://json-schema.org/draft/2019-09/schema",
      "properties": {
        "iterations": {
          "type": "integer",
          "minimum": 0,
          "description": "Number of iterations to run. Each iteration generates a runs a random circuit."
        }
      },
      "required": [
        "iterations"
      ]
    },
    "return_values": {
      "$schema": "https://json-schema.org/draft/2019-09/schema",
      "description": "A string that says 'Hello, World!'.",
      "type": "string"
    },
    "interim_results": {
      "$schema": "https://json-schema.org/draft/2019-09/schema",
      "properties": {
        "iteration": {
          "type": "integer",
          "description": "Iteration number."
        },
        "counts": {
          "description": "Histogram data of the circuit result.",
          "type": "object"
        }
      }
    }
  }
}

Upload a program¶

You can use the QiskitRuntimeService.upload_program() method to upload your program. In the example below, the program data lives in the file hello_world.py, and its metadata, as described above, is in hello_world.json.

import os
from qiskit_ibm_runtime import QiskitRuntimeService

service = QiskitRuntimeService()

hello_world_data = os.path.join(
    os.getcwd(), "../../program_source/hello_world/hello_world.py"
)
hello_world_json = os.path.join(
    os.getcwd(), "../../program_source/hello_world/hello_world.json"
)

program_id = service.upload_program(data=hello_world_data, metadata=hello_world_json)
print(program_id)

hello-world-m92QdGKYro

upload_program() returns a program ID, which uniquely identifies the program. It is derived from the program name, usually with a randomly-generated suffix. Program ID is needed to invoke the program.

Update a program¶

You can use the QiskitRuntimeService.update_program() method to update the source code and/or metadata of a program:

service.update_program(program_id=program_id, description="A new description.")

This method allows you to make changes to your program while retaining the same program ID.

Delete a program¶

You can use the QiskitRuntimeService.delete_program() method to delete a program. Only the person who uploaded the program can delete it.

service.delete_program(program_id)

from qiskit.tools.jupyter import *

%qiskit_copyright