From 7e429dee8726dce82a18ebce85a3d6dff6afce98 Mon Sep 17 00:00:00 2001 From: yassin Date: Wed, 16 Sep 2026 11:18:10 +0000 Subject: [PATCH] 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,