Make the example's __main__ actually run

It claimed to run the pipeline with no engine involved, and could not: the
node bodies it called save and load artifacts, which raise outside a node by
design. Each node is now a thin wrapper over a plain function — make_rows,
train_curve, score — and __main__ calls those, which is the split the sandbox
already demonstrates and the one worth copying.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-26 22:57:36 +02:00
co-authored by Claude Opus 5
parent 4f3eaf950c
commit 8f71b638b6
5 changed files with 45 additions and 19 deletions
+7 -4
View File
@@ -1,7 +1,10 @@
"""A research package that knows nothing about Fluksio, except where it says so. """A research package that knows nothing about Fluksio, except where it says so.
Everything here is ordinary Python: `prepare`, `fit` and `evaluate` are called Each node is a thin wrapper over a plain function: `prepare` stores what
directly by `python -m myresearch.pipeline`, and the decorators on them only `make_rows` made, `fit` yields what `train_curve` computed, `evaluate` reports
say what a node of each would look like. `fluksio sync examples/myresearch` is what `score` worked out. That split is the point — `fluksio.save_artifact` and
what turns that into a flow. `load_artifact` only work *inside* a node, so a body calling them cannot be run
by hand, while `python -m myresearch.pipeline` runs the arithmetic beside them
with no engine anywhere. `fluksio sync examples/myresearch` is what turns the
wrappers into a flow.
""" """
+12 -4
View File
@@ -8,6 +8,14 @@ import fluksio
from fluksio import Port, node from fluksio import Port, node
def make_rows(source: str = "builtin", limit: int = 512) -> dict:
"""The training set itself, with no Fluksio in it."""
return {
"source": source,
"rows": [{"x": index / limit, "y": (index % 7) / 7} for index in range(limit)],
}
@node(provides=[Port("dataset", "artifact"), Port("rows", "int")]) @node(provides=[Port("dataset", "artifact"), Port("rows", "int")])
def prepare(source: str = "builtin", limit: int = 512): def prepare(source: str = "builtin", limit: int = 512):
"""Make the training set and store it, returning a reference to it. """Make the training set and store it, returning a reference to it.
@@ -15,11 +23,11 @@ def prepare(source: str = "builtin", limit: int = 512):
`source` and `limit` have defaults and are not ports, so they become the `source` and `limit` have defaults and are not ports, so they become the
node's settings: the canvas can tune them without touching this file. node's settings: the canvas can tune them without touching this file.
""" """
rows = [{"x": index / limit, "y": (index % 7) / 7} for index in range(limit)] payload = make_rows(source, limit)
payload = json.dumps({"source": source, "rows": rows}).encode() blob = json.dumps(payload).encode()
return { return {
"dataset": fluksio.save_artifact(payload, "dataset.json", "application/json"), "dataset": fluksio.save_artifact(blob, "dataset.json", "application/json"),
"rows": len(rows), "rows": len(payload["rows"]),
} }
+6 -1
View File
@@ -8,6 +8,11 @@ import fluksio
from fluksio import Port, node from fluksio import Port, node
def score(lr: float, epochs: int) -> float:
"""How good the model came out, with no Fluksio in it."""
return round(1 - lr / (1 + epochs), 4)
@node(requires=["weights"], provides=Port("score", "float")) @node(requires=["weights"], provides=Port("score", "float"))
def evaluate(weights): def evaluate(weights):
"""Score the model, returning the number rather than a dict. """Score the model, returning the number rather than a dict.
@@ -16,4 +21,4 @@ def evaluate(weights):
to a bare return: the generated node wraps it in the message it belongs to. to a bare return: the generated node wraps it in the message it belongs to.
""" """
trained = json.loads(open(fluksio.load_artifact(weights)).read()) trained = json.loads(open(fluksio.load_artifact(weights)).read())
return round(1 - trained["lr"] / (1 + trained["epochs"]), 4) return score(trained["lr"], trained["epochs"])
+11 -8
View File
@@ -12,9 +12,9 @@ provides a message another requires, never because one imported the other.
from __future__ import annotations from __future__ import annotations
from fluksio import Flow, Port, use from fluksio import Flow, Port, use
from myresearch.data import augment, prepare from myresearch.data import augment, make_rows, prepare
from myresearch.evaluate import evaluate from myresearch.evaluate import evaluate, score
from myresearch.train import fit from myresearch.train import fit, train_curve
train = Flow( train = Flow(
"train", "train",
@@ -41,8 +41,11 @@ finetune = Flow(
if __name__ == "__main__": if __name__ == "__main__":
# Run it here, with no engine involved: the decorators changed nothing # Run the research here, with no engine involved. Not the node bodies: they
# about calling these functions. # save and load artifacts, and there is nothing to save to out here. The
dataset = prepare(limit=64) # functions they wrap are ordinary Python, which is why they are worth
losses = list(fit(dataset["dataset"], lr=0.05, epochs=5)) # keeping separate — the arithmetic stays yours to run by hand.
print("losses:", [round(step["loss"], 4) for step in losses]) data = make_rows(limit=64)
print("rows:", len(data["rows"]))
print("losses:", [round(loss, 4) for loss in train_curve(0.05, 5)])
print("score:", score(0.05, 5))
+9 -2
View File
@@ -9,6 +9,14 @@ import fluksio
from fluksio import Port, node from fluksio import Port, node
def train_curve(lr: float, epochs: int) -> list[float]:
"""The loss per epoch, with no Fluksio in it."""
return [
math.exp(-lr * epoch * 10) * (1 + 0.05 * (epoch % 3)) / (1 + lr)
for epoch in range(epochs)
]
@node( @node(
requires=["dataset", Port("lr", "float")], requires=["dataset", Port("lr", "float")],
provides=[ provides=[
@@ -38,8 +46,7 @@ def fit(dataset, lr, epochs=25):
""" """
rows = json.loads(open(fluksio.load_artifact(dataset)).read())["rows"] rows = json.loads(open(fluksio.load_artifact(dataset)).read())["rows"]
loss = 1.0 loss = 1.0
for epoch in range(epochs): for loss in train_curve(lr, epochs):
loss = math.exp(-lr * epoch * 10) * (1 + 0.05 * (epoch % 3)) / (1 + lr)
yield {"loss": loss} yield {"loss": loss}
weights = json.dumps({"lr": lr, "epochs": epochs, "n": len(rows)}).encode() weights = json.dumps({"lr": lr, "epochs": epochs, "n": len(rows)}).encode()
return { return {