Add optimizers to qiboml

src/qiboml/models/__init__.py

@@ -0,0 +1 @@
+from qiboml.models.pqc import PQC

src/qiboml/models/encodings.py

@@ -0,0 +1,22 @@
+"""Strategies for data encoding into a Parametrized Quantum Circuit."""
+
+from qibo import Circuit
+
+
+class EncodingCircuit:
+    """
+    An encoding circuit is a quantum circuit with a data encoding strategy.
+
+    Args:
+        circuit (Circuit): a Qibo circuit.
+        encoding_strategy (callable): a callable function which defines the encoding
+            strategy of the data inside the circuit.
+    """
+
+    def __init__(self, circuit: Circuit, encoding_strategy: callable):
+        self.circuit = circuit
+        self.encoding_strategy = encoding_strategy
+
+    def inject_data(self, data):
+        """Encode the data into ``circuit`` according to the chosen encoding strategy."""
+        return self.encoding_strategy(self.circuit, data)

src/qiboml/models/pqc.py

@@ -0,0 +1,160 @@
+"""Parametric Quantum Circuit"""
+
+from typing import Dict, List, Optional, Union
+
+from numpy import array, ndarray
+from qibo import Circuit
+from qibo.config import raise_error
+from qibo.gates import Gate
+from qibo.hamiltonians import Hamiltonian
+
+from qiboml.models.encodings import EncodingCircuit
+from qiboml.optimizers import Optimizer
+
+
+class PQC(Circuit):
+    """Parametric Quantum Circuit built on top of ``qibo.Circuit``."""
+
+    def __init__(
+        self,
+        nqubits: int,
+        accelerators: Optional[Dict] = None,
+        density_matrix: bool = False,
+        wire_names: Optional[Union[list, dict]] = None,
+    ):
+        super().__init__(nqubits, accelerators, density_matrix, wire_names)
+
+        self.parameters = []
+        self.nparams = 0
+
+    def add(self, gate: Gate):
+        """
+        Add a gate to the PQC.
+
+        Args:
+            gate (qibo.Gate): Qibo gate to be added to the PQC.
+        """
+        super().add(gate)
+        if len(gate.parameters) != 0:
+            self.parameters.extend(gate.parameters)
+            self.nparams += gate.nparams
+
+    def set_parameters(self, parameters: Union[List, ndarray]):
+        """
+        Set model parameters.
+
+        Args:
+            parameters (Union[List, ndarray]): new set of parameters to be set
+                into the PQC.
+        """
+        self.parameters = parameters
+        super().set_parameters(parameters)
+
+    def setup(
+        self,
+        optimizer: Optimizer,
+        loss: callable,
+        observable: Union[Hamiltonian, List[Hamiltonian]],
+        encoding_config: EncodingCircuit,
+    ):
+        """
+        Compile the PQC to perform a training.
+
+        Args:
+            optimizer (qiboml.optimizers.Optimizer): optimizer to be used.
+            loss (callable): loss function to be minimizer.
+            observable (qibo.hamiltonians.Hamiltonian): observable, or list of
+                observables, whose expectation value is used to compute predictions.
+            encoding_config (qiboml.models.EncodingCircuit): encoding circuit, which
+                is a Qibo circuit defined together with an encoding strategy.
+
+        """
+        self.optimizer = optimizer
+        self.loss = loss
+        self.observable = observable
+        self.encoding_circuit = encoding_config
+        self._compiled = True
+
+    def fit(
+        self,
+        input_data: ndarray,
+        output_data: ndarray,
+        nshots: Optional[int] = None,
+        options: Optional[Dict] = None,
+    ):
+        """
+        Perform the PQC training according to the chosen trainig setup.
+
+        Args:
+            input_data (np.ndarray): input data to train on.
+            output_data (np.ndarray): output data used as labels in the training process.
+            nshots (Optional[int]): number of shots for circuit evaluations.
+            options (Optional[Dict]): extra fit options eventually needed by the
+                chosen optimizer.
+        """
+
+        if not self._compiled:
+            raise_error(
+                ValueError,
+                "Please compile the model through the `PQC.setup` method to train it.",
+            )
+
+        if options is None:
+            fit_options = {}
+        else:
+            fit_options = options
+
+        def _loss(parameters, input_data, output_data):
+            self.set_parameters(parameters)
+
+            predictions = []
+            for x in input_data:
+                predictions.append(self.predict(input_datum=x, nshots=nshots))
+            loss_value = self.loss(predictions, output_data)
+            return loss_value
+
+        results = self.optimizer.fit(
+            initial_parameters=self.parameters,
+            loss=_loss,
+            args=(input_data, output_data),
+            **fit_options
+        )
+
+        return results
+
+    def predict(self, input_datum: Union[array, List, tuple], nshots: int = None):
+        """
+        Perform prediction associated to a single ``input_datum``.
+
+        Args:
+            input_datum (Union[array, List, tuple]): one single element of the
+                input dataset.
+            nshots (int): number of circuit execution to compute the prediction.
+        """
+
+        if not self._compiled:
+            raise_error(
+                ValueError,
+                "Please compile the model through the `PQC.compile` method to perform predictions.",
+            )
+
+        encoding_state = self.encoding_circuit.inject_data(input_datum)().state()
+        return self.observable.expectation(
+            self(initial_state=encoding_state, nshots=nshots).state()
+        )
+
+    def predict_sample(self, input_data: ndarray, nshots: int = None):
+        """
+        Compute predictions for a set of data ``input_data``.
+
+        Args:
+            input_data (np.ndarray): input data.
+            nshots (int): number of times the circuit is executed to compute the
+                predictions.
+        """
+
+        predictions = []
+        for x in input_data:
+            predictions.append(self.predict(input_datum=x, nshots=nshots))
+
+        return predictions

src/qiboml/models/reuploading/u3.py

@@ -14,7 +14,7 @@ def __init__(
         nlayers: int,
         data_dimensionality: Tuple,
         actf1: Callable = lambda x: x,
-        actf2: Callable = lambda x: log(x),
+        actf2: Callable = lambda x: log(np.abs(x) + 1e-5),
         actf3: Callable = lambda x: exp(x),
     ):
         """Reuplading U3 ansatz."""
@@ -46,16 +46,17 @@ def build_circuit(self):
     def inject_data(self, x):
         new_parameters = []
         k = 0
+
         for _ in range(self.nlayers):
-            for _ in range(self.nqubits):
+            for q in range(self.nqubits):
                 new_parameters.append(
-                    self.parameters[k] * self.actf1(x) + self.parameters[k + 1]
+                    self.parameters[k] * self.actf1(x[q]) + self.parameters[k + 1]
                 )
                 new_parameters.append(
-                    self.parameters[k + 2] * self.actf2(x) + self.parameters[k + 3]
+                    self.parameters[k + 2] * self.actf2(x[q]) + self.parameters[k + 3]
                 )
                 new_parameters.append(
-                    self.parameters[k + 4] * self.actf3(x) + self.parameters[k + 5]
+                    self.parameters[k + 4] * self.actf3(x[q]) + self.parameters[k + 5]
                 )
             k += 6
         self.circuit.set_parameters(new_parameters)

src/qiboml/optimizers/__init__.py

@@ -0,0 +1 @@
+from qiboml.optimizers.abstract import Optimizer

src/qiboml/optimizers/abstract.py

@@ -0,0 +1,16 @@
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+
+from qibo.config import raise_error
+
+
+@dataclass
+class Optimizer(ABC):
+
+    verbosity: bool = field(default=True)
+    """Verbosity of the optimization process. If True, logging messages will be displayed."""
+
+    @abstractmethod
+    def fit(self):
+        """Compute the optimization strategy."""
+        raise_error(NotImplementedError)

