Another change to support Gemini Live in Pipecat Flows: now even in the case of programmatically-appended context messages, do a full reconnection—I've tried many things to get "live updates" to context working reliably and have been unable to (Gemini code comments also warn against doing "live updates" after the initial context seeding prior to starting audio input).

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: