app/docs/code/nodes.md

Writing node code

A Function node is a Python file. That is all it is — no base class, no decorator, no framework import unless you want one.

def process(temperature, setpoint=21.0):
    """Ask for heat when the room is below the comfort point."""
    return {"heat": temperature < setpoint}

The rules

One function called process. If the file defines exactly one public function under another name, that one is used instead. Two, and the node refuses to load rather than guessing.

Arguments come from ports and settings, by name. temperature above is an input port; setpoint is a setting typed into the node's panel. Both arrive as keyword arguments, which is why a setting may not share a name with a port. See Where a node's values come from.

The return value is a dict keyed by output ports. Every value is checked against the port's declared type before it is published. A key that is not a declared port is an error, not a silent drop — nothing leaves a node except through a port it declared.

Nothing else is importable from the engine. Node code runs in a separate process, on a separate interpreter, with none of Fluksio's own modules on its path. What it can import is what the Modules screen installed — which can include your own project, so a node need not be a self-contained file.

A node is a pure function of its inputs. No context object, no global store, no handle to reach for. A running total or a debounce timer has a specific shape — see Keeping state in a flow.

Producing values over time

A node that produces values during its execution is a generator. Every yield is a dict keyed by output port, published the instant it happens:

def process(lr, steps):
    loss = 1.0
    for _ in range(steps):
        loss = train_one_step(lr)
        yield {"loss": loss}          # published now
    return {"final_loss": loss}

Whatever the generator returns at the end is the node's result — what downstream nodes read. If you never return, the last thing you yield is the result instead.

Mark the port so the flow says what it does:

{"name": "loss", "dtype": "float", "stream": true}

Two consequences. In a run, the whole series is kept as that run's metrics — this is why there is no log_metric() anywhere in the API. And the node's timeout starts measuring silence rather than duration: each emission resets the deadline, so a node yielding every few seconds can run for hours under a timeout of 300.

fluksio.emit

Where a yield cannot reach — the value comes from inside somebody else's callback, and they call you rather than the other way round:

import fluksio


def process():
    model.fit(callbacks=[LambdaCallback(
        on_epoch_end=lambda epoch, logs: fluksio.emit(loss=logs["loss"])
    )])
    return {"weights": ...}

Same ports, same type checking, same publication. Prefer yield where you can reach it; emit where you cannot.

Bytes: artifacts

Messages are JSON, which is what lets the same value pass through Redis, the work queue and the worker protocol unchanged. A checkpoint is not that.

import fluksio


def process(dataset):
    path = fluksio.load_artifact(dataset)     # → a local path to read
    ...
    return {
        "weights": fluksio.save_artifact("model.pt", media_type="application/octet-stream"),
        "score": 0.94,
    }

save_artifact takes bytes or a path, stores them by their SHA-256 digest, and returns a small reference — digest, size, media type, name — which is what an artifact-typed port carries.

Because the address is the content's hash, a sweep whose fifty configs share one preprocessed input stores it once, and a reference stays valid wherever the store is reachable from — including on another machine.

Media

Say what the bytes are and the port can be typed for them:

def process(speech):                     # an `audio` port
    clip = fluksio.load_artifact(speech)
    words = transcribe(clip)
    return {"transcript": words}         # a `str` port

def process(camera_url):
    for index, jpeg in enumerate(grab(camera_url)):   # a generator
        frame = fluksio.save_artifact(
            jpeg, f"frame-{index:05d}.jpg", media_type="image/jpeg"
        )
        frame["meta"] = {"seq": index}
        yield {"frame": frame}                        # an `image` stream port

An image, audio or video port is an artifact reference whose media type has to match, so a node declaring audio never receives a video by accident. See Payload types for what each carries and what rates are realistic.

!!! warning "Emitted media is not kept; returned media is"

Only what a node *returns* is recorded against its run. Frames yielded
along the way are replaced in state by the next one, and the artifact sweep
removes bytes nothing refers to any more — which is what stops a camera
filling the disk. If a particular frame matters, return it.

Printing

print works and is captured. The first 16 KB per call is kept and shown in the flow editor's log panel and on the run's per-node record; the rest is dropped, so a node printing in a loop cannot fill anything up.

Use it to debug. Do not use it to record results — a number worth keeping is an output port, not a line of text.

Errors

An exception fails that node's execution, not the flow. The message you see is one line from the frame in your code, not a stack through the engine — that is a deliberate choice about what is actionable.

The node keeps its last error visible after it recovers, so a failure that fired an alert at 03:00 still says what it was at 09:00. It can also be acknowledged from the canvas.

Timeouts

timeout on a node is how many seconds its code may be silent before it is stopped. A yield or an emit resets the clock, and the first call's imports are not charged to it — a node importing torch is not being slow, it is loading.

There is no timeout by default. Training runs for hours and a node that reports nothing is usually working, so the engine waits: what fails a call is the worker dying, which arrives at once rather than after a deadline. Set a timeout on the nodes where silence means stuck — an HTTP call that should answer in seconds, a loop that can spin — either per node or, for all of them, with FLOW_NODE_TIMEOUT. timeout = 0 says explicitly that this node has none.

Running a node somewhere else

A node declares the label of the machine it needs:

{"id": "train", "device": "gpu", "device_policy": "require", "timeout": 7200}

require (the default) waits for a worker carrying that label; prefer runs locally when none is attached. A node bound to a device is compiled on that machine — a node importing torch is correct on the GPU box and a missing module on the engine, so checking it here would fail something that is fine.

See Remote workers.

Sharing code between flows

A node's source can be promoted to the shared library from its panel, and other flows can then use it by reference. One copy, one place to edit — and every flow using it runs the edit, which is the point and also the caution.

Shared sources live in _lib/ in the flow repository, so they are versioned with everything else.

Packages

Node code runs in a virtual environment of its own, on the installation's data volume — deliberately separate from the one Fluksio itself runs on.

Declare what you import in Modules, or over the API:

curl -X POST $FLUKSIO/modules/apply -H "Authorization: Bearer $TOKEN" \
  -H 'Content-Type: application/json' \
  -d '{"requirements": "numpy>=2\npandas\n"}'

It is a pip manifest installed with uv pip sync, versioned alongside your flows. An install takes effect immediately; nothing restarts.

Your own code as a package

!!! tip "If Fluksio is installed in the venv you work in, skip this"

Node code then runs on that environment, so your project and everything it
imports are already importable — see
[Getting started: data science](../getting-started/data-science.md). What
follows is for a Fluksio with a venv of its own, which is what a container
always has.

A manifest line can name a directory, so the project you already have is installable like any other dependency:

-e /home/you/my-research
numpy>=2

A node body then imports it, and the logic stays where it already lives — in your repository, under your own version control, importing its own siblings:

from myresearch.train import fit


def process(lr, epochs):
    return fit(lr, epochs)

That is the whole of it. The node is three lines, myresearch can be as many modules as it likes, and nothing was copied.

!!! tip "You can have those three lines written for you"

Decorate `fit` with `@node(...)` where it is defined, say which nodes make
a flow with `Flow(...)`, and `fluksio sync` generates the body above —
along with the flow document, so there is nothing to PUT by hand. The
declaration lives beside the function it describes and is checked against
its signature. See
[Getting started: data science](../getting-started/data-science.md).

!!! warning "Editable, but not live"

`-e` means edits reach the venv without reinstalling — but a node's process
already holds the imported module in memory. The engine's workers are
long-lived, so a change to your code is picked up when they are retired,
which is what **Apply** does. Pressing it after an edit is the loop.

A [worker](workers.md) you attach yourself is the exception: it starts a
process per call, so it reads your code fresh every run. If you are
iterating on the code many times an hour, point one at your own
interpreter — `fluksio-worker --python "$(which python)"` — and mark the
node with its label.

!!! note "The path is a deployment detail"

It is resolved on whichever machine runs the node, and the manifest is
committed to the flow repository — so an absolute path from your laptop
means nothing inside a container or on a GPU box. Those need their own
install of the same project; a VCS requirement
(`myresearch @ git+ssh://…@a1b2c3d`) travels where a path does not.

A worked example

The repository ships a small supervised fit as a seedable demo — three nodes, a batch flow, streaming metrics, artifacts between stages, and a GPU-labelled node that falls back to the engine when no worker is attached. It is the shortest complete thing to read:

prepare ──dataset(artifact)──▶ train ──weights(artifact)──▶ evaluate
                                 │
                                 └── loss (streaming float) ──▶ chart

The example is built by a seed script the maintainers run against a live stack; the flow above is what it produces.

See also

• Where a node's values come from
• Keeping state in a flow
• Runs: pipelines that finish
• Node types — the ones you do not have to write