diff --git a/litellm/litellm_core_utils/realtime_streaming.py b/litellm/litellm_core_utils/realtime_streaming.py index ea912ba62a0..eefa918926a 100644 --- a/litellm/litellm_core_utils/realtime_streaming.py +++ b/litellm/litellm_core_utils/realtime_streaming.py @@ -254,25 +254,32 @@ class RealTimeStreaming: ## SYNC LOGGING executor.submit(self.logging_obj.success_handler(self.messages)) - async def _send_to_backend(self, message: str) -> None: + async def _send_to_backend(self, message: str) -> bool: """Send a message to the backend WebSocket. If a provider_config is set the message is first passed through transform_realtime_request so that provider-specific translation (e.g. dropping session.update for Vertex AI) is applied even for guardrail-injected messages. + + Returns True if at least one message was actually delivered to the + backend, False if the provider transformation produced no output and + the message was effectively dropped. """ if self.provider_config: transformed = self.provider_config.transform_realtime_request( message, self.model, self.session_configuration_request ) + sent = False for msg in transformed: # Cache setup immediately once we send it so concurrent client # session.update messages don't emit duplicate setup packets. self._cache_session_configuration_request(msg) await self.backend_ws.send(msg) # type: ignore[union-attr, attr-defined] - else: - await self.backend_ws.send(message) # type: ignore[union-attr, attr-defined] + sent = True + return sent + await self.backend_ws.send(message) # type: ignore[union-attr, attr-defined] + return True def _cache_session_configuration_request(self, transformed_message: str) -> None: """Store setup payload once sent to backend.""" @@ -308,8 +315,13 @@ class RealTimeStreaming: return if not self._has_audio_transcription_guardrails(): return - await self._send_to_backend(self._make_disable_auto_response_message()) - self._guardrail_turn_detection_update_sent = True + sent = await self._send_to_backend(self._make_disable_auto_response_message()) + # Only mark as sent when the provider transformation actually delivered + # the update to the backend. Otherwise (e.g. Gemini drops session.update + # after the initial setup), leave the flag unset so future opportunities + # — such as a duplicate session.created — can retry. + if sent: + self._guardrail_turn_detection_update_sent = True def _has_realtime_guardrails(self) -> bool: """Return True if any callback is registered for realtime guardrail event types.""" @@ -879,7 +891,7 @@ class RealTimeStreaming: ## LOGGING self.store_input(message=message) - + ## GUARDRAIL: Inject turn_detection into first session.update if needed try: msg_obj = json.loads(message) @@ -891,7 +903,9 @@ class RealTimeStreaming: ): # Inject turn_detection into the first session.update session = msg_obj.setdefault("session", {}) - session.setdefault("turn_detection", {})["create_response"] = False + session.setdefault("turn_detection", {})[ + "create_response" + ] = False message = json.dumps(msg_obj) self._guardrail_turn_detection_update_sent = True verbose_logger.debug( @@ -899,7 +913,7 @@ class RealTimeStreaming: ) except (json.JSONDecodeError, AttributeError): pass - + ## FORWARD TO BACKEND if self.provider_config: message = self.provider_config.transform_realtime_request( diff --git a/litellm/llms/gemini/realtime/transformation.py b/litellm/llms/gemini/realtime/transformation.py index 0c6b04c41d4..917f13eefb6 100644 --- a/litellm/llms/gemini/realtime/transformation.py +++ b/litellm/llms/gemini/realtime/transformation.py @@ -972,14 +972,16 @@ class GeminiRealtimeConfig(BaseRealtimeConfig): delta_type=current_delta_type ) else: - # Check if this key or any nested key matches our mapping - for map_key, openai_event in MAP_GEMINI_FIELD_TO_OPENAI_EVENT.items(): + # Check if this key or any nested key matches our mapping. Use a + # distinct loop variable so we don't shadow ``openai_event`` and + # leak the last dict value when no entry matches. + for map_key, candidate_event in MAP_GEMINI_FIELD_TO_OPENAI_EVENT.items(): if map_key == key or ( "." in map_key and GeminiRealtimeConfig.get_nested_value(json_message, map_key) is not None ): - openai_event = openai_event + openai_event = candidate_event break if openai_event is None: raise ValueError(f"Unknown openai event: {key}, value: {value}")