diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 498c6e485..bc9e8a06d 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -1076,6 +1076,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. diff --git a/src/pipecat/services/gemini_multimodal_live/events.py b/src/pipecat/services/gemini_multimodal_live/events.py index 0e229c22f..1766cf806 100644 --- a/src/pipecat/services/gemini_multimodal_live/events.py +++ b/src/pipecat/services/gemini_multimodal_live/events.py @@ -114,13 +114,15 @@ class RealtimeInputConfig(BaseModel): class RealtimeInput(BaseModel): - """Contains realtime input media chunks. + """Contains realtime input media chunks and text. Parameters: mediaChunks: List of media chunks for realtime processing. + text: Text for realtime processing. """ - mediaChunks: List[MediaChunk] + mediaChunks: Optional[List[MediaChunk]] = None + text: Optional[str] = None class ClientContent(BaseModel): @@ -190,6 +192,24 @@ class VideoInputMessage(BaseModel): ) +class TextInputMessage(BaseModel): + """Message containing text input data.""" + + realtimeInput: RealtimeInput + + @classmethod + def from_text(cls, text: str) -> "TextInputMessage": + """Create a text input message from a string. + + Args: + text: The text to send. + + Returns: + A TextInputMessage instance. + """ + return cls(realtimeInput=RealtimeInput(text=text)) + + class ClientContentMessage(BaseModel): """Message containing client content for the API. diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index d1ae3ee33..238fdd5db 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/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: