From 685f951ae218dfd7756229631e997dc09ef8951b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 29 Apr 2025 15:40:18 -0400 Subject: [PATCH 1/9] Fix: GeminiMultimodalLiveLLMService was appending tokens to the context --- CHANGELOG.md | 3 ++ .../processors/aggregators/llm_response.py | 5 +-- .../services/gemini_multimodal_live/gemini.py | 34 +++++++------------ 3 files changed, 18 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 08e9d30ed..c3f8fc3fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -80,6 +80,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. diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index fa163ac0a..5d8b6a32c 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -36,6 +36,7 @@ from pipecat.frames.frames import ( StartInterruptionFrame, TextFrame, TranscriptionFrame, + TTSTextFrame, UserImageRawFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, @@ -493,7 +494,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): await self._handle_llm_start(frame) elif isinstance(frame, LLMFullResponseEndFrame): await self._handle_llm_end(frame) - elif isinstance(frame, TextFrame): + elif isinstance(frame, TTSTextFrame): await self._handle_text(frame) elif isinstance(frame, LLMMessagesAppendFrame): self.add_messages(frame.messages) @@ -620,7 +621,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): self._started -= 1 await self.push_aggregation() - async def _handle_text(self, frame: TextFrame): + async def _handle_text(self, frame: TTSTextFrame): if not self._started: return diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 3881f7c7e..79cc556f5 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -344,7 +344,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 @@ -427,7 +426,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 @@ -839,14 +840,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 +854,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 +880,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()) - await self.push_frame(LLMTextFrame(text=text)) - await self.push_frame(TTSTextFrame(text=text)) - await self.push_frame(LLMFullResponseEndFrame()) + + if not text: + return + + await self.push_frame(LLMTextFrame(text=text)) + await self.push_frame(TTSTextFrame(text=text)) def create_context_aggregator( self, @@ -934,6 +924,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) From 2ba60fc41f47bddea8e868faa644be8fc6812582 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 29 Apr 2025 15:50:52 -0400 Subject: [PATCH 2/9] Update TranscriptProcessor to handle GeminiMultimodalLiveLLMService changes --- .../processors/transcript_processor.py | 64 ++++++++++--------- 1 file changed, 35 insertions(+), 29 deletions(-) diff --git a/src/pipecat/processors/transcript_processor.py b/src/pipecat/processors/transcript_processor.py index a2ad22223..85bcd072d 100644 --- a/src/pipecat/processors/transcript_processor.py +++ b/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 - the appropriate joining strategy. It handles fragments from different TTS - services with different formatting patterns. + contain embedded spacing (spaces at the beginning or end of fragments) or not, + and applies the appropriate joining strategy. It handles fragments from different + 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: [" going"] - TTSTextFrame: ["?"] + TTSTextFrame: [" it"] # Leading space ``` - 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: ["How"] - TTSTextFrame: ["is"] - TTSTextFrame: ["it"] - TTSTextFrame: ["going?"] + TTSTextFrame: ["there"] + TTSTextFrame: ["how"] + TTSTextFrame: ["are"] + TTSTextFrame: ["you"] ``` - 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 - uses_prespacing = False - if len(self._current_text_parts) > 1: - # Check if any fragment after the first one starts with whitespace - has_spaced_parts = any( - part and part[0].isspace() for part in self._current_text_parts[1:] - ) - if has_spaced_parts: - uses_prespacing = True + has_leading_spaces = any( + part and part[0].isspace() for part in self._current_text_parts[1:] + ) + has_trailing_spaces = any( + part and part[-1].isspace() for part in self._current_text_parts[:-1] + ) - # Apply appropriate joining method - if uses_prespacing: - # Pre-spaced fragments - just concatenate + # If there are embedded spaces in the fragments, use direct concatenation + contains_spacing_between_fragments = has_leading_spaces or has_trailing_spaces + + # 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 From facbebb15fa1d67545336fe99f0c8e4df0d58ca6 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 29 Apr 2025 15:58:34 -0400 Subject: [PATCH 3/9] Transcribe user audio in 26b --- .../foundational/26b-gemini-multimodal-live-function-calling.py | 1 + 1 file changed, 1 insertion(+) diff --git a/examples/foundational/26b-gemini-multimodal-live-function-calling.py b/examples/foundational/26b-gemini-multimodal-live-function-calling.py index 29017ba00..2224d4dad 100644 --- a/examples/foundational/26b-gemini-multimodal-live-function-calling.py +++ b/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, ) From 82a9d7f9921b87f8111d5e23de68ec8a9ac0c323 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 30 Apr 2025 16:24:04 -0400 Subject: [PATCH 4/9] Gemini Mulitmodal Live to push TTSTextFrame only --- src/pipecat/processors/aggregators/llm_response.py | 5 ++--- src/pipecat/services/gemini_multimodal_live/gemini.py | 2 -- 2 files changed, 2 insertions(+), 5 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 5d8b6a32c..fa163ac0a 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -36,7 +36,6 @@ from pipecat.frames.frames import ( StartInterruptionFrame, TextFrame, TranscriptionFrame, - TTSTextFrame, UserImageRawFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, @@ -494,7 +493,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): await self._handle_llm_start(frame) elif isinstance(frame, LLMFullResponseEndFrame): await self._handle_llm_end(frame) - elif isinstance(frame, TTSTextFrame): + elif isinstance(frame, TextFrame): await self._handle_text(frame) elif isinstance(frame, LLMMessagesAppendFrame): self.add_messages(frame.messages) @@ -621,7 +620,7 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): self._started -= 1 await self.push_aggregation() - async def _handle_text(self, frame: TTSTextFrame): + async def _handle_text(self, frame: TextFrame): if not self._started: return diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 79cc556f5..b5b650ef5 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -30,7 +30,6 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMMessagesAppendFrame, LLMSetToolsFrame, - LLMTextFrame, LLMUpdateSettingsFrame, StartFrame, StartInterruptionFrame, @@ -892,7 +891,6 @@ class GeminiMultimodalLiveLLMService(LLMService): if not text: return - await self.push_frame(LLMTextFrame(text=text)) await self.push_frame(TTSTextFrame(text=text)) def create_context_aggregator( From 7e09933070afd75ba39c1ee77a314bf562267a4f Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 30 Apr 2025 16:25:02 -0400 Subject: [PATCH 5/9] OpenAI Realtime should push TTSTextFrame only --- src/pipecat/services/openai_realtime_beta/openai.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 94d848b77..a4b339362 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -24,7 +24,6 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMMessagesAppendFrame, LLMSetToolsFrame, - LLMTextFrame, LLMUpdateSettingsFrame, StartFrame, StartInterruptionFrame, @@ -525,7 +524,6 @@ class OpenAIRealtimeBetaLLMService(LLMService): async def _handle_evt_audio_transcript_delta(self, evt): if evt.delta: - await self.push_frame(LLMTextFrame(evt.delta)) await self.push_frame(TTSTextFrame(evt.delta)) async def _handle_evt_speech_started(self, evt): From ef29800fe9cbe460b8cfb99b9b824a47f5e7e326 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 30 Apr 2025 16:26:38 -0400 Subject: [PATCH 6/9] Update the changelog --- CHANGELOG.md | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index c3f8fc3fb..855ef02dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `OpenAIRealtimeBetaLLMService` and `GeminiMultimodalLiveLLMService` no longer + push `LLMTextFrame`. Instead, they both push only `TTSTextFrame`, which is + used to aggregate the assistant context and generate a transcript. + - Function calls now receive a single parameter `FunctionCallParams` instead of `(function_name, tool_call_id, args, llm, context, result_callback)` which is now deprecated. From 9c5878af3d1a9ce839f6278ce3019705cda1cf0a Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 30 Apr 2025 17:12:16 -0400 Subject: [PATCH 7/9] OpenAI Realtime and Gemini Live push LLMTextFrame again, overwrite the assitant context aggregator for LLMTextFrame --- CHANGELOG.md | 4 ---- .../services/gemini_multimodal_live/gemini.py | 12 +++++++++++- src/pipecat/services/openai_realtime_beta/context.py | 9 +++++++++ src/pipecat/services/openai_realtime_beta/openai.py | 2 ++ 4 files changed, 22 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 855ef02dc..c3f8fc3fb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,10 +34,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed -- `OpenAIRealtimeBetaLLMService` and `GeminiMultimodalLiveLLMService` no longer - push `LLMTextFrame`. Instead, they both push only `TTSTextFrame`, which is - used to aggregate the assistant context and generate a transcript. - - Function calls now receive a single parameter `FunctionCallParams` instead of `(function_name, tool_call_id, args, llm, context, result_callback)` which is now deprecated. diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index b5b650ef5..01f55407e 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -30,6 +30,7 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMMessagesAppendFrame, LLMSetToolsFrame, + LLMTextFrame, LLMUpdateSettingsFrame, StartFrame, StartInterruptionFrame, @@ -222,6 +223,14 @@ 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. + async def process_frame(self, frame: Frame, direction: FrameDirection): + if not isinstance(frame, LLMTextFrame): + 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. @@ -365,7 +374,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 {}, } @@ -891,6 +900,7 @@ class GeminiMultimodalLiveLLMService(LLMService): if not text: return + await self.push_frame(LLMTextFrame(text=text)) await self.push_frame(TTSTextFrame(text=text)) def create_context_aggregator( diff --git a/src/pipecat/services/openai_realtime_beta/context.py b/src/pipecat/services/openai_realtime_beta/context.py index 80cffbef2..d9e105098 100644 --- a/src/pipecat/services/openai_realtime_beta/context.py +++ b/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) diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index a4b339362..94d848b77 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -24,6 +24,7 @@ from pipecat.frames.frames import ( LLMFullResponseStartFrame, LLMMessagesAppendFrame, LLMSetToolsFrame, + LLMTextFrame, LLMUpdateSettingsFrame, StartFrame, StartInterruptionFrame, @@ -524,6 +525,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): async def _handle_evt_audio_transcript_delta(self, evt): if evt.delta: + await self.push_frame(LLMTextFrame(evt.delta)) await self.push_frame(TTSTextFrame(evt.delta)) async def _handle_evt_speech_started(self, evt): From c4f758725ec82e913ab03ffc0a8a056d48a36e22 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 30 Apr 2025 18:22:43 -0400 Subject: [PATCH 8/9] Ignore TranscriptionFrames too --- src/pipecat/services/gemini_multimodal_live/gemini.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 01f55407e..5e6c6ec7c 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -227,8 +227,10 @@ class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggre # 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): + if not isinstance(frame, (LLMTextFrame, TranscriptionFrame)): await super().process_frame(frame, direction) async def handle_user_image_frame(self, frame: UserImageRawFrame): From a1e1255f16e7256cb8731fc4634f3b9f9cbc8606 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 30 Apr 2025 18:26:31 -0400 Subject: [PATCH 9/9] Strip newlines from generated user transcript --- src/pipecat/services/gemini_multimodal_live/gemini.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 5e6c6ec7c..791888993 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -461,10 +461,12 @@ class GeminiMultimodalLiveLLMService(LLMService): text = await self._transcribe_audio(audio, context) if not text: return - logger.debug(f"[Transcription:user] {text}") - context.add_message({"role": "user", "content": [{"type": "text", "text": text}]}) + # Sometimes the transcription contains newlines; we want to remove them. + 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):