Stream - Collect

Type identifier: stream:collect Category: Stream Operations

Description

The Stream - Collect node consumes an asynchronous stream (such as LLM streaming responses) and processes each chunk as it arrives. It enables accumulator patterns where results build up incrementally, useful for streaming text aggregation, real-time processing, or tracking stream progress.

Input Handles

Output Handles

Configuration Options

This node has no additional configuration options.

Behaviour

1. Receives an async iterable stream from the input handle
2. Triggers the start control handle when processing begins
3. Initialises the accumulated value to null

4. For each chunk in the stream:

   • Sets the current chunk value
   • Triggers the process control handle
   • Reads the result input to update the accumulator
   • Updates accumulated with the new value

5. When the stream ends:

   • Sets data to the final accumulated value
   • Triggers the done control handle

Iteration Pattern

Each chunk creates a processing cycle:

1. chunk becomes available with the current stream value
2. process fires to allow handling
3. Connected nodes process and return a result
4. accumulated updates with that result

Stream Requirements

The input must be a stream. The only current source is LLM streaming responses.

Examples

Accumulate Streaming Text

Use case: Collect an LLM's streaming response into a complete string.

Flow:

1. AI - Chat Message (streaming mode) → stream

2. Stream - Collect:

   • On process: Concatenate chunk with accumulated
   • Return via result
3. data contains the complete response

Track Stream Progress

Use case: Count items in a stream and update progress.

Flow:

1. Stream source → stream

2. Stream - Collect:

   • start triggers progress initialisation
   • process increments counter and updates UI
   • done triggers completion notification

Transform Each Chunk

Use case: Process and store each chunk independently while tracking totals.

Flow:

1. Data stream → stream

2. Stream - Collect:

   • process → Database insert for each chunk
   • result → Running count/statistics
3. data contains final statistics