Gather nodes

A gather node runs a tool once per item of an upstream collection and gathers the per-run outputs into parallel lists. Use it when an upstream produces many items but the tool’s input is cardinality = "one", i.e. the tool can only consume one item per invocation.

rf   = g.add_node("rfdiffusion")                  # output `designs`: a dir of N backbones
mpnn = g.add_gather_node("proteinmpnn", split_key="designs")
g.add_edge((rf, mpnn))

split_key names the upstream output key whose value is the collection to fan out over.

Finding split_key

split_key is the name of the upstream tool’s output that holds the collection, read it from the producing tool’s manifest, not the consuming tool’s inputs:

bv show rfdiffusion --format json

RFdiffusion declares one output, designs (a directory of backbones), so split_key="designs". The rule of thumb: pick the upstream output that is a directory or list of the items you want to process one at a time. You also need a gather node (rather than a plain tool node) precisely when that upstream output is a collection but the downstream tool’s matching input is cardinality = "one".

        flowchart LR
    RF["rfdiffusion<br/>runs once → N backbones"] --> SC{{"split<br/>into N inputs"}}
    K(["knobs<br/>num_seq_per_target=2"]) -. into every run .-> RUNS
    SC --> RUNS["proteinmpnn ×N<br/>one backbone each"]
    RUNS --> GA{{"gather<br/>N outputs → lists"}}
    GA --> OUT(["sequences: [s1 … sN]"])
    

How it runs

For rfdiffusion → gather(proteinmpnn) with N backbones, the upstream runs once, its output is split into N items, the tool runs once per item, and the N results are gathered into lists, e.g. {"sequences": [s0, s1, …]}.

The split is over the items in the upstream output: the files in a directory, the elements of a list, or the records of a multi-record file. It is handled automatically, and the tool itself is unaware of the fan-out.

Shared parameters (a separate input node) are applied to every run, not split:

mpnn_in = g.add_input_node(num_seq_per_target="2", sampling_temp="0.1")
g.add_edge((rfdiffusion, proteinmpnn), (mpnn_in, proteinmpnn))

A gather node requires exactly one non-input upstream (the collection source); additional input nodes for scalars are allowed.

Chaining gather nodes

gather emits lists, so a gather node can feed another gather node:

proteinmpnn = g.add_gather_node("proteinmpnn", split_key="designs")
colabfold   = g.add_gather_node("colabfold",   split_key="sequences")
g.add_edge((rfdiffusion, proteinmpnn))
g.add_edge((proteinmpnn, colabfold))

The second stage receives the gathered list and processes it item by item. This two-level pattern (backbones → one sequence design each → one structure prediction each) is the basis of the self-consistency example (Example pipelines).

Reference

See the API reference for the full method and parameter reference.