Migrate third-party provider interfaces from `backend.run` to primitives

Why should third-party providers implement primitives?

The Qiskit SDK is built to support third parties in creating external providers of quantum resources. The evolution of user access patterns for QPUs led to the Qiskit primitives (Sampler and Estimator) superseding the backend.run interface. To ensure Qiskit users can seamlessly transition between providers, all third-party providers should implement native support of the primitives for performant access to their hardware or simulators.

This guide is for you if the following applies:

You are in charge of a third-party provider.
Your interface is still based on backend.run.
You are looking for guidance to implement your own native Sampler and Estimator primitives.

After you migrate your interface to primitives, your users with primitives-based code can seamlessly transition between providers.

Exposing native `Sampler` and `Estimator` implementations

Qiskit defines base interfaces (BaseEstimatorV2 and BaseSamplerV2) that set the standards for input and output handling in the primitives. Native primitive implementations can be developed to wrap any service provider hardware access function (for example, execute_workload(QPU) or resource.access()) or local simulator, as long as the final inputs and outputs conform to the established standards set by the base interfaces.

The Estimator interface exposes a tool to calculate expectation values of observables with respect to quantum states, while the Sampler interface outputs the counts or probability distributions obtained by sampling a quantum state in a QPU. These base interfaces also provide standard methods for execution, querying job status, and returning the job results. See the qiskit.primitives API documentation for more information.

The primitive unified bloc (PUB) data model defines the primitive input format, as described in Overview of PUBs . While different types of data can be accepted in a PUB, both SamplerPub and EstimatorPub have a coerce() method that can be used to convert them into the respective model. The following code snippets provide an example for each of the primitives.

The PrimitiveResult defines the output of a primitive job. It is comprised of additional data models such as PubResult and DataBin . See Overview of primitive results for more information.

Example of a third-party `Sampler` implementation

An example of a custom Sampler implementation for a third-party provider might look like:

from qiskit.primitives import (
    BaseSamplerV2
    BitArray,
    DataBin,
    PrimitiveResult,
    SamplerPubResult,
    SamplerPubLike,
    PrimitiveJob
)
from qiskit.primitives.containers.sampler_pub import SamplerPub
 
class CustomSampler(BaseSamplerV2):
    """
    Custom Sampler implementation for a third-party `backend`.
    """
    def __init__(
        self,
        *,
        backend: BackendV2, # object representing QPU (hardware resource)
        options: dict # options can be a dict or a subclass of the qiskit Options class
    ):
        self._hw_resource = backend
        self._options = options
        self._default_shots = 1024
 
    @property
    def backend(self) -> BackendV2:
        """ Return the Sampler's backend (hardware resource)"""
        return self._hw_resource
 
    def run(
        self, pubs: Iterable[SamplerPubLike], *, shots: int | None = None
    ) -> BasePrimitiveJob[PrimitiveResult[SamplerPubResult]]:
        """ 
        Execution method. Steps to implement: 
            1. Define a default number of shots if none is given. 
            2. Validate pub format.
            3. Instantiate an object that inherits from BasePrimitiveJob 
                containing pub and runtime information.
            4. Send the job to the provider's execution service.
            5. Return the data in some format.
        """
 
        # 1. Define default shots.
        if shots is None:
            shots = self._default_shots
 
        # 2. Validate SamplerPub format.
        coerced_pubs = [SamplerPub.coerce(pub, shots) for pub in pubs]
 
        # 3. Instantiate job object. See _run and _run_pub for more details.
        job = PrimitiveJob(self._run, coerced_pubs)
 
        # 4. Send the job to the provider's execution service.
        job._submit()
 
        # 5. Return the data.
        return job
 
    def _run(self, pubs: Iterable[SamplerPub]) -> PrimitiveResult[SamplerPubResult]:
        """
        Iterate over input pubs and call _run_pub.
        """
        results = [self._run_pub(pub) for pub in pubs]
        # return results wrapped in PrimitiveResult
        return PrimitiveResult(results, metadata={"version": 2})
 
    def _run_pub(self, pub: SamplerPub) -> SamplerPubResult:
        """
        This internal method handles the provider resource access.
        """
        # pre-process the sampling inputs to fit the required format
        circuit, qargs, meas_info = _preprocess_circuit(pub.circuit)
        bound_circuits = pub.parameter_values.bind_all(circuit)
        arrays = {
            item.creg_name: np.zeros(
                bound_circuits.shape + (pub.shots, item.num_bytes), dtype=np.uint8
            )
            for item in meas_info
        }
 
        # iterate over pub elements
        for index, bound_circuit in enumerate(bound_circuits):
            # ACCESS PROVIDER RESOURCE HERE
            # This case shows an illustrative implementation
            samples_array = self._hw_resource.sample(bound_circuit)
 
            # post-process the sampling output to fit the required format
            for item in meas_info:
                ary = _samples_to_packed_array(samples_array, item.num_bits, item.qreg_indices)
                arrays[item.creg_name][index] = ary
        meas = {
            item.creg_name: BitArray(arrays[item.creg_name], item.num_bits) for item in meas_info
        }
        # return results wrapped in SamplerPubResult
        return SamplerPubResult(
            DataBin(**meas, shape=pub.shape),
            metadata={"shots": pub.shots, "circuit_metadata": pub.circuit.metadata},
        )

