A tour of the core

Everything in pipekit-ops (import pipekit_ops), group by group: what it is, when to reach for it, and what it does today.

The core is pure Python with no dependencies. Its public surface is the 105 names of pipekit_ops.__all__, from 21 modules, one module per job, so you can learn it a group at a time. For why it is shaped this way (operators as classes, the carrier left open), read the background; for the shapes real pipelines take, read Patterns.

Every output on this page is from a real run of pipekit-ops 0.2.1.

The map

Group	Module	Names	Reach for it when
Foundations	_base/	Operator, ConfigMixin, Carrier, Sequential, Graph, Input, Node	always: writing a step, chaining steps, wiring a graph
Composition	compose	pipe, compose, compose_left, juxt, complement	you think in toolz
Building blocks	blocks	Identity, Const, Lambda, Sink	a no-op, a constant, a quick function, a write
Control flow	control	Branch, Switch, Try, Coalesce, Retry	a decision or a failure inside the pipeline
Flow limits	flow	Timeout, RateLimit, Until	a call that may hang, a service with a quota, a step repeated until done
Observers	observe	Tap, Snapshot, ShapeTrace, Profile, Histogram, Log	looking without changing the flow
Combination	combine	Fanout	one input, several named outputs
Dict plumbing	dicts	Get, GetIn, Pick, Dissoc, Rename, Merge, Assoc, UpdateIn, ValMap	reshaping the dicts a Graph, a Fanout or a STAC item carries, without a Lambda
Iterables	seq	Not, Filter, Remove, Reduce, PartitionAll, Concat, Zip, GroupBy, CountBy, Count, Unique, Take, SortBy	filtering, grouping and counting a list of items inside the pipeline
Caching	cache	Cache, Memoize	the same expensive call twice
Stores	store	Store, MemoryStore, DirectoryStore, store_scope	a cache that survives a restart, or that several processes share
Quality control	qc	Quarantine, AssertShape, AssertDType, AssertHasAttribute, AssertCallable, AssertThat, AssertSignature, QCError	a guard that fails loudly, or sets bad inputs aside
Shape inference	signature	Signature, compute_output_signature	catching a shape mismatch before running
Parallelism	parallel	ThreadMap, ProcessMap, AsyncMap, BatchedMap, check_pickleable	many items, one operator
Execution	execution	Map, executor_scope, MapError, Failed, Prefetch, Fold	many items, one operator, on any executor, with a bounded window
Serialisation	serial	dumps, loads, loads_sandboxed, from_state_sandboxed, register, unregister, allowed, upgrade_state	configuration as data; untrusted pipelines
Hashing	hashing	pipeline_hash	one key for "the same pipeline again": a registry, a cache, an orchestrator task
State	state	StatefulOperator, CarryState	state carried across calls: time steps, training, streams
Protocols	protocols	Predictor, FittableTransformer	accepting a model without importing its library
Named-array adapters	adapters	NamedArrayAdapter, LabelledAdapter, register_adapter, register_lazy_adapter, unregister_adapter, get_adapter, signature_of, MissingAdapterError, backend_of, backend_name, BackendError	reading a carrier's named dims, or the array backend it sits on, without importing its library
Import hooks	imports	on_import	running code once a module is imported, without importing it (JAX pytree registration)

pipekit_ops.testing (assert_round_trip) is the 22nd module; it is imported on its own and adds nothing to __all__.

The sister packages build on this core: time stepping and data assimilation in pipekit-cycle, training in pipekit-train, the model registry in pipekit-experiment, benchmarks in pipekit-evaluate, Array API dispatch in pipekit-array, JAX models in pipekit-jax.

The examples share one small operator:

from pipekit_ops import Operator


class Scale(Operator):
    """x ↦ a · x."""

    def __init__(self, factor: float) -> None:
        self.factor = factor  # stored under its argument's name → its config

    def _apply(self, x):
        return x * self.factor

Foundations

Operator

The class every step subclasses. You write two things: a constructor that stores its arguments under the same names, and _apply. The base class supplies the rest:

• configuration: get_config() is derived from the constructor's signature (ConfigMixin), and state is a JSON-safe record of the class and its config;
• rebuilding: Operator.from_state(state) finds the class among every imported subclass and calls it with the config;
• the pipe: a | b is a Sequential, flattened;
• two modes in one __call__: on data it runs _apply; on a graph placeholder (Input or Node) it records a node instead;
• class flags: forbid_in_yaml (holds a closure, so its config cannot be rebuilt) and _terminal (may only end a Sequential);
• backends: the array backends the operator runs on, such as ("numpy",) or ("numpy", "jax"). Left at None, nothing is checked. When set, an eager call whose carriers or arrays sit on another backend, or on several, raises BackendError before _apply runs, so inputs never move between backends silently.

import pipekit_ops as pk

Scale(2.0).state
# → {'module': '__main__', 'class': 'Scale', 'config': {'factor': 2.0}}
pk.loads(pk.dumps(Scale(2.0)))  # → Scale(factor=2.0)

Sequential

Apply operators left to right. It checks at construction that every item is an operator and that a _terminal one comes last; | flattens, so (a | b) | c and a | (b | c) are the same three steps. It has len, indexing and slicing, a Keras-style summary(signature) and a describe() tree, and the empty Sequential is the identity.

from pipekit_ops import Sequential, Signature

pipe = Scale(0.5) | Scale(2.0) | Scale(3.0)
pipe(10.0)  # → 30.0
len(pipe)  # → 3
pipe[1:]  # → Sequential([Scale(factor=2.0), Scale(factor=3.0)])
Sequential([])(7)  # → 7

print(pipe.summary(Signature(dims=(("y", 4), ("x", 4)), dtype="float32")))
# Operator   Input sig                      Output sig
# -----------------------------------------------------------------------
# [0] Scale  Signature((y=4, x=4) float32)  Signature((y=4, x=4) float32)
# [1] Scale  Signature((y=4, x=4) float32)  Signature((y=4, x=4) float32)
# [2] Scale  Signature((y=4, x=4) float32)  Signature((y=4, x=4) float32)

Graph, Input and Node

For flows that branch and merge, with several inputs and named outputs. Calling operators on Inputs (and on what those calls return) records the graph; Graph(inputs=…, outputs=…) wraps it, rejects cycles, and evaluates in topological order. Inputs bind by keyword or by position; outputs come back as a dict.

from pipekit_ops import Graph, Input, Lambda

a, b = Input("a"), Input("b")
s = Scale(2.0)(a)  # s = 2a
total = Lambda(lambda u, v: u + v, name="add")(s, b)  # s + b

g = Graph(inputs={"a": a, "b": b}, outputs={"s": s, "sum": total})
g(a=1.0, b=10.0)  # → {'s': 2.0, 'sum': 12.0}
g(1.0, 10.0)  # → the same; positional arguments bind in declaration order

A single-input graph composes into a Sequential ((Scale(2.0) | g1)(1.0) → {'y': 6.0} for a graph g1 with output y = 3x). The bigger-graph tutorial builds a full one.

Composition

Free functions that carry the toolz surface. They are not operators themselves (except what they return). Below, inc = Lambda(lambda v: v + 1).

Function	toolz ancestor	What it does	Example
pipe(value, *ops)	toolz.pipe	Sequential(ops)(value)
compose(*ops)	toolz.compose	right to left: compose(f, g, h) is Sequential([h, g, f])	compose(Scale(2.0), inc)(3.0) → 8.0
compose_left(*ops)	toolz.compose_left	left to right, like Sequential	compose_left(Scale(2.0), inc)(3.0) → 7.0
juxt(*ops)	toolz.juxt	each operator on the same input; a tuple	juxt(Identity(), Scale(2.0))(5.0) → (5.0, 10.0)
complement(pred)	toolz.complement	negates a predicate	complement(str.isdigit)("a") → True

juxt returns a plain function, so it cannot sit inside a Sequential; Fanout is the operator-shaped version.

Building blocks

Operator	Purpose	Holds a closure
Identity()	an explicit no-op; the default branch of Branch and Switch	no
Const(value)	return a fixed value, whatever the input	no
Lambda(fn, name=…)	wrap any callable; the escape hatch	yes
Sink(write_fn, name=…)	write as a side effect, return the input unchanged	yes

Sink returns its input, so a pipeline can continue past it. An operator that really ends a pipeline (writes and returns nothing) sets _terminal = True.

Control flow

Decisions and failures as steps inside the pipeline, rather than if and try around it.

Operator	Purpose
Branch(predicate, if_true, if_false=Identity())	a binary choice
Switch(key, cases, default=Identity())	dispatch on key(carrier)
Try(primary, fallback, on)	run fallback when primary raises one of on
Coalesce(sources, is_ok, on=())	the first source whose output passes is_ok
Retry(op, attempts, base_delay, on)	retry on on; attempt i waits base_delay · 2^(i−1)

from pipekit_ops import Coalesce, Const, Lambda, Retry, Try


def down(_):
    raise IOError("down")


Try(Lambda(down), Const(0.0), on=(IOError,))(1.0)  # → 0.0
Coalesce(
    [Lambda(down), Const(-1), Const(5)],
    is_ok=lambda r: r > 0,  # -1 fails the check
    on=(IOError,),  # down() falls through
)(0)  # → 5
Retry(flaky, attempts=5, base_delay=0.0, on=(ConnectionError,))(42)
# → 42, where flaky is an operator that raises ConnectionError twice, then succeeds

Try and Retry require their exception list, so neither is ever a silent catch-all; when every source fails, Coalesce raises RuntimeError. The ETL tutorial nests all three around a reader.

Flow limits

Operator	Purpose
Timeout(op, seconds, max_abandoned=8)	raise TimeoutError if op has not returned after seconds
RateLimit(op, calls, period)	at most calls starts of op in any period seconds, shared by every thread in the process
Until(op, pred, max_iter, on_max="raise")	apply op until pred holds, at most max_iter times

from pipekit_ops import RateLimit, Retry, Timeout, Until

read = Retry(
    RateLimit(Timeout(fetch, seconds=10.0), calls=10, period=1.0),
    attempts=3,
    base_delay=0.5,
    on=(TimeoutError,),
)
Until(NextPage(), IsLastPage(), max_iter=50)(first_page)

fetch, NextPage and IsLastPage are operators. A timed-out call's thread is abandoned, not killed, and finishes in the background; Timeout logs a warning once more than max_abandoned of them are still running. The rate limit does not cross processes, and placement matters: inside Retry, as here, every attempt waits for its own slot. pred is an operator, like every predicate in the core, and none of the three accepts an operator with an async def _apply.

Observers

The carrier flows through unchanged; something useful happens on the side.

Name	Purpose
Tap(fn, name=…)	call fn on the carrier; return the carrier
Snapshot() + .at(key)	capture intermediates by name in snapshot.captures
ShapeTrace(printer=print, mode="every" \| "diff_only", label=None, extra=None)	log shape and dtype at each step; extra(carrier) adds a line (a coordinate system, units)
Profile() + .wrap(op)	time each call; profile.report() gives calls, total and mean per operator
Histogram(bins=10, to_array=list) + .at(key)	capture distributions; to_array turns the carrier into numbers
Log(message=…, level="INFO", logger="pipekit_ops")	write one logging record per call, with the call count {n}, {type} and {signature}; holds only strings, so it round-trips

Snapshot, Profile and Histogram are controllers, not operators: the operators you place in the pipeline come from .at() and .wrap(), and the data they collect lives on the controller.

from pipekit_ops import Histogram, Lambda, Profile, Snapshot

snap, prof, hist = Snapshot(), Profile(), Histogram(bins=2)
pipe = Lambda(lambda v: [v, 2 * v], name="pair") | snap.at("mid") | hist.at("out")
pipe(1.0)  # → [1.0, 2.0]
snap.captures  # → {'mid': [1.0, 2.0]}
prof.wrap(Scale(3.0))(2.0)  # → 6.0;  prof.report() → {'Scale': {'calls': 1.0, ...}}

Combination

Fanout runs several operators on the same input and returns a dict of their outputs:

from pipekit_ops import Fanout, Lambda

Fanout({"n": Lambda(len), "u": Lambda(str.upper)})("ab")  # → {'n': 2, 'u': 'AB'}

Caching

Cache(inner) (alias Memoize) memoises an operator in memory, keyed by a SHA-256 of the input and the inner operator's pipeline_hash, so two operator classes with equal configs never share an entry. Arrays are keyed by their bytes, dtype and shape, so two large arrays never collide, and the bookkeeping is thread-safe, so one Cache can be shared by ThreadMap workers.

from pipekit_ops import Cache, Memoize

c = Cache(Scale(2.0))
c(1.0)
c(1.0)
c(2.0)
c.hits, c.misses  # → (1, 2)
Memoize is Cache  # → True

Stores

Cache(inner, store="scenes", key=Get("id")) keeps its entries in a named store instead of a private dict. The name is in the config; the store is bound around the call with store_scope, so the pipeline hashes the same on a laptop and on a cluster. key is an operator that picks what to key on, here a STAC item's ID, so two dicts with the same id share one entry. A DirectoryStore writes one pickle file per key, atomically, so a rerun after a crash skips the work already done, and processes on one file system can share it. Pickle loads run code, so point a DirectoryStore only at a directory you control.

from pipekit_ops import Cache, DirectoryStore, Get, Identity, store_scope

cached = Cache(Identity(), store="scenes", key=Get("id"))
with store_scope(scenes=DirectoryStore("/tmp/scenes")):
    cached({"id": "S2A", "cloud": 3})
    cached({"id": "S2A", "cloud": 9})  # → {'id': 'S2A', 'cloud': 3}
cached.hits, cached.misses  # → (1, 1)

MemoryStore() is the in-process store; any object with __contains__, load and save is a Store. An unbound name raises LookupError, and Map carries the bound stores to its workers: threads see the scope, and process workers get the stores pickled (a DirectoryStore pickles as its path).

Quality control

Assertions as pass-through operators: a research-time "this won't break" becomes a permanent guard.

Operator	Purpose	Holds a closure
Quarantine(check, sentinel, on_quarantine=None)	on failure call on_quarantine and return sentinel; never raises	yes
AssertShape(expected_shape)	raise QCError on a shape mismatch; None matches any size	no
AssertDType(expected_dtype)	raise QCError on a dtype mismatch	no
AssertHasAttribute(name, value=…)	raise QCError if an attribute is missing or wrong	no
AssertCallable(predicate, message=…)	raise QCError if a predicate fails	yes
AssertThat(pred, message=…)	raise QCError if the operator pred returns a falsy value	no
AssertSignature(pattern, dtype=None, missing="raise")	raise QCError unless the adapter's dims fit pattern ("band y x", "... y x"); also checked by Sequential.summary	no

AssertShape((2, 4))(x)  # x.shape == (2, 3)
# QCError: AssertShape: axis 1 expected 4, got 3 (expected=(2, 4), actual=(2, 3)).

The checks read shape, dtype and attributes and never look inside the data, so they work on any carrier. QCError subclasses AssertionError. Quarantine is the batch-friendly one: the inference and ETL tutorials use it to set bad records aside and keep going.

Shape inference

