Lesson 3: Using predefined workers

In this example we’re going beyoind writing our workers. For this we’re going to extend the graph from the previous lesson.

We’re going to additionally

  • compile the circuit using the tkr-pytket-worker

  • run it on a simulator using the tkr-aer-worker (based on the pytket-qiskit extension)

  • calculate the expected value of our results The workers can be installed using uv add tkr-pytket-worker-impl tkr-aer-worker-impl

In this example we’re also covering storage and execution in a bit more detail. First we start with the graph from before. Since we will add more tasks we will not define the output just yet:

from typing import NamedTuple
from tierkreis.builder import Graph
from tierkreis.controller.data.models import TKR

from my_example_worker import substitute, symbolic_circuit


class PytketInputs(NamedTuple):
    a: TKR[float]
    b: TKR[float]
    c: TKR[float]


graph = Graph(PytketInputs, TKR[float])

circuit = graph.task(symbolic_circuit())
substituted = graph.task(
    substitute(circuit, graph.inputs.a, graph.inputs.a, graph.inputs.a)  # type: ignore
)

Now we add the compilation step. For generic a generic the tkr-pytket-worker has compile_generic_with_fixed_pass. You can find a list of all available tasks here

from pytket_worker import compile_generic_with_fixed_pass, add_measure_all

compiled_circuit = graph.task(
    compile_generic_with_fixed_pass(circuit, optimisation_level=graph.const(2))
)
measured = graph.task(add_measure_all(compiled_circuit))  # type: ignore

Next we will run the circuit on aer with the tkr-aer-workers run_circuit (see docs) And calculate the expected value. The expectation tasks simply assumes the computational basis. For other observables you could defined another task in the my-example-worker as before.

from aer_worker import run_circuit
from pytket_worker import expectation

results = graph.task(
    run_circuit(
        circuit=compiled_circuit,  # type: ignore
        n_shots=graph.const(10),
    ),
)
exp_val = graph.task(expectation(results))
workflow = graph.finish_with_outputs(exp_val)

Now we will define our own storage and executor. Storage is responsible for setting up the checkpointing; it stores the state of the computation. We have to provide a uuid, and optionally a name, as before.

from uuid import UUID

from tierkreis.storage import FileStorage

storage = FileStorage(UUID(int=209), do_cleanup=True, name="quantinuum_submission")

An executor lives in context, where it can access workers to run their main entrypoints. We define this by providing a path to the directory our workers live in, in this case in the tierkreis_workers directory.

If you cloned the directory you now have the source of both workers. You could run them using uv run main.py. For this use case we have the UvExecutor.

from tierkreis.consts import PACKAGE_PATH
from tierkreis.executor import UvExecutor

executor = UvExecutor(PACKAGE_PATH.parent / "tierkreis_workers", storage.logs_path)

As an alternative (or if you only installed the worker), the worker exports a script tkr-pyktet-worker(tkr-aer-worker). It functions as a shell binary, hence we can use the ShellExecutor for it. Since using uv add will make this script available from within our python environment, we don’t need to point to a directory.

from tierkreis.executor import ShellExecutor

executor = ShellExecutor(registry_path=None, workflow_dir=storage.workflow_dir)

Once we provide the workflow inputs we can now run it by providing a storage and an executor. And finally we can print the outputs.

from tierkreis import run_graph
from tierkreis.storage import read_outputs

run_graph(storage, executor, workflow, {"a": -1, "b": 0, "c": 1})
outputs = read_outputs(workflow, storage)
print(outputs)