The service architecture of the QLM provides all the tools to describe quite complicated sequences of compilation steps and post-processings, like so:
stack = plugin1 | plugin2 | qpu
However, the information flow along a stack composed of Plugins and a QPU remains linear: the job goes down the stack, reaches the QPU and a result comes back up the stack. Many applications require to be able to iterate some step of job evaluation in an adaptive manner. One may think of, for example, any variational eigensolving procedure.
That's exactly what the Junction object is here for. A Junction can be seen as a Plugin that will adaptively keep sending jobs down the stack and analyze the corresponding results until it decides to stop and returns a final result. For instance, a variational optimizer will optimize variational circuits until some criterion on the variational energy is reached.
Junctions can be used to compose stacks with the same pipe syntax as Plugins and QPUs:
stack = plugin1 | somejunction | plugin2 | qpu
Let us try and program some Junction called IterativeExplorer that will:
import numpy as np
from qat.plugins import Junction
from qat.core import Result, Job
from qat.comm.exceptions.ttypes import PluginException
# Junctions are built by inheriting from the Junction class
class IterativeExplorer(Junction):
r"""
Iteratively explores the (0, 2$\pi$) range for the incoming job's parameter
Args:
nsteps (int): the number of values to try
"""
def __init__(self, nsteps=100):
self.nsteps = nsteps
# Here the 'collective' parameter tells the junction that we would like to handle jobs one by one
# If set to True, we would have to process the full incoming Batch in one go. Let us keep things simple.
super(IterativeExplorer, self).__init__(collective=False)
# Junctions are abstract classes that require you to implement the following method
def run(self, qlm_object: Job, meta_data: dict) -> Result:
parameters = qlm_object.get_variables() # This returns the list of variables of the job
if len(parameters) != 1:
raise PluginException(message="Can't handle Jobs with more than 1 variable")
vname = parameters.pop() # getting the first and only variable name
angles = np.linspace(0, 2 * np.pi, self.nsteps)
values = []
for angle in angles:
# We bind the value of the variable to `angle`
job = qlm_object(**{vname: angle})
# We evaluate the job using the `self.execute` method
result = self.execute(job)
# We extract the energy and push it to our result list
values.append(result.value)
# Extracting the best value
best_value = min(values)
# and the best angle
best_angle = angles[values.index(best_value)]
# We need to return a QLM Result object
return Result(
value=best_value,
meta_data={"best_angle": best_angle}
)
As you can see, Junctions are built by inheriting from the Junction class and implementing its run method.
This method will receive the incoming job object, together with some optional meta data, from the higher part of the stack and has access to the lower part of the stack via the execute method.
Let us now run our junction!
# first we need a qpu
from qat.qpus import get_default_qpu
qpu = get_default_qpu()
# and build a stack
stack = IterativeExplorer(nsteps=25) | qpu
# Now lets build a simple parametrized Job
from qat.core import Observable
from qat.lang.AQASM import Program, RY
prog = Program()
qbits = prog.qalloc(1)
theta = prog.new_var(float, "\\theta")
prog.apply(RY(theta), qbits)
circuit = prog.to_circ()
job = circuit.to_job(observable=Observable.sigma_z(0, 1)) # Z on qbit 0
circuit.display()
# Run!
result = stack.submit(job)
print("Final energy:", result.value, " | best angle:", result.meta_data["best_angle"])
In the previous example, we managed to achieve the expected behavior for our simple optimizer. However, some line of codes are purely here for administrative purposes.
This is why the Optimizer class exists. It provides a very similar interface to the Junction class, but takes care of some of the administrative burden.
Let us see how the previous example can be rewritten using an Optimizer:
from qat.plugins import Optimizer
class IterativeExplorer(Optimizer):
r"""
Iteratively explores the (0, 2$\pi$) range for the incoming job's parameter
Args:
nsteps (int): the number of values to try
"""
def __init__(self, nsteps=100):
self.nsteps = nsteps
super(IterativeExplorer, self).__init__(collective=False)
# The run method changed name and is now called `optimize`
def optimize(self, variables: list) -> tuple:
# the argument `variables` contains the list of variables of the job
if len(variables) != 1:
raise PluginException(message="Can't handle Jobs with more than 1 variable")
vname = variables.pop() # getting the first and only variable name
angles = np.linspace(0, 2 * np.pi, self.nsteps)
values = []
for angle in angles:
values.append(self.evaluate({vname: angle}))
best_value = min(values)
best_angle = angles[values.index(best_value)]
return best_value, [best_angle], "hello :)"
As you can see, we need to implement a method called optimize and have access to a method called evaluate that takes a value map and returns a float.
The optimize method receives the list of variables of the job and should return:
stack = IterativeExplorer(nsteps=25) | qpu
result = stack.submit(job)
print("Final energy:", result.value, " | best angle:", result.meta_data["parameters"])
As you can see the final result is quite similar as in the first implementation. We can have a deep look at its meta data:
for key, value in result.meta_data.items():
print(key, ":", value)
The entry "optimizer_data" contains our third returned value. Most importantly, the optimizer kept track of the different evaluation and returned this trace in the "optimization_trace" entry.
When we implemented our Junction (or our Optimizer), we didn't make any use of the meta_data parameter. In order to improve our plugin, we could make it so the job itself could control the resolution of our exploration.
To do so, we will assume that the user will transmit the number of steps via the "IterativeExplorer_nsteps" entry of the meta data:
# Here most of the code is the same as in the first cell
class IterativeExplorer(Junction):
r"""
Iteratively explores the (0, 2$\pi$) range for the incoming job's parameter
Args:
nsteps (int): the number of values to try
"""
def __init__(self, nsteps=100):
self.nsteps = nsteps # This is now a default value
super(IterativeExplorer, self).__init__(collective=False)
def run(self, qlm_object: Job, meta_data: dict) -> Result:
nsteps = meta_data.get("IterativeExplorer_nsteps", None)
if nsteps is not None:
nsteps = int(nsteps)
else:
nsteps = self.nsteps
print(self.__class__.__name__, ": using", nsteps, 'steps')
parameters = qlm_object.get_variables()
if len(parameters) != 1:
raise PluginException(message="Can't handle Jobs with more than 1 variable")
vname = parameters.pop()
angles = np.linspace(0, 2 * np.pi, nsteps)
values = []
for angle in angles:
job = qlm_object(**{vname: angle})
result = self.execute(job)
values.append(result.value)
best_value = min(values)
best_angle = angles[values.index(best_value)]
return Result(
value=best_value,
meta_data={"best_angle": best_angle}
)
stack = IterativeExplorer(nsteps=1) | qpu
result = stack.submit(job)
print("Final energy:", result.value, " | best angle:", result.meta_data["best_angle"])
result = stack.submit(job, meta_data={"IterativeExplorer_nsteps": "32"})
print("Final energy:", result.value, " | best angle:", result.meta_data["best_angle"])