Skip to content

Commit

Permalink
Rewrite Amplitude Estimators with Primitives
Browse files Browse the repository at this point in the history
  • Loading branch information
@manoelmarques
manoelmarques committed Sep 2, 2022
1 parent 9ddc97d commit 4008094
Show file tree
Hide file tree
Showing 5 changed files with 361 additions and 69 deletions.
89 changes: 73 additions & 16 deletions qiskit/algorithms/amplitude_estimators/ae.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
# This code is part of Qiskit.
#
# (C) Copyright IBM 2018, 2020.
# (C) Copyright IBM 2018, 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
Expand All @@ -12,18 +12,23 @@

"""The Quantum Phase Estimation-based Amplitude Estimation algorithm."""

from typing import Optional, Union, List, Tuple, Dict
from __future__ import annotations
from typing import Union, List, Tuple, Dict
from collections import OrderedDict
import warnings
import numpy as np
from scipy.stats import chi2, norm
from scipy.optimize import bisect

from qiskit import QuantumCircuit, ClassicalRegister
from qiskit.providers import Backend
from qiskit.providers import Backend, JobStatus
from qiskit.primitives import BaseSampler
from qiskit.utils import QuantumInstance
from qiskit.utils.deprecation import deprecate_function
from .amplitude_estimator import AmplitudeEstimator, AmplitudeEstimatorResult
from .ae_utils import pdf_a, derivative_log_pdf_a, bisect_max
from .estimation_problem import EstimationProblem
from ..exceptions import AlgorithmError


class AmplitudeEstimation(AmplitudeEstimator):
Expand Down Expand Up @@ -56,9 +61,10 @@ class AmplitudeEstimation(AmplitudeEstimator):
def __init__(
self,
num_eval_qubits: int,
phase_estimation_circuit: Optional[QuantumCircuit] = None,
iqft: Optional[QuantumCircuit] = None,
quantum_instance: Optional[Union[QuantumInstance, Backend]] = None,
phase_estimation_circuit: None | QuantumCircuit = None,
iqft: None | QuantumCircuit = None,
quantum_instance: None | Union[QuantumInstance, Backend] = None,
sampler: None | BaseSampler = None,
) -> None:
r"""
Args:
Expand All @@ -68,7 +74,9 @@ def __init__(
`qiskit.circuit.library.PhaseEstimation` when None.
iqft: The inverse quantum Fourier transform component, defaults to using a standard
implementation from `qiskit.circuit.library.QFT` when None.
quantum_instance: The backend (or `QuantumInstance`) to execute the circuits on.
quantum_instance: Pending deprecation\: The backend (or `QuantumInstance`) to execute
the circuits on.
sampler: base sampler
Raises:
ValueError: If the number of evaluation qubits is smaller than 1.
Expand All @@ -79,27 +87,49 @@ def __init__(
super().__init__()

# set quantum instance
self.quantum_instance = quantum_instance
if quantum_instance is not None:
warnings.warn(
"The quantum_instance argument has been superseded by the sampler argument. "
"This argument will be deprecated in a future release and subsequently "
"removed after that.",
category=PendingDeprecationWarning,
)
with warnings.catch_warnings():
warnings.simplefilter("ignore")
self.quantum_instance = quantum_instance

# get parameters
self._m = num_eval_qubits # pylint: disable=invalid-name
self._M = 2**num_eval_qubits # pylint: disable=invalid-name

self._iqft = iqft
self._pec = phase_estimation_circuit
self._sampler = sampler

@property
def quantum_instance(self) -> Optional[QuantumInstance]:
"""Get the quantum instance.
@deprecate_function(
"The AmplitudeEstimation.quantum_instance getter is pending deprecation. "
"This property will be deprecated in a future release and subsequently "
"removed after that.",
category=PendingDeprecationWarning,
)
def quantum_instance(self) -> None | QuantumInstance:
"""Pending deprecation: Get the quantum instance.
Returns:
The quantum instance used to run this algorithm.
"""
return self._quantum_instance

@quantum_instance.setter
@deprecate_function(
"The AmplitudeEstimation.quantum_instance setter is pending deprecation. "
"This property will be deprecated in a future release and subsequently "
"removed after that.",
category=PendingDeprecationWarning,
)
def quantum_instance(self, quantum_instance: Union[QuantumInstance, Backend]) -> None:
"""Set quantum instance.
"""Pending deprecation: Set quantum instance.
Args:
quantum_instance: The quantum instance used to run this algorithm.
Expand Down Expand Up @@ -201,7 +231,9 @@ def _evaluate_count_results(self, counts):
# construct probabilities
measurements = OrderedDict()
samples = OrderedDict()
shots = self._quantum_instance._run_config.shots
shots = (
self._quantum_instance._run_config.shots if self._quantum_instance is not None else 1
)

for state, count in counts.items():
y = int(state.replace(" ", "")[: self._m][::-1], 2)
Expand Down Expand Up @@ -283,12 +315,16 @@ def estimate(self, estimation_problem: EstimationProblem) -> "AmplitudeEstimatio
Raises:
ValueError: If `state_preparation` or `objective_qubits` are not set in the
`estimation_problem`.
ValueError: A quantum instance or Sampler must be provided.
AlgorithmError: Sampler run error
"""
# check if A factory or state_preparation has been set
if estimation_problem.state_preparation is None:
raise ValueError(
"The state_preparation property of the estimation problem must be set."
)
if self._quantum_instance is None and self._sampler is None:
raise ValueError("A quantum instance or sampler must be provided.")

if estimation_problem.objective_qubits is None:
raise ValueError("The objective_qubits property of the estimation problem must be set.")
Expand All @@ -297,15 +333,37 @@ def estimate(self, estimation_problem: EstimationProblem) -> "AmplitudeEstimatio
result.num_evaluation_qubits = self._m
result.post_processing = estimation_problem.post_processing

if self._quantum_instance.is_statevector:
measurements = OrderedDict()
samples = OrderedDict()
if self._sampler is not None:
circuit = self.construct_circuit(estimation_problem, measurement=True)
job = self._sampler.run([circuit])
if job.status() is not JobStatus.DONE:
raise AlgorithmError("The job was not completed successfully. ")
ret = job.result()
result.shots = 1
circuit_results = {
np.binary_repr(k, circuit.num_qubits): v for k, v in ret.quasi_dists[0].items()
}
for state, probability in circuit_results.items():
state = state[::-1]
y = int(state[: self._m], 2)
measurements[y] = probability
a = np.round(np.power(np.sin(y * np.pi / 2**self._m), 2), decimals=7)
samples[a] = samples.get(a, 0.0) + probability
# cutoff probabilities below the threshold
threshold: float = 1e-6
samples = {a: p for a, p in samples.items() if p > threshold}
measurements = {y: p for y, p in measurements.items() if p > threshold}
elif self._quantum_instance.is_statevector:
circuit = self.construct_circuit(estimation_problem, measurement=False)
# run circuit on statevector simulator
statevector = self._quantum_instance.execute(circuit).get_statevector()
result.circuit_results = statevector

# store number of shots: convention is 1 shot for statevector,
# needed so that MLE works!
result.shots = 1
samples, measurements = self.evaluate_measurements(result.circuit_results)
else:
# run circuit on QASM simulator
circuit = self.construct_circuit(estimation_problem, measurement=True)
Expand All @@ -314,8 +372,7 @@ def estimate(self, estimation_problem: EstimationProblem) -> "AmplitudeEstimatio

# store shots
result.shots = sum(counts.values())

samples, measurements = self.evaluate_measurements(result.circuit_results)
samples, measurements = self.evaluate_measurements(result.circuit_results)

result.samples = samples
result.samples_processed = {
Expand Down
101 changes: 84 additions & 17 deletions qiskit/algorithms/amplitude_estimators/fae.py
View file
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
# This code is part of Qiskit.
#
# (C) Copyright IBM 2017, 2020.
# (C) Copyright IBM 2017, 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
Expand All @@ -12,12 +12,16 @@

"""Faster Amplitude Estimation."""

from typing import Optional, Union, List, Tuple
from __future__ import annotations
from typing import Union, List, Tuple
import warnings
import numpy as np

from qiskit.circuit import QuantumCircuit, ClassicalRegister
from qiskit.providers import Backend
from qiskit.providers import Backend, JobStatus
from qiskit.primitives import BaseSampler
from qiskit.utils import QuantumInstance
from qiskit.utils.deprecation import deprecate_function
from qiskit.algorithms.exceptions import AlgorithmError

from .amplitude_estimator import AmplitudeEstimator, AmplitudeEstimatorResult
Expand Down Expand Up @@ -50,14 +54,17 @@ def __init__(
delta: float,
maxiter: int,
rescale: bool = True,
quantum_instance: Optional[Union[QuantumInstance, Backend]] = None,
quantum_instance: None | Union[QuantumInstance, Backend] = None,
sampler: None | BaseSampler = None,
) -> None:
r"""
Args:
delta: The probability that the true value is outside of the final confidence interval.
maxiter: The number of iterations, the maximal power of Q is `2 ** (maxiter - 1)`.
rescale: Whether to rescale the problem passed to `estimate`.
quantum_instance: The quantum instance or backend to run the circuits.
quantum_instance: Pending deprecation\: The quantum instance or backend
to run the circuits.
sampler: base sampler
.. note::
Expand All @@ -66,25 +73,48 @@ def __init__(
"""
super().__init__()
self.quantum_instance = quantum_instance
# set quantum instance
if quantum_instance is not None:
warnings.warn(
"The quantum_instance argument has been superseded by the sampler argument. "
"This argument will be deprecated in a future release and subsequently "
"removed after that.",
category=PendingDeprecationWarning,
)
with warnings.catch_warnings():
warnings.simplefilter("ignore")
self.quantum_instance = quantum_instance
self._shots = (int(1944 * np.log(2 / delta)), int(972 * np.log(2 / delta)))
self._rescale = rescale
self._delta = delta
self._maxiter = maxiter
self._num_oracle_calls = 0
self._sampler = sampler

@property
def quantum_instance(self) -> Optional[QuantumInstance]:
"""Get the quantum instance.
@deprecate_function(
"The FasterAmplitudeEstimation.quantum_instance getter is pending deprecation. "
"This property will be deprecated in a future release and subsequently "
"removed after that.",
category=PendingDeprecationWarning,
)
def quantum_instance(self) -> None | QuantumInstance:
"""Pending deprecation: Get the quantum instance.
Returns:
The quantum instance used to run this algorithm.
"""
return self._quantum_instance

@quantum_instance.setter
@deprecate_function(
"The FasterAmplitudeEstimation.quantum_instance setter is pending deprecation. "
"This property will be deprecated in a future release and subsequently "
"removed after that.",
category=PendingDeprecationWarning,
)
def quantum_instance(self, quantum_instance: Union[QuantumInstance, Backend]) -> None:
"""Set quantum instance.
"""Pending deprecation: Set quantum instance.
Args:
quantum_instance: The quantum instance used to run this algorithm.
Expand All @@ -94,10 +124,30 @@ def quantum_instance(self, quantum_instance: Union[QuantumInstance, Backend]) ->
self._quantum_instance = quantum_instance

def _cos_estimate(self, estimation_problem, k, shots):
if self._quantum_instance is None:
raise AlgorithmError("Quantum instance must be set.")
if self._quantum_instance is None and self._sampler is None:
raise ValueError("A quantum instance or sampler must be provided.")

if self._quantum_instance.is_statevector:
if self._sampler is not None:
circuit = self.construct_circuit(estimation_problem, k, measurement=True)
job = self._sampler.run([circuit])
if job.status() is not JobStatus.DONE:
raise AlgorithmError("The job was not completed successfully. ")
result = job.result()

# sum over all probabilities where the objective qubits are 1
prob = 0
for bit, probabilities in result.quasi_dists[0].items():
i = int(bit)
# get bitstring of objective qubits
full_state = bin(i)[2:].zfill(circuit.num_qubits)[::-1]
state = "".join([full_state[i] for i in estimation_problem.objective_qubits])

# check if it is a good state
if estimation_problem.is_good_state(state[::-1]):
prob = prob + probabilities

cos_estimate = 1 - 2 * prob
elif self._quantum_instance.is_statevector:
circuit = self.construct_circuit(estimation_problem, k, measurement=False)
statevector = self._quantum_instance.execute(circuit).get_statevector()

Expand Down Expand Up @@ -179,21 +229,37 @@ def construct_circuit(
return circuit

def estimate(self, estimation_problem: EstimationProblem) -> "FasterAmplitudeEstimationResult":
"""Run the amplitude estimation algorithm on provided estimation problem.
Args:
estimation_problem: The estimation problem.
Returns:
An amplitude estimation results object.
Raises:
ValueError: A quantum instance or Sampler must be provided.
AlgorithmError: Sampler run error
"""
if self._quantum_instance is None and self._sampler is None:
raise ValueError("A quantum instance or sampler must be provided.")

self._num_oracle_calls = 0
user_defined_shots = self.quantum_instance._run_config.shots
user_defined_shots = (
self._quantum_instance._run_config.shots if self._quantum_instance is not None else 0
)

if self._rescale:
problem = estimation_problem.rescale(0.25)
else:
problem = estimation_problem

if self._quantum_instance.is_statevector:
if self._sampler is not None or self._quantum_instance.is_statevector:
cos = self._cos_estimate(problem, k=0, shots=1)
theta = np.arccos(cos) / 2
theta_ci = [theta, theta]
theta_cis = [theta_ci]
num_steps = num_first_stage_steps = 1

else:
theta_ci = [0, np.arcsin(0.25)]
first_stage = True
Expand Down Expand Up @@ -240,7 +306,7 @@ def cos_estimate(power, shots):
result.num_oracle_queries = self._num_oracle_calls
result.num_steps = num_steps
result.num_first_state_steps = num_first_stage_steps
if self._quantum_instance.is_statevector:
if self._sampler is not None or self._quantum_instance.is_statevector:
result.success_probability = 1
else:
result.success_probability = 1 - (2 * self._maxiter - j_0) * self._delta
Expand All @@ -252,7 +318,8 @@ def cos_estimate(power, shots):
result.theta_intervals = theta_cis

# reset shots to what the user had defined
self.quantum_instance._run_config.shots = user_defined_shots
if self._quantum_instance is not None:
self._quantum_instance._run_config.shots = user_defined_shots
return result


Expand Down
Loading

0 comments on commit 4008094

Please sign in to comment.