mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-06 02:48:13 +00:00
Merge e5cff43c82 into f285229b51
This commit is contained in:
Caduri
GitHub
commit
493d167eb0
14 changed files with 2622 additions and 17 deletions
|
|
@ -8,9 +8,13 @@ skip the other shapes — these helpers normalise that so every hook sees
|
|||
every text fragment.
|
||||
"""
|
||||
|
||||
import json
|
||||
from collections.abc import Callable, Iterator, Mapping, Sequence
|
||||
from typing import Any, Final
|
||||
|
||||
from pydantic import JsonValue
|
||||
from pydantic_core import to_jsonable_python
|
||||
|
||||
# Call types whose body carries free-form chat / prompt text that
|
||||
# text-content guardrails (banned keywords, content moderation, secret
|
||||
# detection, …) should inspect. The proxy ingress passes ``route_type``
|
||||
|
|
@ -146,6 +150,51 @@ def iter_message_text(data: Mapping[str, object]) -> Iterator[str]:
|
|||
yield from _iter_text_parts_in_content(message.get("content"))
|
||||
|
||||
|
||||
def image_part_url(part: JsonValue) -> str | None:
|
||||
"""The URL of a Chat Completions ``image_url`` part, in either its object or bare-string form."""
|
||||
if not isinstance(part, dict) or part.get("type") != "image_url":
|
||||
return None
|
||||
image_url: Final = part.get("image_url")
|
||||
if isinstance(image_url, str):
|
||||
return image_url
|
||||
url: Final = image_url.get("url") if isinstance(image_url, dict) else None
|
||||
return url if isinstance(url, str) else None
|
||||
|
||||
|
||||
def map_content_image_urls(content: JsonValue, transform: Callable[[str], str]) -> JsonValue:
|
||||
"""Return a copy of ``content`` with every ``image_url`` part's URL replaced by ``transform(url)``."""
|
||||
if not isinstance(content, list):
|
||||
return content
|
||||
return [_image_part_with_mapped_url(part, transform) for part in content] # mutable-ok: JSON content is an array
|
||||
|
||||
|
||||
def _image_part_with_mapped_url(part: JsonValue, transform: Callable[[str], str]) -> JsonValue:
|
||||
if not isinstance(part, dict) or part.get("type") != "image_url":
|
||||
return part
|
||||
image_url: Final = part.get("image_url")
|
||||
if isinstance(image_url, str):
|
||||
return {**part, "image_url": transform(image_url)} # mutable-ok: JSON parts are objects
|
||||
if not isinstance(image_url, dict):
|
||||
return part
|
||||
url: Final = image_url.get("url")
|
||||
if not isinstance(url, str):
|
||||
return part
|
||||
return {**part, "image_url": {**image_url, "url": transform(url)}} # mutable-ok: JSON parts are objects
|
||||
|
||||
|
||||
def map_messages_image_urls(messages: JsonValue, transform: Callable[[str], str]) -> JsonValue:
|
||||
"""Return a new message list with :func:`map_content_image_urls` applied to every ``content``."""
|
||||
if not isinstance(messages, list):
|
||||
return messages
|
||||
return [_message_with_mapped_image_urls(message, transform) for message in messages] # mutable-ok: JSON array
|
||||
|
||||
|
||||
def _message_with_mapped_image_urls(message: JsonValue, transform: Callable[[str], str]) -> JsonValue:
|
||||
if not isinstance(message, dict) or not isinstance(message.get("content"), list):
|
||||
return message
|
||||
return {**message, "content": map_content_image_urls(message["content"], transform)} # mutable-ok: JSON object
|
||||
|
||||
|
||||
def walk_user_text(data: dict[str, Any], visit: Callable[[str], str]) -> int:
|
||||
"""Rewrite every text fragment in place via ``visit``.
|
||||
|
||||
|
|
@ -307,3 +356,12 @@ def build_inspection_messages(data: dict[str, Any]) -> list[dict[str, str]]:
|
|||
role = message.get("role", "user") or "user"
|
||||
flattened.append({"role": role, "content": text})
|
||||
return flattened
|
||||
|
||||
|
||||
def as_json_value(value: object) -> JsonValue:
|
||||
"""Round-trips through the stdlib codec because pydantic's serializer turns anything nested
|
||||
past 254 levels into "...", while this keeps about the depth the proxy's request parser accepts"""
|
||||
parsed: Final[JsonValue] = json.loads( # pyright: ignore[reportAny] # untyped stdlib parse of json.dumps output
|
||||
json.dumps(value, default=to_jsonable_python)
|
||||
)
|
||||
return parsed
|
||||
|
|
|
|||
|
|
@ -39,6 +39,11 @@ def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"
|
|||
streaming_end_of_stream_only=_get_config_value(litellm_params, optional_params, "streaming_end_of_stream_only"),
|
||||
streaming_sampling_rate=_get_config_value(litellm_params, optional_params, "streaming_sampling_rate"),
|
||||
streaming_transform_mode=_get_config_value(litellm_params, optional_params, "streaming_transform_mode"),
|
||||
send_images=_get_config_value(litellm_params, optional_params, "send_images"),
|
||||
exclude_payload_fields=_get_config_value(litellm_params, optional_params, "exclude_payload_fields"),
|
||||
max_messages=_get_config_value(litellm_params, optional_params, "max_messages"),
|
||||
max_text_chars=_get_config_value(litellm_params, optional_params, "max_text_chars"),
|
||||
strip_patterns=_get_config_value(litellm_params, optional_params, "strip_patterns"),
|
||||
)
|
||||
|
||||
litellm.logging_callback_manager.add_litellm_callback(_generic_guardrail_api_callback)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,15 @@
|
|||
import re
|
||||
from collections.abc import Sequence
|
||||
|
||||
|
||||
def config_values(raw: Sequence[str] | None, *, option_name: str) -> tuple[str, ...]:
|
||||
if isinstance(raw, str):
|
||||
raise ValueError(f"{option_name} must be a list of strings, got the single string {raw!r}")
|
||||
return tuple(raw or ())
|
||||
|
||||
|
||||
def compile_patterns(raw: Sequence[str] | None, *, option_name: str) -> tuple[re.Pattern[str], ...]:
|
||||
try:
|
||||
return tuple(re.compile(pattern) for pattern in config_values(raw, option_name=option_name))
|
||||
except re.error as e:
|
||||
raise ValueError(f"{option_name} contains an invalid regex: {e}") from e
|
||||
|
|
@ -11,6 +11,8 @@ from collections.abc import Mapping, Sequence
|
|||
from typing import TYPE_CHECKING, Any, Final, Literal, Optional
|
||||
|
||||
import httpx
|
||||
from pydantic import JsonValue
|
||||
from typing_extensions import TypeIs
|
||||
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm._version import version as litellm_version
|
||||
|
|
@ -19,10 +21,13 @@ from litellm.integrations.custom_guardrail import (
|
|||
CustomGuardrail,
|
||||
log_guardrail_information,
|
||||
)
|
||||
from litellm.llms.base_llm.guardrail_translation.utils import UnappliableRequestRewrite
|
||||
from litellm.llms.custom_httpx.http_handler import (
|
||||
AsyncHTTPHandler,
|
||||
get_async_httpx_client,
|
||||
httpxSpecialProvider,
|
||||
)
|
||||
from litellm.proxy.guardrails._content_utils import as_json_value
|
||||
from litellm.types.guardrails import GuardrailEventHooks
|
||||
from litellm.types.llms.openai import AllMessageValues, ChatCompletionToolParam
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.generic_guardrail_api import (
|
||||
|
|
@ -33,6 +38,16 @@ from litellm.types.proxy.guardrails.guardrail_hooks.generic_guardrail_api import
|
|||
)
|
||||
from litellm.types.utils import GenericGuardrailAPIInputs
|
||||
|
||||
from .payload_policy import (
|
||||
PayloadLoss,
|
||||
accepted_rewrites,
|
||||
block_only_response,
|
||||
raise_if_intervention_was_refused,
|
||||
resolve_payload_policy,
|
||||
restore_unseen_rows,
|
||||
shape_payload,
|
||||
)
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.base import GuardrailConfigModel
|
||||
|
|
@ -150,6 +165,28 @@ def _extract_inbound_headers(
|
|||
return None
|
||||
|
||||
|
||||
def _is_part_list(value: object) -> TypeIs[list[object]]: # guard-ok: trivial isinstance narrowing
|
||||
return isinstance(value, list)
|
||||
|
||||
|
||||
def _row_as_sent(dumped: JsonValue, caller: Mapping[str, object]) -> JsonValue:
|
||||
"""The request model validates list content lazily and dumps a list holding
|
||||
any part it rejects as [], so such a row is sent with the caller's content."""
|
||||
caller_content: Final = caller.get("content")
|
||||
if not isinstance(dumped, dict) or not _is_part_list(caller_content):
|
||||
return dumped
|
||||
dumped_content: Final = dumped.get("content")
|
||||
if isinstance(dumped_content, list) and len(dumped_content) == len(caller_content):
|
||||
return dumped
|
||||
return {**dumped, "content": as_json_value(caller_content)}
|
||||
|
||||
|
||||
def _rows_as_sent(dumped_rows: JsonValue, caller_rows: Sequence[Mapping[str, object]] | None) -> JsonValue:
|
||||
if caller_rows is None or not isinstance(dumped_rows, list):
|
||||
return dumped_rows
|
||||
return [_row_as_sent(dumped, caller) for dumped, caller in zip(dumped_rows, caller_rows, strict=True)]
|
||||
|
||||
|
||||
def _structured_rows_to_write_back(
|
||||
original_rows: Sequence[AllMessageValues] | None,
|
||||
shown_rows: Sequence[AllMessageValues] | None,
|
||||
|
|
@ -204,9 +241,17 @@ class GenericGuardrailAPI(CustomGuardrail):
|
|||
streaming_end_of_stream_only: bool | None = None,
|
||||
streaming_sampling_rate: int | None = None,
|
||||
streaming_transform_mode: Literal["block_only", "incremental_diff"] | None = None,
|
||||
send_images: bool | None = None,
|
||||
exclude_payload_fields: Sequence[str] | None = None,
|
||||
max_messages: int | None = None,
|
||||
max_text_chars: int | None = None,
|
||||
strip_patterns: Sequence[str] | None = None,
|
||||
async_handler: AsyncHTTPHandler | None = None,
|
||||
**kwargs,
|
||||
):
|
||||
self.async_handler = get_async_httpx_client(llm_provider=httpxSpecialProvider.GuardrailCallback)
|
||||
self.async_handler = async_handler or get_async_httpx_client(
|
||||
llm_provider=httpxSpecialProvider.GuardrailCallback
|
||||
)
|
||||
self.headers = headers or {}
|
||||
self.extra_headers = extra_headers or []
|
||||
|
||||
|
|
@ -251,6 +296,15 @@ class GenericGuardrailAPI(CustomGuardrail):
|
|||
"block_only" if streaming_transform_mode is None else streaming_transform_mode
|
||||
)
|
||||
|
||||
self._payload_policy: Final = resolve_payload_policy(
|
||||
send_images=send_images,
|
||||
exclude_payload_fields=exclude_payload_fields,
|
||||
max_messages=max_messages,
|
||||
max_text_chars=max_text_chars,
|
||||
strip_patterns=strip_patterns,
|
||||
guardrail_name=kwargs.get("guardrail_name"),
|
||||
)
|
||||
|
||||
# Set supported event hooks
|
||||
kwargs.setdefault("supported_event_hooks", list(self.get_supported_event_hooks()))
|
||||
|
||||
|
|
@ -345,24 +399,42 @@ class GenericGuardrailAPI(CustomGuardrail):
|
|||
structured_messages: Sequence[AllMessageValues] | None,
|
||||
shown_messages: Sequence[AllMessageValues] | None,
|
||||
guardrail_response: GenericGuardrailAPIResponse,
|
||||
loss: PayloadLoss,
|
||||
) -> GenericGuardrailAPIInputs:
|
||||
# Action is NONE or no modifications needed
|
||||
return_inputs: Final = GenericGuardrailAPIInputs(texts=texts)
|
||||
if guardrail_response.texts:
|
||||
return_inputs["texts"] = guardrail_response.texts
|
||||
if guardrail_response.images:
|
||||
return_inputs["images"] = guardrail_response.images
|
||||
name: Final = self.guardrail_name
|
||||
accepted: Final = accepted_rewrites(guardrail_response, loss, guardrail_name=name)
|
||||
if accepted.texts:
|
||||
return_inputs["texts"] = accepted.texts
|
||||
if accepted.images:
|
||||
return_inputs["images"] = accepted.images
|
||||
elif images:
|
||||
return_inputs["images"] = images
|
||||
if guardrail_response.tools:
|
||||
return_inputs["tools"] = guardrail_response.tools
|
||||
if accepted.tools:
|
||||
return_inputs["tools"] = accepted.tools
|
||||
elif tools:
|
||||
return_inputs["tools"] = tools
|
||||
rows_to_write_back: Final = (
|
||||
_structured_rows_to_write_back(structured_messages, shown_messages, guardrail_response.structured_messages)
|
||||
if guardrail_response.structured_messages
|
||||
restore_unseen_rows(
|
||||
rows=_structured_rows_to_write_back(structured_messages, shown_messages, accepted.rows),
|
||||
caller=structured_messages,
|
||||
sent=shown_messages,
|
||||
loss=loss,
|
||||
guardrail_name=name,
|
||||
)
|
||||
if accepted.rows
|
||||
else None
|
||||
)
|
||||
raise_if_intervention_was_refused(
|
||||
action=guardrail_response.action,
|
||||
accepted=accepted,
|
||||
original_texts=texts,
|
||||
original_images=images,
|
||||
original_tools=tools,
|
||||
rows_written_back=rows_to_write_back is not None,
|
||||
guardrail_name=name,
|
||||
)
|
||||
if rows_to_write_back is not None:
|
||||
return_inputs["structured_messages"] = list(rows_to_write_back) # mutable-ok: guardrail inputs take a list
|
||||
if guardrail_response.stream_holdback_chars is not None:
|
||||
|
|
@ -470,12 +542,15 @@ class GenericGuardrailAPI(CustomGuardrail):
|
|||
)
|
||||
|
||||
headers: Final = self._build_request_headers()
|
||||
# The model's list content is a lazy iterator that this dump consumes, so it cannot be read again
|
||||
dumped: Final[Mapping[str, JsonValue]] = guardrail_request.model_dump(mode="json")
|
||||
sent_messages: Final = _rows_as_sent(dumped.get("structured_messages"), structured_messages)
|
||||
request_json: Final = {**dumped, "structured_messages": sent_messages} # mutable-ok: JSON POST body
|
||||
payload: Final = shape_payload(request_json, self._payload_policy, guardrail_name=self.guardrail_name)
|
||||
|
||||
# Make the API request
|
||||
# Use mode="json" to ensure all iterables are converted to lists
|
||||
response: Final = await self.async_handler.post(
|
||||
url=self.api_base,
|
||||
json=guardrail_request.model_dump(mode="json"),
|
||||
json=payload.body,
|
||||
headers=headers,
|
||||
)
|
||||
|
||||
|
|
@ -503,11 +578,14 @@ class GenericGuardrailAPI(CustomGuardrail):
|
|||
images=images,
|
||||
tools=tools,
|
||||
structured_messages=structured_messages,
|
||||
shown_messages=guardrail_request.structured_messages,
|
||||
guardrail_response=guardrail_response,
|
||||
shown_messages=payload.sent_messages,
|
||||
guardrail_response=block_only_response(
|
||||
guardrail_response, payload, input_type=input_type, guardrail_name=self.guardrail_name
|
||||
),
|
||||
loss=payload.loss,
|
||||
)
|
||||
|
||||
except GuardrailRaisedException:
|
||||
except (GuardrailRaisedException, UnappliableRequestRewrite):
|
||||
raise
|
||||
except Timeout as e:
|
||||
return self._handle_guardrail_request_error(e, inputs, input_type, logging_obj)
|
||||
|
|
|
|||
|
|
@ -0,0 +1,512 @@
|
|||
"""What the Generic Guardrail API endpoint is sent, and what it may write back.
|
||||
|
||||
Shaping is lossy, so every shaped payload carries a ``PayloadLoss``. Caller content the
|
||||
guardrail did not see in full is never replaced by the guardrail's response.
|
||||
"""
|
||||
|
||||
import time
|
||||
from collections.abc import Callable, Iterable, Mapping, Sequence
|
||||
from dataclasses import dataclass
|
||||
from functools import reduce
|
||||
from itertools import accumulate, chain
|
||||
from types import MappingProxyType
|
||||
from typing import Final, Literal, assert_never
|
||||
|
||||
import regex
|
||||
from pydantic import JsonValue
|
||||
|
||||
from litellm._logging import verbose_proxy_logger
|
||||
from litellm.exceptions import GuardrailRaisedException
|
||||
from litellm.llms.base_llm.guardrail_translation.utils import (
|
||||
message_slot_texts,
|
||||
message_with_slot_texts,
|
||||
unappliable_request_rewrite,
|
||||
)
|
||||
from litellm.proxy.guardrails._content_utils import as_json_value, image_part_url, map_messages_image_urls
|
||||
from litellm.types.llms.openai import AllMessageValues
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.generic_guardrail_api import (
|
||||
GenericGuardrailAPIRequest,
|
||||
GenericGuardrailAPIResponse,
|
||||
GuardrailToolParam,
|
||||
structured_messages_from_json,
|
||||
)
|
||||
|
||||
from .config_parsing import config_values
|
||||
|
||||
PROTECTED_PAYLOAD_FIELDS: Final = frozenset({"input_type", "litellm_call_id"})
|
||||
|
||||
IMAGE_OMITTED_PLACEHOLDER: Final = "[omitted]"
|
||||
|
||||
MAX_STRIP_SUBSTITUTIONS: Final = 64
|
||||
|
||||
MAX_STRIP_CALL_CHARS: Final = 100_000
|
||||
|
||||
STRIP_TIMEOUT_SECONDS: Final = 0.1
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class PayloadPolicy:
|
||||
send_images: bool = True
|
||||
exclude_fields: frozenset[str] = frozenset()
|
||||
max_messages: int | None = None
|
||||
max_text_chars: int | None = None
|
||||
strip_patterns: tuple[regex.Pattern[str], ...] = ()
|
||||
|
||||
@property
|
||||
def omitted_fields(self) -> frozenset[str]:
|
||||
return self.exclude_fields if self.send_images else self.exclude_fields | frozenset(("images",))
|
||||
|
||||
@property
|
||||
def shapes_messages(self) -> bool:
|
||||
return not self.send_images
|
||||
|
||||
@property
|
||||
def shapes_text(self) -> bool:
|
||||
return self.max_text_chars is not None or bool(self.strip_patterns)
|
||||
|
||||
@property
|
||||
def lossy_options(self) -> tuple[str, ...]:
|
||||
options: Final = (
|
||||
("send_images=False", not self.send_images),
|
||||
(f"exclude_payload_fields={sorted(self.exclude_fields)}", bool(self.exclude_fields)),
|
||||
(f"max_messages={self.max_messages}", self.max_messages is not None),
|
||||
(f"max_text_chars={self.max_text_chars}", self.max_text_chars is not None),
|
||||
("strip_patterns", bool(self.strip_patterns)),
|
||||
)
|
||||
return tuple(option for option, is_set in options if is_set)
|
||||
|
||||
@property
|
||||
def is_lossy(self) -> bool:
|
||||
return bool(self.lossy_options)
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class PayloadLoss:
|
||||
altered_message_indices: frozenset[int] = frozenset()
|
||||
texts_omitted: bool = False
|
||||
messages_omitted: bool = False
|
||||
images_omitted: bool = False
|
||||
tools_omitted: bool = False
|
||||
text_shaped: bool = False
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class ShapedPayload:
|
||||
body: dict[str, JsonValue] # mutable-ok: the HTTP client takes the POST body as a dict
|
||||
sent_messages: Sequence[AllMessageValues] | None
|
||||
loss: PayloadLoss
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class AcceptedRewrites:
|
||||
texts: list[str] | None # mutable-ok: guardrail inputs take a list
|
||||
images: list[str] | None # mutable-ok: guardrail inputs take a list
|
||||
tools: list[GuardrailToolParam] | None # mutable-ok: guardrail inputs take a list
|
||||
rows: Sequence[AllMessageValues] | None
|
||||
refused_any: bool
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class _Unappliable:
|
||||
pass
|
||||
|
||||
|
||||
_UNAPPLIABLE: Final = _Unappliable()
|
||||
|
||||
|
||||
def resolve_payload_policy(
|
||||
*,
|
||||
send_images: object,
|
||||
exclude_payload_fields: Sequence[str] | None,
|
||||
max_messages: object,
|
||||
max_text_chars: object,
|
||||
strip_patterns: Sequence[str] | None,
|
||||
guardrail_name: str | None,
|
||||
) -> PayloadPolicy:
|
||||
policy: Final = PayloadPolicy(
|
||||
send_images=_send_images(send_images),
|
||||
exclude_fields=_resolve_exclude_fields(
|
||||
config_values(exclude_payload_fields, option_name="exclude_payload_fields"), guardrail_name=guardrail_name
|
||||
),
|
||||
max_messages=_positive_int(max_messages, option_name="max_messages"),
|
||||
max_text_chars=_positive_int(max_text_chars, option_name="max_text_chars"),
|
||||
strip_patterns=_compile_strip_patterns(strip_patterns),
|
||||
)
|
||||
if policy.is_lossy:
|
||||
verbose_proxy_logger.warning(
|
||||
"Generic Guardrail API (%s): %s keep part of the request from the guardrail, so it can only enforce "
|
||||
"on what it is sent, and it cannot rewrite what it did not see.",
|
||||
guardrail_name,
|
||||
", ".join(policy.lossy_options),
|
||||
)
|
||||
return policy
|
||||
|
||||
|
||||
def _positive_int(value: object, *, option_name: str) -> int | None:
|
||||
match value:
|
||||
case None:
|
||||
return None
|
||||
case bool():
|
||||
raise ValueError(f"{option_name} must be an int, got {value!r}")
|
||||
case int() if value >= 1:
|
||||
return value
|
||||
case int():
|
||||
raise ValueError(f"{option_name} must be >= 1 (got {value})")
|
||||
case _:
|
||||
raise ValueError(f"{option_name} must be an int, got {value!r}")
|
||||
|
||||
|
||||
def _compile_strip_patterns(raw: Sequence[str] | None) -> tuple[regex.Pattern[str], ...]:
|
||||
try:
|
||||
return tuple(regex.compile(pattern) for pattern in config_values(raw, option_name="strip_patterns"))
|
||||
except regex.error as e:
|
||||
raise ValueError(f"strip_patterns contains an invalid regex: {e}") from e
|
||||
|
||||
|
||||
def _send_images(value: object) -> bool:
|
||||
match value:
|
||||
case None:
|
||||
return True
|
||||
case bool():
|
||||
return value
|
||||
case _:
|
||||
raise ValueError(f"send_images must be a bool, got {value!r}")
|
||||
|
||||
|
||||
def _resolve_exclude_fields(raw: Sequence[str], *, guardrail_name: str | None) -> frozenset[str]:
|
||||
known: Final = frozenset(GenericGuardrailAPIRequest.model_fields)
|
||||
unknown: Final = tuple(field for field in raw if field not in known)
|
||||
if unknown:
|
||||
verbose_proxy_logger.warning(
|
||||
"Generic Guardrail API (%s): ignoring unknown exclude_payload_fields %s. Known fields: %s",
|
||||
guardrail_name,
|
||||
unknown,
|
||||
sorted(known),
|
||||
)
|
||||
protected: Final = tuple(field for field in raw if field in PROTECTED_PAYLOAD_FIELDS)
|
||||
if protected:
|
||||
verbose_proxy_logger.warning(
|
||||
"Generic Guardrail API (%s): exclude_payload_fields cannot drop %s, the guardrail needs them to "
|
||||
"interpret the payload; they are still sent.",
|
||||
guardrail_name,
|
||||
protected,
|
||||
)
|
||||
excludable: Final = known - PROTECTED_PAYLOAD_FIELDS
|
||||
return frozenset(field for field in raw if field in excludable)
|
||||
|
||||
|
||||
def _omit_image(_url: str) -> str:
|
||||
return IMAGE_OMITTED_PLACEHOLDER
|
||||
|
||||
|
||||
def _content(row: JsonValue) -> JsonValue:
|
||||
return row.get("content") if isinstance(row, dict) else None
|
||||
|
||||
|
||||
def _holds_placeholder(part: JsonValue) -> bool:
|
||||
return image_part_url(part) == IMAGE_OMITTED_PLACEHOLDER
|
||||
|
||||
|
||||
def _altered_message_indices(unshaped: JsonValue, sent: JsonValue) -> frozenset[int]:
|
||||
if not isinstance(unshaped, list) or not isinstance(sent, list):
|
||||
return frozenset()
|
||||
return frozenset(index for index, (row, sent_row) in enumerate(zip(unshaped, sent, strict=True)) if row != sent_row)
|
||||
|
||||
|
||||
def _string_list(value: JsonValue) -> tuple[str, ...]:
|
||||
return tuple(item for item in value if isinstance(item, str)) if isinstance(value, list) else ()
|
||||
|
||||
|
||||
def _windowed_messages(messages: JsonValue, max_messages: int | None) -> JsonValue:
|
||||
if max_messages is None or not isinstance(messages, list) or len(messages) <= max_messages:
|
||||
return messages
|
||||
return messages[-max_messages:]
|
||||
|
||||
|
||||
def _strip(text: str, patterns: tuple[regex.Pattern[str], ...], deadline: float) -> str | None:
|
||||
def strip_one(acc: str | None, pattern: regex.Pattern[str]) -> str | None:
|
||||
remaining: Final = deadline - time.monotonic()
|
||||
if acc is None or remaining <= 0:
|
||||
return None
|
||||
try:
|
||||
return pattern.sub("", acc, count=MAX_STRIP_SUBSTITUTIONS, timeout=remaining)
|
||||
except TimeoutError:
|
||||
return None
|
||||
|
||||
return reduce(strip_one, patterns, text)
|
||||
|
||||
|
||||
def _stripped_fragments(
|
||||
fragments: Iterable[str], policy: PayloadPolicy, guardrail_name: str | None
|
||||
) -> Mapping[str, str]:
|
||||
if not policy.strip_patterns:
|
||||
return MappingProxyType({})
|
||||
distinct: Final = tuple(dict.fromkeys(fragments))
|
||||
spent: Final = tuple(accumulate(len(fragment) for fragment in distinct))
|
||||
within_budget: Final = tuple(fragment for fragment, total in zip(distinct, spent) if total <= MAX_STRIP_CALL_CHARS)
|
||||
deadline: Final = time.monotonic() + STRIP_TIMEOUT_SECONDS
|
||||
stripped: Final = tuple((fragment, _strip(fragment, policy.strip_patterns, deadline)) for fragment in within_budget)
|
||||
unstripped: Final = len(distinct) - sum(1 for _, text in stripped if text is not None)
|
||||
if unstripped:
|
||||
verbose_proxy_logger.warning(
|
||||
"Generic Guardrail API (%s): %d text(s) are sent unstripped, because strip_patterns only run on the "
|
||||
"first %d characters of distinct text per guardrail call and for at most %s seconds.",
|
||||
guardrail_name,
|
||||
unstripped,
|
||||
MAX_STRIP_CALL_CHARS,
|
||||
STRIP_TIMEOUT_SECONDS,
|
||||
)
|
||||
return MappingProxyType({fragment: text for fragment, text in stripped if text is not None})
|
||||
|
||||
|
||||
def _text_shaper(stripped: Mapping[str, str], max_text_chars: int | None) -> Callable[[str], str]:
|
||||
def shape(text: str) -> str:
|
||||
kept: Final = stripped.get(text, text)
|
||||
return kept if max_text_chars is None else kept[:max_text_chars]
|
||||
|
||||
return shape
|
||||
|
||||
|
||||
def _row_with_shaped_text(row: AllMessageValues, shape: Callable[[str], str]) -> AllMessageValues:
|
||||
return message_with_slot_texts(row, tuple(shape(text) for text in message_slot_texts(row))) or row
|
||||
|
||||
|
||||
def shape_payload(
|
||||
dumped: Mapping[str, JsonValue], policy: PayloadPolicy, *, guardrail_name: str | None
|
||||
) -> ShapedPayload:
|
||||
omitted: Final = policy.omitted_fields
|
||||
dumped_messages: Final = dumped.get("structured_messages")
|
||||
retained_messages: Final = _windowed_messages(dumped_messages, policy.max_messages)
|
||||
rows_windowed: Final = retained_messages is not dumped_messages
|
||||
retained_rows: Final = structured_messages_from_json(retained_messages) or ()
|
||||
dumped_texts: Final = dumped.get("texts")
|
||||
texts: Final = _string_list(dumped_texts)
|
||||
windowed_texts: Final = (
|
||||
tuple(chain.from_iterable(message_slot_texts(row) for row in retained_rows)) if rows_windowed else texts
|
||||
)
|
||||
shape: Final = _text_shaper(
|
||||
_stripped_fragments(
|
||||
chain(windowed_texts, chain.from_iterable(message_slot_texts(row) for row in retained_rows)),
|
||||
policy,
|
||||
guardrail_name,
|
||||
),
|
||||
policy.max_text_chars,
|
||||
)
|
||||
sent_texts: Final = tuple(shape(text) for text in windowed_texts) if policy.shapes_text else windowed_texts
|
||||
texted_messages: Final = (
|
||||
as_json_value([_row_with_shaped_text(row, shape) for row in retained_rows])
|
||||
if policy.shapes_text and retained_rows
|
||||
else retained_messages
|
||||
)
|
||||
sent_messages: Final = (
|
||||
map_messages_image_urls(texted_messages, _omit_image) if policy.shapes_messages else texted_messages
|
||||
)
|
||||
shaped: Final = MappingProxyType(
|
||||
{
|
||||
**dumped,
|
||||
"structured_messages": sent_messages,
|
||||
"texts": None if dumped_texts is None else list(sent_texts), # mutable-ok: JSON texts is an array
|
||||
}
|
||||
)
|
||||
return ShapedPayload(
|
||||
body={key: value for key, value in shaped.items() if key not in omitted}, # mutable-ok: JSON POST body
|
||||
sent_messages=structured_messages_from_json(sent_messages),
|
||||
loss=PayloadLoss(
|
||||
altered_message_indices=(
|
||||
frozenset() if rows_windowed else _altered_message_indices(dumped_messages, sent_messages)
|
||||
),
|
||||
texts_omitted="texts" in omitted,
|
||||
messages_omitted="structured_messages" in omitted,
|
||||
images_omitted="images" in omitted,
|
||||
tools_omitted="tools" in omitted,
|
||||
text_shaped=rows_windowed or sent_texts != texts or texted_messages != retained_messages,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _without_nulls(value: JsonValue) -> JsonValue:
|
||||
if isinstance(value, dict):
|
||||
return {key: _without_nulls(item) for key, item in value.items() if item is not None} # mutable-ok: JSON
|
||||
if isinstance(value, list):
|
||||
return [_without_nulls(item) for item in value] # mutable-ok: JSON array
|
||||
return value
|
||||
|
||||
|
||||
def _rewrites(response: GenericGuardrailAPIResponse, body: Mapping[str, JsonValue]) -> bool:
|
||||
returned: Final = (
|
||||
("texts", response.texts),
|
||||
("structured_messages", response.structured_messages),
|
||||
("images", response.images),
|
||||
("tools", response.tools),
|
||||
)
|
||||
return response.action == "GUARDRAIL_INTERVENED" or any(
|
||||
value and _without_nulls(as_json_value(value)) != _without_nulls(body.get(field)) for field, value in returned
|
||||
)
|
||||
|
||||
|
||||
def block_only_response(
|
||||
response: GenericGuardrailAPIResponse,
|
||||
payload: ShapedPayload,
|
||||
*,
|
||||
input_type: Literal["request", "response"],
|
||||
guardrail_name: str | None,
|
||||
) -> GenericGuardrailAPIResponse:
|
||||
if not payload.loss.text_shaped:
|
||||
return response
|
||||
if not _rewrites(response, payload.body):
|
||||
return GenericGuardrailAPIResponse(action=response.action, stream_holdback_chars=response.stream_holdback_chars)
|
||||
verbose_proxy_logger.warning(
|
||||
"Generic Guardrail API (%s): the guardrail rewrote a %s that max_messages, max_text_chars or strip_patterns "
|
||||
"shaped before it was sent. A rewrite of content it saw only in part cannot be applied, so the %s is "
|
||||
"rejected. These options are for block-only guardrails.",
|
||||
guardrail_name,
|
||||
input_type,
|
||||
input_type,
|
||||
)
|
||||
match input_type:
|
||||
case "request":
|
||||
raise unappliable_request_rewrite(guardrail_name)
|
||||
case "response":
|
||||
raise GuardrailRaisedException(
|
||||
guardrail_name=guardrail_name,
|
||||
message=f"Guardrail '{guardrail_name}' returned a rewrite that cannot be applied to this response",
|
||||
should_wrap_with_default_message=False,
|
||||
)
|
||||
case _:
|
||||
assert_never(input_type)
|
||||
|
||||
|
||||
def _log_refused(field: str, detail: str, guardrail_name: str | None) -> None:
|
||||
verbose_proxy_logger.warning(
|
||||
"Generic Guardrail API (%s): ignoring the returned %s, %s.", guardrail_name, field, detail
|
||||
)
|
||||
|
||||
|
||||
def _refused(returned: object, *, field: str, omitted: bool, guardrail_name: str | None) -> bool:
|
||||
if not returned or not omitted:
|
||||
return False
|
||||
_log_refused(field, "it was not sent to the guardrail", guardrail_name)
|
||||
return True
|
||||
|
||||
|
||||
def accepted_rewrites(
|
||||
response: GenericGuardrailAPIResponse, loss: PayloadLoss, *, guardrail_name: str | None
|
||||
) -> AcceptedRewrites:
|
||||
refused: Final = MappingProxyType(
|
||||
{
|
||||
field: _refused(returned, field=field, omitted=omitted, guardrail_name=guardrail_name)
|
||||
for field, returned, omitted in (
|
||||
("texts", response.texts, loss.texts_omitted),
|
||||
("images", response.images, loss.images_omitted),
|
||||
("tools", response.tools, loss.tools_omitted),
|
||||
("structured_messages", response.structured_messages, loss.messages_omitted),
|
||||
)
|
||||
}
|
||||
)
|
||||
return AcceptedRewrites(
|
||||
texts=None if refused["texts"] else response.texts or None,
|
||||
images=None if refused["images"] else response.images or None,
|
||||
tools=None if refused["tools"] else response.tools or None,
|
||||
rows=None if refused["structured_messages"] else response.structured_messages or None,
|
||||
refused_any=any(refused.values()),
|
||||
)
|
||||
|
||||
|
||||
def _changes(accepted: object, original: object) -> bool:
|
||||
return accepted is not None and as_json_value(accepted) != as_json_value(original)
|
||||
|
||||
|
||||
def raise_if_intervention_was_refused(
|
||||
*,
|
||||
action: str,
|
||||
accepted: AcceptedRewrites,
|
||||
original_texts: Sequence[str],
|
||||
original_images: Sequence[str] | None,
|
||||
original_tools: Sequence[Mapping[str, object]] | None,
|
||||
rows_written_back: bool,
|
||||
guardrail_name: str | None,
|
||||
) -> None:
|
||||
"""A guardrail whose only real rewrite went to a field it was not sent would otherwise let the request
|
||||
through unchanged, so it is rejected instead. An echo of what the caller sent changes nothing."""
|
||||
applied: Final = rows_written_back or any(
|
||||
_changes(accepted_value, original_value)
|
||||
for accepted_value, original_value in (
|
||||
(accepted.texts, original_texts),
|
||||
(accepted.images, original_images),
|
||||
(accepted.tools, original_tools),
|
||||
)
|
||||
)
|
||||
if action == "GUARDRAIL_INTERVENED" and accepted.refused_any and not applied:
|
||||
raise unappliable_request_rewrite(guardrail_name)
|
||||
|
||||
|
||||
def _omitted_image_count(content: JsonValue) -> int:
|
||||
if not isinstance(content, list):
|
||||
return 0
|
||||
return sum(1 for part in content if _holds_placeholder(part))
|
||||
|
||||
|
||||
def _restored_part(returned: JsonValue, caller: JsonValue, sent: JsonValue) -> JsonValue | _Unappliable:
|
||||
if returned == sent:
|
||||
return caller
|
||||
return _UNAPPLIABLE if _holds_placeholder(sent) or _holds_placeholder(returned) else returned
|
||||
|
||||
|
||||
def _restored_content(returned: JsonValue, caller: JsonValue, sent: JsonValue) -> JsonValue | _Unappliable:
|
||||
if returned == sent:
|
||||
return caller
|
||||
if not isinstance(returned, list) or not isinstance(caller, list) or not isinstance(sent, list):
|
||||
return _UNAPPLIABLE
|
||||
if not len(returned) == len(caller) == len(sent):
|
||||
return _UNAPPLIABLE
|
||||
parts: Final = tuple(_restored_part(*aligned) for aligned in zip(returned, caller, sent))
|
||||
restored: Final = [part for part in parts if not isinstance(part, _Unappliable)] # mutable-ok: JSON array
|
||||
return restored if len(restored) == len(parts) else _UNAPPLIABLE
|
||||
|
||||
|
||||
def _restored_row(
|
||||
returned: AllMessageValues,
|
||||
caller: AllMessageValues,
|
||||
sent: AllMessageValues,
|
||||
*,
|
||||
altered: bool,
|
||||
) -> AllMessageValues | JsonValue | _Unappliable:
|
||||
if returned is caller:
|
||||
return caller
|
||||
returned_json: Final = as_json_value(returned)
|
||||
caller_content: Final = _content(as_json_value(caller))
|
||||
if not altered:
|
||||
adds_placeholder: Final = _omitted_image_count(_content(returned_json)) > _omitted_image_count(caller_content)
|
||||
return _UNAPPLIABLE if adds_placeholder else returned
|
||||
content: Final = _restored_content(_content(returned_json), caller_content, _content(as_json_value(sent)))
|
||||
if isinstance(content, _Unappliable) or not isinstance(returned_json, dict):
|
||||
return _UNAPPLIABLE
|
||||
return {**returned_json, "content": content} # mutable-ok: JSON row
|
||||
|
||||
|
||||
def restore_unseen_rows(
|
||||
*,
|
||||
rows: tuple[AllMessageValues, ...] | None,
|
||||
caller: Sequence[AllMessageValues] | None,
|
||||
sent: Sequence[AllMessageValues] | None,
|
||||
loss: PayloadLoss,
|
||||
guardrail_name: str | None,
|
||||
) -> tuple[AllMessageValues, ...] | None:
|
||||
"""At a row the guardrail saw only in part, accept its rewrite only where the parts line up, and put the
|
||||
caller's image back into each part that still holds the placeholder. Anything else blocks the request,
|
||||
since keeping the caller's row would silently drop the guardrail's rewrite."""
|
||||
altered: Final = loss.altered_message_indices
|
||||
if rows is None or not altered:
|
||||
return rows
|
||||
if caller is None or sent is None or not len(rows) == len(caller) == len(sent):
|
||||
raise unappliable_request_rewrite(guardrail_name)
|
||||
restored: Final = tuple(
|
||||
_restored_row(row, caller_row, sent_row, altered=index in altered)
|
||||
for index, (row, caller_row, sent_row) in enumerate(zip(rows, caller, sent))
|
||||
)
|
||||
accepted: Final = structured_messages_from_json(
|
||||
[row for row in restored if not isinstance(row, _Unappliable)] # mutable-ok: JSON array
|
||||
)
|
||||
if accepted is None or len(accepted) != len(restored):
|
||||
raise unappliable_request_rewrite(guardrail_name)
|
||||
return tuple(accepted)
|
||||
|
|
@ -103,6 +103,77 @@ class GenericGuardrailAPIOptionalParams(BaseModel):
|
|||
),
|
||||
)
|
||||
|
||||
send_images: bool | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"If False, the top-level images field is not sent, and every inline image_url part in "
|
||||
"structured_messages keeps its place but has its URL replaced by '[omitted]'. File, "
|
||||
"audio and video parts are still sent as they are. Saves payload size for guardrails "
|
||||
"that only inspect text. The guardrail cannot replace the caller's images: a rewritten "
|
||||
"message gets the caller's image back in each part still holding '[omitted]', and a "
|
||||
"rewrite that changes or moves an image part is rejected. Defaults to True in "
|
||||
"GenericGuardrailAPI.__init__ when None."
|
||||
),
|
||||
)
|
||||
|
||||
exclude_payload_fields: tuple[str, ...] | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Top-level guardrail request fields to leave out of the payload, e.g. "
|
||||
"['request_headers', 'tools'], for guardrails that do not use them. Unknown fields "
|
||||
"are ignored with a warning at init, and input_type and litellm_call_id are always "
|
||||
"sent. A field that is not sent (texts, structured_messages, images or tools) cannot "
|
||||
"be rewritten by the guardrail response."
|
||||
),
|
||||
)
|
||||
|
||||
max_messages: int | None = Field(
|
||||
default=None,
|
||||
ge=1,
|
||||
description=(
|
||||
"If set and a request has more than N structured_messages, only the last N are sent, "
|
||||
"and texts is rebuilt from the text of those N messages. Calls without "
|
||||
"structured_messages, such as embeddings, rerank or an LLM response, are not affected. "
|
||||
"images and tool_calls are not windowed. Bounds payload size when the whole conversation "
|
||||
"is re-sent every turn, but the system prompt and early turns fall out of the window. "
|
||||
"For block-only or observe-only guardrails: on a windowed call BLOCKED still applies, "
|
||||
"but any rewrite the guardrail returns fails the call. A failed request or response is "
|
||||
"rejected with an error, and a failed stream is cut off after the chunks already sent."
|
||||
),
|
||||
)
|
||||
|
||||
max_text_chars: int | None = Field(
|
||||
default=None,
|
||||
ge=1,
|
||||
description=(
|
||||
"If set, every text in texts and in structured_messages content is cut to this many "
|
||||
"characters before sending, so a caller can put content the guardrail never sees after "
|
||||
"the first N characters. For block-only or observe-only guardrails: when any text was "
|
||||
"cut, BLOCKED still applies, but any rewrite the guardrail returns fails the call, with "
|
||||
"the same errors as max_messages."
|
||||
),
|
||||
)
|
||||
|
||||
strip_patterns: tuple[str, ...] | None = Field(
|
||||
default=None,
|
||||
description=(
|
||||
"Regexes whose matches are removed from every text in texts and in structured_messages "
|
||||
"content before sending, e.g. volatile boilerplate the guardrail does not need. Roles, "
|
||||
"ids, tool calls, tools and metadata are never touched. A caller can hide content from "
|
||||
"the guardrail by wrapping it in something a pattern matches. For block-only or "
|
||||
"observe-only guardrails: when any text was stripped, BLOCKED still applies, but any "
|
||||
"rewrite the guardrail returns fails the call, with the same errors as max_messages. "
|
||||
"An invalid regex raises at init. Patterns use the regex package and run against "
|
||||
"caller requests and LLM responses alike. Each pattern removes at most 64 matches per "
|
||||
"text. Per guardrail call, only the first 100,000 characters of distinct text are "
|
||||
"stripped and stripping stops after 0.1 seconds. A text past either limit is sent "
|
||||
"unstripped in full with a warning. Stripping runs on the worker's event loop, so a slow "
|
||||
"pattern blocks that worker, and every request on it, for up to 0.1 seconds per "
|
||||
"guardrail call. Keep patterns linear-time: no nested quantifiers such as (a+)+ and no "
|
||||
"lazy match up to a closing delimiter such as <!--.*?-->."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
class GenericGuardrailAPIConfigModel(
|
||||
GuardrailConfigModel[GenericGuardrailAPIOptionalParams],
|
||||
|
|
@ -159,7 +230,7 @@ def coerce_stream_holdback_value(value: Any) -> int:
|
|||
return 0
|
||||
|
||||
|
||||
def structured_messages_from_response(value: object) -> Sequence[AllMessageValues] | None:
|
||||
def structured_messages_from_json(value: object) -> Sequence[AllMessageValues] | None:
|
||||
if not isinstance(value, list):
|
||||
return None
|
||||
if not all(isinstance(message, Mapping) and isinstance(message.get("role"), str) for message in value):
|
||||
|
|
@ -212,5 +283,5 @@ class GenericGuardrailAPIResponse:
|
|||
images=data.get("images"),
|
||||
tools=data.get("tools"),
|
||||
stream_holdback_chars=stream_holdback_chars,
|
||||
structured_messages=structured_messages_from_response(data.get("structured_messages")),
|
||||
structured_messages=structured_messages_from_json(data.get("structured_messages")),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ dependencies = [
|
|||
"pydantic-settings>=2.14.1,<3.0",
|
||||
"jsonschema>=4.0.0,<5.0",
|
||||
"boto3>=1.43.1,<2.0",
|
||||
"regex>=2022.1.18",
|
||||
]
|
||||
|
||||
[project.urls]
|
||||
|
|
|
|||
|
|
@ -178,3 +178,4 @@ hypothesis: >=6.165.10 # MPL 2.0 license
|
|||
pytest-rerunfailures: >=15.1 # MPL 2.0 license
|
||||
pytest-recording: >=0.13.4 # MIT license
|
||||
expression: >=5.6.0 # MIT License - https://github.com/cognitedata/Expression/blob/main/LICENSE
|
||||
regex: >=2022.1.18 # Apache-2.0 AND CNRI-Python, both permissive and OSI approved - https://github.com/mrabarnett/mrab-regex/blob/hg/LICENSE.txt
|
||||
|
|
|
|||
|
|
@ -5,16 +5,23 @@ This test file tests the Generic Guardrail API implementation,
|
|||
specifically focusing on metadata extraction and passing.
|
||||
"""
|
||||
|
||||
import json
|
||||
import os
|
||||
from collections.abc import Callable, Mapping
|
||||
from typing import Final
|
||||
from unittest.mock import AsyncMock, MagicMock, patch
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from pydantic import JsonValue
|
||||
|
||||
import litellm
|
||||
from litellm import ModelResponse
|
||||
from litellm._version import version as litellm_version
|
||||
from litellm.exceptions import GuardrailRaisedException, Timeout
|
||||
from litellm.llms.anthropic.chat.guardrail_translation.handler import AnthropicMessagesHandler
|
||||
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
|
||||
from litellm.llms.openai.chat.guardrail_translation.handler import OpenAIChatCompletionsHandler
|
||||
from litellm.proxy._types import UserAPIKeyAuth
|
||||
from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api import (
|
||||
GenericGuardrailAPI,
|
||||
|
|
@ -22,6 +29,8 @@ from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api import (
|
|||
from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.generic_guardrail_api import (
|
||||
_HEADER_PRESENT_PLACEHOLDER,
|
||||
)
|
||||
from litellm.types.llms.anthropic import AllAnthropicMessageValues
|
||||
from litellm.types.llms.openai import AllMessageValues, ChatCompletionImageObject
|
||||
from litellm.types.utils import Choices, Message
|
||||
|
||||
|
||||
|
|
@ -721,6 +730,287 @@ class TestStructuredMessagesInResponse:
|
|||
assert guardrailed_inputs["texts"] == ["[REDACTED]"]
|
||||
|
||||
|
||||
_SSN: Final = "123-45-6789"
|
||||
|
||||
|
||||
def _image_part() -> ChatCompletionImageObject:
|
||||
return {"type": "image_url", "image_url": {"url": "data:image/png;base64,iVBORw0KGgo="}}
|
||||
|
||||
|
||||
GuardrailAnswer = Callable[[Mapping[str, JsonValue]], Mapping[str, JsonValue]]
|
||||
|
||||
|
||||
def _guardrail_answering(answer: GuardrailAnswer) -> GenericGuardrailAPI:
|
||||
def serve(request: httpx.Request) -> httpx.Response:
|
||||
return httpx.Response(200, json=answer(json.loads(request.content)))
|
||||
|
||||
return GenericGuardrailAPI(
|
||||
api_base="https://guardrail.test/beta/litellm_basic_guardrail_api",
|
||||
guardrail_name="pii-masker",
|
||||
event_hook="pre_call",
|
||||
default_on=True,
|
||||
async_handler=AsyncHTTPHandler(transport=httpx.MockTransport(serve)),
|
||||
)
|
||||
|
||||
|
||||
def _masked(text: str) -> str:
|
||||
return text.replace(_SSN, "[SSN]")
|
||||
|
||||
|
||||
def _echo_every_row_and_mask_texts(request_json: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:
|
||||
return {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"structured_messages": request_json["structured_messages"],
|
||||
"texts": [_masked(text) for text in request_json["texts"]],
|
||||
}
|
||||
|
||||
|
||||
def _echo_first_row_and_mask_the_rest(request_json: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:
|
||||
first_row, *other_rows = request_json["structured_messages"]
|
||||
return {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"structured_messages": [first_row, *({**row, "content": _masked(row["content"])} for row in other_rows)],
|
||||
}
|
||||
|
||||
|
||||
def _masked_part(part: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:
|
||||
return {**part, "text": _masked(part["text"])} if part["type"] == "text" else part
|
||||
|
||||
|
||||
def _masked_row(row: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:
|
||||
content: Final = row["content"]
|
||||
if isinstance(content, str):
|
||||
return {**row, "content": _masked(content)}
|
||||
return {**row, "content": [_masked_part(part) for part in content]}
|
||||
|
||||
|
||||
def _mask_every_row_and_text(request_json: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:
|
||||
return {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"texts": [_masked(text) for text in request_json["texts"]],
|
||||
"structured_messages": [_masked_row(row) for row in request_json["structured_messages"]],
|
||||
}
|
||||
|
||||
|
||||
def _guarded_text_part() -> Mapping[str, JsonValue]:
|
||||
return {"type": "guarded_text", "text": "keep this guarded"}
|
||||
|
||||
|
||||
def _pdf_document_part() -> Mapping[str, JsonValue]:
|
||||
return {"type": "document", "source": {"type": "base64", "media_type": "application/pdf", "data": "JVBERi0="}}
|
||||
|
||||
|
||||
def _nested(depth: int) -> JsonValue:
|
||||
return {"leaf": "x"} if depth == 0 else {"nested": _nested(depth - 1)}
|
||||
|
||||
|
||||
async def _llm_bound_messages(guardrail: GenericGuardrailAPI, messages: list[AllMessageValues]) -> object:
|
||||
data: Final = await OpenAIChatCompletionsHandler().process_input_messages(
|
||||
data={"model": "gpt-5.6", "messages": messages}, guardrail_to_apply=guardrail
|
||||
)
|
||||
return data["messages"]
|
||||
|
||||
|
||||
async def _structured_messages_posted_for(rows: list[AllMessageValues]) -> list[JsonValue]:
|
||||
posted: Final[list[JsonValue]] = [] # mutable-ok: records what the endpoint received
|
||||
|
||||
def record(request_json: Mapping[str, JsonValue]) -> Mapping[str, JsonValue]:
|
||||
posted.append(request_json["structured_messages"])
|
||||
return {"action": "NONE"}
|
||||
|
||||
await _guardrail_answering(record).apply_guardrail(
|
||||
inputs={"texts": ["hi"], "structured_messages": rows}, request_data={}, input_type="request"
|
||||
)
|
||||
return posted
|
||||
|
||||
|
||||
async def _llm_bound_anthropic_messages(
|
||||
guardrail: GenericGuardrailAPI, messages: list[AllAnthropicMessageValues]
|
||||
) -> object:
|
||||
data: Final = await AnthropicMessagesHandler().process_input_messages(
|
||||
data={"model": "claude-opus-5-5", "max_tokens": 64, "messages": messages}, guardrail_to_apply=guardrail
|
||||
)
|
||||
return data["messages"]
|
||||
|
||||
|
||||
class TestEchoedRowsReachingTheLLM:
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("messages", "expected"),
|
||||
[
|
||||
(
|
||||
[{"role": "user", "content": [{"type": "text", "text": f"my ssn is {_SSN}"}, _image_part()]}],
|
||||
[{"role": "user", "content": [{"type": "text", "text": "my ssn is [SSN]"}, _image_part()]}],
|
||||
),
|
||||
(
|
||||
[
|
||||
{"role": "system", "content": "You are helpful."},
|
||||
{"role": "user", "content": [{"type": "text", "text": f"ssn {_SSN}"}, _image_part()]},
|
||||
{"role": "user", "content": f"again {_SSN}"},
|
||||
],
|
||||
[
|
||||
{"role": "system", "content": "You are helpful."},
|
||||
{"role": "user", "content": [{"type": "text", "text": "ssn [SSN]"}, _image_part()]},
|
||||
{"role": "user", "content": "again [SSN]"},
|
||||
],
|
||||
),
|
||||
(
|
||||
[{"role": "user", "content": f"my ssn is {_SSN}"}],
|
||||
[{"role": "user", "content": "my ssn is [SSN]"}],
|
||||
),
|
||||
],
|
||||
ids=["multipart", "multipart_among_string_rows", "string_only"],
|
||||
)
|
||||
async def test_every_row_echoed_applies_the_masked_texts(
|
||||
self, messages: list[AllMessageValues], expected: list[AllMessageValues]
|
||||
) -> None:
|
||||
guardrail: Final = _guardrail_answering(_echo_every_row_and_mask_texts)
|
||||
|
||||
llm_bound: Final = await _llm_bound_messages(guardrail, messages)
|
||||
|
||||
assert llm_bound == expected, "an unchanged echo of every row must leave the rewrite to texts"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_every_anthropic_content_block_row_echoed_applies_the_masked_texts(self) -> None:
|
||||
guardrail: Final = _guardrail_answering(_echo_every_row_and_mask_texts)
|
||||
|
||||
llm_bound: Final = await _llm_bound_anthropic_messages(
|
||||
guardrail,
|
||||
[
|
||||
{
|
||||
"role": "user",
|
||||
"content": [{"type": "text", "text": f"my ssn is {_SSN}"}, {"type": "text", "text": "ok"}],
|
||||
}
|
||||
],
|
||||
)
|
||||
|
||||
assert llm_bound == [
|
||||
{"role": "user", "content": [{"type": "text", "text": "my ssn is [SSN]"}, {"type": "text", "text": "ok"}]}
|
||||
], "an unchanged echo of every content block row must leave the rewrite to texts"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_echoed_multipart_row_is_restored_to_the_callers_row(self) -> None:
|
||||
guardrail: Final = _guardrail_answering(_echo_first_row_and_mask_the_rest)
|
||||
|
||||
llm_bound: Final = await _llm_bound_messages(
|
||||
guardrail,
|
||||
[
|
||||
{"role": "user", "name": "pat", "content": [{"type": "text", "text": "what is this?"}, _image_part()]},
|
||||
{"role": "user", "content": f"my ssn is {_SSN}"},
|
||||
],
|
||||
)
|
||||
|
||||
assert llm_bound == [
|
||||
{"role": "user", "name": "pat", "content": [{"type": "text", "text": "what is this?"}, _image_part()]},
|
||||
{"role": "user", "content": "my ssn is [SSN]"},
|
||||
], "the echoed row must keep the caller's keys the request model drops"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"unvalidated_part", [_guarded_text_part, _pdf_document_part], ids=["guarded_text", "pdf_document"]
|
||||
)
|
||||
async def test_a_row_holding_a_part_the_request_model_rejects_is_masked(
|
||||
self, unvalidated_part: Callable[[], Mapping[str, JsonValue]]
|
||||
) -> None:
|
||||
guardrail: Final = _guardrail_answering(_mask_every_row_and_text)
|
||||
|
||||
llm_bound: Final = await _llm_bound_messages(
|
||||
guardrail,
|
||||
[
|
||||
{"role": "user", "content": [{"type": "text", "text": f"my ssn is {_SSN}"}, unvalidated_part()]},
|
||||
{"role": "user", "content": f"also {_SSN}"},
|
||||
],
|
||||
)
|
||||
|
||||
assert llm_bound == [
|
||||
{"role": "user", "content": [{"type": "text", "text": "my ssn is [SSN]"}, unvalidated_part()]},
|
||||
{"role": "user", "content": "also [SSN]"},
|
||||
], "the guardrail must see the whole row so its masking of it reaches the LLM"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_echoed_row_holding_a_part_the_request_model_rejects_is_restored_to_the_callers_row(
|
||||
self,
|
||||
) -> None:
|
||||
guardrail: Final = _guardrail_answering(_echo_first_row_and_mask_the_rest)
|
||||
|
||||
llm_bound: Final = await _llm_bound_messages(
|
||||
guardrail,
|
||||
[
|
||||
{"role": "user", "name": "pat", "content": [{"type": "text", "text": "hi"}, _guarded_text_part()]},
|
||||
{"role": "user", "content": f"my ssn is {_SSN}"},
|
||||
],
|
||||
)
|
||||
|
||||
assert llm_bound == [
|
||||
{"role": "user", "name": "pat", "content": [{"type": "text", "text": "hi"}, _guarded_text_part()]},
|
||||
{"role": "user", "content": "my ssn is [SSN]"},
|
||||
], "the echoed row must keep the caller's keys the request model drops"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rows_the_request_model_accepts_are_posted_as_it_dumps_them(self) -> None:
|
||||
posted: Final = await _structured_messages_posted_for(
|
||||
[
|
||||
{
|
||||
"role": "user",
|
||||
"name": "pat",
|
||||
"content": [
|
||||
{"type": "text", "text": "hi"},
|
||||
{**_image_part(), "cache_control": {"type": "ephemeral"}},
|
||||
],
|
||||
},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{
|
||||
"id": "call_1",
|
||||
"type": "function",
|
||||
"function": {"name": "lookup", "arguments": "{}"},
|
||||
"index": 0,
|
||||
}
|
||||
],
|
||||
},
|
||||
{"role": "tool", "tool_call_id": "call_1", "content": "done"},
|
||||
]
|
||||
)
|
||||
|
||||
assert posted == [
|
||||
[
|
||||
{"role": "user", "content": [{"type": "text", "text": "hi"}, _image_part()]},
|
||||
{
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{"id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}
|
||||
],
|
||||
},
|
||||
{"role": "tool", "tool_call_id": "call_1", "content": "done"},
|
||||
]
|
||||
], "rows the request model dumps in full must reach the guardrail exactly as before"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_row_holding_a_part_the_request_model_rejects_is_posted_with_the_callers_content(self) -> None:
|
||||
posted: Final = await _structured_messages_posted_for(
|
||||
[{"role": "user", "name": "pat", "content": [{"type": "text", "text": "hi"}, _pdf_document_part()]}]
|
||||
)
|
||||
|
||||
assert posted == [[{"role": "user", "content": [{"type": "text", "text": "hi"}, _pdf_document_part()]}]], (
|
||||
"only the content the request model emptied is taken from the caller, the row keys stay as dumped"
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_deeply_nested_part_the_proxy_accepts_is_posted_in_full(self) -> None:
|
||||
deep_document: Final = {"type": "document", "source": _nested(300)}
|
||||
|
||||
posted: Final = await _structured_messages_posted_for(
|
||||
[{"role": "user", "content": [{"type": "text", "text": "hi"}, deep_document]}]
|
||||
)
|
||||
|
||||
assert posted == [[{"role": "user", "content": [{"type": "text", "text": "hi"}, deep_document]}]], (
|
||||
"content nested as deep as the proxy's own JSON parser allows must still reach the guardrail"
|
||||
)
|
||||
|
||||
|
||||
class TestImageSupport:
|
||||
"""Test image handling in guardrail requests"""
|
||||
|
||||
|
|
|
|||
|
|
@ -1,12 +1,20 @@
|
|||
"""Tests for the shared guardrail content extraction helpers."""
|
||||
|
||||
import copy
|
||||
|
||||
import pytest
|
||||
|
||||
from litellm.proxy.guardrails._content_utils import (
|
||||
apply_redacted_messages_back,
|
||||
as_json_value,
|
||||
build_inspection_messages,
|
||||
has_non_string_content,
|
||||
image_part_url,
|
||||
is_non_conversational_call_type,
|
||||
is_string_batch_input,
|
||||
iter_message_text,
|
||||
map_content_image_urls,
|
||||
map_messages_image_urls,
|
||||
walk_user_text,
|
||||
)
|
||||
|
||||
|
|
@ -741,3 +749,96 @@ def test_is_non_conversational_call_type_defaults_to_inspecting_unknown_call_typ
|
|||
"""A call type this module has never heard of must still be inspected —
|
||||
failing closed is the point of the deny-list."""
|
||||
assert is_non_conversational_call_type("some_future_call_type") is False
|
||||
|
||||
|
||||
def _nested(depth: int) -> dict[str, object]:
|
||||
return {"leaf": "x"} if depth == 0 else {"nested": _nested(depth - 1)}
|
||||
|
||||
|
||||
def test_as_json_value_keeps_content_nested_past_the_pydantic_serializer_limit_and_decodes_bytes():
|
||||
assert as_json_value([{"type": "document", "source": _nested(600), "data": b"raw"}, ("a", 1)]) == [
|
||||
{"type": "document", "source": _nested(600), "data": "raw"},
|
||||
["a", 1],
|
||||
], "nothing may be truncated to '...' and non-JSON types must become their JSON form"
|
||||
|
||||
|
||||
# ── map_content_image_urls / map_messages_image_urls ─────────────────────────────
|
||||
|
||||
|
||||
def _omit(url: str) -> str:
|
||||
return f"[omitted {len(url)}]"
|
||||
|
||||
|
||||
def test_map_messages_image_urls_covers_every_image_part_shape():
|
||||
messages = [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": "describe"},
|
||||
{"type": "image_url", "image_url": {"url": "data:AAA", "detail": "low"}},
|
||||
{"type": "image_url", "image_url": "data:BBBB"},
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
assert map_messages_image_urls(messages, _omit) == [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": "describe"},
|
||||
{"type": "image_url", "image_url": {"url": "[omitted 8]", "detail": "low"}},
|
||||
{"type": "image_url", "image_url": "[omitted 9]"},
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
def test_map_messages_image_urls_does_not_mutate_its_input():
|
||||
messages = [
|
||||
{"role": "user", "content": [{"type": "image_url", "image_url": {"url": "data:AAA"}}]},
|
||||
{"role": "user", "content": [{"type": "image_url", "image_url": "data:BBB"}]},
|
||||
]
|
||||
snapshot = copy.deepcopy(messages)
|
||||
|
||||
map_messages_image_urls(messages, _omit)
|
||||
|
||||
assert messages == snapshot
|
||||
|
||||
|
||||
def test_map_content_image_urls_passes_non_image_parts_through_by_identity():
|
||||
text_part = {"type": "text", "text": "describe"}
|
||||
audio_part = {"type": "input_audio", "input_audio": {"data": "AAA", "format": "wav"}}
|
||||
file_id_image = {"type": "input_image", "file_id": "file-1"}
|
||||
content = [text_part, "bare text", audio_part, file_id_image]
|
||||
|
||||
mapped = map_content_image_urls(content, _omit)
|
||||
|
||||
assert all(after is before for after, before in zip(mapped, content, strict=True))
|
||||
|
||||
|
||||
def test_map_messages_image_urls_passes_messages_without_list_content_through_by_identity():
|
||||
string_message = {"role": "user", "content": "just text"}
|
||||
tool_call_message = {"role": "assistant", "content": None, "tool_calls": []}
|
||||
messages = [string_message, tool_call_message]
|
||||
|
||||
mapped = map_messages_image_urls(messages, _omit)
|
||||
|
||||
assert mapped[0] is string_message
|
||||
assert mapped[1] is tool_call_message
|
||||
assert map_messages_image_urls(None, _omit) is None
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("part", "url"),
|
||||
[
|
||||
({"type": "image_url", "image_url": {"url": "data:AAA", "detail": "low"}}, "data:AAA"),
|
||||
({"type": "image_url", "image_url": "data:BBB"}, "data:BBB"),
|
||||
({"type": "image_url", "image_url": {"detail": "low"}}, None),
|
||||
({"type": "input_image", "image_url": "data:CCC"}, None),
|
||||
({"type": "text", "text": "data:DDD"}, None),
|
||||
("data:EEE", None),
|
||||
],
|
||||
ids=["object", "bare_string", "no_url", "responses_shape", "text", "bare_text"],
|
||||
)
|
||||
def test_image_part_url_reads_only_chat_image_url_parts(part, url):
|
||||
assert image_part_url(part) == url
|
||||
|
|
|
|||
|
|
@ -0,0 +1,725 @@
|
|||
import copy
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from litellm.exceptions import GuardrailRaisedException
|
||||
from litellm.llms.anthropic.chat.guardrail_translation.handler import AnthropicMessagesHandler
|
||||
from litellm.llms.base_llm.guardrail_translation.utils import UnappliableRequestRewrite
|
||||
from litellm.llms.openai.chat.guardrail_translation.handler import OpenAIChatCompletionsHandler
|
||||
from litellm.llms.openai.responses.guardrail_translation.handler import OpenAIResponsesHandler
|
||||
from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api import GenericGuardrailAPI
|
||||
from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.payload_policy import (
|
||||
IMAGE_OMITTED_PLACEHOLDER,
|
||||
)
|
||||
from litellm.types.proxy.guardrails.guardrail_hooks.generic_guardrail_api import GenericGuardrailAPIRequest
|
||||
|
||||
SSN = "123-45-6789"
|
||||
IMAGE_URL = "data:image/png;base64,SECRETPIXELS"
|
||||
OTHER_IMAGE_URL = "data:image/png;base64,OTHERPIXELS"
|
||||
|
||||
|
||||
def _answer(body: dict) -> Callable[[dict], dict]:
|
||||
return lambda _payload: body
|
||||
|
||||
|
||||
class _FakeGuardrailEndpoint:
|
||||
def __init__(self, respond: Callable[[dict], dict] = _answer({"action": "NONE"})):
|
||||
self.payloads: list[dict] = []
|
||||
self._respond = respond
|
||||
|
||||
async def post(self, *, url: str, json: dict, headers: dict, **_kwargs: object) -> httpx.Response:
|
||||
sent = _wire_copy(json)
|
||||
self.payloads.append(sent)
|
||||
return httpx.Response(200, json=self._respond(sent), request=httpx.Request("POST", url))
|
||||
|
||||
|
||||
class _LoggingObj:
|
||||
litellm_call_id = "call-abc"
|
||||
litellm_trace_id = "trace-abc"
|
||||
|
||||
|
||||
def _wire_copy(payload: dict) -> dict:
|
||||
return json.loads(json.dumps(payload))
|
||||
|
||||
|
||||
def _guardrail(endpoint: _FakeGuardrailEndpoint, **options: object) -> GenericGuardrailAPI:
|
||||
return GenericGuardrailAPI(
|
||||
api_base="https://guardrail.example",
|
||||
guardrail_name="payload-policy-test",
|
||||
event_hook="pre_call",
|
||||
default_on=True,
|
||||
async_handler=endpoint,
|
||||
**options,
|
||||
)
|
||||
|
||||
|
||||
def _image_message() -> dict:
|
||||
return {
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": f"my ssn is {SSN}"},
|
||||
{"type": "image_url", "image_url": {"url": IMAGE_URL, "detail": "low"}},
|
||||
{"type": "image_url", "image_url": OTHER_IMAGE_URL},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _text_message() -> dict:
|
||||
return {"role": "user", "content": f"also {SSN}"}
|
||||
|
||||
|
||||
def _texts() -> list[str]:
|
||||
return [f"my ssn is {SSN}", f"also {SSN}"]
|
||||
|
||||
|
||||
def _masked_text(text: str) -> str:
|
||||
return text.replace(SSN, "[SSN]")
|
||||
|
||||
|
||||
def _masked_part(part: dict) -> dict:
|
||||
return {**part, "text": _masked_text(part["text"])} if part.get("type") == "text" else part
|
||||
|
||||
|
||||
def _masked_row(row: dict) -> dict:
|
||||
content = row["content"]
|
||||
if isinstance(content, str):
|
||||
return {**row, "content": _masked_text(content)}
|
||||
return {**row, "content": [_masked_part(part) for part in content]}
|
||||
|
||||
|
||||
def _masking_guardrail(payload: dict) -> dict:
|
||||
return {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"texts": [_masked_text(text) for text in payload.get("texts", [])],
|
||||
"structured_messages": [_masked_row(row) for row in payload.get("structured_messages") or []],
|
||||
}
|
||||
|
||||
|
||||
def _rewrite_rows(rewrite: Callable[[int, dict], dict]) -> Callable[[dict], dict]:
|
||||
def respond(payload: dict) -> dict:
|
||||
rows = payload["structured_messages"]
|
||||
return {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"structured_messages": [rewrite(i, row) for i, row in enumerate(rows)],
|
||||
}
|
||||
|
||||
return respond
|
||||
|
||||
|
||||
async def _apply(guardrail: GenericGuardrailAPI, inputs: dict) -> dict:
|
||||
return await guardrail.apply_guardrail(
|
||||
inputs=inputs,
|
||||
request_data={"proxy_server_request": {"headers": {"user-agent": "curl/8"}}},
|
||||
input_type="request",
|
||||
logging_obj=_LoggingObj(),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_defaults_send_every_request_field_unchanged():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
tools = [{"type": "function", "function": {"name": "lookup"}}]
|
||||
|
||||
await _apply(
|
||||
_guardrail(endpoint),
|
||||
{"texts": ["describe this"], "images": [IMAGE_URL], "tools": tools, "structured_messages": [_image_message()]},
|
||||
)
|
||||
|
||||
payload = endpoint.payloads[0]
|
||||
assert set(payload) == set(GenericGuardrailAPIRequest.model_fields)
|
||||
assert payload["images"] == [IMAGE_URL]
|
||||
assert payload["texts"] == ["describe this"]
|
||||
assert payload["tools"] == tools
|
||||
assert payload["structured_messages"] == [_image_message()]
|
||||
assert payload["request_headers"] == {"user-agent": "curl/8"}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_defaults_still_accept_every_rewrite():
|
||||
rewritten_tools = [{"type": "function", "function": {"name": "safe_lookup"}}]
|
||||
endpoint = _FakeGuardrailEndpoint(
|
||||
_answer(
|
||||
{
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"texts": ["[MASKED]"],
|
||||
"images": [OTHER_IMAGE_URL],
|
||||
"tools": rewritten_tools,
|
||||
}
|
||||
)
|
||||
)
|
||||
|
||||
result = await _apply(
|
||||
_guardrail(endpoint),
|
||||
{"texts": ["my ssn is 123"], "images": [IMAGE_URL], "tools": [{"type": "function", "function": {"name": "x"}}]},
|
||||
)
|
||||
|
||||
assert result == {"texts": ["[MASKED]"], "images": [OTHER_IMAGE_URL], "tools": rewritten_tools}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_images_false_withholds_every_image_but_keeps_the_parts():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
messages = [_image_message(), _text_message()]
|
||||
|
||||
await _apply(
|
||||
_guardrail(endpoint, send_images=False),
|
||||
{"texts": _texts(), "images": [IMAGE_URL], "structured_messages": messages},
|
||||
)
|
||||
|
||||
payload = endpoint.payloads[0]
|
||||
assert "SECRETPIXELS" not in json.dumps(payload)
|
||||
assert "OTHERPIXELS" not in json.dumps(payload)
|
||||
assert "images" not in payload
|
||||
assert payload["texts"] == _texts()
|
||||
assert payload["structured_messages"] == [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": f"my ssn is {SSN}"},
|
||||
{"type": "image_url", "image_url": {"url": IMAGE_OMITTED_PLACEHOLDER, "detail": "low"}},
|
||||
{"type": "image_url", "image_url": IMAGE_OMITTED_PLACEHOLDER},
|
||||
],
|
||||
},
|
||||
_text_message(),
|
||||
]
|
||||
assert messages == [_image_message(), _text_message()]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_images_false_withholds_the_image_in_a_row_the_request_model_cannot_validate():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
row = {
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "guarded_text", "text": "hi"},
|
||||
{"type": "image_url", "image_url": {"url": IMAGE_URL}},
|
||||
],
|
||||
}
|
||||
|
||||
await _apply(_guardrail(endpoint, send_images=False), {"texts": ["hi"], "structured_messages": [row]})
|
||||
|
||||
assert "SECRETPIXELS" not in json.dumps(endpoint.payloads[0])
|
||||
assert endpoint.payloads[0]["structured_messages"] == [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "guarded_text", "text": "hi"},
|
||||
{"type": "image_url", "image_url": {"url": IMAGE_OMITTED_PLACEHOLDER}},
|
||||
],
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_images_false_keeps_the_callers_images():
|
||||
endpoint = _FakeGuardrailEndpoint(
|
||||
_answer({"action": "GUARDRAIL_INTERVENED", "texts": ["[MASKED]"], "images": [OTHER_IMAGE_URL]})
|
||||
)
|
||||
|
||||
result = await _apply(_guardrail(endpoint, send_images=False), {"texts": ["ssn 1"], "images": [IMAGE_URL]})
|
||||
|
||||
assert result == {"texts": ["[MASKED]"], "images": [IMAGE_URL]}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_images_false_masks_the_llm_bound_text_and_keeps_the_callers_image():
|
||||
guardrail = _guardrail(_FakeGuardrailEndpoint(_masking_guardrail), send_images=False)
|
||||
data = {"model": "gpt-x", "messages": [_image_message(), _text_message()]}
|
||||
|
||||
result = await OpenAIChatCompletionsHandler().process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert result["messages"] == [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": "my ssn is [SSN]"},
|
||||
{"type": "image_url", "image_url": {"url": IMAGE_URL, "detail": "low"}},
|
||||
{"type": "image_url", "image_url": OTHER_IMAGE_URL},
|
||||
],
|
||||
},
|
||||
{"role": "user", "content": "also [SSN]"},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_images_false_masks_the_responses_input_and_keeps_the_callers_input_image():
|
||||
guardrail = _guardrail(_FakeGuardrailEndpoint(_masking_guardrail), send_images=False)
|
||||
data = {
|
||||
"model": "gpt-x",
|
||||
"input": [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "input_text", "text": f"my ssn is {SSN}"},
|
||||
{"type": "input_image", "image_url": IMAGE_URL},
|
||||
],
|
||||
},
|
||||
{"role": "user", "content": f"also {SSN}"},
|
||||
],
|
||||
}
|
||||
|
||||
result = await OpenAIResponsesHandler().process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert result["input"] == [
|
||||
{
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "input_text", "text": "my ssn is [SSN]"},
|
||||
{"type": "input_image", "image_url": IMAGE_URL, "detail": "auto"},
|
||||
],
|
||||
},
|
||||
{"role": "user", "content": "also [SSN]"},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_send_images_false_masks_the_anthropic_messages_and_keeps_the_callers_base64_image():
|
||||
endpoint = _FakeGuardrailEndpoint(_masking_guardrail)
|
||||
guardrail = _guardrail(endpoint, send_images=False)
|
||||
image_block = {"type": "image", "source": {"type": "base64", "media_type": "image/png", "data": "SECRETPIXELS"}}
|
||||
data = {
|
||||
"model": "claude-x",
|
||||
"max_tokens": 5,
|
||||
"system": f"sys {SSN}",
|
||||
"messages": [
|
||||
{"role": "user", "content": [{"type": "text", "text": f"my ssn is {SSN}"}, image_block]},
|
||||
{"role": "assistant", "content": "ok"},
|
||||
{"role": "user", "content": f"also {SSN}"},
|
||||
],
|
||||
}
|
||||
|
||||
result = await AnthropicMessagesHandler().process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert "SECRETPIXELS" not in json.dumps(endpoint.payloads[0])
|
||||
assert result["system"] == [{"type": "text", "text": "sys [SSN]"}]
|
||||
assert result["messages"] == [
|
||||
{"role": "user", "content": [{"type": "text", "text": "my ssn is [SSN]"}, image_block]},
|
||||
{"role": "assistant", "content": [{"type": "text", "text": "ok"}]},
|
||||
{"role": "user", "content": [{"type": "text", "text": "also [SSN]"}]},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_rewritten_image_row_gets_the_callers_images_back():
|
||||
messages = [_image_message(), _text_message()]
|
||||
endpoint = _FakeGuardrailEndpoint(_rewrite_rows(lambda i, row: _masked_row(row) if i == 0 else row))
|
||||
|
||||
result = await _apply(_guardrail(endpoint, send_images=False), {"texts": _texts(), "structured_messages": messages})
|
||||
|
||||
assert result == {"texts": _texts(), "structured_messages": [_masked_row(_image_message()), messages[1]]}
|
||||
assert result["structured_messages"][1] is messages[1]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_echoed_placeholder_row_is_restored_to_the_callers_row():
|
||||
messages = [_image_message(), _text_message()]
|
||||
endpoint = _FakeGuardrailEndpoint(_rewrite_rows(lambda i, row: _masked_row(row) if i == 1 else row))
|
||||
|
||||
result = await _apply(_guardrail(endpoint, send_images=False), {"texts": _texts(), "structured_messages": messages})
|
||||
|
||||
assert result == {"texts": _texts(), "structured_messages": [messages[0], _masked_row(_text_message())]}
|
||||
assert result["structured_messages"][0] is messages[0]
|
||||
|
||||
|
||||
_TEXT_PART = {"type": "text", "text": "my ssn is [SSN]"}
|
||||
_OBJECT_PLACEHOLDER = {"type": "image_url", "image_url": {"url": IMAGE_OMITTED_PLACEHOLDER, "detail": "low"}}
|
||||
_BARE_PLACEHOLDER = {"type": "image_url", "image_url": IMAGE_OMITTED_PLACEHOLDER}
|
||||
|
||||
|
||||
def _with_content(content: object) -> Callable[[int, dict], dict]:
|
||||
return lambda i, row: {**row, "content": content} if i == 0 else row
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
"rewrite",
|
||||
[
|
||||
_with_content([_TEXT_PART, {"type": "image_url", "image_url": {"url": OTHER_IMAGE_URL}}, _BARE_PLACEHOLDER]),
|
||||
_with_content([_TEXT_PART, {"type": "text", "text": "y"}, _BARE_PLACEHOLDER]),
|
||||
_with_content([_BARE_PLACEHOLDER, _OBJECT_PLACEHOLDER, _BARE_PLACEHOLDER]),
|
||||
_with_content([{"type": "text", "text": "x"}]),
|
||||
_with_content("my ssn is [SSN]"),
|
||||
],
|
||||
ids=[
|
||||
"image_part_changed",
|
||||
"image_part_replaced_by_text",
|
||||
"text_part_replaced_by_placeholder",
|
||||
"parts_dropped",
|
||||
"content_flattened",
|
||||
],
|
||||
)
|
||||
async def test_a_rewrite_that_does_not_line_up_with_a_withheld_image_blocks_the_request(rewrite):
|
||||
endpoint = _FakeGuardrailEndpoint(_rewrite_rows(rewrite))
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _apply(
|
||||
_guardrail(endpoint, send_images=False),
|
||||
{"texts": _texts(), "structured_messages": [_image_message(), _text_message()]},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_misaligned_rows_block_the_request_when_an_image_was_withheld():
|
||||
endpoint = _FakeGuardrailEndpoint(
|
||||
_answer({"action": "GUARDRAIL_INTERVENED", "structured_messages": [{"role": "user", "content": "[MASKED]"}]})
|
||||
)
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _apply(
|
||||
_guardrail(endpoint, send_images=False),
|
||||
{"texts": _texts(), "structured_messages": [_text_message(), _image_message()]},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_placeholder_copied_to_a_row_without_images_blocks_the_request():
|
||||
endpoint = _FakeGuardrailEndpoint(
|
||||
lambda payload: {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"structured_messages": [payload["structured_messages"][0], payload["structured_messages"][0]],
|
||||
}
|
||||
)
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _apply(
|
||||
_guardrail(endpoint, send_images=False),
|
||||
{"texts": _texts(), "structured_messages": [_image_message(), _text_message()]},
|
||||
)
|
||||
|
||||
|
||||
def _row_with_a_part_the_request_model_cannot_validate() -> dict:
|
||||
return {
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": f"my ssn is {SSN}"},
|
||||
{"type": "image_url", "image_url": {"url": IMAGE_URL}},
|
||||
{"type": "input_audio", "input_audio": {"data": "AAAA", "format": "flac"}},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
async def _chat_messages_reaching_the_llm(guardrail: GenericGuardrailAPI, messages: list[dict]) -> list[dict]:
|
||||
data = {"model": "gpt-x", "messages": messages}
|
||||
return (await OpenAIChatCompletionsHandler().process_input_messages(data=data, guardrail_to_apply=guardrail))[
|
||||
"messages"
|
||||
]
|
||||
|
||||
|
||||
def _masked_row_with_a_part_the_request_model_cannot_validate() -> dict:
|
||||
return {
|
||||
"role": "user",
|
||||
"content": [
|
||||
{"type": "text", "text": "my ssn is [SSN]"},
|
||||
{"type": "image_url", "image_url": {"url": IMAGE_URL}},
|
||||
{"type": "input_audio", "input_audio": {"data": "AAAA", "format": "flac"}},
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_row_with_a_part_the_request_model_cannot_validate_reaches_the_llm_masked():
|
||||
guardrail = _guardrail(_FakeGuardrailEndpoint(_masking_guardrail), send_images=False)
|
||||
|
||||
messages = await _chat_messages_reaching_the_llm(
|
||||
guardrail, [_row_with_a_part_the_request_model_cannot_validate(), _text_message()]
|
||||
)
|
||||
|
||||
assert messages == [
|
||||
_masked_row_with_a_part_the_request_model_cannot_validate(),
|
||||
{"role": "user", "content": "also [SSN]"},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_row_with_a_part_the_request_model_cannot_validate_reaches_the_llm_masked_at_defaults():
|
||||
guardrail = _guardrail(_FakeGuardrailEndpoint(_masking_guardrail))
|
||||
|
||||
messages = await _chat_messages_reaching_the_llm(
|
||||
guardrail, [_row_with_a_part_the_request_model_cannot_validate(), _text_message()]
|
||||
)
|
||||
|
||||
assert messages == [
|
||||
_masked_row_with_a_part_the_request_model_cannot_validate(),
|
||||
{"role": "user", "content": "also [SSN]"},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_defaults_still_write_back_rewritten_image_rows():
|
||||
messages = [_image_message(), _text_message()]
|
||||
endpoint = _FakeGuardrailEndpoint(_rewrite_rows(lambda _i, row: {**row, "content": "[MASKED]"}))
|
||||
|
||||
result = await _apply(_guardrail(endpoint), {"texts": _texts(), "structured_messages": messages})
|
||||
|
||||
assert result == {
|
||||
"texts": _texts(),
|
||||
"structured_messages": [{"role": "user", "content": "[MASKED]"}, {"role": "user", "content": "[MASKED]"}],
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_exclude_payload_fields_drops_only_the_named_fields(caplog):
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
guardrail = _guardrail(
|
||||
endpoint, exclude_payload_fields=["request_headers", "litellm_version", "input_type", "not_a_field"]
|
||||
)
|
||||
|
||||
await _apply(guardrail, {"texts": ["hello"]})
|
||||
|
||||
payload = endpoint.payloads[0]
|
||||
assert set(GenericGuardrailAPIRequest.model_fields) - set(payload) == {"request_headers", "litellm_version"}
|
||||
assert payload["input_type"] == "request"
|
||||
assert payload["litellm_call_id"] == "call-abc"
|
||||
assert payload["texts"] == ["hello"]
|
||||
assert any("not_a_field" in message for message in caplog.messages)
|
||||
assert any("input_type" in message and "still sent" in message for message in caplog.messages)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_litellm_call_id_cannot_be_excluded():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _apply(_guardrail(endpoint, exclude_payload_fields=["litellm_call_id"]), {"texts": ["hello"]})
|
||||
|
||||
assert endpoint.payloads[0]["litellm_call_id"] == "call-abc"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_exclude_config_naming_only_unknown_or_protected_fields_is_not_lossy(caplog):
|
||||
endpoint = _FakeGuardrailEndpoint(_answer({"action": "GUARDRAIL_INTERVENED", "texts": ["[MASKED]"]}))
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
guardrail = _guardrail(endpoint, exclude_payload_fields=["not_a_field", "input_type"])
|
||||
|
||||
result = await _apply(guardrail, {"texts": ["ssn 123"]})
|
||||
|
||||
assert set(endpoint.payloads[0]) == set(GenericGuardrailAPIRequest.model_fields)
|
||||
assert result == {"texts": ["[MASKED]"]}
|
||||
assert not any("can only enforce on what it is sent" in message for message in caplog.messages)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("field", "inputs", "response", "expected"),
|
||||
[
|
||||
("texts", {"texts": ["ssn 1", "ssn 2"]}, {"texts": ["[M]", "[M]"]}, {"texts": ["ssn 1", "ssn 2"]}),
|
||||
("texts", {"texts": []}, {"texts": ["INVENTED"]}, {"texts": []}),
|
||||
(
|
||||
"tools",
|
||||
{"texts": ["hi"], "tools": [{"type": "function", "function": {"name": "run"}}]},
|
||||
{"tools": [{"type": "function"}]},
|
||||
{"texts": ["hi"], "tools": [{"type": "function", "function": {"name": "run"}}]},
|
||||
),
|
||||
(
|
||||
"images",
|
||||
{"texts": ["hi"], "images": [IMAGE_URL]},
|
||||
{"images": [OTHER_IMAGE_URL]},
|
||||
{"texts": ["hi"], "images": [IMAGE_URL]},
|
||||
),
|
||||
(
|
||||
"structured_messages",
|
||||
{"texts": ["hi"], "structured_messages": [{"role": "user", "content": "hi"}]},
|
||||
{"structured_messages": [{"role": "user", "content": "INVENTED"}]},
|
||||
{"texts": ["hi"]},
|
||||
),
|
||||
(
|
||||
"structured_messages",
|
||||
{"texts": ["hi"], "structured_messages": []},
|
||||
{"structured_messages": [{"role": "user", "content": "INVENTED"}]},
|
||||
{"texts": ["hi"]},
|
||||
),
|
||||
(
|
||||
"structured_messages",
|
||||
{"texts": ["hi"]},
|
||||
{"structured_messages": [{"role": "user", "content": "INVENTED"}]},
|
||||
{"texts": ["hi"]},
|
||||
),
|
||||
],
|
||||
ids=[
|
||||
"texts",
|
||||
"texts_empty",
|
||||
"tools",
|
||||
"images",
|
||||
"structured_messages",
|
||||
"structured_messages_empty",
|
||||
"structured_messages_none",
|
||||
],
|
||||
)
|
||||
async def test_an_excluded_field_is_not_sent_and_its_rewrite_is_refused(caplog, field, inputs, response, expected):
|
||||
endpoint = _FakeGuardrailEndpoint(_answer({"action": "NONE", **response}))
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
result = await _apply(_guardrail(endpoint, exclude_payload_fields=[field]), inputs)
|
||||
|
||||
assert field not in endpoint.payloads[0]
|
||||
assert result == expected
|
||||
assert any(f"ignoring the returned {field}" in message for message in caplog.messages)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("excluded", "response"),
|
||||
[
|
||||
(["texts"], {"texts": ["[MASKED]"]}),
|
||||
(["texts"], {"texts": ["[MASKED]"], "structured_messages": [{"role": "user", "content": "ssn 1"}]}),
|
||||
(
|
||||
["texts"],
|
||||
{
|
||||
"texts": ["[MASKED]"],
|
||||
"structured_messages": [{"role": "user", "content": "ssn 1"}],
|
||||
"tools": [{"type": "function", "function": {"name": "run"}}],
|
||||
"images": [IMAGE_URL],
|
||||
},
|
||||
),
|
||||
(["images"], {"images": [OTHER_IMAGE_URL]}),
|
||||
(["tools"], {"tools": [{"type": "function"}]}),
|
||||
(["structured_messages"], {"structured_messages": [{"role": "user", "content": "[MASKED]"}]}),
|
||||
],
|
||||
ids=[
|
||||
"texts",
|
||||
"texts_with_rows_echoed",
|
||||
"texts_with_every_other_field_echoed",
|
||||
"images",
|
||||
"tools",
|
||||
"structured_messages",
|
||||
],
|
||||
)
|
||||
async def test_an_intervention_made_only_through_fields_that_were_not_sent_blocks_the_request(excluded, response):
|
||||
endpoint = _FakeGuardrailEndpoint(_answer({"action": "GUARDRAIL_INTERVENED", **response}))
|
||||
inputs = {
|
||||
"texts": ["ssn 1"],
|
||||
"images": [IMAGE_URL],
|
||||
"tools": [{"type": "function", "function": {"name": "run"}}],
|
||||
"structured_messages": [{"role": "user", "content": "ssn 1"}],
|
||||
}
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _apply(_guardrail(endpoint, exclude_payload_fields=excluded), inputs)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_echo_counts_as_no_change_whatever_container_the_caller_used():
|
||||
endpoint = _FakeGuardrailEndpoint(
|
||||
_answer({"action": "GUARDRAIL_INTERVENED", "texts": ["[MASKED]"], "images": [IMAGE_URL]})
|
||||
)
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _apply(
|
||||
_guardrail(endpoint, exclude_payload_fields=["texts"]), {"texts": ["ssn 1"], "images": (IMAGE_URL,)}
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_chat_request_is_rejected_when_the_only_masking_went_to_texts_that_were_not_sent():
|
||||
def echo_rows_and_tools_and_mask_texts(payload: dict) -> dict:
|
||||
return {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"texts": [_masked_text(row["content"]) for row in payload["structured_messages"]],
|
||||
"structured_messages": payload["structured_messages"],
|
||||
"tools": payload["tools"],
|
||||
}
|
||||
|
||||
guardrail = _guardrail(_FakeGuardrailEndpoint(echo_rows_and_tools_and_mask_texts), exclude_payload_fields=["texts"])
|
||||
data = {
|
||||
"model": "gpt-x",
|
||||
"messages": [_text_message()],
|
||||
"tools": [{"type": "function", "function": {"name": "lookup", "parameters": {"type": "object"}}}],
|
||||
}
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await OpenAIChatCompletionsHandler().process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize(
|
||||
("response", "expected"),
|
||||
[
|
||||
({"texts": ["[MASKED]"], "images": [OTHER_IMAGE_URL]}, {"texts": ["[MASKED]"], "images": [IMAGE_URL]}),
|
||||
(
|
||||
{"images": [OTHER_IMAGE_URL], "structured_messages": [{"role": "user", "content": "[MASKED]"}]},
|
||||
{
|
||||
"texts": ["ssn 1"],
|
||||
"images": [IMAGE_URL],
|
||||
"structured_messages": [{"role": "user", "content": "[MASKED]"}],
|
||||
},
|
||||
),
|
||||
(
|
||||
{"images": [OTHER_IMAGE_URL], "tools": [{"type": "function", "function": {"name": "safe"}}]},
|
||||
{"texts": ["ssn 1"], "images": [IMAGE_URL], "tools": [{"type": "function", "function": {"name": "safe"}}]},
|
||||
),
|
||||
({}, {"texts": ["ssn 1"], "images": [IMAGE_URL]}),
|
||||
],
|
||||
ids=["texts_applied", "rows_applied", "tools_applied", "nothing_returned"],
|
||||
)
|
||||
async def test_an_intervention_that_does_not_depend_on_an_unsent_field_goes_through(response, expected):
|
||||
endpoint = _FakeGuardrailEndpoint(_answer({"action": "GUARDRAIL_INTERVENED", **response}))
|
||||
|
||||
result = await _apply(
|
||||
_guardrail(endpoint, exclude_payload_fields=["images"]),
|
||||
{"texts": ["ssn 1"], "images": [IMAGE_URL], "structured_messages": [{"role": "user", "content": "ssn 1"}]},
|
||||
)
|
||||
|
||||
assert result == expected
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_responses_handler_rejects_rows_it_did_not_send_even_when_texts_are_echoed():
|
||||
endpoint = _FakeGuardrailEndpoint(
|
||||
_answer(
|
||||
{
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"texts": ["hello"],
|
||||
"structured_messages": [{"role": "user", "content": "INVENTED"}],
|
||||
}
|
||||
)
|
||||
)
|
||||
guardrail = _guardrail(endpoint, exclude_payload_fields=["structured_messages"])
|
||||
data = {"model": "gpt-x", "input": [{"role": "user", "content": [{"type": "input_text", "text": "hello"}]}]}
|
||||
original_input = copy.deepcopy(data["input"])
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await OpenAIResponsesHandler().process_input_messages(data=data, guardrail_to_apply=guardrail)
|
||||
|
||||
assert data["input"] == original_input
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_blocked_still_blocks_a_shaped_payload():
|
||||
endpoint = _FakeGuardrailEndpoint(_answer({"action": "BLOCKED", "blocked_reason": "no"}))
|
||||
guardrail = _guardrail(endpoint, send_images=False, exclude_payload_fields=["texts", "tools"])
|
||||
|
||||
with pytest.raises(GuardrailRaisedException):
|
||||
await _apply(guardrail, {"texts": ["hello"], "structured_messages": [_image_message()]})
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"options",
|
||||
[{"send_images": False}, {"exclude_payload_fields": ["request_headers"]}],
|
||||
ids=["send_images", "exclude_payload_fields"],
|
||||
)
|
||||
def test_lossy_options_warn_that_the_guardrail_only_enforces_on_what_it_is_sent(caplog, options):
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
_guardrail(_FakeGuardrailEndpoint(), **options)
|
||||
|
||||
assert any("can only enforce on what it is sent" in message for message in caplog.messages)
|
||||
|
||||
|
||||
def test_defaults_do_not_warn(caplog):
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
_guardrail(_FakeGuardrailEndpoint(), send_images=True, exclude_payload_fields=[])
|
||||
|
||||
assert not any("can only enforce on what it is sent" in message for message in caplog.messages)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"options",
|
||||
[{"send_images": "false"}, {"send_images": 0}, {"exclude_payload_fields": "tools"}],
|
||||
ids=["send_images_string", "send_images_int", "exclude_payload_fields_string"],
|
||||
)
|
||||
def test_a_mistyped_option_fails_at_init(options):
|
||||
with pytest.raises(ValueError, match=next(iter(options))):
|
||||
_guardrail(_FakeGuardrailEndpoint(), **options)
|
||||
|
|
@ -0,0 +1,746 @@
|
|||
import copy
|
||||
import json
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from litellm.exceptions import GuardrailRaisedException
|
||||
from litellm.llms.anthropic.chat.guardrail_translation.handler import AnthropicMessagesHandler
|
||||
from litellm.llms.base_llm.guardrail_translation.utils import UnappliableRequestRewrite
|
||||
from litellm.llms.cohere.rerank.guardrail_translation.handler import CohereRerankHandler
|
||||
from litellm.llms.openai.chat.guardrail_translation.handler import OpenAIChatCompletionsHandler
|
||||
from litellm.llms.openai.embeddings.guardrail_translation.handler import OpenAIEmbeddingsHandler
|
||||
from litellm.llms.openai.responses.guardrail_translation.handler import OpenAIResponsesHandler
|
||||
from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api import GenericGuardrailAPI
|
||||
from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.payload_policy import (
|
||||
MAX_STRIP_CALL_CHARS,
|
||||
MAX_STRIP_SUBSTITUTIONS,
|
||||
)
|
||||
from litellm.types.utils import Choices, Message, ModelResponse, ModelResponseStream
|
||||
|
||||
SSN = "123-45-6789"
|
||||
TIMESTAMP = r"<ts>\d+</ts>"
|
||||
IMAGE_URL = "data:image/png;base64,PIXELS"
|
||||
|
||||
|
||||
def _answer(body: dict) -> Callable[[dict], dict]:
|
||||
return lambda _payload: body
|
||||
|
||||
|
||||
class _FakeGuardrailEndpoint:
|
||||
def __init__(self, respond: Callable[[dict], dict] = _answer({"action": "NONE"})):
|
||||
self.payloads: list[dict] = []
|
||||
self._respond = respond
|
||||
|
||||
async def post(self, *, url: str, json: dict, headers: dict, **_kwargs: object) -> httpx.Response:
|
||||
sent = _wire_copy(json)
|
||||
self.payloads.append(sent)
|
||||
return httpx.Response(200, json=self._respond(sent), request=httpx.Request("POST", url))
|
||||
|
||||
|
||||
class _LoggingObj:
|
||||
litellm_call_id = "call-abc"
|
||||
litellm_trace_id = "trace-abc"
|
||||
|
||||
def __init__(self) -> None:
|
||||
self.model_call_details: dict = {}
|
||||
|
||||
|
||||
def _wire_copy(payload: dict) -> dict:
|
||||
return json.loads(json.dumps(payload))
|
||||
|
||||
|
||||
def _guardrail(endpoint: _FakeGuardrailEndpoint, **options: object) -> GenericGuardrailAPI:
|
||||
return GenericGuardrailAPI(
|
||||
api_base="https://guardrail.example",
|
||||
guardrail_name="text-shaping-test",
|
||||
event_hook="pre_call",
|
||||
default_on=True,
|
||||
async_handler=endpoint,
|
||||
**options,
|
||||
)
|
||||
|
||||
|
||||
async def _apply(guardrail: GenericGuardrailAPI, inputs: dict, input_type: str = "request") -> dict:
|
||||
return await guardrail.apply_guardrail(
|
||||
inputs=inputs, request_data={}, input_type=input_type, logging_obj=_LoggingObj()
|
||||
)
|
||||
|
||||
|
||||
async def _chat_request(guardrail: GenericGuardrailAPI, messages: list[dict]) -> list[dict]:
|
||||
data = await OpenAIChatCompletionsHandler().process_input_messages(
|
||||
data={"model": "gpt-x", "messages": copy.deepcopy(messages)}, guardrail_to_apply=guardrail
|
||||
)
|
||||
return data["messages"]
|
||||
|
||||
|
||||
def _user(content: object) -> dict:
|
||||
return {"role": "user", "content": content}
|
||||
|
||||
|
||||
def _assistant(content: object) -> dict:
|
||||
return {"role": "assistant", "content": content}
|
||||
|
||||
|
||||
def _tool_call_turn() -> dict:
|
||||
return {
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "lookup", "arguments": "{}"}}],
|
||||
}
|
||||
|
||||
|
||||
def _multimodal_turn() -> dict:
|
||||
return _user(
|
||||
[
|
||||
{"type": "text", "text": "compare"},
|
||||
{"type": "image_url", "image_url": {"url": IMAGE_URL}},
|
||||
{"type": "text", "text": "these two"},
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def _conversation() -> list[dict]:
|
||||
return [
|
||||
{"role": "system", "content": "rules <ts>1</ts>"},
|
||||
_user("hi"),
|
||||
_assistant("noted <ts>2</ts>"),
|
||||
_user(f"my ssn is {SSN}"),
|
||||
]
|
||||
|
||||
|
||||
def _mask(text: str) -> str:
|
||||
return text.replace(SSN, "[SSN]")
|
||||
|
||||
|
||||
def _masked_row(row: dict) -> dict:
|
||||
content = row.get("content")
|
||||
if isinstance(content, str):
|
||||
return {**row, "content": _mask(content)}
|
||||
if not isinstance(content, list):
|
||||
return row
|
||||
return {
|
||||
**row,
|
||||
"content": [
|
||||
{**part, "text": _mask(part["text"])} if isinstance(part.get("text"), str) else part for part in content
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def _mask_ssn(payload: dict) -> dict:
|
||||
rows = payload.get("structured_messages")
|
||||
return {
|
||||
"action": "GUARDRAIL_INTERVENED",
|
||||
"texts": [_mask(text) for text in payload.get("texts") or ()],
|
||||
**({"structured_messages": [_masked_row(row) for row in rows]} if rows else {}),
|
||||
}
|
||||
|
||||
|
||||
def _echo(payload: dict) -> dict:
|
||||
return {"action": "NONE", "texts": payload.get("texts"), "structured_messages": payload.get("structured_messages")}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_defaults_send_the_whole_conversation_unchanged():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
long_text = "x <ts>1</ts> " * 20_000
|
||||
messages = [*_conversation(), _user(long_text)]
|
||||
|
||||
await _chat_request(_guardrail(endpoint, max_messages=None, max_text_chars=None, strip_patterns=None), messages)
|
||||
|
||||
payload = endpoint.payloads[0]
|
||||
assert payload["texts"] == ["rules <ts>1</ts>", "hi", "noted <ts>2</ts>", f"my ssn is {SSN}", long_text]
|
||||
assert payload["structured_messages"] == messages
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_messages_sends_the_last_rows_and_only_their_texts():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _chat_request(_guardrail(endpoint, max_messages=2), _conversation())
|
||||
|
||||
payload = endpoint.payloads[0]
|
||||
assert payload["structured_messages"] == _conversation()[-2:]
|
||||
assert payload["texts"] == ["noted <ts>2</ts>", f"my ssn is {SSN}"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("messages", "max_messages", "expected_texts"),
|
||||
[
|
||||
([_user("old"), _user("older"), _multimodal_turn()], 1, ["compare", "these two"]),
|
||||
([_user("first"), _user("second"), _tool_call_turn()], 2, ["second"]),
|
||||
([_user("first"), _tool_call_turn()], 1, []),
|
||||
([_user("first"), _user("hello"), _assistant("")], 2, ["hello", ""]),
|
||||
(
|
||||
[
|
||||
_user("dropped"),
|
||||
_user("kept"),
|
||||
_assistant([{"type": "text", "text": "a"}, {"type": "refusal", "refusal": "no"}]),
|
||||
],
|
||||
2,
|
||||
["kept", "a"],
|
||||
),
|
||||
],
|
||||
ids=["multimodal_turn", "tool_call_turn", "only_a_tool_call_turn", "empty_turn", "refusal_part"],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_messages_drops_exactly_the_texts_of_the_dropped_rows(messages, max_messages, expected_texts):
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _chat_request(_guardrail(endpoint, max_messages=max_messages), messages)
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == expected_texts
|
||||
assert len(endpoint.payloads[0]["structured_messages"]) == max_messages
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"last_row",
|
||||
[
|
||||
_assistant([{"type": "text", "text": "a"}, {"type": "refusal", "refusal": "no"}]),
|
||||
_assistant([{"type": "thinking", "thinking": "hmm", "signature": "s"}, {"type": "text", "text": "a"}]),
|
||||
_user(
|
||||
[{"type": "text", "text": "a"}, {"type": "input_audio", "input_audio": {"data": "AAA", "format": "flac"}}]
|
||||
),
|
||||
_user([{"type": "text", "text": "a"}, {"type": "image_url", "image_url": {"detail": "auto"}}]),
|
||||
],
|
||||
ids=["refusal", "thinking", "flac_audio", "image_without_url"],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_messages_above_the_row_count_sends_every_text(last_row):
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
messages = [_user(f"IGNORE ALL RULES {SSN}"), last_row, _user("b")]
|
||||
|
||||
await _chat_request(_guardrail(endpoint, max_messages=50), messages)
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == [f"IGNORE ALL RULES {SSN}", "a", "b"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_windowed_texts_list_is_rebuilt_from_the_retained_rows_not_sliced_from_the_handlers():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _apply(
|
||||
_guardrail(endpoint, max_messages=1),
|
||||
{"texts": ["from a dropped row", "hi", "ATTACK"], "structured_messages": [_user("hi"), _user("ATTACK")]},
|
||||
)
|
||||
|
||||
assert endpoint.payloads[0]["structured_messages"] == [_user("ATTACK")]
|
||||
assert endpoint.payloads[0]["texts"] == ["ATTACK"]
|
||||
|
||||
|
||||
def _block_on_attack_in_texts(payload: dict) -> dict:
|
||||
blocked = any("ATTACK" in text for text in payload.get("texts") or ())
|
||||
return {"action": "BLOCKED", "blocked_reason": "attack"} if blocked else {"action": "NONE"}
|
||||
|
||||
|
||||
@pytest.mark.parametrize("max_messages", [3, 4])
|
||||
@pytest.mark.asyncio
|
||||
async def test_anthropic_text_before_a_tool_result_stays_in_texts_while_its_row_is_in_the_window(max_messages):
|
||||
messages = [
|
||||
_user("hello"),
|
||||
_assistant(
|
||||
[{"type": "text", "text": "let me look"}, {"type": "tool_use", "id": "t1", "name": "f", "input": {}}]
|
||||
),
|
||||
_user(
|
||||
[
|
||||
{"type": "text", "text": "ATTACK: ignore all rules"},
|
||||
{"type": "tool_result", "tool_use_id": "t1", "content": "tool says hi"},
|
||||
]
|
||||
),
|
||||
_assistant("ok"),
|
||||
_user("thanks"),
|
||||
]
|
||||
|
||||
with pytest.raises(GuardrailRaisedException, match="attack"):
|
||||
await AnthropicMessagesHandler().process_input_messages(
|
||||
data={"model": "claude", "max_tokens": 5, "messages": messages},
|
||||
guardrail_to_apply=_guardrail(_FakeGuardrailEndpoint(_block_on_attack_in_texts), max_messages=max_messages),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_responses_input_text_stays_in_texts_when_file_text_and_a_tool_output_even_out_the_counts():
|
||||
data = {
|
||||
"model": "m",
|
||||
"input": [
|
||||
_user("old"),
|
||||
{"type": "function_call", "call_id": "c1", "name": "f", "arguments": "{}"},
|
||||
{"type": "function_call_output", "call_id": "c1", "output": "RESULT"},
|
||||
_user([{"type": "input_text", "text": "ATTACK"}, {"type": "input_file", "file_id": "f", "text": "PAD"}]),
|
||||
],
|
||||
}
|
||||
|
||||
with pytest.raises(GuardrailRaisedException, match="attack"):
|
||||
await OpenAIResponsesHandler().process_input_messages(
|
||||
data=data, guardrail_to_apply=_guardrail(_FakeGuardrailEndpoint(_block_on_attack_in_texts), max_messages=1)
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_messages_drops_the_texts_of_dropped_anthropic_turns_including_the_system_prompt():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
data = {"model": "claude", "system": "rules", "messages": [_user("hi"), _assistant("hello"), _user("bye")]}
|
||||
|
||||
await AnthropicMessagesHandler().process_input_messages(
|
||||
data=data, guardrail_to_apply=_guardrail(endpoint, max_messages=2)
|
||||
)
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == ["hello", "bye"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_messages_leaves_embedding_inputs_alone():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await OpenAIEmbeddingsHandler().process_input_messages(
|
||||
data={"model": "e", "input": ["ATTACK", "pad"]}, guardrail_to_apply=_guardrail(endpoint, max_messages=1)
|
||||
)
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == ["ATTACK", "pad"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_messages_leaves_rerank_inputs_alone():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
data = {"model": "r", "query": "ATTACK query", "instruction": "rank", "documents": ["a"]}
|
||||
|
||||
await CohereRerankHandler().process_input_messages(
|
||||
data=data, guardrail_to_apply=_guardrail(endpoint, max_messages=1)
|
||||
)
|
||||
|
||||
assert "ATTACK query" in endpoint.payloads[0]["texts"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_messages_leaves_every_choice_of_a_response_alone():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
response = ModelResponse(
|
||||
choices=[
|
||||
Choices(index=0, message=Message(role="assistant", content="LEAKED SECRET")),
|
||||
Choices(index=1, message=Message(role="assistant", content="fine")),
|
||||
]
|
||||
)
|
||||
|
||||
await OpenAIChatCompletionsHandler().process_output_response(
|
||||
response=response, guardrail_to_apply=_guardrail(endpoint, max_messages=1)
|
||||
)
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == ["LEAKED SECRET", "fine"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"options",
|
||||
[{"max_messages": 2}, {"max_text_chars": 12}, {"strip_patterns": [TIMESTAMP]}],
|
||||
ids=["windowed", "truncated", "stripped"],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_mask_of_a_shaped_request_fails_the_call(options):
|
||||
endpoint = _FakeGuardrailEndpoint(_mask_ssn)
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _chat_request(_guardrail(endpoint, **options), _conversation())
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_mask_of_a_shaped_response_fails_the_call():
|
||||
response = ModelResponse(choices=[Choices(index=0, message=Message(role="assistant", content=f"ssn {SSN} ..."))])
|
||||
|
||||
with pytest.raises(GuardrailRaisedException, match="cannot be applied to this response") as raised:
|
||||
await OpenAIChatCompletionsHandler().process_output_response(
|
||||
response=response, guardrail_to_apply=_guardrail(_FakeGuardrailEndpoint(_mask_ssn), max_text_chars=5)
|
||||
)
|
||||
|
||||
assert raised.value.status_code == 400
|
||||
assert raised.value.blocked_content is False
|
||||
|
||||
|
||||
def _long_answer() -> str:
|
||||
return f"your ssn is {SSN}. " + "and more text " * 20
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_echo_of_a_shaped_response_leaves_the_llm_output_whole():
|
||||
response = ModelResponse(choices=[Choices(index=0, message=Message(role="assistant", content=_long_answer()))])
|
||||
|
||||
result = await OpenAIChatCompletionsHandler().process_output_response(
|
||||
response=response, guardrail_to_apply=_guardrail(_FakeGuardrailEndpoint(_echo), max_text_chars=30)
|
||||
)
|
||||
|
||||
assert result.choices[0].message.content == _long_answer()
|
||||
|
||||
|
||||
def _answer_chunks() -> list[ModelResponseStream]:
|
||||
words = _long_answer().split(" ")
|
||||
deltas = [
|
||||
ModelResponseStream(choices=[{"index": 0, "delta": {"role": "assistant", "content": f"{word} "}}])
|
||||
for word in words
|
||||
]
|
||||
return [*deltas, ModelResponseStream(choices=[{"index": 0, "delta": {}, "finish_reason": "stop"}])]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_mask_of_a_shaped_stream_fails_the_stream():
|
||||
with pytest.raises(GuardrailRaisedException, match="cannot be applied to this response"):
|
||||
await OpenAIChatCompletionsHandler().process_output_streaming_response(
|
||||
responses_so_far=_answer_chunks(),
|
||||
guardrail_to_apply=_guardrail(_FakeGuardrailEndpoint(_mask_ssn), max_text_chars=30),
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_echo_of_a_shaped_stream_passes_the_stream_through():
|
||||
chunks = _answer_chunks()
|
||||
|
||||
result = await OpenAIChatCompletionsHandler().process_output_streaming_response(
|
||||
responses_so_far=chunks, guardrail_to_apply=_guardrail(_FakeGuardrailEndpoint(_echo), max_text_chars=30)
|
||||
)
|
||||
|
||||
assert [chunk.choices[0].delta.content for chunk in result] == [chunk.choices[0].delta.content for chunk in chunks]
|
||||
assert "".join(chunk.choices[0].delta.content or "" for chunk in result) == f"{_long_answer()} "
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_echo_of_a_shaped_payload_keeps_the_stream_holdback():
|
||||
endpoint = _FakeGuardrailEndpoint(_answer({"action": "NONE", "texts": ["hel"], "stream_holdback_chars": [2]}))
|
||||
|
||||
result = await _apply(_guardrail(endpoint, max_text_chars=3), {"texts": ["hello"]}, input_type="response")
|
||||
|
||||
assert result == {"texts": ["hello"], "stream_holdback_chars": [2]}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_echo_without_null_fields_is_not_a_rewrite():
|
||||
def echo_without_nulls(payload: dict) -> dict:
|
||||
return json.loads(
|
||||
json.dumps(_echo(payload)), object_hook=lambda obj: {k: v for k, v in obj.items() if v is not None}
|
||||
)
|
||||
|
||||
messages = [
|
||||
_user("hi " * 10),
|
||||
_assistant([{"type": "text", "text": "calling"}, {"type": "tool_use", "id": "t1", "name": "f", "input": {}}]),
|
||||
_user([{"type": "tool_result", "tool_use_id": "t1", "content": "r"}]),
|
||||
]
|
||||
endpoint = _FakeGuardrailEndpoint(echo_without_nulls)
|
||||
|
||||
data = await AnthropicMessagesHandler().process_input_messages(
|
||||
data={"model": "claude", "max_tokens": 5, "messages": copy.deepcopy(messages)},
|
||||
guardrail_to_apply=_guardrail(endpoint, max_text_chars=5),
|
||||
)
|
||||
|
||||
assert endpoint.payloads[0]["structured_messages"][1]["thinking_blocks"] is None
|
||||
assert data["messages"] == messages
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_windowing_combined_with_withheld_images_still_sends_and_blocks():
|
||||
endpoint = _FakeGuardrailEndpoint(_answer({"action": "BLOCKED", "blocked_reason": "no"}))
|
||||
messages = [_user("old"), _multimodal_turn(), _user("new")]
|
||||
|
||||
with pytest.raises(GuardrailRaisedException, match="no"):
|
||||
await _apply(
|
||||
_guardrail(endpoint, max_messages=2, send_images=False, fail_on_error=False),
|
||||
{"texts": ["old", "compare", "these two", "new"], "structured_messages": messages},
|
||||
)
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == ["compare", "these two", "new"]
|
||||
assert IMAGE_URL not in json.dumps(endpoint.payloads[0])
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"respond",
|
||||
[
|
||||
_answer({"action": "GUARDRAIL_INTERVENED"}),
|
||||
_answer({"action": "NONE", "texts": ["[A]"]}),
|
||||
_answer({"action": "NONE", "images": ["data:image/png;base64,OTHER"]}),
|
||||
_answer({"action": "NONE", "tools": [{"type": "function", "function": {"name": "other"}}]}),
|
||||
_answer({"action": "NONE", "structured_messages": [_user("[A]")]}),
|
||||
],
|
||||
ids=["intervened", "texts", "images", "tools", "rows"],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_any_returned_change_to_a_shaped_payload_fails_the_call(respond):
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _apply(
|
||||
_guardrail(_FakeGuardrailEndpoint(respond), max_text_chars=3),
|
||||
{"texts": ["hello"], "structured_messages": [_user("hello")], "images": [IMAGE_URL]},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"options",
|
||||
[{"max_messages": 2}, {"max_text_chars": 12}, {"strip_patterns": [TIMESTAMP]}],
|
||||
ids=["windowed", "truncated", "stripped"],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_an_echo_of_a_shaped_request_passes_the_callers_request_through(options):
|
||||
endpoint = _FakeGuardrailEndpoint(_echo)
|
||||
|
||||
messages = await _chat_request(_guardrail(endpoint, **options), _conversation())
|
||||
|
||||
assert messages == _conversation()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_blocked_still_blocks_a_shaped_request():
|
||||
endpoint = _FakeGuardrailEndpoint(_answer({"action": "BLOCKED", "blocked_reason": "no"}))
|
||||
|
||||
with pytest.raises(GuardrailRaisedException) as raised:
|
||||
await _chat_request(
|
||||
_guardrail(endpoint, max_messages=1, max_text_chars=3, strip_patterns=["ssn"]), _conversation()
|
||||
)
|
||||
|
||||
assert raised.value.blocked_content is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"options",
|
||||
[{"max_messages": 4}, {"max_text_chars": 100}, {"strip_patterns": [r"\[debug\]"]}],
|
||||
ids=["window_covers_all", "nothing_too_long", "nothing_matches"],
|
||||
)
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_mask_applies_when_the_configured_shaping_left_the_payload_untouched(options):
|
||||
endpoint = _FakeGuardrailEndpoint(_mask_ssn)
|
||||
|
||||
messages = await _chat_request(_guardrail(endpoint, **options), _conversation())
|
||||
|
||||
assert messages == [*_conversation()[:-1], _user("my ssn is [SSN]")]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_text_chars_truncates_every_text():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _chat_request(_guardrail(endpoint, max_text_chars=4), [_user("abcdefgh"), _multimodal_turn(), _user("abc")])
|
||||
|
||||
payload = endpoint.payloads[0]
|
||||
assert payload["texts"] == ["abcd", "comp", "thes", "abc"]
|
||||
assert payload["structured_messages"] == [
|
||||
_user("abcd"),
|
||||
_user(
|
||||
[
|
||||
{"type": "text", "text": "comp"},
|
||||
{"type": "image_url", "image_url": {"url": IMAGE_URL}},
|
||||
{"type": "text", "text": "thes"},
|
||||
]
|
||||
),
|
||||
_user("abc"),
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_strip_patterns_remove_matches_from_every_text():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _chat_request(_guardrail(endpoint, strip_patterns=[TIMESTAMP, r" \[debug\]"]), _conversation())
|
||||
|
||||
payload = endpoint.payloads[0]
|
||||
assert payload["texts"] == ["rules ", "hi", "noted ", f"my ssn is {SSN}"]
|
||||
assert [row["content"] for row in payload["structured_messages"]] == payload["texts"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_strip_patterns_touch_only_text_never_roles_ids_tool_calls_or_tools():
|
||||
tools = [{"type": "function", "function": {"name": "SECRET_lookup", "description": "SECRET"}}]
|
||||
tool_call_turn = {
|
||||
"role": "assistant",
|
||||
"content": None,
|
||||
"tool_calls": [
|
||||
{"id": "call_SECRET", "type": "function", "function": {"name": "SECRET", "arguments": '"SECRET"'}}
|
||||
],
|
||||
}
|
||||
messages = [
|
||||
_user("SECRET question"),
|
||||
tool_call_turn,
|
||||
{"role": "tool", "tool_call_id": "call_SECRET", "content": "SECRET answer"},
|
||||
]
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _apply(
|
||||
_guardrail(endpoint, strip_patterns=["SECRET"]),
|
||||
{"texts": ["SECRET question", "SECRET answer"], "structured_messages": messages, "tools": tools},
|
||||
)
|
||||
|
||||
payload = endpoint.payloads[0]
|
||||
assert payload["texts"] == [" question", " answer"]
|
||||
assert payload["tools"] == tools
|
||||
assert payload["structured_messages"] == [
|
||||
_user(" question"),
|
||||
tool_call_turn,
|
||||
{"role": "tool", "tool_call_id": "call_SECRET", "content": " answer"},
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_each_pattern_removes_a_bounded_number_of_matches_per_text():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _apply(_guardrail(endpoint, strip_patterns=["x"]), {"texts": ["x" * (MAX_STRIP_SUBSTITUTIONS + 3)]})
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == ["xxx"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_text_past_the_request_strip_budget_is_sent_unstripped(caplog):
|
||||
first = "<ts>1</ts>" + "a" * (MAX_STRIP_CALL_CHARS - len("<ts>1</ts>") - 20)
|
||||
second = "<ts>2</ts>" + "b" * 10
|
||||
third = "<ts>3</ts>" + "c" * 20
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
await _apply(_guardrail(endpoint, strip_patterns=[TIMESTAMP]), {"texts": [first, second, third]})
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == [first[len("<ts>1</ts>") :], "b" * 10, third]
|
||||
assert any("text-shaping-test" in message and "sent unstripped" in message for message in caplog.messages)
|
||||
|
||||
|
||||
@pytest.mark.timeout(10)
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_catastrophic_pattern_times_out_and_sends_the_text_unstripped(caplog):
|
||||
backtracking = "PREFIX " + "a" * 40 + "!"
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
await _apply(
|
||||
_guardrail(endpoint, strip_patterns=[r"PREFIX ", r"(a|aa)+$"]), {"texts": [backtracking, "PREFIX short"]}
|
||||
)
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == [backtracking, "PREFIX short"]
|
||||
assert any("text-shaping-test" in message and "sent unstripped" in message for message in caplog.messages)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_text_repeated_in_texts_and_rows_is_charged_to_the_strip_budget_once():
|
||||
text = "<ts>1</ts>" + "a" * (MAX_STRIP_CALL_CHARS * 2 // 3)
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _apply(
|
||||
_guardrail(endpoint, strip_patterns=[TIMESTAMP]),
|
||||
{"texts": [text, text, "<ts>2</ts>b"], "structured_messages": [_user(text), _user(text), _user("<ts>2</ts>b")]},
|
||||
)
|
||||
|
||||
stripped = text[len("<ts>1</ts>") :]
|
||||
assert endpoint.payloads[0]["texts"] == [stripped, stripped, "b"]
|
||||
assert endpoint.payloads[0]["structured_messages"] == [_user(stripped), _user(stripped), _user("b")]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_row_rewrite_fails_when_only_text_free_rows_were_windowed_out():
|
||||
endpoint = _FakeGuardrailEndpoint(_mask_ssn)
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _chat_request(_guardrail(endpoint, max_messages=1), [_tool_call_turn(), _user(f"my ssn is {SSN}")])
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == [f"my ssn is {SSN}"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_row_rewrite_fails_when_only_row_text_was_stripped():
|
||||
endpoint = _FakeGuardrailEndpoint(_mask_ssn)
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _apply(
|
||||
_guardrail(endpoint, strip_patterns=[TIMESTAMP]),
|
||||
{"texts": [f"ssn {SSN}"], "structured_messages": [_user(f"<ts>1</ts>ssn {SSN}")]},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_texts_rewrite_fails_when_only_row_text_was_stripped():
|
||||
endpoint = _FakeGuardrailEndpoint(_answer({"action": "NONE", "texts": ["ssn [SSN]"]}))
|
||||
|
||||
with pytest.raises(UnappliableRequestRewrite):
|
||||
await _apply(
|
||||
_guardrail(endpoint, strip_patterns=[TIMESTAMP]),
|
||||
{"texts": [f"ssn {SSN}"], "structured_messages": [_user(f"<ts>1</ts>ssn {SSN}")]},
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_max_messages_keeps_the_handlers_texts_when_no_row_is_dropped():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _apply(_guardrail(endpoint, max_messages=5), {"texts": ["extra", "a"], "structured_messages": [_user("a")]})
|
||||
|
||||
assert endpoint.payloads[0]["texts"] == ["extra", "a"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_combined_options_window_then_strip_then_truncate():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _chat_request(
|
||||
_guardrail(endpoint, max_messages=2, strip_patterns=[TIMESTAMP], max_text_chars=6), _conversation()
|
||||
)
|
||||
|
||||
payload = endpoint.payloads[0]
|
||||
assert payload["texts"] == ["noted ", "my ssn"]
|
||||
assert payload["structured_messages"] == [_assistant("noted "), _user("my ssn")]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("options", "message"),
|
||||
[
|
||||
({"strip_patterns": ["("]}, "strip_patterns contains an invalid regex"),
|
||||
({"strip_patterns": "ssn"}, "strip_patterns must be a list of strings"),
|
||||
({"max_messages": 0}, "max_messages must be >= 1"),
|
||||
({"max_text_chars": 0}, "max_text_chars must be >= 1"),
|
||||
({"max_text_chars": 10.5}, "max_text_chars must be an int"),
|
||||
({"max_messages": True}, "max_messages must be an int"),
|
||||
({"max_messages": "3"}, "max_messages must be an int"),
|
||||
],
|
||||
ids=["invalid_regex", "bare_string", "zero_messages", "zero_chars", "float", "bool", "string"],
|
||||
)
|
||||
def test_invalid_options_raise_at_init(options, message):
|
||||
with pytest.raises(ValueError, match=message):
|
||||
_guardrail(_FakeGuardrailEndpoint(), **options)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("options", "named"),
|
||||
[
|
||||
({"max_messages": 3}, "max_messages=3"),
|
||||
({"max_text_chars": 100}, "max_text_chars=100"),
|
||||
({"strip_patterns": [TIMESTAMP]}, "strip_patterns"),
|
||||
],
|
||||
ids=["max_messages", "max_text_chars", "strip_patterns"],
|
||||
)
|
||||
def test_text_shaping_options_warn_that_the_guardrail_only_enforces_on_what_it_is_sent(caplog, options, named):
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
_guardrail(_FakeGuardrailEndpoint(), **options)
|
||||
|
||||
assert any("can only enforce on what it is sent" in message and named in message for message in caplog.messages)
|
||||
|
||||
|
||||
def test_unset_text_shaping_options_do_not_warn(caplog):
|
||||
with caplog.at_level(logging.WARNING, logger="LiteLLM Proxy"):
|
||||
_guardrail(_FakeGuardrailEndpoint(), max_messages=None, max_text_chars=None, strip_patterns=[])
|
||||
|
||||
assert not any("can only enforce on what it is sent" in message for message in caplog.messages)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_absent_texts_stay_absent_when_the_rows_are_windowed():
|
||||
endpoint = _FakeGuardrailEndpoint()
|
||||
|
||||
await _apply(
|
||||
_guardrail(endpoint, max_messages=1, max_text_chars=2), {"texts": None, "structured_messages": _conversation()}
|
||||
)
|
||||
|
||||
assert endpoint.payloads[0]["texts"] is None
|
||||
assert endpoint.payloads[0]["structured_messages"] == [_user("my")]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_shaping_never_mutates_the_callers_inputs():
|
||||
messages = [_user("old <ts>1</ts>"), _multimodal_turn()]
|
||||
texts = ["old <ts>1</ts>", "compare", "these two"]
|
||||
snapshot = copy.deepcopy((messages, texts))
|
||||
|
||||
await _apply(
|
||||
_guardrail(_FakeGuardrailEndpoint(), max_messages=1, max_text_chars=3, strip_patterns=[TIMESTAMP]),
|
||||
{"texts": texts, "structured_messages": messages},
|
||||
)
|
||||
|
||||
assert (messages, texts) == snapshot
|
||||
2
uv.lock
generated
2
uv.lock
generated
|
|
@ -4519,6 +4519,7 @@ dependencies = [
|
|||
{ name = "pydantic-settings" },
|
||||
{ name = "python-dotenv" },
|
||||
{ name = "pyyaml" },
|
||||
{ name = "regex" },
|
||||
{ name = "tiktoken" },
|
||||
{ name = "tokenizers" },
|
||||
]
|
||||
|
|
@ -4835,6 +4836,7 @@ requires-dist = [
|
|||
{ name = "pyyaml", marker = "extra == 'cli'", specifier = ">=6.0.3,<7.0" },
|
||||
{ name = "pyyaml", marker = "extra == 'proxy'", specifier = ">=6.0.3,<7.0" },
|
||||
{ name = "redisvl", marker = "extra == 'extra-proxy'", specifier = ">=0.4.1,<1.0" },
|
||||
{ name = "regex", specifier = ">=2022.1.18" },
|
||||
{ name = "requests", marker = "extra == 'cli'", specifier = ">=2.32.0,<3.0" },
|
||||
{ name = "resend", marker = "extra == 'extra-proxy'", specifier = ">=2.23.0,<3.0" },
|
||||
{ name = "restrictedpython", marker = "extra == 'proxy'", specifier = ">=8.5,<9.0" },
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue