Implementation:Guardrails ai Guardrails ValidationOutcome Stream

Knowledge Sources	Guardrails
Domains	Streaming, Data_Structure
Last Updated	2026-02-14 00:00 GMT

Overview

Concrete pattern for consuming streaming validation results provided by the guardrails package.

Description

During streaming, each yielded ValidationOutcome contains the incrementally validated text segment. After the stream completes, guard.history.last provides access to the full Call object containing the complete raw and validated output, validator logs, and iteration details.

Usage

Iterate over the streaming generator to consume chunks in real time. After iteration completes, access guard.history.last for the full result.

Code Reference

Source Location

• Repository: guardrails
• File: guardrails/classes/validation_outcome.py (L19-135), guardrails/run/stream_runner.py (L236-239)

Signature

# Per-chunk (same ValidationOutcome class)
class ValidationOutcome(IValidationOutcome, ArbitraryModel, Generic[OT]):
    validated_output: Optional[OT]  # The validated text chunk
    validation_passed: bool          # Per-chunk pass/fail
    raw_llm_output: Optional[str]    # Accumulated raw text so far

# Post-stream (via Guard.history)
guard.history  # Stack[Call]
guard.history.last  # Call object with full execution details

Import

from guardrails.classes.validation_outcome import ValidationOutcome
# guard.history is accessed as an attribute of the Guard instance

I/O Contract

Inputs

Name	Type	Required	Description
stream generator	Iterator[ValidationOutcome[OT]]	Yes	Generator from Guard.__call__(stream=True)

Outputs

Name	Type	Description
Per-chunk .validated_output	Optional[str]	The validated text chunk
Per-chunk .validation_passed	bool	Whether this chunk passed validation
guard.history.last	Call	Full call object after stream completes

Usage Examples

Real-Time Consumption

from guardrails import Guard
from guardrails.hub import ToxicLanguage

guard = Guard().use(ToxicLanguage(on_fail="fix"))

stream = guard(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Tell me a story."}],
    stream=True,
)

# Real-time per-chunk consumption
for chunk in stream:
    if chunk.validated_output:
        print(chunk.validated_output, end="", flush=True)

# Post-stream full result access
last_call = guard.history.last
print("\n\nFull validated output:", last_call.guarded_output)

Related Pages

Implements Principle

• Principle:Guardrails_ai_Guardrails_Stream_Result_Handling