Signature is an immutable record of named dimensions and a dtype; a size of None marks a dimension dynamic. An operator declares how it changes a signature by overriding compute_output_signature; returning None means "this operator does not track shape", and Sequential.summary shows ? for it.

from pipekit_ops import Signature

sig = Signature(dims=(("time", 12), ("y", 256), ("x", 256)), dtype="float32")
sig.format()  # → '(time=12, y=256, x=256) float32'
sig.drop_dims(("time",))  # → Signature((y=256, x=256) float32)
sig.replace_dims({"x": 128})  # → Signature((time=12, y=256, x=128) float32)

Containers report what they return: - Cache, Retry and Profile.wrap report their inner operator's signature. - Try, Branch, Switch and Coalesce report the signature their branches agree on, else None. - Fanout and the parallel maps report None, because they return a dict or a list.

A Graph propagates signatures node by node: output_signatures gives each output's, and summary prints the table. A node with several inputs gets its operator's signature only when the inputs agree; otherwise it shows ?:

from pipekit_ops import Graph, Input, Operator


class MeanOver(Operator):
    """Mean over one named dimension: drops it from the signature."""

    def __init__(self, dim: str) -> None:
        self.dim = dim

    def _apply(self, x):
        return x.mean(axis=0)

    def compute_output_signature(self, signature):
        return None if signature is None else signature.drop_dims((self.dim,))


x = Input("x")
g = Graph(inputs={"x": x}, outputs={"climatology": MeanOver("time")(x)})
print(g.summary(sig))
# Node                 Input sig                        Output sig
# ----------------------------------------------------------------
# [0] MeanOver(x)      (time=12, y=256, x=256) float32  (y=256, x=256) float32
# → climatology = [0]                                   (y=256, x=256) float32

The payoff grows as the operators you write (and domain libraries) declare theirs.

Parallelism

Four shapes, one wrapper each. Every wrapper takes an iterable and returns a list in input order. Each is a preset of Map, so it pulls its input lazily and, on on_error="raise", cancels what has not started. ProcessMap keeps the platform's start method for one more minor release and warns (FutureWarning) where that is not spawn.

Shape	When	Constraint	Operator
Threads	I/O-bound: object-store reads, downloads	the I/O must release the GIL	ThreadMap(op, n_workers=8)
Processes	CPU-bound work	every step must pickle	ProcessMap(op, n_workers=8, on_error="raise" \| "log_and_continue")
Async	I/O on one thread: web services	async def _apply, or a sync operator run in a thread	AsyncMap(op, semaphore=8)
Batches	GPU inference, vectorised code	the inner operator takes a list of items	BatchedMap(op, batch_size=8, flatten=True)

from pipekit_ops import AsyncMap, BatchedMap, Lambda, ThreadMap

ThreadMap(Scale(2.0), n_workers=2)([1, 2, 3])  # → [2.0, 4.0, 6.0]
add_up = Lambda(lambda b: [sum(b)], name="sum")
BatchedMap(add_up, batch_size=2)([1, 2, 3, 4, 5])  # → [3, 7, 5]
AsyncMap(Scale(2.0), semaphore=2)([1, 2])  # → [2.0, 4.0]

Pickling. Process pools need every operator to pickle, and closures (Lambda, Tap, Sink, a Branch predicate), open handles and captured random generators do not. Operators that hold a closure carry forbid_in_yaml = True, and check_pickleable(op) walks a whole pipeline, however deeply nested, and returns the flagged operators before you fan out:

from pipekit_ops import Branch, Cache, Identity, Tap, ThreadMap, check_pickleable

pipe = ThreadMap(Cache(Scale(2.0) | Tap(print) | Branch(bool, Identity())))
[type(op).__name__ for op in check_pickleable(pipe)]  # → ['Tap', 'Branch']

Async. AsyncMap(...)(items) drives its own event loop, so it works in scripts. Inside a running loop (a FastAPI or LitServe handler) it raises a RuntimeError that says to await its run method instead:

