Merge pull request #1786 from getchannel/main

Add File API to GeminiMultimodalLive

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)

src/pipecat/services/gemini_multimodal_live/__init__.py

@@ -1 +1,2 @@
+from .file_api import GeminiFileAPI
 from .gemini import GeminiMultimodalLiveLLMService

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):

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

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: