Implementation:Guardrails ai Guardrails SequentialValidatorService

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

Overview

SequentialValidatorService is a synchronous implementation of the validator execution service that runs validators sequentially, with comprehensive support for both standard and streaming validation modes including fix-on-stream capabilities.

Description

The SequentialValidatorService class extends ValidatorServiceBase to provide synchronous, sequential validator execution. Unlike the AsyncValidatorService which runs validators concurrently, this service processes validators one at a time in order, making it suitable for use with synchronous Guards and contexts where validator ordering matters.

Key Capabilities

Standard Validation

The validate method performs a depth-first traversal of the value structure:

Recursively validates all list elements and dictionary values (children first).
Then validates the parent value against its assigned validators.
Supports all on-fail actions: NOOP, FIX, FIX_REASK, REASK, FILTER, REFRAIN, EXCEPTION, and CUSTOM.

When FIX_REASK is used, the fixed value is re-validated synchronously. If the recheck fails, the value is treated as a reask.

Streaming Validation

The module provides three streaming validation strategies:

run_validators_stream -- Dispatcher that selects between fix and noop stream strategies based on whether any validator uses OnFailAction.FIX.

run_validators_stream_fix -- Handles streaming validation with fix support. Accumulates chunks, runs each validator on each chunk, tracks partial accumulations per validator, and merges fixed values when all validators have produced concrete results. Supports refrain/filter abort. Handles the case where the LLM stream ends without a finished flag by validating remaining accumulated chunks.

run_validators_stream_noop -- Simpler streaming validation that yields each chunk after running it through all validators sequentially. Does not attempt to merge or accumulate across chunks.

Stream Validation Restrictions

When the stream flag is set during non-streaming run_validators, the service enforces restrictions: REASK, FIX, FIX_REASK, FILTER, and REFRAIN on-fail actions are not supported and raise ValueError.

Async Guard Check

The run_validator_sync method explicitly checks for coroutine results and raises a UserFacingException if an async validator is used with a synchronous Guard, providing a clear error message directing the user to use AsyncGuard.

Usage

Use SequentialValidatorService when you need synchronous, ordered validator execution. This is the default service used by the standard Guard class. It is also the only service that supports streaming validation with fix behavior.

Code Reference

Source Location

Repository: Guardrails
File: guardrails/validator_service/sequential_validator_service.py
Lines: 1-496

Signature

class SequentialValidatorService(ValidatorServiceBase):
    def run_validator_sync(
        self,
        validator: Validator,
        value: Any,
        metadata: Dict,
        validator_logs: ValidatorLogs,
        stream: Optional[bool] = False,
        *,
        validation_session_id: str,
        **kwargs,
    ) -> Optional[ValidationResult]: ...

    def run_validator(
        self,
        iteration: Iteration,
        validator: Validator,
        value: Any,
        metadata: Dict,
        property_path: str,
        stream: Optional[bool] = False,
        **kwargs,
    ) -> ValidatorLogs: ...

    def run_validators(
        self,
        iteration: Iteration,
        validator_map: ValidatorMap,
        value: Any,
        metadata: Dict[str, Any],
        absolute_property_path: str,
        reference_property_path: str,
        stream: Optional[bool] = False,
        **kwargs,
    ) -> Tuple[Any, Dict[str, Any]]: ...

    def run_validators_stream(
        self, iteration, validator_map, value_stream, metadata,
        absolute_property_path, reference_property_path, **kwargs,
    ) -> Iterator[StreamValidationResult]: ...

    def run_validators_stream_fix(
        self, iteration, validator_map, value_stream, metadata,
        absolute_property_path, reference_property_path, **kwargs,
    ) -> Iterator[StreamValidationResult]: ...

    def run_validators_stream_noop(
        self, iteration, validator_map, value_stream, metadata,
        absolute_property_path, reference_property_path, **kwargs,
    ) -> Iterator[StreamValidationResult]: ...

    def validate(
        self,
        value: Any,
        metadata: dict,
        validator_map: ValidatorMap,
        iteration: Iteration,
        absolute_path: str,
        reference_path: str,
        stream: Optional[bool] = False,
        **kwargs,
    ) -> Tuple[Any, dict]: ...

    def validate_stream(
        self,
        value_stream: Iterator[Tuple[Any, bool]],
        metadata: dict,
        validator_map: ValidatorMap,
        iteration: Iteration,
        absolute_path: str,
        reference_path: str,
        **kwargs,
    ) -> Iterator[StreamValidationResult]: ...

Import

from guardrails.validator_service.sequential_validator_service import SequentialValidatorService

I/O Contract

Inputs

validate

Name	Type	Required	Description
value	`Any`	Yes	The value to validate. Can be a primitive, list, or dictionary for recursive validation.
metadata	`dict`	Yes	Metadata dictionary passed to validators and updated with validator-returned metadata.
validator_map	`ValidatorMap`	Yes	Mapping of JSON paths to lists of Validator instances.
iteration	`Iteration`	Yes	The current iteration/step object for logging and session tracking.
absolute_path	`str`	Yes	The absolute JSON path (e.g., `"$.items.0.name"`).
reference_path	`str`	Yes	The reference JSON path for validator lookup (e.g., `"$.items.*.name"`).
stream	`Optional[bool]`	No	Whether to use streaming validation. Defaults to `False`.

validate_stream

Name	Type	Required	Description
value_stream	`Iterator[Tuple[Any, bool]]`	Yes	An iterator of (chunk, finished) tuples from the LLM stream. The boolean indicates whether this is the last chunk.
metadata	`dict`	Yes	Metadata dictionary.
validator_map	`ValidatorMap`	Yes	Mapping of JSON paths to Validator instances.
iteration	`Iteration`	Yes	The current iteration object.
absolute_path	`str`	Yes	The absolute JSON path.
reference_path	`str`	Yes	The reference JSON path for validator lookup.

Outputs

validate

Name	Type	Description
value	`Any`	The validated value, possibly corrected. May be `Filter`, `Refrain`, or `ReAsk`.
metadata	`dict`	Updated metadata dictionary.

validate_stream

Name	Type	Description
(yields)	`StreamValidationResult`	Each yielded result contains `chunk` (the validated/fixed text), `original_text` (the original chunk), and `metadata`.

Usage Examples

from guardrails.validator_service.sequential_validator_service import SequentialValidatorService

service = SequentialValidatorService()

# Standard validation
value, metadata = service.validate(
    value={"name": "John", "age": -5},
    metadata={},
    validator_map=my_validator_map,
    iteration=current_iteration,
    absolute_path="$",
    reference_path="$",
)

# Streaming validation
for result in service.validate_stream(
    value_stream=llm_stream,
    metadata={},
    validator_map=my_validator_map,
    iteration=current_iteration,
    absolute_path="$",
    reference_path="$",
):
    print(result.chunk)  # Validated chunk

Related Pages

Page Connections

Double-click a node to navigate. Hold to expand connections.

Principle

Implementation

Heuristic

Environment