diff --git a/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py b/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py index e26fbed51..b96e3ab26 100644 --- a/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py +++ b/examples/foundational/26g-gemini-multimodal-live-groundingMetadata.py @@ -162,4 +162,4 @@ async def run_example(transport: BaseTransport, _: argparse.Namespace, handle_si if __name__ == "__main__": from pipecat.examples.run import main - main(run_example, transport_params=transport_params) + main(run_example, transport_params=transport_params) diff --git a/src/pipecat/services/gemini_multimodal_live/events.py b/src/pipecat/services/gemini_multimodal_live/events.py index 63efdc31a..ddac795e7 100644 --- a/src/pipecat/services/gemini_multimodal_live/events.py +++ b/src/pipecat/services/gemini_multimodal_live/events.py @@ -256,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 @@ -279,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 @@ -286,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 @@ -476,8 +482,6 @@ class ServerEvent(BaseModel): usageMetadata: Optional[UsageMetadata] = None - - def parse_server_event(str): """Parse a server event from JSON string. diff --git a/src/pipecat/services/gemini_multimodal_live/file_api.py b/src/pipecat/services/gemini_multimodal_live/file_api.py index 67871a714..5ae7fdbb7 100644 --- a/src/pipecat/services/gemini_multimodal_live/file_api.py +++ b/src/pipecat/services/gemini_multimodal_live/file_api.py @@ -186,4 +186,4 @@ class GeminiFileAPI: logger.error(f"Error deleting file: {error_text}") raise Exception(f"Failed to delete file: {response.status}") - return True \ No newline at end of file + return True diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index f6cc51a5f..89fe3b257 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -227,9 +227,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 @@ -482,7 +482,7 @@ class GeminiMultimodalLiveLLMService(LLMService): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter - + def __init__( self, *, @@ -1024,10 +1024,9 @@ class GeminiMultimodalLiveLLMService(LLMService): self._needs_turn_complete_message = True async def _create_single_response(self, messages_list): - """Create a single response from a list of messages.""" - - # Refactor to combine this logic with same logic in GeminiMultimodalLiveContext + + # Refactor to combine this logic with same logic in GeminiMultimodalLiveContext messages = [] for item in messages_list: role = item.get("role") @@ -1206,7 +1205,9 @@ class GeminiMultimodalLiveLLMService(LLMService): # Process grounding metadata if we have accumulated any if self._accumulated_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 + ) # Reset grounding tracking for next response self._search_result_buffer = "" @@ -1305,29 +1306,32 @@ class GeminiMultimodalLiveLLMService(LLMService): # 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.""" if not grounding_metadata: return # 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) @@ -1341,20 +1345,16 @@ class GeminiMultimodalLiveLLMService(LLMService): # 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 ) - + await self.push_frame(search_frame) + async def _handle_evt_usage_metadata(self, evt): """Handle the usage metadata event.""" if not evt.usageMetadata: