From f590a476e77311d5d55b76b1c81ad2b918896da3 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 2 Jul 2025 07:41:15 -0700 Subject: [PATCH] Gemini Live fixes, plus additional docstrings --- .../gemini_multimodal_live/file_api.py | 7 ++ .../services/gemini_multimodal_live/gemini.py | 68 +++++++++++++------ 2 files changed, 55 insertions(+), 20 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/file_api.py b/src/pipecat/services/gemini_multimodal_live/file_api.py index f0f23ab83..5ae7fdbb7 100644 --- a/src/pipecat/services/gemini_multimodal_live/file_api.py +++ b/src/pipecat/services/gemini_multimodal_live/file_api.py @@ -4,6 +4,13 @@ # SPDX-License-Identifier: BSD 2-Clause License # +"""Gemini File API client for uploading and managing files. + +This module provides a client for Google's Gemini File API, enabling file +uploads, metadata retrieval, listing, and deletion. Files uploaded through +this API can be referenced in Gemini generative model calls. +""" + import mimetypes from typing import Any, Dict, Optional diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 19860387e..a31bbfe70 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -72,7 +72,6 @@ from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt from . import events - from .file_api import GeminiFileAPI try: @@ -223,9 +222,9 @@ class GeminiMultimodalLiveContext(OpenAILLMContext): def add_file_reference(self, file_uri: str, mime_type: str, text: Optional[str] = None): """Add a file reference to the context. - + This adds a user message with a file reference that will be sent during context initialization. - + Args: file_uri: URI of the uploaded file mime_type: MIME type of the file @@ -235,15 +234,17 @@ class GeminiMultimodalLiveContext(OpenAILLMContext): parts = [] if text: parts.append({"type": "text", "text": text}) - + # Add file reference part - parts.append({"type": "file_data", "file_data": {"mime_type": mime_type, "file_uri": file_uri}}) - + parts.append( + {"type": "file_data", "file_data": {"mime_type": mime_type, "file_uri": file_uri}} + ) + # Add to messages message = {"role": "user", "content": parts} self.messages.append(message) logger.info(f"Added file reference to context: {file_uri}") - + def get_messages_for_initializing_history(self): """Get messages formatted for Gemini history initialization. @@ -270,12 +271,14 @@ class GeminiMultimodalLiveContext(OpenAILLMContext): parts.append({"text": part.get("text")}) elif part.get("type") == "file_data": file_data = part.get("file_data", {}) - parts.append({ - "fileData": { - "mimeType": file_data.get("mime_type"), - "fileUri": file_data.get("file_uri") + parts.append( + { + "fileData": { + "mimeType": file_data.get("mime_type"), + "fileUri": file_data.get("file_uri"), + } } - }) + ) else: logger.warning(f"Unsupported content type: {str(part)[:80]}") else: @@ -379,7 +382,14 @@ class GeminiMultimodalModalities(Enum): class GeminiMediaResolution(str, Enum): - """Media resolution options for Gemini Multimodal Live.""" + """Media resolution options for Gemini Multimodal Live. + + Parameters: + UNSPECIFIED: Use default resolution setting. + LOW: Low resolution with 64 tokens. + MEDIUM: Medium resolution with 256 tokens. + HIGH: High resolution with zoomed reframing and 256 tokens. + """ UNSPECIFIED = "MEDIA_RESOLUTION_UNSPECIFIED" # Use default LOW = "MEDIA_RESOLUTION_LOW" # 64 tokens @@ -465,7 +475,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter - + def __init__( self, *, @@ -496,6 +506,7 @@ class GeminiMultimodalLiveLLMService(LLMService): params: Configuration parameters for the model. Defaults to InputParams(). inference_on_context_initialization: Whether to generate a response when context is first set. Defaults to True. + file_api_base_url: Base URL for the Gemini File API. Defaults to the official endpoint. **kwargs: Additional arguments passed to parent LLMService. """ super().__init__(base_url=base_url, **kwargs) @@ -557,7 +568,7 @@ class GeminiMultimodalLiveLLMService(LLMService): else {}, "extra": params.extra if isinstance(params.extra, dict) else {}, } - + # Initialize the File API client self.file_api = GeminiFileAPI(api_key=api_key, base_url=file_api_base_url) @@ -757,6 +768,7 @@ class GeminiMultimodalLiveLLMService(LLMService): await self._ws_send(event.model_dump(exclude_none=True)) async def _connect(self): + """Establish WebSocket connection to Gemini Live API.""" if self._websocket: # Here we assume that if we have a websocket, we are connected. We # handle disconnections in the send/recv code paths. @@ -861,6 +873,7 @@ class GeminiMultimodalLiveLLMService(LLMService): self._websocket = None async def _disconnect(self): + """Disconnect from Gemini Live API and clean up resources.""" logger.info("Disconnecting from Gemini service") try: self._disconnecting = True @@ -877,6 +890,7 @@ class GeminiMultimodalLiveLLMService(LLMService): logger.error(f"{self} error disconnecting: {e}") async def _ws_send(self, message): + """Send a message to the WebSocket connection.""" # logger.debug(f"Sending message to websocket: {message}") try: if self._websocket: @@ -897,6 +911,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # async def _receive_task_handler(self): + """Handle incoming messages from the WebSocket connection.""" async for message in WatchdogAsyncIterator(self._websocket, manager=self.task_manager): evt = events.parse_server_event(message) # logger.debug(f"Received event: {message[:500]}") @@ -925,6 +940,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # async def _send_user_audio(self, frame): + """Send user audio frame to Gemini Live API.""" if self._audio_input_paused: return # Send all audio to Gemini @@ -941,6 +957,7 @@ class GeminiMultimodalLiveLLMService(LLMService): self._user_audio_buffer = self._user_audio_buffer[-length:] async def _send_user_video(self, frame): + """Send user video frame to Gemini Live API.""" if self._video_input_paused: return @@ -954,6 +971,7 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.send_client_event(evt) async def _create_initial_response(self): + """Create initial response based on context history.""" if not self._api_session_ready: self._run_llm_when_api_session_ready = True return @@ -979,6 +997,7 @@ class GeminiMultimodalLiveLLMService(LLMService): self._needs_turn_complete_message = True async def _create_single_response(self, messages_list): + """Create a single response from a list of messages.""" # Refactor to combine this logic with same logic in GeminiMultimodalLiveContext messages = [] for item in messages_list: @@ -1000,12 +1019,14 @@ class GeminiMultimodalLiveLLMService(LLMService): parts.append({"text": part.get("text")}) elif part.get("type") == "file_data": file_data = part.get("file_data", {}) - parts.append({ - "fileData": { - "mimeType": file_data.get("mime_type"), - "fileUri": file_data.get("file_uri") + parts.append( + { + "fileData": { + "mimeType": file_data.get("mime_type"), + "fileUri": file_data.get("file_uri"), + } } - }) + ) else: logger.warning(f"Unsupported content type: {str(part)[:80]}") else: @@ -1029,6 +1050,7 @@ class GeminiMultimodalLiveLLMService(LLMService): @traced_gemini_live(operation="llm_tool_result") async def _tool_result(self, tool_result_message): + """Send tool result back to the API.""" # For now we're shoving the name into the tool_call_id field, so this # will work until we revisit that. id = tool_result_message.get("tool_call_id") @@ -1054,6 +1076,7 @@ class GeminiMultimodalLiveLLMService(LLMService): @traced_gemini_live(operation="llm_setup") async def _handle_evt_setup_complete(self, evt): + """Handle the setup complete event.""" # If this is our first context frame, run the LLM self._api_session_ready = True # Now that we've configured the session, we can run the LLM if we need to. @@ -1062,6 +1085,7 @@ class GeminiMultimodalLiveLLMService(LLMService): await self._create_initial_response() async def _handle_evt_model_turn(self, evt): + """Handle the model turn event.""" part = evt.serverContent.modelTurn.parts[0] if not part: return @@ -1103,6 +1127,7 @@ class GeminiMultimodalLiveLLMService(LLMService): @traced_gemini_live(operation="llm_tool_call") async def _handle_evt_tool_call(self, evt): + """Handle tool call events.""" function_calls = evt.toolCall.functionCalls if not function_calls: return @@ -1123,6 +1148,7 @@ class GeminiMultimodalLiveLLMService(LLMService): @traced_gemini_live(operation="llm_response") async def _handle_evt_turn_complete(self, evt): + """Handle the turn complete event.""" self._bot_is_speaking = False text = self._bot_text_buffer @@ -1206,6 +1232,7 @@ class GeminiMultimodalLiveLLMService(LLMService): ) async def _handle_evt_output_transcription(self, evt): + """Handle the output transcription event.""" if not evt.serverContent.outputTranscription: return @@ -1224,6 +1251,7 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.push_frame(TTSTextFrame(text=text)) async def _handle_evt_usage_metadata(self, evt): + """Handle the usage metadata event.""" if not evt.usageMetadata: return