mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-06 08:16:43 +00:00
Avoid duplicate realtime response.create injection
This commit is contained in:
parent
13b590c8ec
commit
3fa5e26e7e
2 changed files with 58 additions and 12 deletions
|
|
@ -903,6 +903,8 @@ class RealTimeStreaming:
|
|||
self._collect_user_input_from_backend_event(cast(dict, event))
|
||||
self.store_message(event_str)
|
||||
await self._send_event_to_client(event, event_str)
|
||||
if not self._has_audio_transcription_guardrails():
|
||||
continue
|
||||
blocked = await self.run_realtime_guardrails(
|
||||
cast(str, transcript),
|
||||
item_id=cast(Optional[str], event.get("item_id")),
|
||||
|
|
@ -953,6 +955,8 @@ class RealTimeStreaming:
|
|||
if self._is_transcription_session:
|
||||
self._capture_transcription_usage(event_obj)
|
||||
return True
|
||||
if not self._has_audio_transcription_guardrails():
|
||||
return True
|
||||
|
||||
blocked = await self.run_realtime_guardrails(
|
||||
transcript,
|
||||
|
|
|
|||
|
|
@ -17,10 +17,6 @@ from litellm.litellm_core_utils.realtime_streaming import (
|
|||
)
|
||||
from litellm.llms.xai.realtime.transformation import XAIRealtimeNormalizer
|
||||
from litellm.types.guardrails import GuardrailEventHooks
|
||||
from litellm.types.llms.openai import (
|
||||
OpenAIRealtimeStreamResponseBaseObject,
|
||||
OpenAIRealtimeStreamSessionEvents,
|
||||
)
|
||||
|
||||
|
||||
def _make_transcript_event(text: str, item_id: str = "item_x") -> bytes:
|
||||
|
|
@ -809,8 +805,6 @@ async def test_transcription_captured_in_backend_to_client():
|
|||
Test that conversation.item.input_audio_transcription.completed events
|
||||
from the backend are captured as user input during the WebSocket session.
|
||||
"""
|
||||
import litellm
|
||||
|
||||
client_ws = MagicMock()
|
||||
client_ws.send_text = AsyncMock()
|
||||
|
||||
|
|
@ -903,10 +897,11 @@ async def test_transcription_session_captures_usage_and_skips_response_create():
|
|||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_transcription_completed_event_still_triggers_response_create():
|
||||
async def test_non_transcription_completed_event_without_guardrails_skips_response_create():
|
||||
"""
|
||||
Regression guard: a normal (non-transcription) session with no guardrails must
|
||||
keep triggering response.create on a completed transcription event.
|
||||
Regression guard: without realtime_input_transcription guardrails, the proxy
|
||||
must not inject response.create on transcription completion because the
|
||||
backend's own server-VAD auto-response is still responsible for the turn.
|
||||
"""
|
||||
client_ws = MagicMock()
|
||||
client_ws.send_text = AsyncMock()
|
||||
|
|
@ -932,7 +927,56 @@ async def test_non_transcription_completed_event_still_triggers_response_create(
|
|||
|
||||
assert streaming._is_transcription_session is False
|
||||
sent_to_backend = [json.loads(c.args[0]) for c in backend_ws.send.call_args_list if c.args]
|
||||
assert any(e.get("type") == "response.create" for e in sent_to_backend)
|
||||
assert all(e.get("type") != "response.create" for e in sent_to_backend), (
|
||||
f"proxy must not inject response.create without transcription guardrails, got: {sent_to_backend}"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_provider_path_completed_event_without_guardrails_skips_response_create():
|
||||
client_ws = MagicMock()
|
||||
client_ws.send_text = AsyncMock()
|
||||
|
||||
backend_ws = MagicMock()
|
||||
backend_ws.send = AsyncMock()
|
||||
|
||||
provider_config = MagicMock()
|
||||
provider_config.transform_realtime_response = MagicMock(
|
||||
return_value={
|
||||
"response": [
|
||||
{
|
||||
"type": "conversation.item.input_audio_transcription.completed",
|
||||
"transcript": "hi",
|
||||
"item_id": "item_1",
|
||||
}
|
||||
],
|
||||
"current_output_item_id": None,
|
||||
"current_response_id": None,
|
||||
"current_delta_chunks": [],
|
||||
"current_conversation_id": None,
|
||||
"current_item_chunks": [],
|
||||
"current_delta_type": None,
|
||||
"session_configuration_request": None,
|
||||
}
|
||||
)
|
||||
|
||||
logging_obj = MagicMock()
|
||||
logging_obj.litellm_trace_id = "trace_1"
|
||||
logging_obj.async_success_handler = AsyncMock()
|
||||
logging_obj.success_handler = MagicMock()
|
||||
|
||||
streaming = RealTimeStreaming(
|
||||
websocket=client_ws,
|
||||
backend_ws=backend_ws,
|
||||
logging_obj=logging_obj,
|
||||
provider_config=provider_config,
|
||||
model="gemini-2.5-flash",
|
||||
)
|
||||
streaming._send_to_backend = AsyncMock() # type: ignore[method-assign]
|
||||
|
||||
await streaming._handle_provider_config_message("{}")
|
||||
|
||||
assert streaming._send_to_backend.await_count == 0
|
||||
|
||||
|
||||
def test_client_session_update_marks_transcription_session():
|
||||
|
|
@ -1097,8 +1141,6 @@ def test_capture_transcription_usage_deduplicates_when_already_stored():
|
|||
When the event is already in messages (logged via store_message), it must not
|
||||
be appended a second time by _capture_transcription_usage.
|
||||
"""
|
||||
import litellm
|
||||
|
||||
streaming = RealTimeStreaming(MagicMock(), MagicMock(), MagicMock())
|
||||
# Add the event type to the default logged list so _should_store_message returns True.
|
||||
streaming.logged_real_time_event_types = ["conversation.item.input_audio_transcription.completed"]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue