Merge pull request #1696 from pipecat-ai/mb/fix-gemini-live-context

Fix: GeminiMultimodalLiveLLMService was appending tokens to the context

CHANGELOG.md

@@ -87,6 +87,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue with `GeminiMultimodalLiveLLMService` where the context
+  contained tokens instead of words.
+
 - Fixed an issue with HTTP Smart Turn handling, where the service returns a 500
   error. Previously, this would cause an unhandled exception. Now, a 500 error
   is treated as an incomplete response.

examples/foundational/26b-gemini-multimodal-live-function-calling.py

@@ -89,6 +89,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
     llm = GeminiMultimodalLiveLLMService(
         api_key=os.getenv("GOOGLE_API_KEY"),
         system_instruction=system_instruction,
+        transcribe_user_audio=True,
         tools=tools,
     )
 

src/pipecat/processors/transcript_processor.py

@@ -93,49 +93,55 @@ class AssistantTranscriptProcessor(BaseTranscriptProcessor):
         """Aggregates and emits text fragments as a transcript message.
 
         This method uses a heuristic to automatically detect whether text fragments
-        use pre-spacing (spaces at the beginning of fragments) or not, and applies
+        contain embedded spacing (spaces at the beginning or end of fragments) or not,
-        the appropriate joining strategy. It handles fragments from different TTS
+        and applies the appropriate joining strategy. It handles fragments from different
-        services with different formatting patterns.
+        TTS services with different formatting patterns.
 
         Examples:
-            Pre-spaced fragments (concatenated):
+            Fragments with embedded spacing (concatenated):
                 ```
                 TTSTextFrame: ["Hello"]
-                TTSTextFrame: [" there"]
+                TTSTextFrame: [" there"]  # Leading space
                 TTSTextFrame: ["!"]
-                TTSTextFrame: [" How"]
+                TTSTextFrame: [" How"]    # Leading space
                 TTSTextFrame: ["'s"]
-                TTSTextFrame: [" it"]
+                TTSTextFrame: [" it"]     # Leading space
-                TTSTextFrame: [" going"]
-                TTSTextFrame: ["?"]
                 ```
-                Result: "Hello there! How's it going?"
+                Result: "Hello there! How's it"
 
-            Word-by-word fragments (joined with spaces):
+            Fragments with trailing spaces (concatenated):
+                ```
+                TTSTextFrame: ["Hel"]
+                TTSTextFrame: ["lo "]     # Trailing space
+                TTSTextFrame: ["to "]     # Trailing space
+                TTSTextFrame: ["you"]
+                ```
+                Result: "Hello to you"
+
+            Word-by-word fragments without spacing (joined with spaces):
                 ```
                 TTSTextFrame: ["Hello"]
-                TTSTextFrame: ["there!"]
+                TTSTextFrame: ["there"]
-                TTSTextFrame: ["How"]
+                TTSTextFrame: ["how"]
-                TTSTextFrame: ["is"]
+                TTSTextFrame: ["are"]
-                TTSTextFrame: ["it"]
+                TTSTextFrame: ["you"]
-                TTSTextFrame: ["going?"]
                 ```
-                Result: "Hello there! How is it going?"
+                Result: "Hello there how are you"
         """
         if self._current_text_parts and self._aggregation_start_time:
-            # Heuristic to detect pre-spaced fragments
+            has_leading_spaces = any(
-            uses_prespacing = False
+                part and part[0].isspace() for part in self._current_text_parts[1:]
-            if len(self._current_text_parts) > 1:
+            )
-                # Check if any fragment after the first one starts with whitespace
+            has_trailing_spaces = any(
-                has_spaced_parts = any(
+                part and part[-1].isspace() for part in self._current_text_parts[:-1]
-                    part and part[0].isspace() for part in self._current_text_parts[1:]
+            )
-                )
-                if has_spaced_parts:
-                    uses_prespacing = True
 
-            # Apply appropriate joining method
+            # If there are embedded spaces in the fragments, use direct concatenation
-            if uses_prespacing:
+            contains_spacing_between_fragments = has_leading_spaces or has_trailing_spaces
-                # Pre-spaced fragments - just concatenate
+
+            # Apply corresponding joining method
+            if contains_spacing_between_fragments:
+                # Fragments already have spacing - just concatenate
                 content = "".join(self._current_text_parts)
             else:
                 # Word-by-word fragments - join with spaces

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -223,6 +223,16 @@ class GeminiMultimodalLiveUserContextAggregator(OpenAIUserContextAggregator):
 
 
 class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggregator):
+    # The LLMAssistantContextAggregator uses TextFrames to aggregate the LLM output,
+    # but the GeminiMultimodalLiveAssistantContextAggregator pushes LLMTextFrames and TTSTextFrames. We
+    # need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames
+    # are process. This ensures that the context gets only one set of messages.
+    # GeminiMultimodalLiveLLMService also pushes TranscriptionFrames, so we need to
+    # ignore pushing those as well, as they're also TextFrames.
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        if not isinstance(frame, (LLMTextFrame, TranscriptionFrame)):
+            await super().process_frame(frame, direction)
+
     async def handle_user_image_frame(self, frame: UserImageRawFrame):
         # We don't want to store any images in the context. Revisit this later
         # when the API evolves.
@@ -344,7 +354,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
         self._bot_is_speaking = False
         self._user_audio_buffer = bytearray()
         self._bot_audio_buffer = bytearray()
-        self._bot_text_buffer = ""
 
         self._sample_rate = 24000
 
@@ -367,7 +376,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
             "vad": params.vad,
             "context_window_compression": params.context_window_compression.model_dump()
             if params.context_window_compression
-            else None,
+            else {},
             "extra": params.extra if isinstance(params.extra, dict) else {},
         }
 
@@ -427,7 +436,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
     #
 
     async def _handle_interruption(self):
-        pass
+        self._bot_is_speaking = False
+        await self.push_frame(TTSStoppedFrame())
+        await self.push_frame(LLMFullResponseEndFrame())
 
     async def _handle_user_started_speaking(self, frame):
         self._user_is_speaking = True
@@ -450,10 +461,12 @@ class GeminiMultimodalLiveLLMService(LLMService):
         text = await self._transcribe_audio(audio, context)
         if not text:
             return
-        logger.debug(f"[Transcription:user] {text}")
+        # Sometimes the transcription contains newlines; we want to remove them.
-        context.add_message({"role": "user", "content": [{"type": "text", "text": text}]})
+        cleaned_text = text.rstrip("\n")
+        logger.debug(f"[Transcription:user] {cleaned_text}")
+        context.add_message({"role": "user", "content": [{"type": "text", "text": cleaned_text}]})
         await self.push_frame(
-            TranscriptionFrame(text=text, user_id="user", timestamp=time_now_iso8601())
+            TranscriptionFrame(text=cleaned_text, user_id="user", timestamp=time_now_iso8601())
         )
 
     async def _transcribe_audio(self, audio, context):
@@ -839,14 +852,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
         if not part:
             return
 
-        text = part.text
-        if text:
-            if not self._bot_text_buffer:
-                await self.push_frame(LLMFullResponseStartFrame())
-
-            self._bot_text_buffer += text
-            await self.push_frame(LLMTextFrame(text=text))
-
         inline_data = part.inlineData
         if not inline_data:
             return
@@ -861,6 +866,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
         if not self._bot_is_speaking:
             self._bot_is_speaking = True
             await self.push_frame(TTSStartedFrame())
+            await self.push_frame(LLMFullResponseStartFrame())
 
         self._bot_audio_buffer.extend(audio)
         frame = TTSAudioRawFrame(
@@ -886,24 +892,20 @@ class GeminiMultimodalLiveLLMService(LLMService):
 
     async def _handle_evt_turn_complete(self, evt):
         self._bot_is_speaking = False
-        text = self._bot_text_buffer
-        self._bot_text_buffer = ""
-
-        if text:
-            await self.push_frame(LLMFullResponseEndFrame())
-
         await self.push_frame(TTSStoppedFrame())
+        await self.push_frame(LLMFullResponseEndFrame())
 
     async def _handle_evt_output_transcription(self, evt):
         if not evt.serverContent.outputTranscription:
             return
 
         text = evt.serverContent.outputTranscription.text
-        if text:
+
-            await self.push_frame(LLMFullResponseStartFrame())
+        if not text:
-            await self.push_frame(LLMTextFrame(text=text))
+            return
-            await self.push_frame(TTSTextFrame(text=text))
+
-            await self.push_frame(LLMFullResponseEndFrame())
+        await self.push_frame(LLMTextFrame(text=text))
+        await self.push_frame(TTSTextFrame(text=text))
 
     def create_context_aggregator(
         self,
@@ -934,6 +936,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
         GeminiMultimodalLiveContext.upgrade(context)
         user = GeminiMultimodalLiveUserContextAggregator(context, params=user_params)
 
-        assistant_params.expect_stripped_words = True
+        assistant_params.expect_stripped_words = False
         assistant = GeminiMultimodalLiveAssistantContextAggregator(context, params=assistant_params)
         return GeminiMultimodalLiveContextAggregatorPair(_user=user, _assistant=assistant)

src/pipecat/services/openai_realtime_beta/context.py

@@ -14,6 +14,7 @@ from pipecat.frames.frames import (
     FunctionCallResultFrame,
     LLMMessagesUpdateFrame,
     LLMSetToolsFrame,
+    LLMTextFrame,
 )
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection
@@ -170,6 +171,14 @@ class OpenAIRealtimeUserContextAggregator(OpenAIUserContextAggregator):
 
 
 class OpenAIRealtimeAssistantContextAggregator(OpenAIAssistantContextAggregator):
+    # The LLMAssistantContextAggregator uses TextFrames to aggregate the LLM output,
+    # but the OpenAIRealtimeLLMService pushes LLMTextFrames and TTSTextFrames. We
+    # need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames
+    # are process. This ensures that the context gets only one set of messages.
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        if not isinstance(frame, LLMTextFrame):
+            await super().process_frame(frame, direction)
+
     async def handle_function_call_result(self, frame: FunctionCallResultFrame):
         await super().handle_function_call_result(frame)