Another change to support Gemini Live in Pipecat Flows: if the only change to the context is newly-appended messages, send them to the server.

This requires us to distinguish between newly-appended "bookkeeping" messages that just reflect what Gemini Live already said, and messages that were programmatically inserted, such as from the the transition to the new Pipecat Flows node.

This change makes it so that using `LLMMessagesAppendFrame` will have the desired effect, of updating the Gemini Live conversation.

src/pipecat/frames/frames.py

@@ -667,9 +667,16 @@ class LLMContextFrame(Frame):
 
     Parameters:
         context: The LLM context containing messages, tools, and configuration.
+        messages_programmatically_edited: Whether the context messages were
+            programmatically edited (e.g. via LLMMessagesAppendFrame or
+            LLMMessagesUpdateFrame) since the last context frame was pushed.
+            This is used by speech-to-speech LLM services (like Gemini Live) to
+            distinguish between messages that originated from the LLM output
+            itself vs. messages that were externally injected.
     """
 
     context: "LLMContext"
+    messages_programmatically_edited: bool = False
 
 
 @dataclass

src/pipecat/processors/aggregators/llm_response_universal.py

@@ -231,7 +231,16 @@ class LLMContextAggregator(FrameProcessor):
         Returns:
             LLMContextFrame containing the current context.
         """
-        return LLMContextFrame(context=self._context)
+        # Check if messages were programmatically edited since the last push.
+        # This flag is stored as a runtime attribute on the shared context
+        # object so that both user and assistant aggregators can see it.
+        messages_programmatically_edited = getattr(
+            self._context, "_pipecat_messages_programmatically_edited", False
+        )
+        return LLMContextFrame(
+            context=self._context,
+            messages_programmatically_edited=messages_programmatically_edited,
+        )
 
     async def push_context_frame(self, direction: FrameDirection = FrameDirection.DOWNSTREAM):
         """Push a context frame in the specified direction.
@@ -241,6 +250,9 @@ class LLMContextAggregator(FrameProcessor):
         """
         frame = self._get_context_frame()
         await self.push_frame(frame, direction)
+        # Clear the programmatic edit flag after pushing, since the context
+        # frame now carries this information to downstream processors.
+        self._context._pipecat_messages_programmatically_edited = False
 
     def add_messages(self, messages):
         """Add messages to the context.
@@ -573,11 +585,19 @@ class LLMUserAggregator(LLMContextAggregator):
 
     async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
         self.add_messages(frame.messages)
+        # Mark the context as programmatically edited. This flag is stored as a
+        # runtime attribute on the shared context object so that both user and
+        # assistant aggregators can see it.
+        self._context._pipecat_messages_programmatically_edited = True
         if frame.run_llm:
             await self.push_context_frame()
 
     async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
         self.set_messages(frame.messages)
+        # Mark the context as programmatically edited. This flag is stored as a
+        # runtime attribute on the shared context object so that both user and
+        # assistant aggregators can see it.
+        self._context._pipecat_messages_programmatically_edited = True
         if frame.run_llm:
             await self.push_context_frame()
 
@@ -909,11 +929,19 @@ class LLMAssistantAggregator(LLMContextAggregator):
 
     async def _handle_llm_messages_append(self, frame: LLMMessagesAppendFrame):
         self.add_messages(frame.messages)
+        # Mark the context as programmatically edited. This flag is stored as a
+        # runtime attribute on the shared context object so that both user and
+        # assistant aggregators can see it.
+        self._context._pipecat_messages_programmatically_edited = True
         if frame.run_llm:
             await self.push_context_frame(FrameDirection.UPSTREAM)
 
     async def _handle_llm_messages_update(self, frame: LLMMessagesUpdateFrame):
         self.set_messages(frame.messages)
+        # Mark the context as programmatically edited. This flag is stored as
+        # a runtime attribute on the shared context object so that both user
+        # and assistant aggregators can see it.
+        self._context._pipecat_messages_programmatically_edited = True
         if frame.run_llm:
             await self.push_context_frame(FrameDirection.UPSTREAM)
 

src/pipecat/services/google/gemini_live/llm.py

@@ -60,7 +60,12 @@ from pipecat.frames.frames import (
     UserStoppedSpeakingFrame,
 )
 from pipecat.metrics.metrics import LLMTokenUsage
-from pipecat.processors.aggregators.llm_context import NOT_GIVEN, LLMContext, LLMContextDiff
+from pipecat.processors.aggregators.llm_context import (
+    NOT_GIVEN,
+    LLMContext,
+    LLMContextDiff,
+    LLMContextMessage,
+)
 from pipecat.processors.aggregators.llm_response import (
     LLMAssistantAggregatorParams,
     LLMUserAggregatorParams,
@@ -920,7 +925,12 @@ class GeminiLiveLLMService(LLMService):
                 if isinstance(frame, LLMContextFrame)
                 else LLMContext.from_openai_context(frame.context)
             )
-            await self._handle_context(context)
+            messages_programmatically_edited = (
+                frame.messages_programmatically_edited
+                if isinstance(frame, LLMContextFrame)
+                else False
+            )
+            await self._handle_context(context, messages_programmatically_edited)
         elif isinstance(frame, InputTextRawFrame):
             await self._send_user_text(frame.text)
             await self.push_frame(frame, direction)
@@ -965,7 +975,7 @@ class GeminiLiveLLMService(LLMService):
         else:
             await self.push_frame(frame, direction)
 
-    async def _handle_context(self, context: LLMContext):
+    async def _handle_context(self, context: LLMContext, messages_programmatically_edited: bool):
         if not self._context:
             # We got our initial context
             self._context = context
@@ -1032,13 +1042,11 @@ class GeminiLiveLLMService(LLMService):
 
             if self._context_update_requires_reconnect(diff):
                 # Reconnect
-                print("[pk] Context update requires reconnect. Reconnecting...")
+                logger.debug("Context update requires reconnect. Reconnecting...")
 
                 # TODO: necessary?
                 self._session_resumption_handle = None
 
-                # TODO: do something special here to handle the context like it's the initial one again?
-
                 # Reconnect
                 await self._reconnect()
 
@@ -1049,10 +1057,23 @@ 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.
-                print(
+                logger.debug("Checking for newly-completed function call results...")
-                    "[pk] Context update does not require reconnect. Sending any newly-completed function call results..."
-                )
                 await self._process_completed_function_calls(send_new_results=True)
 
     def _context_update_requires_reconnect(self, diff: LLMContextDiff) -> bool:
@@ -1063,26 +1084,16 @@ class GeminiLiveLLMService(LLMService):
         Returns:
             True if reconnection is required, False otherwise.
         """
-        # During the course of a "normal" conversation with no external context
+        if diff.history_edited or diff.tools_diff.has_changes():
-        # manipulation (like adding tools, changing system instructions,
-        # rewriting history), context updates will contain either:
-        # - new messages (that the Gemini Live service, with its own internal
-        #   context management, is already aware of), or
-        # - tool call results (that we need to tell the remote service
-        #   about).
-        # Any other changes to the context require reconnection.
-        # TODO: the below
-        # Note that it's possible that the developer is trying to
-        # programmatically append messages, in which case we'd miss that
-        # update...maybe we can check the number of appended messages...
-        if diff.history_edited or diff.tools_diff.has_changes() or diff.tool_choice_changed:
             if diff.history_edited:
                 print("[pk] Context diff: history edited.")
             if diff.tools_diff.has_changes():
                 print("[pk] Context diff: tools changed.")
-            if diff.tool_choice_changed:
-                print("[pk] Context diff: tool choice changed.")
             return True
+        elif diff.messages_appended:
+            print(f"[pk] Context diff: messages appended: {diff.messages_appended}")
+
+        return False
 
     async def _process_completed_function_calls(self, send_new_results: bool):
         # Check for set of completed function calls in the context
@@ -1103,6 +1114,9 @@ class GeminiLiveLLMService(LLMService):
                         ):
                             # Found a newly-completed function call - send the result to the service
                             if send_new_results:
+                                logger.debug(
+                                    f"Sending newly-completed tool call result for tool '{tool_name}'"
+                                )
                                 await self._tool_result(
                                     tool_call_id, tool_name, part.function_response.response
                                 )
@@ -1467,6 +1481,32 @@ 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: