Stream results with pipeline nodes

Most nodes answer once: one input message in, one output message out. Some answers are not one value but a series — the chunks of a document, the tokens of a model's reply, the records in a large file, the pages of an API. Buffering a series into one response makes the caller wait for the slowest part of it and puts the whole thing in memory at once.

A pipeline node streams instead. It reads a stream of input frames and emits as many output frames as it likes, and every frame is delivered to the next node — or to your terminal — the moment it is produced.

Prerequisites: a package you can build and push — see Create a node for the authoring loop in your language.

Declare the node#

One knob in axiom.yaml:

nodes:
  - name: Chunk
    type: pipeline          # "" or "unary" is the default
    description: Splits text into word tokens, one frame per word.
    input: TextRequest
    output: TokensResult

Scaffold it with the flag and you get a streaming handler and a matching streaming unit test:

axiom create node Chunk --type pipeline --input TextRequest --output TokensResult

type: pipeline composes with everything else a node can declare — required_secrets, mutation_capable, kind: generic.

Write the handler#

The handler signature is fixed per language and axiom validate enforces it. Go is the odd one out: it takes a receive channel plus an emit callback rather than returning an iterator.

// Go — nodes/chunk.go
func Chunk(ctx context.Context, ax axiom.Context, in <-chan *gen.TextRequest, emit func(*gen.TokensResult) error) error {
    for input := range in {
        for _, w := range strings.Fields(input.GetText()) {
            if err := emit(&gen.TokensResult{Tokens: []string{w}, Count: 1}); err != nil {
                return err
            }
        }
    }
    return nil
}

# Python — yield, do not return a list
def chunk(ax: AxiomContext, inputs: Iterator[TextRequest]) -> Iterator[TokensResult]:
    for inp in inputs:
        for word in inp.text.split():
            yield TokensResult(tokens=[word], count=1)

// TypeScript
export async function* chunk(
  ax: AxiomContext,
  inputs: AsyncIterable<TextRequest>,
): AsyncGenerator<TokensResult> {
  for await (const inp of inputs) {
    for (const word of inp.text.split(/\s+/)) {
      yield new TokensResult({ tokens: [word], count: 1 });
    }
  }
}

// Rust
pub fn chunk<I>(ax: &dyn AxiomContext, inputs: I)
    -> impl Iterator<Item = Result<TokensResult, Box<dyn std::error::Error>>>
where I: Iterator<Item = TextRequest> { /* … */ }

// Java
public static Stream<TokensResult> chunk(AxiomContext ax, Iterator<TextRequest> inputs) { /* … */ }

// C#
public static async IAsyncEnumerable<TokensResult> Chunk(
    IAxiomContext ax,
    IAsyncEnumerable<TextRequest> inputs,
    [EnumeratorCancellation] CancellationToken ct = default) { /* … */ }

How many items the input stream yields depends on where the node sits:

• Start node of a pipeline flow: exactly one item — the flow's input.
• Any node further down: as many frames as its upstream emitted.
• Direct invoke: one item for an ordinary invoke, or one item per line when the caller streams NDJSON into it — see Stream input into a node below.

So a start node must not assume a single item: the same node reached by direct invoke can be handed a live, unbounded stream. Write the loop, not the inputs[0].

Give the output message its own is_final field#

When your function returns, the platform appends a transport-level terminal frame with an empty payload. A consumer therefore cannot read the transport flag as a business signal: by the time it sees "final", the payload is gone.

If "this is the last one" matters to whoever consumes your stream, put the marker on your own message and set it yourself:

message TokensResult {
  repeated string tokens = 1;
  int32 count = 2;
  // True on the LAST frame of the stream and no other.
  bool is_final = 3;
}

Make every frame idempotent#

Delivery is at-least-once, per frame, exactly as it is for a unary call. A pipeline flow is durable but not resumable: its checkpoint is written once per node at stream completion, so a redelivery re-runs the whole graph and replays every frame. Upsert on a stable key derived from the frame — never "append one row per frame".

Test it before you push#

Seed the input, collect every emitted frame, and assert the exact count and order — plus is_final on only the last one. This is what axiom create node --type pipeline scaffolds:

func TestChunk(t *testing.T) {
    in := make(chan *gen.TextRequest, 1)
    in <- &gen.TextRequest{Text: "hello world"}
    close(in) // an unclosed channel leaves the node's range loop blocked forever

    var frames []*gen.TokensResult
    err := nodes.Chunk(context.Background(), ax, in, func(f *gen.TokensResult) error {
        frames = append(frames, f)
        return nil
    })
    if err != nil {
        t.Fatalf("unexpected error: %v", err)
    }
    if len(frames) != 2 {
        t.Fatalf("expected 2 frames, got %d", len(frames))
    }
    if frames[0].GetTokens()[0] != "hello" || frames[1].GetTokens()[0] != "world" {
        t.Errorf("frames out of order: %v", frames)
    }
}

Then run the real generated service and watch the frames arrive:

axiom dev
curl -N localhost:8083/nodes/Chunk -d '{"text":"hello world"}'
# data: {"tokens":["hello"],"count":1}
# data: {"tokens":["world"],"count":1}

-N is load-bearing — without it curl buffers the response and you learn nothing about when each frame appeared.

Invoke a published pipeline node#

axiom invoke prints one JSON frame per line, as each arrives:

axiom invoke me/text-ops/Chunk --input '{"text":"hello world"}'
#   ···  streaming — pipeline node, one JSON frame per line
# {"frame_index":0,"payload":{"tokens":["hello"],"count":1},"is_final":false,"success":true}
# {"frame_index":1,"payload":{"tokens":["world"],"count":1},"is_final":false,"success":true}
# {"frame_index":2,"is_final":true,"success":true}

Frames go to stdout and diagnostics to stderr, so the stream pipes straight into jq. Over raw HTTP the same invoke answers with Content-Type: text/event-stream — see HTTP API.

Stream input into a node#

Output streaming is only half of it. The same invoke accepts a streamed request body: NDJSON — one JSON object per LINE — where each line becomes its own input frame, dispatched as it is read.

axiom invoke me/text-ops/Chunk --input-stream frames.ndjson
#   ···  streaming input — NDJSON, one frame per line
# {"frame_index":0,"payload":{"tokens":["hello"],"count":1},"is_final":false,"success":true}
# {"frame_index":1,"payload":{"tokens":["world"],"count":1},"is_final":false,"success":true}
# {"frame_index":2,"is_final":true,"success":true,"input_frames":2}
#   ···  2 input frame(s) sent and accepted

- reads stdin, so the input can be produced live and never end:

tail -f events.ndjson | axiom invoke me/tools/StreamRecords --input-stream -

That last line — sent and accepted — is the CLI reconciling the frames it wrote against the count the platform reports back. If they ever disagree the invoke fails loudly rather than printing a clean run over a lossy one.

Under the hood: three requests, not one#

--input-stream drives three separate HTTP requests: it creates a stream, attaches the output channel, then uploads frames. Each is half-duplex — bytes flow one way per request.

That shape is not incidental. NDJSON up and Server-Sent Events down on ONE request is full-duplex HTTP, and an HTTP client — including the reverse proxy in front of the API — stops writing a request body once the response arrives. The one-request form therefore truncated the caller's upload the moment the node answered: 7 frames paced 3s apart delivered 4, with the run still reporting success. Splitting the legs removes the interaction entirely.

Two consequences worth knowing:

• The output channel replays from the beginning, so attaching late — or reconnecting after a drop — still gets you every frame from the first.
• Posting NDJSON straight to the node endpoint still works, but its response is buffered: you get every frame, only after your upload has finished. Use the three-leg shape (or just the CLI) when you want output as it happens.

Driving it by hand sets Content-Type: application/x-ndjson and streams the body with curl -N -X POST -T - — not --data-binary @-, which reads all of stdin into memory before sending and therefore cannot stream at all. See HTTP API for the three legs in full.

Pipeline nodes only. A unary node consumes exactly one input, so later frames would be silently discarded; streaming into one is rejected with a clear error instead.

Limits. 16 MiB per frame, 64 MiB and 100 000 frames per stream, and the same 5-minute node ceiling as any other invoke. A frame that fails validation is rejected at that frame — the stream ends with an error naming the frame index, and output already emitted for earlier frames stands. A streamed direct invoke is synchronous and is not redelivered: if the connection drops, re-stream.

Live mode: full duplex, and nothing written down#

--input-stream is durable bulk ingestion. When what you want is an interactive session — a token relay, a live tail, a dialogue where you send the next thing based on what came back — use --live:

axiom invoke me/tools/Chat --live -

That opens one WebSocket. Frames go up and come down on the same connection at the same time, so you can send frame k+1 in response to frame k's output. Under the hood it is the same node stream, the same frame JSON, the same per-frame validation and reshape — only the client-edge transport differs.

The trade is durability, and it is total:

A live session is ephemeral. No durable execution row, no replay, no redelivery, and it is pinned to a single ingress pod — if that pod restarts, the session dies. axiom executions get will never find it. Recovery means reconnecting and re-sending.

Use --input-stream when you need a record of what was ingested. Use --live when the value is in the round trip and re-sending is cheap.

A slow reader loses its session. The platform buffers only a small bounded amount of output per live session; a client that stops draining is closed with a reason rather than buffered indefinitely. Live sessions also cap output at 64 MiB and hold one inflight execution slot for their lifetime. The per-frame 16 MiB cap and the 5-minute ceiling are the same as everywhere else.

--live needs a write-scoped token, like any other invoke — running a node is a write, even though the WebSocket handshake itself is a GET.

Incremental consumption is per-language#

Every language receives every frame. What differs is when — and it is a property of the generated node service, not of your handler:

If your node has to react to frame k before frame k+1 is sent — a token relay, a live tail, an interactive session — write it in Go, Python or TypeScript. For a bounded batch that simply arrives as N frames, any language works.

Compose pipeline nodes into a streaming flow#

A flow that should stream sets one top-level field:

# doc-chunk-stream.flow.yaml
name: me/doc-chunk-stream
version: 0.1.0
pipeline_mode: true
nodes:
  - alias: chunk
    package: me/text-ops@0.1.0
    node: Chunk
  - alias: prep
    package: me/embed-prep@0.1.0
    node: Prepare
edges:
  - from: chunk
    to: prep

Compile it as usual, then run it with --stream:

axiom flow compile doc-chunk-stream.flow.yaml
axiom flow run 01J… --stream -d '{"text":"hello world"}'
# {"tokens":["hello"],"count":1}
# {"tokens":["world"],"count":1}
# ✓ succeeded  (execution 4bf9…, 2 frame(s))

--stream --json prints the full frame envelope instead — including frame_index and is_final, which is what lets you assert the sequence.

Without --stream, a pipeline flow is sent to the unary endpoint, which returns a single aggregated result and discards every later frame. The CLI detects the mismatch and warns loudly; that warning means the output you are looking at is not the flow's output. Re-run with --stream.

The raw HTTP form is POST /v1/flows/invoke/stream:

curl -N -X POST 'https://api.axiomide.com/invocations/v1/flows/invoke/stream' \
  -H "Authorization: Bearer $AXIOM_API_KEY" \
  -H 'Content-Type: application/json' \
  -d '{"graph_id":"01J…","input":{"text":"hello world"}}'

Mixed graphs are allowed — with one rule#

pipeline_mode: true does not require every node in the flow to be type: pipeline. The pipeline executor drives a unary node over the same stream using single-frame semantics, so mixing compiles and runs.

The rule that decides your design: a unary node receives the first frame and every later frame is discarded. A unary node is therefore fine as the start of a pipeline flow, or anywhere upstream of the first real stream — and is silent data loss anywhere downstream of one.

Instances of a pipeline generic#

A generic node can be type: pipeline, and an Instance you bind to it is an ordinary pipeline node: it streams. The facade reshape runs once per frame — your input facade on the way in, your output facade on every frame that comes back — so each frame arrives at the caller in the shape the Instance declares, not the generic's:

axiom instance create me/docs/Tokens \
  --generic-package me/text-ops --generic-node Split \
  --input-message Doc   --input-field text=string \
  --input-map body=text \
  --output-message Token --output-field position=int64 --output-field value=string \
  --output-map position=index --output-map value=body

axiom invoke me/docs/Tokens --input '{"text":"hello world"}'
# {"frame_index":0,"payload":{"position":"1","value":"seen: hello"},"is_final":false,"success":true}
# {"frame_index":1,"payload":{"position":"2","value":"seen: world"},"is_final":false,"success":true}
# {"frame_index":2,"is_final":true,"success":true}

The same Instance node drops into a pipeline_mode flow like any other pipeline node, and streams there too.

One consequence of the empty terminal frame above: the reshape is applied to data frames only. The platform's terminal frame carries no payload, so there is nothing there for your declared facade fields to honor — which is another reason to put your own is_final on your own message.

What a pipeline flow cannot do#

These are rejected at compile time, with a diagnostic:

Absent at run time: human-in-the-loop pauses, node retry, mutation, gates and quorum joins. Fan-out and fan-in both work, and fan-in is order-preserving — frames are merged positionally by their source frame index. Conditions and edge adapters are evaluated once per frame, so a condition can drop individual frames out of a stream.

Limits: a single frame must be ≤24 MiB, and a node has a 5-minute ceiling for its whole stream.

Prove it actually streamed#

The execution timeline is the definitive evidence — a flow that silently degraded to one aggregated result looks completely different there:

axiom executions get <execution-id>          # the event sequence
axiom executions get <execution-id> --json   # …with each event's full payload

Look for per-frame NODE_FRAME and NODE_INPUT_FRAME events, per-frame per-edge EDGE_EVALUATED, and NODE_STREAM_COMPLETE — which replaces NODE_COMPLETED for a pipeline node. The human view lists the events; the per-event counters (total_frames, total_bytes, frames_dropped) are in the payload, so use --json to read them.

frames_dropped is the one to watch. A frame whose outgoing edge's condition or adapter throws is dropped: the stream gets shorter and the flow still succeeds. Each drop is also recorded as an EDGE_EVALUATED event carrying the frame_index and an error naming the failure (and deliberately no condition_pass — the condition never returned a verdict), so you can see exactly which frame died on which edge. Assert exact frame counts in your tests and a non-zero frames_dropped will never surprise you in production.

One wrinkle worth knowing: a unary node inside a pipeline flow reports NODE_STREAM_COMPLETE before it forwards its single result, so its frames_dropped is always 0. For those nodes the per-drop EDGE_EVALUATED event is the record.

See Debug a flow for the rest of the timeline surfaces.