From 79e51051c7cfa16344bd456c995c1baa53d22186 Mon Sep 17 00:00:00 2001 From: vipyne Date: Tue, 1 Jul 2025 16:25:59 -0500 Subject: [PATCH] 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(