diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index c1af46745..5c2871fdc 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -1040,7 +1040,7 @@ class GeminiLiveLLMService(LLMService): context ) # Take a static snapshot guaranteed not to change - if self._context_update_requires_reconnect(diff): + if self._context_update_requires_reconnect(diff, messages_programmatically_edited): # Reconnect logger.debug("Context update requires reconnect. Reconnecting...") @@ -1057,41 +1057,46 @@ class GeminiLiveLLMService(LLMService): # Trigger "initial" response with new connection await self._create_initial_response() else: - # If messages were appended, they might be: - # - Messages generated by Gemini Live itself, which the remote - # service and its internal context management is already - # aware of. We can ignore these. - # - Messages appended programmatically by the user (e.g. via - # an LLMMessagesAppendFrame). We need to send these to the - # remote Gemini Live service. - # - Function call results, which are handled separately below. - if diff.messages_appended and messages_programmatically_edited: - # Messages were programmatically edited - send to the API - logger.debug( - "Context update includes programmatically-appended messages. Creating incremental response..." - ) - await self._create_incremental_response(diff.messages_appended) - # Send results for newly-completed function calls, if any. logger.debug("Checking for newly-completed function call results...") await self._process_completed_function_calls(send_new_results=True) - def _context_update_requires_reconnect(self, diff: LLMContextDiff) -> bool: + def _context_update_requires_reconnect( + self, diff: LLMContextDiff, messages_programmatically_edited: bool + ) -> bool: """Check if an update to our LLM context requires reconnection. Args: diff: The context diff representing the update + messages_programmatically_edited: Whether context messages were + programmatically edited (e.g. with LLMMessagesAppendFrame) + Returns: True if reconnection is required, False otherwise. """ - if diff.history_edited or diff.tools_diff.has_changes(): + # We need to reconnect in 3 cases: + # 1. If the conversation history was edited + # 2. If the tools available to the model were changed + # 3. If messages were appended programmatically by the user, e.g. with + # LLMMessagesAppendFrame (NOT if they originated from Gemini Live + # itself, since the service is already aware of those) + # + # Note that *ideally* in this 3rd case we would just send the newly + # appended messages without reconnecting, but in all my testing so far, + # I haven't been able to get that to work reliably. + # TODO: add more detail about what's been attempted. + if ( + diff.history_edited + or diff.tools_diff.has_changes() + or (diff.messages_appended and messages_programmatically_edited) + ): if diff.history_edited: print("[pk] Context diff: history edited.") if diff.tools_diff.has_changes(): print("[pk] Context diff: tools changed.") + if diff.messages_appended and messages_programmatically_edited: + print(f"[pk] Context diff: messages appended: {diff.messages_appended}") return True - elif diff.messages_appended: - print(f"[pk] Context diff: messages appended: {diff.messages_appended}") return False @@ -1481,32 +1486,6 @@ class GeminiLiveLLMService(LLMService): if not self._inference_on_context_initialization: self._needs_turn_complete_message = True - async def _create_incremental_response(self, new_messages: List[LLMContextMessage]): - """Create a new response mid-conversation with newly-appended messages. - - Args: - new_messages: The new messages to send. - """ - if self._disconnecting or not self._session or not new_messages: - return - - # Create a throwaway context just for the purpose of getting messages - adapter: GeminiLLMAdapter = self.get_llm_adapter() - messages = adapter.get_llm_invocation_params( - LLMContext(messages=new_messages), strip_function_messages=True - ).get("messages", []) - if not messages: - return - - logger.debug(f"Creating response with new messages: {messages}") - - await self.start_ttfb_metrics() - - try: - await self._session.send_client_content(turns=messages, turn_complete=True) - except Exception as e: - await self._handle_send_error(e) - async def _create_single_response(self, messages_list): """Create a single response from a list of messages.""" if self._disconnecting or not self._session: