AI/참고/guardrails-main/guardrails/guard.py

import contextvars
import json
import os
import uuid
from builtins import id as object_id
from typing import (
    Any,
    Callable,
    Dict,
    Generic,
    Iterator,
    List,
    Optional,
    Sequence,
    Union,
    cast,
    Iterable,
)
import warnings
from langchain_core.runnables import Runnable

from guardrails_ai.types import (
    Guard as IGuard,
    ValidationOutcome as IValidationOutcome,
    JSONSchema,
    Validator as ValidatorReference,
)
from opentelemetry import context as otel_context
from pydantic import ValidationError, field_validator
from pydantic.config import ConfigDict

from guardrails.api_client import GuardrailsApiClient
from guardrails.classes.output_type import OT
from guardrails.classes.rc import RC
from guardrails_ai.types import ErrorSpan
from guardrails.classes.validation.validation_summary import ValidationSummary
from guardrails.classes.validation_outcome import ValidationOutcome
from guardrails.classes.execution import GuardExecutionOptions
from guardrails.classes.generic import Stack
from guardrails.classes.history import Call
from guardrails.classes.history.call_inputs import CallInputs
from guardrails.classes.output_type import OutputTypes
from guardrails.classes.schema.processed_schema import ProcessedSchema
from guardrails.formatters import BaseFormatter, get_formatter
from guardrails.llm_providers import (
    get_llm_ask,
    model_is_supported_server_side,
)
from guardrails.logger import logger, set_scope
from guardrails.run import Runner, StreamRunner
from guardrails.schema.primitive_schema import primitive_to_schema
from guardrails.schema.pydantic_schema import pydantic_model_to_schema
from guardrails.schema.rail_schema import rail_file_to_schema, rail_string_to_schema
from guardrails.schema.validator import SchemaValidationError, validate_json_schema
from guardrails.stores.context import get_call_kwarg, set_call_kwargs, set_guard_name
from guardrails.hub_telemetry.hub_tracing import trace
from guardrails.types.on_fail import OnFailAction
from guardrails.types.simple import SimpleTypes
from guardrails.types.pydantic import ModelOrListOfModels
from guardrails.utils.safe_get import safe_get
from guardrails.utils.api_utils import extract_serializeable_metadata
from guardrails.utils.hub_telemetry_utils import HubTelemetry
from guardrails.telemetry import (
    trace_guard_execution,
    wrap_with_otel_context,
)
from guardrails.utils.validator_utils import (
    parse_validator_reference,
    verify_metadata_requirements,
)
from guardrails.validator_base import Validator
from guardrails.types import (
    ValidatorMap,
)

from guardrails.utils.structured_data_utils import (
    # Prevent duplicate declaration in the docs
    json_function_calling_tool as json_function_calling_tool_util,
    output_format_json_schema as output_format_json_schema,
)

from guardrails.settings import settings
from guardrails.decorators.experimental import experimental


class Guard(IGuard, Generic[OT]):
    """The Guard class.

    This class is the main entry point for using Guardrails. It can be
    initialized by one of the following patterns:

    - `Guard().use(...)`
    - `Guard.for_string(...)`
    - `Guard.for_pydantic(...)`
    - `Guard.for_rail(...)`
    - `Guard.for_rail_string(...)`

    The `__call__`
    method functions as a wrapper around LLM APIs. It takes in an LLM
    API, and optional prompt parameters, and returns a ValidationOutcome
    class that contains the raw output from
    the LLM, the validated output, as well as other helpful information.
    """

    history: Stack[Call]
    _history_max_length: int
    _use_server: bool

    # Pydantic Config
    model_config = ConfigDict(arbitrary_types_allowed=True)

    def __init__(
        self,
        *,
        id: Optional[str] = None,
        name: Optional[str] = None,
        description: Optional[str] = None,
        validators: Optional[List[ValidatorReference]] = None,
        output_schema: Optional[Dict[str, Any]] = None,
        base_url: Optional[str] = None,
        api_key: Optional[str] = None,
        history_max_length: Optional[int] = None,
        use_server: Optional[bool] = None,
    ):
        """Initialize the Guard with serialized validator references and an
        output schema.

        Output schema must be a valid JSON Schema.
        """
        # Shared Interface Properties
        id = id or str(uuid.uuid4())
        name = name or f"gr-{id}"

        # Defaults
        validators = validators or []
        output_schema = output_schema or {"type": "string"}
        history_max_length = history_max_length or 10

        # Init JSONSchema class
        json_schema = JSONSchema.model_validate(output_schema)

        # TODO: Support a sink for history so that it is not solely held in memory
        history: Stack[Call] = Stack(max_length=history_max_length)

        # Super Init
        super().__init__(
            id=id,
            name=name,
            description=description,
            validators=[],
            output_schema=json_schema,
            history=history,  # type: ignore - pyright doesn't understand pydantic overrides
        )

        ### Overrides ###
        self.validators = validators

        ### Legacy ##
        self._num_reasks = None
        self._rail: Optional[str] = None
        self._base_model: Optional[ModelOrListOfModels] = None

        ### Private ###
        self._validator_map: ValidatorMap = {}
        self._validators: List[Validator] = []
        self._output_type: OutputTypes = OutputTypes.__from_json_schema__(output_schema)
        self._exec_opts: GuardExecutionOptions = GuardExecutionOptions()
        self._hub_telemetry: HubTelemetry
        self._user_id: Optional[str] = None
        self._allow_metrics_collection: Optional[bool] = None
        self._output_formatter: Optional[BaseFormatter] = None
        self._api_key: Optional[str] = api_key
        self._base_url: Optional[str] = base_url
        self._api_client: Optional[GuardrailsApiClient] = GuardrailsApiClient(
            api_key=self._api_key, base_url=self._base_url
        )
        self._history_max_length = history_max_length
        self._use_server = (
            use_server if use_server is not None else settings.use_server or False
        )

        self.configure()

    @field_validator("output_schema")
    @classmethod
    def must_be_valid_json_schema(
        cls, output_schema: Optional[JSONSchema] = None
    ) -> Optional[JSONSchema]:
        if output_schema:
            try:
                validate_json_schema(
                    output_schema.model_dump(exclude_none=True, by_alias=True)
                )
            except SchemaValidationError as e:
                raise ValueError(f"{str(e)}\n{json.dumps(e.fields, indent=2)}")
        return output_schema

    def configure(
        self,
        *,
        num_reasks: Optional[int] = None,
        allow_metrics_collection: Optional[bool] = None,
    ):
        """Configure the Guard.

        Args:
            num_reasks (int, optional): The max times to re-ask the LLM
                if validation fails. Defaults to None.
            allow_metrics_collection (bool, optional): Whether to allow
                Guardrails to collect anonymous metrics.
                Defaults to None, and falls back to waht is
                    set via the `guardrails configure` command.
        """
        if num_reasks:
            self._set_num_reasks(num_reasks)
        self._load_rc()
        self._configure_hub_telemtry(allow_metrics_collection)

    def _set_num_reasks(self, num_reasks: Optional[int] = None) -> None:
        # Configure may check if num_reasks is none, but this method still needs to be
        # defensive for when it's called internally.  Setting a default parameter
        # doesn't help the case where the method is explicitly passed a 'None'.
        if num_reasks is None:
            logger.debug("_set_num_reasks called with 'None'.  Defaulting to 1.")
            self._num_reasks = 1
        else:
            self._num_reasks = num_reasks

    def _load_rc(self) -> None:
        rc = RC.load(logger)
        settings.rc = rc

    def _configure_hub_telemtry(
        self, allow_metrics_collection: Optional[bool] = None
    ) -> None:
        allow_metrics_collection = (
            settings.rc.enable_metrics is True
            if allow_metrics_collection is None
            else allow_metrics_collection
        )

        self._allow_metrics_collection = allow_metrics_collection

        # Initialize Hub Telemetry singleton and get the tracer
        self._hub_telemetry = HubTelemetry()
        self._hub_telemetry._enabled = allow_metrics_collection

        if allow_metrics_collection is True:
            # Get unique id of user from rc file
            self._user_id = settings.rc.id or ""

    def _fill_validator_map(self):
        for ref in self.validators:
            entry: List[Validator] = self._validator_map.get(ref.on, [])  # type: ignore
            # Check if the validator from the reference
            #   has an instance in the validator_map
            existing_instance: Optional[Validator] = None
            for v in entry:
                same_id = v.rail_alias == ref.id
                same_on_fail = v.on_fail_descriptor == ref.on_fail or (  # is default
                    v.on_fail_descriptor == OnFailAction.NOOP and not ref.on_fail
                )
                same_args = v.get_args() == ref.kwargs or (  # Both are empty
                    not v.get_args() and not ref.kwargs
                )
                if same_id and same_on_fail and same_args:
                    existing_instance = v
                    break
            if not existing_instance:
                validator = parse_validator_reference(ref)
                if validator:
                    entry.append(validator)
                self._validator_map[ref.on] = entry  # type: ignore

    def _fill_validators(self):
        self._validators = [
            v
            for v_list in [self._validator_map[k] for k in self._validator_map]
            for v in v_list
        ]

    def _fill_exec_opts(
        self,
        *,
        num_reasks: Optional[int] = None,
        messages: Optional[List[Dict]] = None,
        reask_messages: Optional[List[Dict]] = None,
        **kwargs,  # noqa
    ):
        """Backfill execution options from kwargs."""
        if num_reasks is not None:
            self._exec_opts.num_reasks = num_reasks
        if messages is not None:
            self._exec_opts.messages = messages
        if reask_messages is not None:
            self._exec_opts.reask_messages = reask_messages

    @classmethod
    def _for_rail_schema(
        cls,
        schema: ProcessedSchema,
        rail: str,
        *,
        name: Optional[str] = None,
        description: Optional[str] = None,
    ):
        guard = cls(
            name=name,
            description=description,
            output_schema=schema.json_schema,
            validators=schema.validators,
        )
        if schema.output_type == OutputTypes.STRING:
            guard = cast(Guard[str], guard)
        elif schema.output_type == OutputTypes.LIST:
            guard = cast(Guard[List], guard)
        else:
            guard = cast(Guard[Dict], guard)
        guard.configure()
        guard._validator_map = schema.validator_map
        guard._exec_opts = schema.exec_opts
        guard._output_type = schema.output_type
        guard._rail = rail
        guard._fill_validators()
        return guard

    @classmethod
    def for_rail(
        cls,
        rail_file: str,
        *,
        name: Optional[str] = None,
        description: Optional[str] = None,
    ):
        """Create a Guard using a `.rail` file to specify the output schema,
        prompt, etc.

        Args:
            rail_file: The path to the `.rail` file.
            name (str, optional): A unique name for this Guard. Defaults to `gr-` + the object id.
            description (str, optional): A description for this Guard. Defaults to None.

        Returns:
            An instance of the `Guard` class.
        """  # noqa

        schema = rail_file_to_schema(rail_file)
        return cls._for_rail_schema(
            schema,
            rail=rail_file,
            name=name,
            description=description,
        )

    @classmethod
    def for_rail_string(
        cls,
        rail_string: str,
        *,
        name: Optional[str] = None,
        description: Optional[str] = None,
    ):
        """Create a Guard using a `.rail` string to specify the output schema,
        prompt, etc..

        Args:
            rail_string: The `.rail` string.
            name (str, optional): A unique name for this Guard. Defaults to `gr-` + the object id.
            description (str, optional): A description for this Guard. Defaults to None.

        Returns:
            An instance of the `Guard` class.
        """  # noqa

        schema = rail_string_to_schema(rail_string)
        return cls._for_rail_schema(
            schema,
            rail=rail_string,
            name=name,
            description=description,
        )

    @classmethod
    def for_pydantic(
        cls,
        output_class: ModelOrListOfModels,
        *,
        reask_messages: Optional[List[Dict]] = None,
        messages: Optional[List[Dict]] = None,
        name: Optional[str] = None,
        description: Optional[str] = None,
        output_formatter: Optional[Union[str, BaseFormatter]] = None,
    ):
        """Create a Guard instance using a Pydantic model to specify the output
        schema.

        Args:
            output_class: (Union[Type[BaseModel], List[Type[BaseModel]]]): The pydantic model that describes
            the desired structure of the output.
            messages (List[Dict], optional): A list of messages to give to the llm. Defaults to None.
            reask_messages (List[Dict], optional): A list of messages to use during reasks. Defaults to None.
            name (str, optional): A unique name for this Guard. Defaults to `gr-` + the object id.
            description (str, optional): A description for this Guard. Defaults to None.
            output_formatter (str | Formatter, optional): 'none' (default), 'jsonformer', or a Guardrails Formatter.
        """  # noqa

        schema = pydantic_model_to_schema(output_class)
        exec_opts = GuardExecutionOptions(
            reask_messages=reask_messages,
            messages=messages,
        )
        guard = cls(
            name=name,
            description=description,
            output_schema=schema.json_schema,
            validators=schema.validators,
        )
        if schema.output_type == OutputTypes.LIST:
            guard = cast(Guard[List], guard)
        else:
            guard = cast(Guard[Dict], guard)
        guard.configure()
        guard._validator_map = schema.validator_map
        guard._exec_opts = exec_opts
        guard._output_type = schema.output_type
        guard._base_model = output_class
        if isinstance(output_formatter, str):
            if isinstance(output_class, list):
                raise Exception("""Root-level arrays are not supported with the 
                jsonformer argument, but can be used with other json generation methods.
                Omit the output_formatter argument to use the other methods.""")
            output_formatter = get_formatter(
                output_formatter,
                schema=output_class.model_json_schema(),  # type: ignore
            )
        guard._output_formatter = output_formatter
        guard._fill_validators()
        return guard

    @classmethod
    def for_string(
        cls,
        validators: Sequence[Validator],
        *,
        string_description: Optional[str] = None,
        reask_messages: Optional[List[Dict]] = None,
        messages: Optional[List[Dict]] = None,
        name: Optional[str] = None,
        description: Optional[str] = None,
    ):
        """Create a Guard instance for a string response.

        Args:
            validators: (List[Validator]): The list of validators to apply to the string output.
            string_description (str, optional): A description for the string to be generated. Defaults to None.
            messages (List[Dict], optional): A list of messages to pass to llm. Defaults to None.
            reask_messages (List[Dict], optional): A list of messages to use during reasks. Defaults to None.
            name (str, optional): A unique name for this Guard. Defaults to `gr-` + the object id.
            description (str, optional): A description for this Guard. Defaults to None.
        """  # noqa

        schema = primitive_to_schema(
            list(validators), type=SimpleTypes.STRING, description=string_description
        )
        exec_opts = GuardExecutionOptions(
            messages=messages,
            reask_messages=reask_messages,
        )
        guard = cast(
            Guard[str],
            cls(
                name=name,
                description=description,
                output_schema=schema.json_schema,
                validators=schema.validators,
            ),
        )
        guard.configure()
        guard._validator_map = schema.validator_map
        guard._exec_opts = exec_opts
        guard._output_type = schema.output_type
        guard._fill_validators()
        return guard

    def _execute(
        self,
        *args,
        llm_api: Optional[Callable] = None,
        llm_output: Optional[str] = None,
        prompt_params: Optional[Dict] = None,
        num_reasks: Optional[int] = None,
        messages: Optional[List[Dict]] = None,
        reask_messages: Optional[List[Dict]] = None,
        metadata: Optional[Dict],
        full_schema_reask: Optional[bool] = None,
        **kwargs,
    ) -> Union[ValidationOutcome[OT], Iterator[ValidationOutcome[OT]]]:
        self._fill_validator_map()
        self._fill_validators()
        self._fill_exec_opts(
            num_reasks=num_reasks,
            messages=messages,
            reask_messages=reask_messages,
        )
        metadata = metadata or {}
        # if not llm_output and llm_api and not (messages):
        #     raise RuntimeError("'messages' must be provided in order to call an LLM!")

        # check if validator requirements are fulfilled
        missing_keys = verify_metadata_requirements(metadata, self._validators)
        if missing_keys:
            raise ValueError(
                f"Missing required metadata keys: {', '.join(missing_keys)}"
            )

        def __exec(
            self: Guard,
            *args,
            llm_api: Optional[Callable] = None,
            llm_output: Optional[str] = None,
            prompt_params: Optional[Dict] = None,
            num_reasks: Optional[int] = None,
            messages: Optional[List[Dict]] = None,
            metadata: Optional[Dict] = None,
            full_schema_reask: Optional[bool] = None,
            **kwargs,
        ):
            prompt_params = prompt_params or {}
            metadata = metadata or {}
            if full_schema_reask is None:
                full_schema_reask = self._base_model is not None

            set_call_kwargs(kwargs)
            set_guard_name(self.name)

            self._set_num_reasks(num_reasks=num_reasks)
            if self._num_reasks is None:
                raise RuntimeError(
                    "`num_reasks` is `None` after calling `configure()`. "
                    "This should never happen."
                )

            input_messages = messages or self._exec_opts.messages
            call_inputs = CallInputs(
                llmApi=llm_api,
                messages=input_messages,
                promptParams=prompt_params,
                numReasks=self._num_reasks,
                metadata=metadata,
                fullSchemaReask=full_schema_reask,
                args=list(args),
                kwargs=kwargs,
            )

            if self._use_server and model_is_supported_server_side(
                llm_api, *args, **kwargs
            ):
                return self._call_server(
                    llm_output=llm_output,
                    llm_api=llm_api,
                    num_reasks=self._num_reasks,
                    prompt_params=prompt_params,
                    metadata=metadata,
                    full_schema_reask=full_schema_reask,
                    *args,
                    **kwargs,
                )

            call_log = Call(inputs=call_inputs)
            set_scope(str(object_id(call_log)))
            self.history.push(call_log)
            # Otherwise, call the LLM synchronously
            return self._exec(
                llm_api=llm_api,
                llm_output=llm_output,
                prompt_params=prompt_params,
                num_reasks=self._num_reasks,
                messages=messages,
                metadata=metadata,
                full_schema_reask=full_schema_reask,
                call_log=call_log,
                *args,
                **kwargs,
            )

        guard_context = contextvars.Context()

        # get the current otel context and wrap the subsequent call
        #   to preserve otel context if guard call is being called be another
        # framework upstream
        current_otel_context = otel_context.get_current()
        wrapped__exec = wrap_with_otel_context(current_otel_context, __exec)

        return guard_context.run(
            wrapped__exec,
            self,
            llm_api=llm_api,
            llm_output=llm_output,
            prompt_params=prompt_params,
            num_reasks=num_reasks,
            messages=messages,
            metadata=metadata,
            full_schema_reask=full_schema_reask,
            *args,
            **kwargs,
        )

    def _exec(
        self,
        *args,
        llm_api: Optional[Callable] = None,
        llm_output: Optional[str] = None,
        call_log: Call,  # Not optional, but internal
        prompt_params: Dict,  # Should be defined at this point
        num_reasks: int = 0,  # Should be defined at this point
        metadata: Dict,  # Should be defined at this point
        full_schema_reask: bool = False,  # Should be defined at this point
        messages: Optional[List[Dict]] = None,
        **kwargs,
    ) -> Union[ValidationOutcome[OT], Iterator[ValidationOutcome[OT]]]:
        api = None

        if llm_api is not None or kwargs.get("model") is not None:
            api = get_llm_ask(llm_api, *args, **kwargs)

        if self._output_formatter is not None:
            # Type suppression here? ArbitraryCallable is a subclass of PromptCallable!?
            api = self._output_formatter.wrap_callable(api)  # type: ignore

        # Check whether stream is set
        if kwargs.get("stream", False):
            # If stream is True, use StreamRunner
            runner = StreamRunner(
                output_type=self._output_type,
                output_schema=self.output_schema.model_dump(
                    exclude_none=True, by_alias=True
                ),
                num_reasks=num_reasks,
                validation_map=self._validator_map,
                messages=messages,
                api=api,
                metadata=metadata,
                output=llm_output,
                base_model=self._base_model,
                full_schema_reask=full_schema_reask,
                disable_tracer=(
                    not self._allow_metrics_collection
                    if isinstance(self._allow_metrics_collection, bool)
                    else None
                ),
                exec_options=self._exec_opts,
            )
            return runner(call_log=call_log, prompt_params=prompt_params)
        else:
            # Otherwise, use Runner
            runner = Runner(
                output_type=self._output_type,
                output_schema=self.output_schema.model_dump(
                    exclude_none=True, by_alias=True
                ),
                num_reasks=num_reasks,
                validation_map=self._validator_map,
                messages=messages,
                api=api,
                metadata=metadata,
                output=llm_output,
                base_model=self._base_model,
                full_schema_reask=full_schema_reask,
                disable_tracer=(
                    not self._allow_metrics_collection
                    if isinstance(self._allow_metrics_collection, bool)
                    else None
                ),
                exec_options=self._exec_opts,
            )
            call = runner(call_log=call_log, prompt_params=prompt_params)
            return ValidationOutcome[OT].from_guard_history(call)

    @trace(name="/guard_call", origin="Guard.__call__")
    def __call__(
        self,
        llm_api: Optional[Callable] = None,
        *args,
        prompt_params: Optional[Dict] = None,
        num_reasks: Optional[int] = 1,
        messages: Optional[List[Dict]] = None,
        metadata: Optional[Dict] = None,
        full_schema_reask: Optional[bool] = None,
        **kwargs,
    ) -> Union[ValidationOutcome[OT], Iterator[ValidationOutcome[OT]]]:
        """Call the LLM and validate the output.

        Args:
            llm_api: The LLM API to call
                     (e.g. openai.completions.create or openai.Completion.acreate)
            prompt_params: The parameters to pass to the prompt.format() method.
            num_reasks: The max times to re-ask the LLM for invalid output.
            messages: The message history to pass to the LLM.
            metadata: Metadata to pass to the validators.
            full_schema_reask: When reasking, whether to regenerate the full schema
                               or just the incorrect values.
                               Defaults to `True` if a base model is provided,
                               `False` otherwise.

        Returns:
            ValidationOutcome
        """

        messages = messages or self._exec_opts.messages or []

        if messages is not None and not len(messages):
            raise RuntimeError(
                "You must provide messages. "
                "Alternatively, you can provide messages in the Schema constructor."
            )

        return trace_guard_execution(
            self.name,
            self.history,
            self._execute,
            *args,
            llm_api=llm_api,
            prompt_params=prompt_params,
            num_reasks=num_reasks,
            messages=messages,
            metadata=metadata,
            full_schema_reask=full_schema_reask,
            **kwargs,
        )

    @trace(name="/guard_call", origin="Guard.parse")
    def parse(
        self,
        llm_output: str,
        *args,
        metadata: Optional[Dict] = None,
        llm_api: Optional[Callable] = None,
        num_reasks: Optional[int] = None,
        prompt_params: Optional[Dict] = None,
        full_schema_reask: Optional[bool] = None,
        **kwargs,
    ) -> ValidationOutcome[OT]:
        """Alternate flow to using Guard where the llm_output is known.

        Args:
            llm_output: The output being parsed and validated.
            metadata: Metadata to pass to the validators.
            llm_api: The LLM API to call
                     (e.g. openai.completions.create or openai.Completion.acreate)
            num_reasks: The max times to re-ask the LLM for invalid output.
            prompt_params: The parameters to pass to the prompt.format() method.
            full_schema_reask: When reasking, whether to regenerate the full schema
                               or just the incorrect values.

        Returns:
            ValidationOutcome
        """
        final_num_reasks = (
            num_reasks
            if num_reasks is not None
            else self._num_reasks
            if self._num_reasks is not None
            else 0
            if llm_api is None
            else 1
        )

        default_messages = self._exec_opts.messages if llm_api else None
        messages = kwargs.pop("messages", default_messages)

        return trace_guard_execution(
            self.name,
            self.history,
            self._execute,  # type: ignore # streams are supported for parse
            *args,
            llm_output=llm_output,
            llm_api=llm_api,
            prompt_params=prompt_params,
            num_reasks=final_num_reasks,
            messages=messages,
            metadata=metadata,
            full_schema_reask=full_schema_reask,
            **kwargs,
        )

    def error_spans_in_output(self) -> List[ErrorSpan]:
        """Get the error spans in the last output."""
        try:
            call = self.history.last
            if call:
                iter = call.iterations.last
                if iter:
                    llm_spans = iter.error_spans_in_output
                    return llm_spans
                return []
            return []
        except (AttributeError, TypeError):
            return []

    def __add_validators(self, validators: List[Validator], on: str = "output"):
        if on not in [
            "output",
            "messages",
        ] and not on.startswith("$"):
            warnings.warn(
                f"Unusual 'on' value: {on}!"
                "This value is typically one of "
                "'output', 'messages') "
                "or a JSON path starting with '$.'",
                UserWarning,
            )

        if on == "output":
            on = "$"

        validator_references = [
            ValidatorReference(
                id=validator.rail_alias,
                on=on,
                on_fail=validator.on_fail_descriptor,  # type: ignore
                kwargs=validator.get_args(),
            )
            for validator in validators
        ]
        retained_validator_refs = [v for v in self.validators if v.on != on]
        retained_validator_refs.extend(validator_references)

        self.validators = retained_validator_refs

        self._validator_map[on] = validators

        self._validators = [v for on_vs in self._validator_map.values() for v in on_vs]

    def use(
        self,
        *validator_spread: Validator,
        validators: List[Validator] = [],
        on: str = "output",
    ) -> "Guard":
        """Applies validators to the property specified in the `on` argument.
        Calling `Guard.use` with the same `on` value multiple times will
        overwrite previously configured validators on the specified property.

        Args:
            *validator_spread:
                One or more validators passed as positional arguments to use.
            validators:
                Keyword argument that allows explicitly setting a list of
                validators to use.
            on:
                The property to validate. Valid options include "output", "messages",
             or a JSON path starting with "$.". Defaults to "output".
        """
        vals = [*list(validator_spread), *validators]
        self.__add_validators(vals, on=on)
        return self

    def get_validators(self, on: str) -> List[Validator]:
        """The read-only counterpart to `Guard.use`. Retrieves the validators
        applied to the specified property.

        Args:
            on: The property for which to return configured validators.
                Valid options include "output", "messages",
                or a JSON path starting with "$.".
        """
        if on == "output":
            on = "$"
        return self._validator_map.get(on) or []

    @trace(name="/guard_call", origin="Guard.validate")
    def validate(self, llm_output: str, *args, **kwargs) -> ValidationOutcome[OT]:
        return self.parse(llm_output=llm_output, *args, **kwargs)

    # No call support for this until
    # https://github.com/guardrails-ai/guardrails/pull/525 is merged
    # def __call__(self, llm_output: str, *args, **kwargs) -> ValidationOutcome[str]:
    #     return self.validate(llm_output, *args, **kwargs)

    # TODO: Test generated history and override to_dict if necessary
    # def to_dict(self) -> Dict[str, Any]:
    #     pass

    def _single_server_call(self, *, payload: Dict[str, Any]) -> ValidationOutcome[OT]:
        if self._use_server and self._api_client:
            validation_output: IValidationOutcome = self._api_client.validate(
                guard=self,  # type: ignore
                openai_api_key=get_call_kwarg("api_key"),
                **payload,
            )
            if not validation_output:
                return ValidationOutcome[OT](
                    call_id="0",  # type: ignore
                    rawLlmOutput=None,
                    validatedOutput=None,
                    validationPassed=False,
                    error="The response from the server was empty!",
                )
            if os.environ.get("GUARD_HISTORY_ENABLED", "true").lower() == "true":
                guard_history = self._api_client.get_history(
                    self.name, validation_output.call_id
                )
                call_log = safe_get(
                    [
                        c
                        for c in guard_history
                        if c.get("id") == validation_output.call_id
                    ],
                    0,
                )
                if call_log:
                    try:
                        call = Call.model_validate(call_log)
                        # Only append the history from this call
                        self.history.append(call)
                    except ValidationError:
                        pass

            validation_summaries = validation_output.validation_summaries or []
            validation_summaries = [
                ValidationSummary(**v.model_dump()) for v in validation_summaries
            ]
            if not validation_summaries:
                call_log: Call = safe_get(
                    [c for c in self.history if c.id == validation_output.call_id], 0
                )
                if call_log and call_log.iterations.last:
                    validator_logs = call_log.iterations.last.validator_logs
                    validation_summaries = (
                        ValidationSummary.from_validator_logs_only_fails(validator_logs)
                    )

            # TODO: See if the below statement is still true
            # Our interfaces are too different for this to work right now.
            # Once we move towards shared interfaces for both the open source
            # and the api we can re-enable this.
            # return ValidationOutcome[OT].from_guard_history(call_log)
            validated_output = (
                cast(OT, validation_output.validated_output)
                if validation_output.validated_output
                else None
            )
            return ValidationOutcome[OT](
                call_id=validation_output.call_id,  # type: ignore
                rawLlmOutput=validation_output.raw_llm_output,
                validatedOutput=validated_output,
                validationPassed=(validation_output.validation_passed is True),
                validationSummaries=validation_summaries,
            )
        else:
            raise ValueError("Guard does not have an api client!")

    def _stream_server_call(
        self,
        *,
        payload: Dict[str, Any],
    ) -> Iterator[ValidationOutcome[OT]]:
        if self._use_server and self._api_client:
            validation_output: Optional[IValidationOutcome] = None
            response = self._api_client.stream_validate(
                guard=self,  # type: ignore
                openai_api_key=get_call_kwarg("api_key"),
                **payload,
            )
            for fragment in response:
                validation_output = fragment
                if validation_output is None:
                    yield ValidationOutcome[OT](
                        call_id="0",  # type: ignore
                        rawLlmOutput=None,
                        validatedOutput=None,
                        validationPassed=False,
                        error="The response from the server was empty!",
                    )
                else:
                    validated_output = (
                        cast(OT, validation_output.validated_output)
                        if validation_output.validated_output
                        else None
                    )
                    yield ValidationOutcome[OT](
                        call_id=validation_output.call_id,  # type: ignore
                        rawLlmOutput=validation_output.raw_llm_output,
                        validatedOutput=validated_output,
                        validationPassed=(validation_output.validation_passed is True),
                    )

            if os.environ.get("GUARD_HISTORY_ENABLED", "true").lower() == "true":
                if validation_output:
                    guard_history = self._api_client.get_history(
                        self.id, validation_output.call_id
                    )
                    if isinstance(guard_history, Sequence):
                        self.history.extend(
                            [Call.model_validate(call) for call in guard_history]
                        )
        else:
            raise ValueError("Guard does not have an api client!")

    def _call_server(
        self,
        *args,
        llm_output: Optional[str] = None,
        llm_api: Optional[Callable] = None,
        num_reasks: Optional[int] = None,
        prompt_params: Optional[Dict] = None,
        metadata: Optional[Dict] = {},
        full_schema_reask: Optional[bool] = True,
        **kwargs,
    ) -> Union[ValidationOutcome[OT], Iterator[ValidationOutcome[OT]]]:
        if self._use_server and self._api_client:
            payload: Dict[str, Any] = {
                "args": list(args),
                "full_schema_reask": full_schema_reask,
            }
            payload.update(**kwargs)
            if metadata:
                payload["metadata"] = extract_serializeable_metadata(metadata)
            if llm_output is not None:
                payload["llm_output"] = llm_output
            if num_reasks is not None:
                payload["num_reasks"] = num_reasks or self._exec_opts.num_reasks
            if prompt_params is not None:
                payload["prompt_params"] = prompt_params

            if not payload.get("messages"):
                payload["messages"] = self._exec_opts.messages
            if not payload.get("reask_messages"):
                payload["reask_messages"] = self._exec_opts.reask_messages

            should_stream = kwargs.get("stream", False)
            if should_stream:
                return self._stream_server_call(payload=payload)
            else:
                return self._single_server_call(
                    payload=payload,
                )
        else:
            raise ValueError("Guard does not have an api client!")

    def save(self):
        """Upserts a Guard to your guardrails-api server.

        Only valid for servers using a database to persist Guards. Not
        valid for servers using a config.py file.
        """
        if self.name is None:
            self.name = f"gr-{str(self.id)}"
            logger.warning("No name passed to guard!")
            logger.warning(
                "Use this auto-generated name to re-use this guard: {name}".format(
                    name=self.name
                )
            )
        if not self._api_client:
            self._api_client = GuardrailsApiClient(
                api_key=self._api_key, base_url=self._base_url
            )
        try:
            saved_guard = self._api_client.upsert_guard(self)
            self.id = saved_guard.id
            self._use_server = True
        except Exception as e:
            logger.error(
                f"Failed to save Guard with name {self.name}! Make sure your Guard is"
                " properly configured with an API Key, and base url."
            )
            raise e

    def to_runnable(self) -> Runnable:
        """Convert a Guard to a LangChain Runnable."""
        from guardrails.integrations.langchain.guard_runnable import GuardRunnable

        return GuardRunnable(self)

    # override IGuard.to_dict
    def to_dict(self) -> Dict[str, Any]:
        return self.model_dump(exclude_none=True, by_alias=True)

    @experimental
    def response_format_json_schema(self) -> Dict[str, Any]:
        return output_format_json_schema(schema=self._base_model)  # type: ignore

    def json_function_calling_tool(
        self,
        tools: Optional[list] = None,
    ) -> List[Dict[str, Any]]:
        """Appends an OpenAI tool that specifies the output structure using
        JSON Schema for chat models."""
        tools = json_function_calling_tool_util(
            tools=tools,
            # todo to_dict has a slight bug workaround here
            # but should fix in the long run dont have to
            # serialize and deserialize
            schema=json.loads(
                self.output_schema.model_dump_json(exclude_none=True, by_alias=True)
            ),
        )
        return tools

    # override IGuard.from_dict
    @classmethod
    def from_dict(cls, obj: Optional[Dict[str, Any]]) -> Optional["Guard"]:
        if obj is None:
            return None
        i_guard = IGuard.model_validate(obj)
        if not i_guard:
            return i_guard
        output_schema = (
            i_guard.output_schema.model_dump(exclude_none=True, by_alias=True)
            if i_guard.output_schema
            else None
        )

        guard = cls(
            id=i_guard.id,
            name=i_guard.name,
            description=i_guard.description,
            validators=[
                ValidatorReference.model_validate(i_val)
                for i_val in i_guard.validators or []
            ],
            output_schema=output_schema,
        )

        i_history = obj.get("history")
        history = []
        if isinstance(i_history, Iterable):
            for h in i_history:
                call = Call.model_validate(h)
                call._id = h.get("id") or call._id
                history.append(call)
        guard.history = Stack(*history, max_length=guard._history_max_length)

        guard._fill_validator_map()
        guard._fill_validators()
        return guard

    @classmethod
    def load(
        cls,
        name: str,
        *,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        history_max_length: Optional[int] = None,
    ) -> Optional["Guard"]:
        """Fetches and loads a Guard from your guardrails-api server."""
        api_client = GuardrailsApiClient(api_key=api_key, base_url=base_url)
        guard = api_client.fetch_guard(name)
        if guard:
            validators = guard.validators

            output_schema = (
                guard.output_schema.model_dump(exclude_none=True, by_alias=True)
                if guard.output_schema
                else {"type": "string"}
            )

            guard = cls(
                id=guard.id,
                name=guard.name,
                description=guard.description,
                validators=validators,
                output_schema=output_schema,
                base_url=base_url,
                api_key=api_key,
                history_max_length=history_max_length,
                use_server=True,
            )
            guard._fill_validator_map()
            guard._fill_validators()
            return guard
        return None

    def delete(self):
        """Deletes a Guard to your guardrails-api server.

        Only valid for servers using a database to persist Guards. Not
        valid for servers using a config.py file.
        """
        if self.name is None:
            self.name = f"gr-{str(self.id)}"
        if not self._api_client:
            self._api_client = GuardrailsApiClient(
                api_key=self._api_key, base_url=self._base_url
            )
        self._api_client.delete_guard(self.name)