Getting started: data science¶

Turn a Python experiment into a flow with recorded parameters, training curves and files you can find again. Goodbye, results_final_v3.csv.

This example uses a toy training curve, so you can try the whole workflow without a dataset or GPU. You need Python 3.12 or newer on Linux or macOS. Install Git too if you want flow changes recorded in version history.

1. Start your instance¶

Create a project directory and install Fluksio in a virtual environment:

mkdir research-flow
cd research-flow
python3 -m venv .venv
. .venv/bin/activate
pip install fluksio
fluksio serve

Fluksio creates .fluksio/ for this project's flows, run history and artifacts. It prints the initial admin password once and saves a token for local CLI commands. Keep the password somewhere safe.

At a terminal, serve opens a dashboard. Leave it running, then open a second terminal in research-flow and activate the same environment:

. .venv/bin/activate

For the browser dashboard, sign up at fluksio.com, add an instance, and enter the code it gives you:

fluksio enroll <code>

Select the instance in the portal. The CLI also works without enrolling.

2. Write a small flow¶

Save this as study.py in research-flow:

import json
import math
from pathlib import Path

import fluksio
from fluksio import Flow, Port, node


@node(
    requires=[Port("lr", "float"), Port("epochs", "int")],
    provides=[
        Port("loss", "float", stream=True),
        Port("weights", "artifact"),
        Port("final_loss", "float"),
    ],
)
def fit(lr, epochs):
    loss = 1.0
    for epoch in range(epochs):
        loss = math.exp(-lr * (epoch + 1))
        yield {"loss": loss}
    checkpoint = json.dumps({"loss": loss}).encode()
    return {
        "weights": fluksio.save_artifact(
            checkpoint, "weights.json", "application/json"
        ),
        "final_loss": loss,
    }


@node(requires=["weights"], provides=Port("score", "float"))
def evaluate(weights):
    checkpoint = json.loads(Path(fluksio.load_artifact(weights)).read_text())
    return 1.0 - checkpoint["loss"]


train = Flow(
    "train",
    nodes=[fit, evaluate],
    inputs=[Port("lr", "float", initial=0.05), Port("epochs", "int", initial=10)],
    outputs=["score", "final_loss", "weights"],
)

@node declares inputs and outputs. Matching message names connect the nodes: fit produces weights, and evaluate reads it. Flow lists the nodes, run parameters and final results.

Each yield records a loss value. The final return sends the checkpoint to the next node. Artifact ports carry a small reference; the file stays in the artifact store.

3. Sync and run¶

fluksio sync study
fluksio run train --lr 0.1 --epochs 20 --follow

sync uploads the flow and generates node code that imports your functions. Keep the project files available to the engine. Later run commands sync edits before submitting; --follow waits and prints metrics as they arrive. This run should finish with final_loss around 0.1353 and score around 0.8647.

In the portal, open Flows → train to see the graph, or Runs to inspect the parameters, metrics and results. From the CLI:

fluksio runs --flow train
fluksio artifacts <run-id>
fluksio artifacts <run-id> train.weights -o weights.json

Use the ID printed by run. Artifact names are flow-qualified; the list command shows the exact names to download.

4. Try a few learning rates¶

fluksio sweep train --param lr=0.01,0.05,0.1 --wait
fluksio export metrics --flow train -o curves.csv
fluksio export runs --flow train --status ok -o results.csv

The sweep creates three runs in one group. Select them on Runs to compare their curves. The exports give you a row per metric point and a row per run, ready for your analysis.

Python can read the same tables:

from fluksio.sdk.client import Client

client = Client()
results = client.export_runs(flow="train", status="ok")
curves = client.export_metrics(flow="train", name="train.loss")

Pass either list to pandas.DataFrame if you use pandas.

Bring your own experiment¶

Replace the toy calculation in fit with your training code. Keep ordinary Python helpers separate from artifact storage so you can still test them without an engine: save_artifact and load_artifact work only inside nodes. The public repository's examples/myresearch/ package shows this pattern across several files.

• Defaulted function arguments that are not input ports become node settings. Use flow inputs for values you want to vary between runs.
• Use fluksio.emit(loss=value) when metrics arrive through a framework callback instead of a loop you can yield from.
• Runs can reuse unchanged stages. Use --no-cache when you want every node to execute; see Stage caching.
• --seed records a seed and fills an input named seed when declared. Your code still needs to pass that value to its random number generator.

Declaring what a node needs¶

For heavy computation, declare resources on the node:

@node(requires=["dataset"], provides=Port("score", "float"),
      resources={"cpus": 2, "ram": "2G"})
def score_dataset(dataset):
    return {"score": calculate_score(dataset)}

The engine accounts for these resources and sets worker thread limits. GPU nodes can request resources={"gpus": 1}; check that the Workers screen reports your GPU, or set fluksio serve --gpus 1 when automatic detection does not cover it. Resource accounting does not impose a hard memory limit.

To run on another machine, attach a remote worker and install your project and its dependencies there.

Next steps¶

• Writing node code — ports, streaming, artifacts and packages.
• Runs — caching, comparisons and retries.
• CLI reference — command options and instance settings.