Complex types, optional values

So far we have only build graphs with a single input and output and defined everything else as constants. Tierkreis allows for multiple inputs and outputs similar to python, by defining tuples of multiple fields. As an invariance, a graph will still always have exactly one output node which also mimics python.

In this example we will look at defining graph which compiles and runs a single circuit on an IBM simulator. For this you will need valid IBMQ credentials.

Inputs and Outputs

In tierkreis, inputs and outputs are defined as named tuples of TKR types. Were building a graph circuit -> int -> (BackendResult, int)

%pip install tierkreis pytket

/home/runner/work/tierkreis/tierkreis/.venv/bin/python3: No module named pip

Note: you may need to restart the kernel to use updated packages.

from typing import NamedTuple

from tierkreis.controller.data.models import TKR, OpaqueType


class IBMInput(NamedTuple):
    circuit: TKR[OpaqueType["pytket._tket.circuit.Circuit"]]  # noqa: F821
    n_shots: TKR[int]


class IBMOutput(NamedTuple):
    results: TKR[OpaqueType["pytket._tket.backends.BackendResult"]]  # noqa: F821
    n_shots: TKR[int]

Now we construct the graph using the tkr-aer-worker. Note how we address inputs by their name g.inputs.circuit and for the output we construct an output object.

from tierkreis.aer_worker import get_compiled_circuit, submit_single
from tierkreis.builder import GraphBuilder


def compile_run_single() -> GraphBuilder[IBMInput, IBMOutput]:
    g = GraphBuilder(IBMInput, IBMOutput)

    compiled_circuit = g.task(
        get_compiled_circuit(
            circuit=g.inputs.circuit,
            optimisation_level=g.const(2),
        ),
    )
    res = g.task(submit_single(compiled_circuit, g.inputs.n_shots))
    g.outputs(IBMOutput(res, g.inputs.n_shots))
    return g

We now can run this code as before. Inputs currently still have to be defined as inputs

from uuid import UUID

from pytket._tket.circuit import Circuit

from tierkreis.consts import PACKAGE_PATH
from tierkreis.controller import run_graph
from tierkreis.executor import UvExecutor
from tierkreis.storage import FileStorage, read_outputs

storage = FileStorage(UUID(int=109), do_cleanup=True, name="ibmq_example")
executor = UvExecutor(PACKAGE_PATH.parent / "tierkreis_workers", storage.logs_path)


def ghz() -> Circuit:
    circ1 = Circuit(2)
    circ1.H(0)
    circ1.CX(0, 1)
    circ1.measure_all()
    return circ1


inputs = {
    "circuit": ghz(),
    "n_shots": 1024,
    "backend": "ibm_torino",
}
graph = compile_run_single()
run_graph(
    storage,
    executor,
    graph,
    inputs,
)
res = read_outputs(graph, storage)

Optional Values

Tierkreis can also deal with optional values which are indicated python style bny providing the type hint type | None = None. Previously we have defined the optimization level by setting a constant. Inspecting the type declaration of compile_circuit_ibmq reveals the following type: optimisation_level: TKR[str] | None = None which means we can set it optionally. We will also pull this up into the inputs:

class IBMInputOptional(NamedTuple):
    circuit: TKR[OpaqueType["pytket._tket.circuit.Circuit"]]  # noqa: F821
    n_shots: TKR[int]
    optimisation_level: TKR[int] | None = None  # Optional input


def compile_optional() -> GraphBuilder[IBMInputOptional, IBMOutput]:
    g = GraphBuilder(IBMInputOptional, IBMOutput)

    compiled_circuit = g.task(
        get_compiled_circuit(
            circuit=g.inputs.circuit,
            optimisation_level=g.inputs.optimisation_level,
        ),
    )
    res = g.task(submit_single(compiled_circuit, g.inputs.n_shots))
    g.outputs(IBMOutput(res, g.inputs.n_shots))  # type: ignore
    return g

You can now run the the graph with or without th optional input. Note, that not providing all inputs will produce some error messages.

inputs = {
    "circuit": ghz(),
    "n_shots": 1024,
    # "optimisation_level": 1, Uncomment to provide the optional input
}

graph = compile_optional()
storage.clean_graph_files()
run_graph(
    storage,
    executor,
    graph,
    inputs,
)
res = read_outputs(graph, storage)