Hello World: Tierkreis Edition

In this example we will construct our own worker and a workflow graph that uses the its tasks. We will first write the hello_world_worker with a greet function.

Writing tierkreis workers

One of the core concepts in tierkreis are workers. In this example, we will cover the conceptual ideas of workers and how to write one. For this we will write a simple worker from scratch. The tkr init cli provides a convenient set up for this which will automate most of the tasks described after. If you followed the previous tutorial, you can also update the example worker to your liking.

Note

As explained in the previous section, workers are independent projects. The worker we are going to implement already exists in the project, so that we can use its package name. If you’re using the example_worker from the previous tutorial, make sure to change hello_world_worker to example_worker. If you don’t want to set up the worker yourself, you can clone the Tierkreis project and use the worker from there. To have python recognize the package you have to run uv sync once.

Concepts

Workers encapsulate a set of functionalities which are described in their API (stubs). The API contains a list of typed function interfaces which are implemented by the worker. Tierkreis expects the functions to be stateless and will only checkpoint inputs and outputs of the function call. Although, it is still possible to introduce state, but this then has to be managed by the programmer.

Each worker is an independent program, can have multiple implementations (e.g. based on architecture) of these interfaces and has its own dependencies. There are two main ways to write workers:

  • Using the tierkreis Worker class for python based workers

  • Using a spec file to define the interface for non-python workers

In this example, we will write a simple worker, that greets a subject. greet: str -> str -> str

Worker Setup

A worker consist of four conceptual parts which we will set up in this example.

  1. Dependency information. For python workers, we recommend using uv and a pyproject.toml.

  2. A main entry point into the worker. For python workers this is main.py. Alternatively any runnable binary will work; Tierkreis sets this up when using the cli.

  3. The api definition of the worker typically called stubs.py or api.py. For python workers this can be generated from the main file. For other workers you can use a typespec file.

  4. (Optional) Library code containing the core logic.

To set this up, you can run:

tkr init worker --worker-name hello_world_worker

1. Dependencies

Dependency information should be provide, such that the worker can run in its installed location. For python, this is canonically done through the pyproject.toml which tierkreis also uses. A minimal example contains

[project]
name = "hello-world-worker"
version = "0.1.0"
requires-python = ">=3.12"
dependencies = ["tierkreis"]

[dependency-groups]
# Optional dev dependency group
dev = ["ruff"]

Ruff is an optional dependency to format the code later.

2. Worker entry point

The entrypoint contains the implementations of the interface. We are going to use the build in Worker class to define the worker. Each worker needs a unique name, here we chose hello_world_worker. In tierkreis the worker is defined by its name and the directory it lives in.

In tkr_hello_world_worker_impl/impl.py we define the worker functionality by instantiating a worker

import logging

from tierkreis import Worker

logger = logging.getLogger(__name__)
worker = Worker("hello_world_worker")

From here we can add functions to the worker by decorating python functions with @worker.task(). To make use of the type safety in tierkreis, you should provide narrow types to the function.

@worker.task()
def greet(greeting: str, subject: str) -> str:
    logger.info("%s %s", greeting, subject)
    return greeting + subject

Now we have defined a greet function that concatenates a two strings, similar to the concat builtin. It now lives in the namespace hello_world_worker.greet. To make sure the worker runs correctly we need to add the following lines in main.py which is usually autogenerated:

%%script false --no-raise-error
from sys import argv

from tkr_hello_world_worker_impl import worker 

def main():
    worker.app(argv)


if __name__ == "__main__":
    main()

They make sure that tierkreis can call the worker correctly.

3. Api definition

For python workers, tierkreis can automatically generate the api definition. You can either run the app as uv run main.py --stubs-path ../api/api.py or call the function directly.

%%script false --no-raise-error
from pathlib import Path
worker.namespace.write_stubs(Path("../api/api.py"))

This will generate a file api.py that contains the interface definition that can now be used as task in a Tierkreis workflow. The code will look like this:

"""Code generated from hello_world_worker namespace. Please do not edit."""

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


class greet(NamedTuple):
    greeting: TKR[str]  # noqa: F821 # fmt: skip
    subject: TKR[str]  # noqa: F821 # fmt: skip

    @staticmethod
    def out() -> type[TKR[str]]:  # noqa: F821 # fmt: skip
        return TKR[str]  # noqa: F821 # fmt: skip

    @property
    def namespace(self) -> str:
        return "hello_world_worker"

Which defines a class describing the function.

4. Using an existing library

For simple workers we can directly write our code as python functions. If you already have a library we recommend keeping the structure, having a single file for your worker tasks. You can import you library functions there and wrap them with a small function, e.g.:

from my_lib import my_func

@worker.task()
def my_func_wrapper() -> None:
    my_func()

Writing the workflow

We want to write a workflow that produces the output Hello <name>, which is similar to the one from the initial example. But instead of the builtins, we’re going to use the newly defined tasks. First we provide the necessary imports, and declare the Graph.

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

graph = Graph(inputs_type=TKR[str], outputs_type=TKR[str])

As before were going to construct the string Hello world from an constant value and a user input. First we define the constant part, by adding a constant node to the graph.

hello = graph.const("Hello ")

We capture the output of that node in the variable hello, which we can use as input to other nodes to insert an edge in the graph. Inputs to the graph cah be referenced in two ways depending if we have a single input (graph.inputs) or multiple inputs (graph.inputs.<input_name>). We will cover multiple inputs in a later example. For now, we want to call the greet function from the hello_world_worker we defined early. To use the type hints from that worker we can import the interface and then use it as a task.

Important

During definition, we use the api which means we don’t need to have the worker available in our project. During runtime, an executor will invoke the workers entry point (main.py).

from hello_world_worker import greet  # noqa: F811

output = graph.task(greet(greeting=hello, subject=graph.inputs))

The benefit of using the function interface is the type checking. The python typechecker will warn us if we provide a type of an incorrect input. To complete our graph, we now have to define an output. As an invariant, a well-formed graph has exactly one output. Although, this output can consist of multiple values, we will cover this in a later example.

workflow = graph.finish_with_outputs(output)

Now that we have defined a workflow we can run the graph. The most generic way to do this is the run_workflow function which requires the following inputs:

  • The graph to run.

  • A dictionary providing the inputs to the graph. A single input will always have the key "value". In this case we want to set it to {"value": "world!"}

To make this reproducible we can define a name and a run_id.

In contrast to before, we now also have to define an executor that will find our worker. In run_workflow we have to define the registry_path to the location that contains the worker directory. We point it to the parent directory of the hello_world_worker.

from pathlib import Path

registry_path = (
    Path().parent / "example_workers"
)  # Look for workers in the `example_workers` directory.

# registry_path = Path().parent / "workers"
# TODO: uncomment this if you're using the init example

Finally we can run the code, which will print the desired output Hello world!.

from tierkreis.cli.run_workflow import run_workflow

input_value = " world!"  # TODO change this
run_workflow(
    workflow,
    {"value": input_value},
    name="hello_world",
    run_id=100,  # Assign a fixed uuid for our workflow.
    registry_path=registry_path,
    use_uv_executor=True,
    print_output=True,
)

value: 'Hello  world!'