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..160cd5e1b --- /dev/null +++ b/examples/foundational/26f-gemini-multimodal-live-files-api.py @@ -0,0 +1,242 @@ +# +# 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 ( + GeminiMultimodalLiveContext, + GeminiMultimodalLiveLLMService, +) +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(): + 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. + + ## 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_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}") + + system_instruction = """ + You are a helpful AI assistant with access to a document that has been uploaded for analysis. + + 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 + + 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 pipecat.examples.run import 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) diff --git a/src/pipecat/services/gemini_multimodal_live/__init__.py b/src/pipecat/services/gemini_multimodal_live/__init__.py index 61bdf58dd..513d9fd66 100644 --- a/src/pipecat/services/gemini_multimodal_live/__init__.py +++ b/src/pipecat/services/gemini_multimodal_live/__init__.py @@ -1 +1,2 @@ +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 160ff1174..8fea91666 100644 --- a/src/pipecat/services/gemini_multimodal_live/events.py +++ b/src/pipecat/services/gemini_multimodal_live/events.py @@ -44,6 +44,17 @@ 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): diff --git a/src/pipecat/services/gemini_multimodal_live/file_api.py b/src/pipecat/services/gemini_multimodal_live/file_api.py new file mode 100644 index 000000000..f0f23ab83 --- /dev/null +++ b/src/pipecat/services/gemini_multimodal_live/file_api.py @@ -0,0 +1,182 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import mimetypes +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" + ): + """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 + # 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 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", + } + + 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 + ) 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", + } + + 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: + 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] + + 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 diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 3c5ed92dc..19860387e 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -59,6 +59,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.google.frames import LLMSearchOrigin, LLMSearchResponseFrame, LLMSearchResult from pipecat.services.llm_service import FunctionCallFromLLM, LLMService from pipecat.services.openai.llm import ( OpenAIAssistantContextAggregator, @@ -72,6 +73,8 @@ from pipecat.utils.tracing.service_decorators import traced_gemini_live, traced_ from . import events +from .file_api import GeminiFileAPI + try: import websockets except ModuleNotFoundError as e: @@ -218,6 +221,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 +268,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: @@ -431,7 +465,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter - + def __init__( self, *, @@ -445,6 +479,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 +557,12 @@ 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) def can_generate_metrics(self) -> bool: """Check if the service can generate usage metrics. @@ -938,7 +979,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 +998,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: