Fluksio/app
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

docs

Browse Source
This commit is contained in:
stroblme
2026-10-13 10:02:20 +02:00
parent a93f865e91
commit 80e595c02e
24 changed files with 2210 additions and 4310 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/concepts/flows.md
+104 -157
View File
@@ -1,204 +1,151 @@
# Flows, nodes and messages
Three ideas hold the system up. Almost everything else follows from them.
A **flow** is a group of **nodes** that exchange **messages**. Nodes declare
what they consume (`requires`) and produce (`provides`); matching message names
connect them. No wires to draw.
## A flow is a graph you did not draw
## How nodes connect
A **flow** is a set of **nodes**. Each node declares the messages it needs
(`requires`) and the messages it produces (`provides`). The graph is whatever
those declarations imply:
For a Function node, declare input and output ports in its panel, then write
`process(...)`. Arguments use input port names; the returned dictionary uses
output port names:
```python
# node "read"
# Node "read": provides temperature
def process():
return {"temperature": read_sensor()}
return {"temperature": 18.5}
# node "decide"
# Node "decide": consumes temperature, provides heat
def process(temperature, setpoint=21.0):
return {"heat": temperature < setpoint}
```
`decide` is downstream of `read` because it needs `temperature` and `read`
produces it. Nobody drew a wire.
Bind both temperature ports to the same message and `decide` follows `read`.
Here `setpoint` uses a Python default; you can also add it as a node
[setting](values.md#settings-constants-of-this-node).
The consequences:
- **Fan-in is free.** Two nodes providing `temperature` are two producers of
one message. The consumer does not change.
- **A node runs when something it reads was published.** Not merely when
something upstream of it ran: a node that produced nothing this time (held
back by a rate limit, say) leaves what reads it on the value it already
has, and so does everything behind that.
- **A wire cannot be wrong.** There is no wire. There is a name that either
matches or does not, and the canvas tells you at edit time which it is.
- **Layout is not a document.** The canvas computes the arrangement, so a flow
has no stored positions to maintain, merge or fight over.
- **Flows stay small.** A graph nobody can hand-arrange is one worth keeping
small, which is the intent. Several atomic flows that name each other beat
one flow with sixty nodes in it.
- Several nodes can produce the same message. Consumers use the latest value.
- Publishing an input wakes its consumers once their required values are
available. An upstream node that publishes nothing leaves downstream nodes
waiting for a new publication.
- The editor arranges the graph automatically. Keep flows focused and share
messages between them when needed.
### Message names are namespaced
Inside flow `house`, a message named `temperature` is really
`house.temperature`. A bare name is qualified with its own flow; a dotted name
is used as written. That is how two flows share a value:
In flow `house`, `temperature` becomes `house.temperature`. A dotted name is
used as written, so another flow can consume it directly. Bind an input port
named `temperature` to `house.temperature`:
```python
# in flow `dashboard`, reading a message that flow `house` produces
def process(house_temperature): # port bound to "house.temperature"
...
def process(temperature):
return {"shown": temperature}
```
The canvas draws messages arriving from another flow as labelled endpoints, so
you can see where they come from without opening the other flow.
The canvas shows cross-flow messages as labelled endpoints.
## A node is a function with declared ports
## Node types and payloads
Most nodes are **Function** nodes: a Python file defining `process(...)`. Its
arguments are its input ports by name; its return value is a dict keyed by
output ports.
```python
def process(reading, unit="C"):
return {"shown": reading if unit == "C" else reading * 1.8 + 32}
```
`reading` is a port. `unit` is a **setting**: a constant of this node's code,
typed into its panel and stored with the flow. Both arrive as arguments, which
is why a setting may not share a name with a port. See
[Where a node's values come from](values.md).
The rest of the node types are the ones that would be tedious or unsafe to
write yourself: MQTT, HTTP, InfluxDB, schedules, switches, notifications. Each
one is configured by filling in a form the editor generates from its parameter
schema, so they all behave the same way. The full list is in
Function nodes run your Python code. Built-in nodes handle MQTT, HTTP,
InfluxDB, schedules, filtering and more through forms in their panels. See
[Node types](../reference/node-types.md).
### Ports are typed
Every port declares a payload type (`dtype`). Values are checked against it,
and dashboard widgets offer compatible messages. Scalars, structured JSON and
media references are covered in [Payload types](../reference/payload-types.md).
A port declares a `dtype`: `float`, `int`, `str`, `bool`, `json`, `record`,
`list`, `series`, `artifact`, or one of the media types `image`, `audio` and
`video`. Every value that passes through is checked against it.
Types are not decoration. They are what lets the dashboard editor offer you
only the messages a gauge can actually draw, and what lets the canvas refuse a
binding before anything runs. See [Payload types](../reference/payload-types.md).
Everything on the wire is JSON. Bytes (a checkpoint, an image, a model)
travel as an `artifact`: the bytes go to a content-addressed store and the
message carries a small reference to them. The media types are that same
reference, saying what kind of bytes are behind it.
Messages carry JSON. For binary data, store the bytes as an **artifact** and
publish its reference. `image`, `audio` and `video` are artifact references
restricted to their media family.
### Streaming ports
A port marked `stream` produces repeatedly *during* one execution rather than
once at the end: a training loss, a progress fraction, a frame from a camera, a
second of speech. A node publishes on one by being a generator and yielding, or
by calling `fluksio.emit`.
Mark an output **streaming** when it publishes during an execution: progress,
training loss or media chunks. A Function node can yield values or call
`fluksio.emit`.
Each value is delivered to the nodes reading it, in the order it was produced,
so a recogniser slower than the microphone in front of it still sees every
chunk rather than only the newest. What is in state remains the latest value,
which is what everything else reads, and what a run keeps is the whole series.
Downstream nodes receive each publication in order; message state holds the
latest value. A batch run records the full series. An output `interval` limits
live publication without thinning the run's recorded series.
An `interval` on a streaming port thins what reaches the canvas without
thinning the run's record of it.
### Keeping node code predictable
### Nodes are pure
Write Function nodes as functions of their declared inputs and settings.
Fluksio provides no hidden `context` or global flow store. To remember a value
between executions, use a message the node reads and writes; see
[Keeping state in a flow](state.md).
A node is called with the values of the messages it declares and returns the
values of the messages it provides. There is no context object, no global
store, no handle to reach for.
## Live and batch flows
That is deliberate: a node with hidden state cannot run twice in parallel,
cannot be replayed, and cannot be moved to another machine. Plenty of real
automations do need to remember something, and there is a specific way to say
so; see [Keeping state in a flow](state.md).
## Two shapes of flow
Set `mode` on the flow:
Choose **Mode** in the flow settings:
| | `live` (default) | `batch` |
|---|---|---|
| Runs | continuously | once per run, on request |
| Started by | subscriptions, schedules, webhooks | `POST /runs/flows/{name}` |
| Ends | never | when the graph drains |
| Keeps | the last value of each message | a run record: params, result, metrics, artifacts |
| Is | a thermostat, an ETL job on a cron | an experiment, a CI-style job |
| Runs | reacts continuously | once per requested run |
| Started by | subscriptions, schedules, webhooks or manual execution | Run in the editor, SDK or API |
| Finishes | stays active until stopped | when its work finishes |
| Keeps | latest message values | parameters, results, metrics and artifacts |
| Example | thermostat | experiment or build job |
A batch flow is built and validated like any other, appears on the same canvas
and is type-checked the same way. It is simply never *activated*: no
subscriptions, no schedules, no webhooks. See
[Runs: pipelines that finish](runs.md).
Batch flows use the same editor and validation, but do not activate background
subscriptions, schedules or webhooks. See [Runs](runs.md).
## Editing is separate from running
## Flow inputs
Every flow has a published version and, while you are working, a draft.
- **Saving** writes the draft. The engine keeps running the published version.
- **Publishing** promotes the draft. The engine reloads and picks it up.
- **Discarding** throws the draft away.
The store is a git repository (`flow.json` for the structure, `nodes/*.py` for
the code) and each save is a commit. So a flow's history is readable with
ordinary git tooling, and copying a flow between instances is copying a
directory.
Saving carries the version you last saw. If someone else saved in between, you
get a 409 instead of quietly overwriting their work.
## What can be wrong, and when you find out
The canvas validates continuously and names problems on the nodes they belong
to:
| Issue | What it means |
|---|---|
| `unconnected_input` | a port needs a message nothing in reach provides |
| `missing_initial_value` | the message exists but has never held a value, and nothing will give it one. Not reported on a batch flow: its inputs arrive with the run |
| `cycle` | A waits for B and B waits for A — nothing could ever start |
| `self_loop_needs_initial` | a node reads a message it also writes, with no starting value |
| `node_error` | the node's code did not load: a syntax error, a missing import |
| `unauthenticated_hook` | advisory — a webhook with no shared secret is open to anyone |
| `node_unhealthy` | the node loaded but is not working: a connector that cannot reach its device, or whose last publication failed |
A flow with any of these except the advisory one and `node_unhealthy` does not
run. A node reporting itself down is a live condition, not a build error, so
the rest of the flow keeps going and the issue clears by itself once the node
reports well again. The health summary on Home counts them, so "why is nothing
happening?" has an answer that does not involve reading logs.
## What happens at runtime
- **A flow can be started and stopped.** Stopped means its subscriptions and
schedules are torn down.
- **A flow can be paused and stepped.** Paused holds messages instead of
running them; step releases exactly one. This is how you test something
before it moves a relay.
- **A failing node does not take the flow down.** It reports an error, keeps
its last error visible after it recovers, and can fire an alert.
- **A flow whose background tasks keep crashing is quarantined.** The engine
stops restarting them and says so, rather than spinning. Publishing a change
gives it another chance.
## Values that arrive from outside
Some messages are not computed by any node: a dashboard control writes them,
the API publishes them, a batch run passes them in. Declare those as the flow's
**inputs**, with the value they start from:
Declare messages supplied by dashboard controls, the API or run parameters as
the flow's **Inputs**. Give live inputs a starting value:
```json
{"inputs": [{"spec": {"name": "setpoint", "dtype": "float"}, "initial": 21.0}]}
```
Without that, the node reading `setpoint` waits for something nothing provides,
and the canvas says so. With it, the flow starts at 21.0 and whatever writes
the message afterwards takes over.
This seeds `house.setpoint` in flow `house` if it has no value yet. Later
publications replace it. Batch inputs are run parameters, and batch outputs
select the run's result. See [Where a node's values come from](values.md).
## Drafts and publishing
- **Saving** stores a draft; the engine keeps running the published version.
- **Publishing** promotes the draft and reloads the flow.
- **Discarding** removes the draft.
Shared node code is an exception: saving it updates every flow using it
immediately. See [The flow editor](../interface/flow-editor.md#code-and-reuse).
Flows are stored in a git repository: `flow.json` describes the graph and
`nodes/*.py` holds code. Saves are versioned, and stale versions return a
conflict instead of overwriting another editor's changes.
Python installations need Git on the host to record history. Without it,
flows still save as files; Docker includes Git.
## When a flow needs attention
The canvas reports issues on the affected nodes:
| Issue | Meaning |
|---|---|
| `unconnected_input` | no node or declared flow input supplies the message |
| `missing_initial_value` | a live input has no starting value |
| `cycle` | nodes depend on each other in a loop |
| `self_loop_needs_initial` | a node reads and writes a message with no starting value |
| `missing_source` | a Function node has no saved code |
| `node_error` | code or configuration could not load |
| `unauthenticated_hook` | a webhook has no shared secret; advisory |
| `node_unhealthy` | a loaded node reports a runtime problem, such as a lost connection |
Build issues prevent the affected flow from running. The last two are runtime
or advisory issues: other work continues, and node health clears when the
connection recovers.
You can **stop** a live flow to remove its subscriptions and schedules, or
**pause** it to hold pending work while values still arrive. **Step** releases
one held item. Repeated background crashes can quarantine a flow; publishing
a change gives it another chance.
## Where to next
- [Where a node's values come from](values.md) — ports, settings, flow inputs
- [Keeping state in a flow](state.md) — the sanctioned way to remember something
- [Runs: pipelines that finish](runs.md) — the batch half of the engine
- [Writing node code](../code/nodes.md) — the practical side of `process()`
- [Where a node's values come from](values.md) — ports, settings and flow inputs
- [Keeping state in a flow](state.md) — remembering a value
- [Runs](runs.md) — batch flows
- [Writing node code](../code/nodes.md) — practical Python examples
docs/concepts/runs.md
+178 -389
View File
@@ -1,476 +1,265 @@
# Runs: pipelines that finish
A flow that automates a house never ends: a value arrives, nodes fire, and it
waits for the next one. A research pipeline is the other shape: parameters go
in, stages execute in order, and at some point it is *done* and has produced
something worth keeping. Fluksio calls the second one a **run**, and it is the
same engine either way.
A **live flow** reacts to events. A **batch flow** starts with parameters,
runs its graph once and returns a result. Each execution is a **run**, with
its own state, parameters, node records, metrics and artifacts.
This is what makes Fluksio usable where Kedro, MLflow or ClearML would be: a
run has parameters that identify it, a result, per-step metrics, artifacts and
a place in a queryable history, with no second server and no
project bootstrap on every execution.
Use runs for experiments, reports and other jobs with a finish line. For a
complete first example, start with the
[data science guide](../getting-started/data-science.md).
## A batch flow
Set `mode: "batch"` on the flow and name the messages its result should hold:
In Python, `Flow(...)` defaults to `mode="batch"`. Declare `inputs` for
run parameters and `outputs` for the final result. On the canvas, choose batch
mode and define the same inputs and outputs.
```json
{
"name": "train_polymer_gnn",
"mode": "batch",
"outputs": ["final_loss", "report"],
"inputs": [
{"spec": {"name": "lr", "dtype": "float"}, "initial": 0.1},
{"spec": {"name": "steps", "dtype": "int"}, "initial": 3000}
],
"nodes": [{"id": "train", "timeout": 7200, "device": "gpu", "...": "..."}]
}
```
A batch flow does not need activation and does not open subscriptions,
schedules or webhooks. A run executes each node once after its dependencies
finish. Missing parameters use their declared initial values.
A batch flow is built and validated like any other (it appears on the canvas,
its ports are type-checked) but it is never *activated*: no subscriptions, no
schedules, no webhooks. It runs when a run asks it to, and not otherwise.
Its `inputs` are its parameters. A run supplies values for them; anything it
does not supply keeps the declared initial value.
One thing a batch flow may not do is rate-limit a port (`interval`). A rate
limit holds a value back for a timer to release, and a run has no timer, so the
value would be dropped rather than delayed, so submitting is refused instead.
Non-streaming ports cannot use `interval` in a batch flow: there is no timer
to deliver a delayed value. Streaming ports can use it to reduce display
updates while keeping every metric point.
## Submitting
**Runs → New run** in the app is the same call with a form in front of it: pick
a flow, fill in its declared inputs, press Run. A comma-separated list in a
number field runs every combination of them as a sweep. **Use draft** runs the
working copy rather than what is published, which is how a change is tried
before it is deployed, and each run keeps its own state, so nothing it computes
reaches the live flow.
In **Runs → New run**, choose a flow, enter its inputs and press **Run**.
**Use draft** runs the working copy without publishing it.
```bash
curl -X POST $FLUKSIO/runs/flows/train_polymer_gnn \
-H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
-d '{"params": {"lr": 0.3, "steps": 4000}, "seed": 7}'
From a terminal:
```sh
fluksio run train --lr 0.1 --epochs 20 --wait
```
The answer is immediate and the run is `queued`; a training run is measured in
hours, so nothing waits for it. Poll `GET /api/v1/runs/{id}` for its status,
result, per-node record and artifacts.
Or submit through the API:
Wrong parameters are refused before anything executes: an undeclared name, or
a value of the wrong type, comes back as a 422 naming the problem.
```sh
curl -X POST "$FLUKSIO/runs/flows/train" \
-H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' \
-d '{"params": {"lr": 0.1, "epochs": 20}, "seed": 7}'
```
`"draft": true` on the body runs the unpublished working copy — the flow
document and the node sources the editor is showing. The copy is read again
when the run starts, so a draft published while it waited runs as the
published one.
Here `$FLUKSIO` is the API base URL ending in `/api/v1`, and `$TOKEN`
is an access token. See [The HTTP API](../code/api.md) for authentication.
Submission returns immediately with status `queued`. Read
`GET /api/v1/runs/{id}` for status and results. Unknown parameters or wrong
types return 422 before execution.
The API body also accepts `"draft": true`, `"no_cache": true` and an
`idempotency_key`. Reusing an idempotency key returns the original submission
instead of creating another run. Drafts are read when execution starts, so
edits made while a run is queued can affect it.
A seed is recorded on the run and fills a declared input named `seed`.
Your node code must use that input to seed its random number generator.
### Sweeps
An ensemble is the same parameters at different seeds; a grid search is the
parameters spread out. Both are one call, and the caller builds the list:
A sweep submits every combination of input values as one group:
```bash
curl -X POST $FLUKSIO/runs/flows/train_polymer_gnn/sweep \
-H "Authorization: Bearer $TOKEN" -H 'Content-Type: application/json' -d '{
"runs": [{"params": {"lr": 0.1}, "seed": 1}, {"params": {"lr": 0.3}, "seed": 1}]
}'
```sh
fluksio sweep train --param lr=0.01,0.1 --param epochs=10,50 --wait
```
From a terminal that is `fluksio sweep train_polymer_gnn --param lr=0.1,0.3`,
which builds the product of the lists you give it and posts the same call.
These four runs share a `group_id` and execute concurrently within the
engine's limits. Each has its own state, so one run cannot overwrite another's
messages. Vary seeds with `--param seed=1,2,3` if the flow declares that input.
They share a `group_id`, so `GET /api/v1/runs?group=…` is the sweep, and they
execute in parallel. That is safe because **each run has a state backend of its
own**: message names are global keys, so two runs of one flow would otherwise
overwrite each other's values. They do not.
The API equivalent is `POST /api/v1/runs/flows/{name}/sweep` with a
`runs` list of parameter dictionaries and optional seeds. Filter history
with `GET /api/v1/runs?group=<group-id>`.
## Producing values before you are finished
A training loop has numbers worth keeping long before it has a result. Those
numbers are **outputs**, not logs: a node declares a port for them and produces
them over time, which in Python is a generator.
Training curves and progress values leave through output ports. A Python
generator publishes them with `yield`; callbacks can use `fluksio.emit`.
Declare the port with `stream: true`. See
[Writing node code](../code/nodes.md#producing-values-over-time) for examples.
```python
def process(lr, steps):
loss = 1.0
for _ in range(steps):
loss = train_one_step(lr)
yield {"loss": loss} # published now, on the `loss` port
return {
"weights": fluksio.save_artifact(dump(model), "weights.npz"),
"final_loss": loss,
}
Every numeric emission is stored under its flow-qualified message name,
such as `train.loss`, with a step counter and timestamp. Retrieve it with:
```text
GET /api/v1/runs/{id}/metrics?name=train.loss
GET /api/v1/runs/series/compare?ids=<a>,<b>&metric=train.loss
```
Mark the port it streams on, so the flow says what it does:
A streaming port's `interval` reduces updates sent to the canvas, while the
run history keeps every point. In a run, intermediate emissions do not start
downstream nodes; those wait for the node's final result.
```json
{"name": "loss", "dtype": "float", "stream": true}
```
Every yield is published the instant it happens: same port, same type check,
same place on the canvas as any other value. Whatever the generator `return`s
is the node's result, and is what downstream nodes read. If you never
`return`, the last thing you yield is the result instead.
There is no `log_metric()`. A metric leaves through a port, which makes it a
message: a chart binds to it directly, a downstream node can consume it, and the
run keeps its series.
**Where a yield cannot reach** — the value comes from inside somebody else's
callback, and they call you rather than the other way round. `fluksio.emit`
writes the same ports the same way:
```python
import fluksio
def process():
model.fit(callbacks=[LambdaCallback(
on_epoch_end=lambda epoch, logs: fluksio.emit(loss=logs["loss"])
)])
return {"weights": ...}
```
### What a run does with them
Every number a node emits is kept as the run's series, stepped by the count of
emissions on that message. Read one back with
`GET /api/v1/runs/{id}/metrics?name=<flow>.loss`, or leave `name` off for
every series the run kept, or compare runs:
```
GET /api/v1/runs/series/compare?ids=<a>,<b>,<c>&metric=<flow>.loss
```
That answers in the `series` shape a chart widget already draws, so three
training curves side by side is a widget binding. During a run the values also
arrive live on the flow socket, so a chart bound to the port fills in as the
training goes.
A streaming port may set `interval` to thin out what reaches the canvas. The
run's history still keeps every value, because the interval is asking for the
display not to be flooded, not for the curve to have holes in it.
Emitting has a second effect: **a node's timeout measures silence, not
duration**. A node that yields every few seconds can run for hours under a
`timeout` of 300; one that says nothing for longer than its timeout is killed.
There is no timeout unless one is set, since a training node that reports
nothing is usually working. Set `timeout` where going quiet means stuck, at how long
the node may plausibly be.
In a *live* flow, an emission also wakes whatever is downstream of it, exactly
as a subscriber publishing does. In a run it does not: a run's graph is
scheduled once, and three thousand mid-node cascades would leave "the run has
finished" with nothing to mean.
A node's `timeout` measures silence. Each yield or emit resets it, so a node
can run for hours while reporting progress. No timeout is imposed by default.
## Artifacts
Bytes never travel as a message. `save_artifact` writes them to a
content-addressed store and returns a small reference (digest, size, media
type, name) which is what an `artifact`-typed port carries:
Use `fluksio.save_artifact` for checkpoints, datasets, plots and other bytes.
Return its reference through an `artifact` port. Another node calls
`fluksio.load_artifact` to obtain a local path.
```python
def process(weights): # requires: weights, dtype "artifact"
path = fluksio.load_artifact(weights)
...
Identical bytes share one stored copy. Returned artifacts are listed on the
run and can be downloaded with:
```sh
fluksio artifacts <run-id>
fluksio artifacts <run-id> train.weights -o weights.json
```
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. Artifacts a run produced are listed on it and
downloadable at `GET /api/v1/artifacts/{digest}`.
Use the exact message name shown by the list command. Emitted media frames
are transient; return a frame you need to keep as a run artifact.
## Stage caching
A run mostly does not redo what an earlier one already did. Before a node
executes it is fingerprinted with a sha256 over its source, its settings, the
ports it declares and the values it is about to read. If some earlier run of that
same fingerprint finished, what that one returned is restored into this run's
state and the node is skipped. It is recorded with the status `cached` and a
duration of zero, and its artifacts are listed on the new run as well, so they
stay downloadable from either.
Before executing a Python node, the engine checks for a successful previous
execution with matching source, settings, port declarations and input values.
On a hit, it restores the outputs and artifacts and records the node as
`cached`.
The settings go into the key raw, so a secret contributes its `{"$secret":
name}` reference and never its value. An artifact input counts as its content
digest: the same bytes under a different filename are the same input. Each node
on a run carries the `cache_key` it was looked up by.
For code-defined flows, sync also records digests of the project modules
reachable from each node. Editing those modules invalidates the corresponding
stage. Static import tracking does not follow computed import names or track
installed dependency versions; use `--no-cache` after changes it cannot see.
For a [code-defined flow](../getting-started/data-science.md), "its source" is
the generated shim, which imports the real function and does not change when
that function does. So the key carries one thing more: a digest of the project
modules that node's function reaches, worked out by `fluksio sync`, which is
the only side that imports your code and can see what it imports, and read
again from those files when the run starts.
Artifact inputs are compared by content digest. Secret settings contribute
their references rather than secret values. External state is not part of a
cache key, so disable caching on a node that reads changing data or performs
side effects:
Editing a helper three calls down from the node invalidates it, which is the
point: the alternative is a re-run answering with the previous code's numbers.
Editing something the node does not reach leaves the hit standing, which is
the other half: a notebook two directories away is not a reason to retrain.
The walk follows imports statically and stops at the standard library, at
anything installed, and at Fluksio itself; a module imported under a name the
code computes is not followed. An engine that cannot see the files keeps what
sync recorded instead of nothing, so a worker on another machine no longer
keys every run the same.
```python
@node(provides=Port("reading", "float"), cache=False)
def read_sensor():
return get_current_reading()
```
The run history *is* the cache; there is no second store. A node's returned
outputs are kept on its run record as canonical JSON, up to 256000 characters.
A node returning more than that is not cacheable that run. An entry
whose artifact bytes have since left the store is a miss, not an error.
Use `fluksio run ... --no-cache`, `fluksio sweep ... --no-cache`, or
`"no_cache": true` in an API submission to execute every node.
Only `python` nodes are cached, and by default all of them are. A built-in node
type or a connector node has side effects and no source to fingerprint, so
neither is ever a candidate. Turn it off for one node with
`@node(..., cache=False)` (the flow document carries it as `cache`, so the
canvas and the API can change it too) or for one run with
`fluksio run --no-cache`, `fluksio sweep --no-cache`, or `"no_cache": true` in
the submission body.
Only Python nodes are cached. Built-in and connector nodes are not. The cache
uses run history; outputs too large to store there are not cacheable, and
missing artifact bytes make an entry a miss.
A cached node replays no emissions, since those values were the story of an
execution that is not happening this time, so its series is not rewritten
either. The run it was restored from is recorded instead, and that is where the
curve is read back from: asking the reusing run for its metrics answers with
the same points, under its own flow's names. The one way to be left with a
result and no curve is for that earlier run to have been deleted. Deleting a
flow does not do that — its runs stay — so this only happens when the run
itself was deleted.
A cache hit does not replay emissions. Its metrics are read from the original
run, so comparisons still show the recorded curve. Deleting that original run
leaves cached results available but removes their curve.
## Objects that cannot be serialized
A live model, a `DataLoader`, a JAX-compiled function: none of these cross a
node boundary, and no framework flag will make them. There are exactly two
patterns, and they are both deliberate:
* **Keep them in one node.** Stages that must share live memory are one node.
Building the model and training it is one stage; the fact that Kedro would
make them two nodes is Kedro's problem, not a structure worth reproducing.
* **Cross at a checkpoint.** Save what matters as an artifact and rebuild from
it on the other side. That is the boundary that also survives the next node
running on a different machine.
Live models, data loaders and compiled functions cannot cross a JSON message
boundary. Keep stages that share live objects in one node, or save a checkpoint
artifact and rebuild the object in the next node.
## Running a node somewhere else
A node that needs a GPU declares the label of the machine that has one:
Attach a [remote worker](../code/workers.md) to the engine and declare the
worker's label as the node's `device`. `device_policy="require"` waits for
that label; `"prefer"` uses the local engine when none is attached.
```json
{"id": "train", "device": "gpu", "device_policy": "require", "timeout": 7200}
```
A worker on that machine dials **out** to the engine, because the engine
generally cannot reach it (different network, no inbound route) and because
nothing should expose Redis across hosts. Install it on the box, mint it a
token, and start it:
```bash
curl -X POST $FLUKSIO/workers/tokens -d '{"name": "gpu-dev", "labels": ["gpu", "cuda12"]}' # once, as an admin
pip install fluksio-worker
fluksio-worker \
--url wss://api.${DOMAIN}/api/v1/workers/attach \
--labels gpu,cuda12 \
--python /opt/torch-venv/bin/python
```
`fluksio-worker` is its own distribution: the agent, the node runner, and
`websockets`. Nothing of the engine, so a GPU box does not install a database
driver to run a training step. Where pip is not an option, the two files still
work copied into one directory and run with `python agent.py …`; the engine
serves the runner at `GET /api/v1/workers/runtime`.
`--python` is the interpreter node code runs on, which is how the GPU box keeps
its CUDA wheels without the engine ever installing them. The node's source
travels with every call, so nothing has to be deployed there.
A few consequences worth knowing:
* `import fluksio` inside a node is the worker's own reporter — `emit`,
`save_artifact`, `load_artifact`, installed before the node's code runs, so
the installed `fluksio` package (if the box has one) never shadows it.
* 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 on the engine would fail a node that is fine.
* If nothing carrying the label is attached, the run stays `queued` and says
what it is waiting for. Submit first, switch the GPU box on later.
* Cancelling a run kills what it is executing, there or here, and leaves other
runs of the same node alone.
* If the worker disappears mid-call, the run fails in seconds with
`worker went away mid-call` rather than waiting out its timeout.
* `device_policy: "prefer"` runs locally when no such worker is attached;
`"require"` (the default) waits for one.
The selected machine needs your imported packages. Code-defined flows also
need your project installed or available there: generated import shims travel
with the call, but your package does not. Workers connect outward to the
engine; see the worker guide for tokens and resource requests.
## Durability
Submitting journals the run to a Redis stream of its own, separate from the
one the automations use: a burst of five hundred sweep runs must not stand
between a house and its heating. An engine that is down when a run is
submitted picks it up when it starts.
Queued runs are stored in the database and picked up when the engine restarts.
Running jobs refresh a lease. If the engine dies and the lease expires, those
runs become `abandoned` rather than being executed again automatically.
The row is what makes that true rather than the queue. A run is written before
the work item is added, so an engine reads its own history at startup and
wakes itself for anything still `queued` — which is what a run submitted
seconds before a restart is, and what an in-memory queue would otherwise have
lost with the process.
Retry a finished run or the unfinished results of a sweep:
From the moment a run is claimed, its database row is the record and the queue
is finished with it. Redelivering two hours of training because an
acknowledgement was late is not recovery; instead a running run refreshes a
lease, and one whose lease goes stale is marked `abandoned`, which is what a
run whose engine was killed mid-training becomes. Starting it over is a
decision rather than something that happens: [`fluksio
retry`](../code/cli.md#fluksio-retry) submits it again, and `--group` does
that for the runs of a sweep that did not finish.
```sh
fluksio retry <run-id>
fluksio retry --group <group-id>
```
A retry creates a new run with the previous inputs, seed and group, linked
to its parent. It uses the flow's current definition, and can reuse completed
cached stages. Cancel active runs before retrying them.
## Looking at what ran
**New run** is where one is started, and the **Runs** screen is the experiment
log: every run newest-first, filtered by flow, by status, or down to one
sweep. A sweep is worth filtering to, since the table then draws a column per
parameter that actually varied, which is what makes fifty runs of one flow
readable.
The **Runs** screen filters by flow, status or sweep. Each run shows its
parameters, per-node timings and logs, traceback, metrics, artifacts and
result.
Tick two or more and their curves go side by side. Shift-click to take a
range, or the header box to take everything on screen. That comparison is the
address, so a link to it is a link someone else can open.
Select two or more runs to compare curves and the parameters that differ.
Plot against step, elapsed seconds or another metric from the same runs.
Comparisons have shareable URLs.
The curves are drawn against the step by default. They can also be drawn
against elapsed seconds, which answers "which one got there sooner" and is
measured from each run's own first reading so that runs started hours apart
still lie on top of each other; or against another metric of the same runs (an
epoch, or samples seen) joined on the step the two share.
One run in full is params, the per-node record with its logs and traceback,
the artifacts it made, its metrics and its result. **Retry** submits it again
as a run of its own, keeping the flow, the inputs, the seed and the sweep it
belonged to.
A run outlives the flow it ran. Deleting a flow leaves its history standing —
being the record of what ran is the point of keeping one — so the Runs screen
still lists it, and opening it says **flow deleted** with Retry disabled,
because there is nothing left to run it against. Clearing that history is a
decision of its own: a run at a time on the screen, or
`DELETE /api/v1/runs?flow=<name>` for a whole flow's worth, which is what a
reseed wants.
Deleting a flow keeps its run history. You can still inspect those runs, but
cannot retry them until the flow exists again. Delete history separately,
after active runs finish or are cancelled.
### Taking it into a dataframe
An analysis wants a table rather than a screen, and there are two it usually
wants. `fluksio export metrics` is the long one, a row per run, metric and
step; `fluksio export runs` is the wide one, a row per run with the
inputs that varied as columns beside its final numbers:
Export metrics as a long table and results as a row per run:
```sh
fluksio export metrics --flow train -o curves.csv
fluksio export runs --flow train --status ok -o results.csv
```
Both include run IDs. Result rows also include input values, final numeric
outputs, status, duration and code identifiers. Nested values use dotted paths,
such as `metric.final_metrics.train_loss`.
Python returns the same rows directly:
```python
import pandas as pd
from fluksio.sdk.client import Client
client = Client()
curves = pd.DataFrame(client.export_metrics(flow="train"))
arms = pd.DataFrame(client.export_runs(flow="train", status="ok"))
curves = client.export_metrics(flow="train")
results = client.export_runs(flow="train", status="ok")
```
The run id is on every row of both, so a curve joins to the arm it came from
and to the run page it was recorded on, and the wide table carries the commit
and the code digest, so an exported file says what produced its numbers.
[The CLI](../code/cli.md#fluksio-export) writes the same rows as csv, jsonl or
parquet, which is where an export belongs: in the script beside the analysis.
A node's numbers usually arrive inside a record, and the wide table follows
them in: every number a result carries is a column of its own, named by its
path (`metric.final_metrics.train_loss`) and `metrics="final_metrics.train_loss"`
selects one. Inputs work the same way, and are compared leaf by leaf, so two
configurations differing in one field give that field as the axis.
Pass either list to `pandas.DataFrame`. The
[CLI reference](../code/cli.md#fluksio-export) covers selection filters,
JSON Lines and Parquet.
### Bringing in runs made elsewhere
Runs are made by running a flow, so a history recorded before the move (an
MLflow experiment, say) has to be imported instead. Each entry names the flow,
the id the run had where it ran, and what it ended with:
Import historical runs without executing them:
```python
Client().import_runs([
{
"flow": "train",
"external_id": "mlflow:study-1/3f2a…",
"params": {"lr": 0.01},
"seed": 1001,
"status": "ok",
"created_at": "2026-03-01T10:00:00Z",
"started_at": "2026-03-01T10:00:01Z",
"finished_at": "2026-03-01T10:40:12Z",
"result": {"final_metrics": {"train_loss": 0.08}},
"metrics": [{"name": "train.train_loss", "step": 0, "ts": 1772359201.0, "value": 2.3}],
"artifacts": [{"name": "weights", "filename": "model.txt",
"media_type": "text/plain", "content_base64": "…"}],
"labels": ["imported"],
"group_id": "mlflow:study-1",
},
])
from fluksio.sdk.client import Client
Client().import_runs([{
"flow": "train",
"external_id": "previous-study:run-42",
"params": {"lr": 0.05, "epochs": 10},
"status": "ok",
"created_at": "2026-03-01T10:00:00Z",
"result": {"score": 0.94},
}])
```
The parameters are completed with the flow's declared values and checked the
way a submit checks them. The run is stored with the cause `import` and no
commit or code digest, since this engine never saw the code it ran. Importing
the same `external_id` again adds nothing, and `fluksio export runs` carries
it as a column, so an exported row joins back to where it came from.
The flow must exist, and parameters are checked against its inputs. Repeated
imports of the same `external_id` reuse the existing record. Imported runs
have no code commit or digest because the engine did not execute them.
Entries may also carry metric points and base64-encoded artifacts; see the
API schema for their fields.
### A dashboard, read against runs
A run records values under the same names a dashboard binds to (a run of
`study` writes `study.loss`) so a dashboard is already a way of looking at
one. **Open in dashboard** from a run or a comparison opens any dashboard with
`?runs=a,b,c` on it, and the widgets resolve from those runs instead of from
the live engine: each chart draws a line per run, the tiles that show one
number show the first, and the controls go quiet because there is nothing left
to publish to.
Choose **Open in dashboard** from a run or comparison to view those results
with a dashboard's widgets. Charts draw one line per run, scalar widgets use
the first run, and controls are disabled.
Nothing about the dashboard is specific to runs. The page built to watch a
training run happen is the page that shows the finished ones.
If the flow has no dashboard, the same menu can create one from its streaming
ports and outputs. The API equivalent is
`POST /api/v1/dashboards/from-flow/{name}`. The generated dashboard is
published and can be edited like any other dashboard.
If a flow has no dashboard yet, the same menu offers to build one from what
the flow declares: a chart per streaming port, a tile per output:
```
POST /api/v1/dashboards/from-flow/study
```
It publishes `study_results` and is an ordinary dashboard afterwards; editing
it is how it stops being generic.
### A tile that always shows the last few
The other direction is a chart pinned to runs rather than a page opened
against them, which is what a panel over a bench wants. Set a chart's source
to **Runs**, name the metric, and pick either the latest N of a flow, one
sweep, or specific runs. It re-reads on its own and whenever a run finishes.
### When a run draws nothing
A node restored from the [stage cache](#stage-caching) has its curve read back
from the run that recorded it. Delete that run and the reusing run is left with
a result and an empty curve, and the chart says so rather than looking broken.
## What this costs, compared
The repository ships a benchmark that measures submitting a run against a
Kedro project doing the same nothing:
```
fluksio — submit accepted median 15.3 ms
submit -> result median 60.8 ms
kedro — kedro run median 1109.5 ms
```
The difference is not the orchestration; it is that Fluksio does not boot a
project per run. The engine is already up, and the workers already have the
node's code compiled. On a 510-run sweep, that gap is about nine minutes of
pure startup that never happens.
To keep a chart showing recent experiments, set its source to **Runs** and
select the latest runs of a flow, a sweep or specific run IDs.
## See also
- [Getting started: data science](../getting-started/data-science.md) — the same
material as a first setup
- [Writing node code](../code/nodes.md) — generators, `fluksio.emit`, artifacts
- [Remote workers](../code/workers.md) — attaching the machine with the GPU
- [The HTTP API](../code/api.md) — every endpoint used above
- [Getting started: data science](../getting-started/data-science.md)
- [Writing node code](../code/nodes.md)
- [Remote workers](../code/workers.md)
- [The HTTP API](../code/api.md)
docs/concepts/state.md
+32 -51
View File
@@ -1,75 +1,56 @@
# Keeping state in a flow
Logic nodes are pure functions of their inputs. There is no `context`, no
`global`, no handle to a store. A node is called with the values of the
messages it declares and returns the values of the messages it provides.
Need a running total or a last-seen value? Make it a message. Function nodes
have no hidden `context` or global flow store.
That is deliberate: a node with hidden state cannot be run twice in parallel,
cannot be replayed, and cannot be moved to another worker. But plenty of real
automations need to remember something. This is how.
## Read and write the same message
## State is a message the node both reads and writes
A running total, a debounce timer, a last-seen reading: each is a value that
survives between runs. Give it a message name, declare it as both an input and
an output, and it is state:
Declare `reading` and `total` as inputs, and `total` as an output:
```python
def process(reading, total=0.0):
def process(reading, total):
return {"total": total + reading}
```
with `reading` and `total` as inputs and `total` as the output. Each run reads
the total the previous run wrote.
Add `total` to the flow's **Inputs** with a starting value of `0.0`. Each
execution reads the previous total and publishes the next one.
Two things make this work rather than loop forever:
- A node is never downstream of itself, so its own publication does not run
it again.
- A triggering input needs a starting value. Without one, validation reports
`self_loop_needs_initial`; a Python default alone does not supply the message.
- **A node never depends on itself.** The graph is built from message names, but
a node providing a message it also requires is not placed downstream of
itself. Publishing `total` does not re-run the node that wrote it.
- **The value must start somewhere.** The first run has nothing to read. Declare
the message as a flow input with a starting value; otherwise the flow reports
`self_loop_needs_initial` at edit time and refuses to publish.
## Feedback between two nodes
The port needs a default in the function signature (`total=0.0` above) so the
first call is well-formed even if the value is missing.
## Feeding a value back between two nodes
Sometimes the value comes from a different node, such as a controller reading back what
an estimator computed from its own last output. Written plainly that is a cycle,
and the validator rejects it, because a graph where A waits for B and B waits
for A can never start.
What is actually meant is that the fed-back value is *state*, not a trigger.
Say so, by marking the input non-triggering:
A controller may read an estimate that depends on its own previous output.
Ordinary triggering inputs would form a cycle. Mark the feedback input
**non-triggering** in the flow definition instead:
```json
{"name": "estimate", "port": "estimate", "dtype": "float", "trigger": false}
```
A non-triggering input:
It creates no dependency and does not wake the node. Whenever another input
runs the node, it reads the latest estimate. If no estimate exists yet, that
argument is omitted, so give it a Python default:
- creates no dependency, so it cannot form a cycle;
- never makes the node wait. If the message has no value yet, the port is
simply left out of the call, and the function's default applies;
- is read fresh from state whenever the node does run, for whatever reason.
```python
def process(reading, estimate=0.0):
return {"command": reading - estimate}
```
Use it for the back edge only. An input that should wake the node is an ordinary
input, and marking it non-triggering would mean the node never runs at all.
Use this on the feedback input; keep the inputs that should wake the node
triggering. Configure `trigger` through the API or flow JSON; the node panel
does not currently expose it.
## What still holds engine-side state
## Built-in state
Built-in nodes that are *about* time or change (rate limiting, filter-on-change,
delay, cron) keep what they need in the engine's own state backend, under keys
that never appear as messages. That is the engine's business, not a flow's: the
line is that node *code* you write never reaches for state, while node *types*
the engine ships may.
Storage and direct-I/O nodes are the other sanctioned exception, since talking
to a device or a database is stateful by nature.
Built-in nodes such as delay, rate limit and filter-on-change keep their own
timing or comparison state in the engine. Storage and device nodes also manage
the state needed for their connections. Your Function code still exchanges
state through declared messages.
## See also
- [Where a node's values come from](values.md) — ports, settings and flow inputs
- [Node types](../reference/node-types.md) — the built-ins that do keep engine-side state
- [Where a node's values come from](values.md)
- [Node types](../reference/node-types.md)
docs/concepts/values.md
+26 -33
View File
@@ -1,67 +1,60 @@
# Where a node's values come from
A node function is called with one argument per thing it needs, and every one
of them is declared somewhere you can see. There are two kinds.
Function nodes receive values as named arguments. Each comes from a message
port, a node setting or a Python default.
## Ports: what the graph carries
A port binds to a message name. Whatever last published that message is the
value the node is called with, and the wiring follows from the name rather than
from a line somebody drew.
A port binds to a message name and reads its current value. Declare inputs
`reading` and `setpoint`, and output `heat`:
```python
def process(reading, setpoint):
return {"heat": reading < setpoint}
```
with `reading` and `setpoint` as inputs and `heat` as an output.
Matching message names connect nodes. A port can have a different name from
its message: bind port `reading` to `house.temperature`, for example.
## Settings: constants of this node
A setting is a value that belongs to this node's code rather than to the graph:
how many retries, which unit, how long to wait. It is typed into the node's
**Settings** section, stored with the flow, and arrives as an argument like a
port:
Use settings for values that stay fixed while the flow runs: units, retries,
thresholds. Add them in the Function node's **Settings** section; they are
stored with the flow and passed beside the ports:
```python
def process(reading, unit="C"):
return {"shown": reading if unit == "C" else reading * 1.8 + 32}
```
Because both arrive by name, a setting cannot share a name with a port. The
node reports it rather than picking one.
With a `unit` setting, its configured value is passed to the function. Without
one, the Python default applies. A setting cannot share a name with a port.
Settings are part of the flow document, so changing one is an edit that gets
published, not something that happens at runtime. **A value that should change
while the flow runs is a message, not a setting.**
There is no `global` and no `flow` context. A constant several nodes need is a
node that provides it: one place it is set, an ordinary message out, and every
consumer visibly downstream of it.
Changing a setting edits the flow draft, so publish it to apply the change.
For a value that should change at runtime, use a message. To share a constant
between nodes, publish it from one node and let the others consume it.
## Flow inputs: what arrives from outside
Some messages are not computed by any node in the flow, such as a dashboard control
writes them, the API publishes them, a batch run passes them in. Declare those
as the flow's **inputs**, with the value they start from:
Dashboard controls, the API and batch runs can supply messages without a
producer node. Declare them as **Inputs** in the flow settings:
```json
{"inputs": [{"spec": {"name": "setpoint", "dtype": "float"}, "initial": 21.0}]}
```
Without that, the node reading `setpoint` waits for something nothing provides,
and the canvas says so. With it, the flow starts from 21.0 and whatever writes
the message afterwards takes over.
The initial value seeds an unset live message. Later publications take over;
changing the starting value does not overwrite a value already in state.
Use **Publish this value now** in the input row to send it immediately.
The canvas draws each one as a labelled endpoint feeding the nodes that read
it, the same way it draws a dashboard tile or another flow, so a value never
appears from nowhere. They are edited in the flow's own panel.
The canvas draws flow inputs as labelled endpoints. Without a producer or
declared input, a consuming node reports `unconnected_input`.
A batch flow's inputs are also its run parameters, and its **outputs** name what
a run reports as its result; see [Runs](runs.md).
A batch flow's inputs are run parameters. Its **Result** section selects the
messages returned when the run finishes. See [Runs](runs.md).
## See also
- [Flows, nodes and messages](flows.md) — how the graph is built from the names
- [Keeping state in a flow](state.md) — a value a node reads *and* writes
- [Writing node code](../code/nodes.md) — the practical side
- [Flows, nodes and messages](flows.md)
- [Keeping state in a flow](state.md)
- [Writing node code](../code/nodes.md)