Validating a stream

Every reply so far arrived whole. A support desk that makes the customer wait for the last word before showing the first one feels broken, and Guardrails validates a stream without waiting either.

A streaming model is a callable that yields pieces instead of returning a string. The stand-in from lesson 13 becomes one in three lines.

from guardrails import Guard
from guardrails_ai.valid_length import ValidLength


def stream(**kwargs):
    for part in ["Order 8821 ships today. ", "We can refund the postage. "]:
        yield part


desk = Guard().use(ValidLength(min=1, max=200, on_fail="noop"))

for chunk in desk(stream, messages=[{"role": "user", "content": "Where is 8821?"}], stream=True):
    print(repr(chunk.validated_output))

stream=True turns the Guard call into a generator of ValidationOutcome objects, one per validated piece. Each one has the same fields as the single outcome from lesson 14.

What a validator sees

The pieces the model yields are not the pieces the validator gets. Validators accumulate text until they have a sentence, which the Streaming page calls the default chunking strategy, and a validator can override it by defining _chunking_function.

@register_validator(name="shop/watching", data_type="string")
class Watching(Validator):
    def _validate(self, value: str, metadata: Dict) -> ValidationResult:
        print("validator saw:", repr(value))
        return PassResult()


desk = Guard().use(Watching(on_fail="noop"))

for chunk in desk(stream, messages=[{"role": "user", "content": "x"}], stream=True):
    pass

Two sentences, trimmed, arriving one at a time. For a validator that runs a model of its own, that is the difference between paying per sentence and paying for the whole answer again on every token.

Repairing a sentence mid-flight

desk = Guard().use(NoRefunds(on_fail="fix"))

for chunk in desk(stream, messages=[{"role": "user", "content": "x"}], stream=True):
    print(repr(chunk.validated_output))

The second sentence was repaired before it left the Guard, and the first was passed through untouched. Guardrails merges the fixes from every validator that touched a sentence; the Handling Fix Results for Streaming page describes the merge and admits it has edge cases when two repairs overlap.

This one is worth flagging. The Error and Remediation table says fix does not support streaming. On 0.11.0 it plainly does, and Guardrails has a whole documentation page about how. Where two pages disagree, run it.

The one that breaks

desk = Guard().use(NoRefunds(on_fail="refrain"))

try:
    for chunk in desk(stream, messages=[{"role": "user", "content": "x"}], stream=True):
        print(repr(chunk.validated_output))
except TypeError as error:
    print("TypeError:", error)

The first sentence streams, then the second one fails and Guardrails tries to add a Refrain marker to a string. The table said refrain does not support streaming, and this is what that means in practice: not ignored, not degraded, but a TypeError out of the middle of your response.

noop, exception and fix are the three that behave in a stream. If a rule needs to withhold an answer, withhold it before you start streaming.

Where the failures are

desk = Guard().use(NoRefunds(on_fail="noop"))

for chunk in desk(stream, messages=[{"role": "user", "content": "x"}], stream=True):
    pass

print(desk.error_spans_in_output())

error_spans_in_output gives character ranges into the answer so far, which is what you highlight in a chat window while the rest is still arriving.