From 62ecb11ab9d87963c18c5da8aa3233352412a3f3 Mon Sep 17 00:00:00 2001 From: yassin Date: Wed, 16 Sep 2026 09:59:53 +0000 Subject: [PATCH 1/6] perf(content_filter): scan a bounded window per streamed chunk The streaming post-call hook rescanned the whole accumulated choice buffer on every chunk, so scan cost grew quadratically with output length. Keep a bounded per-choice buffer instead: once it exceeds twice the scan context, drop the head when masking the head and tail separately yields the same output as masking the whole buffer, so no pattern, phrase or exception straddles the cut. Detections from the dropped head are kept and merged, deduplicated, into the final log row Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/constants.py | 2 + .../litellm_content_filter/content_filter.py | 118 +++++++----- .../content_filter/test_content_filter.py | 177 ++++++++++++++++++ 3 files changed, 254 insertions(+), 43 deletions(-) diff --git a/litellm/constants.py b/litellm/constants.py index 745a4d9294e..ce5b65080ee 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -364,6 +364,8 @@ GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS: Final = int( os.getenv("GUARDRAIL_SCANNED_MESSAGES_CACHE_TTL_SECONDS", 24 * 60 * 60) ) BEDROCK_APPLY_GUARDRAIL_CHUNK_BUDGET_CHARS: Final = 25_000 +CONTENT_FILTER_STREAMING_HOLDBACK_CHARS: Final = 50 +CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS: Final = 512 DEFAULT_PRESIDIO_ANALYZE_CHUNK_SIZE_BYTES: Final = 500_000 PRESIDIO_ANALYZE_CHUNK_OVERLAP_CHARS: Final = 4096 PRESIDIO_ANALYZE_CHUNK_CONCURRENCY: Final = 8 diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py index 1e684c514de..1ffe200169a 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py @@ -11,6 +11,7 @@ import os import re import time from collections.abc import AsyncGenerator, Coroutine, Mapping, Sequence +from dataclasses import dataclass, replace from datetime import datetime from re import Pattern from typing import TYPE_CHECKING, Any, Final, Literal, Optional, TypedDict, cast @@ -20,7 +21,11 @@ from fastapi import HTTPException from litellm import Router from litellm._logging import verbose_proxy_logger -from litellm.constants import DEFAULT_MAX_RECURSE_DEPTH +from litellm.constants import ( + CONTENT_FILTER_STREAMING_HOLDBACK_CHARS, + CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS, + DEFAULT_MAX_RECURSE_DEPTH, +) from litellm.integrations.custom_guardrail import CustomGuardrail from litellm.proxy._types import UserAPIKeyAuth from litellm.types.utils import ( @@ -112,6 +117,14 @@ class _CategoryConfigView(TypedDict): category_file: str | None +@dataclass(frozen=True, slots=True) +class _StreamedChoiceState: + buffered_text: str = "" + yielded_masked_text_len: int = 0 + committed_detections: tuple[ContentFilterDetection, ...] = () + latest_detections: tuple[ContentFilterDetection, ...] = () + + class CategoryFileData(TypedDict, total=False): category_name: str description: str @@ -1950,6 +1963,40 @@ class ContentFilterGuardrail(CustomGuardrail): exception_str=exception_str, ) + def _trim_streamed_choice_buffer(self, state: _StreamedChoiceState, masked_text: str) -> _StreamedChoiceState: + """ + Bound the per-choice buffer rescanned on every streamed chunk. + + Once the buffer exceeds twice the scan context, drop everything but the last + context-sized tail, provided the two halves mask to the same output as the whole + (so no match, phrase or exception straddles the cut) and the dropped prefix has + already been yielded. Otherwise keep the buffer and retry on the next chunk. + + Detections found in the dropped prefix move to the state's committed detections. + """ + if len(state.buffered_text) <= 2 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS: + return state + head: Final = state.buffered_text[:-CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS] + tail: Final = state.buffered_text[-CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS:] + head_detections: Final[list[ContentFilterDetection]] = [] # mutable-ok: filled by _filter_single_text + try: + masked_head: Final = self._filter_single_text(head, detections=head_detections) + masked_tail: Final = self._filter_single_text(tail) + except Exception: + return state + if masked_head + masked_tail != masked_text or len(masked_head) > state.yielded_masked_text_len: + return state + return replace( + state, + buffered_text=tail, + yielded_masked_text_len=state.yielded_masked_text_len - len(masked_head), + committed_detections=state.committed_detections + tuple(head_detections), + ) + + @staticmethod + def _merge_detections(detections: Sequence[ContentFilterDetection]) -> tuple[ContentFilterDetection, ...]: + return tuple(detection for index, detection in enumerate(detections) if detection not in detections[:index]) + async def async_post_call_streaming_iterator_hook( self, user_api_key_dict: UserAPIKeyAuth, @@ -1968,10 +2015,7 @@ class ContentFilterGuardrail(CustomGuardrail): and the UI Request Lifecycle panel. Mirrors apply_guardrail's finally-block contract. """ - accumulated_text_by_choice: Final[dict[int, str]] = {} - yielded_masked_text_len_by_choice: Final[dict[int, int]] = {} - latest_detections_by_choice: Final[dict[int, list[ContentFilterDetection]]] = {} - buffer_size: Final = 50 # Increased buffer to catch patterns split across many chunks + state_by_choice: Final[dict[int, _StreamedChoiceState]] = {} start_time: Final = datetime.now() scan_seconds: float = 0.0 # rebind-ok: accumulates per-chunk scan time across the stream @@ -1997,69 +2041,57 @@ class ContentFilterGuardrail(CustomGuardrail): content = getattr(choice.delta, "content", None) is_final = bool(getattr(choice, "finish_reason", None)) - if isinstance(content, str) and content: - accumulated_text_by_choice[choice_index] = ( - accumulated_text_by_choice.get(choice_index, "") + content - ) - elif not is_final: + new_content = content if isinstance(content, str) else "" + if not new_content and not is_final: continue - text_to_check = accumulated_text_by_choice.get(choice_index, "") - if not text_to_check: + previous_state = state_by_choice.get(choice_index, _StreamedChoiceState()) + buffered_text = previous_state.buffered_text + new_content + if not buffered_text: continue # Add a space at the end if it's the final chunk to trigger word boundaries (\b) - text_to_scan = text_to_check + (" " if is_final else "") + text_to_scan = buffered_text + (" " if is_final else "") choice_detections: list[ContentFilterDetection] = [] scan_started = time.perf_counter() try: - # _filter_single_text scans the whole accumulated - # choice buffer every chunk, so previous-chunk - # matches are guaranteed to be re-found. Keeping - # only each choice's latest scan avoids duplicate - # detections in the final log row. masked_text = self._filter_single_text(text_to_scan, detections=choice_detections) if is_final and masked_text.endswith(" "): masked_text = masked_text[:-1] - latest_detections_by_choice[choice_index] = choice_detections + latest_detections = tuple(choice_detections) except HTTPException: - latest_detections_by_choice[choice_index] = choice_detections + state_by_choice[choice_index] = replace( + previous_state, latest_detections=tuple(choice_detections) + ) raise except Exception as e: verbose_proxy_logger.error("ContentFilterGuardrail: Error in masking: %s", e) masked_text = text_to_scan # Fallback to current text + latest_detections = previous_state.latest_detections finally: scan_seconds += time.perf_counter() - scan_started - # Determine how much can be safely yielded + safe_to_yield_len = max( + previous_state.yielded_masked_text_len, + len(masked_text) - (0 if is_final else CONTENT_FILTER_STREAMING_HOLDBACK_CHARS), + ) + choice.delta.content = masked_text[previous_state.yielded_masked_text_len : safe_to_yield_len] + next_state = _StreamedChoiceState( + buffered_text, safe_to_yield_len, previous_state.committed_detections, latest_detections + ) if is_final: - safe_to_yield_len = len(masked_text) - else: - safe_to_yield_len = max(0, len(masked_text) - buffer_size) + state_by_choice[choice_index] = next_state + continue - yielded_masked_text_len = yielded_masked_text_len_by_choice.get(choice_index, 0) - if safe_to_yield_len > yielded_masked_text_len: - new_masked_content = masked_text[yielded_masked_text_len:safe_to_yield_len] - choice.delta.content = new_masked_content - yielded_masked_text_len_by_choice[choice_index] = safe_to_yield_len - else: - # Hold content by yielding empty content on this choice - # while preserving chunk metadata and other choices. - choice.delta.content = "" + trim_started = time.perf_counter() + state_by_choice[choice_index] = self._trim_streamed_choice_buffer(next_state, masked_text) + scan_seconds += time.perf_counter() - trim_started yield item else: # Not a ModelResponseStream or no choices - yield as is yield item - - # Any remaining content (should have been handled by is_final, but just in case) - if any( - yielded_masked_text_len_by_choice.get(choice_index, 0) < len(accumulated_text) - for choice_index, accumulated_text in accumulated_text_by_choice.items() - ): - # We already reached the end of the generator - pass except HTTPException: status = "guardrail_intervened" raise @@ -2070,8 +2102,8 @@ class ContentFilterGuardrail(CustomGuardrail): finally: detections = [ detection - for choice_detections in latest_detections_by_choice.values() - for detection in choice_detections + for state in state_by_choice.values() + for detection in self._merge_detections((*state.committed_detections, *state.latest_detections)) ] self._count_masked_entities(detections, masked_entity_count) self._log_guardrail_information( diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py index 130b0da000b..71f46ca3047 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py @@ -11,6 +11,10 @@ import pytest from fastapi import HTTPException +from litellm.constants import ( + CONTENT_FILTER_STREAMING_HOLDBACK_CHARS, + CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS, +) from litellm.proxy.guardrails.guardrail_hooks.litellm_content_filter.content_filter import ( ContentFilterGuardrail, ) @@ -22,6 +26,7 @@ from litellm.types.guardrails import ( ) from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter import ( ContentFilterCategoryConfig, + ContentFilterDetection, ) @@ -900,6 +905,178 @@ class TestContentFilterGuardrail: # masked_entity_count for email is the real count, not N×. assert entry["masked_entity_count"].get("email") == 1 + @staticmethod + async def _collect_streamed_text( + guardrail: ContentFilterGuardrail, chunks: list[str], request_data: dict + ) -> str: + from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices + + async def mock_stream(): + for i, content in enumerate(chunks): + yield ModelResponseStream( + id=f"c{i}", + choices=[StreamingChoices(delta=Delta(content=content), index=0)], + model="gpt-4", + ) + yield ModelResponseStream( + id="final", + choices=[ + StreamingChoices( + delta=Delta(content=""), index=0, finish_reason="stop" + ) + ], + model="gpt-4", + ) + + yielded = [] + async for chunk in guardrail.async_post_call_streaming_iterator_hook( + user_api_key_dict=MagicMock(), + response=mock_stream(), + request_data=request_data, + ): + yielded.append(chunk.choices[0].delta.content or "") + return "".join(yielded) + + @pytest.mark.asyncio + async def test_streaming_hook_scans_bounded_window_per_chunk(self): + """ + Regression: the streaming hook used to re-scan the whole accumulated + buffer on every chunk, so scan work grew quadratically with the length + of the response. Each scan must now cover only the new chunk plus a + bounded tail of what came before, without dropping any output. + """ + scanned_lengths: list[int] = [] + + class RecordingGuardrail(ContentFilterGuardrail): + def _filter_single_text( + self, + text: str, + detections: list[ContentFilterDetection] | None = None, + ) -> str: + scanned_lengths.append(len(text)) + return super()._filter_single_text(text, detections=detections) + + guardrail = RecordingGuardrail( + guardrail_name="test-streaming-bounded-scan", + patterns=[ + ContentFilterPattern( + pattern_type="prebuilt", + pattern_name="email", + action=ContentFilterAction.MASK, + ) + ], + event_hook=GuardrailEventHooks.post_call, + ) + chunk = "Item: a plain household object description. " + chunks = [chunk] * 200 + request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + + streamed = await self._collect_streamed_text(guardrail, chunks, request_data) + + assert streamed == chunk * 200 + assert len(chunk) * 200 > 4 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + window_bound = 2 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + len(chunk) + 1 + assert max(scanned_lengths) <= window_bound, ( + f"scan input grew to {max(scanned_lengths)} chars for a " + f"{len(chunk)}-char chunk; expected at most {window_bound}" + ) + + @pytest.mark.asyncio + async def test_streaming_hook_blocks_match_longer_than_holdback_across_chunks( + self, + ): + """ + A blocked phrase longer than the holdback window arrives in small chunks, + so its start has already been yielded before its end shows up. The scan + still has to see the whole phrase and block. + """ + phrase = "alpha bravo charlie delta echo foxtrot golf hotel india juliet kilo lima" + assert len(phrase) > CONTENT_FILTER_STREAMING_HOLDBACK_CHARS + guardrail = ContentFilterGuardrail( + guardrail_name="test-streaming-long-block", + blocked_words=[BlockedWord(keyword=phrase, action=ContentFilterAction.BLOCK)], + event_hook=GuardrailEventHooks.post_call, + ) + text = "Here is the codeword list: " + phrase + " and that is all." + chunks = [text[i : i + 4] for i in range(0, len(text), 4)] + request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + + with pytest.raises(HTTPException) as exc_info: + await self._collect_streamed_text(guardrail, chunks, request_data) + + assert exc_info.value.detail["keyword"] == phrase + entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + assert entry["guardrail_status"] == "guardrail_intervened" + assert [d["keyword"] for d in entry["guardrail_response"]] == [phrase] + + @pytest.mark.asyncio + async def test_streaming_hook_masks_every_email_in_long_stream_and_logs_once( + self, + ): + """ + A response made of nothing but emails, several times longer than the + rescanned buffer, must come out as nothing but redaction tags, and the log + must carry one email detection, matching what a single scan of the full + text reports. Wherever the buffer is cut, an email sits on the cut, so + dropping text without checking that the cut leaves the masked output + unchanged corrupts the stream. + """ + guardrail = ContentFilterGuardrail( + guardrail_name="test-streaming-many-emails", + patterns=[ + ContentFilterPattern( + pattern_type="prebuilt", + pattern_name="email", + action=ContentFilterAction.MASK, + ) + ], + event_hook=GuardrailEventHooks.post_call, + ) + emails = [f"user{i:03d}@example.com" for i in range(200)] + text = " ".join(emails) + assert len(text) > 4 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + chunks = [text[i : i + 3] for i in range(0, len(text), 3)] + request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + + streamed = await self._collect_streamed_text(guardrail, chunks, request_data) + + assert streamed == " ".join(["[EMAIL_REDACTED]"] * len(emails)) + entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + assert entry["guardrail_status"] == "success" + assert [d["pattern_name"] for d in entry["guardrail_response"]] == ["email"] + assert entry["masked_entity_count"] == {"email": 1} + + @pytest.mark.asyncio + async def test_streaming_hook_logs_detection_masked_long_before_stream_end(self): + """ + An email at the start of a long response is masked and then falls out of + the rescanned buffer well before the stream ends. The final log entry must + still report it, as a scan of the full text would. + """ + guardrail = ContentFilterGuardrail( + guardrail_name="test-streaming-early-detection", + patterns=[ + ContentFilterPattern( + pattern_type="prebuilt", + pattern_name="email", + action=ContentFilterAction.MASK, + ) + ], + event_hook=GuardrailEventHooks.post_call, + ) + filler = "filler text " * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + text = f"Contact one@example.com for details. {filler}" + chunks = [text[i : i + 40] for i in range(0, len(text), 40)] + request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + + streamed = await self._collect_streamed_text(guardrail, chunks, request_data) + + assert streamed == text.replace("one@example.com", "[EMAIL_REDACTED]") + entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + assert entry["guardrail_status"] == "success" + assert [d["pattern_name"] for d in entry["guardrail_response"]] == ["email"] + assert entry["masked_entity_count"] == {"email": 1} + def test_init_with_plain_dicts(self): """ Test initialization with plain dicts (DB format). From 52f06906fe78d6c9848ac71c2c2bead83aaa5da6 Mon Sep 17 00:00:00 2001 From: yassin Date: Wed, 16 Sep 2026 10:58:30 +0000 Subject: [PATCH 2/6] fix(content_filter): widen the streamed scan tail to the longest configured keyword Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../litellm_content_filter/content_filter.py | 23 ++++++++++++---- .../content_filter/test_content_filter.py | 27 +++++++++++++++++++ 2 files changed, 45 insertions(+), 5 deletions(-) diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py index 1ffe200169a..a210762d1df 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py @@ -1963,7 +1963,17 @@ class ContentFilterGuardrail(CustomGuardrail): exception_str=exception_str, ) - def _trim_streamed_choice_buffer(self, state: _StreamedChoiceState, masked_text: str) -> _StreamedChoiceState: + def _streamed_scan_context_chars(self) -> int: + """Retained tail length: the default context, widened to the longest configured keyword.""" + longest_keyword: Final = max( + map(len, (*self.blocked_words, *self.category_keywords, *self.always_block_category_keywords)), + default=0, + ) + return max(CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS, longest_keyword) + + def _trim_streamed_choice_buffer( + self, state: _StreamedChoiceState, masked_text: str, scan_context_chars: int + ) -> _StreamedChoiceState: """ Bound the per-choice buffer rescanned on every streamed chunk. @@ -1974,10 +1984,10 @@ class ContentFilterGuardrail(CustomGuardrail): Detections found in the dropped prefix move to the state's committed detections. """ - if len(state.buffered_text) <= 2 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS: + if len(state.buffered_text) <= 2 * scan_context_chars: return state - head: Final = state.buffered_text[:-CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS] - tail: Final = state.buffered_text[-CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS:] + head: Final = state.buffered_text[:-scan_context_chars] + tail: Final = state.buffered_text[-scan_context_chars:] head_detections: Final[list[ContentFilterDetection]] = [] # mutable-ok: filled by _filter_single_text try: masked_head: Final = self._filter_single_text(head, detections=head_detections) @@ -2016,6 +2026,7 @@ class ContentFilterGuardrail(CustomGuardrail): contract. """ state_by_choice: Final[dict[int, _StreamedChoiceState]] = {} + scan_context_chars: Final = self._streamed_scan_context_chars() start_time: Final = datetime.now() scan_seconds: float = 0.0 # rebind-ok: accumulates per-chunk scan time across the stream @@ -2085,7 +2096,9 @@ class ContentFilterGuardrail(CustomGuardrail): continue trim_started = time.perf_counter() - state_by_choice[choice_index] = self._trim_streamed_choice_buffer(next_state, masked_text) + state_by_choice[choice_index] = self._trim_streamed_choice_buffer( + next_state, masked_text, scan_context_chars + ) scan_seconds += time.perf_counter() - trim_started yield item diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py index 71f46ca3047..b91abb98dbf 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py @@ -1009,6 +1009,33 @@ class TestContentFilterGuardrail: assert entry["guardrail_status"] == "guardrail_intervened" assert [d["keyword"] for d in entry["guardrail_response"]] == [phrase] + @pytest.mark.asyncio + async def test_streaming_hook_blocks_keyword_longer_than_scan_context(self): + """ + A blocked keyword longer than the default retained context arrives after + enough text that the buffer has already been trimmed at least once. The + retained tail must be wide enough that the keyword's start is still in the + buffer when its end arrives, so the stream is blocked. + """ + phrase = " ".join(f"token{i:03d}" for i in range(80)) + assert len(phrase) > CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + guardrail = ContentFilterGuardrail( + guardrail_name="test-streaming-keyword-wider-than-context", + blocked_words=[BlockedWord(keyword=phrase, action=ContentFilterAction.BLOCK)], + event_hook=GuardrailEventHooks.post_call, + ) + filler = "plain filler sentence. " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 23) + text = filler + phrase + " and that is all." + chunks = [text[i : i + 16] for i in range(0, len(text), 16)] + request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + + with pytest.raises(HTTPException) as exc_info: + await self._collect_streamed_text(guardrail, chunks, request_data) + + assert exc_info.value.detail["keyword"] == phrase + entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + assert entry["guardrail_status"] == "guardrail_intervened" + @pytest.mark.asyncio async def test_streaming_hook_masks_every_email_in_long_stream_and_logs_once( self, From 7e429dee8726dce82a18ebce85a3d6dff6afce98 Mon Sep 17 00:00:00 2001 From: yassin Date: Wed, 16 Sep 2026 11:18:10 +0000 Subject: [PATCH 3/6] fix(content_filter): keep exception phrases and open conditional sentences in the streamed buffer Trimming the streamed buffer to the retained tail could drop a category exception phrase that suppresses a later keyword, or the identifier word of an unfinished sentence that a conditional category pairs with a later block word. Refuse the cut while either would leave the buffer so the bounded scan masks and blocks exactly like a scan of the full text Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../litellm_content_filter/content_filter.py | 63 ++++++++++++++----- .../content_filter/test_content_filter.py | 54 ++++++++++++++++ 2 files changed, 102 insertions(+), 15 deletions(-) diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py index a210762d1df..111f1d4d59a 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py @@ -66,6 +66,7 @@ from .patterns import PATTERN_EXTRA_CONFIG, get_compiled_pattern MAX_KEYWORD_VALUE_GAP_WORDS: Final = 1 GAP_WORD_TOKENIZER: Final = re.compile(r"\b\w+\b") +SENTENCE_TERMINATORS: Final = re.compile(r"[.!?]+") WORD_NUMBER_MAP: Final = { @@ -125,6 +126,13 @@ class _StreamedChoiceState: latest_detections: tuple[ContentFilterDetection, ...] = () +@dataclass(frozen=True, slots=True) +class _StreamedScanPlan: + context_chars: int + exception_phrases: tuple[str, ...] + conditional_words: tuple[str, ...] + + class CategoryFileData(TypedDict, total=False): category_name: str description: str @@ -989,7 +997,7 @@ class ContentFilterGuardrail(CustomGuardrail): # Split text into sentences for more precise matching # Simple sentence splitting on common terminators - sentences: Final = re.split(r"[.!?]+", text) + sentences: Final = SENTENCE_TERMINATORS.split(text) for category_name, config in self.conditional_categories.items(): identifier_words = config["identifier_words"] @@ -1963,31 +1971,58 @@ class ContentFilterGuardrail(CustomGuardrail): exception_str=exception_str, ) - def _streamed_scan_context_chars(self) -> int: - """Retained tail length: the default context, widened to the longest configured keyword.""" + def _streamed_scan_plan(self) -> _StreamedScanPlan: + """ + Per-stream inputs for buffer trimming: the retained tail length (the default + context, widened to the longest configured keyword), the category exception + phrases, which suppress matches anywhere in the scanned text, and the conditional + category words, which only match when paired inside one sentence. + """ longest_keyword: Final = max( map(len, (*self.blocked_words, *self.category_keywords, *self.always_block_category_keywords)), default=0, ) - return max(CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS, longest_keyword) + return _StreamedScanPlan( + context_chars=max(CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS, longest_keyword), + exception_phrases=tuple( + phrase for category in self.loaded_categories.values() for phrase in category.exceptions + ), + conditional_words=tuple( + word + for config in self.conditional_categories.values() + for word in (*config["identifier_words"], *config["block_words"]) + ), + ) + + @staticmethod + def _cut_breaks_wider_context(buffered_text: str, head: str, tail: str, plan: _StreamedScanPlan) -> bool: + buffered_lower: Final = buffered_text.lower() + tail_lower: Final = tail.lower() + if any(phrase in buffered_lower and phrase not in tail_lower for phrase in plan.exception_phrases): + return True + open_sentence: Final = SENTENCE_TERMINATORS.split(head.lower())[-1] + return any(word in open_sentence for word in plan.conditional_words) def _trim_streamed_choice_buffer( - self, state: _StreamedChoiceState, masked_text: str, scan_context_chars: int + self, state: _StreamedChoiceState, masked_text: str, plan: _StreamedScanPlan ) -> _StreamedChoiceState: """ Bound the per-choice buffer rescanned on every streamed chunk. Once the buffer exceeds twice the scan context, drop everything but the last - context-sized tail, provided the two halves mask to the same output as the whole - (so no match, phrase or exception straddles the cut) and the dropped prefix has - already been yielded. Otherwise keep the buffer and retry on the next chunk. + context-sized tail, provided no exception phrase or unfinished conditional sentence + would leave the buffer, the two halves mask to the same output as the whole (so no + match or phrase straddles the cut), and the dropped prefix has already been yielded. + Otherwise keep the buffer and retry on the next chunk. Detections found in the dropped prefix move to the state's committed detections. """ - if len(state.buffered_text) <= 2 * scan_context_chars: + if len(state.buffered_text) <= 2 * plan.context_chars: + return state + head: Final = state.buffered_text[: -plan.context_chars] + tail: Final = state.buffered_text[-plan.context_chars :] + if self._cut_breaks_wider_context(state.buffered_text, head, tail, plan): return state - head: Final = state.buffered_text[:-scan_context_chars] - tail: Final = state.buffered_text[-scan_context_chars:] head_detections: Final[list[ContentFilterDetection]] = [] # mutable-ok: filled by _filter_single_text try: masked_head: Final = self._filter_single_text(head, detections=head_detections) @@ -2026,7 +2061,7 @@ class ContentFilterGuardrail(CustomGuardrail): contract. """ state_by_choice: Final[dict[int, _StreamedChoiceState]] = {} - scan_context_chars: Final = self._streamed_scan_context_chars() + plan: Final = self._streamed_scan_plan() start_time: Final = datetime.now() scan_seconds: float = 0.0 # rebind-ok: accumulates per-chunk scan time across the stream @@ -2096,9 +2131,7 @@ class ContentFilterGuardrail(CustomGuardrail): continue trim_started = time.perf_counter() - state_by_choice[choice_index] = self._trim_streamed_choice_buffer( - next_state, masked_text, scan_context_chars - ) + state_by_choice[choice_index] = self._trim_streamed_choice_buffer(next_state, masked_text, plan) scan_seconds += time.perf_counter() - trim_started yield item diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py index b91abb98dbf..a35f6f26983 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py @@ -1036,6 +1036,60 @@ class TestContentFilterGuardrail: entry = request_data["metadata"]["standard_logging_guardrail_information"][0] assert entry["guardrail_status"] == "guardrail_intervened" + @pytest.mark.asyncio + async def test_streaming_hook_keeps_early_exception_phrase_suppressing_later_keyword(self): + """ + Category exception phrases suppress category matches anywhere in the + scanned text. An exception phrase at the start of a long response must keep + suppressing a category keyword that arrives long after the buffer would + otherwise have been trimmed, exactly as one scan of the full text does. + """ + guardrail = ContentFilterGuardrail( + guardrail_name="test-streaming-exception-context", + categories=[{"category": "harmful_self_harm", "enabled": True, "action": "BLOCK"}], + event_hook=GuardrailEventHooks.post_call, + ) + exception_phrase = guardrail.loaded_categories["harmful_self_harm"].exceptions[0] + keyword = next(iter(guardrail.category_keywords)) + filler = "plain filler sentence. " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 23) + text = f"Resources on {exception_phrase} matter. {filler}Someone said {keyword} in a novel." + chunks = [text[i : i + 16] for i in range(0, len(text), 16)] + request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + + streamed = await self._collect_streamed_text(guardrail, chunks, request_data) + + full_scan = await guardrail.apply_guardrail(inputs={"texts": [text]}, request_data={}, input_type="response") + assert streamed == full_scan["texts"][0] == text + + @pytest.mark.asyncio + async def test_streaming_hook_blocks_conditional_pair_split_by_long_sentence(self): + """ + Conditional categories block an identifier word and a block word that + share one sentence. When the sentence runs longer than the retained + context, the identifier at its start must still be in the buffer when the + block word arrives, so the stream is blocked like a scan of the full text. + """ + guardrail = ContentFilterGuardrail( + guardrail_name="test-streaming-conditional-context", + categories=[{"category": "harmful_child_safety", "enabled": True, "action": "BLOCK"}], + event_hook=GuardrailEventHooks.post_call, + ) + conditional = guardrail.conditional_categories["harmful_child_safety"] + identifier, block_word = conditional["identifier_words"][0], conditional["block_words"][-1] + filler = "and then more plain words " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 26) + text = f"In this chapter the {identifier} {filler}shared an {block_word} moment. The end." + chunks = [text[i : i + 16] for i in range(0, len(text), 16)] + request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + + with pytest.raises(HTTPException): + await guardrail.apply_guardrail(inputs={"texts": [text]}, request_data={}, input_type="response") + with pytest.raises(HTTPException) as exc_info: + await self._collect_streamed_text(guardrail, chunks, request_data) + + assert "harmful_child_safety" in str(exc_info.value.detail) + entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + assert entry["guardrail_status"] == "guardrail_intervened" + @pytest.mark.asyncio async def test_streaming_hook_masks_every_email_in_long_stream_and_logs_once( self, From 60642e875bff8838d452d3db29e017df4d3651b4 Mon Sep 17 00:00:00 2001 From: yassin Date: Wed, 16 Sep 2026 11:51:06 +0000 Subject: [PATCH 4/6] perf(content_filter): back off refused streamed buffer cuts by one context length Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../litellm_content_filter/content_filter.py | 20 ++++++--- .../content_filter/test_content_filter.py | 43 +++++++++++++++++++ 2 files changed, 56 insertions(+), 7 deletions(-) diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py index 111f1d4d59a..d4edf618d1b 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py @@ -124,6 +124,7 @@ class _StreamedChoiceState: yielded_masked_text_len: int = 0 committed_detections: tuple[ContentFilterDetection, ...] = () latest_detections: tuple[ContentFilterDetection, ...] = () + next_trim_len: int = 0 @dataclass(frozen=True, slots=True) @@ -2013,29 +2014,31 @@ class ContentFilterGuardrail(CustomGuardrail): context-sized tail, provided no exception phrase or unfinished conditional sentence would leave the buffer, the two halves mask to the same output as the whole (so no match or phrase straddles the cut), and the dropped prefix has already been yielded. - Otherwise keep the buffer and retry on the next chunk. + Otherwise keep the buffer and retry once it has grown by another context length. Detections found in the dropped prefix move to the state's committed detections. """ - if len(state.buffered_text) <= 2 * plan.context_chars: + if len(state.buffered_text) <= max(2 * plan.context_chars, state.next_trim_len): return state + deferred: Final = replace(state, next_trim_len=len(state.buffered_text) + plan.context_chars) head: Final = state.buffered_text[: -plan.context_chars] tail: Final = state.buffered_text[-plan.context_chars :] if self._cut_breaks_wider_context(state.buffered_text, head, tail, plan): - return state + return deferred head_detections: Final[list[ContentFilterDetection]] = [] # mutable-ok: filled by _filter_single_text try: masked_head: Final = self._filter_single_text(head, detections=head_detections) masked_tail: Final = self._filter_single_text(tail) except Exception: - return state + return deferred if masked_head + masked_tail != masked_text or len(masked_head) > state.yielded_masked_text_len: - return state + return deferred return replace( state, buffered_text=tail, yielded_masked_text_len=state.yielded_masked_text_len - len(masked_head), committed_detections=state.committed_detections + tuple(head_detections), + next_trim_len=0, ) @staticmethod @@ -2123,8 +2126,11 @@ class ContentFilterGuardrail(CustomGuardrail): len(masked_text) - (0 if is_final else CONTENT_FILTER_STREAMING_HOLDBACK_CHARS), ) choice.delta.content = masked_text[previous_state.yielded_masked_text_len : safe_to_yield_len] - next_state = _StreamedChoiceState( - buffered_text, safe_to_yield_len, previous_state.committed_detections, latest_detections + next_state = replace( + previous_state, + buffered_text=buffered_text, + yielded_masked_text_len=safe_to_yield_len, + latest_detections=latest_detections, ) if is_final: state_by_choice[choice_index] = next_state diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py index a35f6f26983..ca0f9c474a8 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py @@ -981,6 +981,49 @@ class TestContentFilterGuardrail: f"{len(chunk)}-char chunk; expected at most {window_bound}" ) + @pytest.mark.asyncio + async def test_streaming_hook_retries_refused_cut_once_per_context_length(self): + """ + A single URL that keeps growing crosses every proposed cut, so no cut is + ever safe. The trim check must then back off instead of adding two extra + scans on every chunk, and the whole URL must still come out masked. + """ + scanned_lengths: list[int] = [] + + class RecordingGuardrail(ContentFilterGuardrail): + def _filter_single_text( + self, + text: str, + detections: list[ContentFilterDetection] | None = None, + ) -> str: + scanned_lengths.append(len(text)) + return super()._filter_single_text(text, detections=detections) + + guardrail = RecordingGuardrail( + guardrail_name="test-streaming-refused-cut-backoff", + patterns=[ + ContentFilterPattern( + pattern_type="prebuilt", + pattern_name="url", + action=ContentFilterAction.MASK, + ) + ], + event_hook=GuardrailEventHooks.post_call, + ) + text = "See https://example.com/" + "a" * (8 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS) + " now." + chunks = [text[i : i + 16] for i in range(0, len(text), 16)] + request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + + streamed = await self._collect_streamed_text(guardrail, chunks, request_data) + streamed_scans = len(scanned_lengths) + + full_scan = await guardrail.apply_guardrail(inputs={"texts": [text]}, request_data={}, input_type="response") + assert streamed == full_scan["texts"][0] == "See [URL_REDACTED] now." + extra_scans = streamed_scans - len(chunks) + assert extra_scans <= 2 * (len(text) // CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS), ( + f"{extra_scans} scans beyond one per chunk for {len(chunks)} chunks; the refused cut must back off" + ) + @pytest.mark.asyncio async def test_streaming_hook_blocks_match_longer_than_holdback_across_chunks( self, From 4fbe63114626db307a6094c6951ee98b2e7db853 Mon Sep 17 00:00:00 2001 From: yassin Date: Wed, 16 Sep 2026 12:17:47 +0000 Subject: [PATCH 5/6] test(content_filter): annotate streaming test locals as Final and type the logging metadata Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../content_filter/test_content_filter.py | 135 +++++++++--------- 1 file changed, 70 insertions(+), 65 deletions(-) diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py index ca0f9c474a8..7dad44a8dc7 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py @@ -4,6 +4,7 @@ Tests for the Content Filter Guardrail import json import os +from typing import Final from unittest.mock import MagicMock import pytest @@ -28,6 +29,7 @@ from litellm.types.proxy.guardrails.guardrail_hooks.litellm_content_filter impor ContentFilterCategoryConfig, ContentFilterDetection, ) +from litellm.types.utils import StandardLoggingGuardrailInformation class TestContentFilterGuardrail: @@ -907,7 +909,9 @@ class TestContentFilterGuardrail: @staticmethod async def _collect_streamed_text( - guardrail: ContentFilterGuardrail, chunks: list[str], request_data: dict + guardrail: ContentFilterGuardrail, + chunks: list[str], + metadata: dict[str, list[StandardLoggingGuardrailInformation]], ) -> str: from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices @@ -928,11 +932,11 @@ class TestContentFilterGuardrail: model="gpt-4", ) - yielded = [] + yielded: Final[list[str]] = [] async for chunk in guardrail.async_post_call_streaming_iterator_hook( user_api_key_dict=MagicMock(), response=mock_stream(), - request_data=request_data, + request_data={"messages": [], "model": "gpt-4o", "metadata": metadata}, ): yielded.append(chunk.choices[0].delta.content or "") return "".join(yielded) @@ -945,7 +949,7 @@ class TestContentFilterGuardrail: of the response. Each scan must now cover only the new chunk plus a bounded tail of what came before, without dropping any output. """ - scanned_lengths: list[int] = [] + scanned_lengths: Final[list[int]] = [] class RecordingGuardrail(ContentFilterGuardrail): def _filter_single_text( @@ -956,7 +960,7 @@ class TestContentFilterGuardrail: scanned_lengths.append(len(text)) return super()._filter_single_text(text, detections=detections) - guardrail = RecordingGuardrail( + guardrail: Final = RecordingGuardrail( guardrail_name="test-streaming-bounded-scan", patterns=[ ContentFilterPattern( @@ -967,15 +971,14 @@ class TestContentFilterGuardrail: ], event_hook=GuardrailEventHooks.post_call, ) - chunk = "Item: a plain household object description. " - chunks = [chunk] * 200 - request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + chunk: Final = "Item: a plain household object description. " + chunks: Final = [chunk] * 200 - streamed = await self._collect_streamed_text(guardrail, chunks, request_data) + streamed: Final = await self._collect_streamed_text(guardrail, chunks, {}) assert streamed == chunk * 200 assert len(chunk) * 200 > 4 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS - window_bound = 2 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + len(chunk) + 1 + window_bound: Final = 2 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + len(chunk) + 1 assert max(scanned_lengths) <= window_bound, ( f"scan input grew to {max(scanned_lengths)} chars for a " f"{len(chunk)}-char chunk; expected at most {window_bound}" @@ -988,7 +991,7 @@ class TestContentFilterGuardrail: ever safe. The trim check must then back off instead of adding two extra scans on every chunk, and the whole URL must still come out masked. """ - scanned_lengths: list[int] = [] + scanned_lengths: Final[list[int]] = [] class RecordingGuardrail(ContentFilterGuardrail): def _filter_single_text( @@ -999,7 +1002,7 @@ class TestContentFilterGuardrail: scanned_lengths.append(len(text)) return super()._filter_single_text(text, detections=detections) - guardrail = RecordingGuardrail( + guardrail: Final = RecordingGuardrail( guardrail_name="test-streaming-refused-cut-backoff", patterns=[ ContentFilterPattern( @@ -1010,16 +1013,17 @@ class TestContentFilterGuardrail: ], event_hook=GuardrailEventHooks.post_call, ) - text = "See https://example.com/" + "a" * (8 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS) + " now." - chunks = [text[i : i + 16] for i in range(0, len(text), 16)] - request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + text: Final = "See https://example.com/" + "a" * (8 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS) + " now." + chunks: Final = [text[i : i + 16] for i in range(0, len(text), 16)] - streamed = await self._collect_streamed_text(guardrail, chunks, request_data) - streamed_scans = len(scanned_lengths) + streamed: Final = await self._collect_streamed_text(guardrail, chunks, {}) + streamed_scans: Final = len(scanned_lengths) - full_scan = await guardrail.apply_guardrail(inputs={"texts": [text]}, request_data={}, input_type="response") + full_scan: Final = await guardrail.apply_guardrail( + inputs={"texts": [text]}, request_data={}, input_type="response" + ) assert streamed == full_scan["texts"][0] == "See [URL_REDACTED] now." - extra_scans = streamed_scans - len(chunks) + extra_scans: Final = streamed_scans - len(chunks) assert extra_scans <= 2 * (len(text) // CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS), ( f"{extra_scans} scans beyond one per chunk for {len(chunks)} chunks; the refused cut must back off" ) @@ -1033,22 +1037,22 @@ class TestContentFilterGuardrail: so its start has already been yielded before its end shows up. The scan still has to see the whole phrase and block. """ - phrase = "alpha bravo charlie delta echo foxtrot golf hotel india juliet kilo lima" + phrase: Final = "alpha bravo charlie delta echo foxtrot golf hotel india juliet kilo lima" assert len(phrase) > CONTENT_FILTER_STREAMING_HOLDBACK_CHARS - guardrail = ContentFilterGuardrail( + guardrail: Final = ContentFilterGuardrail( guardrail_name="test-streaming-long-block", blocked_words=[BlockedWord(keyword=phrase, action=ContentFilterAction.BLOCK)], event_hook=GuardrailEventHooks.post_call, ) - text = "Here is the codeword list: " + phrase + " and that is all." - chunks = [text[i : i + 4] for i in range(0, len(text), 4)] - request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + text: Final = "Here is the codeword list: " + phrase + " and that is all." + chunks: Final = [text[i : i + 4] for i in range(0, len(text), 4)] + metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {} with pytest.raises(HTTPException) as exc_info: - await self._collect_streamed_text(guardrail, chunks, request_data) + await self._collect_streamed_text(guardrail, chunks, metadata) assert exc_info.value.detail["keyword"] == phrase - entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + entry: Final = metadata["standard_logging_guardrail_information"][0] assert entry["guardrail_status"] == "guardrail_intervened" assert [d["keyword"] for d in entry["guardrail_response"]] == [phrase] @@ -1060,23 +1064,23 @@ class TestContentFilterGuardrail: retained tail must be wide enough that the keyword's start is still in the buffer when its end arrives, so the stream is blocked. """ - phrase = " ".join(f"token{i:03d}" for i in range(80)) + phrase: Final = " ".join(f"token{i:03d}" for i in range(80)) assert len(phrase) > CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS - guardrail = ContentFilterGuardrail( + guardrail: Final = ContentFilterGuardrail( guardrail_name="test-streaming-keyword-wider-than-context", blocked_words=[BlockedWord(keyword=phrase, action=ContentFilterAction.BLOCK)], event_hook=GuardrailEventHooks.post_call, ) - filler = "plain filler sentence. " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 23) - text = filler + phrase + " and that is all." - chunks = [text[i : i + 16] for i in range(0, len(text), 16)] - request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + filler: Final = "plain filler sentence. " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 23) + text: Final = filler + phrase + " and that is all." + chunks: Final = [text[i : i + 16] for i in range(0, len(text), 16)] + metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {} with pytest.raises(HTTPException) as exc_info: - await self._collect_streamed_text(guardrail, chunks, request_data) + await self._collect_streamed_text(guardrail, chunks, metadata) assert exc_info.value.detail["keyword"] == phrase - entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + entry: Final = metadata["standard_logging_guardrail_information"][0] assert entry["guardrail_status"] == "guardrail_intervened" @pytest.mark.asyncio @@ -1087,21 +1091,22 @@ class TestContentFilterGuardrail: suppressing a category keyword that arrives long after the buffer would otherwise have been trimmed, exactly as one scan of the full text does. """ - guardrail = ContentFilterGuardrail( + guardrail: Final = ContentFilterGuardrail( guardrail_name="test-streaming-exception-context", categories=[{"category": "harmful_self_harm", "enabled": True, "action": "BLOCK"}], event_hook=GuardrailEventHooks.post_call, ) - exception_phrase = guardrail.loaded_categories["harmful_self_harm"].exceptions[0] - keyword = next(iter(guardrail.category_keywords)) - filler = "plain filler sentence. " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 23) - text = f"Resources on {exception_phrase} matter. {filler}Someone said {keyword} in a novel." - chunks = [text[i : i + 16] for i in range(0, len(text), 16)] - request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + exception_phrase: Final = guardrail.loaded_categories["harmful_self_harm"].exceptions[0] + keyword: Final = next(iter(guardrail.category_keywords)) + filler: Final = "plain filler sentence. " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 23) + text: Final = f"Resources on {exception_phrase} matter. {filler}Someone said {keyword} in a novel." + chunks: Final = [text[i : i + 16] for i in range(0, len(text), 16)] - streamed = await self._collect_streamed_text(guardrail, chunks, request_data) + streamed: Final = await self._collect_streamed_text(guardrail, chunks, {}) - full_scan = await guardrail.apply_guardrail(inputs={"texts": [text]}, request_data={}, input_type="response") + full_scan: Final = await guardrail.apply_guardrail( + inputs={"texts": [text]}, request_data={}, input_type="response" + ) assert streamed == full_scan["texts"][0] == text @pytest.mark.asyncio @@ -1112,25 +1117,25 @@ class TestContentFilterGuardrail: context, the identifier at its start must still be in the buffer when the block word arrives, so the stream is blocked like a scan of the full text. """ - guardrail = ContentFilterGuardrail( + guardrail: Final = ContentFilterGuardrail( guardrail_name="test-streaming-conditional-context", categories=[{"category": "harmful_child_safety", "enabled": True, "action": "BLOCK"}], event_hook=GuardrailEventHooks.post_call, ) - conditional = guardrail.conditional_categories["harmful_child_safety"] + conditional: Final = guardrail.conditional_categories["harmful_child_safety"] identifier, block_word = conditional["identifier_words"][0], conditional["block_words"][-1] - filler = "and then more plain words " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 26) - text = f"In this chapter the {identifier} {filler}shared an {block_word} moment. The end." - chunks = [text[i : i + 16] for i in range(0, len(text), 16)] - request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + filler: Final = "and then more plain words " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 26) + text: Final = f"In this chapter the {identifier} {filler}shared an {block_word} moment. The end." + chunks: Final = [text[i : i + 16] for i in range(0, len(text), 16)] + metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {} with pytest.raises(HTTPException): await guardrail.apply_guardrail(inputs={"texts": [text]}, request_data={}, input_type="response") with pytest.raises(HTTPException) as exc_info: - await self._collect_streamed_text(guardrail, chunks, request_data) + await self._collect_streamed_text(guardrail, chunks, metadata) assert "harmful_child_safety" in str(exc_info.value.detail) - entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + entry: Final = metadata["standard_logging_guardrail_information"][0] assert entry["guardrail_status"] == "guardrail_intervened" @pytest.mark.asyncio @@ -1145,7 +1150,7 @@ class TestContentFilterGuardrail: dropping text without checking that the cut leaves the masked output unchanged corrupts the stream. """ - guardrail = ContentFilterGuardrail( + guardrail: Final = ContentFilterGuardrail( guardrail_name="test-streaming-many-emails", patterns=[ ContentFilterPattern( @@ -1156,16 +1161,16 @@ class TestContentFilterGuardrail: ], event_hook=GuardrailEventHooks.post_call, ) - emails = [f"user{i:03d}@example.com" for i in range(200)] - text = " ".join(emails) + emails: Final = [f"user{i:03d}@example.com" for i in range(200)] + text: Final = " ".join(emails) assert len(text) > 4 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS - chunks = [text[i : i + 3] for i in range(0, len(text), 3)] - request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + chunks: Final = [text[i : i + 3] for i in range(0, len(text), 3)] + metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {} - streamed = await self._collect_streamed_text(guardrail, chunks, request_data) + streamed: Final = await self._collect_streamed_text(guardrail, chunks, metadata) assert streamed == " ".join(["[EMAIL_REDACTED]"] * len(emails)) - entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + entry: Final = metadata["standard_logging_guardrail_information"][0] assert entry["guardrail_status"] == "success" assert [d["pattern_name"] for d in entry["guardrail_response"]] == ["email"] assert entry["masked_entity_count"] == {"email": 1} @@ -1177,7 +1182,7 @@ class TestContentFilterGuardrail: the rescanned buffer well before the stream ends. The final log entry must still report it, as a scan of the full text would. """ - guardrail = ContentFilterGuardrail( + guardrail: Final = ContentFilterGuardrail( guardrail_name="test-streaming-early-detection", patterns=[ ContentFilterPattern( @@ -1188,15 +1193,15 @@ class TestContentFilterGuardrail: ], event_hook=GuardrailEventHooks.post_call, ) - filler = "filler text " * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS - text = f"Contact one@example.com for details. {filler}" - chunks = [text[i : i + 40] for i in range(0, len(text), 40)] - request_data = {"messages": [], "model": "gpt-4o", "metadata": {}} + filler: Final = "filler text " * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + text: Final = f"Contact one@example.com for details. {filler}" + chunks: Final = [text[i : i + 40] for i in range(0, len(text), 40)] + metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {} - streamed = await self._collect_streamed_text(guardrail, chunks, request_data) + streamed: Final = await self._collect_streamed_text(guardrail, chunks, metadata) assert streamed == text.replace("one@example.com", "[EMAIL_REDACTED]") - entry = request_data["metadata"]["standard_logging_guardrail_information"][0] + entry: Final = metadata["standard_logging_guardrail_information"][0] assert entry["guardrail_status"] == "success" assert [d["pattern_name"] for d in entry["guardrail_response"]] == ["email"] assert entry["masked_entity_count"] == {"email": 1} From a8fff5b0912607ad435c1ab35960f62f064cc766 Mon Sep 17 00:00:00 2001 From: yassin Date: Wed, 16 Sep 2026 12:45:58 +0000 Subject: [PATCH 6/6] fix(content_filter): refuse a trim that splits a conditional word across the cut Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- .../litellm_content_filter/content_filter.py | 6 ++-- .../content_filter/test_content_filter.py | 36 +++++++++++++++++++ 2 files changed, 40 insertions(+), 2 deletions(-) diff --git a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py index d4edf618d1b..092e8eaafa1 100644 --- a/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py +++ b/litellm/proxy/guardrails/guardrail_hooks/litellm_content_filter/content_filter.py @@ -2001,8 +2001,10 @@ class ContentFilterGuardrail(CustomGuardrail): tail_lower: Final = tail.lower() if any(phrase in buffered_lower and phrase not in tail_lower for phrase in plan.exception_phrases): return True - open_sentence: Final = SENTENCE_TERMINATORS.split(head.lower())[-1] - return any(word in open_sentence for word in plan.conditional_words) + cut_sentence: Final = ( + SENTENCE_TERMINATORS.split(head.lower())[-1] + SENTENCE_TERMINATORS.split(tail_lower, maxsplit=1)[0] + ) + return any(word in cut_sentence for word in plan.conditional_words) def _trim_streamed_choice_buffer( self, state: _StreamedChoiceState, masked_text: str, plan: _StreamedScanPlan diff --git a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py index 7dad44a8dc7..d0ad068aeb4 100644 --- a/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py +++ b/tests/test_litellm/proxy/guardrails/guardrail_hooks/content_filter/test_content_filter.py @@ -1138,6 +1138,42 @@ class TestContentFilterGuardrail: entry: Final = metadata["standard_logging_guardrail_information"][0] assert entry["guardrail_status"] == "guardrail_intervened" + @pytest.mark.asyncio + async def test_streaming_hook_blocks_conditional_identifier_straddling_cut(self): + """ + The buffer is cut at a character offset, so a conditional identifier word + can sit half in the dropped head and half in the retained tail. That cut + must be refused: otherwise the block word arriving later in the same + sentence finds no identifier and the stream passes where a scan of the + full text blocks. + """ + guardrail: Final = ContentFilterGuardrail( + guardrail_name="test-streaming-conditional-straddle", + categories=[{"category": "harmful_child_safety", "enabled": True, "action": "BLOCK"}], + event_hook=GuardrailEventHooks.post_call, + ) + conditional: Final = guardrail.conditional_categories["harmful_child_safety"] + identifier, block_word = conditional["identifier_words"][0], conditional["block_words"][-1] + chunk_size: Final = 16 + first_cut: Final = ( + 2 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // chunk_size + 1 + ) * chunk_size - CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS + prefix: Final = ("plain words " * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS)[: first_cut - 2] + filler: Final = "and then more plain words " * (3 * CONTENT_FILTER_STREAMING_SCAN_CONTEXT_CHARS // 26) + text: Final = f"{prefix}{identifier} {filler}shared an {block_word} moment. The end." + assert text[first_cut - 2 : first_cut - 2 + len(identifier)] == identifier + chunks: Final = [text[i : i + chunk_size] for i in range(0, len(text), chunk_size)] + metadata: Final[dict[str, list[StandardLoggingGuardrailInformation]]] = {} + + with pytest.raises(HTTPException): + await guardrail.apply_guardrail(inputs={"texts": [text]}, request_data={}, input_type="response") + with pytest.raises(HTTPException) as exc_info: + await self._collect_streamed_text(guardrail, chunks, metadata) + + assert "harmful_child_safety" in str(exc_info.value.detail) + entry: Final = metadata["standard_logging_guardrail_information"][0] + assert entry["guardrail_status"] == "guardrail_intervened" + @pytest.mark.asyncio async def test_streaming_hook_masks_every_email_in_long_stream_and_logs_once( self,