Tutorial: Streaming Validation

Post-generation validation waits until the model has finished writing before checking the output. That is fine for short responses, but wastes time and compute when a violation appears in the first sentence of a ten-paragraph reply. Streaming validation moves the check into the generation loop: each chunk is validated as soon as it arrives, and generation is cancelled the moment a requirement fails.

By the end you will have covered:

• stream() — the streaming validation entry point
• Consuming validated chunks with async for inside async with
• Early-exit cancellation and reading streaming_failures
• Choosing between "word", "sentence", and "paragraph" chunking
• Requirement chunking= — validators at different granularities on one stream
• Observing the typed event vocabulary (ChunkEvent, QuickCheckEvent, …) by iterating stream(as_events=True)
• Subclassing ChunkingStrategy to define a custom split boundary

Prerequisites: Tutorial 02 (async and streaming), Tutorial 04 (requirements and validation), pip install mellea, Ollama running locally with granite4.1:3b downloaded.

---

Step 1: Your first streaming validation call

stream() starts a streaming generation and returns a Streamer. Consume it with async for inside async with: each iteration yields the next chunk, already validated against every requirement, and the async with block cancels the generation on every exit path — including an early break or an exception — so an abandoned stream never keeps running in the background.

# Requires: mellea
# Returns: None
import asyncio
import re

from mellea.core.backend import Backend
from mellea.core.base import Context
from mellea.core.requirement import PartialValidationResult, Requirement, ValidationResult
from mellea.stdlib.components import Instruction
from mellea.stdlib.streaming import stream

_SENTENCE_END = re.compile(r"[.!?]+")


class MaxSentencesReq(Requirement):
    """Fails the stream if the model writes more sentences than *limit*."""

    def __init__(self, limit: int) -> None:
        super().__init__()
        self._limit = limit
        self._count = 0

    def format_for_llm(self) -> str:
        return f"The response must be at most {self._limit} sentences."

    async def _stream_validate(
        self, chunk: str, *, backend: Backend, ctx: Context
    ) -> PartialValidationResult:
        self._count += len(_SENTENCE_END.findall(chunk))
        if self._count > self._limit:
            return PartialValidationResult(
                "fail", reason=f"Exceeded {self._limit}-sentence limit"
            )
        return PartialValidationResult("unknown")

    async def validate(
        self, backend: Backend, ctx: Context, *, format=None, model_options=None
    ) -> ValidationResult:
        return ValidationResult(result=self._count <= self._limit)


async def main() -> None:
    from mellea.stdlib.session import start_session

    m = start_session()

    async with await stream(
        Instruction("Write a two-sentence summary of how photosynthesis works."),
        m.backend,
        m.ctx,
        requirements=[MaxSentencesReq(limit=3)],
        chunking="sentence",
    ) as streamer:
        async for chunk in streamer:
            print(f"  chunk: {chunk!r}")

    print(f"\nCompleted normally: {streamer.completed_normally}")
    print(f"Full text: {streamer.full_text!r}")


asyncio.run(main())

  chunk: 'Photosynthesis is the process by which plants use sunlight, water, and carbon dioxide to produce glucose and oxygen.'
  chunk: 'This reaction takes place in the chloroplasts and is essential to nearly all life on Earth.'

Completed normally: True
Full text: 'Photosynthesis is the process by which plants use sunlight, water, and carbon dioxide to produce glucose and oxygen. This reaction takes place in the chloroplasts and is essential to nearly all life on Earth.'

Note: LLM output is non-deterministic. Your result will vary in wording.

Two things to notice:

• stream() is awaited to obtain the Streamer, but generation begins eagerly — it is already running by the time you enter the async for loop.
• Terminal state is read from the Streamer after the loop: failed_early, full_text, streaming_failures, and final_validations.

---

Step 2: Early exit on failure

When stream_validate() returns "fail", the backend generation is cancelled immediately and the loop ends. No further chunks are delivered, failed_early becomes True, and the failure is recorded in streamer.streaming_failures.

Lower the sentence limit so the model is likely to exceed it:

# Requires: mellea
# Returns: None
import asyncio
import re

from mellea.core.backend import Backend
from mellea.core.base import Context
from mellea.core.requirement import PartialValidationResult, Requirement, ValidationResult
from mellea.stdlib.components import Instruction
from mellea.stdlib.streaming import stream

_SENTENCE_END = re.compile(r"[.!?]+")


class MaxSentencesReq(Requirement):
    def __init__(self, limit: int) -> None:
        super().__init__()
        self._limit = limit
        self._count = 0

    def format_for_llm(self) -> str:
        return f"The response must be at most {self._limit} sentences."

    async def _stream_validate(
        self, chunk: str, *, backend: Backend, ctx: Context
    ) -> PartialValidationResult:
        self._count += len(_SENTENCE_END.findall(chunk))
        if self._count > self._limit:
            return PartialValidationResult(
                "fail", reason=f"Exceeded {self._limit}-sentence limit"
            )
        return PartialValidationResult("unknown")

    async def validate(
        self, backend: Backend, ctx: Context, *, format=None, model_options=None
    ) -> ValidationResult:
        return ValidationResult(result=self._count <= self._limit)


async def main() -> None:
    from mellea.stdlib.session import start_session

    m = start_session()

    # Ask for five sentences but cap the requirement at two.
    # The stream should be cancelled after the third sentence arrives.
    async with await stream(
        Instruction("Write five sentences about the history of the internet."),
        m.backend,
        m.ctx,
        requirements=[MaxSentencesReq(limit=2)],
        chunking="sentence",
    ) as streamer:
        async for chunk in streamer:
            print(f"  chunk: {chunk[:60]!r}...")

    if streamer.streaming_failures:
        _req, pvr = streamer.streaming_failures[0]
        print(f"\nStreaming failure: {pvr.reason}")
        print(f"Text at cancellation:\n{streamer.full_text!r}")
    else:
        print(f"\nFull text: {streamer.full_text!r}")


asyncio.run(main())

  chunk: 'The internet began as ARPANET, a U.S. Defense Department pr'...
  chunk: 'In the 1980s, the network expanded beyond government use an'...

Streaming failure: Exceeded 2-sentence limit
Text at cancellation:
'The internet began as ARPANET, a U.S. Defense Department project in the late 1960s. In the 1980s, the network expanded beyond government use and began connecting universities and research centres.'

Note: Whether the stream is cancelled depends on whether the model exceeds the limit. If the model happens to comply, streaming_failures will be empty and failed_early will be False.

streamer.full_text always contains the text accumulated up to the point where generation stopped — useful for debugging what the model produced before the requirement failed.

---

Step 3: Choosing a chunking strategy

The built-in strategies cover a coarse-to-fine spectrum:

Alias	Splits on	Good for
"word"	Whitespace	Token-local checks: forbidden words, numeric limits
"sentence"	., !, ? followed by whitespace	Grammar, coherence, per-sentence content rules
"paragraph"	Two or more consecutive newlines	Topic coherence, citation presence, heading structure

The trade-off is latency vs context. Word chunking fires after every word — maximum reaction speed, but each chunk carries only a single word. Paragraph chunking waits for blank lines — full paragraph context for the validator, but detection is later and may happen after the model has produced a large amount of invalid content.

To see the granularity difference concretely, switch to word chunking and count how many chunks arrive compared to Step 1's two sentences:

# Requires: mellea
# Returns: None
import asyncio

from mellea.core.backend import Backend
from mellea.core.base import Context
from mellea.core.requirement import PartialValidationResult, Requirement, ValidationResult
from mellea.stdlib.components import Instruction
from mellea.stdlib.streaming import stream

_FORBIDDEN = {"deprecated", "legacy", "obsolete"}


class ForbiddenWordReq(Requirement):
    """Cancels the stream the moment any forbidden word appears."""

    def format_for_llm(self) -> str:
        return f"Do not use any of the following words: {', '.join(sorted(_FORBIDDEN))}."

    async def _stream_validate(
        self, chunk: str, *, backend: Backend, ctx: Context
    ) -> PartialValidationResult:
        word = chunk.strip().lower().strip(".,!?;:\"'")
        if word in _FORBIDDEN:
            return PartialValidationResult("fail", reason=f"Forbidden word: {chunk.strip()!r}")
        return PartialValidationResult("unknown")

    async def validate(
        self, backend: Backend, ctx: Context, *, format=None, model_options=None
    ) -> ValidationResult:
        return ValidationResult(result=True)


async def main() -> None:
    from mellea.stdlib.session import start_session

    m = start_session()

    word_count = 0
    async with await stream(
        Instruction(
            "Describe three advantages of cloud-native development in two sentences."
        ),
        m.backend,
        m.ctx,
        requirements=[ForbiddenWordReq()],
        chunking="word",
    ) as streamer:
        async for word in streamer:
            word_count += 1
            # Print every fifth word to show how many chunks arrive.
            if word_count % 5 == 1:
                print(f"  word {word_count:>3}: {word!r}")

    if streamer.streaming_failures:
        print(f"Failure: {streamer.streaming_failures[0][1].reason}")
    else:
        print(f"{word_count} word chunks total")


asyncio.run(main())

  word   1: 'Cloud-native'
  word   6: 'resilient'
  word  11: 'and'
  word  16: 'allows'
  word  21: 'horizontally,'
  word  26: 'costs,'
  word  31: 'deployments,'
  word  36: 'services.'
38 word chunks total

Note: LLM output is non-deterministic. Your result will vary in wording.

The same two-sentence response that produced 2 chunks with sentence chunking now produces 38. The validator fires on every word — maximum reaction speed at the cost of per-chunk context.

If a forbidden word appears, the stream stops at that word and no further chunks are delivered. To see early exit in action, change _FORBIDDEN to include a common English word like "and" or "the".

---

Step 4: Observing the event lifecycle

The async for loop above yields validated chunks. To observe the full lifecycle instead — per-chunk validation results, stream completion, final validation, errors — pass as_events=True: stream() then returns an EventStreamer that yields one typed StreamEvent per lifecycle moment in place of chunks.

# Requires: mellea
# Returns: None
import asyncio
import re

from mellea.core.backend import Backend
from mellea.core.base import Context
from mellea.core.requirement import PartialValidationResult, Requirement, ValidationResult
from mellea.stdlib.components import Instruction
from mellea.stdlib.streaming import (
    ChunkEvent,
    CompletedEvent,
    FullValidationEvent,
    QuickCheckEvent,
    StreamingDoneEvent,
    stream,
)

_SENTENCE_END = re.compile(r"[.!?]+")


class MaxSentencesReq(Requirement):
    def __init__(self, limit: int) -> None:
        super().__init__()
        self._limit = limit
        self._count = 0

    def format_for_llm(self) -> str:
        return f"The response must be at most {self._limit} sentences."

    async def _stream_validate(
        self, chunk: str, *, backend: Backend, ctx: Context
    ) -> PartialValidationResult:
        self._count += len(_SENTENCE_END.findall(chunk))
        if self._count > self._limit:
            return PartialValidationResult(
                "fail", reason=f"Exceeded {self._limit}-sentence limit"
            )
        return PartialValidationResult("unknown")

    async def validate(
        self, backend: Backend, ctx: Context, *, format=None, model_options=None
    ) -> ValidationResult:
        return ValidationResult(result=self._count <= self._limit)


async def main() -> None:
    from mellea.stdlib.session import start_session

    m = start_session()

    async with await stream(
        Instruction("Write a two-sentence summary of the water cycle."),
        m.backend,
        m.ctx,
        requirements=[MaxSentencesReq(limit=3)],
        chunking="sentence",
        as_events=True,
    ) as streamer:
        async for event in streamer:
            match event:
                case ChunkEvent():
                    print(f"  chunk[{event.chunk_index}]: {event.text!r}")
                case QuickCheckEvent(passed=False):
                    print(f"  FAIL at chunk {event.chunk_index}: {event.results[0].reason}")
                case StreamingDoneEvent():
                    print(f"  stream done — {len(event.full_text)} chars")
                case FullValidationEvent():
                    print(f"  final validation: {'pass' if event.passed else 'fail'}")
                case CompletedEvent():
                    print(f"  completed — success={event.success}")
                case _:
                    pass


asyncio.run(main())

  chunk[0]: 'Water evaporates from oceans and lakes, rises into the atmosphere, and condenses into clouds.'
  chunk[1]: 'Precipitation then falls back to Earth as rain or snow, replenishing rivers, lakes, and groundwater.'
  stream done — 195 chars
  final validation: pass
  completed — success=True

Note: LLM output is non-deterministic. Your result will vary in wording.

The event vocabulary:

• ChunkEvent — a validated chunk was delivered to the consumer.
• QuickCheckEvent — the result of validating one chunk; passed=False marks the requirement failure that ends the stream.
• StreamingDoneEvent — the token stream finished (natural completion only).
• FullValidationEvent — the final validate() pass over the whole output.
• CompletedEvent — the stream exited; always the last event, on every path.
• ErrorEvent — an exception occurred mid-stream.

Because the hook is global, a single plugin can observe many concurrent streams at once — use payload.streaming_id to tell their events apart. See docs/examples/streaming/multi_stream_events.py for a multi-stream consumer.

---

Step 5: A custom chunking strategy

The built-in strategies cover the most common boundaries. For structured output — numbered lists, code blocks, CSV rows — you can subclass ChunkingStrategy and define your own split boundary.

Two methods to implement:

• split(text) — return all complete chunks in text, withholding any trailing fragment. Must be stateless and idempotent.
• flush(text) — called once at natural end of stream. Release the withheld trailing fragment, or return [] to discard it.

Here is a LineChunking strategy that splits on single newlines — natural for numbered list output where each line is one item:

# Requires: mellea
# Returns: None
import asyncio
import re

from mellea.core.backend import Backend
from mellea.core.base import Context
from mellea.core.requirement import PartialValidationResult, Requirement, ValidationResult
from mellea.core.chunking import ChunkingStrategy
from mellea.stdlib.components import Instruction
from mellea.stdlib.streaming import stream

_NUMBERED_LINE = re.compile(r"^\s*\d+[\.\)]\s")


class LineChunking(ChunkingStrategy):
    """Emits one complete line per chunk, splitting on single newlines."""

    def split(self, text: str) -> list[str]:
        if "\n" not in text:
            return []
        last_nl = text.rfind("\n")
        return [line for line in text[:last_nl].split("\n") if line.strip()]

    def flush(self, text: str) -> list[str]:
        if not text:
            return []
        last_nl = text.rfind("\n")
        trailing = (text if last_nl == -1 else text[last_nl + 1 :]).strip()
        return [trailing] if trailing else []


class NumberedLineReq(Requirement):
    """Cancels the stream if any line does not begin with a number."""

    def format_for_llm(self) -> str:
        return "Every line must begin with a number followed by a period (e.g. '1. ')."

    async def _stream_validate(
        self, chunk: str, *, backend: Backend, ctx: Context
    ) -> PartialValidationResult:
        if not _NUMBERED_LINE.match(chunk):
            return PartialValidationResult(
                "fail", reason=f"Line does not start with a number: {chunk.strip()!r}"
            )
        return PartialValidationResult("unknown")

    async def validate(
        self, backend: Backend, ctx: Context, *, format=None, model_options=None
    ) -> ValidationResult:
        # All format checking happens during streaming. Lines that reach validate()
        # are guaranteed to have passed stream_validate() already.
        return ValidationResult(result=True)


async def main() -> None:
    from mellea.stdlib.session import start_session

    m = start_session()

    async with await stream(
        Instruction(
            "List five world capitals, one per line, numbered 1 through 5. "
            "Use the format: '1. City'. Output only the numbered list, nothing else."
        ),
        m.backend,
        m.ctx,
        requirements=[NumberedLineReq()],
        chunking=LineChunking(),
    ) as streamer:
        async for line in streamer:
            print(f"  line: {line.strip()!r}")

    if streamer.streaming_failures:
        print(f"FAIL: {streamer.streaming_failures[0][1].reason}")
    else:
        print("Completed normally")


asyncio.run(main())

  line: '1. London'
  line: '2. Paris'
  line: '3. Tokyo'
  line: '4. Ottawa'
  line: '5. Canberra'
Completed normally

Note: LLM output is non-deterministic. Your result will vary in wording.

validate() on NumberedLineReq always returns True because all format checking happens during streaming. If any line fails, the stream is cancelled before reaching validate(). Lines that do reach it have already passed _stream_validate(). This pattern — enforce in _stream_validate, pass in validate — is common for requirements whose invariant is a property of individual chunks rather than the full output.

Pass a ChunkingStrategy instance (not a string alias) to use a custom chunker. The built-in strategies (WordChunking, SentenceChunking, ParagraphChunking) are also available as instances if you need to pass one explicitly or subclass to override flush().

See also: docs/examples/streaming/custom_chunking.py for an annotated version of this pattern with a more detailed split()/flush() contract walkthrough.

---

Step 6: Requirement chunking

So far one chunking= on stream() decides the granularity for the consumer and every requirement. But a requirement can also carry its own chunking=, so different validators can validate at different granularities on the same stream — a sentence-level check and a word-level check side by side:

class MaxWordsPerSentence(Requirement):
    def __init__(self, limit: int) -> None:
        super().__init__(chunking="sentence")  # this requirement validates whole sentences
        self._limit = limit

    async def _stream_validate(
        self, chunk: str, *, backend: Backend, ctx: Context
    ) -> PartialValidationResult:
        if len(chunk.split()) > self._limit:
            return PartialValidationResult("fail", reason="sentence too long")
        return PartialValidationResult("unknown")


class NoBannedWord(Requirement):
    def __init__(self, banned: set[str]) -> None:
        super().__init__(chunking="word")  # this one validates word by word
        self._banned = banned

    async def _stream_validate(
        self, chunk: str, *, backend: Backend, ctx: Context
    ) -> PartialValidationResult:
        if chunk.strip(".,!?").lower() in self._banned:
            return PartialValidationResult("fail", reason=f"banned word: {chunk!r}")
        return PartialValidationResult("unknown")

When a requirement sets its own chunking=, the stream_validate driver feeds the stream chunks it receives to the requirement's own chunker and calls _stream_validate on each chunk it produces. Until that chunker completes a chunk, the requirement reports nothing yet (the neutral state). This composes with the stream's own chunking=, including chunking=None, where the consumer receives raw deltas while each requirement chunks independently.

Mixing granularities

A requirement's chunker is fed the stream's chunks, not the raw deltas, and it accumulates them in one stateful chunker — so the stream's chunks must carry the boundaries the requirement needs. The built-in strategies — WordChunking, SentenceChunking, and ParagraphChunking — all split on a form of whitespace and discard it, so when the stream and a requirement chunk differently, those boundaries never reach the requirement's chunker; no built-in pairing is safe.

A requirement coarser than the stream may never reach a boundary mid-stream; it stays "unknown" until the end-of-stream flush, which then validates the leftover as a single chunk, fused wherever the stream's chunker dropped a separator.

A finer requirement hits the same fusion mid-stream, and sentence → word is the clearest case: sentence chunking eats the space between sentences, which is exactly the boundary word chunking needs. Feed a word-level requirement the sentence chunks "A cat jumped the dog." then "The dog caught the horse."; with no space between them in the requirement's accumulated buffer, it sees:

["A", "cat", "jumped", "the", "dog.The", "dog", "caught", "the", "horse."]

"dog.The" fused across the boundary, so a check for the word "The" never fires. So set the stream to chunking=None whenever a requirement carries its own chunking and let each requirement re-chunk the raw deltas independently. A custom chunking strategy that keeps its separator can still compose with a different stream granularity; the constraint is specific to the built-in three.

See also: docs/examples/streaming/per_requirement_chunking.py for a runnable version with two requirements validating one stream at different granularities.

---

What you built

Concept	What it gives you
stream() + requirements=	Per-chunk validation with automatic early exit
async for chunk in streamer	Validated chunks as they arrive, inside async with for safe cleanup
streamer.failed_early / streamer.streaming_failures	Detect and inspect a mid-stream requirement failure
stream(as_events=True)	Typed event stream (an EventStreamer) — observe every chunk, validation result, and lifecycle signal
"word" / "sentence" / "paragraph"	Built-in chunking strategies trading reaction speed for context
ChunkingStrategy subclass	Custom split boundaries for structured output (lists, code, CSV)
Requirement chunking=	Each requirement validates at its own granularity, independent of the stream

---

See also: How-to: Streaming with per-chunk validation | Concepts: The Requirements System — Streaming validation | Examples: streaming/