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)