""" This file contains the transformation logic for the Gemini realtime API. """ import json from collections import OrderedDict from collections.abc import Mapping, Sequence from typing import Any, Final, cast import litellm from litellm import verbose_logger from litellm._uuid import uuid from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.base_llm.realtime.transformation import BaseRealtimeConfig from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import ( VertexGeminiConfig, ) from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) from litellm.types.llms.gemini import ( AutomaticActivityDetection, BidiGenerateContentRealtimeInput, BidiGenerateContentRealtimeInputConfig, BidiGenerateContentServerContent, BidiGenerateContentServerMessage, BidiGenerateContentSetup, ) from litellm.types.llms.openai import ( OpenAIRealtimeContentPartDone, OpenAIRealtimeDoneEvent, OpenAIRealtimeEvents, OpenAIRealtimeEventTypes, OpenAIRealtimeFunctionCallArgumentsDone, OpenAIRealtimeOutputItemDone, OpenAIRealtimeResponseAudioDone, OpenAIRealtimeResponseContentPartAdded, OpenAIRealtimeResponseDelta, OpenAIRealtimeResponseDoneObject, OpenAIRealtimeResponseTextDone, OpenAIRealtimeStreamResponseBaseObject, OpenAIRealtimeStreamResponseOutputItem, OpenAIRealtimeStreamResponseOutputItemAdded, OpenAIRealtimeStreamSession, OpenAIRealtimeStreamSessionEvents, OpenAIRealtimeTurnDetection, ResponsesAPIStreamEvents, ) from litellm.types.llms.vertex_ai import ( GeminiResponseModalities, HttpxBlobType, HttpxContentType, ) from litellm.types.realtime import ( ALL_DELTA_TYPES, RealtimeInputAudioTranscriptionUsage, RealtimeModalityResponseTransformOutput, RealtimeResponseTransformInput, RealtimeResponseTypedDict, ) from litellm.utils import get_empty_usage from ..common_utils import encode_unserializable_types, get_api_key_from_env MAP_GEMINI_FIELD_TO_OPENAI_EVENT: Final[dict[str, OpenAIRealtimeEventTypes | ResponsesAPIStreamEvents]] = { "setupComplete": OpenAIRealtimeEventTypes.SESSION_CREATED, "serverContent.generationComplete": OpenAIRealtimeEventTypes.RESPONSE_TEXT_DONE, "serverContent.turnComplete": OpenAIRealtimeEventTypes.RESPONSE_DONE, "serverContent.interrupted": OpenAIRealtimeEventTypes.RESPONSE_DONE, "toolCall": ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE, } # Keys the main transform loop handles; siblings like ``usageMetadata`` are skipped. _KNOWN_GEMINI_TOP_LEVEL_KEYS: Final[set] = {map_key.split(".", 1)[0] for map_key in MAP_GEMINI_FIELD_TO_OPENAI_EVENT} OPENAI_STOCK_REALTIME_VOICES: Final[frozenset[str]] = frozenset( {"alloy", "ash", "ballad", "cedar", "coral", "echo", "marin", "sage", "shimmer", "verse"} ) def _gemini_live_speech_config(voice: object) -> Mapping[str, object] | None: """Build the Gemini Live speechConfig for a client-requested voice. OpenAI stock voice names have no Gemini equivalent and Gemini Live closes the session on an unknown voice, so they are dropped with a warning and the model keeps its default voice. Every other name is forwarded verbatim. """ if isinstance(voice, str) and voice.lower() in OPENAI_STOCK_REALTIME_VOICES: verbose_logger.warning( "Gemini Realtime: voice %s is an OpenAI voice with no Gemini equivalent; " "dropping it so the session keeps the model's default voice.", voice, ) return None return VertexGeminiConfig()._map_audio_params({"voice": voice}) # Google bills Live transcription at an estimated 25 audio tokens/sec of input and # 175 text tokens/min of output (ai.google.dev/gemini-api/docs/pricing). GEMINI_LIVE_TRANSCRIBE_AUDIO_TOKENS_PER_SECOND: Final = 25 GEMINI_LIVE_TRANSCRIBE_OUTPUT_TEXT_TOKENS_PER_MINUTE: Final = 175 PCM16_INPUT_AUDIO_BYTES_PER_SECOND: Final = 48000 def _base64_decoded_byte_count(data: str) -> int: padding: Final = 2 if data.endswith("==") else 1 if data.endswith("=") else 0 return max(len(data) * 3 // 4 - padding, 0) class GeminiRealtimeConfig(BaseRealtimeConfig): _TOOL_CALL_ID_TO_NAME_MAX = 256 # LRU cap for call_id→name mapping def __init__(self): super().__init__() self._tool_call_id_to_name: OrderedDict[str, str] = OrderedDict() # Gemini Live sometimes emits usageMetadata in a standalone frame between # turns; buffer it here so the next response.done carries the token counts. self._pending_usage_metadata: dict | None = None self._unbilled_input_audio_bytes: int = 0 def is_setup_message(self, msg_obj: dict) -> bool: return "setup" in msg_obj def is_content_message(self, msg_obj: dict) -> bool: return any(k in msg_obj for k in ("realtimeInput", "clientContent", "toolResponse")) def _include_function_response_id(self) -> bool: """Google AI Studio Gemini 3.5+ accepts ``id`` on functionResponses; Vertex AI rejects it.""" return True @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: str | None = None) -> dict: return headers def get_complete_url(self, api_base: str | None, model: str, api_key: str | None = None) -> str: """ Example output: "BACKEND_WS_URL = "wss://generativelanguage.googleapis.com/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent""; """ if api_base is None: api_base = "wss://generativelanguage.googleapis.com" if api_key is None: api_key = get_api_key_from_env() if api_key is None: raise ValueError("api_key is required for Gemini API calls") api_base = api_base.replace("https://", "wss://") api_base = api_base.replace("http://", "ws://") # WebSocket connections do not support custom HTTP headers in all clients, # so the API key must remain as a query parameter here. This is an accepted # limitation; httpx is not used for WebSocket so MaskedHTTPStatusError # already covers the main leak vector. return f"{api_base}/ws/google.ai.generativelanguage.v1beta.GenerativeService.BidiGenerateContent?key={api_key}" def map_model_turn_event(self, model_turn: HttpxContentType) -> OpenAIRealtimeEventTypes: """ Map the model turn event to the OpenAI realtime events. Returns either: - response.text.delta - model_turn: {"parts": [{"text": "..."}]} - response.audio.delta - model_turn: {"parts": [{"inlineData": {"mimeType": "audio/pcm", "data": "..."}}]} Assumes parts is a single element list. """ if "parts" in model_turn: parts: Final = model_turn["parts"] if len(parts) != 1: verbose_logger.warning("Realtime: Expected 1 part, got %s for Gemini model turn event.", len(parts)) part: Final = parts[0] if "text" in part: return OpenAIRealtimeEventTypes.RESPONSE_TEXT_DELTA elif "inlineData" in part: return OpenAIRealtimeEventTypes.RESPONSE_AUDIO_DELTA else: raise ValueError(f"Unexpected part type: {part}") raise ValueError(f"Unexpected model turn event, no 'parts' key: {model_turn}") def map_generation_complete_event(self, delta_type: ALL_DELTA_TYPES | None) -> OpenAIRealtimeEventTypes: if delta_type == "text": return OpenAIRealtimeEventTypes.RESPONSE_TEXT_DONE elif delta_type == "audio": return OpenAIRealtimeEventTypes.RESPONSE_AUDIO_DONE else: raise ValueError(f"Unexpected delta type: {delta_type}") def get_audio_mime_type(self, input_audio_format: str = "pcm16"): mime_types: Final = { "pcm16": "audio/pcm;rate=24000", "g711_ulaw": "audio/pcmu", "g711_alaw": "audio/pcma", } return mime_types.get(input_audio_format, "application/octet-stream") def _manual_turn_detection_enabled(self, session_configuration_request: str | None) -> bool: if not session_configuration_request: return False try: setup: Final = json.loads(session_configuration_request).get("setup", {}) automatic_detection: Final = setup.get("realtimeInputConfig", {}).get("automaticActivityDetection", {}) return isinstance(automatic_detection, dict) and automatic_detection.get("disabled") is True except (json.JSONDecodeError, TypeError, AttributeError): return False def _handle_input_audio_buffer_commit_or_end(self, session_configuration_request: str | None) -> list[str]: """Map OpenAI buffer commit/end to Gemini Live turn-boundary signals.""" if self._manual_turn_detection_enabled(session_configuration_request): realtime_input_dict: BidiGenerateContentRealtimeInput = { "activityEnd": True, } verbose_logger.debug("Gemini Realtime: Sending activityEnd realtimeInput to backend") else: realtime_input_dict = {"audioStreamEnd": True} verbose_logger.debug("Gemini Realtime: Sending audioStreamEnd realtimeInput to backend") return [json.dumps({"realtimeInput": realtime_input_dict})] 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: Final = 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): automatic_activity_dection["prefixPaddingMs"] = value["prefix_padding_ms"] if "silence_duration_ms" in value and isinstance(value["silence_duration_ms"], int): automatic_activity_dection["silenceDurationMs"] = value["silence_duration_ms"] return automatic_activity_dection def get_supported_openai_params(self, model: str) -> list[str]: return [ "instructions", "temperature", "max_response_output_tokens", "modalities", "tools", "input_audio_transcription", "turn_detection", "voice", ] def map_openai_params(self, optional_params: dict, non_default_params: dict) -> dict: if "generationConfig" not in optional_params: optional_params["generationConfig"] = {} for key, value in non_default_params.items(): if key == "instructions": optional_params["systemInstruction"] = HttpxContentType(role="user", parts=[{"text": value}]) elif key == "temperature": optional_params["generationConfig"]["temperature"] = value elif key == "max_response_output_tokens": optional_params["generationConfig"]["maxOutputTokens"] = value elif key == "modalities": optional_params["generationConfig"]["responseModalities"] = [ modality.upper() for modality in cast(list[str], value) ] elif key == "tools": from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import ( VertexGeminiConfig, ) vertex_gemini_config = VertexGeminiConfig() # Tools should be at the top level of setup, not inside generationConfig optional_params["tools"] = vertex_gemini_config._map_function( value=value, optional_params=optional_params ) elif key == "input_audio_transcription" and value is not None: 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 transformed_audio_activity_config: optional_params["realtimeInputConfig"] = BidiGenerateContentRealtimeInputConfig( automaticActivityDetection=transformed_audio_activity_config ) elif key == "voice": speech_config = _gemini_live_speech_config(value) if speech_config: optional_params["generationConfig"]["speechConfig"] = speech_config if len(optional_params["generationConfig"]) == 0: optional_params.pop("generationConfig") return optional_params @staticmethod def _extract_turn_detection(session: dict) -> dict | None: """Extract turn_detection from a session.update payload. Handles both the flat beta shape (``session.turn_detection``) and the GA shape (``session.audio.input.turn_detection``). """ if not isinstance(session, dict): return None td = session.get("turn_detection") if isinstance(td, dict): return td audio: Final = session.get("audio") if isinstance(audio, dict): input_cfg: Final = audio.get("input") if isinstance(input_cfg, dict): td = input_cfg.get("turn_detection") if isinstance(td, dict): return td return None @staticmethod def _normalize_session_payload_for_mapping(session: dict) -> dict: """Normalize GA-remapped session fields back to their beta keys. ``map_openai_params`` only recognises the flat OpenAI-beta key names (``modalities``, ``input_audio_transcription``, ``turn_detection``). For GA clients the upstream shim renames these into the nested GA schema (``output_modalities``, ``audio.input.transcription``, ``audio.input.turn_detection``), which would otherwise be silently dropped here. Surface them back at the top level so the existing mapping logic picks them up without duplicating provider-specific knowledge of the GA schema in ``map_openai_params``. """ if not isinstance(session, dict): return session normalized: Final = dict(session) if "modalities" not in normalized and "output_modalities" in normalized: normalized["modalities"] = normalized["output_modalities"] audio: Final = normalized.get("audio") if isinstance(audio, dict): input_cfg: Final = audio.get("input") if isinstance(input_cfg, dict): if "input_audio_transcription" not in normalized and "transcription" in input_cfg: normalized["input_audio_transcription"] = input_cfg["transcription"] output_cfg: Final = audio.get("output") if isinstance(output_cfg, dict) and output_cfg.get("voice"): normalized["voice"] = output_cfg["voice"] extracted_turn_detection: Final = GeminiRealtimeConfig._extract_turn_detection(normalized) if extracted_turn_detection is not None and not isinstance(normalized.get("turn_detection"), dict): normalized["turn_detection"] = extracted_turn_detection return normalized @staticmethod def _model_cost_entry(model: str) -> dict: entry = litellm.model_cost.get(model) if entry is None: stripped: Final = model.split("/", 1)[-1] entry = litellm.model_cost.get(stripped) or litellm.model_cost.get(f"gemini/{stripped}") return entry or {} @staticmethod def _is_audio_only_live_model(model: str) -> bool: entry: Final = GeminiRealtimeConfig._model_cost_entry(model) return bool(entry.get("gemini_native_audio") or entry.get("gemini_audio_only_live")) @staticmethod def _is_text_only_live_model(model: str) -> bool: return GeminiRealtimeConfig._model_cost_entry(model).get("mode") == "audio_transcription" @staticmethod def _default_response_modality(model: str) -> GeminiResponseModalities: return "TEXT" if GeminiRealtimeConfig._is_text_only_live_model(model) else "AUDIO" @staticmethod def _coerce_response_modalities(model: str, modalities: Sequence[Any]) -> tuple[str, ...]: """Swap responseModalities a Live model cannot produce: TEXT to AUDIO for audio-only models, AUDIO to TEXT for text-only ones (e.g. transcribe-live).""" normalized: Final = tuple( modality.upper() if isinstance(modality, str) else str(modality).upper() for modality in modalities ) if GeminiRealtimeConfig._is_audio_only_live_model(model) and "TEXT" in normalized: return tuple(modality for modality in normalized if modality != "TEXT") or ("AUDIO",) if GeminiRealtimeConfig._is_text_only_live_model(model) and "AUDIO" in normalized: return tuple(modality for modality in normalized if modality != "AUDIO") or ("TEXT",) return normalized @staticmethod def _finalize_gemini_live_setup(model: str, setup: dict[str, Any]) -> dict[str, Any]: generation_config: Final = setup.get("generationConfig") if isinstance(generation_config, dict): modalities: Final = generation_config.get("responseModalities") if isinstance(modalities, list): generation_config["responseModalities"] = GeminiRealtimeConfig._coerce_response_modalities( model, modalities ) return setup def _handle_session_update( self, json_message: dict, model: str, session_configuration_request: str | None, ) -> list[str]: """ Handle session.update by sending setup to Gemini. On the FIRST session.update (when session_configuration_request is None), the full setup with all configuration is sent. Gemini Live accepts setup as the first-and-only client message, so every later session.update is dropped rather than forwarded as a second setup (which Gemini rejects with a 1007, tearing the session down). To carry tools/instructions, send them on the first session.update before any conversation content. """ session_payload = json_message.get("session") or {} # Normalize GA-remapped fields (``output_modalities``, # nested ``audio.input.transcription``, # ``audio.input.turn_detection``) back to their flat beta keys so # ``map_openai_params`` picks them up. Without this, GA clients' # explicit modality / transcription / turn-detection settings # would be silently dropped because ``map_openai_params`` only # recognises the flat OpenAI-beta key names. session_payload = self._normalize_session_payload_for_mapping(session_payload) new_overrides: Final = self.map_openai_params(optional_params={}, non_default_params=session_payload) if session_configuration_request is None: generation_config: Final = new_overrides.setdefault("generationConfig", {}) generation_config.setdefault("responseModalities", [GeminiRealtimeConfig._default_response_modality(model)]) new_overrides.setdefault("inputAudioTranscription", {}) new_overrides["model"] = f"models/{model}" verbose_logger.debug("Gemini Realtime: Sending initial setup with tools to backend") return [json.dumps({"setup": self._finalize_gemini_live_setup(model, new_overrides)})] # Gemini Live accepts exactly one ``setup`` message: the first and only # client message. A second ``setup`` closes the socket with # ``1007 Request contains an invalid argument``, so a session.update # after the initial setup must not be forwarded as a follow-up setup. # Every GA client (pipecat included) sends several session.updates while # configuring the session; forwarding a second one tears the session down # before the first turn, which surfaces to callers as silence after the # first response, reconnect/retry latency churn, and 1011 errors. Drop # it. The Vertex subclass already drops subsequent setups for this exact # reason; the constraint is identical on AI Studio. client_turn_detection: Final = self._extract_turn_detection(session_payload) if isinstance(client_turn_detection, dict) and client_turn_detection.get("create_response") is False: verbose_logger.warning( "Gemini Realtime: Dropping subsequent session.update " "(turn_detection.create_response=False) — Gemini Live rejects a " "second setup message, so audio-transcription guardrails cannot " "suppress the model's auto-response mid-session." ) else: verbose_logger.debug("Gemini Realtime: Ignoring session.update (setup already sent)") return [] def _handle_conversation_item(self, json_message: dict) -> list[str]: """ Handle conversation.item.create for user text or function call output. Converts OpenAI format to Gemini's clientContent (for user text) or toolResponse (for function outputs). """ item: Final = json_message.get("item", {}) item_type: Final = item.get("type") if item_type == "function_call_output": return self._handle_function_call_output(item) return self._handle_user_text_content(item) def _handle_function_call_output(self, item: dict) -> list[str]: """Transform function_call_output to Gemini toolResponse format.""" call_id: Final = item.get("call_id", "") output: Final = item.get("output", "{}") verbose_logger.debug("Gemini Realtime: Transforming function_call_output for call_id=%s", call_id) # Gemini functionResponses[].response must be a dict; wrap non-dicts. try: parsed_output = json.loads(output) if isinstance(output, str) else output except json.JSONDecodeError: parsed_output = output output_dict: Final = parsed_output if isinstance(parsed_output, dict) else {"result": parsed_output} # Keep the entry (don't delete) so retried tool responses still find the name. function_name: Final = self._tool_call_id_to_name.get(call_id) if function_name: self._tool_call_id_to_name.move_to_end(call_id) else: verbose_logger.warning( "Gemini Realtime: Function name not found for call_id=%s. This may cause Gemini to reject the response.", call_id, ) function_response: Final[dict[str, Any]] = {"response": output_dict} if self._include_function_response_id() and call_id: function_response["id"] = call_id if function_name: function_response["name"] = function_name tool_response_message: Final = {"toolResponse": {"functionResponses": [function_response]}} return [json.dumps(tool_response_message)] def _handle_user_text_content(self, item: dict) -> list[str]: """Transform user text content to Gemini clientContent format.""" content_list: Final = item.get("content", []) text_parts = [c.get("text", "") for c in content_list if isinstance(c, dict) and c.get("type") == "input_text"] text: Final = " ".join(filter(None, text_parts)) if not text: return [] client_content_message: Final = { "clientContent": { "turns": [{"role": "user", "parts": [{"text": text}]}], "turnComplete": True, } } return [json.dumps(client_content_message)] def transform_realtime_request( self, message: str, model: str, session_configuration_request: str | None = None, ) -> list[str]: realtime_input_dict: BidiGenerateContentRealtimeInput = {} try: json_message: Final = json.loads(message) except json.JSONDecodeError: if isinstance(message, bytes): message_str = message.decode("utf-8", errors="replace") else: message_str = str(message) raise ValueError(f"Invalid JSON message: {message_str}") messages: Final[list[str]] = [] msg_type: Final = json_message.get("type") if msg_type == "session.update": return self._handle_session_update(json_message, model, session_configuration_request) if msg_type == "response.create": return [] # Gemini responds automatically; nothing to forward if msg_type == "conversation.item.create": return self._handle_conversation_item(json_message) if msg_type == "input_audio_buffer.append": audio_b64: Final = json_message["audio"] if isinstance(audio_b64, str): self._unbilled_input_audio_bytes += _base64_decoded_byte_count(audio_b64) realtime_input_dict["audio"] = HttpxBlobType(mimeType=self.get_audio_mime_type(), data=audio_b64) realtime_input_dict = cast( BidiGenerateContentRealtimeInput, encode_unserializable_types(cast(dict[str, object], realtime_input_dict)), ) gemini_msg: Final = json.dumps({"realtimeInput": realtime_input_dict}) verbose_logger.debug("Gemini Realtime: Sending audio realtimeInput to backend") messages.append(gemini_msg) return messages if msg_type in ("input_audio_buffer.commit", "input_audio_buffer.end"): return self._handle_input_audio_buffer_commit_or_end(session_configuration_request) if msg_type == "input_audio_buffer.clear": return [] # local buffer op, nothing to forward return [] # unknown/unsupported event type def transform_session_created_event( self, model: str, logging_session_id: str, session_configuration_request: str | None = None, ) -> OpenAIRealtimeStreamSessionEvents: if session_configuration_request: session_configuration_request_dict: BidiGenerateContentSetup = json.loads( session_configuration_request ).get("setup", {}) else: session_configuration_request_dict = {} _model: Final = session_configuration_request_dict.get("model") or model generation_config: Final = session_configuration_request_dict.get("generationConfig", {}) or {} gemini_modalities: Final = generation_config.get("responseModalities", ["AUDIO"]) _modalities: Final = [modality.lower() for modality in cast(list[str], gemini_modalities)] _system_instruction: Final = session_configuration_request_dict.get("systemInstruction") session: Final = OpenAIRealtimeStreamSession( id=logging_session_id, modalities=_modalities, ) if _system_instruction is not None and isinstance(_system_instruction, str): session["instructions"] = _system_instruction if _model is not None and isinstance(_model, str): # Strip Vertex/AI Studio path prefixes to expose the bare model name. if "/models/" in _model: session["model"] = _model.split("/models/")[-1] elif _model.startswith("models/"): session["model"] = _model[len("models/") :] else: session["model"] = _model return OpenAIRealtimeStreamSessionEvents( type="session.created", session=session, event_id=str(uuid.uuid4()), ) def _is_new_content_delta( self, previous_messages: list[OpenAIRealtimeEvents] | None = None, ) -> bool: if previous_messages is None or len(previous_messages) == 0: return True if "type" in previous_messages[-1] and previous_messages[-1]["type"].endswith("delta"): return False return True def return_new_content_delta_events( self, response_id: str, output_item_id: str, conversation_id: str, delta_type: ALL_DELTA_TYPES, session_configuration_request: str | None = None, ) -> list[OpenAIRealtimeEvents]: session_configuration_request_dict: BidiGenerateContentSetup = {} if session_configuration_request is not None: try: session_configuration_request_dict = json.loads(session_configuration_request).get("setup", {}) except json.JSONDecodeError: session_configuration_request_dict = {} generation_config: Final = session_configuration_request_dict.get("generationConfig", {}) gemini_modalities: Final = generation_config.get("responseModalities", ["AUDIO"]) _modalities: Final = [modality.lower() for modality in cast(list[str], gemini_modalities)] _temperature: Final = generation_config.get("temperature") _max_output_tokens: Final = generation_config.get("maxOutputTokens") response_items: Final[list[OpenAIRealtimeEvents]] = [] response_created: Final = OpenAIRealtimeStreamResponseBaseObject( type="response.created", event_id=f"event_{uuid.uuid4()}", response={ "object": "realtime.response", "id": response_id, "status": "in_progress", "status_details": None, "output": [], "conversation_id": conversation_id, "modalities": _modalities, "temperature": _temperature, "max_output_tokens": _max_output_tokens, }, ) response_items.append(response_created) ## - return response.output_item.added response_output_item_added: Final = OpenAIRealtimeStreamResponseOutputItemAdded( type="response.output_item.added", event_id=f"event_{uuid.uuid4()}", response_id=response_id, output_index=0, item={ "id": output_item_id, "object": "realtime.item", "type": "message", "status": "in_progress", "role": "assistant", "content": [], }, ) response_items.append(response_output_item_added) ## - 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": f"event_{uuid.uuid4()}", "previous_item_id": None, "item": { "id": output_item_id, "object": "realtime.item", "type": "message", "status": "in_progress", "role": "assistant", "content": [], }, }, ) ) ## - return response.content_part.added response_content_part_added: Final = OpenAIRealtimeResponseContentPartAdded( type="response.content_part.added", content_index=0, output_index=0, event_id=f"event_{uuid.uuid4()}", item_id=output_item_id, part=( { "type": "text", "text": "", } if delta_type == "text" else { "type": "audio", "transcript": "", } ), response_id=response_id, ) response_items.append(response_content_part_added) return response_items def transform_content_delta_events( self, message: BidiGenerateContentServerContent, output_item_id: str, response_id: str, delta_type: ALL_DELTA_TYPES, ) -> OpenAIRealtimeResponseDelta: delta = "" try: if "modelTurn" in message and "parts" in message["modelTurn"]: for part in message["modelTurn"]["parts"]: if "text" in part: delta += part["text"] elif "inlineData" in part: delta += part["inlineData"].get("data", "") except Exception as e: raise ValueError(f"Error transforming content delta events: {e}, got message: {message}") return OpenAIRealtimeResponseDelta( type=("response.output_text.delta" if delta_type == "text" else "response.output_audio.delta"), content_index=0, event_id=f"event_{uuid.uuid4()}", item_id=output_item_id, output_index=0, response_id=response_id, delta=delta, ) def transform_content_done_event( self, delta_chunks: list[OpenAIRealtimeResponseDelta] | None, current_output_item_id: str | None, current_response_id: str | None, delta_type: ALL_DELTA_TYPES, ) -> OpenAIRealtimeResponseTextDone | OpenAIRealtimeResponseAudioDone: if delta_chunks: delta = "".join([delta_chunk["delta"] for delta_chunk in delta_chunks]) else: delta = "" if current_output_item_id is None: current_output_item_id = f"item_{uuid.uuid4()}" if current_response_id is None: current_response_id = f"resp_{uuid.uuid4()}" if delta_type == "text": return OpenAIRealtimeResponseTextDone( type="response.output_text.done", content_index=0, event_id=f"event_{uuid.uuid4()}", item_id=current_output_item_id, output_index=0, response_id=current_response_id, text=delta, ) elif delta_type == "audio": return OpenAIRealtimeResponseAudioDone( type="response.output_audio.done", content_index=0, event_id=f"event_{uuid.uuid4()}", item_id=current_output_item_id, output_index=0, response_id=current_response_id, ) def return_additional_content_done_events( self, current_output_item_id: str | None, current_response_id: str | None, delta_done_event: OpenAIRealtimeResponseTextDone | OpenAIRealtimeResponseAudioDone, delta_type: ALL_DELTA_TYPES, ) -> list[OpenAIRealtimeEvents]: """ - return response.content_part.done - return response.output_item.done """ if current_output_item_id is None: current_output_item_id = f"item_{uuid.uuid4()}" if current_response_id is None: current_response_id = f"resp_{uuid.uuid4()}" returned_items: Final[list[OpenAIRealtimeEvents]] = [] delta_done_event_text: Final = cast(str | None, delta_done_event.get("text")) # response.content_part.done response_content_part_done: Final = OpenAIRealtimeContentPartDone( type="response.content_part.done", content_index=0, event_id=f"event_{uuid.uuid4()}", item_id=current_output_item_id, output_index=0, part=( {"type": "text", "text": delta_done_event_text} if delta_done_event_text and delta_type == "text" else { "type": "audio", "transcript": "", # gemini doesn't return transcript for audio } ), response_id=current_response_id, ) returned_items.append(response_content_part_done) # response.output_item.done response_output_item_done: Final = OpenAIRealtimeOutputItemDone( type="response.output_item.done", event_id=f"event_{uuid.uuid4()}", output_index=0, response_id=current_response_id, item={ "id": current_output_item_id, "object": "realtime.item", "type": "message", "status": "completed", "role": "assistant", "content": [ ( {"type": "text", "text": delta_done_event_text} if delta_done_event_text and delta_type == "text" else { "type": "audio", "transcript": "", } ) ], }, ) returned_items.append(response_output_item_done) return returned_items def _consume_usage_metadata_for_response_done(self, frame: dict) -> dict | None: """Pop usageMetadata from the frame (authoritative) or drain the pending buffer. Uses pop so a frame with both ``toolCall`` and ``turnComplete`` can't attribute the same counts to two response.done events. """ in_frame: Final = frame.pop("usageMetadata", None) if isinstance(frame, dict) else None if isinstance(in_frame, dict): self._pending_usage_metadata = None return in_frame buffered: Final = self._pending_usage_metadata self._pending_usage_metadata = None return buffered def transform_tool_call_events( self, tool_call_message: dict, response_id: str | None = None, output_item_id: str | None = None, ) -> list[OpenAIRealtimeFunctionCallArgumentsDone]: function_calls: Final = tool_call_message.get("functionCalls", []) resolved_response_id: Final = response_id or f"resp_{uuid.uuid4()}" resolved_output_item_id: Final = output_item_id or f"item_{uuid.uuid4()}" verbose_logger.debug("Gemini Realtime: Transforming %s tool call(s) to OpenAI format", len(function_calls)) events: Final[list[OpenAIRealtimeFunctionCallArgumentsDone]] = [] for idx, fc in enumerate(function_calls): call_id = fc.get("id", "") or f"call_{uuid.uuid4().hex[:16]}" name = fc.get("name", "") if call_id and name: self._tool_call_id_to_name[call_id] = name self._tool_call_id_to_name.move_to_end(call_id) while len(self._tool_call_id_to_name) > self._TOOL_CALL_ID_TO_NAME_MAX: self._tool_call_id_to_name.popitem(last=False) events.append( OpenAIRealtimeFunctionCallArgumentsDone( type="response.function_call_arguments.done", event_id=f"event_{uuid.uuid4()}", response_id=resolved_response_id, item_id=f"{resolved_output_item_id}_tool_{idx}", output_index=idx, call_id=call_id, name=name, arguments=json.dumps(fc.get("args", {})), ) ) return events @staticmethod def get_nested_value(obj: dict, path: str) -> Any: keys: Final = path.split(".") current = obj for key in keys: if isinstance(current, dict) and key in current: current = current[key] else: return None return current def update_current_delta_chunks( self, transformed_message: OpenAIRealtimeEvents | list[OpenAIRealtimeEvents], current_delta_chunks: list[OpenAIRealtimeResponseDelta] | None, ) -> list[OpenAIRealtimeResponseDelta] | None: try: if isinstance(transformed_message, list): current_delta_chunks = [] any_delta_chunk = False for event in transformed_message: if event["type"] == "response.output_text.delta": current_delta_chunks.append(cast(OpenAIRealtimeResponseDelta, event)) any_delta_chunk = True if not any_delta_chunk: current_delta_chunks = None else: if ( transformed_message["type"] == "response.output_text.delta" ): # audio deltas are not accumulated (memory) if current_delta_chunks is None: current_delta_chunks = [] current_delta_chunks.append(cast(OpenAIRealtimeResponseDelta, transformed_message)) else: current_delta_chunks = None return current_delta_chunks except Exception as e: raise ValueError( f"Error updating current delta chunks: {e}, got transformed_message: {transformed_message}" ) def update_current_item_chunks( self, transformed_message: OpenAIRealtimeEvents | list[OpenAIRealtimeEvents], current_item_chunks: list[OpenAIRealtimeOutputItemDone] | None, ) -> list[OpenAIRealtimeOutputItemDone] | None: try: if isinstance(transformed_message, list): current_item_chunks = [] any_item_chunk = False for event in transformed_message: if event["type"] == "response.output_item.done": current_item_chunks.append(cast(OpenAIRealtimeOutputItemDone, event)) any_item_chunk = True if not any_item_chunk: current_item_chunks = None else: if transformed_message["type"] == "response.output_item.done": if current_item_chunks is None: current_item_chunks = [] current_item_chunks.append(cast(OpenAIRealtimeOutputItemDone, transformed_message)) else: current_item_chunks = None return current_item_chunks except Exception as e: raise ValueError(f"Error updating current item chunks: {e}, got transformed_message: {transformed_message}") def transform_response_done_event( self, message: BidiGenerateContentServerMessage, current_response_id: str | None, current_conversation_id: str | None, output_items: list[OpenAIRealtimeOutputItemDone] | None, session_configuration_request: str | None = None, ) -> OpenAIRealtimeDoneEvent: if current_conversation_id is None: current_conversation_id = f"conv_{uuid.uuid4()}" if current_response_id is None: current_response_id = f"resp_{uuid.uuid4()}" if session_configuration_request: session_configuration_request_dict: BidiGenerateContentSetup = json.loads( session_configuration_request ).get("setup", {}) else: session_configuration_request_dict = {} generation_config: Final = session_configuration_request_dict.get("generationConfig", {}) temperature: Final = generation_config.get("temperature") max_output_tokens: Final = generation_config.get("maxOutputTokens") gemini_modalities: Final = generation_config.get("responseModalities", ["AUDIO"]) _modalities: Final = [modality.lower() for modality in cast(list[str], gemini_modalities)] resolved_usage_metadata: Final = self._consume_usage_metadata_for_response_done(cast(dict, message)) if resolved_usage_metadata is not None: _chat_completion_usage = VertexGeminiConfig._calculate_usage( completion_response=cast( BidiGenerateContentServerMessage, {**cast(dict, message), "usageMetadata": resolved_usage_metadata}, ), ) else: _chat_completion_usage = get_empty_usage() responses_api_usage = LiteLLMCompletionResponsesConfig._transform_chat_completion_usage_to_responses_usage( _chat_completion_usage, ) _usage_dict: Final = responses_api_usage.model_dump() self._add_pipecat_usage_detail_aliases(_usage_dict) response_done_event: Final = OpenAIRealtimeDoneEvent( type="response.done", event_id=f"event_{uuid.uuid4()}", response=OpenAIRealtimeResponseDoneObject( object="realtime.response", id=current_response_id, status="completed", status_details=None, output=([output_item["item"] for output_item in output_items] if output_items else []), conversation_id=current_conversation_id, modalities=_modalities, usage=_usage_dict, ), ) if temperature is not None: response_done_event["response"]["temperature"] = temperature if max_output_tokens is not None: response_done_event["response"]["max_output_tokens"] = cast(int, max_output_tokens) return response_done_event def handle_openai_modality_event( self, openai_event: OpenAIRealtimeEventTypes, json_message: dict, realtime_response_transform_input: RealtimeResponseTransformInput, delta_type: ALL_DELTA_TYPES, ) -> RealtimeModalityResponseTransformOutput: current_output_item_id = realtime_response_transform_input["current_output_item_id"] current_response_id = realtime_response_transform_input["current_response_id"] current_conversation_id = realtime_response_transform_input["current_conversation_id"] current_delta_chunks: Final = realtime_response_transform_input["current_delta_chunks"] session_configuration_request: Final = realtime_response_transform_input["session_configuration_request"] returned_message: list[OpenAIRealtimeEvents] = [] if ( openai_event == OpenAIRealtimeEventTypes.RESPONSE_TEXT_DELTA or openai_event == OpenAIRealtimeEventTypes.RESPONSE_AUDIO_DELTA ): current_response_id = current_response_id or f"resp_{uuid.uuid4()}" if not current_output_item_id: # send the list of standard 'new' content.delta events current_output_item_id = f"item_{uuid.uuid4()}" current_conversation_id = current_conversation_id or f"conv_{uuid.uuid4()}" returned_message = 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=delta_type, ) # send the list of standard 'new' content.delta events transformed_message: Final = self.transform_content_delta_events( BidiGenerateContentServerContent(**json_message["serverContent"]), current_output_item_id, current_response_id, delta_type=delta_type, ) returned_message.append(transformed_message) elif ( openai_event == OpenAIRealtimeEventTypes.RESPONSE_TEXT_DONE or openai_event == OpenAIRealtimeEventTypes.RESPONSE_AUDIO_DONE ): transformed_content_done_event: Final = self.transform_content_done_event( current_output_item_id=current_output_item_id, current_response_id=current_response_id, delta_chunks=current_delta_chunks, delta_type=delta_type, ) returned_message = [transformed_content_done_event] # Use IDs from the done event — transform_content_done_event may have # generated UUID fallbacks when the originals were None. resolved_item_id: Final = transformed_content_done_event.get("item_id") or current_output_item_id resolved_response_id: Final = transformed_content_done_event.get("response_id") or current_response_id additional_items: Final = self.return_additional_content_done_events( current_output_item_id=resolved_item_id, current_response_id=resolved_response_id, delta_done_event=transformed_content_done_event, delta_type=delta_type, ) returned_message.extend(additional_items) return { "returned_message": returned_message, "current_output_item_id": current_output_item_id, "current_response_id": current_response_id, "current_conversation_id": current_conversation_id, "current_delta_chunks": current_delta_chunks, "current_delta_type": delta_type, } def map_openai_event( self, key: str, value: Any, current_delta_type: ALL_DELTA_TYPES | None, ) -> OpenAIRealtimeEventTypes | ResponsesAPIStreamEvents: if isinstance(value, dict): model_turn_event = value.get("modelTurn") generation_complete_event = value.get("generationComplete") else: model_turn_event = None generation_complete_event = None openai_event: OpenAIRealtimeEventTypes | ResponsesAPIStreamEvents | None = None if model_turn_event: # check if model turn event openai_event = self.map_model_turn_event(model_turn_event) elif generation_complete_event: openai_event = self.map_generation_complete_event(delta_type=current_delta_type) else: # 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. Scope dotted-key # lookups to the current ``key``/``value`` pair — checking the # whole ``json_message`` would let a sibling key (e.g. # ``serverContent.turnComplete``) misclassify the event currently # being processed (e.g. ``toolCall``). for map_key, candidate_event in MAP_GEMINI_FIELD_TO_OPENAI_EVENT.items(): if map_key == key: openai_event = candidate_event break if "." in map_key: prefix, _, nested_path = map_key.partition(".") if ( prefix == key and isinstance(value, dict) and GeminiRealtimeConfig.get_nested_value(value, nested_path) is not None ): openai_event = candidate_event break if openai_event is None: raise ValueError(f"Unknown openai event: {key}, value: {value}") return openai_event def _consume_input_transcription_usage_estimate(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None: """Gemini Live sends no usageMetadata for transcribe sessions; estimate billing from streamed audio duration.""" if self._unbilled_input_audio_bytes <= 0 or not self._is_text_only_live_model(model): return None audio_seconds: Final = self._unbilled_input_audio_bytes / PCM16_INPUT_AUDIO_BYTES_PER_SECOND self._unbilled_input_audio_bytes = 0 audio_tokens: Final = round(audio_seconds * GEMINI_LIVE_TRANSCRIBE_AUDIO_TOKENS_PER_SECOND) output_tokens: Final = round(audio_seconds * GEMINI_LIVE_TRANSCRIBE_OUTPUT_TEXT_TOKENS_PER_MINUTE / 60) usage: Final[RealtimeInputAudioTranscriptionUsage] = { "type": "tokens", "input_tokens": audio_tokens, "output_tokens": output_tokens, "total_tokens": audio_tokens + output_tokens, "input_token_details": {"text_tokens": 0, "audio_tokens": audio_tokens}, } return usage def unbilled_usage_on_session_close(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None: return self._consume_input_transcription_usage_estimate(model) def transform_realtime_response( self, message: str | bytes, model: str, logging_obj: LiteLLMLoggingObj, realtime_response_transform_input: RealtimeResponseTransformInput, ) -> RealtimeResponseTypedDict: """ Keep this state less - leave the state management (e.g. tracking current_output_item_id, current_response_id, current_conversation_id, current_delta_chunks) to the caller. """ try: json_message: Final = json.loads(message) except json.JSONDecodeError: if isinstance(message, bytes): message_str = message.decode("utf-8", errors="replace") else: message_str = str(message) raise ValueError(f"Invalid JSON message: {message_str}") verbose_logger.debug( "Realtime Response Transform: Gemini frame keys=%s", (sorted(json_message.keys()) if isinstance(json_message, dict) else type(json_message).__name__), ) logging_session_id: Final = logging_obj.litellm_trace_id current_output_item_id = realtime_response_transform_input["current_output_item_id"] current_response_id = realtime_response_transform_input["current_response_id"] current_conversation_id = realtime_response_transform_input["current_conversation_id"] current_delta_chunks = realtime_response_transform_input["current_delta_chunks"] session_configuration_request: Final = realtime_response_transform_input["session_configuration_request"] current_item_chunks = realtime_response_transform_input["current_item_chunks"] current_delta_type: ALL_DELTA_TYPES | None = realtime_response_transform_input["current_delta_type"] returned_message: Final[list[OpenAIRealtimeEvents]] = [] server_content: Final = json_message.get("serverContent") if isinstance(server_content, dict): input_tx: Final = server_content.get("inputTranscription") if isinstance(input_tx, dict) and input_tx.get("text"): transcription_usage: Final = self._consume_input_transcription_usage_estimate(model) returned_message.append( cast( OpenAIRealtimeEvents, { "type": "conversation.item.input_audio_transcription.completed", "event_id": f"event_{uuid.uuid4()}", "transcript": input_tx["text"], "item_id": f"item_{uuid.uuid4()}", "content_index": 0, **({} if transcription_usage is None else {"usage": transcription_usage}), }, ) ) output_tx: Final = server_content.get("outputTranscription") if isinstance(output_tx, dict) and output_tx.get("text"): if current_response_id is None: current_response_id = f"resp_{uuid.uuid4()}" if current_output_item_id is None: current_output_item_id = f"item_{uuid.uuid4()}" current_conversation_id = current_conversation_id or f"conv_{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", ) ) returned_message.append( cast( OpenAIRealtimeEvents, { "type": "response.output_audio_transcript.delta", "event_id": f"event_{uuid.uuid4()}", "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"], }, ) ) # Transcription-only models emit generationComplete with no prior # modelTurn delta; there is no started OpenAI response to close, so # drop it and let siblings (turnComplete, usageMetadata) process. if current_delta_type is None and "modelTurn" not in server_content: server_content.pop("generationComplete", None) # Mark transcription-only serverContent as handled so the main loop # skips it; sibling keys like toolCall are still processed below. _model_content_keys: Final = { "modelTurn", "turnComplete", "interrupted", "generationComplete", } server_content_handled = not any(k in server_content for k in _model_content_keys) else: server_content_handled = False tool_call_handled = False for key, value in list(json_message.items()): # snapshot: handlers may mutate json_message # Skip sibling metadata keys (e.g. ``usageMetadata``) that can # accompany a primary payload like ``toolCall`` or ``serverContent``. # ``map_openai_event`` raises ValueError on unknown keys, which # would otherwise terminate the WebSocket session. if key not in _KNOWN_GEMINI_TOP_LEVEL_KEYS: continue # serverContent was a transcription-only payload already emitted # above; skip it here so map_openai_event doesn't raise on the # missing model-content subkeys. if key == "serverContent" and server_content_handled: continue # Check if this key or any nested key matches our mapping openai_event = self.map_openai_event( key=key, value=value, current_delta_type=current_delta_type, ) if openai_event == OpenAIRealtimeEventTypes.SESSION_CREATED: transformed_message = self.transform_session_created_event( model, logging_session_id, realtime_response_transform_input["session_configuration_request"], ) returned_message.append(transformed_message) elif openai_event == ResponsesAPIStreamEvents.FUNCTION_CALL_ARGUMENTS_DONE: if not value.get("functionCalls"): # Empty toolCall — mark consumed so the post-loop guard doesn't raise. tool_call_handled = True continue if current_conversation_id is None: current_conversation_id = f"conv_{uuid.uuid4()}" session_setup: BidiGenerateContentSetup = {} if session_configuration_request is not None: try: session_setup = json.loads(session_configuration_request).get("setup", {}) except (json.JSONDecodeError, TypeError): session_setup = {} tool_call_generation_config = session_setup.get("generationConfig", {}) or {} tool_call_modalities = [ modality.lower() for modality in cast( list[str], tool_call_generation_config.get("responseModalities", ["AUDIO"]), ) ] if current_response_id is None: current_response_id = f"resp_{uuid.uuid4()}" current_output_item_id = f"item_{uuid.uuid4()}" returned_message.append( { "type": "response.created", "event_id": f"event_{uuid.uuid4()}", "response": { "object": "realtime.response", "id": current_response_id, "status": "in_progress", "status_details": None, "output": [], "conversation_id": current_conversation_id, "modalities": tool_call_modalities, "temperature": tool_call_generation_config.get("temperature"), "max_output_tokens": tool_call_generation_config.get("maxOutputTokens"), }, } ) tool_call_events = self.transform_tool_call_events( value, response_id=current_response_id, output_item_id=current_output_item_id, ) for idx, tool_call in enumerate(tool_call_events): item_id = tool_call["item_id"] function_call_item: OpenAIRealtimeStreamResponseOutputItem = { "id": item_id, "object": "realtime.item", "type": "function_call", "status": "completed", "call_id": tool_call["call_id"], "name": tool_call["name"], "arguments": tool_call["arguments"], } returned_message.append( OpenAIRealtimeStreamResponseOutputItemAdded( type="response.output_item.added", event_id=f"event_{uuid.uuid4()}", response_id=current_response_id, output_index=idx, item={ **function_call_item, "status": "in_progress", "arguments": "", }, ) ) # conversation.item.added is required for Pipecat 1.3.x to # register the call_id into _pending_function_calls before # response.function_call_arguments.done fires. 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": "", }, }, ) ) # Gemini delivers args in one shot; emit a single delta before .done # so clients that accumulate deltas get the full payload. returned_message.append( cast( OpenAIRealtimeEvents, { "type": "response.function_call_arguments.delta", "event_id": f"event_{uuid.uuid4()}", "response_id": current_response_id, "item_id": item_id, "output_index": idx, "call_id": tool_call["call_id"], "delta": tool_call["arguments"], }, ) ) returned_message.append(tool_call) # Fresh copy — downstream handlers may mutate the item dict. returned_message.append( OpenAIRealtimeOutputItemDone( type="response.output_item.done", event_id=f"event_{uuid.uuid4()}", response_id=current_response_id, output_index=idx, item={**function_call_item}, ) ) resolved_tool_call_usage_metadata = self._consume_usage_metadata_for_response_done(json_message) if resolved_tool_call_usage_metadata is not None: _tool_call_chat_completion_usage = VertexGeminiConfig._calculate_usage( completion_response=cast( BidiGenerateContentServerMessage, { **json_message, "usageMetadata": resolved_tool_call_usage_metadata, }, ), ) else: _tool_call_chat_completion_usage = get_empty_usage() 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()}", response=OpenAIRealtimeResponseDoneObject( id=current_response_id, object="realtime.response", status="completed", status_details=None, output=[ { "id": te["item_id"], "object": "realtime.item", "type": "function_call", "status": "completed", "call_id": te["call_id"], "name": te["name"], "arguments": te["arguments"], } for te in tool_call_events ], conversation_id=current_conversation_id, modalities=tool_call_modalities, usage=_tool_usage_dict, ), ) tool_call_temperature = tool_call_generation_config.get("temperature") if tool_call_temperature is not None: tool_call_done_event["response"]["temperature"] = tool_call_temperature tool_call_max_output_tokens = tool_call_generation_config.get("maxOutputTokens") if tool_call_max_output_tokens is not None: tool_call_done_event["response"]["max_output_tokens"] = cast(int, tool_call_max_output_tokens) returned_message.append(tool_call_done_event) current_output_item_id = None current_response_id = None elif openai_event == OpenAIRealtimeEventTypes.RESPONSE_DONE: _has_pending_function_call = current_item_chunks and any( chunk.get("item", {}).get("type") == "function_call" for chunk in current_item_chunks ) if current_response_id is None and _has_pending_function_call: # Trailing bare turnComplete after a toolCall (Vertex emits ~5 # bookkeeping tokens before the follow-up answer). Suppress the # empty response.done so collect_until("response.done") clients # don't stop prematurely; buffer usage for the next real turn. standalone_usage_metadata = json_message.get("usageMetadata") if isinstance(standalone_usage_metadata, dict): self._pending_usage_metadata = standalone_usage_metadata server_content_handled = True continue transformed_response_done_event = self.transform_response_done_event( message=BidiGenerateContentServerMessage(**json_message), current_response_id=current_response_id, current_conversation_id=current_conversation_id, session_configuration_request=session_configuration_request, output_items=None, ) returned_message.append(transformed_response_done_event) current_output_item_id = None current_response_id = None elif ( openai_event == OpenAIRealtimeEventTypes.RESPONSE_TEXT_DELTA or openai_event == OpenAIRealtimeEventTypes.RESPONSE_TEXT_DONE or openai_event == OpenAIRealtimeEventTypes.RESPONSE_AUDIO_DELTA or openai_event == OpenAIRealtimeEventTypes.RESPONSE_AUDIO_DONE ): # Use locally-updated state so prior loop iterations' ID resets are visible. _modality_input: RealtimeResponseTransformInput = { **realtime_response_transform_input, "current_output_item_id": current_output_item_id, "current_response_id": current_response_id, "current_conversation_id": current_conversation_id, "current_delta_chunks": current_delta_chunks, "current_item_chunks": current_item_chunks, "current_delta_type": current_delta_type, "session_configuration_request": session_configuration_request, } _returned_message = self.handle_openai_modality_event( openai_event, json_message, _modality_input, delta_type="text" if "text" in openai_event.value else "audio", ) returned_message.extend(_returned_message["returned_message"]) current_output_item_id = _returned_message["current_output_item_id"] current_response_id = _returned_message["current_response_id"] current_conversation_id = _returned_message["current_conversation_id"] current_delta_chunks = _returned_message["current_delta_chunks"] current_delta_type = _returned_message["current_delta_type"] else: raise ValueError(f"Unknown openai event: {openai_event}") if len(returned_message) == 0: unhandled_known_keys: Final = [ key for key in json_message if key in _KNOWN_GEMINI_TOP_LEVEL_KEYS and not (key == "serverContent" and server_content_handled) and not (key == "toolCall" and tool_call_handled) ] standalone_usage_metadata = json_message.get("usageMetadata") if isinstance(standalone_usage_metadata, dict): self._pending_usage_metadata = standalone_usage_metadata if not unhandled_known_keys: return { "response": returned_message, "current_output_item_id": current_output_item_id, "current_response_id": current_response_id, "current_delta_chunks": current_delta_chunks, "current_conversation_id": current_conversation_id, "current_item_chunks": current_item_chunks, "current_delta_type": current_delta_type, "session_configuration_request": session_configuration_request, } if isinstance(message, bytes): message_str = message.decode("utf-8", errors="replace") else: message_str = str(message) raise ValueError(f"Unknown message type: {message_str}") current_delta_chunks = self.update_current_delta_chunks( transformed_message=returned_message, current_delta_chunks=current_delta_chunks, ) current_item_chunks = self.update_current_item_chunks( transformed_message=returned_message, current_item_chunks=current_item_chunks, ) for msg in returned_message: event_type = msg.get("type") if isinstance(msg, dict) else "unknown" verbose_logger.debug("Realtime Response Transform: OpenAI event=%s", event_type) return { "response": returned_message, "current_output_item_id": current_output_item_id, "current_response_id": current_response_id, "current_delta_chunks": current_delta_chunks, "current_conversation_id": current_conversation_id, "current_item_chunks": current_item_chunks, "current_delta_type": current_delta_type, "session_configuration_request": session_configuration_request, } def requires_session_configuration(self) -> bool: # Deferred setup opt-in: litellm.gemini_live_defer_setup = True return not litellm.gemini_live_defer_setup def session_configuration_request(self, model: str) -> str: """ ``` { "model": string, "generationConfig": { "candidateCount": integer, "maxOutputTokens": integer, "temperature": number, "topP": number, "topK": integer, "presencePenalty": number, "frequencyPenalty": number, "responseModalities": [string], "speechConfig": object, "mediaResolution": object }, "systemInstruction": string, "tools": [object] } ``` """ response_modalities: Final[list[GeminiResponseModalities]] = [ GeminiRealtimeConfig._default_response_modality(model) ] output_audio_transcription: Final = False # if "audio" in model: ## UNCOMMENT THIS WHEN AUDIO IS SUPPORTED # output_audio_transcription = True setup_config: Final[BidiGenerateContentSetup] = { "model": f"models/{model}", "generationConfig": {"responseModalities": response_modalities}, # Return input transcript so guardrails can inspect user speech. "inputAudioTranscription": {}, } if output_audio_transcription: setup_config["outputAudioTranscription"] = {} return json.dumps( { "setup": setup_config, } )