Implementation:Guardrails ai Guardrails StreamRunner Step

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

Overview

Concrete method for processing streaming LLM chunks with incremental validation provided by the guardrails package.

Description

The StreamRunner.step method is the core streaming validation loop. It receives the LLM stream, processes each chunk by accumulating text, applying validator chunking functions for boundary detection, and yielding ValidationOutcome objects for each validated segment. For string output, chunks are split at sentence boundaries; for JSON output, progressive parsing validates partial structures.

Usage

This method is called internally by StreamRunner.__call__ during streaming Guard execution. Users do not call it directly but interact with it through the Guard's streaming generator.

Code Reference

Source Location

• Repository: guardrails
• File: guardrails/run/stream_runner.py
• Lines: L65-239

Signature

@trace_stream(name="/step", origin="StreamRunner.step")
@trace_stream_step
def step(
    self,
    index: int,
    api: Optional[PromptCallableBase],
    messages: Optional[List[Dict]],
    prompt_params: Dict,
    output_schema: Dict[str, Any],
    call_log: Call,
    output: Optional[str] = None,
) -> Iterator[ValidationOutcome[OT]]:
    """Run a full step."""

Import

from guardrails.run.stream_runner import StreamRunner
# Note: StreamRunner is typically used internally by Guard

I/O Contract

Inputs

Name	Type	Required	Description
index	int	Yes	Step index in the execution loop
api	Optional[PromptCallableBase]	No	LLM API callable with streaming support
messages	Optional[List[Dict]]	No	Chat messages for the LLM
prompt_params	Dict	Yes	Parameters for prompt template formatting
output_schema	Dict[str, Any]	Yes	JSON schema for output validation
call_log	Call	Yes	Call trace log for recording execution
output	Optional[str]	No	Pre-existing output to validate (no LLM call)

Outputs

Name	Type	Description
generator	Iterator[ValidationOutcome[OT]]	Yields ValidationOutcome for each validated chunk with raw_llm_output, validated_output, validation_passed

Usage Examples

Internal Usage (via Guard)

# StreamRunner is used internally; users interact via Guard:
from guardrails import Guard
from guardrails.hub import ToxicLanguage

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

# This internally creates StreamRunner and calls .step()
stream = guard(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Tell me a story."}],
    stream=True,
)

full_output = ""
for chunk in stream:
    full_output += chunk.validated_output or ""
    print(chunk.validated_output, end="", flush=True)

Related Pages

Implements Principle

• Principle:Guardrails_ai_Guardrails_Stream_Chunk_Processing

Uses Heuristic

• Heuristic:Guardrails_ai_Guardrails_Sentence_Tokenizer_Optimization