async def handler(items):
    return await AsyncMap(Scale(2.0)).run(items)  # → [2.0, 4.0] for [1, 2]

Map

Map(op, executor=...) is the one engine the four wrappers are built on: executor is "serial", "thread", "process" (spawn by default), "async", or "scope", which borrows whatever concurrent.futures.Executor an executor_scope binds (a dask or Ray executor included) and never shuts it down. The executor is not in the config, so the pipeline hashes the same wherever it runs. Input is pulled lazily and at most max_in_flight items are held at once; on_error is "raise" (fail fast, cancel what has not started), "collect" (raise a MapError with the partial results) or "log_and_continue". Process pools and borrowed executors check that the operator pickles before any worker starts.

from concurrent.futures import ThreadPoolExecutor

from pipekit_ops import Identity, Map, MapError, Prefetch, executor_scope

Map(Scale(2.0), executor="thread", n_workers=2)([1, 2, 3])  # → [2.0, 4.0, 6.0]
try:
    Map(Inverse(), on_error="collect")([1, 0, 2])
except MapError as error:
    error.results
    # → [1.0, Failed(index=1, exception=ZeroDivisionError('division by zero')), 0.5]
with executor_scope(ThreadPoolExecutor(max_workers=4)):
    Map(Scale(2.0), executor="scope", max_in_flight=8)([1, 2])  # → [2.0, 4.0]
list(Map(Scale(2.0), executor="thread").iter([1, 2]))  # → [2.0, 4.0], lazily
(Prefetch(Scale(2.0), depth=4) | Map(Identity()))([1, 2, 3])  # → [2.0, 4.0, 6.0]

Inverse is an operator returning 1 / x. Prefetch(op, depth) reads up to depth items ahead of the consumer, and Fold(op, initial, chunksize) folds chunks in parallel through Map. The details are in the execution design.

Deliberately left out. There is no automatic parallelism: Sequential is sequential, and Graph evaluates in one process. There is no graph scheduler either. Map(executor="scope") can send the items of one step to a Dask or Ray executor, but scheduling whole pipelines across machines is left to Dask, Ray, Beam and orchestrators; pipekit keeps operators pickleable so they can.

Serialisation

Name	What it does
dumps(op)	JSON-encode op.state, keys sorted, nested operators included
loads(s)	rebuild the whole tree through Operator.from_state; the defining modules must be imported
register(cls)	add a class to the sandbox allowlist (also a class decorator)
unregister(cls), allowed()	remove from, and list, the allowlist
loads_sandboxed(s)	loads, but PermissionError for any class in the tree that is not registered
from_state_sandboxed(state)	the same check on a record already in memory

The sandboxed loader is for pipelines you did not write (uploaded by users, read from a queue): only classes you registered can be built, at any depth, and every class is checked before anything is constructed. Nothing is allowed by default, pipekit's own containers included.

import pipekit_ops as pk

pk.dumps(pk.Const(3))
# → '{"class": "Const", "config": {"value": 3}, "module": "pipekit_ops.blocks"}'
pk.loads(pk.dumps(Scale(2.0)))  # → Scale(factor=2.0)
pk.loads_sandboxed(pk.dumps(Scale(2.0)))  # PermissionError: not registered
pk.register(Scale)
pk.loads_sandboxed(pk.dumps(Scale(2.0)))  # → Scale(factor=2.0)
pk.loads(pk.dumps(pk.Lambda(abs)))  # RuntimeError: marked forbid_in_yaml

Nested pipelines round-trip too. A child operator appears in its parent's config as its own state record, a Graph records every node and its wiring, and an exception class (Try.on, Retry.on) is written as {"exception": "builtins.KeyError"}. pipeline_hash hashes that whole record, so two pipelines share a hash exactly when they are configured the same:

from pipekit_ops import Cache, Retry, Sequential, ThreadMap, pipeline_hash

pipe = ThreadMap(Cache(Retry(Scale(2.0), attempts=3, base_delay=0.5, on=(OSError,))))
back = pk.loads(pk.dumps(pipe))
back.state == pipe.state  # → True
pipeline_hash(back) == pipeline_hash(pipe)  # → True

pk.register(Sequential)
pk.loads_sandboxed(pk.dumps(Sequential([Scale(2.0), pk.Identity()])))
# PermissionError: '__main__.Scale' is not in the allowlist (checked first)

pipekit-experiment's HydraConfigLoader does the same with _target_ YAML: to_yaml(op) writes any pipeline, from_yaml loads it through Hydra, and from_yaml_sandboxed loads untrusted YAML through the allowlist without ever calling Hydra's instantiate.

State

Most operators are stateless: _apply(carrier) → carrier. Some work carries state from call to call: a data-assimilation cycle (forecast, analysis, forecast…), a training loop (optimiser state, step, metrics), a streaming statistic, a recurrent model. A StatefulOperator implements _apply(carrier, state) → (carrier, state), and a CarryState is the base for that state, with to_dict / from_dict for checkpointing. A Sequential that contains a stateful step threads the state through and returns (carrier, state); stateless steps pass it along untouched.

A streaming mean: after values x₁, …, xₙ the state is the running sum Sₙ = ∑ᵢ xᵢ = Sₙ₋₁ + xₙ and the count n, and the step emits x̄ₙ = Sₙ / n.

from pipekit_ops import CarryState, StatefulOperator


class MeanState(CarryState):
    def __init__(self, total: float, n: int) -> None:
        self.total = total  # Sₙ = ∑ᵢ xᵢ
        self.n = n  # n


class StreamMean(StatefulOperator):
    def _apply(self, x, state):
        st = MeanState(state.total + x, state.n + 1)  # Sₙ = Sₙ₋₁ + xₙ;  n ← n + 1
        return st.total / st.n, st  # x̄ₙ = Sₙ / n


pipe = Scale(1.0) | StreamMean()  # a stateless step, then a stateful one
x_bar, st = pipe(4.0, state=MeanState(0.0, 0))  # x̄₁ = 4.0
x_bar, st = pipe(8.0, state=st)  # x̄₂ = 6.0
st  # → MeanState(total=12.0, n=2)
MeanState.from_dict(st.to_dict()) == st  # → True
pipe(4.0)  # TypeError: ... called without a `state` argument

A Graph threads state the same way. Call it with state= and it returns (outputs, state); the state passes through its stateful nodes in topological order (the order describe() lists):

from pipekit_ops import Graph, Input

x = Input("x")
g = Graph(inputs={"x": x}, outputs={"mean": StreamMean()(Scale(1.0)(x))})
out, st = g(x=4.0, state=MeanState(0.0, 0))  # out == {"mean": 4.0}
out, st = g(x=8.0, state=st)  # out == {"mean": 6.0}

pipekit-cycle's Cycle, DACycle and EnsembleDACycle and pipekit-train's TrainingLoop are built on these two classes; see the data-assimilation and training tutorials.

Protocols

Model wrappers in the sister libraries rely on two duck-typed seams, written down once as runtime_checkable Protocols: Predictor (anything with predict(x)) and FittableTransformer (anything with fit(x) and transform(x)). isinstance(model, Predictor) checks that the methods exist, not their signatures, so a scikit-learn estimator, a JAX model wrapper or your own class all qualify without importing pipekit. The cycle and experiment packages define more such seams (ForwardModel, AnalysisStep, ModelRegistry, …); see Protocols.

Limits today

The limits this page used to list have been fixed: - nested pipelines round-trip; - a Graph's state records every node, threads carry state and has a summary; - containers report the signature of what they return; - CarryState is consistently unhashable; - ProcessMap reports failures through logging; - AsyncMap explains itself inside a running loop; - ShapeTrace takes an extra line.

The design questions still open are tracked in the GeoStack design reports and pipekit's issues.