diff --git a/litellm/litellm_core_utils/realtime_guardrails.py b/litellm/litellm_core_utils/realtime_guardrails.py index 6228e2893b7..7c242f0696a 100644 --- a/litellm/litellm_core_utils/realtime_guardrails.py +++ b/litellm/litellm_core_utils/realtime_guardrails.py @@ -62,6 +62,11 @@ def _json_object(value: object) -> Mapping[str, object] | None: return None +def _turn_copy(value: object) -> object: + mapping: Final = _json_object(value) + return value if mapping is None else dict(mapping) + + def _typed_text(part: object) -> str | None: part_object: Final = _json_object(part) if part_object is None or part_object.get("type") not in _TYPED_TEXT_PART_TYPES: @@ -264,7 +269,10 @@ async def realtime_guardrail_verdict( if guardrail is not None and _in_role_scope(guardrail, role) } in_scope: Final = tuple(guardrails.values()) - guardrail_request_data: Final = {**request_data, "user_api_key_dict": user_api_key_dict} + guardrail_request_data: Final = { + **{key: _turn_copy(value) for key, value in request_data.items()}, + "user_api_key_dict": user_api_key_dict, + } joined: Final = _joins(texts) screened: Final = await _screen_each(in_scope, joined, guardrail_request_data) if isinstance(screened, RealtimeGuardrailBlock): diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index 2b75e7c9392..438d18da2be 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -1670,12 +1670,18 @@ class RealTimeStreaming: Returns True if blocked (synthetic warning already sent to client). Returns False if clean (caller should send response.create to the backend). """ - if not isinstance(transcript, str): - return False - verdict: Final = await self.apply_realtime_guardrails( - (transcript,), "user", item_id, pre_block_backend_message, event_hooks - ) - return isinstance(verdict, RealtimeGuardrailBlock) + from litellm.types.guardrails import GuardrailEventHooks + + if isinstance(transcript, str): + verdict: Final = await self.apply_realtime_guardrails( + (transcript,), "user", item_id, pre_block_backend_message, event_hooks + ) + return isinstance(verdict, RealtimeGuardrailBlock) + if self._has_realtime_guardrails_for_event_hooks( + event_hooks or (GuardrailEventHooks.realtime_input_transcription,) + ): + raise TypeError(f"Realtime guardrails cannot screen a {type(transcript).__name__} transcript") + return False async def _screen_client_item(self, guarded_item: RealtimeGuardedItem | None) -> RealtimeGuardrailVerdict: from litellm.types.guardrails import GuardrailEventHooks diff --git a/tests/unit/litellm_core_utils/test_realtime_guardrails.py b/tests/unit/litellm_core_utils/test_realtime_guardrails.py index f227ffba296..26ab2887883 100644 --- a/tests/unit/litellm_core_utils/test_realtime_guardrails.py +++ b/tests/unit/litellm_core_utils/test_realtime_guardrails.py @@ -339,6 +339,44 @@ async def test_a_mask_in_a_message_with_more_text_parts_than_the_recursion_limit assert verdict == RealtimeGuardrailRewrite(texts=(*parts[:-1], "my code is [MASKED].")) +class GuardrailLogRecordingGuardrail(CustomGuardrail): + def __init__(self) -> None: + super().__init__(guardrail_name="log-recorder", event_hook=GuardrailEventHooks.pre_call, default_on=True) + self.guardrail_log_lengths: list[int] = [] + + async def apply_guardrail( + self, + inputs: GenericGuardrailAPIInputs, + request_data: dict, + input_type: Literal["request", "response"], + logging_obj: LiteLLMLogging | None = None, + ) -> GenericGuardrailAPIInputs: + self.guardrail_log_lengths.append( + len(request_data["litellm_metadata"].get("standard_logging_guardrail_information", [])) + ) + return inputs + + +@pytest.mark.asyncio +async def test_a_guardrail_sees_only_the_current_turns_guardrail_log_however_long_the_session_runs() -> None: + session_request_data: Final = {"litellm_metadata": {"user_api_key_team_id": "team-a"}} + content_filter: Final = ContentFilterGuardrail( + guardrail_name="phrase-filter", + blocked_words=[BlockedWord(keyword="hunter2", action=ContentFilterAction.BLOCK)], + event_hook=GuardrailEventHooks.pre_call, + default_on=True, + ) + recorder: Final = GuardrailLogRecordingGuardrail() + + for turn in ("hello", "how are you", "goodbye"): + await realtime_guardrail_verdict( + (turn,), "user", (GuardrailEventHooks.pre_call,), session_request_data, None, (content_filter, recorder) + ) + + assert recorder.guardrail_log_lengths == [1, 1, 1] + assert session_request_data == {"litellm_metadata": {"user_api_key_team_id": "team-a"}} + + class ScopedGuardrail(RaisingGuardrail): def __init__( self, *, skip_system_message: bool | None, skip_tool_message: bool | None, scan_only_tool_results: bool | None diff --git a/tests/unit/litellm_core_utils/test_realtime_streaming.py b/tests/unit/litellm_core_utils/test_realtime_streaming.py index 77392ec969c..fa66cf0dcb6 100644 --- a/tests/unit/litellm_core_utils/test_realtime_streaming.py +++ b/tests/unit/litellm_core_utils/test_realtime_streaming.py @@ -1081,6 +1081,35 @@ async def test_a_non_string_transcript_does_not_end_the_session(transcript: obje backend_ws.send.assert_not_called() +@pytest.mark.asyncio +@pytest.mark.parametrize("transcript", [None, 123, ["a"]]) +async def test_a_non_string_transcript_fails_the_session_when_a_transcript_guardrail_would_screen_it( + transcript: object, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(litellm, "callbacks", [_transcription_guardrail()]) + client_ws: Final = MagicMock() + client_ws.send_text = AsyncMock() + transcript_event: Final = json.dumps( + {"type": "conversation.item.input_audio_transcription.completed", "transcript": transcript, "item_id": "i"} + ) + later_event: Final = json.dumps({"type": "response.done", "response": {"id": "resp_1"}}) + backend_ws: Final = MagicMock() + backend_ws.recv = AsyncMock( + side_effect=[transcript_event.encode(), later_event.encode(), ConnectionClosed(None, None)] + ) + backend_ws.send = AsyncMock() + logging_obj: Final = _mock_logging() + logging_obj.model_call_details = {} + logging_obj.dispatch_success_handlers = AsyncMock() + + close: Final = await RealTimeStreaming(client_ws, backend_ws, logging_obj).backend_to_client_send_messages() + + assert close.code == 1011 + sent_types: Final = [json.loads(call.args[0])["type"] for call in client_ws.send_text.call_args_list] + assert sent_types == ["conversation.item.input_audio_transcription.completed"] + backend_ws.send.assert_not_called() + + @pytest.mark.asyncio async def test_transcription_session_captures_usage_and_skips_response_create(): """