diff --git a/litellm/types/guardrails.py b/litellm/types/guardrails.py index 2725653d9ac..cbc54015aca 100644 --- a/litellm/types/guardrails.py +++ b/litellm/types/guardrails.py @@ -2,12 +2,14 @@ from collections.abc import Mapping from datetime import datetime from enum import Enum from types import MappingProxyType -from typing import Final, Literal +from typing import Final, Literal, cast -from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator +from pydantic import ConfigDict, Field, field_validator, model_validator from typing_extensions import ReadOnly, Required, TypedDict +from litellm._logging import verbose_logger from litellm.constants import BEDROCK_APPLY_GUARDRAIL_CHUNK_BUDGET_CHARS +from litellm.types.llms.base import LiteLLMBaseModel from litellm.types.proxy.guardrails.guardrail_hooks.agent_365 import ( Agent365GuardrailConfigModel, ) @@ -23,6 +25,9 @@ from litellm.types.proxy.guardrails.guardrail_hooks.cisco_ai_defense import ( from litellm.types.proxy.guardrails.guardrail_hooks.compresr import ( CompresrGuardrailConfigModel, ) +from litellm.types.proxy.guardrails.guardrail_hooks.decision_model import ( + DecisionModelCheck, +) from litellm.types.proxy.guardrails.guardrail_hooks.enkryptai import ( EnkryptAIGuardrailConfigs, ) @@ -145,9 +150,10 @@ class SupportedGuardrailIntegrations(Enum): STRAIKER = "straiker" ALICE = "alice" AGENT_365 = "agent_365" + LLM_SHIELD_PROXY = "llm_shield_proxy" CONDUCT = "conduct" - TEALTIGER = "tealtiger" DECISION_MODEL = "decision_model" + TEALTIGER = 'tealtiger' class Role(Enum): @@ -167,7 +173,7 @@ class GuardrailItemSpec(TypedDict, total=False): callback_args: dict[str, dict] -class GuardrailItem(BaseModel): +class GuardrailItem(LiteLLMBaseModel): callbacks: list[str] default_on: bool logging_only: bool | None @@ -437,7 +443,7 @@ class GuardrailParamUITypes(str, Enum): PERCENTAGE = "percentage" -class PresidioPresidioConfigModelUserInterface(BaseModel): +class PresidioPresidioConfigModelUserInterface(LiteLLMBaseModel): """Configuration parameters for the Presidio PII masking guardrail on LiteLLM UI""" presidio_analyzer_api_base: str | None = Field( @@ -546,31 +552,31 @@ BedrockChecksSensitiveInformationEntity = Literal[ ] -class BedrockChecksContentFilterCategoryItem(BaseModel): +class BedrockChecksContentFilterCategoryItem(LiteLLMBaseModel): category: BedrockChecksContentFilterCategory -class BedrockChecksContentFilterModel(BaseModel): +class BedrockChecksContentFilterModel(LiteLLMBaseModel): categories: list[BedrockChecksContentFilterCategoryItem] -class BedrockChecksPromptAttackCategoryItem(BaseModel): +class BedrockChecksPromptAttackCategoryItem(LiteLLMBaseModel): category: BedrockChecksPromptAttackCategory -class BedrockChecksPromptAttackModel(BaseModel): +class BedrockChecksPromptAttackModel(LiteLLMBaseModel): categories: list[BedrockChecksPromptAttackCategoryItem] -class BedrockChecksSensitiveInformationEntityItem(BaseModel): +class BedrockChecksSensitiveInformationEntityItem(LiteLLMBaseModel): type: BedrockChecksSensitiveInformationEntity -class BedrockChecksSensitiveInformationModel(BaseModel): +class BedrockChecksSensitiveInformationModel(LiteLLMBaseModel): entities: list[BedrockChecksSensitiveInformationEntityItem] -class BedrockChecksConfigModel(BaseModel): +class BedrockChecksConfigModel(LiteLLMBaseModel): """Inline `checks` config for the resource-less Bedrock InvokeGuardrailChecks API. Include only the checks you want to run; at least one must be set. @@ -589,7 +595,7 @@ class BedrockChecksConfigModel(BaseModel): return self -class BedrockGuardrailConfigModel(BaseModel): +class BedrockGuardrailConfigModel(LiteLLMBaseModel): """Configuration parameters for the AWS Bedrock guardrail""" guardrailIdentifier: str | None = Field(default=None, description="The ID of your guardrail on Bedrock") @@ -664,7 +670,7 @@ class BedrockGuardrailConfigModel(BaseModel): ) -class BedrockGuardrailStreamingParams(BaseModel): +class BedrockGuardrailStreamingParams(LiteLLMBaseModel): streaming_buffer_until_moderated: bool = Field( default=True, description="If True (default), withhold every streamed chunk until the end-of-stream " @@ -704,7 +710,7 @@ class BedrockGuardrailStreamingParams(BaseModel): ) -class LakeraV2GuardrailConfigModel(BaseModel): +class LakeraV2GuardrailConfigModel(LiteLLMBaseModel): """Configuration parameters for the Lakera AI v2 guardrail""" api_key: str | None = Field(default=None, description="API key for the Lakera AI service") @@ -729,7 +735,7 @@ class LakeraV2GuardrailConfigModel(BaseModel): ) -class LassoGuardrailConfigModel(BaseModel): +class LassoGuardrailConfigModel(LiteLLMBaseModel): """Configuration parameters for the Lasso guardrail""" lasso_user_id: str | None = Field(default=None, description="User ID for the Lasso guardrail") @@ -737,7 +743,7 @@ class LassoGuardrailConfigModel(BaseModel): mask: bool | None = Field(default=False, description="Enable content masking using Lasso classifix API") -class DeepKeepGuardrailConfigModel(BaseModel): +class DeepKeepGuardrailConfigModel(LiteLLMBaseModel): """Configuration parameters for the DeepKeep AI Firewall guardrail""" deepkeep_firewall_id: str | None = Field( @@ -749,7 +755,7 @@ class DeepKeepGuardrailConfigModel(BaseModel): ) -class PillarGuardrailConfigModel(BaseModel): +class PillarGuardrailConfigModel(LiteLLMBaseModel): """Configuration parameters for the Pillar Security guardrail""" on_flagged_action: str | None = Field( @@ -774,7 +780,7 @@ class PillarGuardrailConfigModel(BaseModel): ) -class NomaGuardrailConfigModel(BaseModel): +class NomaGuardrailConfigModel(LiteLLMBaseModel): """Configuration parameters for the Noma Security guardrail""" use_v2: bool | None = Field( @@ -803,7 +809,7 @@ class NomaGuardrailConfigModel(BaseModel): ) -class ZscalerAIGuardConfigModel(BaseModel): +class ZscalerAIGuardConfigModel(LiteLLMBaseModel): """Configuration parameters for the Zscaler AI Guard guardrail""" policy_id: int | None = Field( @@ -821,7 +827,7 @@ class ZscalerAIGuardConfigModel(BaseModel): ) -class JavelinGuardrailConfigModel(BaseModel): +class JavelinGuardrailConfigModel(LiteLLMBaseModel): """Configuration parameters for the Javelin guardrail""" guard_name: str | None = Field(default=None, description="Name of the Javelin guard to use") @@ -838,7 +844,7 @@ class ContentFilterAction(str, Enum): MASK = "MASK" -class BlockedWord(BaseModel): +class BlockedWord(LiteLLMBaseModel): """Represents a blocked word with its action and optional description""" keyword: str = Field(description="The keyword to block or mask") @@ -849,7 +855,7 @@ class BlockedWord(BaseModel): ) -class ContentFilterPattern(BaseModel): +class ContentFilterPattern(LiteLLMBaseModel): """Represents a content filter pattern (prebuilt or custom regex)""" pattern_type: Literal["prebuilt", "regex"] = Field( @@ -870,7 +876,7 @@ class ContentFilterPattern(BaseModel): action: ContentFilterAction = Field(description="Action to take when pattern matches (BLOCK or MASK)") -class ContentFilterConfigModel(BaseModel): +class ContentFilterConfigModel(LiteLLMBaseModel): """Configuration parameters for the content filter guardrail""" patterns: list[ContentFilterPattern] | None = Field( @@ -901,6 +907,95 @@ class ContentFilterConfigModel(BaseModel): MCP_SECURITY_ON_VIOLATION: Final = frozenset({"block", "alert"}) +GuardrailStreamScope = Literal["streaming", "non_streaming", "both"] +DEFAULT_GUARDRAIL_STREAM_SCOPE: Final[GuardrailStreamScope] = "both" + + +class GuardrailEventHooks(str, Enum): + pre_call = "pre_call" + post_call = "post_call" + during_call = "during_call" + logging_only = "logging_only" + pre_mcp_call = "pre_mcp_call" + during_mcp_call = "during_mcp_call" + post_mcp_call = "post_mcp_call" + realtime_input_transcription = "realtime_input_transcription" + + +GUARDRAIL_EVENT_HOOK_VALUES: Final = frozenset(member.value for member in GuardrailEventHooks) + +_GUARDRAIL_STREAM_SCOPES: Final[Mapping[str, GuardrailStreamScope]] = MappingProxyType( + { + "streaming": "streaming", + "non_streaming": "non_streaming", + "both": "both", + } +) + + +def _as_guardrail_stream_scope(value: object) -> GuardrailStreamScope: + if not isinstance(value, str): + raise ValueError(f"stream_scope values must be strings, got {type(value).__name__}") + scope: Final = _GUARDRAIL_STREAM_SCOPES.get(value.lower()) + if scope is None: + raise ValueError(f"stream_scope must be one of both, streaming, non_streaming, got {value!r}") + return scope + + +def _validated_stream_scope_hook(key: object) -> str: + if not isinstance(key, str): + raise ValueError(f"stream_scope keys must be strings, got {type(key).__name__}") + hook: Final = key.lower() + if hook not in GUARDRAIL_EVENT_HOOK_VALUES: + raise ValueError( + f"stream_scope keys must be guardrail modes ({sorted(GUARDRAIL_EVENT_HOOK_VALUES)}), got {key!r}" + ) + return hook + + +def coerce_stream_scope(value: object) -> GuardrailStreamScope | dict[str, GuardrailStreamScope] | None: + if value is None: + return None + if isinstance(value, str): + return _as_guardrail_stream_scope(value) + if isinstance(value, Mapping): + scope_map: Final[Mapping[str, object]] = cast(Mapping[str, object], value) # cast-ok: keys validated below + return { + _validated_stream_scope_hook(key): _as_guardrail_stream_scope(scope) for key, scope in scope_map.items() + } + raise ValueError(f"stream_scope must be a string or mapping, got {type(value).__name__}") + + +def stored_stream_scope(value: object) -> GuardrailStreamScope | dict[str, GuardrailStreamScope] | None: + try: + return coerce_stream_scope(value) + except ValueError: + verbose_logger.warning("Ignoring invalid stored stream_scope value of type %s", type(value).__name__) + return None + + +def with_tolerated_stream_scope(params: Mapping[str, object]) -> dict[str, object]: + if "stream_scope" not in params: + return dict(params) + return { + **params, + "stream_scope": stored_stream_scope(params["stream_scope"]), + } + + +def runtime_stream_scope( + stream_scope: object, +) -> tuple[GuardrailStreamScope, MappingProxyType[str, GuardrailStreamScope]]: + coerced: Final = coerce_stream_scope(stream_scope) + if coerced is None: + return DEFAULT_GUARDRAIL_STREAM_SCOPE, MappingProxyType({}) + if isinstance(coerced, str): + return coerced, MappingProxyType({}) + return DEFAULT_GUARDRAIL_STREAM_SCOPE, MappingProxyType(coerced) + + +LoggingOnlyScope = Literal["input", "output"] + class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch update guardrails api_key: str | None = Field(default=None, description="API key for the guardrail service") @@ -1043,9 +1138,11 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up skip_unscannable_attachments: bool | None = Field( default=False, description=( - "Implemented by guardrail='model_armor'. When True, attachment references that carry no " - "inline bytes (file_id, gs://, or http(s) URLs) pass through unscanned instead of blocking, " - "while fail_on_error still governs real Model Armor API errors. Default False blocks them." + "Implemented by guardrail='model_armor' and guardrail='bedrock'. When True, attachments the " + "guardrail cannot scan pass through unscanned instead of blocking. For Model Armor these are " + "references with no inline bytes (file_id, gs://, or http(s) URLs), and fail_on_error still " + "governs real Model Armor API errors. For Bedrock these are documents, files, audio, video, " + "and images that are not inline PNG or JPEG up to 4 MB. Default False blocks them." ), ) sanitize_error_detail: bool | None = Field( @@ -1066,10 +1163,23 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up description=( "Behavior when a guardrail endpoint is unreachable due to network errors. " "Implemented by guardrail='generic_guardrail_api', 'agent_365', 'akto', 'vigil_guard', 'repelloai', 'headroom', 'compresr', and 'typesafe'. " - "'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed." + "'fail_closed' raises an error (default). 'fail_open' logs a critical error and allows the request to proceed. " + "Also implemented by guardrail='decision_model'." ), ) + max_input_chars: int | None = Field( + default=None, + gt=0, + description="Character budget for each text sent to the guardrail's model. Implemented by guardrail='decision_model'.", + ) + + max_concurrent_decision_calls: int | None = Field( + default=None, + gt=0, + description="Maximum decisions calls in flight at once on the guardrail in each proxy worker. Implemented by guardrail='decision_model'.", + ) + extra_headers: list[str] | None = Field( default=None, description=( @@ -1142,6 +1252,39 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up ), ) + stream_scope: GuardrailStreamScope | dict[str, GuardrailStreamScope] | None = Field( + default=None, + description=( + "Whether this guardrail runs on streaming requests, non-streaming requests, or both. " + "A string applies to every configured mode. A map overrides named modes " + "(pre_call, during_call, post_call, ...); omitted keys default to both. " + "Unset means both, matching historical behavior." + ), + ) + + @field_validator("stream_scope", mode="before") + @classmethod + def normalize_stream_scope(cls, v: object) -> GuardrailStreamScope | dict[str, GuardrailStreamScope] | None: + return coerce_stream_scope(v) + + logging_only_scope: LoggingOnlyScope | None = Field( + default=None, + description=( + "which direction a logging_only scan observes: 'input' (request) or 'output' (response); " + "unset scans both directions. Only applies to mode logging_only; pre_call/post_call on the " + "same guardrail keep blocking." + ), + ) + + logging_only_continue_on_input_failure: bool | None = Field( + default=None, + description=( + "when True, a flagged or raising logging_only request scan is logged and the response is " + "still scanned, so both verdicts land. Only applies to mode logging_only and is ignored " + "when logging_only_scope is 'input' or 'output'." + ), + ) + @field_validator( "mode", "default_action", @@ -1169,7 +1312,7 @@ class BaseLitellmParams(ContentFilterConfigModel): # works for new and patch up model_config = ConfigDict(extra="allow", protected_namespaces=()) -class Mode(BaseModel): +class Mode(LiteLLMBaseModel): tags: dict[str, str | list[str]] = Field(description="Tags for the guardrail mode") default: str | list[str] | None = Field(default=None, description="Default mode when no tags match") @@ -1207,6 +1350,17 @@ class LitellmParams( # pyright: ignore[reportIncompatibleVariableOverride] # o Agent365GuardrailConfigModel, ): guardrail: str = Field(description="The type of guardrail integration to use") + decision_model: str | None = Field( + default=None, + description="For guardrail='decision_model': the Decisions-API model that scores each check", + ) + checks: BedrockChecksConfigModel | tuple[DecisionModelCheck, ...] | None = Field( # pyright: ignore[reportIncompatibleVariableOverride] # widened to also accept decision-model checks + default=None, + description=( + "Inline Bedrock InvokeGuardrailChecks config for guardrail='bedrock', or the predicate " + "checks a decision_model guardrail scores for guardrail='decision_model'" + ), + ) mode: str | list[str] | Mode = Field( description="When to apply the guardrail (pre_call, post_call, during_call, logging_only)" ) @@ -1232,6 +1386,14 @@ class LitellmParams( # pyright: ignore[reportIncompatibleVariableOverride] # o raise ValueError(f"on_violation={self.on_violation!r} is only supported by guardrail='mcp_security'") return self + @model_validator(mode="after") + def validate_checks_list_for_guardrail(self) -> "LitellmParams": + if isinstance(self.checks, tuple) and self.guardrail != SupportedGuardrailIntegrations.DECISION_MODEL.value: + raise ValueError( + f"checks as a list is only supported by guardrail='decision_model', got guardrail={self.guardrail!r}" + ) + return self + def __init__(self, **kwargs) -> None: default_on: Final = kwargs.pop("default_on", None) if default_on is not None: @@ -1268,17 +1430,6 @@ class guardrailConfig(TypedDict): guardrails: list[Guardrail] -class GuardrailEventHooks(str, Enum): - pre_call = "pre_call" - post_call = "post_call" - during_call = "during_call" - logging_only = "logging_only" - pre_mcp_call = "pre_mcp_call" - during_mcp_call = "during_mcp_call" - post_mcp_call = "post_mcp_call" - realtime_input_transcription = "realtime_input_transcription" - - class DynamicGuardrailParams(TypedDict): extra_body: ReadOnly[dict[str, object]] @@ -1288,7 +1439,7 @@ class GUARDRAIL_DEFINITION_LOCATION(str, Enum): CONFIG = "config" -class GuardrailInfoResponse(BaseModel): +class GuardrailInfoResponse(LiteLLMBaseModel): guardrail_id: str | None = None guardrail_name: str litellm_params: BaseLitellmParams | None = None @@ -1301,20 +1452,22 @@ class GuardrailInfoResponse(BaseModel): super().__init__(**kwargs) -class ListGuardrailsResponse(BaseModel): +class ListGuardrailsResponse(LiteLLMBaseModel): guardrails: list[GuardrailInfoResponse] -class GuardrailUIAddGuardrailSettings(BaseModel): +class GuardrailUIAddGuardrailSettings(LiteLLMBaseModel): supported_entities: list[str] supported_actions: list[str] supported_modes: list[str] supported_modes_by_provider: dict[str, list[str]] + providers_without_directional_logging_only_scope: tuple[str, ...] pii_entity_categories: list[PiiEntityCategoryMap] content_filter_settings: dict[str, object] | None = None + decision_model_providers: tuple[str, ...] = Field(default_factory=tuple) -class PresidioPerRequestConfig(BaseModel): +class PresidioPerRequestConfig(LiteLLMBaseModel): """ presdio params that can be controlled per request, api key """ @@ -1323,7 +1476,7 @@ class PresidioPerRequestConfig(BaseModel): entities: list[PiiEntityType] | None = None -class ApplyGuardrailRequest(BaseModel): +class ApplyGuardrailRequest(LiteLLMBaseModel): guardrail_name: str text: str language: str | None = None @@ -1333,11 +1486,11 @@ class ApplyGuardrailRequest(BaseModel): metadata: dict[str, object] | None = None -class ApplyGuardrailResponse(BaseModel): +class ApplyGuardrailResponse(LiteLLMBaseModel): response_text: str -class PatchGuardrailRequest(BaseModel): +class PatchGuardrailRequest(LiteLLMBaseModel): guardrail_name: str | None = None litellm_params: BaseLitellmParams | None = None guardrail_info: dict[str, object] | None = None