This implementation is based on the StatevectorSampler source code, with some additions to represent the backend (hardware) resource access, as the StatevectorSampler class is designed for local statevector simulators.

Example of a third-party `Estimator` implementation

An example of a custom Estimator implementation for a third-party provider might look like:

from .base import BaseEstimatorV2
from .containers import DataBin, EstimatorPubLike, PrimitiveResult, PubResult
from .containers.estimator_pub import EstimatorPub
from .primitive_job import PrimitiveJob
...
 
class CustomEstimator(BaseEstimatorV2):
 
    """
    Custom Estimator implementation for a third-party `backend`.
    """
    def __init__(
        self,
        *,
        backend: BackendV2, # object representing QPU (hardware) resource
        options: dict # options can be a dict or a subclass of the qiskit Options class
    ):
        self._hw_resource = backend
        self._options = options
        self._default_precision = 1024
 
    @property
    def backend(self) -> BackendV2:
        """ Return the Estimator's backend (hardware) resource"""
        return self._hw_resource
 
    def run(
        self, pubs: Iterable[EstimatorPubLike], *, precision: float | None = None
    ) -> PrimitiveJob[PrimitiveResult[PubResult]]:
        """ 
        Execution method. Steps to implement: 
            1. Define a default precision if none is given. 
            2. Validate the PUB format.
            3. Instantiate an object that inherits from BasePrimitiveJob 
                containing PUB and runtime information.
            4. Send the job to the provider's execution service.
            5. Return the data in some format.
        """
 
        # 1. Define default precision
        if precision is None:
            precision = self._default_precision
 
        # 2. Validate EstimatorPub format
        coerced_pubs = [EstimatorPub.coerce(pub, precision) for pub in pubs]
 
        # 3. Instantiate the job object. See _run and _run_pub for more details
        job = PrimitiveJob(self._run, coerced_pubs)
 
        # 4. Send the job to the provider's execution service
        job._submit()
 
        # 5. Return the data
        return job
 
    def _run(self, pubs: list[EstimatorPub]) -> PrimitiveResult[PubResult]:
        """
        Iterate over input pubs and call _run_pub.
        """
        return PrimitiveResult([self._run_pub(pub) for pub in pubs], metadata={"version": 2})
 
    def _run_pub(self, pub: EstimatorPub) -> PubResult:
        """
        This internal method handles the provider resource access.
        """
        # pre-process the estimator inputs to fit the required format
        rng = np.random.default_rng(self._seed)
        circuit = pub.circuit
        observables = pub.observables
        parameter_values = pub.parameter_values
        precision = pub.precision
        bound_circuits = parameter_values.bind_all(circuit)
        bc_circuits, bc_obs = np.broadcast_arrays(bound_circuits, observables)
        evs = np.zeros_like(bc_circuits, dtype=np.float64)
        stds = np.zeros_like(bc_circuits, dtype=np.float64)
 
        # iterate over PUB elements
        for index in np.ndindex(*bc_circuits.shape):
            # further pre-processing of the inputs to fit the required format
            bound_circuit = bc_circuits[index]
            observable = bc_obs[index]
            paulis, coeffs = zip(*observable.items())
            obs = SparsePauliOp(paulis, coeffs)
 
            # ACCESS PROVIDER RESOURCE HERE
            # This case shows an illustrative implementation
            samples_array = self._hw_resource.sample(bound_circuit, rng, precision)
 
            # post-process the sampling output to extract expectation value
            expectation_value = compute_expectation_value(samples_array, obs)
            evs[index] = expectation_value
 
        # return results wrapped in PubResult
        data = DataBin(evs=evs, stds=stds, shape=evs.shape)
        return PubResult(
            data, metadata={"target_precision": precision, "circuit_metadata": pub.circuit.metadata}
        )

Similar to the Sampler example, this implementation is based on the StatevectorEstimator source code, with some additions to represent the backend (hardware) resource access, as the StatevectorEstimator class is designed for local statevector simulators.

Migrate third-party provider interfaces from backend.run to primitives

Why should third-party providers implement primitives?

Exposing native Sampler and Estimator implementations

Example of a third-party Sampler implementation

Example of a third-party Estimator implementation

Migrate third-party provider interfaces from `backend.run` to primitives

Exposing native `Sampler` and `Estimator` implementations

Example of a third-party `Sampler` implementation

Example of a third-party `Estimator` implementation