Skip to main content
Version: Next

mellea.stdlib.streaming

Streaming generation: a single-task async for primitive.

stream() starts a streaming generation and returns a Streamer you consume with async for. It drives token draining, chunking, and (when requirements are given) per-chunk and final validation.

Consume inside async with so cleanup always runs: the stream runs on the caller's task, and leaving the block — normally, on early break, or on exception — cancels the generation and fires the STREAMING_END hook.

Typed StreamEvent objects are emitted via the STREAMING_EVENT hook; subscribe a plugin to observe them (see docs/examples/streaming/).

Functions​

FUNC stream ​

stream(action: Component[Any] | CBlock, backend: Backend, ctx: Context) -> Streamer

Start a streaming generation.

Generation begins eagerly, before this call returns. Consume the returned Streamer inside async with so the stream is always released. On early exit or break, async with cancels the in-flight generation:

async with await stream(action, backend, ctx) as s:
    async for chunk in s:
        ...

Each iteration yields a chunk — a unit produced by the chunking strategy, or the raw model delta when chunking is None. A chunk is delivered once it has passed every requirement's stream_validate; a "fail" stops the stream early and cancels the backend. On natural completion, validate() runs on the full output. With no requirements, chunks are yielded without validation.

Args:

  • action: The component or content block to generate from.
  • backend: Backend used for generation and, unless validation_backend is set, validation.
  • ctx: The generation context.
  • chunking: A ChunkingStrategy, a recognized alias string, or None (default) to yield raw deltas unchunked.
  • requirements: Requirements validated against each chunk during streaming and against the full output at stream end. None yields chunks without validation.
  • validation_backend: Backend for validation calls; defaults to backend.

Returns:

  • An async-iterable handle over the validated chunks.

Raises:

  • ValueError: If chunking is a string that is not a known alias.
  • RuntimeError: If the backend returns an already-computed thunk instead of a streaming one — i.e. it is not honouring ModelOption.STREAM.

Classes​

CLASS StreamEvent ​

Base class for all streaming events emitted by stream.

The timestamp field is auto-populated at instantiation time; callers do not set it. Because timestamp has init=False it is never part of __init__, so subclasses may declare additional fields in any order without conflict. Any new init=False fields on subclasses must also use field(..., init=False).

Attributes:

  • timestamp: Unix timestamp (seconds) at the moment the event was created.

CLASS ChunkEvent ​

Emitted after each validated chunk is delivered to the consumer.

Fired after all active requirements' stream_validate calls return non-"fail" for this chunk and the chunk has been yielded to the consumer.

Args:

  • text: The chunk text that was validated and emitted.
  • chunk_index: Zero-based position of this chunk in the stream.
  • attempt: Sampling attempt number; currently always 1.

CLASS QuickCheckEvent ​

Emitted after each per-chunk streaming validation batch.

One event per chunk, covering all active requirements in parallel. Not emitted when there are no requirements.

Args:

  • chunk_index: Zero-based position of the chunk that was validated.
  • attempt: Sampling attempt number; currently always 1.
  • passed: True if all active requirements returned non-"fail" for this chunk.
  • results: PartialValidationResult from each active requirement, in the same order as the active slice of requirements.

CLASS StreamingDoneEvent ​

Emitted after all chunks have been validated and delivered to the consumer.

Fired after the regular token stream and any trailing fragment released by the chunker's flush() have both been processed. Only emitted on natural completion — not on early exit (a requirement returned "fail") or on exception.

Args:

  • attempt: Sampling attempt number; currently always 1.
  • full_text: Complete accumulated text at stream end.

CLASS FullValidationEvent ​

Emitted after the final Requirement.validate calls complete.

Only emitted when the stream completed naturally (no requirement failed during streaming). Not emitted on early exit.

Args:

  • attempt: Sampling attempt number; currently always 1.
  • passed: True if all final ValidationResult objects passed.
  • results: ValidationResult from each requirement, in requirement order.

CLASS RetryEvent ​

Reserved for future use.

Defined for API completeness — RetryEvent is not currently emitted; today retry is caller-driven re-invocation of stream. If retry is added to streaming itself, this event will fire before each re-attempt.

Args:

  • attempt: Attempt number being started (1-based).
  • reason: Human-readable reason for the retry.

CLASS CompletedEvent ​

Emitted when the stream exits, including early-exit cases.

Always the last StreamEvent on every exit path. success reflects whether the stream completed with no "fail" result and no exception.

Args:

  • success: True if the stream completed normally (no "fail" result and no unhandled exception); False otherwise.
  • full_text: Validated-and-emitted output. On early exit or exception, reflects whatever passed validation before the stop.
  • attempts_used: Number of stream attempts; currently always 1.

CLASS ErrorEvent ​

Emitted when an unhandled exception occurs while streaming.

Args:

  • exception_type: Python class name of the exception (e.g. "ValueError").
  • detail: String representation of the exception.

CLASS Streamer ​

Async-iterable handle for a stream call.

Iterate the returned Streamer object with async for to receive the output as validated chunks, ideally inside async with so the stream is released on every exit. Each chunk is a str segment of the model output text, sized by the chunking strategy (or a raw model delta when chunking is None). The attributes below track progress and outcome. Instances are created by stream; do not instantiate directly.

Args:

  • mot: The in-flight streaming thunk from the backend generation call.
  • ctx: The generation context, used for validation calls.
  • chunking: Resolved chunking strategy, or None for raw deltas.
  • requirements: Requirements to validate against; pre-copied by stream.
  • validation_backend: Backend used for validation calls.

Attributes:

  • failed_early: True if a requirement returned "fail" during streaming and the stream stopped before natural completion.
  • completed_normally: True only if the stream reached its natural end, prior to final validation. False on requirement failure, an early break, or an exception — unlike not failed_early, which stays True after an early break.
  • failure_reason: Human-readable reason when failed_early is True.
  • streaming_failures: (Requirement, PartialValidationResult) pairs for every requirement that failed the offending chunk.
  • full_text: Validated-and-emitted output. On natural completion, the full accumulated text; on early exit, the accumulated text through the last emitted chunk.
  • mot: The computed thunk, set on natural completion; None otherwise.
  • final_validations: ValidationResult objects from the stream-end validate() calls; empty on early exit.
  • streaming_id: UUID correlating this stream's START/EVENT/END hooks.

Methods:

FUNC aclose ​

aclose(self) -> None

Release the stream, cancelling generation if it is still in flight.

Runs the driver's cleanup (cancelling the backend generation and firing STREAMING_END). Safe and idempotent on every path: after natural completion, after an early exit/break, and on a Streamer that was never iterated — the eager generation is still cancelled in every case.

Prefer consuming with async with stream(...) as s: so this runs automatically on every exit path; call aclose() directly only when not using the context manager.