Skip to content

Tutorial — Level 1: Hello World

Let's build a pipeline step by step. We start with a single step that splits a message into its individual characters and prints them.

from typing import NamedTuple
from synaflow import pipeline, step, run

class Params(NamedTuple):
    message: str

def hello(message: str) -> list[str]:
    return list(message)

def printer(hello: list[str]) -> None:
    print(hello)

p = pipeline(
    name="tutorial",
    params=Params,
    steps=[
        step("hello", fn=hello),
        step("printer", fn=printer),
    ],
)

run(p, Params(message="SynaFlow"))
# Output: ['S', 'y', 'n', 'a', 'F', 'l', 'o', 'w']

from typing import NamedTuple
from synaflow import pipeline, step, async_run

class Params(NamedTuple):
    message: str

async def hello(message: str) -> list[str]:
    return list(message)

async def printer(hello: list[str]) -> None:
    print(hello)

p = pipeline(
    name="tutorial",
    params=Params,
    steps=[
        step("hello", fn=hello),
        step("printer", fn=printer),
    ],
)

async_run(p, Params(message="SynaFlow"))
# Output: ['S', 'y', 'n', 'a', 'F', 'l', 'o', 'w']

What happened?

  • pipeline() builds the DAG from the steps.
  • run() / async_run() executes it in topological order.
  • The param message: str is injected into hello, which splits it into a list.
  • printer receives the list and prints it.
flowchart TD
    hello["hello<br/><i>list[str]</i>"]
    printer["printer<br/><i>None</i>"]
    message --> hello
    hello --> printer

Next

Add a transformation in Level 2.