diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 8df97d313..b415f9c11 100644 --- a/src/pipecat/frames/frames.py +++ b/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 diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index db28b85c1..871b1cbca 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/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) diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 712624db8..c1af46745 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/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( - "[pk] Context update does not require reconnect. Sending any newly-completed function call results..." - ) + 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: @@ -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 - # 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 or diff.tools_diff.has_changes(): 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: