Long-Stream Processing

Some data must never be resident all at once: a ten-hour audio file, a 200 MB document, a token stream, a dataset larger than memory. Sema answers this with Stream[T] — a first-class, pull-driven, bounded-memory type — and with a stream stdlib module that folds a whole document larger than any context window into a bounded digest. This guide covers both: the streaming type and the long-context pattern.

The unit doctrine

A Sema stream has no byte-level unit. The element type T is the unit of meaning — AudioFrame, Row, Utterance, Bytes — and declaring a stream is choosing that unit. Transport framing (packet coalescing, chunking oversized elements) belongs to the runtime and is invisible to programs. Re-unitizing for consumption — 30-second audio windows, sliding text windows — happens at the consumer via windowing adapters, never by a producer guessing.

Generators: stream def

stream def declares a lazy, pull-driven generator whose return type is Stream[U]; yield suspends until the consumer pulls, and a bare return ends the stream:

stream def rows(path: Path) -> Stream[Row] !{fs.read}:   # generator: lazy, pull-driven
    with open_csv(path) as f:
        while f.has_next():
            yield f.next_row()                           # suspends until the consumer pulls

def totals(path: Path) -> Money !{fs.read}:
    mut sum = Money.zero
    for batch in rows(path).batch(4096):                 # 100 GB file, O(batch) resident
        sum = sum + batch_total(batch)
    return sum

The bounded-memory law: resident memory per stage is O(queue + window), independent of stream length — nothing materializes unless you write collect.

Note

stream def is not async — suspension is a pull-driven frame in the runtime, not a coroutine that colors every caller. Effects execute at pull time in the consumer’s dynamic extent, charged to the puller’s budget.

Pipelines: the adapter chain

Prelude adapters are lazy, fused where provable, and O(window) in memory. Chaining them is the transform surface — open a stream, pipe it through, and what falls out the end is already clean:

clean = follow(feed)
    .filter(a => a.lang == "en", label="english")
    .distinct(within=10_000, label="dedupe")      # sliding dedupe; bound in the signature
    .map(normalize)
    .window(size=256, stride=256)

The vocabulary includes map / filter / flat_map / scan / take / take_while / distinct(within=); batch(n) -> Stream[list[T]]; window(size=, stride=, by=) where by is an optional measure (by=f => f.duration for time windows); lift() for the error channel; buffer(n) to override the queue bound; and collect() -> list[T] as the one explicit materialization point.

Grouping and sorting are bounded-scope operations

group_by and sorting exist on list — and therefore inside a window or batch (w.items.group_by(key)) — not on a raw stream. The fix-it is always “window first”. distinct(within=) is the sliding exception, its bound in the signature.

Fallibility, without hiding it

Stream[T] cannot fail mid-flow — by construction, not convention. A producer that can fail types its elements Result[T, E], so “can this pipe break?” is answered by the type. The terminator law: a producer that cannot continue yields exactly one terminal Err, then ends — a stream never just stops silently. Consumers stop by leaving the for, satisfying .take(n), or scope exit; cancellation propagates upstream.

Stream values are affine scoped resources: consumed at most once, never duplicated, and released — with their producers cancelled — at scope exit. A live stream can’t be stored in a struct, emitted in an event, or serialized (only service signatures carry one, because there the runtime manages the wire form).

Live model streaming

Two ready-to-use streaming builtins let a program show partial output instead of blocking. generate_stream streams an LLM completion token-by-token — each decoded piece is emitted live and the chunks are returned as a list:

streamed = generate_stream("Summarize the logistics plan", 24)
log.info("streamed", chunks=len(streamed))

transcribe_stream(audio) transcribes a file in 30-second windows, emitting each window’s transcript live and returning the segment list — so a long recording transcribes progressively. generate(prompt, max_tokens) is the non-streaming form. The real GGUF backend streams true model tokens on-device; with no model configured streaming fails typed (ModelUnavailable); under the opt-in deterministic engine a word-by-word deterministic stream keeps tests hermetic. The ai-console example exercises generate_stream alongside batched generation.

Note

The tree-walking interpreter is single-threaded, so these builtins are eager (the chunk list is materialized) — “streaming” means live incremental emission via a callback during generation, the observable win without coloring callers.

Long-document folds with compaction

The classic LLM problem — a document larger than the model’s context window — is a language primitive. import stream gives you a streaming fold with automatic compaction, so processing an entire book with an 8k-context model is one call at constant memory:

import stream

# Fold a whole on-disk book into a bounded digest, streaming from disk.
digest = stream.fold_file("book.txt", window=1000, budget=4000)

# Or over an in-memory string:
digest = stream.fold(text, window=1000, budget=4000)

# Semantic full-text search over a document too big to embed at once:
hits = stream.search(book, "apple gpu inference", window=500, k=5)

# Process each window and collect results:
names = stream.map(book, lambda w: semantic.extract(w, "person names"), window=800)

fold walks the text in window-token pieces, keeps a running digest, and the moment the digest exceeds budget it compacts it — so memory is O(window + budget), independent of input length. fold_file streams from disk, holding only a line plus the current window plus the digest. Measured: an 11 MB / 2.8-million-token book folds to a 127-token digest at 3.4 MB peak RSS.

The compaction is driven by the configured engine — a real model summarizes semantically; the opt-in deterministic engine gives a reproducible proxy so tests are stable. It is model-agnostic: the same code works whether the engine is a 360M local model or a frontier API, and window / budget adapt it to any context size. Set the defaults once in sema.toml (see Packaging & Providers):

[stream]
window = 1000
budget = 4000

Run and verify

Streaming builtins and the stream module check like any project:

sema check <your-project>
SEMA_STRICT=1 sema run <your-project>

The stream primitives are crash-proof by construction (char-safe truncation, saturating arithmetic, clamped windows). SEMA_STRICT=1 turns any degradation — a truncation, a skipped window — into a hard, typed error while you verify; leave it off in production so the system self-heals.

Materializing defeats the point

sema doctor flags collect on a wire or generator stream with no upstream bound — materializing an unbounded source is exactly what streams exist to avoid. Bound it with a window, take, or a lane budget first.

Variations

• Sliding window into a generative function. The sanctioned long-context pattern is text.window(size=4096, stride=3584, by=tok) feeding a simulate function per window.
• Progressive speech-to-speech. transcribe_stream → generate_stream → per-sentence TTS makes a spoken exchange responsive.
• Search before you fold. stream.search finds the relevant windows of a huge document so you only fold the parts that matter.

See also

• ai-console example (generated)
• Packaging & Providers · Multimodal
• Building an Agent Loop