# Long-Stream Processing > Handle unbounded data in Sema — generators and Stream[T], bounded-memory pipelines, live token/speech streaming, and whole-document folds with compaction. 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: ```sema 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: ```sema 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. :::caution[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: ```sema 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`](/reference/examples-api/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: ```sema 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](/guides/packaging/)): ```toml [stream] window = 1000 budget = 4000 ``` ## Run and verify Streaming builtins and the `stream` module check like any project: ```bash sema check SEMA_STRICT=1 sema run ``` 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. :::caution[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)](/reference/examples-api/ai-console/) - [Packaging & Providers](/guides/packaging/) · [Multimodal](/guides/multimodal/) - [Building an Agent Loop](/guides/agent-loops/)