Add InputTextRawFrame class for handling raw text input in frames

- Introduced `InputTextRawFrame` to represent raw text input from users or programs.
- Updated `GeminiMultimodalLiveLLMService` to process `InputTextRawFrame` and send user text via the Gemini Live API's realtime input stream.
- Enhanced `_send_user_text` method documentation for clarity on its functionality and usage.

src/pipecat/frames/frames.py

@@ -1074,6 +1074,20 @@ class InputImageRawFrame(SystemFrame, ImageRawFrame):
         return f"{self.name}(pts: {pts}, source: {self.transport_source}, size: {self.size}, format: {self.format})"
 
 
+@dataclass  
+class InputTextRawFrame(SystemFrame, TextFrame):
+    """Raw text input frame from transport.
+    
+    Text input usually coming from user typing or programmatic text injection
+    that should be sent to LLM services as input, similar to how InputAudioRawFrame 
+    and InputImageRawFrame represent user audio and video input.
+    """
+
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return f"{self.name}(pts: {pts}, source: {self.transport_source}, text: [{self.text}])"
+
+
 @dataclass
 class UserAudioRawFrame(InputAudioRawFrame):
     """Raw audio input frame associated with a specific user.

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -32,6 +32,7 @@ from pipecat.frames.frames import (
     Frame,
     InputAudioRawFrame,
     InputImageRawFrame,
+    InputTextRawFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMMessagesAppendFrame,
@@ -733,8 +734,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, LLMTextFrame):
+        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)
@@ -967,7 +969,19 @@ class GeminiMultimodalLiveLLMService(LLMService):
             self._user_audio_buffer = self._user_audio_buffer[-length:]
 
     async def _send_user_text(self, text: str):
-        """Send user text to Gemini Live API."""
+        """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.