Stream

View as MarkdownView as Markdown

A Stream delivers messages from a Flow to a reader in real time. A Step or a Client writes messages. A Client reads them in order and can resume where it stopped. Dex keeps Stream messages only on a best-effort basis.

When to use a Stream

Use a Stream for data that improves the user experience but is not business-critical:

• LLM reasoning summaries, generated-text chunks, and agent progress
• progress bars, diagnostic events, and live previews
• transient status shown while a Step is still running

Use a Channel or an Attribute for data that must not be lost.

Guarantees

A Stream gives up durability for speed:

• Dex keeps messages in the memory of the server. A server restart or a standby takeover loses them.
• Each Stream has a capacity in bytes. When it fills up, Dex removes the oldest messages. A slow reader can miss them.
• A Step write goes out at once, while the Step still runs. A later Step failure does not roll it back.
• A failed Stream write never fails the Step.
• A reader sees messages in the order Dex stored them.

The capacity is shared by all Flow runs with the same Flow type and Stream name. It is not a limit for each Flow ID. See Write semantics, Capacity and trimming, and Best-effort storage.

Define and write a Stream

A Stream has a stable name, a message type, and a capacity. Add it to the Flow persistence schema.

This example writes progress text through a buffered text writer. Dex sends the buffered text before it saves the Step result. The Step does not wait for the server to store the text. See Buffer text chunks.

stream_flow.pyPython

12345678910111213141516171819202122

class RenderPreview(Step[str]):
    def __init__(self, progress: Stream[str]) -> None:
        self.progress = progress

    async def execute(self, context: AsyncContext, input: str) -> StepDecision:
        progress = self.progress.buffered_text(context)
        progress.write(f"Rendering preview for {input}")
        progress.write(f"Preview ready for {input}")
        return graceful_complete(f"Rendered {input}")


class StreamFlow(Flow[str]):
    progress = Stream("Progress", str, 10 * 1024 * 1024)

    def __init__(self) -> None:
        self.render_preview = RenderPreview(self.progress)

    def get_steps(self) -> StepList[str]:
        return StepList.start_step(self.render_preview)

    def get_persistence_schema(self) -> PersistenceSchema:
        return PersistenceSchema.of(self.progress)

Example: examples/python/dex_examples/primitives/stream/stream_flow.py

Read a Stream

A Client has two ways to read a Stream:

• ReadStream reads forward, one message at a time. Each message carries a resume token. Pass the token back to read the next message. When no new message exists, the call waits for one.
• ListStreamMessages pages backward from the newest message. It returns at once.

See Resumable reads and List messages.

A Client can also write to a Stream, even when the Flow has closed. An RPC handler writes through an injected Client.

Reference

The Stream page of each SDK reference lists the API names in that language.

Data primitive semantics covers write semantics, buffered text, reads, listing, capacity, and storage in detail. The server settings are in Streams.