src/qiboml/optimizers/gradient_based.py

@@ -0,0 +1,113 @@
+"""Gradient descent strategies to optimize quantum models."""
+
+from typing import List, Optional, Tuple, Union
+
+from numpy import ndarray
+from qibo.backends import construct_backend
+from qibo.config import log
+
+from qiboml.optimizers.abstract import Optimizer
+
+
+class TensorflowSGD(Optimizer):
+    """
+    Stochastic Gradient Descent (SGD) optimizer using Tensorflow backpropagation.
+    See `tf.keras.Optimizers https://www.tensorflow.org/api_docs/python/tf/keras/optimizers.
+    for a list of the available optimizers.
+
+    Args:
+        optimizer_name (str): `tensorflow.keras.optimizer`, see
+            https://www.tensorflow.org/api_docs/python/tf/keras/optimizers
+            for the list of available optimizers.
+        compile (bool):  if ``True`` the Tensorflow optimization graph is compiled.
+        **optimizer_options (dict): a dictionary containing the keywords arguments
+            to customize the selected keras optimizer. In order to properly
+            customize your optimizer please refer to https://www.tensorflow.org/api_docs/python/tf/keras/optimizers.
+    """
+
+    def __init__(
+        self, optimizer_name: str = "Adagrad", compile: bool = True, **optimizer_options
+    ):
+
+        self.optimizer_name = optimizer_name
+        self.compile = compile
+
+        if optimizer_options is None:
+            options = {}
+        else:
+            options = optimizer_options
+
+        self.backend = construct_backend("tensorflow")
+        self.optimizer = getattr(
+            self.backend.tf.optimizers.legacy, self.optimizer_name
+        )(**options)
+
+    def __str__(self):
+        return f"tensorflow_{self.optimizer_name}"
+
+    def fit(
+        self,
+        initial_parameters: Union[List, ndarray],
+        loss: callable,
+        args: Union[Tuple] = None,
+        epochs: int = 10000,
+        nmessage: int = 100,
+        loss_threshold: Optional[float] = None,
+    ):
+        """
+        Compute the SGD optimization according to the chosen optimizer.
+
+        Args:
+            initial_parameters (np.ndarray or list): array with initial values
+                for gate parameters.
+            loss (callable): loss function to train on.
+            args (tuple): tuple containing loss function arguments.
+            epochs (int): number of optimization iterations [default 10000].
+            nmessage (int): Every how many epochs to print
+                a message of the loss function [default 100].
+            loss_threshold (float): if this loss function value is reached, training
+                stops [default None].
+
+        Returns:
+            (float): best loss value
+            (np.ndarray): best parameter values
+            (list): loss function history
+        """
+
+        vparams = self.backend.tf.Variable(
+            initial_parameters, dtype=self.backend.tf.float64
+        )
+        print(vparams)
+        loss_history = []
+
+        def sgd_step():
+            """Compute one SGD optimization step according to the chosen optimizer."""
+            with self.backend.tf.GradientTape() as tape:
+                tape.watch(vparams)
+                loss_value = loss(vparams, *args)
+
+            grads = tape.gradient(loss_value, [vparams])
+            self.optimizer.apply_gradients(zip(grads, [vparams]))
+            return loss_value
+
+        if self.compile:
+            self.backend.compile(loss)
+            self.backend.compile(sgd_step)
+
+        # SGD procedure: loop over epochs
+        for epoch in range(epochs):  # pragma: no cover
+            # early stopping if loss_threshold has been set
+            if (
+                loss_threshold is not None
+                and (epoch != 0)
+                and (loss_history[-1] <= loss_threshold)
+            ):
+                break
+
+            loss_value = sgd_step().numpy()
+            loss_history.append(loss_value)
+
+            if epoch % nmessage == 0:
+                log.info("ite %d : loss %f", epoch, loss_value)
+
+        return loss(vparams, *args).numpy(), vparams.numpy(), loss_history