tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-07-17 16:17:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(gemini-realtime): use GA event names for Pipecat 1.3.x compatibility (#29662)

Browse Source 
* fix(gemini-realtime): use GA event names for Pipecat 1.3.x compatibility

Pipecat v1.3.0 adopted the OpenAI Realtime API GA event naming:
  response.audio.delta          -> response.output_audio.delta
  response.text.delta           -> response.output_text.delta
  response.audio.done           -> response.output_audio.done
  response.text.done            -> response.output_text.done

The proxy was still emitting the old beta names; Pipecat's
`parse_server_event` raises "Unimplemented server event type" for any
unknown type, which killed the receive task handler and broke audio
playback and tool-call delivery.

Also:
- conversation.item.created -> conversation.item.added (already handled)
- client audio is buffered until backend setupComplete in deferred mode
- call_id fallback UUID when Gemini returns empty id
- status_details / token detail fields added to Pydantic-strict events

The _GA_TO_BETA_EVENT_TYPES map in RealTimeStreaming already translates
GA names back to beta for clients that opt in with the openai-beta
header, so legacy clients are unaffected.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(gemini-realtime): address greptile review comments

- emit outputTranscription as response.output_audio_transcript.delta
  instead of suppressing it; GA_TO_BETA map handles translation for
  legacy clients
- cap pre-setup audio buffer at 200 frames to prevent memory exhaustion;
  log a warning when the limit is hit and additional frames are dropped
- log remaining dropped message count on flush error

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(gemini-realtime): address veria review comments

- remove unused OpenAIRealtimeConversationItemCreated import
- fix guardrail bypass: semantic_vad early-return now preserves
  create_response when set so a guardrail-injected create_response:false
  is not silently dropped
- add per-connection 10 MB byte cap alongside the 200-frame count cap
  for the pre-setup audio buffer to prevent memory exhaustion

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(gemini-realtime): fix mypy arg-type on _finalize_gemini_live_setup

setup parameter typed as BidiGenerateContentSetup to match the TypedDict
passed at both call sites; was dict which mypy rejected.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(gemini-realtime): widen _finalize_gemini_live_setup to Dict[str, Any]

BidiGenerateContentSetup (TypedDict) is a subtype of Dict[str,Any] so
both call sites (one passing a plain dict, one passing the TypedDict)
satisfy mypy.

Co-authored-by: Cursor <cursoragent@cursor.com>

* fix(gemini-realtime): cast BidiGenerateContentSetup to Dict at _finalize call site

mypy rejects TypedDict as dict[str, Any] argument; cast at the call site
where follow_up_setup is BidiGenerateContentSetup to satisfy the checker.

Co-authored-by: Cursor <cursoragent@cursor.com>

* Fix Gemini realtime beta compatibility

* Fix deferred Gemini setup audio ordering

* fix: preserve Gemini audio transcript ids

* fix(realtime): cap pre-setup client buffer on all append paths

Route every append to the deferred-setup pending buffer through the
per-connection message/byte caps. Previously only the audio-buffer
fast path enforced the caps; once one frame was buffered, a client
that withheld session.update could stream arbitrary frames into
_pending_messages_until_setup unbounded and exhaust proxy memory.

* style(gemini-realtime): apply black formatting to transformation.py

* fix(gemini-realtime): log beta-translation fallback and name native-audio marker

Surface the previously swallowed exception in _send_event_to_client so a
failed GA->beta translation is observable instead of silently forwarding the
untranslated event. Extract the native-audio model substring used by
_finalize_gemini_live_setup into a named constant documenting why speechConfig
is dropped on those setups.

---------

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
Sameer Kankute
2026-06-04 08:03:24 -07:00
committed by GitHub
co-authored by Cursor mateo-berri
parent 20dc6dffa4
commit ed073d382d
5 changed files with 848 additions and 60 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/litellm_core_utils/realtime_streaming.py
+139 -3
View File
@@ -92,8 +92,27 @@ class RealTimeStreaming:
# Track whether we have already sent the guardrail turn-detection update
# that disables provider auto-response for transcription guardrails.
self._guardrail_turn_detection_update_sent: bool = False
# Deferred Gemini Live setup: Pipecat may stream audio before session.update.
# Buffer client audio until the backend acknowledges setup (setupComplete).
self._backend_setup_complete: bool = (
provider_config is None or provider_config.requires_session_configuration()
)
self._flushing_pending_messages_until_setup: bool = False
self._pending_messages_until_setup: List[str] = []
self._pending_messages_byte_total: int = 0
# Per-connection caps for pre-setup audio frames (message count + total bytes).
_MAX_BUFFERED_MESSAGES: int = 200
_MAX_BUFFERED_BYTES: int = 10 * 1024 * 1024 # 10 MB
_SESSION_EVENT_TYPES = frozenset(["session.created", "session.updated"])
_CLIENT_AUDIO_BUFFER_TYPES = frozenset(
[
"input_audio_buffer.append",
"input_audio_buffer.commit",
"input_audio_buffer.clear",
]
)
_AUDIO_FORMAT_MAP: Dict[str, Dict[str, Any]] = {
"pcm16": {"type": "audio/pcm", "rate": 24000},
"g711_ulaw": {"type": "audio/G711-ulaw", "rate": 8000},
@@ -285,6 +304,86 @@ class RealTimeStreaming:
await self.backend_ws.send(message) # type: ignore[union-attr, attr-defined]
return True
def _uses_deferred_backend_setup(self) -> bool:
"""True when setup is deferred until the client's first session.update."""
if self.provider_config is None:
return False
return not self.provider_config.requires_session_configuration()
def _should_buffer_client_message_until_setup(self, message: str) -> bool:
if not self._uses_deferred_backend_setup():
return False
if (
self._backend_setup_complete
and not self._flushing_pending_messages_until_setup
):
return False
try:
msg_obj = json.loads(message)
except (json.JSONDecodeError, TypeError):
return False
return msg_obj.get("type") in RealTimeStreaming._CLIENT_AUDIO_BUFFER_TYPES
def _buffer_pending_message_until_setup(self, message: str) -> None:
msg_bytes = len(message.encode("utf-8"))
if (
len(self._pending_messages_until_setup)
< RealTimeStreaming._MAX_BUFFERED_MESSAGES
and self._pending_messages_byte_total + msg_bytes
<= RealTimeStreaming._MAX_BUFFERED_BYTES
):
self._pending_messages_until_setup.append(message)
self._pending_messages_byte_total += msg_bytes
else:
verbose_logger.warning(
"Pre-setup buffer full (%d messages / %d bytes); dropping frame",
len(self._pending_messages_until_setup),
self._pending_messages_byte_total,
)
async def _flush_pending_messages_until_setup(self) -> bool:
pending = self._pending_messages_until_setup
self._pending_messages_until_setup = []
self._pending_messages_byte_total = 0
for idx, message in enumerate(pending):
try:
await self._send_to_backend(message)
except Exception as e:
unsent = pending[idx:]
self._pending_messages_until_setup = (
unsent + self._pending_messages_until_setup
)
self._pending_messages_byte_total = sum(
len(msg.encode("utf-8"))
for msg in self._pending_messages_until_setup
)
verbose_logger.debug(
"Failed to flush buffered client message after setup: %s "
"(%d buffered message(s) retained)",
e,
len(unsent),
)
return False
return True
async def _send_event_to_client(self, event: Any, event_str: str) -> bool:
if self._client_wants_beta and isinstance(event, dict):
try:
translated = self._translate_event_to_beta(event)
if translated is None:
return False
await self.websocket.send_text(json.dumps(translated))
return True
except Exception as e:
verbose_logger.warning(
"Failed to translate %s to beta protocol, forwarding "
"untranslated event to client: %s",
event.get("type"),
e,
)
await self.websocket.send_text(event_str)
return True
def _cache_session_configuration_request(self, transformed_message: str) -> None:
"""Store setup payload once sent to backend.
@@ -547,6 +646,19 @@ class RealTimeStreaming:
isinstance(event, dict) and event.get("type") == "session.created"
)
if is_session_created_event:
if (
self._uses_deferred_backend_setup()
and not self._backend_setup_complete
):
self._backend_setup_complete = True
self._flushing_pending_messages_until_setup = True
try:
while self._pending_messages_until_setup:
flushed = await self._flush_pending_messages_until_setup()
if not flushed:
break
finally:
self._flushing_pending_messages_until_setup = False
if self._session_created_sent_to_client:
# A synthetic session.created (with placeholder defaults) was
# already forwarded to the client when we connected. The
@@ -569,7 +681,7 @@ class RealTimeStreaming:
## update if a prior attempt was dropped by the provider transform.
if is_session_created_event and self._has_audio_transcription_guardrails():
self.store_message(event_str)
await self.websocket.send_text(event_str)
await self._send_event_to_client(event, event_str)
await self._maybe_send_guardrail_turn_detection_update()
continue
## GUARDRAIL: run on transcription events in provider_config path too
@@ -581,7 +693,7 @@ class RealTimeStreaming:
transcript = event.get("transcript", "")
self._collect_user_input_from_backend_event(cast(dict, event))
self.store_message(event_str)
await self.websocket.send_text(event_str)
await self._send_event_to_client(event, event_str)
blocked = await self.run_realtime_guardrails(
cast(str, transcript),
item_id=cast(Optional[str], event.get("item_id")),
@@ -591,7 +703,7 @@ class RealTimeStreaming:
continue
## LOGGING
self.store_message(event_str)
await self.websocket.send_text(event_str)
await self._send_event_to_client(event, event_str)
async def _handle_raw_backend_message(self, raw_response) -> bool:
"""Process a backend message without provider_config (raw path).
@@ -880,6 +992,7 @@ class RealTimeStreaming:
## GUARDRAIL: intercept conversation.item.create for text-based injection.
guardrail_turn_detection_injected = False
msg_type: Optional[str] = None
try:
msg_obj = json.loads(message)
msg_type = msg_obj.get("type")
@@ -1081,6 +1194,29 @@ class RealTimeStreaming:
# actually forward to the backend.
self.store_input(message=message)
if self._should_buffer_client_message_until_setup(message):
self._buffer_pending_message_until_setup(message)
continue
if self._pending_messages_until_setup:
should_send_setup_before_buffered_messages = (
not self._backend_setup_complete
and not self._flushing_pending_messages_until_setup
and msg_type == "session.update"
)
if not should_send_setup_before_buffered_messages:
self._buffer_pending_message_until_setup(message)
if (
self._backend_setup_complete
and not self._flushing_pending_messages_until_setup
):
await self._flush_pending_messages_until_setup()
continue
if self._flushing_pending_messages_until_setup:
self._buffer_pending_message_until_setup(message)
continue
## FORWARD TO BACKEND
# Only mark the guardrail turn_detection update as sent after the
# backend actually accepted the message. Setting the flag earlier
litellm/llms/gemini/realtime/transformation.py
+185 -44
View File
@@ -27,7 +27,6 @@ from litellm.types.llms.gemini import (
)
from litellm.types.llms.openai import (
OpenAIRealtimeContentPartDone,
OpenAIRealtimeConversationItemCreated,
OpenAIRealtimeDoneEvent,
OpenAIRealtimeEvents,
OpenAIRealtimeEventTypes,
@@ -79,6 +78,12 @@ _KNOWN_GEMINI_TOP_LEVEL_KEYS: set = {
map_key.split(".", 1)[0] for map_key in MAP_GEMINI_FIELD_TO_OPENAI_EVENT
}
# Gemini Live native-audio model ids carry this marker (e.g.
# ``gemini-2.5-flash-native-audio-preview-09-2025``). These models reject a
# ``speechConfig`` on ``setup`` with a 1007 invalid-argument error, so it is
# stripped in ``_finalize_gemini_live_setup``.
_GEMINI_NATIVE_AUDIO_MODEL_MARKER = "native-audio"
class GeminiRealtimeConfig(BaseRealtimeConfig):
# Cap the LRU of in-flight tool calls so long sessions with many tool
@@ -98,6 +103,33 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
# bypassing spend and budget accounting.
self._pending_usage_metadata: Optional[dict] = None
@staticmethod
def _usage_detail_alias(details: Any, defaults: Dict[str, int]) -> Dict[str, Any]:
if not isinstance(details, dict):
return dict(defaults)
return {
**defaults,
**{key: value for key, value in details.items() if value is not None},
}
@staticmethod
def _add_pipecat_usage_detail_aliases(usage_dict: Dict[str, Any]) -> Dict[str, Any]:
usage_dict.setdefault(
"input_token_details",
GeminiRealtimeConfig._usage_detail_alias(
usage_dict.get("input_tokens_details"),
{"cached_tokens": 0, "text_tokens": 0, "audio_tokens": 0},
),
)
usage_dict.setdefault(
"output_token_details",
GeminiRealtimeConfig._usage_detail_alias(
usage_dict.get("output_tokens_details"),
{"text_tokens": 0, "audio_tokens": 0},
),
)
return usage_dict
def validate_environment(
self, headers: dict, model: str, api_key: Optional[str] = None
) -> dict:
@@ -173,9 +205,25 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
def map_automatic_turn_detection(
self, value: OpenAIRealtimeTurnDetection
) -> AutomaticActivityDetection:
"""Map OpenAI ``server_vad`` to Gemini ``automaticActivityDetection``.
OpenAI ``semantic_vad`` has no Gemini Live equivalent — return an empty
dict so callers omit ``realtimeInputConfig`` (mapping it with
``disabled: true`` breaks native-audio sessions).
"""
if (
isinstance(value, dict)
and value.get("type") == "semantic_vad"
and "create_response" not in value
):
return AutomaticActivityDetection()
automatic_activity_dection = AutomaticActivityDetection()
if "create_response" in value and isinstance(value["create_response"], bool):
automatic_activity_dection["disabled"] = not value["create_response"]
elif isinstance(value, dict) and value.get("type") == "server_vad":
# OpenAI server VAD enables activity detection by default.
automatic_activity_dection["disabled"] = False
else:
automatic_activity_dection["disabled"] = True
if "prefix_padding_ms" in value and isinstance(value["prefix_padding_ms"], int):
@@ -197,6 +245,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
"tools",
"input_audio_transcription",
"turn_detection",
"voice",
]
def map_openai_params(
@@ -231,17 +280,33 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
optional_params["inputAudioTranscription"] = {}
elif key == "turn_detection":
value_typed = cast(OpenAIRealtimeTurnDetection, value)
if (
isinstance(value_typed, dict)
and value_typed.get("type") == "semantic_vad"
and "create_response" not in value_typed
):
# Pipecat/OpenAI GA semantic VAD — skip; Gemini uses its own VAD.
# Only skip when there is no create_response override so that
# a guardrail-injected create_response:false is not dropped.
continue
transformed_audio_activity_config = self.map_automatic_turn_detection(
value_typed
)
if (
len(transformed_audio_activity_config) > 0
): # if the config is not empty, add it to the optional params
if transformed_audio_activity_config:
optional_params["realtimeInputConfig"] = (
BidiGenerateContentRealtimeInputConfig(
automaticActivityDetection=transformed_audio_activity_config
)
)
elif key == "voice":
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexGeminiConfig,
)
vertex_gemini_config = VertexGeminiConfig()
speech_config = vertex_gemini_config._map_audio_params({"voice": value})
if speech_config:
optional_params["generationConfig"]["speechConfig"] = speech_config
if len(optional_params["generationConfig"]) == 0:
optional_params.pop("generationConfig")
return optional_params
@@ -297,6 +362,9 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
and "transcription" in input_cfg
):
normalized["input_audio_transcription"] = input_cfg["transcription"]
output_cfg = audio.get("output")
if isinstance(output_cfg, dict) and output_cfg.get("voice"):
normalized["voice"] = output_cfg["voice"]
extracted_turn_detection = GeminiRealtimeConfig._extract_turn_detection(
normalized
@@ -308,6 +376,18 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
return normalized
@staticmethod
def _finalize_gemini_live_setup(
model: str, setup: Dict[str, Any]
) -> Dict[str, Any]:
"""Drop fields Gemini Live native-audio rejects on ``setup``."""
if _GEMINI_NATIVE_AUDIO_MODEL_MARKER not in model.lower():
return setup
generation_config = setup.get("generationConfig")
if isinstance(generation_config, dict):
generation_config.pop("speechConfig", None)
return setup
def _handle_session_update(
self,
json_message: dict,
@@ -351,7 +431,11 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
verbose_logger.debug(
"Gemini Realtime: Sending initial setup with tools to backend"
)
return [json.dumps({"setup": new_overrides})]
return [
json.dumps(
{"setup": self._finalize_gemini_live_setup(model, new_overrides)}
)
]
if not new_overrides:
verbose_logger.debug(
@@ -420,7 +504,15 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
verbose_logger.debug(
"Gemini Realtime: Forwarding session.update as follow-up setup"
)
return [json.dumps({"setup": follow_up_setup})]
return [
json.dumps(
{
"setup": self._finalize_gemini_live_setup(
model, cast(Dict[str, Any], follow_up_setup)
)
}
)
]
def _handle_conversation_item(self, json_message: dict) -> List[str]:
"""
@@ -666,6 +758,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
"object": "realtime.response",
"id": response_id,
"status": "in_progress",
"status_details": None,
"output": [],
"conversation_id": conversation_id,
"modalities": _modalities,
@@ -675,9 +768,10 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
)
response_items.append(response_created)
## - return response.output_item.added ← adds ‘item_id’ same for all subsequent events
## - return response.output_item.added
response_output_item_added = OpenAIRealtimeStreamResponseOutputItemAdded(
type="response.output_item.added",
event_id="event_{}".format(uuid.uuid4()),
response_id=response_id,
output_index=0,
item={
@@ -690,20 +784,28 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
},
)
response_items.append(response_output_item_added)
## - return conversation.item.created
conversation_item_created = OpenAIRealtimeConversationItemCreated(
type="conversation.item.created",
event_id="event_{}".format(uuid.uuid4()),
item={
"id": output_item_id,
"object": "realtime.item",
"type": "message",
"status": "in_progress",
"role": "assistant",
"content": [],
},
## - return conversation.item.added
# Pipecat 1.3.x handles "conversation.item.added" (not ".created").
# Sending ".created" raises "Unimplemented server event type" which
# kills the receive task handler.
response_items.append(
cast(
OpenAIRealtimeEvents,
{
"type": "conversation.item.added",
"event_id": "event_{}".format(uuid.uuid4()),
"previous_item_id": None,
"item": {
"id": output_item_id,
"object": "realtime.item",
"type": "message",
"status": "in_progress",
"role": "assistant",
"content": [],
},
},
)
)
response_items.append(conversation_item_created)
## - return response.content_part.added
response_content_part_added = OpenAIRealtimeResponseContentPartAdded(
type="response.content_part.added",
@@ -749,9 +851,9 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
return OpenAIRealtimeResponseDelta(
type=(
"response.text.delta"
"response.output_text.delta"
if delta_type == "text"
else "response.audio.delta"
else "response.output_audio.delta"
),
content_index=0,
event_id="event_{}".format(uuid.uuid4()),
@@ -778,7 +880,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
current_response_id = "resp_{}".format(uuid.uuid4())
if delta_type == "text":
return OpenAIRealtimeResponseTextDone(
type="response.text.done",
type="response.output_text.done",
content_index=0,
event_id="event_{}".format(uuid.uuid4()),
item_id=current_output_item_id,
@@ -788,7 +890,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
)
elif delta_type == "audio":
return OpenAIRealtimeResponseAudioDone(
type="response.audio.done",
type="response.output_audio.done",
content_index=0,
event_id="event_{}".format(uuid.uuid4()),
item_id=current_output_item_id,
@@ -914,7 +1016,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
events: List[OpenAIRealtimeFunctionCallArgumentsDone] = []
for idx, fc in enumerate(function_calls):
call_id = fc.get("id", "")
call_id = fc.get("id", "") or f"call_{uuid.uuid4().hex[:16]}"
name = fc.get("name", "")
# Store call_id → name mapping for round-trip. Use an LRU so
@@ -962,7 +1064,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
current_delta_chunks = []
any_delta_chunk = False
for event in transformed_message:
if event["type"] == "response.text.delta":
if event["type"] == "response.output_text.delta":
current_delta_chunks.append(
cast(OpenAIRealtimeResponseDelta, event)
)
@@ -973,7 +1075,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
)
else:
if (
transformed_message["type"] == "response.text.delta"
transformed_message["type"] == "response.output_text.delta"
): # ONLY ACCUMULATE TEXT DELTA CHUNKS - AUDIO WILL CAUSE SERVER MEMORY ISSUES
if current_delta_chunks is None:
current_delta_chunks = []
@@ -1067,6 +1169,8 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
responses_api_usage = LiteLLMCompletionResponsesConfig._transform_chat_completion_usage_to_responses_usage(
_chat_completion_usage,
)
_usage_dict = responses_api_usage.model_dump()
self._add_pipecat_usage_detail_aliases(_usage_dict)
response_done_event = OpenAIRealtimeDoneEvent(
type="response.done",
event_id="event_{}".format(uuid.uuid4()),
@@ -1074,6 +1178,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
object="realtime.response",
id=current_response_id,
status="completed",
status_details=None, # type: ignore[typeddict-item]
output=(
[output_item["item"] for output_item in output_items]
if output_items
@@ -1081,7 +1186,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
),
conversation_id=current_conversation_id,
modalities=_modalities,
usage=responses_api_usage.model_dump(),
usage=_usage_dict,
),
)
if temperature is not None:
@@ -1294,19 +1399,36 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
output_tx = server_content.get("outputTranscription")
if isinstance(output_tx, dict) and output_tx.get("text"):
if current_response_id is None:
current_response_id = "resp_{}".format(uuid.uuid4())
if current_output_item_id is None:
current_output_item_id = "item_{}".format(uuid.uuid4())
current_conversation_id = (
current_conversation_id or "conv_{}".format(uuid.uuid4())
)
returned_message.extend(
self.return_new_content_delta_events(
session_configuration_request=session_configuration_request,
response_id=current_response_id,
output_item_id=current_output_item_id,
conversation_id=current_conversation_id,
delta_type="audio",
)
)
# Emit as the GA event name; _GA_TO_BETA_EVENT_TYPES translates
# this back to response.audio_transcript.delta for beta clients.
returned_message.append(
cast(
OpenAIRealtimeEvents,
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"event_id": "event_{}".format(uuid.uuid4()),
"delta": output_tx["text"],
"item_id": current_output_item_id
or "item_{}".format(uuid.uuid4()),
"response_id": current_response_id
or "resp_{}".format(uuid.uuid4()),
"output_index": 0,
"transcript": output_tx["text"],
"item_id": current_output_item_id,
"content_index": 0,
"output_index": 0,
"response_id": current_response_id,
"delta": output_tx["text"],
},
)
)
@@ -1416,6 +1538,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
"object": "realtime.response",
"id": current_response_id,
"status": "in_progress",
"status_details": None,
"output": [],
"conversation_id": current_conversation_id,
"modalities": tool_call_modalities,
@@ -1460,6 +1583,29 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
},
)
)
# conversation.item.added — Pipecat 1.3.x registers the
# call_id into _pending_function_calls inside
# _handle_evt_conversation_item_added, which is triggered
# by this event (NOT by response.output_item.added and NOT
# by the old conversation.item.created which Pipecat 1.3.x
# does not handle). Without this event the subsequent
# response.function_call_arguments.done finds an empty
# pending-calls dict and drops the tool invocation silently.
returned_message.append(
cast(
OpenAIRealtimeEvents,
{
"type": "conversation.item.added",
"event_id": f"event_{uuid.uuid4()}",
"previous_item_id": None,
"item": {
**function_call_item,
"status": "in_progress",
"arguments": "",
},
},
)
)
# response.function_call_arguments.delta — Gemini delivers
# the full arguments string in a single toolCall frame
# rather than streaming partial chunks, so emit one delta
@@ -1496,14 +1642,6 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
item={**function_call_item},
)
)
# conversation.item.created
returned_message.append(
OpenAIRealtimeConversationItemCreated(
type="conversation.item.created",
event_id=f"event_{uuid.uuid4()}",
item={**function_call_item},
)
)
# response.done - close the response so clients can submit tool
# results. Mirror the non-tool-call RESPONSE_DONE path: if Gemini
@@ -1537,6 +1675,8 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
tool_call_responses_api_usage = LiteLLMCompletionResponsesConfig._transform_chat_completion_usage_to_responses_usage(
_tool_call_chat_completion_usage,
)
_tool_usage_dict = tool_call_responses_api_usage.model_dump()
self._add_pipecat_usage_detail_aliases(_tool_usage_dict)
tool_call_done_event = OpenAIRealtimeDoneEvent(
type="response.done",
event_id=f"event_{uuid.uuid4()}",
@@ -1544,6 +1684,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
id=current_response_id,
object="realtime.response",
status="completed",
status_details=None, # type: ignore[typeddict-item]
output=[
{
"id": te["item_id"],
@@ -1558,7 +1699,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
],
conversation_id=current_conversation_id,
modalities=tool_call_modalities,
usage=tool_call_responses_api_usage.model_dump(),
usage=_tool_usage_dict,
),
)
tool_call_temperature = tool_call_generation_config.get("temperature")
tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
+340
View File
@@ -293,6 +293,51 @@ def test_translate_event_to_beta_drops_conversation_item_done():
)
@pytest.mark.asyncio
async def test_provider_config_path_translates_ga_events_for_beta_clients():
client_ws = MagicMock()
client_ws.scope = {"headers": [(b"openai-beta", b"realtime=v1")]}
client_ws.send_text = AsyncMock()
backend_ws = MagicMock()
logging_obj = MagicMock()
provider_config = MagicMock()
provider_config.transform_realtime_response = MagicMock(
return_value={
"response": [
{
"type": "response.output_text.delta",
"event_id": "event_1",
"delta": "hello",
},
{"type": "conversation.item.done", "event_id": "event_2"},
],
"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,
}
)
streaming = RealTimeStreaming(
client_ws,
backend_ws,
logging_obj,
provider_config=provider_config,
model="gemini-2.5-flash",
)
await streaming._handle_provider_config_message("{}")
assert client_ws.send_text.await_count == 1
sent = json.loads(client_ws.send_text.await_args.args[0])
assert sent["type"] == "response.text.delta"
assert sent["delta"] == "hello"
def test_client_sent_openai_beta_realtime_header_detects_header():
ws = MagicMock()
ws.scope = {"headers": [(b"openai-beta", b"realtime=v1")]}
@@ -1770,3 +1815,298 @@ async def test_follow_up_setup_updates_cached_session_configuration_request():
await streaming.client_ack_messages()
assert streaming.session_configuration_request == follow_up_setup
@pytest.mark.asyncio
async def test_deferred_setup_buffers_audio_until_backend_setup_complete(monkeypatch):
"""Pipecat may send audio before session.update when setup is deferred."""
monkeypatch.setattr(litellm, "gemini_live_defer_setup", True, raising=False)
from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
client_ws = MagicMock()
audio_msg = json.dumps({"type": "input_audio_buffer.append", "audio": "AA=="})
client_ws.receive_text = AsyncMock(
side_effect=[audio_msg, ConnectionClosed(None, None)]
)
backend_ws = MagicMock()
backend_ws.send = AsyncMock()
logging_obj = MagicMock()
config = GeminiRealtimeConfig()
streaming = RealTimeStreaming(
client_ws,
backend_ws,
logging_obj,
provider_config=config,
model="gemini-live-2.5-flash-native-audio",
)
assert streaming._backend_setup_complete is False
await streaming.client_ack_messages()
backend_ws.send.assert_not_called()
assert len(streaming._pending_messages_until_setup) == 1
streaming._backend_setup_complete = True
await streaming._flush_pending_messages_until_setup()
assert backend_ws.send.call_count == 1
@pytest.mark.asyncio
async def test_deferred_setup_sends_session_update_before_buffered_audio(monkeypatch):
monkeypatch.setattr(litellm, "gemini_live_defer_setup", True, raising=False)
from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
client_ws = MagicMock()
audio_msg = json.dumps({"type": "input_audio_buffer.append", "audio": "AA=="})
session_update = json.dumps(
{"type": "session.update", "session": {"modalities": ["audio"]}}
)
client_ws.receive_text = AsyncMock(
side_effect=[audio_msg, session_update, ConnectionClosed(None, None)]
)
backend_ws = MagicMock()
backend_ws.send = AsyncMock()
logging_obj = MagicMock()
config = GeminiRealtimeConfig()
streaming = RealTimeStreaming(
client_ws,
backend_ws,
logging_obj,
provider_config=config,
model="gemini-live-2.5-flash-native-audio",
)
await streaming.client_ack_messages()
assert backend_ws.send.await_count == 1
sent_payload = json.loads(backend_ws.send.await_args_list[0].args[0])
assert "setup" in sent_payload
assert "realtimeInput" not in sent_payload
assert streaming._pending_messages_until_setup == [audio_msg]
@pytest.mark.asyncio
async def test_deferred_setup_flush_buffers_audio_received_during_flush():
import asyncio
client_ws = MagicMock()
client_ws.send_text = AsyncMock()
new_audio_msg = json.dumps(
{"type": "input_audio_buffer.append", "audio": "new-audio"}
)
client_ws.receive_text = AsyncMock(
side_effect=[new_audio_msg, ConnectionClosed(None, None)]
)
backend_ws = MagicMock()
logging_obj = MagicMock()
provider_config = MagicMock()
provider_config.requires_session_configuration = MagicMock(return_value=False)
provider_config.transform_realtime_response = MagicMock(
return_value={
"response": {
"type": "session.created",
"event_id": "event_1",
"session": {"id": "sess_1", "modalities": ["audio"]},
},
"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,
}
)
streaming = RealTimeStreaming(
websocket=client_ws,
backend_ws=backend_ws,
logging_obj=logging_obj,
provider_config=provider_config,
model="gemini-live-2.5-flash-native-audio",
)
old_audio_msg = json.dumps(
{"type": "input_audio_buffer.append", "audio": "old-audio"}
)
streaming._pending_messages_until_setup = [old_audio_msg]
streaming._pending_messages_byte_total = len(old_audio_msg.encode("utf-8"))
first_flush_started = asyncio.Event()
release_flush = asyncio.Event()
sent_messages = []
async def send_to_backend(message):
sent_messages.append(message)
if message == old_audio_msg:
first_flush_started.set()
await release_flush.wait()
return True
streaming._send_to_backend = send_to_backend # type: ignore[method-assign]
setup_task = asyncio.create_task(
streaming._handle_provider_config_message(json.dumps({"setupComplete": {}}))
)
await asyncio.wait_for(first_flush_started.wait(), timeout=1)
await streaming.client_ack_messages()
assert sent_messages == [old_audio_msg]
assert streaming._pending_messages_until_setup == [new_audio_msg]
release_flush.set()
await asyncio.wait_for(setup_task, timeout=1)
assert sent_messages == [old_audio_msg, new_audio_msg]
assert streaming._pending_messages_until_setup == []
@pytest.mark.asyncio
async def test_deferred_setup_flush_retains_unsent_messages_after_send_failure():
client_ws = MagicMock()
backend_ws = MagicMock()
logging_obj = MagicMock()
streaming = RealTimeStreaming(client_ws, backend_ws, logging_obj)
buffered_messages = [
json.dumps({"type": "input_audio_buffer.append", "audio": "AA=="}),
json.dumps({"type": "input_audio_buffer.commit"}),
]
streaming._pending_messages_until_setup = list(buffered_messages)
streaming._pending_messages_byte_total = sum(
len(message.encode("utf-8")) for message in buffered_messages
)
streaming._send_to_backend = AsyncMock( # type: ignore[method-assign]
side_effect=Exception("transient")
)
await streaming._flush_pending_messages_until_setup()
assert streaming._pending_messages_until_setup == buffered_messages
assert streaming._pending_messages_byte_total == sum(
len(message.encode("utf-8")) for message in buffered_messages
)
streaming._send_to_backend = AsyncMock(return_value=True) # type: ignore[method-assign]
await streaming._flush_pending_messages_until_setup()
assert streaming._pending_messages_until_setup == []
assert streaming._pending_messages_byte_total == 0
assert streaming._send_to_backend.await_count == 2
@pytest.mark.asyncio
async def test_deferred_setup_flushes_audio_on_backend_session_created(monkeypatch):
"""Buffered audio is released when Gemini setupComplete becomes session.created."""
monkeypatch.setattr(litellm, "gemini_live_defer_setup", True, raising=False)
from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
client_ws = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws = MagicMock()
backend_ws.send = AsyncMock()
backend_ws.recv = AsyncMock(
side_effect=[
json.dumps({"setupComplete": {}}).encode(),
ConnectionClosed(None, None),
]
)
logging_obj = MagicMock()
logging_obj.litellm_trace_id = "trace_defer"
logging_obj.async_success_handler = AsyncMock()
logging_obj.success_handler = MagicMock()
config = GeminiRealtimeConfig()
streaming = RealTimeStreaming(
client_ws,
backend_ws,
logging_obj,
provider_config=config,
model="gemini-live-2.5-flash-native-audio",
)
streaming._pending_messages_until_setup.append(
json.dumps({"type": "input_audio_buffer.append", "audio": "AA=="})
)
await streaming.backend_to_client_send_messages()
assert streaming._backend_setup_complete is True
assert streaming._pending_messages_until_setup == []
assert backend_ws.send.call_count == 1
@pytest.mark.asyncio
async def test_deferred_setup_caps_non_audio_buffered_messages(monkeypatch):
"""A client that withholds session.update cannot grow the pre-setup buffer
without bound by streaming non-audio frames after the first audio frame."""
monkeypatch.setattr(litellm, "gemini_live_defer_setup", True, raising=False)
from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
cap = RealTimeStreaming._MAX_BUFFERED_MESSAGES
audio_msg = json.dumps({"type": "input_audio_buffer.append", "audio": "AA=="})
flood_msg = json.dumps({"type": "foo", "data": "x" * 1024})
client_ws = MagicMock()
client_ws.receive_text = AsyncMock(
side_effect=[audio_msg]
+ [flood_msg] * (cap + 50)
+ [ConnectionClosed(None, None)]
)
backend_ws = MagicMock()
backend_ws.send = AsyncMock()
logging_obj = MagicMock()
streaming = RealTimeStreaming(
client_ws,
backend_ws,
logging_obj,
provider_config=GeminiRealtimeConfig(),
model="gemini-live-2.5-flash-native-audio",
)
assert streaming._backend_setup_complete is False
await streaming.client_ack_messages()
backend_ws.send.assert_not_called()
assert len(streaming._pending_messages_until_setup) == cap
assert (
streaming._pending_messages_byte_total <= RealTimeStreaming._MAX_BUFFERED_BYTES
)
@pytest.mark.asyncio
async def test_deferred_setup_caps_non_audio_buffered_bytes(monkeypatch):
"""Non-audio frames appended after the first audio frame honor the byte budget."""
monkeypatch.setattr(litellm, "gemini_live_defer_setup", True, raising=False)
from litellm.llms.gemini.realtime.transformation import GeminiRealtimeConfig
audio_msg = json.dumps({"type": "input_audio_buffer.append", "audio": "AA=="})
big_non_audio = json.dumps(
{"type": "foo", "data": "x" * (RealTimeStreaming._MAX_BUFFERED_BYTES + 1)}
)
client_ws = MagicMock()
client_ws.receive_text = AsyncMock(
side_effect=[audio_msg, big_non_audio, ConnectionClosed(None, None)]
)
backend_ws = MagicMock()
backend_ws.send = AsyncMock()
logging_obj = MagicMock()
streaming = RealTimeStreaming(
client_ws,
backend_ws,
logging_obj,
provider_config=GeminiRealtimeConfig(),
model="gemini-live-2.5-flash-native-audio",
)
await streaming.client_ack_messages()
assert streaming._pending_messages_until_setup == [audio_msg]
assert (
streaming._pending_messages_byte_total <= RealTimeStreaming._MAX_BUFFERED_BYTES
)
tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py
+182 -11
View File
@@ -235,12 +235,73 @@ def test_gemini_realtime_transformation_audio_delta():
contains_audio_delta = False
for response in responses:
if response["type"] == OpenAIRealtimeEventTypes.RESPONSE_AUDIO_DELTA.value:
if (
response["type"]
== OpenAIRealtimeEventTypes.RESPONSE_OUTPUT_AUDIO_DELTA.value
):
contains_audio_delta = True
break
assert contains_audio_delta, "Expected audio delta event"
def test_gemini_output_audio_transcript_delta_uses_active_response_ids():
config = GeminiRealtimeConfig()
session_configuration_request = {
"setup": {
"model": "gemini-1.5-flash",
"generationConfig": {"responseModalities": ["AUDIO"]},
}
}
session_configuration_request_str = json.dumps(session_configuration_request)
event = {
"serverContent": {
"outputTranscription": {"text": "Hello from Gemini."},
"modelTurn": {
"parts": [
{"inlineData": {"mimeType": "audio/pcm", "data": "my-audio-data"}}
]
},
}
}
result = config.transform_realtime_response(
json.dumps(event),
"gemini-1.5-flash",
MagicMock(),
realtime_response_transform_input={
"session_configuration_request": session_configuration_request_str,
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": [],
"current_item_chunks": [],
"current_delta_type": None,
},
)
responses = result["response"]
response_created = next(
response for response in responses if response["type"] == "response.created"
)
transcript_delta = next(
response
for response in responses
if response["type"] == "response.output_audio_transcript.delta"
)
audio_delta = next(
response
for response in responses
if response["type"] == "response.output_audio.delta"
)
assert transcript_delta["response_id"] == response_created["response"]["id"]
assert transcript_delta["response_id"] == audio_delta["response_id"]
assert transcript_delta["item_id"] == audio_delta["item_id"]
assert result["current_response_id"] == transcript_delta["response_id"]
assert result["current_output_item_id"] == transcript_delta["item_id"]
def test_gemini_realtime_transformation_generation_complete():
from litellm.types.llms.openai import OpenAIRealtimeEventTypes
@@ -278,7 +339,10 @@ def test_gemini_realtime_transformation_generation_complete():
contains_audio_done_event = False
for response in responses:
if response["type"] == OpenAIRealtimeEventTypes.RESPONSE_AUDIO_DONE.value:
if (
response["type"]
== OpenAIRealtimeEventTypes.RESPONSE_OUTPUT_AUDIO_DONE.value
):
contains_audio_done_event = True
break
assert contains_audio_done_event, "Expected audio done event"
@@ -735,7 +799,14 @@ def test_gemini_tool_call_emits_response_created_preamble():
)
responses = result["response"]
# Should have: response.created, output_item.added, function_call_arguments.delta, function_call_arguments.done, output_item.done, conversation.item.created, response.done
# Expected sequence:
# 0: response.created
# 1: response.output_item.added (item status=in_progress)
# 2: conversation.item.added (registers call_id in Pipecat's _pending_function_calls)
# 3: response.function_call_arguments.delta
# 4: response.function_call_arguments.done
# 5: response.output_item.done
# 6: response.done
assert len(responses) >= 7
assert responses[0]["type"] == "response.created"
assert "response" in responses[0]
@@ -749,14 +820,14 @@ def test_gemini_tool_call_emits_response_created_preamble():
assert responses[1]["type"] == "response.output_item.added"
assert responses[1]["item"]["type"] == "function_call"
assert responses[1]["item"]["status"] == "in_progress"
assert responses[2]["type"] == "response.function_call_arguments.delta"
assert responses[2]["call_id"] == "call_123"
assert responses[2]["delta"] == responses[3]["arguments"]
assert responses[3]["type"] == "response.function_call_arguments.done"
assert responses[4]["type"] == "response.output_item.done"
assert responses[4]["item"]["type"] == "function_call"
assert responses[4]["item"]["status"] == "completed"
assert responses[5]["type"] == "conversation.item.created"
assert responses[2]["type"] == "conversation.item.added"
assert responses[2]["item"]["type"] == "function_call"
assert responses[2]["item"]["call_id"] == "call_123"
assert responses[3]["type"] == "response.function_call_arguments.delta"
assert responses[3]["call_id"] == "call_123"
assert responses[3]["delta"] == responses[4]["arguments"]
assert responses[4]["type"] == "response.function_call_arguments.done"
assert responses[5]["type"] == "response.output_item.done"
assert responses[5]["item"]["type"] == "function_call"
assert responses[5]["item"]["status"] == "completed"
assert responses[6]["type"] == "response.done"
@@ -930,6 +1001,12 @@ def test_gemini_tool_call_response_done_includes_usage_from_sibling_metadata():
"promptTokenCount": 17,
"responseTokenCount": 4,
"totalTokenCount": 21,
"promptTokensDetails": [
{"modality": "TEXT", "tokenCount": 17},
],
"responseTokensDetails": [
{"modality": "TEXT", "tokenCount": 4},
],
},
}
),
@@ -953,6 +1030,8 @@ def test_gemini_tool_call_response_done_includes_usage_from_sibling_metadata():
assert usage["input_tokens"] == 17
assert usage["output_tokens"] == 4
assert usage["total_tokens"] == 21
assert usage["input_token_details"]["text_tokens"] == 17
assert usage["output_token_details"]["text_tokens"] == 4
def test_gemini_tool_call_response_done_falls_back_to_empty_usage():
@@ -1120,6 +1199,90 @@ def test_gemini_subsequent_session_update_forwards_tools_merged_with_original_se
assert follow_up["inputAudioTranscription"] == {}
def test_gemini_realtime_pipecat_ga_session_voice_and_tools():
"""Pipecat OpenAIRealtimeSessionProperties: output_modalities, nested tools,
and audio.output.voice (e.g. Kore) must map into Gemini setup."""
config = GeminiRealtimeConfig()
session_update = {
"type": "session.update",
"session": {
"output_modalities": ["audio"],
"instructions": "Follow system instructions.",
"tools": [
{
"type": "function",
"function": {
"name": "terminate_call",
"description": "End the call.",
"parameters": {"type": "object", "properties": {}},
},
}
],
"audio": {
"input": {
"format": {"type": "audio/pcm", "rate": 24000},
"turn_detection": {"type": "server_vad"},
},
"output": {
"format": {"type": "audio/pcm", "rate": 24000},
"voice": "Kore",
},
},
"temperature": 0,
},
}
messages = config.transform_realtime_request(
json.dumps(session_update),
"gemini-2.5-flash-native-audio",
session_configuration_request=None,
)
assert len(messages) == 1
setup = json.loads(messages[0])["setup"]
assert setup["generationConfig"]["responseModalities"] == ["AUDIO"]
# Native-audio Live rejects speechConfig on setup (see _finalize_gemini_live_setup).
assert "speechConfig" not in setup.get("generationConfig", {})
assert setup["tools"][0]["function_declarations"][0]["name"] == "terminate_call"
assert (
setup["realtimeInputConfig"]["automaticActivityDetection"]["disabled"] is False
)
def test_gemini_realtime_pipecat_semantic_vad_omits_realtime_input_config():
"""Pipecat SemanticTurnDetection (semantic_vad) must not map to disabled VAD."""
config = GeminiRealtimeConfig()
session_update = {
"type": "session.update",
"session": {
"output_modalities": ["audio"],
"instructions": "test",
"audio": {
"input": {"turn_detection": {"type": "semantic_vad"}},
},
"tools": [
{
"type": "function",
"function": {
"name": "terminate_call",
"description": "End call.",
"parameters": {"type": "object", "properties": {}},
},
}
],
},
}
messages = config.transform_realtime_request(
json.dumps(session_update),
"gemini-live-2.5-flash-native-audio",
session_configuration_request=None,
)
setup = json.loads(messages[0])["setup"]
assert "realtimeInputConfig" not in setup
assert setup["tools"][0]["function_declarations"][0]["name"] == "terminate_call"
def test_gemini_subsequent_session_update_with_turn_detection_only_preserves_original_tools():
"""A subsequent session.update carrying only turn_detection (the
guardrail-injected disable) must keep the original tools/generationConfig."""
@@ -1407,6 +1570,12 @@ def test_gemini_standalone_usage_metadata_is_attributed_to_next_response_done():
"promptTokenCount": 5,
"responseTokenCount": 11,
"totalTokenCount": 16,
"promptTokensDetails": [
{"modality": "TEXT", "tokenCount": 5},
],
"responseTokensDetails": [
{"modality": "TEXT", "tokenCount": 11},
],
}
}
),
@@ -1447,6 +1616,8 @@ def test_gemini_standalone_usage_metadata_is_attributed_to_next_response_done():
assert usage["input_tokens"] == 5
assert usage["output_tokens"] == 11
assert usage["total_tokens"] == 16
assert usage["input_token_details"]["text_tokens"] == 5
assert usage["output_token_details"]["text_tokens"] == 11
assert config._pending_usage_metadata is None
tests/test_litellm/llms/vertex_ai/realtime/test_vertex_ai_realtime_transformation.py
+2 -2
View File
@@ -278,8 +278,8 @@ async def test_vertex_realtime_text_in_text_out():
assert session_created_msgs, "Expected session.created to be sent to client"
# At least one text delta should have been forwarded
text_delta_msgs = [m for m in sent_to_client if '"response.text.delta"' in m]
assert text_delta_msgs, "Expected response.text.delta to be sent to client"
text_delta_msgs = [m for m in sent_to_client if '"response.output_text.delta"' in m]
assert text_delta_msgs, "Expected response.output_text.delta to be sent to client"
# Verify the delta contains the model's text
delta_obj = json.loads(text_delta_msgs[0])