From d45a07b5e51e2c48192ff4d66a176f9d4f1e121e Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 9 May 2025 10:50:27 -0400 Subject: [PATCH 01/18] add FileAPI to gemini.py --- .../services/gemini_multimodal_live/gemini.py | 103 +++++++++++++++++- 1 file changed, 102 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 3c5ed92dc..cc85799db 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -71,6 +71,8 @@ from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt from . import events +from .audio_transcriber import AudioTranscriber +from .file_api import GeminiFileAPI try: import websockets @@ -218,6 +220,29 @@ class GeminiMultimodalLiveContext(OpenAILLMContext): system_instruction += str(content) return system_instruction + 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 + text: Optional text prompt to accompany the file + """ + # Create parts list with file reference + 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}}) + + # 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. @@ -242,6 +267,14 @@ class GeminiMultimodalLiveContext(OpenAILLMContext): for part in content: if part.get("type") == "text": 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") + } + }) else: logger.warning(f"Unsupported content type: {str(part)[:80]}") else: @@ -432,6 +465,62 @@ class GeminiMultimodalLiveLLMService(LLMService): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter + """Gemini Live LLM Service with multimodal capabilities including File API support. + + This service implements the Gemini Multimodal Live API with support for: + - Audio input and output + - Image/video input + - File API (upload, reference, and management) + - Tools/function calling + + Example usage of File API: + ```python + # Initialize the service + gemini_service = GeminiMultimodalLiveLLMService(api_key="YOUR_API_KEY") + + # Upload a file from the client + file_path = "/path/to/user_uploaded_file.pdf" + file_info = await gemini_service.file_api.upload_file(file_path) + + # Get file URI and mime type from response + file_uri = file_info["file"]["uri"] + mime_type = "application/pdf" # Set appropriate MIME type + + # When starting a new bot session: + # 1. Initialize the context + context = GeminiMultimodalLiveContext() + + # 2. Add file reference to context BEFORE starting the conversation + context.add_file_reference( + file_uri=file_uri, + mime_type=mime_type, + text="Please analyze this document" + ) + + # 3. Now set the context to start the conversation with file reference included + await gemini_service.set_context(context) + + # Gemini now has access to the file reference in its context window + # The file URI remains valid for 48 hours before Google deletes it + + # Optional: List all files for this user + files = await gemini_service.file_api.list_files() + + # Optional: Get metadata for a specific file + file_metadata = await gemini_service.file_api.get_file(file_info["file"]["name"]) + + # Optional: Delete a file when no longer needed + await gemini_service.file_api.delete_file(file_info["file"]["name"]) + ``` + + Notes: + - Files are stored for 48 hours on Google's servers + - Maximum file size is 2GB + - Total storage per project is 20GB + - File references should be added to the context BEFORE starting the conversation + - The same file reference can be reused for multiple sessions within the 48-hour window + """ + def __init__( self, *, @@ -445,6 +534,7 @@ class GeminiMultimodalLiveLLMService(LLMService): tools: Optional[Union[List[dict], ToolsSchema]] = None, params: Optional[InputParams] = None, inference_on_context_initialization: bool = True, + file_api_base_url: str = "https://generativelanguage.googleapis.com/v1beta/files", **kwargs, ): """Initialize the Gemini Multimodal Live LLM service. @@ -522,6 +612,9 @@ 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) def can_generate_metrics(self) -> bool: """Check if the service can generate usage metrics. @@ -938,7 +1031,7 @@ class GeminiMultimodalLiveLLMService(LLMService): self._needs_turn_complete_message = True async def _create_single_response(self, messages_list): - # refactor to combine this logic with same logic in GeminiMultimodalLiveContext + # Refactor to combine this logic with same logic in GeminiMultimodalLiveContext messages = [] for item in messages_list: role = item.get("role") @@ -957,6 +1050,14 @@ class GeminiMultimodalLiveLLMService(LLMService): for part in content: if part.get("type") == "text": 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") + } + }) else: logger.warning(f"Unsupported content type: {str(part)[:80]}") else: From e02b95fca5d6e23d670aae953b21f602ab6b5c31 Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 9 May 2025 10:51:24 -0400 Subject: [PATCH 02/18] Create file_api --- .../services/gemini_multimodal_live/file_api | 179 ++++++++++++++++++ 1 file changed, 179 insertions(+) create mode 100644 src/pipecat/services/gemini_multimodal_live/file_api diff --git a/src/pipecat/services/gemini_multimodal_live/file_api b/src/pipecat/services/gemini_multimodal_live/file_api new file mode 100644 index 000000000..10c8a44db --- /dev/null +++ b/src/pipecat/services/gemini_multimodal_live/file_api @@ -0,0 +1,179 @@ +import aiohttp +import mimetypes +from typing import Dict, Any, Optional + +from loguru import logger + +class GeminiFileAPI: + """Client for the Gemini File API. + + This class provides methods for uploading, fetching, listing, and deleting files + through Google's Gemini File API. + + Files uploaded through this API remain available for 48 hours and can be referenced + in calls to the Gemini generative models. Maximum file size is 2GB, with total + project storage limited to 20GB. + """ + + def __init__(self, api_key: str, base_url: str = "https://generativelanguage.googleapis.com/v1beta/files"): + """Initialize the Gemini File API client. + + Args: + api_key: Google AI API key + base_url: Base URL for the Gemini File API (default is the v1beta endpoint) + """ + self.api_key = api_key + self.base_url = base_url + + async def upload_file(self, file_path: str, display_name: Optional[str] = None) -> Dict[str, Any]: + """Upload a file to the Gemini File API. + + Args: + file_path: Path to the file to upload + display_name: Optional display name for the file + + Returns: + File metadata including uri, name, and display_name + """ + logger.info(f"Uploading file: {file_path}") + + async with aiohttp.ClientSession() as session: + # Determine the file's MIME type + mime_type, _ = mimetypes.guess_type(file_path) + if not mime_type: + mime_type = "application/octet-stream" + + # Read the file + with open(file_path, "rb") as f: + file_data = f.read() + + # First request to initiate the upload + headers = { + "X-Goog-Upload-Protocol": "resumable", + "X-Goog-Upload-Command": "start", + "X-Goog-Upload-Header-Content-Length": str(len(file_data)), + "X-Goog-Upload-Header-Content-Type": mime_type, + "Content-Type": "application/json" + } + + # Create the metadata payload + metadata = {} + if display_name: + metadata = {"file": {"display_name": display_name}} + + # Initial request to get the upload URL + async with session.post( + f"{self.base_url}?key={self.api_key}", + headers=headers, + json=metadata + ) as response: + if response.status != 200: + error_text = await response.text() + logger.error(f"Error initiating file upload: {error_text}") + raise Exception(f"Failed to initiate upload: {response.status}") + + # Get the upload URL from the response header + upload_url = response.headers.get("X-Goog-Upload-URL") + if not upload_url: + raise Exception("No upload URL in response") + + # Upload the actual file + headers = { + "Content-Length": str(len(file_data)), + "X-Goog-Upload-Offset": "0", + "X-Goog-Upload-Command": "upload, finalize" + } + + async with session.post( + upload_url, + headers=headers, + data=file_data + ) as response: + if response.status != 200: + error_text = await response.text() + logger.error(f"Error uploading file: {error_text}") + raise Exception(f"Failed to upload file: {response.status}") + + file_info = await response.json() + logger.info(f"File uploaded successfully: {file_info.get('file', {}).get('name')}") + return file_info + + async def get_file(self, name: str) -> Dict[str, Any]: + """Get metadata for a file. + + Args: + name: File name (or full path) + + Returns: + File metadata + """ + # Extract just the name part if a full path is provided + if '/' in name: + name = name.split('/')[-1] + + async with aiohttp.ClientSession() as session: + async with session.get( + f"{self.base_url}/{name}?key={self.api_key}" + ) as response: + if response.status != 200: + error_text = await response.text() + logger.error(f"Error getting file metadata: {error_text}") + raise Exception(f"Failed to get file metadata: {response.status}") + + file_info = await response.json() + return file_info + + async def list_files(self, page_size: int = 10, page_token: Optional[str] = None) -> Dict[str, Any]: + """List uploaded files. + + Args: + page_size: Number of files to return per page + page_token: Token for pagination + + Returns: + List of files and next page token if available + """ + params = { + "key": self.api_key, + "pageSize": page_size + } + + if page_token: + params["pageToken"] = page_token + + async with aiohttp.ClientSession() as session: + async with session.get( + self.base_url, + params=params + ) as response: + if response.status != 200: + error_text = await response.text() + logger.error(f"Error listing files: {error_text}") + raise Exception(f"Failed to list files: {response.status}") + + result = await response.json() + return result + + async def delete_file(self, name: str) -> bool: + """Delete a file. + + Args: + name: File name (or full path) + + Returns: + True if deleted successfully + """ + # Extract just the name part if a full path is provided + if '/' in name: + name = name.split('/')[-1] + + async with aiohttp.ClientSession() as session: + async with session.delete( + f"{self.base_url}/{name}?key={self.api_key}" + ) as response: + if response.status != 200: + error_text = await response.text() + logger.error(f"Error deleting file: {error_text}") + raise Exception(f"Failed to delete file: {response.status}") + + return True From 9171d4b0408c9c954e55a56f19558f8258d29ecc Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 9 May 2025 10:52:04 -0400 Subject: [PATCH 03/18] add FileData class events.py --- src/pipecat/services/gemini_multimodal_live/events.py | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/src/pipecat/services/gemini_multimodal_live/events.py b/src/pipecat/services/gemini_multimodal_live/events.py index 160ff1174..b70e2bf3e 100644 --- a/src/pipecat/services/gemini_multimodal_live/events.py +++ b/src/pipecat/services/gemini_multimodal_live/events.py @@ -44,6 +44,16 @@ class ContentPart(BaseModel): text: Optional[str] = Field(default=None, validate_default=False) inlineData: Optional[MediaChunk] = Field(default=None, validate_default=False) + fileData: Optional['FileData'] = Field(default=None, validate_default=False) + + +class FileData(BaseModel): + """Represents a file reference in the Gemini File API.""" + mimeType: str + fileUri: str + + +ContentPart.model_rebuild() # Rebuild model to resolve forward reference class Turn(BaseModel): From 77c369c3c76f8bb9c1a44d341b97e0133d1d85fd Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 9 May 2025 10:53:31 -0400 Subject: [PATCH 04/18] add file_api __init__.py --- src/pipecat/services/gemini_multimodal_live/__init__.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/pipecat/services/gemini_multimodal_live/__init__.py b/src/pipecat/services/gemini_multimodal_live/__init__.py index 61bdf58dd..60a226e64 100644 --- a/src/pipecat/services/gemini_multimodal_live/__init__.py +++ b/src/pipecat/services/gemini_multimodal_live/__init__.py @@ -1 +1,2 @@ from .gemini import GeminiMultimodalLiveLLMService +from .file_api import GeminiFileAPI From 1ec1aa76e941a10501edafd4492d7d67e830a6a3 Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Tue, 13 May 2025 22:01:02 -0400 Subject: [PATCH 05/18] Rename file_api to file_api.py added proper .py to file name. --- .../services/gemini_multimodal_live/{file_api => file_api.py} | 0 1 file changed, 0 insertions(+), 0 deletions(-) rename src/pipecat/services/gemini_multimodal_live/{file_api => file_api.py} (100%) diff --git a/src/pipecat/services/gemini_multimodal_live/file_api b/src/pipecat/services/gemini_multimodal_live/file_api.py similarity index 100% rename from src/pipecat/services/gemini_multimodal_live/file_api rename to src/pipecat/services/gemini_multimodal_live/file_api.py From bd7ca9419615e88061c4287a29ddf9dea8f3c8b2 Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 30 May 2025 12:19:40 -0400 Subject: [PATCH 06/18] Update gemini.py --- .../services/gemini_multimodal_live/gemini.py | 56 ------------------- 1 file changed, 56 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index cc85799db..b270f7721 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -464,63 +464,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter - - """Gemini Live LLM Service with multimodal capabilities including File API support. - This service implements the Gemini Multimodal Live API with support for: - - Audio input and output - - Image/video input - - File API (upload, reference, and management) - - Tools/function calling - - Example usage of File API: - ```python - # Initialize the service - gemini_service = GeminiMultimodalLiveLLMService(api_key="YOUR_API_KEY") - - # Upload a file from the client - file_path = "/path/to/user_uploaded_file.pdf" - file_info = await gemini_service.file_api.upload_file(file_path) - - # Get file URI and mime type from response - file_uri = file_info["file"]["uri"] - mime_type = "application/pdf" # Set appropriate MIME type - - # When starting a new bot session: - # 1. Initialize the context - context = GeminiMultimodalLiveContext() - - # 2. Add file reference to context BEFORE starting the conversation - context.add_file_reference( - file_uri=file_uri, - mime_type=mime_type, - text="Please analyze this document" - ) - - # 3. Now set the context to start the conversation with file reference included - await gemini_service.set_context(context) - - # Gemini now has access to the file reference in its context window - # The file URI remains valid for 48 hours before Google deletes it - - # Optional: List all files for this user - files = await gemini_service.file_api.list_files() - - # Optional: Get metadata for a specific file - file_metadata = await gemini_service.file_api.get_file(file_info["file"]["name"]) - - # Optional: Delete a file when no longer needed - await gemini_service.file_api.delete_file(file_info["file"]["name"]) - ``` - - Notes: - - Files are stored for 48 hours on Google's servers - - Maximum file size is 2GB - - Total storage per project is 20GB - - File references should be added to the context BEFORE starting the conversation - - The same file reference can be reused for multiple sessions within the 48-hour window - """ - def __init__( self, *, From 7b1071b30d88e290e446a3fc11ca2a8922a56a8d Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 30 May 2025 13:04:52 -0400 Subject: [PATCH 07/18] Create 26f-gemini-multimodal-live-files-api.py This is an example to test usage of the Files API integration. Specifically with the Gemini Multimodal Live Service. --- .../26f-gemini-multimodal-live-files-api.py | 210 ++++++++++++++++++ 1 file changed, 210 insertions(+) create mode 100644 examples/foundational/26f-gemini-multimodal-live-files-api.py diff --git a/examples/foundational/26f-gemini-multimodal-live-files-api.py b/examples/foundational/26f-gemini-multimodal-live-files-api.py new file mode 100644 index 000000000..413830cf9 --- /dev/null +++ b/examples/foundational/26f-gemini-multimodal-live-files-api.py @@ -0,0 +1,210 @@ +# +# Copyright (c) 2024ā€“2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import os +import tempfile + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.gemini_multimodal_live.gemini import ( + GeminiMultimodalLiveLLMService, + GeminiMultimodalLiveContext, +) +from pipecat.transports.base_transport import TransportParams +from pipecat.transports.network.small_webrtc import SmallWebRTCTransport +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +load_dotenv(override=True) + + +async def create_sample_file(): + """Create a sample text file for testing the File API.""" + content = """# Sample Document for Gemini File API Test + +This is a test document to demonstrate the Gemini File API functionality. + +## Key Information: +- This document was created for testing purposes +- It contains information about AI assistants +- The document should be analyzed by Gemini +- The secret phrase for the test is "Pineapple Pizza" + +## AI Assistant Capabilities: +1. Natural language processing +2. File analysis and understanding +3. Context-aware conversations +4. Multi-modal interactions + +## Conclusion: +This document serves as a test case for the Gemini File API integration with Pipecat. +The AI should be able to reference and discuss the contents of this file. +""" + + # Create a temporary file + with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f: + f.write(content) + return f.name + + +async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): + logger.info(f"Starting File API bot") + + # Create a sample file to upload + sample_file_path = await create_sample_file() + logger.info(f"Created sample file: {sample_file_path}") + + # Initialize the SmallWebRTCTransport with the connection + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=False, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), + ) + + system_instruction = """ + You are a helpful AI assistant with access to a document that has been uploaded for analysis. + + The document contains test information including a secret phrase. You should be able to: + - Reference and discuss the contents of the uploaded document + - Answer questions about what's in the document + - Use the information from the document in our conversation + + Your output will be converted to audio so don't include special characters in your answers. + Be friendly and demonstrate your ability to work with the uploaded file. + """ + + # Initialize Gemini service with File API support + llm = GeminiMultimodalLiveLLMService( + api_key=os.getenv("GOOGLE_API_KEY"), + system_instruction=system_instruction, + voice_id="Charon", # Aoede, Charon, Fenrir, Kore, Puck + transcribe_user_audio=True, + ) + + # Upload the sample file to Gemini File API + logger.info("Uploading file to Gemini File API...") + file_info = None + try: + file_info = await llm.file_api.upload_file( + sample_file_path, + display_name="Sample Test Document" + ) + logger.info(f"File uploaded successfully: {file_info['file']['name']}") + + # Get file URI and mime type + file_uri = file_info["file"]["uri"] + mime_type = "text/plain" + + # Create context with file reference + context = OpenAILLMContext( + [ + { + "role": "user", + "content": [ + { + "type": "text", + "text": "Greet the user and let them know you have access to a document they can ask you about. Mention that you can discuss its contents." + }, + { + "type": "file_data", + "file_data": { + "mime_type": mime_type, + "file_uri": file_uri + } + } + ] + } + ] + ) + + logger.info("File reference added to conversation context") + + except Exception as e: + logger.error(f"Error uploading file: {e}") + # Continue with a basic context if file upload fails + context = OpenAILLMContext( + [ + { + "role": "user", + "content": "Greet the user and explain that there was an issue with file upload, but you're ready to help with other tasks." + } + ] + ) + + # Create context aggregator + context_aggregator = llm.create_context_aggregator(context) + + # Build the pipeline + pipeline = Pipeline([ + transport.input(), + context_aggregator.user(), + llm, + transport.output(), + context_aggregator.assistant(), + ]) + + # Configure the pipeline task + task = PipelineTask( + pipeline, + params=PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + ), + ) + + # Handle client connection event + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation using standard context frame + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + # Handle client disconnection events + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() + + # Run the pipeline + runner = PipelineRunner(handle_sigint=False) + await runner.run(task) + + # Clean up: delete the uploaded file and temporary file + if file_info: + try: + await llm.file_api.delete_file(file_info["file"]["name"]) + logger.info("Cleaned up uploaded file from Gemini") + except Exception as e: + logger.error(f"Error cleaning up file: {e}") + + # Remove temporary file + try: + os.unlink(sample_file_path) + logger.info("Cleaned up temporary file") + except Exception as e: + logger.error(f"Error removing temporary file: {e}") + + +if __name__ == "__main__": + from run import main + + main() From baccf504179629166e1df7f4c709b35c309de8a3 Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 30 May 2025 13:41:55 -0400 Subject: [PATCH 08/18] update correct upload endpoint file_api.py --- .../gemini_multimodal_live/file_api.py | 44 ++++++++++++------- 1 file changed, 28 insertions(+), 16 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/file_api.py b/src/pipecat/services/gemini_multimodal_live/file_api.py index 10c8a44db..39b7d9df9 100644 --- a/src/pipecat/services/gemini_multimodal_live/file_api.py +++ b/src/pipecat/services/gemini_multimodal_live/file_api.py @@ -1,3 +1,9 @@ +# +# Copyright (c) 2024ā€“2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + import aiohttp import mimetypes from typing import Dict, Any, Optional @@ -24,9 +30,11 @@ class GeminiFileAPI: """ self.api_key = api_key self.base_url = base_url + # Upload URL uses the /upload/ path + self.upload_base_url = "https://generativelanguage.googleapis.com/upload/v1beta/files" async def upload_file(self, file_path: str, display_name: Optional[str] = None) -> Dict[str, Any]: - """Upload a file to the Gemini File API. + """Upload a file to the Gemini File API using the correct resumable upload protocol. Args: file_path: Path to the file to upload @@ -47,7 +55,12 @@ class GeminiFileAPI: with open(file_path, "rb") as f: file_data = f.read() - # First request to initiate the upload + # Create the metadata payload + metadata = {} + if display_name: + metadata = {"file": {"display_name": display_name}} + + # Step 1: Initial resumable request to get upload URL headers = { "X-Goog-Upload-Protocol": "resumable", "X-Goog-Upload-Command": "start", @@ -56,43 +69,42 @@ class GeminiFileAPI: "Content-Type": "application/json" } - # Create the metadata payload - metadata = {} - if display_name: - metadata = {"file": {"display_name": display_name}} - - # Initial request to get the upload URL + logger.debug(f"Step 1: Getting upload URL from {self.upload_base_url}") async with session.post( - f"{self.base_url}?key={self.api_key}", + f"{self.upload_base_url}?key={self.api_key}", headers=headers, json=metadata ) as response: if response.status != 200: error_text = await response.text() logger.error(f"Error initiating file upload: {error_text}") - raise Exception(f"Failed to initiate upload: {response.status}") + raise Exception(f"Failed to initiate upload: {response.status} - {error_text}") # Get the upload URL from the response header upload_url = response.headers.get("X-Goog-Upload-URL") if not upload_url: - raise Exception("No upload URL in response") + logger.error(f"Response headers: {dict(response.headers)}") + raise Exception("No upload URL in response headers") + + logger.debug(f"Got upload URL: {upload_url}") - # Upload the actual file - headers = { + # Step 2: Upload the actual file data + upload_headers = { "Content-Length": str(len(file_data)), "X-Goog-Upload-Offset": "0", "X-Goog-Upload-Command": "upload, finalize" } + logger.debug(f"Step 2: Uploading file data to {upload_url}") async with session.post( upload_url, - headers=headers, + headers=upload_headers, data=file_data ) as response: if response.status != 200: error_text = await response.text() - logger.error(f"Error uploading file: {error_text}") - raise Exception(f"Failed to upload file: {response.status}") + logger.error(f"Error uploading file data: {error_text}") + raise Exception(f"Failed to upload file: {response.status} - {error_text}") file_info = await response.json() logger.info(f"File uploaded successfully: {file_info.get('file', {}).get('name')}") From 6e6e932370eb74f36a30ddfaeb35e57900a118a8 Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 30 May 2025 17:36:36 -0400 Subject: [PATCH 09/18] Create 26g-gemini-multimodal-live-groundingMetadata.py --- ...emini-multimodal-live-groundingMetadata.py | 164 ++++++++++++++++++ 1 file changed, 164 insertions(+) create mode 100644 examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py diff --git a/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py b/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py new file mode 100644 index 000000000..0c6d35ff0 --- /dev/null +++ b/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py @@ -0,0 +1,164 @@ + +import argparse +import os + +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.audio.vad.vad_analyzer import VADParams +from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema +from pipecat.frames.frames import Frame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService +from pipecat.services.google.frames import LLMSearchResponseFrame +from pipecat.transports.base_transport import TransportParams +from pipecat.transports.network.small_webrtc import SmallWebRTCTransport +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +load_dotenv(override=True) + +SYSTEM_INSTRUCTION = """ +You are a helpful AI assistant that actively uses Google Search to provide up-to-date, accurate information. + +IMPORTANT: For ANY question about current events, news, recent developments, real-time information, or anything that might have changed recently, you MUST use the google_search tool to get the latest information. + +You should use Google Search for: +- Current news and events +- Recent developments in any field +- Today's weather, stock prices, or other real-time data +- Any question that starts with "what's happening", "latest", "recent", "current", "today", etc. +- When you're not certain about recent information + +Always be proactive about using search when the user asks about anything that could benefit from real-time information. + +Your output will be converted to audio so don't include special characters in your answers. + +Respond to what the user said in a creative and helpful way, always using search for current information. +""" + + +class GroundingMetadataProcessor(FrameProcessor): + """Processor to capture and display grounding metadata from Gemini Live API.""" + + def __init__(self): + super().__init__() + self._grounding_count = 0 + + async def process_frame(self, frame: Frame, direction: FrameDirection): + # Always call super().process_frame first + await super().process_frame(frame, direction) + + # Only log important frame types, not every audio frame + if hasattr(frame, '__class__'): + frame_type = frame.__class__.__name__ + if frame_type in ['LLMTextFrame', 'TTSTextFrame', 'LLMFullResponseStartFrame', 'LLMFullResponseEndFrame']: + logger.debug(f"GroundingProcessor received: {frame_type}") + + if isinstance(frame, LLMSearchResponseFrame): + self._grounding_count += 1 + logger.info(f"\nšŸ” GROUNDING METADATA RECEIVED #{self._grounding_count}") + logger.info(f"šŸ“ Search Result Text: {frame.search_result[:200]}...") + + if frame.rendered_content: + logger.info(f"šŸ”— Rendered Content: {frame.rendered_content}") + + if frame.origins: + logger.info(f"šŸ“ Number of Origins: {len(frame.origins)}") + for i, origin in enumerate(frame.origins): + logger.info(f" Origin {i+1}: {origin.site_title} - {origin.site_uri}") + if origin.results: + logger.info(f" Results: {len(origin.results)} items") + + # Always push the frame downstream + await self.push_frame(frame, direction) + + +async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): + logger.info(f"Starting Gemini Live Grounding Test Bot") + + # Initialize the SmallWebRTCTransport with the connection + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=False, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), + ) + + # Create tools using ToolsSchema with custom tools for Gemini + tools = ToolsSchema( + standard_tools=[], # No standard function declarations needed + custom_tools={ + AdapterType.GEMINI: [ + {"google_search": {}}, + {"code_execution": {}} + ] + } + ) + + llm = GeminiMultimodalLiveLLMService( + api_key=os.getenv("GOOGLE_API_KEY"), + system_instruction=SYSTEM_INSTRUCTION, + voice_id="Charon", # Aoede, Charon, Fenrir, Kore, Puck + transcribe_user_audio=True, + tools=tools, + ) + + # Create a processor to capture grounding metadata + grounding_processor = GroundingMetadataProcessor() + + messages = [ + { + "role": "user", + "content": 'Please introduce yourself and let me know that you can help with current information by searching the web. Ask me what current information I\'d like to know about.', + }, + ] + + # Set up conversation context and management + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), + context_aggregator.user(), + llm, + grounding_processor, # Add our grounding processor here + transport.output(), + context_aggregator.assistant(), + ] + ) + + task = PipelineTask(pipeline) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) + + +if __name__ == "__main__": + from run import main + + main() From 44d3bd30fae6a29d115a6cf767f512a4d18c28f0 Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 30 May 2025 18:01:15 -0400 Subject: [PATCH 10/18] Add groundingMetadata and logging gemini.py --- .../services/gemini_multimodal_live/gemini.py | 109 ++++++++++++++++++ 1 file changed, 109 insertions(+) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index b270f7721..301290b45 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -60,6 +60,8 @@ from pipecat.processors.aggregators.openai_llm_context import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.llm_service import FunctionCallFromLLM, LLMService +from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult +from pipecat.services.llm_service import LLMService from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, OpenAIUserContextAggregator, @@ -560,6 +562,10 @@ class GeminiMultimodalLiveLLMService(LLMService): # Initialize the File API client self.file_api = GeminiFileAPI(api_key=api_key, base_url=file_api_base_url) + # Grounding metadata tracking + self._search_result_buffer = "" + self._accumulated_grounding_metadata = None + def can_generate_metrics(self) -> bool: """Check if the service can generate usage metrics. @@ -909,12 +915,23 @@ class GeminiMultimodalLiveLLMService(LLMService): await self._handle_evt_input_transcription(evt) elif evt.serverContent and evt.serverContent.outputTranscription: await self._handle_evt_output_transcription(evt) + elif evt.serverContent and evt.serverContent.groundingMetadata: + await self._handle_evt_grounding_metadata(evt) elif evt.toolCall: await self._handle_evt_tool_call(evt) elif False: # !!! todo: error events? await self._handle_evt_error(evt) # errors are fatal, so exit the receive loop return + else: + # Log unhandled events that might contain grounding metadata + logger.warning(f"Received unhandled server event type: {evt}") + pass + + async def _transcribe_audio_handler(self): + while True: + audio = await self._transcribe_audio_queue.get() + await self._handle_transcribe_user_audio(audio, self._context) # # @@ -1071,8 +1088,14 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.push_frame(LLMFullResponseStartFrame()) self._bot_text_buffer += text + self._search_result_buffer += text # Also accumulate for grounding await self.push_frame(LLMTextFrame(text=text)) + # Check for grounding metadata in server content + if evt.serverContent and evt.serverContent.groundingMetadata: + self._accumulated_grounding_metadata = evt.serverContent.groundingMetadata + logger.debug("Grounding metadata detected in model turn.") + inline_data = part.inlineData if not inline_data: return @@ -1139,6 +1162,18 @@ class GeminiMultimodalLiveLLMService(LLMService): self._llm_output_buffer = "" # Only push the TTSStoppedFrame if the bot is outputting audio + # Process grounding metadata if we have accumulated any + if self._accumulated_grounding_metadata: + logger.debug("Processing grounding metadata...") + await self._process_grounding_metadata(self._accumulated_grounding_metadata, self._search_result_buffer) + else: + logger.debug("No grounding metadata to process") + + # Reset grounding tracking for next response + self._search_result_buffer = "" + self._accumulated_grounding_metadata = None + + # Only push the TTSStoppedFrame the bot is outputting audio # when text is found, modalities is set to TEXT and no audio # is produced. if not text: @@ -1216,6 +1251,13 @@ class GeminiMultimodalLiveLLMService(LLMService): # Collect text for tracing self._llm_output_buffer += text + # Accumulate text for grounding as well + self._search_result_buffer += text + + # Check for grounding metadata in server content + if evt.serverContent and evt.serverContent.groundingMetadata: + self._accumulated_grounding_metadata = evt.serverContent.groundingMetadata + await self.push_frame(LLMTextFrame(text=text)) await self.push_frame(TTSTextFrame(text=text)) @@ -1238,6 +1280,73 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.start_llm_usage_metrics(tokens) + async def _handle_evt_grounding_metadata(self, evt): + """Handle dedicated grounding metadata events.""" + logger.debug("Received dedicated grounding metadata event.") + + if evt.serverContent and evt.serverContent.groundingMetadata: + grounding_metadata = evt.serverContent.groundingMetadata + logger.debug(f"Grounding data: {len(grounding_metadata.groundingChunks or [])} chunks, {len(grounding_metadata.groundingSupports or [])} supports") + + # Process the grounding metadata immediately + await self._process_grounding_metadata(grounding_metadata, self._search_result_buffer) + + async def _process_grounding_metadata(self, grounding_metadata: events.GroundingMetadata, search_result: str = ""): + """Process grounding metadata and emit LLMSearchResponseFrame.""" + logger.debug(f"Processing grounding metadata. Search result text length: {len(search_result)}") + if not grounding_metadata: + logger.warning("No grounding metadata provided to _process_grounding_metadata") + return + + # logger.debug(f"Processing grounding metadata: {grounding_metadata}") # Too verbose for PR + + # Extract rendered content for search suggestions + rendered_content = None + if grounding_metadata.searchEntryPoint and grounding_metadata.searchEntryPoint.renderedContent: + rendered_content = grounding_metadata.searchEntryPoint.renderedContent + + # Convert grounding chunks and supports to LLMSearchOrigin format + origins = [] + + if grounding_metadata.groundingChunks and grounding_metadata.groundingSupports: + # Create a mapping of chunk indices to origins + chunk_to_origin = {} + + for index, chunk in enumerate(grounding_metadata.groundingChunks): + if chunk.web: + origin = LLMSearchOrigin( + site_uri=chunk.web.uri, + site_title=chunk.web.title, + results=[] + ) + chunk_to_origin[index] = origin + origins.append(origin) + + # Add grounding support results to the appropriate origins + for support in grounding_metadata.groundingSupports: + if support.segment and support.groundingChunkIndices: + text = support.segment.text or "" + confidence_scores = support.confidenceScores or [] + + # Add this result to all origins referenced by this support + for chunk_index in support.groundingChunkIndices: + if chunk_index in chunk_to_origin: + result = LLMSearchResult( + text=text, + confidence=confidence_scores + ) + chunk_to_origin[chunk_index].results.append(result) + + # Create and push the search response frame + search_frame = LLMSearchResponseFrame( + search_result=search_result, + origins=origins, + rendered_content=rendered_content + ) + + logger.debug(f"Emitting LLMSearchResponseFrame with {len(origins)} origins, rendered_content available: {rendered_content is not None}") + await self.push_frame(search_frame) + def create_context_aggregator( self, context: OpenAILLMContext, From 4fd8df208f63733a2733759c3ecb4c172c7402d1 Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 30 May 2025 18:07:09 -0400 Subject: [PATCH 11/18] Add groundingMetadata events.py --- .../services/gemini_multimodal_live/events.py | 68 +++++++++++++++++-- 1 file changed, 64 insertions(+), 4 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/events.py b/src/pipecat/services/gemini_multimodal_live/events.py index b70e2bf3e..6df94148b 100644 --- a/src/pipecat/services/gemini_multimodal_live/events.py +++ b/src/pipecat/services/gemini_multimodal_live/events.py @@ -12,6 +12,7 @@ import json from enum import Enum from typing import List, Literal, Optional +from loguru import logger from PIL import Image from pydantic import BaseModel, Field @@ -247,6 +248,49 @@ class Config(BaseModel): setup: Setup +# +# Grounding metadata models +# + + +class SearchEntryPoint(BaseModel): + """Represents the search entry point with rendered content for search suggestions.""" + renderedContent: Optional[str] = None + + +class WebSource(BaseModel): + """Represents a web source from grounding chunks.""" + uri: Optional[str] = None + title: Optional[str] = None + + +class GroundingChunk(BaseModel): + """Represents a grounding chunk containing web source information.""" + web: Optional[WebSource] = None + + +class GroundingSegment(BaseModel): + """Represents a segment of text that is grounded.""" + startIndex: Optional[int] = None + endIndex: Optional[int] = None + text: Optional[str] = None + + +class GroundingSupport(BaseModel): + """Represents support information for grounded text segments.""" + segment: Optional[GroundingSegment] = None + groundingChunkIndices: Optional[List[int]] = None + confidenceScores: Optional[List[float]] = None + + +class GroundingMetadata(BaseModel): + """Represents grounding metadata from Google Search.""" + searchEntryPoint: Optional[SearchEntryPoint] = None + groundingChunks: Optional[List[GroundingChunk]] = None + groundingSupports: Optional[List[GroundingSupport]] = None + webSearchQueries: Optional[List[str]] = None + + # # Server events # @@ -338,6 +382,7 @@ class ServerContent(BaseModel): turnComplete: Optional[bool] = None inputTranscription: Optional[BidiGenerateContentTranscription] = None outputTranscription: Optional[BidiGenerateContentTranscription] = None + groundingMetadata: Optional[GroundingMetadata] = None class FunctionCall(BaseModel): @@ -430,7 +475,7 @@ class ServerEvent(BaseModel): usageMetadata: Optional[UsageMetadata] = None -def parse_server_event(str): +def parse_server_event(message_str): """Parse a server event from JSON string. Args: @@ -439,11 +484,26 @@ def parse_server_event(str): Returns: ServerEvent instance if parsing succeeds, None otherwise. """ + from loguru import logger # Import logger locally to avoid scoping issues + try: - evt = json.loads(str) - return ServerEvent.model_validate(evt) + evt_dict = json.loads(message_str) + + # Only log grounding metadata detection if truly needed for debugging + # In production, this could be removed entirely or moved to TRACE level + if 'serverContent' in evt_dict: + server_content = evt_dict['serverContent'] + if 'groundingMetadata' in server_content: + # Consider removing this log entirely for production + pass + + evt = ServerEvent.model_validate(evt_dict) + return evt except Exception as e: - print(f"Error parsing server event: {e}") + logger.error(f"Error parsing server event: {e}") + # Truncate raw message to avoid logging potentially sensitive or overly long data + truncated_message = message_str[:200] + "..." if len(message_str) > 200 else message_str + logger.error(f"Raw message (truncated): {truncated_message}") return None From a63d0da528872a07127f6fe0dd4d3c9bf716fbd0 Mon Sep 17 00:00:00 2001 From: Pete <78183014+getchannel@users.noreply.github.com> Date: Sat, 21 Jun 2025 14:29:09 -0400 Subject: [PATCH 12/18] Update gemini.py --- src/pipecat/services/gemini_multimodal_live/gemini.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 301290b45..29d17e689 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -62,6 +62,8 @@ from pipecat.processors.frame_processor import FrameDirection from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult from pipecat.services.llm_service import LLMService +from pipecat.services.llm_service import FunctionCallFromLLM, LLMService + from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, OpenAIUserContextAggregator, From 79e51051c7cfa16344bd456c995c1baa53d22186 Mon Sep 17 00:00:00 2001 From: vipyne Date: Tue, 1 Jul 2025 16:25:59 -0500 Subject: [PATCH 13/18] New lint rules and remove unused example file --- ...emini-multimodal-live-groundingMetadata.py | 164 ------------------ .../gemini_multimodal_live/__init__.py | 2 +- .../services/gemini_multimodal_live/events.py | 19 +- .../gemini_multimodal_live/file_api.py | 125 +++++++------ .../services/gemini_multimodal_live/gemini.py | 98 ++++++----- 5 files changed, 125 insertions(+), 283 deletions(-) delete mode 100644 examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py diff --git a/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py b/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py deleted file mode 100644 index 0c6d35ff0..000000000 --- a/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py +++ /dev/null @@ -1,164 +0,0 @@ - -import argparse -import os - -from dotenv import load_dotenv -from loguru import logger - -from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.adapters.schemas.tools_schema import AdapterType, ToolsSchema -from pipecat.frames.frames import Frame -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService -from pipecat.services.google.frames import LLMSearchResponseFrame -from pipecat.transports.base_transport import TransportParams -from pipecat.transports.network.small_webrtc import SmallWebRTCTransport -from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection - -load_dotenv(override=True) - -SYSTEM_INSTRUCTION = """ -You are a helpful AI assistant that actively uses Google Search to provide up-to-date, accurate information. - -IMPORTANT: For ANY question about current events, news, recent developments, real-time information, or anything that might have changed recently, you MUST use the google_search tool to get the latest information. - -You should use Google Search for: -- Current news and events -- Recent developments in any field -- Today's weather, stock prices, or other real-time data -- Any question that starts with "what's happening", "latest", "recent", "current", "today", etc. -- When you're not certain about recent information - -Always be proactive about using search when the user asks about anything that could benefit from real-time information. - -Your output will be converted to audio so don't include special characters in your answers. - -Respond to what the user said in a creative and helpful way, always using search for current information. -""" - - -class GroundingMetadataProcessor(FrameProcessor): - """Processor to capture and display grounding metadata from Gemini Live API.""" - - def __init__(self): - super().__init__() - self._grounding_count = 0 - - async def process_frame(self, frame: Frame, direction: FrameDirection): - # Always call super().process_frame first - await super().process_frame(frame, direction) - - # Only log important frame types, not every audio frame - if hasattr(frame, '__class__'): - frame_type = frame.__class__.__name__ - if frame_type in ['LLMTextFrame', 'TTSTextFrame', 'LLMFullResponseStartFrame', 'LLMFullResponseEndFrame']: - logger.debug(f"GroundingProcessor received: {frame_type}") - - if isinstance(frame, LLMSearchResponseFrame): - self._grounding_count += 1 - logger.info(f"\nšŸ” GROUNDING METADATA RECEIVED #{self._grounding_count}") - logger.info(f"šŸ“ Search Result Text: {frame.search_result[:200]}...") - - if frame.rendered_content: - logger.info(f"šŸ”— Rendered Content: {frame.rendered_content}") - - if frame.origins: - logger.info(f"šŸ“ Number of Origins: {len(frame.origins)}") - for i, origin in enumerate(frame.origins): - logger.info(f" Origin {i+1}: {origin.site_title} - {origin.site_uri}") - if origin.results: - logger.info(f" Results: {len(origin.results)} items") - - # Always push the frame downstream - await self.push_frame(frame, direction) - - -async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): - logger.info(f"Starting Gemini Live Grounding Test Bot") - - # Initialize the SmallWebRTCTransport with the connection - transport = SmallWebRTCTransport( - webrtc_connection=webrtc_connection, - params=TransportParams( - audio_in_enabled=True, - audio_out_enabled=True, - video_in_enabled=False, - vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), - ), - ) - - # Create tools using ToolsSchema with custom tools for Gemini - tools = ToolsSchema( - standard_tools=[], # No standard function declarations needed - custom_tools={ - AdapterType.GEMINI: [ - {"google_search": {}}, - {"code_execution": {}} - ] - } - ) - - llm = GeminiMultimodalLiveLLMService( - api_key=os.getenv("GOOGLE_API_KEY"), - system_instruction=SYSTEM_INSTRUCTION, - voice_id="Charon", # Aoede, Charon, Fenrir, Kore, Puck - transcribe_user_audio=True, - tools=tools, - ) - - # Create a processor to capture grounding metadata - grounding_processor = GroundingMetadataProcessor() - - messages = [ - { - "role": "user", - "content": 'Please introduce yourself and let me know that you can help with current information by searching the web. Ask me what current information I\'d like to know about.', - }, - ] - - # Set up conversation context and management - context = OpenAILLMContext(messages) - context_aggregator = llm.create_context_aggregator(context) - - pipeline = Pipeline( - [ - transport.input(), - context_aggregator.user(), - llm, - grounding_processor, # Add our grounding processor here - transport.output(), - context_aggregator.assistant(), - ] - ) - - task = PipelineTask(pipeline) - - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected") - # Kick off the conversation. - await task.queue_frames([context_aggregator.user().get_context_frame()]) - - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - - @transport.event_handler("on_client_closed") - async def on_client_closed(transport, client): - logger.info(f"Client closed connection") - await task.cancel() - - runner = PipelineRunner(handle_sigint=False) - - await runner.run(task) - - -if __name__ == "__main__": - from run import main - - main() diff --git a/src/pipecat/services/gemini_multimodal_live/__init__.py b/src/pipecat/services/gemini_multimodal_live/__init__.py index 60a226e64..513d9fd66 100644 --- a/src/pipecat/services/gemini_multimodal_live/__init__.py +++ b/src/pipecat/services/gemini_multimodal_live/__init__.py @@ -1,2 +1,2 @@ -from .gemini import GeminiMultimodalLiveLLMService from .file_api import GeminiFileAPI +from .gemini import GeminiMultimodalLiveLLMService diff --git a/src/pipecat/services/gemini_multimodal_live/events.py b/src/pipecat/services/gemini_multimodal_live/events.py index 6df94148b..4687519c0 100644 --- a/src/pipecat/services/gemini_multimodal_live/events.py +++ b/src/pipecat/services/gemini_multimodal_live/events.py @@ -45,11 +45,12 @@ class ContentPart(BaseModel): text: Optional[str] = Field(default=None, validate_default=False) inlineData: Optional[MediaChunk] = Field(default=None, validate_default=False) - fileData: Optional['FileData'] = Field(default=None, validate_default=False) + fileData: Optional["FileData"] = Field(default=None, validate_default=False) class FileData(BaseModel): """Represents a file reference in the Gemini File API.""" + mimeType: str fileUri: str @@ -255,22 +256,26 @@ class Config(BaseModel): class SearchEntryPoint(BaseModel): """Represents the search entry point with rendered content for search suggestions.""" + renderedContent: Optional[str] = None class WebSource(BaseModel): """Represents a web source from grounding chunks.""" + uri: Optional[str] = None title: Optional[str] = None class GroundingChunk(BaseModel): """Represents a grounding chunk containing web source information.""" + web: Optional[WebSource] = None class GroundingSegment(BaseModel): """Represents a segment of text that is grounded.""" + startIndex: Optional[int] = None endIndex: Optional[int] = None text: Optional[str] = None @@ -278,6 +283,7 @@ class GroundingSegment(BaseModel): class GroundingSupport(BaseModel): """Represents support information for grounded text segments.""" + segment: Optional[GroundingSegment] = None groundingChunkIndices: Optional[List[int]] = None confidenceScores: Optional[List[float]] = None @@ -285,6 +291,7 @@ class GroundingSupport(BaseModel): class GroundingMetadata(BaseModel): """Represents grounding metadata from Google Search.""" + searchEntryPoint: Optional[SearchEntryPoint] = None groundingChunks: Optional[List[GroundingChunk]] = None groundingSupports: Optional[List[GroundingSupport]] = None @@ -485,15 +492,15 @@ def parse_server_event(message_str): ServerEvent instance if parsing succeeds, None otherwise. """ from loguru import logger # Import logger locally to avoid scoping issues - + try: evt_dict = json.loads(message_str) - + # Only log grounding metadata detection if truly needed for debugging # In production, this could be removed entirely or moved to TRACE level - if 'serverContent' in evt_dict: - server_content = evt_dict['serverContent'] - if 'groundingMetadata' in server_content: + if "serverContent" in evt_dict: + server_content = evt_dict["serverContent"] + if "groundingMetadata" in server_content: # Consider removing this log entirely for production pass diff --git a/src/pipecat/services/gemini_multimodal_live/file_api.py b/src/pipecat/services/gemini_multimodal_live/file_api.py index 39b7d9df9..2c79338b5 100644 --- a/src/pipecat/services/gemini_multimodal_live/file_api.py +++ b/src/pipecat/services/gemini_multimodal_live/file_api.py @@ -4,26 +4,29 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import aiohttp import mimetypes -from typing import Dict, Any, Optional +from typing import Any, Dict, Optional +import aiohttp from loguru import logger + class GeminiFileAPI: """Client for the Gemini File API. - + This class provides methods for uploading, fetching, listing, and deleting files through Google's Gemini File API. - + Files uploaded through this API remain available for 48 hours and can be referenced in calls to the Gemini generative models. Maximum file size is 2GB, with total project storage limited to 20GB. """ - - def __init__(self, api_key: str, base_url: str = "https://generativelanguage.googleapis.com/v1beta/files"): + + def __init__( + self, api_key: str, base_url: str = "https://generativelanguage.googleapis.com/v1beta/files" + ): """Initialize the Gemini File API client. - + Args: api_key: Google AI API key base_url: Base URL for the Gemini File API (default is the v1beta endpoint) @@ -32,160 +35,148 @@ class GeminiFileAPI: self.base_url = base_url # Upload URL uses the /upload/ path self.upload_base_url = "https://generativelanguage.googleapis.com/upload/v1beta/files" - - async def upload_file(self, file_path: str, display_name: Optional[str] = None) -> Dict[str, Any]: + + async def upload_file( + self, file_path: str, display_name: Optional[str] = None + ) -> Dict[str, Any]: """Upload a file to the Gemini File API using the correct resumable upload protocol. - + Args: file_path: Path to the file to upload display_name: Optional display name for the file - + Returns: File metadata including uri, name, and display_name """ logger.info(f"Uploading file: {file_path}") - + async with aiohttp.ClientSession() as session: # Determine the file's MIME type mime_type, _ = mimetypes.guess_type(file_path) if not mime_type: mime_type = "application/octet-stream" - + # Read the file with open(file_path, "rb") as f: file_data = f.read() - + # Create the metadata payload metadata = {} if display_name: metadata = {"file": {"display_name": display_name}} - + # Step 1: Initial resumable request to get upload URL headers = { "X-Goog-Upload-Protocol": "resumable", "X-Goog-Upload-Command": "start", "X-Goog-Upload-Header-Content-Length": str(len(file_data)), "X-Goog-Upload-Header-Content-Type": mime_type, - "Content-Type": "application/json" + "Content-Type": "application/json", } - + logger.debug(f"Step 1: Getting upload URL from {self.upload_base_url}") async with session.post( - f"{self.upload_base_url}?key={self.api_key}", - headers=headers, - json=metadata + f"{self.upload_base_url}?key={self.api_key}", headers=headers, json=metadata ) as response: if response.status != 200: error_text = await response.text() logger.error(f"Error initiating file upload: {error_text}") raise Exception(f"Failed to initiate upload: {response.status} - {error_text}") - + # Get the upload URL from the response header upload_url = response.headers.get("X-Goog-Upload-URL") if not upload_url: logger.error(f"Response headers: {dict(response.headers)}") raise Exception("No upload URL in response headers") - + logger.debug(f"Got upload URL: {upload_url}") - + # Step 2: Upload the actual file data upload_headers = { "Content-Length": str(len(file_data)), "X-Goog-Upload-Offset": "0", - "X-Goog-Upload-Command": "upload, finalize" + "X-Goog-Upload-Command": "upload, finalize", } - + logger.debug(f"Step 2: Uploading file data to {upload_url}") - async with session.post( - upload_url, - headers=upload_headers, - data=file_data - ) as response: + async with session.post(upload_url, headers=upload_headers, data=file_data) as response: if response.status != 200: error_text = await response.text() logger.error(f"Error uploading file data: {error_text}") raise Exception(f"Failed to upload file: {response.status} - {error_text}") - + file_info = await response.json() logger.info(f"File uploaded successfully: {file_info.get('file', {}).get('name')}") return file_info - + async def get_file(self, name: str) -> Dict[str, Any]: """Get metadata for a file. - + Args: name: File name (or full path) - + Returns: File metadata """ # Extract just the name part if a full path is provided - if '/' in name: - name = name.split('/')[-1] - + if "/" in name: + name = name.split("/")[-1] + async with aiohttp.ClientSession() as session: - async with session.get( - f"{self.base_url}/{name}?key={self.api_key}" - ) as response: + async with session.get(f"{self.base_url}/{name}?key={self.api_key}") as response: if response.status != 200: error_text = await response.text() logger.error(f"Error getting file metadata: {error_text}") raise Exception(f"Failed to get file metadata: {response.status}") - + file_info = await response.json() return file_info - - async def list_files(self, page_size: int = 10, page_token: Optional[str] = None) -> Dict[str, Any]: + + async def list_files( + self, page_size: int = 10, page_token: Optional[str] = None + ) -> Dict[str, Any]: """List uploaded files. - + Args: page_size: Number of files to return per page page_token: Token for pagination - + Returns: List of files and next page token if available """ - params = { - "key": self.api_key, - "pageSize": page_size - } - + params = {"key": self.api_key, "pageSize": page_size} + if page_token: params["pageToken"] = page_token - + async with aiohttp.ClientSession() as session: - async with session.get( - self.base_url, - params=params - ) as response: + async with session.get(self.base_url, params=params) as response: if response.status != 200: error_text = await response.text() logger.error(f"Error listing files: {error_text}") raise Exception(f"Failed to list files: {response.status}") - + result = await response.json() return result - + async def delete_file(self, name: str) -> bool: """Delete a file. - + Args: name: File name (or full path) - + Returns: True if deleted successfully """ # Extract just the name part if a full path is provided - if '/' in name: - name = name.split('/')[-1] - + if "/" in name: + name = name.split("/")[-1] + async with aiohttp.ClientSession() as session: - async with session.delete( - f"{self.base_url}/{name}?key={self.api_key}" - ) as response: + async with session.delete(f"{self.base_url}/{name}?key={self.api_key}") as response: if response.status != 200: error_text = await response.text() logger.error(f"Error deleting file: {error_text}") raise Exception(f"Failed to delete file: {response.status}") - - return True + + return True diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 29d17e689..4d0090a41 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -59,11 +59,8 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult -from pipecat.services.llm_service import LLMService from pipecat.services.llm_service import FunctionCallFromLLM, LLMService - from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, OpenAIUserContextAggregator, @@ -75,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 .audio_transcriber import AudioTranscriber from .file_api import GeminiFileAPI try: @@ -226,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 @@ -238,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. @@ -273,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: @@ -468,7 +468,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter - + def __init__( self, *, @@ -560,7 +560,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) @@ -1015,12 +1015,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: @@ -1167,7 +1169,9 @@ class GeminiMultimodalLiveLLMService(LLMService): # Process grounding metadata if we have accumulated any if self._accumulated_grounding_metadata: logger.debug("Processing grounding metadata...") - await self._process_grounding_metadata(self._accumulated_grounding_metadata, self._search_result_buffer) + await self._process_grounding_metadata( + self._accumulated_grounding_metadata, self._search_result_buffer + ) else: logger.debug("No grounding metadata to process") @@ -1285,17 +1289,23 @@ class GeminiMultimodalLiveLLMService(LLMService): async def _handle_evt_grounding_metadata(self, evt): """Handle dedicated grounding metadata events.""" logger.debug("Received dedicated grounding metadata event.") - + if evt.serverContent and evt.serverContent.groundingMetadata: grounding_metadata = evt.serverContent.groundingMetadata - logger.debug(f"Grounding data: {len(grounding_metadata.groundingChunks or [])} chunks, {len(grounding_metadata.groundingSupports or [])} supports") - + logger.debug( + f"Grounding data: {len(grounding_metadata.groundingChunks or [])} chunks, {len(grounding_metadata.groundingSupports or [])} supports" + ) + # Process the grounding metadata immediately await self._process_grounding_metadata(grounding_metadata, self._search_result_buffer) - async def _process_grounding_metadata(self, grounding_metadata: events.GroundingMetadata, search_result: str = ""): + async def _process_grounding_metadata( + self, grounding_metadata: events.GroundingMetadata, search_result: str = "" + ): """Process grounding metadata and emit LLMSearchResponseFrame.""" - logger.debug(f"Processing grounding metadata. Search result text length: {len(search_result)}") + logger.debug( + f"Processing grounding metadata. Search result text length: {len(search_result)}" + ) if not grounding_metadata: logger.warning("No grounding metadata provided to _process_grounding_metadata") return @@ -1304,49 +1314,47 @@ class GeminiMultimodalLiveLLMService(LLMService): # Extract rendered content for search suggestions rendered_content = None - if grounding_metadata.searchEntryPoint and grounding_metadata.searchEntryPoint.renderedContent: + if ( + grounding_metadata.searchEntryPoint + and grounding_metadata.searchEntryPoint.renderedContent + ): rendered_content = grounding_metadata.searchEntryPoint.renderedContent # Convert grounding chunks and supports to LLMSearchOrigin format origins = [] - + if grounding_metadata.groundingChunks and grounding_metadata.groundingSupports: # Create a mapping of chunk indices to origins chunk_to_origin = {} - + for index, chunk in enumerate(grounding_metadata.groundingChunks): if chunk.web: origin = LLMSearchOrigin( - site_uri=chunk.web.uri, - site_title=chunk.web.title, - results=[] + site_uri=chunk.web.uri, site_title=chunk.web.title, results=[] ) chunk_to_origin[index] = origin origins.append(origin) - + # Add grounding support results to the appropriate origins for support in grounding_metadata.groundingSupports: if support.segment and support.groundingChunkIndices: text = support.segment.text or "" confidence_scores = support.confidenceScores or [] - + # Add this result to all origins referenced by this support for chunk_index in support.groundingChunkIndices: if chunk_index in chunk_to_origin: - result = LLMSearchResult( - text=text, - confidence=confidence_scores - ) + result = LLMSearchResult(text=text, confidence=confidence_scores) chunk_to_origin[chunk_index].results.append(result) # Create and push the search response frame search_frame = LLMSearchResponseFrame( - search_result=search_result, - origins=origins, - rendered_content=rendered_content + search_result=search_result, origins=origins, rendered_content=rendered_content + ) + + logger.debug( + f"Emitting LLMSearchResponseFrame with {len(origins)} origins, rendered_content available: {rendered_content is not None}" ) - - logger.debug(f"Emitting LLMSearchResponseFrame with {len(origins)} origins, rendered_content available: {rendered_content is not None}") await self.push_frame(search_frame) def create_context_aggregator( From f1c9f5040b7808db2f2194621ce2c7b0fd28ccd5 Mon Sep 17 00:00:00 2001 From: vipyne Date: Tue, 1 Jul 2025 16:26:52 -0500 Subject: [PATCH 14/18] Update examples/foundational/26f-gemini-multimodal-live-files-api.py --- .../26f-gemini-multimodal-live-files-api.py | 158 +++++++++++------- 1 file changed, 95 insertions(+), 63 deletions(-) diff --git a/examples/foundational/26f-gemini-multimodal-live-files-api.py b/examples/foundational/26f-gemini-multimodal-live-files-api.py index 413830cf9..160cd5e1b 100644 --- a/examples/foundational/26f-gemini-multimodal-live-files-api.py +++ b/examples/foundational/26f-gemini-multimodal-live-files-api.py @@ -18,67 +18,87 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.gemini_multimodal_live.gemini import ( - GeminiMultimodalLiveLLMService, GeminiMultimodalLiveContext, + GeminiMultimodalLiveLLMService, ) -from pipecat.transports.base_transport import TransportParams -from pipecat.transports.network.small_webrtc import SmallWebRTCTransport -from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.services.daily import DailyParams load_dotenv(override=True) +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=False, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=False, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_in_enabled=False, + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), +} + + +sample_file_path = "" + + async def create_sample_file(): - """Create a sample text file for testing the File API.""" - content = """# Sample Document for Gemini File API Test + if sample_file_path: + return sample_file_path + else: + """Create a sample text file for testing the File API.""" + content = """# Sample Document for Gemini File API Test -This is a test document to demonstrate the Gemini File API functionality. + This is a test document to demonstrate the Gemini File API functionality. -## Key Information: -- This document was created for testing purposes -- It contains information about AI assistants -- The document should be analyzed by Gemini -- The secret phrase for the test is "Pineapple Pizza" + ## Key Information: + - This document was created for testing purposes + - It contains information about AI assistants + - The document should be analyzed by Gemini + - The secret phrase for the test is "Pineapple Pizza" -## AI Assistant Capabilities: -1. Natural language processing -2. File analysis and understanding -3. Context-aware conversations -4. Multi-modal interactions + ## AI Assistant Capabilities: + 1. Natural language processing + 2. File analysis and understanding + 3. Context-aware conversations + 4. Multi-modal interactions -## Conclusion: -This document serves as a test case for the Gemini File API integration with Pipecat. -The AI should be able to reference and discuss the contents of this file. -""" - - # Create a temporary file - with tempfile.NamedTemporaryFile(mode='w', suffix='.txt', delete=False) as f: - f.write(content) - return f.name + ## Conclusion: + This document serves as a test case for the Gemini File API integration with Pipecat. + The AI should be able to reference and discuss the contents of this file. + """ + + # Create a temporary file + with tempfile.NamedTemporaryFile(mode="w", suffix=".txt", delete=False) as f: + f.write(content) + return f.name -async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): +async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_sigint: bool): logger.info(f"Starting File API bot") # Create a sample file to upload sample_file_path = await create_sample_file() logger.info(f"Created sample file: {sample_file_path}") - # Initialize the SmallWebRTCTransport with the connection - transport = SmallWebRTCTransport( - webrtc_connection=webrtc_connection, - params=TransportParams( - audio_in_enabled=True, - audio_out_enabled=True, - video_in_enabled=False, - vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), - ), - ) - system_instruction = """ You are a helpful AI assistant with access to a document that has been uploaded for analysis. - The document contains test information including a secret phrase. You should be able to: + The document contains test information. + You should be able to: - Reference and discuss the contents of the uploaded document - Answer questions about what's in the document - Use the information from the document in our conversation @@ -100,15 +120,14 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac file_info = None try: file_info = await llm.file_api.upload_file( - sample_file_path, - display_name="Sample Test Document" + sample_file_path, display_name="Sample Test Document" ) logger.info(f"File uploaded successfully: {file_info['file']['name']}") - + # Get file URI and mime type file_uri = file_info["file"]["uri"] mime_type = "text/plain" - + # Create context with file reference context = OpenAILLMContext( [ @@ -117,22 +136,19 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac "content": [ { "type": "text", - "text": "Greet the user and let them know you have access to a document they can ask you about. Mention that you can discuss its contents." + "text": "Greet the user and let them know you have access to a document they can ask you about. Mention that you can discuss its contents.", }, { "type": "file_data", - "file_data": { - "mime_type": mime_type, - "file_uri": file_uri - } - } - ] + "file_data": {"mime_type": mime_type, "file_uri": file_uri}, + }, + ], } ] ) - + logger.info("File reference added to conversation context") - + except Exception as e: logger.error(f"Error uploading file: {e}") # Continue with a basic context if file upload fails @@ -140,7 +156,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac [ { "role": "user", - "content": "Greet the user and explain that there was an issue with file upload, but you're ready to help with other tasks." + "content": "Greet the user and explain that there was an issue with file upload, but you're ready to help with other tasks.", } ] ) @@ -149,13 +165,15 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac context_aggregator = llm.create_context_aggregator(context) # Build the pipeline - pipeline = Pipeline([ - transport.input(), - context_aggregator.user(), - llm, - transport.output(), - context_aggregator.assistant(), - ]) + pipeline = Pipeline( + [ + transport.input(), + context_aggregator.user(), + llm, + transport.output(), + context_aggregator.assistant(), + ] + ) # Configure the pipeline task task = PipelineTask( @@ -195,7 +213,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac logger.info("Cleaned up uploaded file from Gemini") except Exception as e: logger.error(f"Error cleaning up file: {e}") - + # Remove temporary file try: os.unlink(sample_file_path) @@ -205,6 +223,20 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac if __name__ == "__main__": - from run import main + from pipecat.examples.run import main - main() + upload_example_file = input(""" + + Please pass in a TEXT filepath to test upload. + NOTE: Files are stored on Google's servers for 48 hours. + + Press Enter to use a default test file. + + text filepath : """) + if upload_example_file: + print(f"Uploading file: {upload_example_file}") + sample_file_path = upload_example_file.strip() + else: + print(f"Using default file") + + main(run_example, transport_params=transport_params) From c7cbfe7a4ff6878ae4356e003f43a8d34a6d29b1 Mon Sep 17 00:00:00 2001 From: vipyne Date: Tue, 1 Jul 2025 17:17:13 -0500 Subject: [PATCH 15/18] remove grounding metadata commits --- .../services/gemini_multimodal_live/events.py | 74 +----------- .../gemini_multimodal_live/file_api.py | 14 +-- .../services/gemini_multimodal_live/gemini.py | 113 ------------------ 3 files changed, 11 insertions(+), 190 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/events.py b/src/pipecat/services/gemini_multimodal_live/events.py index 4687519c0..8fea91666 100644 --- a/src/pipecat/services/gemini_multimodal_live/events.py +++ b/src/pipecat/services/gemini_multimodal_live/events.py @@ -12,7 +12,6 @@ import json from enum import Enum from typing import List, Literal, Optional -from loguru import logger from PIL import Image from pydantic import BaseModel, Field @@ -249,55 +248,6 @@ class Config(BaseModel): setup: Setup -# -# Grounding metadata models -# - - -class SearchEntryPoint(BaseModel): - """Represents the search entry point with rendered content for search suggestions.""" - - renderedContent: Optional[str] = None - - -class WebSource(BaseModel): - """Represents a web source from grounding chunks.""" - - uri: Optional[str] = None - title: Optional[str] = None - - -class GroundingChunk(BaseModel): - """Represents a grounding chunk containing web source information.""" - - web: Optional[WebSource] = None - - -class GroundingSegment(BaseModel): - """Represents a segment of text that is grounded.""" - - startIndex: Optional[int] = None - endIndex: Optional[int] = None - text: Optional[str] = None - - -class GroundingSupport(BaseModel): - """Represents support information for grounded text segments.""" - - segment: Optional[GroundingSegment] = None - groundingChunkIndices: Optional[List[int]] = None - confidenceScores: Optional[List[float]] = None - - -class GroundingMetadata(BaseModel): - """Represents grounding metadata from Google Search.""" - - searchEntryPoint: Optional[SearchEntryPoint] = None - groundingChunks: Optional[List[GroundingChunk]] = None - groundingSupports: Optional[List[GroundingSupport]] = None - webSearchQueries: Optional[List[str]] = None - - # # Server events # @@ -389,7 +339,6 @@ class ServerContent(BaseModel): turnComplete: Optional[bool] = None inputTranscription: Optional[BidiGenerateContentTranscription] = None outputTranscription: Optional[BidiGenerateContentTranscription] = None - groundingMetadata: Optional[GroundingMetadata] = None class FunctionCall(BaseModel): @@ -482,7 +431,7 @@ class ServerEvent(BaseModel): usageMetadata: Optional[UsageMetadata] = None -def parse_server_event(message_str): +def parse_server_event(str): """Parse a server event from JSON string. Args: @@ -491,26 +440,11 @@ def parse_server_event(message_str): Returns: ServerEvent instance if parsing succeeds, None otherwise. """ - from loguru import logger # Import logger locally to avoid scoping issues - try: - evt_dict = json.loads(message_str) - - # Only log grounding metadata detection if truly needed for debugging - # In production, this could be removed entirely or moved to TRACE level - if "serverContent" in evt_dict: - server_content = evt_dict["serverContent"] - if "groundingMetadata" in server_content: - # Consider removing this log entirely for production - pass - - evt = ServerEvent.model_validate(evt_dict) - return evt + evt = json.loads(str) + return ServerEvent.model_validate(evt) except Exception as e: - logger.error(f"Error parsing server event: {e}") - # Truncate raw message to avoid logging potentially sensitive or overly long data - truncated_message = message_str[:200] + "..." if len(message_str) > 200 else message_str - logger.error(f"Raw message (truncated): {truncated_message}") + print(f"Error parsing server event: {e}") return None diff --git a/src/pipecat/services/gemini_multimodal_live/file_api.py b/src/pipecat/services/gemini_multimodal_live/file_api.py index 2c79338b5..f0f23ab83 100644 --- a/src/pipecat/services/gemini_multimodal_live/file_api.py +++ b/src/pipecat/services/gemini_multimodal_live/file_api.py @@ -31,8 +31,8 @@ class GeminiFileAPI: api_key: Google AI API key base_url: Base URL for the Gemini File API (default is the v1beta endpoint) """ - self.api_key = api_key - self.base_url = base_url + self._api_key = api_key + self._base_url = base_url # Upload URL uses the /upload/ path self.upload_base_url = "https://generativelanguage.googleapis.com/upload/v1beta/files" @@ -76,7 +76,7 @@ class GeminiFileAPI: logger.debug(f"Step 1: Getting upload URL from {self.upload_base_url}") async with session.post( - f"{self.upload_base_url}?key={self.api_key}", headers=headers, json=metadata + f"{self.upload_base_url}?key={self._api_key}", headers=headers, json=metadata ) as response: if response.status != 200: error_text = await response.text() @@ -123,7 +123,7 @@ class GeminiFileAPI: name = name.split("/")[-1] async with aiohttp.ClientSession() as session: - async with session.get(f"{self.base_url}/{name}?key={self.api_key}") as response: + async with session.get(f"{self._base_url}/{name}?key={self._api_key}") as response: if response.status != 200: error_text = await response.text() logger.error(f"Error getting file metadata: {error_text}") @@ -144,13 +144,13 @@ class GeminiFileAPI: Returns: List of files and next page token if available """ - params = {"key": self.api_key, "pageSize": page_size} + params = {"key": self._api_key, "pageSize": page_size} if page_token: params["pageToken"] = page_token async with aiohttp.ClientSession() as session: - async with session.get(self.base_url, params=params) as response: + async with session.get(self._base_url, params=params) as response: if response.status != 200: error_text = await response.text() logger.error(f"Error listing files: {error_text}") @@ -173,7 +173,7 @@ class GeminiFileAPI: name = name.split("/")[-1] async with aiohttp.ClientSession() as session: - async with session.delete(f"{self.base_url}/{name}?key={self.api_key}") as response: + async with session.delete(f"{self._base_url}/{name}?key={self._api_key}") as response: if response.status != 200: error_text = await response.text() logger.error(f"Error deleting file: {error_text}") diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 4d0090a41..9c49fdc21 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -564,10 +564,6 @@ class GeminiMultimodalLiveLLMService(LLMService): # Initialize the File API client self.file_api = GeminiFileAPI(api_key=api_key, base_url=file_api_base_url) - # Grounding metadata tracking - self._search_result_buffer = "" - self._accumulated_grounding_metadata = None - def can_generate_metrics(self) -> bool: """Check if the service can generate usage metrics. @@ -917,23 +913,12 @@ class GeminiMultimodalLiveLLMService(LLMService): await self._handle_evt_input_transcription(evt) elif evt.serverContent and evt.serverContent.outputTranscription: await self._handle_evt_output_transcription(evt) - elif evt.serverContent and evt.serverContent.groundingMetadata: - await self._handle_evt_grounding_metadata(evt) elif evt.toolCall: await self._handle_evt_tool_call(evt) elif False: # !!! todo: error events? await self._handle_evt_error(evt) # errors are fatal, so exit the receive loop return - else: - # Log unhandled events that might contain grounding metadata - logger.warning(f"Received unhandled server event type: {evt}") - pass - - async def _transcribe_audio_handler(self): - while True: - audio = await self._transcribe_audio_queue.get() - await self._handle_transcribe_user_audio(audio, self._context) # # @@ -1092,14 +1077,8 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.push_frame(LLMFullResponseStartFrame()) self._bot_text_buffer += text - self._search_result_buffer += text # Also accumulate for grounding await self.push_frame(LLMTextFrame(text=text)) - # Check for grounding metadata in server content - if evt.serverContent and evt.serverContent.groundingMetadata: - self._accumulated_grounding_metadata = evt.serverContent.groundingMetadata - logger.debug("Grounding metadata detected in model turn.") - inline_data = part.inlineData if not inline_data: return @@ -1166,20 +1145,6 @@ class GeminiMultimodalLiveLLMService(LLMService): self._llm_output_buffer = "" # Only push the TTSStoppedFrame if the bot is outputting audio - # Process grounding metadata if we have accumulated any - if self._accumulated_grounding_metadata: - logger.debug("Processing grounding metadata...") - await self._process_grounding_metadata( - self._accumulated_grounding_metadata, self._search_result_buffer - ) - else: - logger.debug("No grounding metadata to process") - - # Reset grounding tracking for next response - self._search_result_buffer = "" - self._accumulated_grounding_metadata = None - - # Only push the TTSStoppedFrame the bot is outputting audio # when text is found, modalities is set to TEXT and no audio # is produced. if not text: @@ -1257,13 +1222,6 @@ class GeminiMultimodalLiveLLMService(LLMService): # Collect text for tracing self._llm_output_buffer += text - # Accumulate text for grounding as well - self._search_result_buffer += text - - # Check for grounding metadata in server content - if evt.serverContent and evt.serverContent.groundingMetadata: - self._accumulated_grounding_metadata = evt.serverContent.groundingMetadata - await self.push_frame(LLMTextFrame(text=text)) await self.push_frame(TTSTextFrame(text=text)) @@ -1286,77 +1244,6 @@ class GeminiMultimodalLiveLLMService(LLMService): await self.start_llm_usage_metrics(tokens) - async def _handle_evt_grounding_metadata(self, evt): - """Handle dedicated grounding metadata events.""" - logger.debug("Received dedicated grounding metadata event.") - - if evt.serverContent and evt.serverContent.groundingMetadata: - grounding_metadata = evt.serverContent.groundingMetadata - logger.debug( - f"Grounding data: {len(grounding_metadata.groundingChunks or [])} chunks, {len(grounding_metadata.groundingSupports or [])} supports" - ) - - # Process the grounding metadata immediately - await self._process_grounding_metadata(grounding_metadata, self._search_result_buffer) - - async def _process_grounding_metadata( - self, grounding_metadata: events.GroundingMetadata, search_result: str = "" - ): - """Process grounding metadata and emit LLMSearchResponseFrame.""" - logger.debug( - f"Processing grounding metadata. Search result text length: {len(search_result)}" - ) - if not grounding_metadata: - logger.warning("No grounding metadata provided to _process_grounding_metadata") - return - - # logger.debug(f"Processing grounding metadata: {grounding_metadata}") # Too verbose for PR - - # Extract rendered content for search suggestions - rendered_content = None - if ( - grounding_metadata.searchEntryPoint - and grounding_metadata.searchEntryPoint.renderedContent - ): - rendered_content = grounding_metadata.searchEntryPoint.renderedContent - - # Convert grounding chunks and supports to LLMSearchOrigin format - origins = [] - - if grounding_metadata.groundingChunks and grounding_metadata.groundingSupports: - # Create a mapping of chunk indices to origins - chunk_to_origin = {} - - for index, chunk in enumerate(grounding_metadata.groundingChunks): - if chunk.web: - origin = LLMSearchOrigin( - site_uri=chunk.web.uri, site_title=chunk.web.title, results=[] - ) - chunk_to_origin[index] = origin - origins.append(origin) - - # Add grounding support results to the appropriate origins - for support in grounding_metadata.groundingSupports: - if support.segment and support.groundingChunkIndices: - text = support.segment.text or "" - confidence_scores = support.confidenceScores or [] - - # Add this result to all origins referenced by this support - for chunk_index in support.groundingChunkIndices: - if chunk_index in chunk_to_origin: - result = LLMSearchResult(text=text, confidence=confidence_scores) - chunk_to_origin[chunk_index].results.append(result) - - # Create and push the search response frame - search_frame = LLMSearchResponseFrame( - search_result=search_result, origins=origins, rendered_content=rendered_content - ) - - logger.debug( - f"Emitting LLMSearchResponseFrame with {len(origins)} origins, rendered_content available: {rendered_content is not None}" - ) - await self.push_frame(search_frame) - def create_context_aggregator( self, context: OpenAILLMContext, From 0e60385871cd98a3add12911ab58ae2c224f0a60 Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 9 May 2025 10:50:27 -0400 Subject: [PATCH 16/18] add FileAPI to gemini.py --- .../services/gemini_multimodal_live/gemini.py | 94 +++++++++++++++++-- 1 file changed, 87 insertions(+), 7 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 9c49fdc21..5a0f66400 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -72,6 +72,7 @@ from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt from . import events +from .audio_transcriber import AudioTranscriber from .file_api import GeminiFileAPI try: @@ -222,9 +223,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 @@ -234,17 +235,19 @@ 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}") +<<<<<<< HEAD +======= + +>>>>>>> cd4a893c (add FileAPI to gemini.py) def get_messages_for_initializing_history(self): """Get messages formatted for Gemini history initialization. @@ -271,6 +274,7 @@ class GeminiMultimodalLiveContext(OpenAILLMContext): parts.append({"text": part.get("text")}) elif part.get("type") == "file_data": file_data = part.get("file_data", {}) +<<<<<<< HEAD parts.append( { "fileData": { @@ -279,6 +283,14 @@ class GeminiMultimodalLiveContext(OpenAILLMContext): } } ) +======= + parts.append({ + "fileData": { + "mimeType": file_data.get("mime_type"), + "fileUri": file_data.get("file_uri") + } + }) +>>>>>>> cd4a893c (add FileAPI to gemini.py) else: logger.warning(f"Unsupported content type: {str(part)[:80]}") else: @@ -469,6 +481,62 @@ class GeminiMultimodalLiveLLMService(LLMService): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter + """Gemini Live LLM Service with multimodal capabilities including File API support. + + This service implements the Gemini Multimodal Live API with support for: + - Audio input and output + - Image/video input + - File API (upload, reference, and management) + - Tools/function calling + + Example usage of File API: + ```python + # Initialize the service + gemini_service = GeminiMultimodalLiveLLMService(api_key="YOUR_API_KEY") + + # Upload a file from the client + file_path = "/path/to/user_uploaded_file.pdf" + file_info = await gemini_service.file_api.upload_file(file_path) + + # Get file URI and mime type from response + file_uri = file_info["file"]["uri"] + mime_type = "application/pdf" # Set appropriate MIME type + + # When starting a new bot session: + # 1. Initialize the context + context = GeminiMultimodalLiveContext() + + # 2. Add file reference to context BEFORE starting the conversation + context.add_file_reference( + file_uri=file_uri, + mime_type=mime_type, + text="Please analyze this document" + ) + + # 3. Now set the context to start the conversation with file reference included + await gemini_service.set_context(context) + + # Gemini now has access to the file reference in its context window + # The file URI remains valid for 48 hours before Google deletes it + + # Optional: List all files for this user + files = await gemini_service.file_api.list_files() + + # Optional: Get metadata for a specific file + file_metadata = await gemini_service.file_api.get_file(file_info["file"]["name"]) + + # Optional: Delete a file when no longer needed + await gemini_service.file_api.delete_file(file_info["file"]["name"]) + ``` + + Notes: + - Files are stored for 48 hours on Google's servers + - Maximum file size is 2GB + - Total storage per project is 20GB + - File references should be added to the context BEFORE starting the conversation + - The same file reference can be reused for multiple sessions within the 48-hour window + """ + def __init__( self, *, @@ -560,6 +628,9 @@ 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) # Initialize the File API client self.file_api = GeminiFileAPI(api_key=api_key, base_url=file_api_base_url) @@ -1000,6 +1071,7 @@ class GeminiMultimodalLiveLLMService(LLMService): parts.append({"text": part.get("text")}) elif part.get("type") == "file_data": file_data = part.get("file_data", {}) +<<<<<<< HEAD parts.append( { "fileData": { @@ -1008,6 +1080,14 @@ class GeminiMultimodalLiveLLMService(LLMService): } } ) +======= + parts.append({ + "fileData": { + "mimeType": file_data.get("mime_type"), + "fileUri": file_data.get("file_uri") + } + }) +>>>>>>> cd4a893c (add FileAPI to gemini.py) else: logger.warning(f"Unsupported content type: {str(part)[:80]}") else: From 2ee935f784319b9b0c2ef9f14f485a259f5cb3e6 Mon Sep 17 00:00:00 2001 From: getchannel <78183014+getchannel@users.noreply.github.com> Date: Fri, 30 May 2025 12:19:40 -0400 Subject: [PATCH 17/18] Update gemini.py --- .../services/gemini_multimodal_live/gemini.py | 56 ------------------- 1 file changed, 56 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 5a0f66400..9a196765a 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -480,63 +480,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter - - """Gemini Live LLM Service with multimodal capabilities including File API support. - This service implements the Gemini Multimodal Live API with support for: - - Audio input and output - - Image/video input - - File API (upload, reference, and management) - - Tools/function calling - - Example usage of File API: - ```python - # Initialize the service - gemini_service = GeminiMultimodalLiveLLMService(api_key="YOUR_API_KEY") - - # Upload a file from the client - file_path = "/path/to/user_uploaded_file.pdf" - file_info = await gemini_service.file_api.upload_file(file_path) - - # Get file URI and mime type from response - file_uri = file_info["file"]["uri"] - mime_type = "application/pdf" # Set appropriate MIME type - - # When starting a new bot session: - # 1. Initialize the context - context = GeminiMultimodalLiveContext() - - # 2. Add file reference to context BEFORE starting the conversation - context.add_file_reference( - file_uri=file_uri, - mime_type=mime_type, - text="Please analyze this document" - ) - - # 3. Now set the context to start the conversation with file reference included - await gemini_service.set_context(context) - - # Gemini now has access to the file reference in its context window - # The file URI remains valid for 48 hours before Google deletes it - - # Optional: List all files for this user - files = await gemini_service.file_api.list_files() - - # Optional: Get metadata for a specific file - file_metadata = await gemini_service.file_api.get_file(file_info["file"]["name"]) - - # Optional: Delete a file when no longer needed - await gemini_service.file_api.delete_file(file_info["file"]["name"]) - ``` - - Notes: - - Files are stored for 48 hours on Google's servers - - Maximum file size is 2GB - - Total storage per project is 20GB - - File references should be added to the context BEFORE starting the conversation - - The same file reference can be reused for multiple sessions within the 48-hour window - """ - def __init__( self, *, From 9c9d4b35a433f58f1f109febc3a1be349db2736d Mon Sep 17 00:00:00 2001 From: Pete <78183014+getchannel@users.noreply.github.com> Date: Tue, 10 Jun 2025 11:22:16 -0400 Subject: [PATCH 18/18] remove audio_transcriber from gemini.py unecessary import removed. --- .../services/gemini_multimodal_live/gemini.py | 28 +------------------ 1 file changed, 1 insertion(+), 27 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 9a196765a..19860387e 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -72,7 +72,7 @@ from pipecat.utils.time import time_now_iso8601 from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_stt from . import events -from .audio_transcriber import AudioTranscriber + from .file_api import GeminiFileAPI try: @@ -243,11 +243,7 @@ class GeminiMultimodalLiveContext(OpenAILLMContext): message = {"role": "user", "content": parts} self.messages.append(message) logger.info(f"Added file reference to context: {file_uri}") -<<<<<<< HEAD - -======= ->>>>>>> cd4a893c (add FileAPI to gemini.py) def get_messages_for_initializing_history(self): """Get messages formatted for Gemini history initialization. @@ -274,23 +270,12 @@ class GeminiMultimodalLiveContext(OpenAILLMContext): parts.append({"text": part.get("text")}) elif part.get("type") == "file_data": file_data = part.get("file_data", {}) -<<<<<<< HEAD - 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") } }) ->>>>>>> cd4a893c (add FileAPI to gemini.py) else: logger.warning(f"Unsupported content type: {str(part)[:80]}") else: @@ -1015,23 +1000,12 @@ class GeminiMultimodalLiveLLMService(LLMService): parts.append({"text": part.get("text")}) elif part.get("type") == "file_data": file_data = part.get("file_data", {}) -<<<<<<< HEAD - 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") } }) ->>>>>>> cd4a893c (add FileAPI to gemini.py) else: logger.warning(f"Unsupported content type: {str(part)[:80]}") else: