Task support in v2 primitives

qiskit_ibm_runtime/__init__.py

@@ -204,4 +204,4 @@ def result_callback(job_id, result):
 QISKIT_IBM_RUNTIME_LOG_FILE = "QISKIT_IBM_RUNTIME_LOG_FILE"
 """The environment variable name that is used to set the file for the IBM Quantum logger."""
 
-warnings.warn("You are using the experimental branch. Stability is not guaranteed.")
+warnings.warn("You are using the experimental-0.2 branch. Stability is not guaranteed.")

qiskit_ibm_runtime/base_primitive.py

@@ -24,14 +24,18 @@
 
 from qiskit_ibm_provider.session import get_cm_session as get_cm_provider_session
 
-from .options import BaseOptions, Options
+from .options import Options
+from .options.options import BaseOptions, OptionsV2
 from .options.utils import merge_options, set_default_error_levels
 from .runtime_job import RuntimeJob
 from .ibm_backend import IBMBackend
 from .utils.default_session import get_cm_session
 from .constants import DEFAULT_DECODERS
 from .qiskit_runtime_service import QiskitRuntimeService
 
+# TODO: remove when we have real v2 base estimator
+from .qiskit.primitives import EstimatorTask, SamplerTask
+
 # pylint: disable=unused-import,cyclic-import
 from .session import Session
 
@@ -41,7 +45,7 @@
 class BasePrimitiveV2(ABC):
     """Base class for Qiskit Runtime primitives."""
 
