Streaming I/O

BLOGE supports streaming at both node level and graph boundary level. Use streaming when callers or downstream nodes should receive chunks before the whole computation has been materialized.

Streaming Members

The stream prefix can wrap nodes, foreach, and loop members:

graph tokenPipeline {
  stream node generate : LlmStreamingOperator {
    input {
      prompt = ctx.prompt
    }
    buffer = 32
  }

  node collect : TokenCollector {
    input {
      tokens = generate.stream
    }
  }
}

generate.stream passes the live NodeChannel to a downstream node and creates a stream edge in the graph model.

Stream Foreach

stream foreach forwards item work as chunks rather than waiting for the entire collection to finish.

stream foreach processOrders : order in loadOrders.output.orders {
  buffer = 16
  node processItem : OrderProcessor {
    input {
      order = order
    }
  }
}

For large collections, combine streaming with max_concurrency or batch_size to avoid unbounded fan-out.

stream foreach processOrders : order in loadOrders.output.orders {
  max_concurrency = 8
  batch_size = 50
  node processItem : OrderProcessor {
    input {
      order = order
    }
  }
}

Graph-Level Streaming Input

streaming input declares a stream supplied by the caller. The runtime reads a NodeChannel from GraphContext under the same name and exposes it as a virtual source node.

graph streamOrders {
  streaming input orders : Order

  stream node enrich : EnrichOrder {
    input {
      orders = orders.stream
    }
  }
}

The declared type describes each chunk, not a collected List.

Graph-Level Streaming Output

streaming output = <nodeId> exposes a stream node as the graph's live external output.

graph streamOrders {
  streaming input orders : Order

  stream node enrich : EnrichOrder {
    input {
      orders = orders.stream
    }
  }

  streaming output = enrich
}

Callers use the streaming execution API to consume chunks and can still await the final graph result.

Broadcast Behavior

The same streaming node can feed multiple downstream consumers. Runtime creates independent broadcast channels for each stream edge, so consumers receive the full chunk sequence instead of competing for messages from one queue.

Failure and Backpressure Guidance

• Set buffer deliberately; it is part of the runtime pressure boundary.
• Use timeout and retry only where re-emitting chunks is safe.
• Treat streaming fallbacks carefully; downstream consumers need to know whether data is complete, partial, or degraded.
• Prefer batch_size for very large stream foreach inputs.
• Keep streaming operators idempotent when retries are possible.

Next Steps

• See runtime scheduling in Execution Model.
• Learn recovery behavior in Crash Recovery & Checkpoints.
• Explore agent streaming in AI Agents & LLM Operators.