diff --git a/examples/myresearch/__init__.py b/examples/myresearch/__init__.py index 9a979d9..32537b1 100644 --- a/examples/myresearch/__init__.py +++ b/examples/myresearch/__init__.py @@ -1,7 +1,10 @@ """A research package that knows nothing about Fluksio, except where it says so. -Everything here is ordinary Python: `prepare`, `fit` and `evaluate` are called -directly by `python -m myresearch.pipeline`, and the decorators on them only -say what a node of each would look like. `fluksio sync examples/myresearch` is -what turns that into a flow. +Each node is a thin wrapper over a plain function: `prepare` stores what +`make_rows` made, `fit` yields what `train_curve` computed, `evaluate` reports +what `score` worked out. That split is the point — `fluksio.save_artifact` and +`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. """ diff --git a/examples/myresearch/data.py b/examples/myresearch/data.py index ed4824c..52f6578 100644 --- a/examples/myresearch/data.py +++ b/examples/myresearch/data.py @@ -8,6 +8,14 @@ import fluksio 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")]) def prepare(source: str = "builtin", limit: int = 512): """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 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 = json.dumps({"source": source, "rows": rows}).encode() + payload = make_rows(source, limit) + blob = json.dumps(payload).encode() return { - "dataset": fluksio.save_artifact(payload, "dataset.json", "application/json"), - "rows": len(rows), + "dataset": fluksio.save_artifact(blob, "dataset.json", "application/json"), + "rows": len(payload["rows"]), } diff --git a/examples/myresearch/evaluate.py b/examples/myresearch/evaluate.py index 090c373..dcfbc97 100644 --- a/examples/myresearch/evaluate.py +++ b/examples/myresearch/evaluate.py @@ -8,6 +8,11 @@ import fluksio 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")) def evaluate(weights): """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. """ trained = json.loads(open(fluksio.load_artifact(weights)).read()) - return round(1 - trained["lr"] / (1 + trained["epochs"]), 4) + return score(trained["lr"], trained["epochs"]) diff --git a/examples/myresearch/pipeline.py b/examples/myresearch/pipeline.py index e516503..731a78c 100644 --- a/examples/myresearch/pipeline.py +++ b/examples/myresearch/pipeline.py @@ -12,9 +12,9 @@ provides a message another requires, never because one imported the other. from __future__ import annotations from fluksio import Flow, Port, use -from myresearch.data import augment, prepare -from myresearch.evaluate import evaluate -from myresearch.train import fit +from myresearch.data import augment, make_rows, prepare +from myresearch.evaluate import evaluate, score +from myresearch.train import fit, train_curve train = Flow( "train", @@ -41,8 +41,11 @@ finetune = Flow( if __name__ == "__main__": - # Run it here, with no engine involved: the decorators changed nothing - # about calling these functions. - dataset = prepare(limit=64) - losses = list(fit(dataset["dataset"], lr=0.05, epochs=5)) - print("losses:", [round(step["loss"], 4) for step in losses]) + # Run the research here, with no engine involved. Not the node bodies: they + # save and load artifacts, and there is nothing to save to out here. The + # functions they wrap are ordinary Python, which is why they are worth + # keeping separate — the arithmetic stays yours to run by hand. + 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)) diff --git a/examples/myresearch/train.py b/examples/myresearch/train.py index ddcdf7b..2761713 100644 --- a/examples/myresearch/train.py +++ b/examples/myresearch/train.py @@ -9,6 +9,14 @@ import fluksio 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( requires=["dataset", Port("lr", "float")], provides=[ @@ -38,8 +46,7 @@ def fit(dataset, lr, epochs=25): """ rows = json.loads(open(fluksio.load_artifact(dataset)).read())["rows"] loss = 1.0 - for epoch in range(epochs): - loss = math.exp(-lr * epoch * 10) * (1 + 0.05 * (epoch % 3)) / (1 + lr) + for loss in train_curve(lr, epochs): yield {"loss": loss} weights = json.dumps({"lr": lr, "epochs": epochs, "n": len(rows)}).encode() return {