Merge pull request #2223 from getchannel/realtime-text

Add text input handling to unify context for realtimeInput stream of GeminiMultimodalLiveService

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -32,6 +32,7 @@ from pipecat.frames.frames import (
     Frame,
     InputAudioRawFrame,
     InputImageRawFrame,
+    InputTextRawFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
@@ -738,6 +739,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
                 # Support just one tool call per context frame for now
                 tool_result_message = context.messages[-1]
                 await self._tool_result(tool_result_message)
+        elif isinstance(frame, InputTextRawFrame):
+            await self._send_user_text(frame.text)
+            await self.push_frame(frame, direction)
         elif isinstance(frame, InputAudioRawFrame):
             await self._send_user_audio(frame)
             await self.push_frame(frame, direction)
@@ -971,6 +975,26 @@ class GeminiMultimodalLiveLLMService(LLMService):
             length = int((frame.sample_rate * frame.num_channels * 2) * 0.5)
             self._user_audio_buffer = self._user_audio_buffer[-length:]
 
+    async def _send_user_text(self, text: str):
+        """Send user text via Gemini Live API's realtime input stream.
+
+        This method sends text through the realtimeInput stream (via TextInputMessage)
+        rather than the clientContent stream. This ensures text input is synchronized
+        with audio and video inputs, preventing temporal misalignment that can occur
+        when different modalities are processed through separate API pathways.
+
+        After sending the text, we signal turn completion to trigger a model response
+        for text-only interactions.
+
+        Args:
+            text: The text to send as user input.
+        """
+        evt = events.TextInputMessage.from_text(text)
+        await self.send_client_event(evt)
+        # After sending text, we need to signal that the turn is complete.
+        evt = events.ClientContentMessage.model_validate({"clientContent": {"turnComplete": True}})
+        await self.send_client_event(evt)
+
     async def _send_user_video(self, frame):
         """Send user video frame to Gemini Live API."""
         if self._video_input_paused: