feat(guardrails): add fire_and_forget dispatch to generic_guardrail_api

litellm/integrations/custom_guardrail.py

@@ -1153,7 +1153,7 @@ def _event_hook_is_event_type(self, event_type: GuardrailEventHooks) -> bool:
             return False
         return self.event_hook == event_type.value
 
-    def get_guardrail_dynamic_request_body_params(self, request_data: dict) -> dict:
+    def get_guardrail_dynamic_request_body_params(self, request_data: dict) -> dict[str, object]:
         """
         Returns `extra_body` to be added to the request body for the Guardrail API call
 

litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/__init__.py

@@ -40,6 +40,8 @@ def initialize_guardrail(litellm_params: "LitellmParams", guardrail: "Guardrail"
         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"),
         timeout=litellm_params.timeout,
+        fire_and_forget=_get_config_value(litellm_params, optional_params, "fire_and_forget"),
+        fire_and_forget_max_inflight=_get_config_value(litellm_params, optional_params, "fire_and_forget_max_inflight"),
     )
 
     litellm.logging_callback_manager.add_litellm_callback(_generic_guardrail_api_callback)

litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/background_dispatch.py

@@ -0,0 +1,112 @@
+import asyncio
+import contextvars
+from collections.abc import Awaitable, Callable
+from typing import Annotated, Final
+
+from pydantic import Field, TypeAdapter, ValidationError
+
+from litellm._logging import verbose_proxy_logger
+
+DEFAULT_FIRE_AND_FORGET_MAX_INFLIGHT: Final = 100
+
+FIRE_AND_FORGET_POST_TIMEOUT_SECONDS: Final = 30.0
+
+FIRE_AND_FORGET_DISPATCHED_REASON: Final = "fire_and_forget dispatched, verdict not read"
+
+FIRE_AND_FORGET_DROPPED_REASON: Final = "fire_and_forget_max_inflight reached, call dropped"
+
+FIRE_AND_FORGET_NOT_DISPATCHED_REASON: Final = "fire_and_forget payload could not be built, call not dispatched"
+
+_DROP_LOG_INTERVAL: Final = 100
+_FIRE_AND_FORGET_ADAPTER: Final[TypeAdapter[bool]] = TypeAdapter(bool)
+_MAX_INFLIGHT_ADAPTER: Final[TypeAdapter[int]] = TypeAdapter(Annotated[int, Field(ge=1)])
+
+
+def fire_and_forget_from_config(value: object) -> bool:
+    if value is None:
+        return False
+    try:
+        return _FIRE_AND_FORGET_ADAPTER.validate_python(value)
+    except ValidationError:
+        verbose_proxy_logger.warning(
+            "Ignoring fire_and_forget=%r, expected true or false. Awaiting every guardrail call", value
+        )
+        return False
+
+
+def _parsed_max_inflight(value: object) -> int | None:
+    if isinstance(value, bool):
+        return None
+    try:
+        return _MAX_INFLIGHT_ADAPTER.validate_python(value)
+    except ValidationError:
+        return None
+
+
+def max_inflight_from_config(value: object) -> int:
+    if value is None:
+        return DEFAULT_FIRE_AND_FORGET_MAX_INFLIGHT
+    parsed: Final = _parsed_max_inflight(value)
+    if parsed is None:
+        verbose_proxy_logger.warning(
+            "Ignoring fire_and_forget_max_inflight=%r, expected an integer of at least 1. Using %d",
+            value,
+            DEFAULT_FIRE_AND_FORGET_MAX_INFLIGHT,
+        )
+        return DEFAULT_FIRE_AND_FORGET_MAX_INFLIGHT
+    return parsed
+
+
+class BackgroundDispatcher:
+    def __init__(self, *, guardrail_name: str | None, max_inflight: int) -> None:
+        if max_inflight < 1:
+            raise ValueError(f"fire_and_forget_max_inflight must be >= 1 (got {max_inflight})")
+        self._guardrail_name: Final = guardrail_name
+        self._max_inflight: Final = max_inflight
+        self._pending: Final[set[asyncio.Task[None]]] = set()  # mutable-ok: strong refs, asyncio keeps only weak ones
+        self._dropped: int = 0
+
+    @property
+    def pending_count(self) -> int:
+        return len(self._pending)
+
+    @property
+    def dropped_count(self) -> int:
+        return self._dropped
+
+    def dispatch(self, prepare: Callable[[], Callable[[], Awaitable[object]]], *, context: str) -> bool:
+        if len(self._pending) >= self._max_inflight:
+            self._dropped += 1
+            if self._dropped % _DROP_LOG_INTERVAL == 1:
+                verbose_proxy_logger.warning(
+                    "Generic Guardrail API (%s, fire_and_forget): dropped %d call(s) so far, "
+                    "%d already in flight (fire_and_forget_max_inflight=%d). %s",
+                    self._guardrail_name,
+                    self._dropped,
+                    len(self._pending),
+                    self._max_inflight,
+                    context,
+                )
+            return False
+
+        run: Final = prepare()
+        task: Final = contextvars.Context().run(asyncio.create_task, self._run_logging_failures(run, context=context))
+        self._pending.add(task)
+        task.add_done_callback(self._pending.discard)
+        return True
+
+    async def _run_logging_failures(self, run: Callable[[], Awaitable[object]], *, context: str) -> None:
+        try:
+            await run()
+        except Exception as e:  # noqa: BLE001  # a detached task has no caller to raise into
+            verbose_proxy_logger.warning(
+                "Generic Guardrail API (%s, fire_and_forget) call failed. %s: %s",
+                self._guardrail_name,
+                context,
+                e,
+            )
+
+    async def wait_for_pending(self) -> None:
+        pending: Final = tuple(self._pending)
+        if pending:
+            await asyncio.gather(*pending, return_exceptions=True)

litellm/proxy/guardrails/guardrail_hooks/generic_guardrail_api/generic_guardrail_api.py

@@ -7,7 +7,7 @@
 
 import fnmatch
 import os
-from collections.abc import Mapping, Sequence
+from collections.abc import Awaitable, Callable, Mapping, Sequence
 from typing import TYPE_CHECKING, Any, Final, Literal, Optional
 
 import httpx
@@ -20,9 +20,19 @@
     log_guardrail_information,
 )
 from litellm.llms.custom_httpx.http_handler import (
+    AsyncHTTPHandler,
     get_async_httpx_client,
     httpxSpecialProvider,
 )
+from litellm.proxy.guardrails.guardrail_hooks.generic_guardrail_api.background_dispatch import (
+    FIRE_AND_FORGET_DISPATCHED_REASON,
+    FIRE_AND_FORGET_DROPPED_REASON,
+    FIRE_AND_FORGET_NOT_DISPATCHED_REASON,
+    FIRE_AND_FORGET_POST_TIMEOUT_SECONDS,
+    BackgroundDispatcher,
+    fire_and_forget_from_config,
+    max_inflight_from_config,
+)
 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 (
@@ -170,6 +180,15 @@ def _structured_rows_to_write_back(
     )
 
 
+def _passthrough_inputs(inputs: GenericGuardrailAPIInputs) -> GenericGuardrailAPIInputs:
+    return GenericGuardrailAPIInputs(**inputs)
+
+
+def _call_context(input_type: Literal["request", "response"], logging_obj: Optional["LiteLLMLoggingObj"]) -> str:
+    call_id: Final = getattr(logging_obj, "litellm_call_id", None) if logging_obj else None
+    return f"input_type={input_type} litellm_call_id={call_id}"
+
+
 class GenericGuardrailAPI(CustomGuardrail):
     """
     Generic Guardrail API integration for LiteLLM.
@@ -204,9 +223,15 @@ def __init__(
         streaming_end_of_stream_only: bool | None = None,
         streaming_sampling_rate: int | None = None,
         streaming_transform_mode: Literal["block_only", "incremental_diff"] | None = None,
+        fire_and_forget: bool | None = None,
+        fire_and_forget_max_inflight: int | None = None,
+        async_handler: AsyncHTTPHandler | None = None,
+        dispatcher: BackgroundDispatcher | 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 []
 
@@ -235,8 +260,11 @@ def __init__(
 
         self.fail_on_error: bool = True if fail_on_error is None else fail_on_error
 
+        self.fire_and_forget: bool = fire_and_forget_from_config(fire_and_forget)
+
         # Read by UnifiedLLMGuardrails.async_post_call_streaming_iterator_hook
         # via getattr(guardrail_to_apply, "streaming_*", default).
+        self.streaming_observe_only: bool = self.fire_and_forget
         self.streaming_end_of_stream_only: bool = (
             False if streaming_end_of_stream_only is None else streaming_end_of_stream_only
         )
@@ -256,6 +284,22 @@ def __init__(
 
         super().__init__(**kwargs)
 
+        self._dispatcher: Final = dispatcher or BackgroundDispatcher(
+            guardrail_name=self.guardrail_name,
+            max_inflight=max_inflight_from_config(fire_and_forget_max_inflight),
+        )
+
+        if self.fire_and_forget:
+            verbose_proxy_logger.warning(
+                "Generic Guardrail API (%s): fire_and_forget=True makes this guardrail observe-only. "
+                "action=BLOCKED and action=GUARDRAIL_INTERVENED are ignored, fail_on_error=%s and "
+                "unreachable_fallback=%s cannot block the request, and a stream is checked once when it "
+                "closes, with whatever reached the client.",
+                self.guardrail_name,
+                self.fail_on_error,
+                self.unreachable_fallback,
+            )
+
         verbose_proxy_logger.debug("Generic Guardrail API initialized with api_base: %s", self.api_base)
 
     def _extract_user_api_key_metadata(self, request_data: dict) -> GenericGuardrailAPIMetadata:
@@ -339,7 +383,7 @@ def _build_request_headers(self) -> dict:
     def _build_guardrail_return_inputs(
         self,
         *,
-        texts: list,
+        texts: list[str],
         images: list[str] | None,
         tools: list[ChatCompletionToolParam] | None,
         structured_messages: Sequence[AllMessageValues] | None,
@@ -377,6 +421,8 @@ def _handle_guardrail_request_error(
         logging_obj: Optional["LiteLLMLoggingObj"],
         is_unreachable: bool = True,
     ) -> GenericGuardrailAPIInputs:
+        if self.fire_and_forget:
+            raise error
         unreachable_fail_open: Final = is_unreachable and self.unreachable_fallback == "fail_open"
         if unreachable_fail_open or not self.fail_on_error:
             http_status_code: Final = getattr(getattr(error, "response", None), "status_code", None)
@@ -390,6 +436,26 @@ def _handle_guardrail_request_error(
         verbose_proxy_logger.error("Generic Guardrail API: failed to make request: %s", str(error))
         raise Exception(f"Generic Guardrail API failed: {error}")
 
+    def _dispatch_background_post(
+        self,
+        *,
+        guardrail_request: GenericGuardrailAPIRequest,
+        input_type: Literal["request", "response"],
+        logging_obj: Optional["LiteLLMLoggingObj"],
+    ) -> bool:
+        timeout: Final = FIRE_AND_FORGET_POST_TIMEOUT_SECONDS if self.timeout is None else self.timeout
+
+        def _prepare() -> Callable[[], Awaitable[None]]:
+            payload: Final = guardrail_request.model_dump(mode="json")
+            headers: Final = self._build_request_headers()
+
+            async def _post() -> None:
+                await self.async_handler.post(url=self.api_base, json=payload, headers=headers, timeout=timeout)
+
+            return _post
+
+        return self._dispatcher.dispatch(_prepare, context=_call_context(input_type, logging_obj))
+
     @log_guardrail_information
     async def apply_guardrail(
         self,
@@ -418,6 +484,31 @@ async def apply_guardrail(
         Raises:
             Exception: If the guardrail blocks the request
         """
+        if not self.fire_and_forget:
+            return await self._apply_guardrail(inputs, request_data, input_type, logging_obj)
+        try:
+            return await self._apply_guardrail(inputs, request_data, input_type, logging_obj)
+        except Exception as e:  # noqa: BLE001  # an observe-only guardrail must never fail the request
+            verbose_proxy_logger.warning(
+                "Generic Guardrail API (%s, fire_and_forget) call not dispatched. %s: %s",
+                self.guardrail_name,
+                _call_context(input_type, logging_obj),
+                e,
+            )
+            self.add_standard_logging_guardrail_information_to_request_data(
+                guardrail_json_response=FIRE_AND_FORGET_NOT_DISPATCHED_REASON,
+                request_data=request_data,
+                guardrail_status="not_run",
+            )
+            return _passthrough_inputs(inputs)
+
+    async def _apply_guardrail(
+        self,
+        inputs: GenericGuardrailAPIInputs,
+        request_data: dict,
+        input_type: Literal["request", "response"],
+        logging_obj: Optional["LiteLLMLoggingObj"],
+    ) -> GenericGuardrailAPIInputs:
         verbose_proxy_logger.debug("Generic Guardrail API: Applying guardrail to text")
 
         # Extract texts and images from inputs
@@ -469,15 +560,24 @@ async def apply_guardrail(
                 model=model,
             )
 
-            headers: Final = self._build_request_headers()
+            if self.fire_and_forget:
+                dispatched: Final = self._dispatch_background_post(
+                    guardrail_request=guardrail_request, input_type=input_type, logging_obj=logging_obj
+                )
+                self.add_standard_logging_guardrail_information_to_request_data(
+                    guardrail_json_response=(
+                        FIRE_AND_FORGET_DISPATCHED_REASON if dispatched else FIRE_AND_FORGET_DROPPED_REASON
+                    ),
+                    request_data=request_data,
+                    guardrail_status="success" if dispatched else "not_run",
+                )
+                return _passthrough_inputs(inputs)
 
-            # Make the API request
+            headers: Final = self._build_request_headers()
             # Use mode="json" to ensure all iterables are converted to lists
+            payload: Final = guardrail_request.model_dump(mode="json")
             response: Final = await self.async_handler.post(
-                url=self.api_base,
-                json=guardrail_request.model_dump(mode="json"),
-                headers=headers,
-                timeout=self.timeout,
+                url=self.api_base, json=payload, headers=headers, timeout=self.timeout
             )
 
             response.raise_for_status()