diff --git a/CHANGELOG.md b/CHANGELOG.md index 05481767d..249133331 100644 --- a/CHANGELOG.md +++ b/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. 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, ) 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 diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 3881f7c7e..791888993 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/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}") - 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): @@ -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()) - 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 +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) 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)