05 — Advanced patterns

Real research pipelines are graphs, not chains: shared preprocessing feeding several branches, studies driving whole graphs, checkpoint bundles. This notebook shows the patterns we use daily.

5.1 — Shared trunks across graphs

Two investigations often share the same expensive trunk (preprocessing on fixed data) with different heads. You don’t need one mega-graph for that: build one graph per head — the persistent cache dedupes the trunk across graphs, processes, and days. Center below fits exactly once for both heads.

from soma import Filter, Graph

class Center(Filter):
    _cache_version = "nb05-center-v1"

    def fit(self, x, y=None):
        return {"mean": sum(x) / len(x)}
    def forward(self, x, state):
        return [v - state["mean"] for v in x]

class Clamp(Filter):
    _cache_version = "nb05-clamp-v1"

    def __init__(self, limit=1.0, **kwargs):
        super().__init__(limit=limit, **kwargs)
    def forward(self, x, state):
        return [max(-self.limit, min(self.limit, v)) for v in x]

class Threshold(Filter):
    _cache_version = "nb05-threshold-v1"

    def forward(self, x, state):
        return [1.0 if v > 0 else 0.0 for v in x]

TRAIN = [10.0, 20.0, 30.0, 40.0, 50.0]

head_a = Graph()
head_a.node("center", Center())
head_a.node("clamp", Clamp(limit=5.0))
head_a.edge("center", "clamp")
head_a.fit(TRAIN)                    # Center actually fits here...

head_b = Graph()
head_b.node("center", Center())
head_b.node("threshold", Threshold())
head_b.edge("center", "threshold")
head_b.fit(TRAIN)                    # ...and is served from cache here

print("clamped:   ", head_a.forward([12.0, 60.0, -80.0]))
print("thresholded:", head_b.forward([12.0, 60.0, 4.0]))

clamped:    [-5.0, 5.0, -5.0]
thresholded: [0.0, 1.0, 0.0]

Forks with the DSL: `|` and `.collect()`

When you DO want one graph with parallel branches, the DSL builds it: A >> (B | C) >> D runs B and C concurrently (executor threads) and D receives a dict keyed by upstream node id — the fan-in contract:

class SumBranches(Filter):
    _cache_version = "nb05-sum-v1"

    def forward(self, x, state):
        # Fan-in contract: x = {upstream_node_id: plain_json_value}
        print("fan-in keys:", sorted(x.keys()))
        return [sum(sum(branch) for branch in x.values())]

fork = Graph.somatize(
    Center() >> (Clamp(limit=5.0) | Threshold()) >> SumBranches()
)
fork.fit(TRAIN)
print("fork result:", fork.forward([40.0, 20.0]))

fan-in keys: ['clamp', 'threshold']
fan-in keys: ['clamp', 'threshold']
fork result: [1.0]

fork   # el diagrama muestra el fan-out y el fan-in

5.2 — Graph checkpoints (`.somack`)

g.save(path) writes a portable bundle (manifest + safetensors states); Graph.load(path) rebuilds the graph with its trained states — the filter classes must be importable. Use it to ship a trained pipeline to another machine; use the cache (notebook 02) for everything else.

5.3 — A study over a full graph, with seeds

The pattern that puts it all together — HPO over a pipeline where:

the fixed preprocessing is computed once across ALL trials (cache),
every config runs on 3 seeds (seeds=[...]),
a crash at any point resumes with exact hits.

from soma import Study
import random as _random

class Model(Filter):
    _cache_version = "nb05-v1"
    def __init__(self, lr=0.1, seed=0, **kwargs):
        super().__init__(lr=lr, seed=seed, **kwargs)
    def fit(self, x, y=None):
        rng = _random.Random(self.seed)
        w = rng.gauss(1.0, 0.1) * (1.0 - self.lr)   # toy "training"
        return {"w": w}
    def forward(self, x, state):
        return [state["w"] * v for v in x]

X = [1.0, 2.0, 3.0, 4.0]
Y = [1.1, 2.0, 3.2, 3.9]

def train(trial):
    g = Graph()
    g.node("center", Center())                                 # shared → cached once
    g.node("model", Model(lr=trial["lr"], seed=trial["seed"]))
    g.edge("center", "model")
    g.fit(X, Y, seed=trial["seed"])
    pred = g.forward(X, seed=trial["seed"])
    mse = sum((p - yv) ** 2 for p, yv in zip(pred, Y)) / len(Y)
    return {"mse": mse}

study = Study(
    "graph-hpo",
    search_space=[{"type": "float", "name": "lr", "low": 0.01, "high": 0.5, "scale": "log"}],
    strategy="bayesian", n_trials=6,
    objectives=[("mse", "minimize")],
    seeds=[1, 2, 3],
    seed=42, tracking=False,
)
study.run(train)
print(f"{study.n_trials} trials (6 configs x 3 seeds)")
best = study.best_trial
print(f"best: lr={best['params']['lr']:.3f} "
      f"seed={best['params']['seed']} mse={best['metrics']['mse']:.4f}")

18 trials (6 configs x 3 seeds)
best: lr=0.038 seed=3 mse=6.5131

5.4 — Distribution

The same graph runs remotely: g.worker("http://host:8080", token=...) routes execution to a somatize-worker (filters travel as cloudpickle, states come back). Nodes can pin placement with g.node(f, target="gpu"). See the workers guide in the docs.