Implementation:Guardrails ai Guardrails AsyncValidatorService
| Knowledge Sources | |
|---|---|
| Domains | Validation, Async Processing, Concurrency |
| Last Updated | 2026-02-14 00:00 GMT |
Overview
AsyncValidatorService is an asynchronous implementation of the validator execution service that runs validators concurrently using asyncio.gather, enabling parallel validation of values against multiple validators and child properties.
Description
The AsyncValidatorService class extends ValidatorServiceBase to provide fully asynchronous validator execution. Its key characteristics are:
Concurrent Validator Execution
When multiple validators are assigned to the same property path, run_validators collects all validator coroutines and executes them concurrently via asyncio.gather. This provides significant performance benefits when validators involve I/O operations (e.g., remote inference or API calls).
Concurrent Child Validation
The validate_children method concurrently validates all child properties of lists and dictionaries. For lists, each element is validated in parallel; for dictionaries, each key-value pair is validated in parallel. This enables deep concurrent validation of nested data structures.
Validation Flow
The validation pipeline follows a depth-first pattern:
- validate_children -- Recursively validates all child properties concurrently.
- run_validators -- Runs all validators for the current property path concurrently.
- Results are merged: fix values are combined, reasks are aggregated, and Filter/Refrain actions are returned immediately.
FIX_REASK Support
When a validator has on_fail_descriptor == OnFailAction.FIX_REASK, the service first applies the fix, then re-validates the fixed value asynchronously to determine if the fix resolved the issue or if a reask is still needed.
Hub Telemetry Integration
The execute_validator method is decorated with @async_trace for Guardrails Hub telemetry, tracking validator usage with the name /validator_usage.
Synchronous Bridge
The validate method provides a synchronous entry point that wraps the async validation in loop.run_until_complete, allowing the async service to be used from synchronous contexts.
Usage
Use AsyncValidatorService when your Guard is configured for async execution (e.g., via AsyncGuard) or when your validators involve I/O-bound operations that benefit from concurrent execution.
Code Reference
Source Location
- Repository: Guardrails
- File:
guardrails/validator_service/async_validator_service.py - Lines: 1-358
Signature
class AsyncValidatorService(ValidatorServiceBase):
async def execute_validator(
self,
validator: Validator,
value: Any,
metadata: Optional[Dict],
stream: Optional[bool] = False,
*,
validation_session_id: str,
**kwargs,
) -> Optional[ValidationResult]: ...
async def run_validator_async(
self,
validator: Validator,
value: Any,
metadata: Dict,
stream: Optional[bool] = False,
*,
validation_session_id: str,
**kwargs,
) -> ValidationResult: ...
async def run_validator(
self,
iteration: Iteration,
validator: Validator,
value: Any,
metadata: Dict,
absolute_property_path: str,
stream: Optional[bool] = False,
*,
reference_path: Optional[str] = None,
**kwargs,
) -> ValidatorRun: ...
async def run_validators(
self,
iteration: Iteration,
validator_map: ValidatorMap,
value: Any,
metadata: Dict,
absolute_property_path: str,
reference_property_path: str,
stream: Optional[bool] = False,
**kwargs,
) -> Tuple[Any, Dict]: ...
async def validate_children(
self,
value: Any,
metadata: Dict,
validator_map: ValidatorMap,
iteration: Iteration,
abs_parent_path: str,
ref_parent_path: str,
stream: Optional[bool] = False,
**kwargs,
) -> Tuple[Any, Dict]: ...
async def async_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(
self,
value: Any,
metadata: dict,
validator_map: ValidatorMap,
iteration: Iteration,
absolute_path: str,
reference_path: str,
loop: asyncio.AbstractEventLoop,
stream: Optional[bool] = False,
**kwargs,
) -> Tuple[Any, dict]: ...
Import
from guardrails.validator_service.async_validator_service import AsyncValidatorService
I/O Contract
Inputs
async_validate / validate
| Name | Type | Required | Description |
|---|---|---|---|
| value | Any |
Yes | The value to validate. Can be a primitive, list, or dictionary. |
| metadata | dict |
Yes | Metadata dictionary passed to validators and updated with validator-returned metadata. |
| validator_map | ValidatorMap |
Yes | Mapping of JSON paths (e.g., "$.name", "$.*") to lists of Validator instances.
|
| iteration | Iteration |
Yes | The current iteration/step object, used for logging and session tracking. |
| absolute_path | str |
Yes | The absolute JSON path of the current value in the output structure (e.g., "$.items.0.name").
|
| reference_path | str |
Yes | The reference JSON path used for validator lookup (e.g., "$.items.*.name").
|
| loop | asyncio.AbstractEventLoop |
Only for validate |
The event loop to run the async validation on (synchronous bridge only). |
| stream | Optional[bool] |
No | Whether to use streaming validation. Defaults to False.
|
Outputs
| Name | Type | Description |
|---|---|---|
| value | Any |
The validated (and possibly corrected) value. May be a Filter, Refrain, or FieldReAsk if validation triggered those actions.
|
| metadata | dict |
The potentially updated metadata dictionary. |
Usage Examples
import asyncio
from guardrails.validator_service.async_validator_service import AsyncValidatorService
service = AsyncValidatorService()
# Async validation
value, metadata = await service.async_validate(
value={"name": "John", "age": 25},
metadata={},
validator_map=my_validator_map,
iteration=current_iteration,
absolute_path="$",
reference_path="$",
)
# Synchronous bridge
loop = asyncio.get_event_loop()
value, metadata = service.validate(
value={"name": "John", "age": 25},
metadata={},
validator_map=my_validator_map,
iteration=current_iteration,
absolute_path="$",
reference_path="$",
loop=loop,
)