Implementation:Guardrails ai Guardrails ValidatorServiceBase

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

Overview

Defines the base class for validator service implementations, providing common logic for executing validators, handling on-fail actions, and managing validator lifecycle logging.

Description

The ValidatorServiceBase class is the foundational service layer for running validators within the Guardrails framework. It provides the following capabilities:

• execute_validator -- Executes a single validator against a value, wrapping the call with telemetry tracing via trace_validator. It supports both standard validation and streaming validation modes. This method is decorated with @trace for hub telemetry.

• perform_correction -- Handles the on-fail action for a FailResult. It implements the full set of on-fail behaviors:
  • FIX -- Returns the fix_value from the result
  • FIX_REASK -- Returns the fixed value, or a FieldReAsk if re-validation fails
  • CUSTOM -- Invokes the validator's custom on_fail_method
  • REASK -- Returns a FieldReAsk with the incorrect value
  • EXCEPTION -- Raises a ValidationError
  • FILTER -- Returns a Filter action
  • REFRAIN -- Returns a Refrain action
  • NOOP -- Returns the original value unchanged

• before_run_validator -- Creates a ValidatorLogs entry, records the start time, and appends it to the iteration's output logs.

• after_run_validator -- Records the end time and validation result on the ValidatorLogs entry.

• run_validator -- Abstract method (raises NotImplementedError) to be implemented by subclasses for the actual validation orchestration.

• multi_merge -- Merges multiple string results into one by successively applying a three-way merge against the original value.

• merge_results -- Merges multiple result values of any type by serializing, merging, and deserializing, with a fallback to the first new value if deserialization fails.

The module also defines a ValidatorRun dataclass that packages a value, metadata, on-fail action, and validator logs for passing between service methods.

Usage

This class is not used directly; it is subclassed by concrete validator service implementations (e.g., SequentialValidatorService, AsyncValidatorService). Use it as the base when implementing a new validator execution strategy.

Code Reference

Source Location

• Repository: Guardrails
• File: guardrails/validator_service/validator_service_base.py

Signature

ValidatorResult = Optional[Union[ValidationResult, Awaitable[ValidationResult]]]

@dataclass
class ValidatorRun:
    value: Any
    metadata: Dict
    on_fail_action: Union[str, OnFailAction]
    validator_logs: ValidatorLogs

class ValidatorServiceBase:
    def __init__(self, disable_tracer: Optional[bool] = True) -> None: ...

    def execute_validator(
        self,
        validator: Validator,
        value: Any,
        metadata: Optional[Dict],
        stream: Optional[bool] = False,
        *,
        validation_session_id: str,
        **kwargs,
    ) -> ValidatorResult: ...

    def perform_correction(
        self,
        result: FailResult,
        value: Any,
        validator: Validator,
        rechecked_value: Optional[ValidationResult] = None,
    ) -> Any: ...

    def before_run_validator(
        self,
        iteration: Iteration,
        validator: Validator,
        value: Any,
        absolute_property_path: str,
    ) -> ValidatorLogs: ...

    def after_run_validator(
        self,
        validator: Validator,
        validator_logs: ValidatorLogs,
        result: Optional[ValidationResult],
    ) -> ValidatorLogs: ...

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

    def multi_merge(self, original: str, new_values: list[str]) -> Optional[str]: ...

    def merge_results(self, original_value: Any, new_values: list[Any]) -> Any: ...

Import

from guardrails.validator_service.validator_service_base import (
    ValidatorServiceBase,
    ValidatorRun,
    ValidatorResult,
)

I/O Contract

execute_validator

Parameter	Type	Description
validator	Validator	The validator instance to execute
value	Any	The value to validate
metadata	Optional[Dict]	Additional metadata for validation context
stream	Optional[bool]	If True, uses validate_stream instead of validate
validation_session_id	str	Unique session identifier for tracing

Returns: ValidatorResult (Optional[Union[ValidationResult, Awaitable[ValidationResult]]])

perform_correction

Parameter	Type	Description
result	FailResult	The failed validation result
value	Any	The original value that failed validation
validator	Validator	The validator that produced the failure
rechecked_value	Optional[ValidationResult]	Result of re-checking a fixed value (used by FIX_REASK)

Returns: Varies by on-fail action -- may return the fixed value, a FieldReAsk, Filter, Refrain, the original value, or the result of a custom handler.

Raises:

• ValidationError -- When on_fail_descriptor is EXCEPTION
• ValueError -- When on_fail_descriptor is CUSTOM but no handler is set, or when the descriptor is unrecognized

ValidatorRun Dataclass

Field	Type	Description
value	Any	The value after validation/correction
metadata	Dict	Metadata passed through validation
on_fail_action	Union[str, OnFailAction]	The on-fail action that was applied
validator_logs	ValidatorLogs	Logs from the validation run

Usage Examples

from guardrails.validator_service.validator_service_base import ValidatorServiceBase

# Typically used via a subclass
class MyValidatorService(ValidatorServiceBase):
    def run_validator(self, iteration, validator, value, metadata,
                      absolute_property_path, stream=False, **kwargs):
        validator_logs = self.before_run_validator(
            iteration, validator, value, absolute_property_path
        )

        result = self.execute_validator(
            validator, value, metadata,
            stream=stream,
            validation_session_id="session-123",
            **kwargs,
        )

        self.after_run_validator(validator, validator_logs, result)

        from guardrails.classes.validation.validation_result import FailResult
        if isinstance(result, FailResult):
            value = self.perform_correction(result, value, validator)

        return ValidatorRun(
            value=value,
            metadata=metadata,
            on_fail_action=validator.on_fail_descriptor,
            validator_logs=validator_logs,
        )

Related Pages

• Guardrails_ai_Guardrails_Validator_Tracing -- Provides tracing decorators used by execute_validator
• Guardrails_ai_Guardrails_Validator -- The Validator base class executed by this service
• Guardrails_ai_Guardrails_OnFailAction -- Defines the on-fail action enum used by perform_correction