-    _OPTIONS_CLASS: type[BaseOptions] = Options
+    _options_class: Optional[type[BaseOptions]] = OptionsV2
     version = 2
 
     def __init__(
@@ -75,16 +79,7 @@ def __init__(
         self._service: QiskitRuntimeService = None
         self._backend: Optional[IBMBackend] = None
 
-        opt_cls = self._OPTIONS_CLASS
-        if options is None:
-            self.options = opt_cls()
-        elif isinstance(options, opt_cls):
-            self.options = replace(options)
-        elif isinstance(options, dict):
-            default_options = opt_cls()
-            self.options = opt_cls(**merge_options(default_options, options))
-        else:
-            raise ValueError(f"Invalid 'options' type. It can only be a dictionary of {opt_cls}")
+        self._set_options(options)
 
         if isinstance(session, Session):
             self._session = session
@@ -123,36 +118,33 @@ def __init__(
                     "A backend or session must be specified when not using ibm_cloud channel."
                 )
 
-    def _run_primitive(self, primitive_inputs: Dict, user_kwargs: Dict) -> RuntimeJob:
+    def _run(self, tasks: Union[list[EstimatorTask], list[SamplerTask]]) -> RuntimeJob:
         """Run the primitive.
 
         Args:
-            primitive_inputs: Inputs to pass to the primitive.
-            user_kwargs: Individual options to overwrite the default primitive options.
+            tasks: Inputs tasks to pass to the primitive.
 
         Returns:
             Submitted job.
         """
-        logger.debug("Merging current options %s with %s", self.options, user_kwargs)
-        combined = merge_options(self.options, user_kwargs)
+        primitive_inputs = {"tasks": tasks}
+        options_dict = asdict(self.options)
+        self._validate_options(options_dict)
+        primitive_inputs.update(self._options_class._get_program_inputs(options_dict))
+        runtime_options = self._options_class._get_runtime_options(options_dict)
 
-        self._validate_options(combined)
-
-        primitive_inputs.update(self._OPTIONS_CLASS._get_program_inputs(combined))
-        runtime_options = self._OPTIONS_CLASS._get_runtime_options(combined)
+        if self._backend and options_dict["transpilation"]["skip_transpilation"]:
+            for task in tasks:
+                self._backend.check_faulty(task.circuit)
 
-        if self._backend and combined["transpilation"]["skip_transpilation"]:
-            for circ in primitive_inputs["circuits"]:
-                self._backend.check_faulty(circ)
-
-        logger.info("Submitting job using options %s", combined)
+        logger.info("Submitting job using options %s", options_dict)
 
         if self._session:
             return self._session.run(
                 program_id=self._program_id(),
                 inputs=primitive_inputs,
                 options=runtime_options,
-                callback=combined.get("environment", {}).get("callback", None),
+                callback=options_dict.get("environment", {}).get("callback", None),
                 result_decoder=DEFAULT_DECODERS.get(self._program_id()),
             )
 
@@ -165,7 +157,7 @@ def _run_primitive(self, primitive_inputs: Dict, user_kwargs: Dict) -> RuntimeJo
             program_id=self._program_id(),
             options=runtime_options,
             inputs=primitive_inputs,
-            callback=combined.get("environment", {}).get("callback", None),
+            callback=options_dict.get("environment", {}).get("callback", None),
             result_decoder=DEFAULT_DECODERS.get(self._program_id()),
         )
 
@@ -178,15 +170,19 @@ def session(self) -> Optional[Session]:
         """
         return self._session
 
-    def set_options(self, **fields: Any) -> None:
-        """Set options values for the sampler.
-
-        Args:
-            **fields: The fields to update the options
-        """
-        self.options = self._OPTIONS_CLASS(  # pylint: disable=attribute-defined-outside-init
-            **merge_options(self.options, fields)
-        )
+    def _set_options(self, options: Optional[Union[Dict, BaseOptions]] = None) -> None:
+        """Set options."""
+        if options is None:
+            self._options = self._options_class()
+        elif isinstance(options, dict):
+            default_options = self._options_class()
+            self.options = self._options_class(**merge_options(default_options, options))
+        elif isinstance(options, self._options_class):
+            self._options = replace(options)
+        else:
+            raise TypeError(
+                f"Invalid 'options' type. It can only be a dictionary of {self._options_class}"
+            )
 
     @abstractmethod
     def _validate_options(self, options: dict) -> None:

qiskit_ibm_runtime/estimator.py

@@ -14,28 +14,20 @@
 
 from __future__ import annotations
 import os
-from typing import Optional, Dict, Sequence, Any, Union, Mapping
+from typing import Optional, Dict, Sequence, Any, Union
 import logging
 import typing
 
-import numpy as np
-from numpy.typing import ArrayLike
-
 from qiskit.circuit import QuantumCircuit
 from qiskit.quantum_info.operators.base_operator import BaseOperator
 from qiskit.primitives import BaseEstimator
-from qiskit.quantum_info import SparsePauliOp, Pauli
-from qiskit.primitives.utils import init_observable
-from qiskit.circuit import Parameter
-from qiskit.primitives.base.base_primitive import _isreal
 
 from .runtime_job import RuntimeJob
 from .ibm_backend import IBMBackend
 from .options import Options
 from .options.estimator_options import EstimatorOptions
 from .base_primitive import BasePrimitiveV1, BasePrimitiveV2
 from .utils.qctrl import validate as qctrl_validate
-from .utils.deprecation import issue_deprecation_msg
 
 # TODO: remove when we have real v2 base estimator
 from .qiskit.primitives import BaseEstimatorV2
@@ -49,23 +41,6 @@
 logger = logging.getLogger(__name__)
 
 
-BasisObservableLike = Union[str, Pauli, SparsePauliOp, Mapping[Union[str, Pauli], complex]]
-"""Types that can be natively used to construct a :const:`BasisObservable`."""
-
-ObservablesArrayLike = Union[ArrayLike, Sequence[BasisObservableLike], BasisObservableLike]
-
-ParameterMappingLike = Mapping[
-    Parameter, Union[float, np.ndarray, Sequence[float], Sequence[Sequence[float]]]
-]
-BindingsArrayLike = Union[
-    float,
-    np.ndarray,
-    ParameterMappingLike,
-    Sequence[Union[float, Sequence[float], np.ndarray, ParameterMappingLike]],
-]
-"""Parameter types that can be bound to a single circuit."""
-
-
 class Estimator:
     """Base class for Qiskit Runtime Estimator."""
 
@@ -118,8 +93,7 @@ class EstimatorV2(BasePrimitiveV2, Estimator, BaseEstimatorV2):
             print(psi1_H23.result())
     """
 
-    _ALLOWED_BASIS: str = "IXYZ01+-rl"
-    _OPTIONS_CLASS = EstimatorOptions
+    _options_class = EstimatorOptions
 
     version = 2
 
@@ -149,101 +123,20 @@ def __init__(
         Raises:
             NotImplementedError: If "q-ctrl" channel strategy is used.
         """
-        self.options: EstimatorOptions
         BaseEstimatorV2.__init__(self)
         Estimator.__init__(self)
         BasePrimitiveV2.__init__(self, backend=backend, session=session, options=options)
 
         if self._service._channel_strategy == "q-ctrl":
             raise NotImplementedError("EstimatorV2 is not supported with q-ctrl channel strategy.")
 
-    def run(  # pylint: disable=arguments-differ
-        self,
-        circuits: QuantumCircuit | Sequence[QuantumCircuit],
-        observables: Sequence[ObservablesArrayLike]
-        | ObservablesArrayLike
-        | Sequence[BaseOperator]
-        | BaseOperator,
-        parameter_values: BindingsArrayLike | Sequence[BindingsArrayLike] | None = None,
-        **kwargs: Any,
-    ) -> RuntimeJob:
-        """Submit a request to the estimator primitive.
-
-        Args:
-            circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
-                a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
-
-            observables: Observable objects.
-
-            parameter_values: Concrete parameters to be bound.
-
-            **kwargs: Individual options to overwrite the default primitive options.
-                These include the runtime options in :class:`qiskit_ibm_runtime.RuntimeOptions`.
-
-        Returns:
-            Submitted job.
-            The result of the job is an instance of :class:`qiskit.primitives.EstimatorResult`.
-
-        Raises:
-            ValueError: Invalid arguments are given.
-        """
-        # To bypass base class merging of options.
-        user_kwargs = {"_user_kwargs": kwargs}
-
-        return super().run(
-            circuits=circuits,
-            observables=observables,
-            parameter_values=parameter_values,
-            **user_kwargs,
-        )
-
-    def _run(  # pylint: disable=arguments-differ
-        self,
-        circuits: Sequence[QuantumCircuit],
-        observables: Sequence[ObservablesArrayLike],
-        parameter_values: Sequence[Sequence[float]],
-        **kwargs: Any,
-    ) -> RuntimeJob:
-        """Submit a request to the estimator primitive.
-
-        Args:
-            circuits: a (parameterized) :class:`~qiskit.circuit.QuantumCircuit` or
-                a list of (parameterized) :class:`~qiskit.circuit.QuantumCircuit`.
-
-            observables: A list of observable objects.
-
-            parameter_values: An optional list of concrete parameters to be bound.
-
-            **kwargs: Individual options to overwrite the default primitive options.
-                These include the runtime options in :class:`~qiskit_ibm_runtime.RuntimeOptions`.
-
-        Returns:
-            Submitted job
-        """
-        logger.debug(
-            "Running %s with new options %s",
-            self.__class__.__name__,
-            kwargs.get("_user_kwargs", {}),
-        )
-        inputs = {
-            "circuits": circuits,
-            "observables": observables,
-            "parameters": [circ.parameters for circ in circuits],
-            "parameter_values": parameter_values,
-        }
-        return self._run_primitive(
-            primitive_inputs=inputs, user_kwargs=kwargs.get("_user_kwargs", {})
-        )
-
     def _validate_options(self, options: dict) -> None:
         """Validate that program inputs (options) are valid
 
         Raises:
             ValidationError: if validation fails.
             ValueError: if validation fails.
         """
-        self._OPTIONS_CLASS(**options)
-
         # TODO: Server should have different optimization/resilience levels for simulator
 
         if (
@@ -257,83 +150,6 @@ def _validate_options(self, options: dict) -> None:
                 "a coupling map is required."
             )
 
-    @staticmethod
-    def _validate_observables(
-        observables: Sequence[ObservablesArrayLike] | ObservablesArrayLike,
-    ) -> Sequence[ObservablesArrayLike]:
-        def _check_and_init(obs: Any) -> Any:
-            if isinstance(obs, str):
-                if not all(basis in EstimatorV2._ALLOWED_BASIS for basis in obs):
-                    raise ValueError(
-                        f"Invalid character(s) found in observable string. "
-                        f"Allowed basis are {EstimatorV2._ALLOWED_BASIS}."
-                    )
-            elif isinstance(obs, Sequence):
-                return tuple(_check_and_init(obs_) for obs_ in obs)
-            elif not isinstance(obs, (Pauli, SparsePauliOp)) and isinstance(obs, BaseOperator):
-                issue_deprecation_msg(
-                    msg="Only Pauli and SparsePauliOp operators can be used as observables",
-                    version="0.13",
-                    remedy="",
-                )
-                return init_observable(obs)
-            elif isinstance(obs, Mapping):
-                for key in obs.keys():
-                    _check_and_init(key)
-
-            return obs
-
-        if isinstance(observables, str) or not isinstance(observables, Sequence):
-            observables = (observables,)
-
-        if len(observables) == 0:
-            raise ValueError("No observables were provided.")
-
-        return tuple(_check_and_init(obs_array) for obs_array in observables)
-
-    @staticmethod
-    def _validate_parameter_values(
-        parameter_values: BindingsArrayLike | Sequence[BindingsArrayLike] | None,
-        default: Sequence[Sequence[float]] | Sequence[float] | None = None,
-    ) -> Sequence:
-
-        # Allow optional (if default)
-        if parameter_values is None:
-            if default is None:
-                raise ValueError("No default `parameter_values`, optional input disallowed.")
-            parameter_values = default
-
-        # Convert single input types to length-1 lists
-        if _isreal(parameter_values):
-            parameter_values = [[parameter_values]]
-        elif isinstance(parameter_values, Mapping):
-            parameter_values = [parameter_values]
-        elif isinstance(parameter_values, Sequence) and all(
-            _isreal(item) for item in parameter_values
-        ):
-            parameter_values = [parameter_values]
-        return tuple(parameter_values)  # type: ignore[arg-type]
-
-    @staticmethod
-    def _cross_validate_circuits_parameter_values(
-        circuits: tuple[QuantumCircuit, ...], parameter_values: tuple[tuple[float, ...], ...]
-    ) -> None:
-        if len(circuits) != len(parameter_values):
-            raise ValueError(
-                f"The number of circuits ({len(circuits)}) does not match "
-                f"the number of parameter value sets ({len(parameter_values)})."
-            )
-
-    @staticmethod
-    def _cross_validate_circuits_observables(
-        circuits: tuple[QuantumCircuit, ...], observables: tuple[ObservablesArrayLike, ...]
-    ) -> None:
-        if len(circuits) != len(observables):
-            raise ValueError(
-                f"The number of circuits ({len(circuits)}) does not match "
-                f"the number of observables ({len(observables)})."
-            )
-
     @classmethod
     def _program_id(cls) -> str:
         """Return the program ID."""

qiskit_ibm_runtime/options/__init__.py

@@ -57,7 +57,7 @@
 
 from .environment_options import EnvironmentOptions
 from .execution_options import ExecutionOptionsV1 as ExecutionOptions
-from .options import Options, BaseOptions
+from .options import Options
 from .simulator_options import SimulatorOptions
 from .transpilation_options import TranspilationOptions
 from .resilience_options import ResilienceOptionsV1 as ResilienceOptions