qiskit_machine_learning.algorithms.distribution_learners.qgan.qgan का स्रोत निर्देश

# This code is part of Qiskit.
# (C) Copyright IBM 2019, 2022.
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.

"""Quantum Generative Adversarial Network."""

from typing import Optional, Union, List, Dict, Any, Callable
from types import FunctionType
import csv
import os
import logging

import numpy as np
from scipy.stats import entropy

from qiskit.circuit import QuantumCircuit
from qiskit.providers import Backend
from qiskit.utils import QuantumInstance, algorithm_globals
from qiskit.algorithms.optimizers import Optimizer
from qiskit.opflow.gradients import Gradient
from qiskit.utils.validation import validate_min
from ....datasets.dataset_helper import discretize_and_truncate
from ....deprecation import deprecate_function
from ....exceptions import QiskitMachineLearningError
from .discriminative_network import DiscriminativeNetwork
from .generative_network import GenerativeNetwork
from .quantum_generator import QuantumGenerator
from .numpy_discriminator import NumPyDiscriminator

logger = logging.getLogger(__name__)

# pylint: disable=invalid-name

[docs]class QGAN: """The Quantum Generative Adversarial Network algorithm. The qGAN [1] is a hybrid quantum-classical algorithm used for generative modeling tasks. This adaptive algorithm uses the interplay of a generative :class:`~qiskit_machine_learning.neural_networks.GenerativeNetwork` and a discriminative :class:`~qiskit_machine_learning.neural_networks.DiscriminativeNetwork` network to learn the probability distribution underlying given training data. These networks are trained in alternating optimization steps, where the discriminator tries to differentiate between training data samples and data samples from the generator and the generator aims at generating samples which the discriminator classifies as training data samples. Eventually, the quantum generator learns the training data's underlying probability distribution. The trained quantum generator loads a quantum state which is a model of the target distribution. **References:** [1] Zoufal et al., `Quantum Generative Adversarial Networks for learning and loading random distributions <https://www.nature.com/articles/s41534-019-0223-2>`_ """ @deprecate_function( "0.5.0", additional_msg="with no direct replacement for it. " "Instead, please refer to the new QGAN tutorial", stack_level=3, ) def __init__( self, data: Union[np.ndarray, List], bounds: Optional[Union[np.ndarray, List]] = None, num_qubits: Optional[Union[np.ndarray, List]] = None, batch_size: int = 500, num_epochs: int = 3000, seed: int = 7, discriminator: Optional[DiscriminativeNetwork] = None, generator: Optional[GenerativeNetwork] = None, tol_rel_ent: Optional[float] = None, snapshot_dir: Optional[str] = None, quantum_instance: Optional[Union[QuantumInstance, Backend]] = None, penalty: Optional[bool] = False, ) -> None: """ Args: data: Training data of dimension k bounds: k min/max data values [[min_0,max_0],...,[min_k-1,max_k-1]] if univariate data: [min_0,max_0] num_qubits: k numbers of qubits to determine representation resolution, i.e. n qubits enable the representation of 2**n values [num_qubits_0,..., num_qubits_k-1] batch_size: Batch size, has a min. value of 1. num_epochs: Number of training epochs seed: Random number seed discriminator: Discriminates between real and fake data samples generator: Generates 'fake' data samples tol_rel_ent: Set tolerance level for relative entropy. If the training achieves relative entropy equal or lower than tolerance it finishes. snapshot_dir: Directory in to which to store cvs file with parameters, if None (default) then no cvs file is created. quantum_instance: Quantum Instance or Backend penalty : enable or not the gradient penalty in the discriminator Raises: QiskitMachineLearningError: invalid input """ validate_min("batch_size", batch_size, 1) self._quantum_instance = None if quantum_instance: self.quantum_instance = quantum_instance if data is None: raise QiskitMachineLearningError("Training data not given.") self._data = np.array(data) if bounds is None: bounds_min = np.percentile(self._data, 5, axis=0) bounds_max = np.percentile(self._data, 95, axis=0) bounds = [] for i, bound_min in enumerate(bounds_min): bounds.append([bound_min, bounds_max[i]]) if np.ndim(data) > 1: if len(bounds) != (len(num_qubits) or len(data[0])): raise QiskitMachineLearningError( "Dimensions of the data, the length of the data bounds " "and the numbers of qubits per " "dimension are incompatible." ) else: if (np.ndim(bounds) or len(num_qubits)) != 1: raise QiskitMachineLearningError( "Dimensions of the data, the length of the data bounds " "and the numbers of qubits per " "dimension are incompatible." ) self._bounds = np.array(bounds) self._num_qubits = num_qubits # pylint: disable=unsubscriptable-object if np.ndim(data) > 1: if self._num_qubits is None: self._num_qubits = np.ones[len(data[0])] * 3 # type: ignore else: if self._num_qubits is None: self._num_qubits = np.array([3]) ( self._data, self._data_grid, self._grid_elements, self._prob_data, ) = discretize_and_truncate( self._data, self._bounds, self._num_qubits, return_data_grid_elements=True, return_prob=True, prob_non_zero=True, ) self._batch_size = batch_size self._num_epochs = num_epochs self._snapshot_dir = snapshot_dir self._g_loss = [] # type: List[float] self._d_loss = [] # type: List[float] self._rel_entr = [] # type: List[float] self._tol_rel_ent = tol_rel_ent self._penalty = penalty self._random_seed = seed if generator is None: self.set_generator() else: self._generator = generator if discriminator is None: self.set_discriminator() else: self._discriminator = discriminator self.seed = self._random_seed self._ret = {} # type: Dict[str, Any] @property def random(self): """Return a numpy random.""" return algorithm_globals.random
[docs] def run( self, quantum_instance: Optional[Union[QuantumInstance, Backend]] = None, **kwargs, ) -> Dict: """Execute the algorithm with selected backend. Args: quantum_instance: the experimental setting. **kwargs (dict): kwargs Returns: dict: results of an algorithm. Raises: QiskitMachineLearningError: If a quantum instance or backend has not been provided """ if quantum_instance is None and self.quantum_instance is None: raise QiskitMachineLearningError( "A QuantumInstance or Backend must be supplied to run the quantum algorithm." ) if isinstance(quantum_instance, Backend): self.set_backend(quantum_instance, **kwargs) else: if quantum_instance is not None: self.quantum_instance = quantum_instance return self._run()
@property def quantum_instance(self) -> Optional[QuantumInstance]: """Returns quantum instance.""" return self._quantum_instance @quantum_instance.setter def quantum_instance(self, quantum_instance: Union[QuantumInstance, Backend]) -> None: """Sets quantum instance.""" if isinstance(quantum_instance, Backend): quantum_instance = QuantumInstance(quantum_instance) self._quantum_instance = quantum_instance
[docs] def set_backend(self, backend: Backend, **kwargs) -> None: """Sets backend with configuration.""" self.quantum_instance = QuantumInstance(backend) self.quantum_instance.set_config(**kwargs)
@property def backend(self) -> Backend: """Returns backend.""" return self.quantum_instance.backend @backend.setter def backend(self, backend: Backend): """Sets backend without additional configuration.""" self.set_backend(backend) @property def seed(self): """Returns random seed""" return self._random_seed @seed.setter def seed(self, s): """ Sets the random seed for QGAN and updates the algorithm_globals seed at the same time Args: s (int): random seed """ self._random_seed = s algorithm_globals.random_seed = self._random_seed self._discriminator.set_seed(self._random_seed) @property def tol_rel_ent(self): """Returns tolerance for relative entropy""" return self._tol_rel_ent @tol_rel_ent.setter def tol_rel_ent(self, t): """ Set tolerance for relative entropy Args: t (float): or None, Set tolerance level for relative entropy. If the training achieves relative entropy equal or lower than tolerance it finishes. """ self._tol_rel_ent = t @property def generator(self): """Returns generator""" return self._generator @property def penalty(self): """Returns penalty parameter for the discriminator training""" return self._penalty # pylint: disable=unused-argument
[docs] def set_generator( self, generator_circuit: Optional[QuantumCircuit] = None, generator_init_params: Optional[np.ndarray] = None, generator_optimizer: Optional[Optimizer] = None, generator_gradient: Optional[Union[Callable, Gradient]] = None, ): """Initialize generator. Args: generator_circuit: parameterized quantum circuit which sets the structure of the quantum generator generator_init_params: initial parameters for the generator circuit generator_optimizer: optimizer to be used for the training of the generator generator_gradient: A Gradient object, or a function returning partial derivatives of the loss function w.r.t. the generator variational params. Raises: QiskitMachineLearningError: invalid input """ if generator_gradient: if not isinstance(generator_gradient, (Gradient, FunctionType)): raise QiskitMachineLearningError( "Please pass either a Gradient object or a function as " "the generator_gradient argument." ) self._generator = QuantumGenerator( self._bounds, self._num_qubits, generator_circuit, generator_init_params, generator_optimizer, generator_gradient, self._snapshot_dir, )
@property def discriminator(self): """Returns discriminator""" return self._discriminator
[docs] def set_discriminator(self, discriminator=None): """ Initialize discriminator. Args: discriminator (Discriminator): discriminator """ if discriminator is None: self._discriminator = NumPyDiscriminator(len(self._num_qubits)) else: self._discriminator = discriminator self._discriminator.set_seed(self._random_seed)
@property def g_loss(self) -> List[float]: """Returns generator loss""" return self._g_loss @property def d_loss(self) -> List[float]: """Returns discriminator loss""" return self._d_loss @property def rel_entr(self) -> List[float]: """Returns relative entropy between target and trained distribution""" return self._rel_entr
[docs] def get_rel_entr(self) -> float: """Get relative entropy between target and trained distribution""" samples_gen, prob_gen = self._generator.get_output(self._quantum_instance) temp = np.zeros(len(self._grid_elements)) for j, sample in enumerate(samples_gen): for i, element in enumerate(self._grid_elements): if sample == element: temp[i] += prob_gen[j] prob_gen = [1e-8 if x == 0 else x for x in temp] rel_entr = entropy(prob_gen, self._prob_data) return rel_entr
def _store_params(self, e, d_loss, g_loss, rel_entr): with open( os.path.join(self._snapshot_dir, "output.csv"), mode="a", encoding="utf8" ) as csv_file: fieldnames = [ "epoch", "loss_discriminator", "loss_generator", "params_generator", "rel_entropy", ] writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writerow( { "epoch": e, "loss_discriminator": np.average(d_loss), "loss_generator": np.average(g_loss), "params_generator": self._generator.parameter_values, "rel_entropy": rel_entr, } ) self._discriminator.save_model(self._snapshot_dir) # Store discriminator model
[docs] def train(self): """ Train the qGAN Raises: QiskitMachineLearningError: Batch size bigger than the number of items in the truncated data set """ if self._snapshot_dir is not None: with open( os.path.join(self._snapshot_dir, "output.csv"), mode="w", encoding="utf8" ) as csv_file: fieldnames = [ "epoch", "loss_discriminator", "loss_generator", "params_generator", "rel_entropy", ] writer = csv.DictWriter(csv_file, fieldnames=fieldnames) writer.writeheader() if len(self._data) < self._batch_size: raise QiskitMachineLearningError( "The batch size needs to be less than the " f"truncated data size of {len(self._data)}" ) for e in range(self._num_epochs): algorithm_globals.random.shuffle(self._data) index = 0 while (index + self._batch_size) <= len(self._data): real_batch = self._data[index : index + self._batch_size] index += self._batch_size generated_batch, generated_prob = self._generator.get_output( self._quantum_instance, shots=self._batch_size ) # 1. Train Discriminator ret_d = self._discriminator.train( [real_batch, generated_batch], [np.ones(len(real_batch)) / len(real_batch), generated_prob], penalty=self._penalty, ) d_loss_min = ret_d["loss"] # 2. Train Generator self._generator.discriminator = self._discriminator ret_g = self._generator.train(self._quantum_instance, shots=self._batch_size) g_loss_min = ret_g["loss"] self._d_loss.append(np.around(float(d_loss_min), 4)) self._g_loss.append(np.around(g_loss_min, 4)) rel_entr = self.get_rel_entr() self._rel_entr.append(np.around(rel_entr, 4)) self._ret["params_d"] = ret_d["params"] self._ret["params_g"] = ret_g["params"] self._ret["loss_d"] = np.around(float(d_loss_min), 4) self._ret["loss_g"] = np.around(g_loss_min, 4) self._ret["rel_entr"] = np.around(rel_entr, 4) if self._snapshot_dir is not None: self._store_params( e, np.around(d_loss_min, 4), np.around(g_loss_min, 4), np.around(rel_entr, 4), ) logger.debug("Epoch %s/%s...", e + 1, self._num_epochs) logger.debug("Loss Discriminator: %s", np.around(float(d_loss_min), 4)) logger.debug("Loss Generator: %s", np.around(g_loss_min, 4)) logger.debug("Relative Entropy: %s", np.around(rel_entr, 4)) if self._tol_rel_ent is not None: if rel_entr <= self._tol_rel_ent: break
def _run(self): """ Run qGAN training Returns: dict: with generator(discriminator) parameters & loss, relative entropy Raises: QiskitMachineLearningError: invalid backend """ if self._quantum_instance.backend_name == ("unitary_simulator" or "clifford_simulator"): raise QiskitMachineLearningError( "Chosen backend not supported - " "Set backend either to statevector_simulator, qasm_simulator" " or actual quantum hardware" ) self.train() return self._ret