Pipeline Composition

Composition turns small typed Flows into a reusable graph you can register, mutate, and drop inside a larger graph — without a second runtime. This example registers a two-stage pipeline, then does two things to it: extends it after construction, and wraps it as a single Flow.

The pieces

Each payload is an @io record; each stage is a Flow[In, Out]. Nothing here knows how it will be composed.

import retriever
from retriever.flow import Flow, Pipeline, Rate, Latest, io

@io
class CounterOut:
    value: int
    aux: int

@io
class ProcOut:
    value: int

class Counter(Flow[None, CounterOut]):
    def reset(self): self.count = 0
    def step(self, _):
        self.count += 1
        return CounterOut(value=self.count, aux=99)

class BiasPolicy(Flow[ProcIn, ProcOut]):
    def step(self, input: ProcIn) -> ProcOut:
        return ProcOut(value=input.value + input.bias)

Register a pipeline with surfaced ports

@retriever.register_pipeline names a graph and declares which internal ports are visible from the outside. counter.then(policy, map=..., sync=...) wires the stages; surface_policy="explicit" means only the listed ports cross the boundary.

@retriever.register_pipeline(
    "golden.composable_counter",
    surface_policy="explicit",
    input_ports=["policy.bias"],
    output_ports=["counter.aux", "policy.value"],
    overwrite=True,
)
def build_composable_counter() -> Pipeline:
    pipe = Pipeline("golden.composable_counter")
    with pipe:
        counter = (Counter()    @ Rate(hz=10)).named("counter")
        policy  = (BiasPolicy() @ Rate(hz=10)).named("policy")
        counter.then(policy, map={"value": "value"}, sync=Latest())
    return pipe

Extend a declared pipeline

Because the graph is addressable by node name, you can replace a stage and inject an input on a live pipeline, then step it — no rebuild.

pipe = build_composable_counter()
pipe.replace(pipe.select_flow("policy"), OverridePolicy(delta=100) @ Rate(hz=10))
pipe.inject_input("policy", "bias", 3, timestamp=0.0)
result = pipe.step(now=0.0)
print("policy output after replacement:", result.outputs["policy"])

Compose a pipeline as a Flow

retriever.build_pipeline_flow("golden.composable_counter") hands the registered pipeline back as an ordinary Flow, so it nests inside a larger graph and its surfaced output is a typed record.

outer = Pipeline("golden.outer_composable_counter")
with outer:
    bias_source = BiasSource(bias=4) @ Rate(hz=10)
    sub  = (retriever.build_pipeline_flow("golden.composable_counter") @ Rate(hz=10)).named("stage")
    sink = DecisionPrinter() @ Rate(hz=10)
    bias_source.then(sub, sync=Latest())
    sub.then(sink, sync=Latest())
result = outer.step(now=0.0)
print("wrapped stage output:", result.outputs["stage"])

pixi run -e golden-retriever demo-composable-pipelines

What you get

Real output. value=104 is Counter (1) + injected bias (3) + OverridePolicy delta (100); the wrapped stage surfaces both aux and value as one typed record:

=== Extend Declared Pipeline ===
internal flows: ['counter', 'policy']
policy output after replacement: ProcOut(value=104)

=== Compose Pipeline As Flow ===
[outer] value=5 aux=99
wrapped stage output: Golden_Composable_CounterOutput(aux=99, value=5)

Why it’s shaped this way

Registration puts the graph in the runtime registry with an explicit port contract, so a composed pipeline is a reusable unit, not a copy-pasted script. Because payloads stay @io-typed across the boundary, you can:

• replace or inject into a stage by name after construction,
• wrap a whole pipeline as a Flow and nest it, and
• render either level as runtime IR or an HTML graph for inspection.

The rendering path is on the simulation and visualization lane; the field-access rules are in flow contracts.

Source: examples/advanced/core_composition/composable_pipelines.py.