Gemini Live to transcribe user audio

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -4,7 +4,6 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import asyncio
 import base64
 import json
 import time
@@ -59,10 +58,10 @@ from pipecat.services.openai.llm import (
     OpenAIUserContextAggregator,
 )
 from pipecat.transcriptions.language import Language
+from pipecat.utils.string import match_endofsentence
 from pipecat.utils.time import time_now_iso8601
 
 from . import events
-from .audio_transcriber import AudioTranscriber
 
 try:
     import websockets
@@ -316,7 +315,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
         start_video_paused: bool = False,
         system_instruction: Optional[str] = None,
         tools: Optional[Union[List[dict], ToolsSchema]] = None,
-        transcribe_user_audio: bool = False,
         params: InputParams = InputParams(),
         inference_on_context_initialization: bool = True,
         **kwargs,
@@ -339,18 +337,16 @@ class GeminiMultimodalLiveLLMService(LLMService):
         self._context = None
         self._websocket = None
         self._receive_task = None
-        self._transcribe_audio_task = None
-        self._transcribe_audio_queue = asyncio.Queue()
 
         self._disconnecting = False
         self._api_session_ready = False
         self._run_llm_when_api_session_ready = False
 
-        self._transcriber = AudioTranscriber(api_key)
-        self._transcribe_user_audio = transcribe_user_audio
         self._user_is_speaking = False
         self._bot_is_speaking = False
         self._user_audio_buffer = bytearray()
+        self._user_transcription_buffer = ""
+        self._last_transcription_sent = ""
         self._bot_audio_buffer = bytearray()
         self._bot_text_buffer = ""
 
@@ -445,7 +441,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
 
     async def _handle_user_stopped_speaking(self, frame):
         self._user_is_speaking = False
-        audio = self._user_audio_buffer
         self._user_audio_buffer = bytearray()
         if self._needs_turn_complete_message:
             self._needs_turn_complete_message = False
@@ -453,36 +448,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
                 {"clientContent": {"turnComplete": True}}
             )
             await self.send_client_event(evt)
-        if self._transcribe_user_audio and self._context:
-            await self._transcribe_audio_queue.put(audio)
-
-    async def _handle_transcribe_user_audio(self, audio, context):
-        text = await self._transcribe_audio(audio, context)
-        if not text:
-            return
-        # Sometimes the transcription contains newlines; we want to remove them.
-        cleaned_text = text.rstrip("\n")
-        logger.debug(f"[Transcription:user] {cleaned_text}")
-        await self.push_frame(
-            TranscriptionFrame(text=cleaned_text, user_id="user", timestamp=time_now_iso8601()),
-            FrameDirection.UPSTREAM,
-        )
-
-    async def _transcribe_audio(self, audio, context):
-        (text, prompt_tokens, completion_tokens, total_tokens) = await self._transcriber.transcribe(
-            audio, context
-        )
-        if not text:
-            return ""
-        # The only usage metrics we have right now are for the transcriber LLM. The Live API is free.
-        await self.start_llm_usage_metrics(
-            LLMTokenUsage(
-                prompt_tokens=prompt_tokens,
-                completion_tokens=completion_tokens,
-                total_tokens=total_tokens,
-            )
-        )
-        return text
 
     #
     # frame processing
@@ -560,7 +525,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
             uri = f"wss://{self._base_url}?key={self._api_key}"
             self._websocket = await websockets.connect(uri=uri)
             self._receive_task = self.create_task(self._receive_task_handler())
-            self._transcribe_audio_task = self.create_task(self._transcribe_audio_handler())
 
             # Create the basic configuration
             config_data = {
@@ -582,6 +546,7 @@ class GeminiMultimodalLiveLLMService(LLMService):
                         },
                         "media_resolution": self._settings["media_resolution"].value,
                     },
+                    "input_audio_transcription": {},
                     "output_audio_transcription": {},
                 }
             }
@@ -664,9 +629,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
             if self._receive_task:
                 await self.cancel_task(self._receive_task, timeout=1.0)
                 self._receive_task = None
-            if self._transcribe_audio_task:
-                await self.cancel_task(self._transcribe_audio_task)
-                self._transcribe_audio_task = None
             self._disconnecting = False
         except Exception as e:
             logger.error(f"{self} error disconnecting: {e}")
@@ -703,6 +665,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
                 await self._handle_evt_model_turn(evt)
             elif evt.serverContent and evt.serverContent.turnComplete:
                 await self._handle_evt_turn_complete(evt)
+            elif evt.serverContent and evt.serverContent.inputTranscription:
+                await self._handle_evt_input_transcription(evt)
             elif evt.serverContent and evt.serverContent.outputTranscription:
                 await self._handle_evt_output_transcription(evt)
             elif evt.toolCall:
@@ -714,11 +678,6 @@ class GeminiMultimodalLiveLLMService(LLMService):
             else:
                 pass
 
-    async def _transcribe_audio_handler(self):
-        while True:
-            audio = await self._transcribe_audio_queue.get()
-            await self._handle_transcribe_user_audio(audio, self._context)
-
     #
     #
     #
@@ -911,6 +870,48 @@ class GeminiMultimodalLiveLLMService(LLMService):
 
         await self.push_frame(LLMFullResponseEndFrame())
 
+    async def _handle_evt_input_transcription(self, evt):
+        """Handle the input transcription event.
+
+        Gemini Live sends user transcriptions in either single words or multi-word
+        phrases. As a result, we have to aggregate the input transcription. This handler
+        aggregates into sentences, splitting on the end of sentence markers.
+        """
+        if not evt.serverContent.inputTranscription:
+            return
+
+        text = evt.serverContent.inputTranscription.text
+
+        if not text:
+            return
+
+        # Strip leading space from sentence starts if buffer is empty
+        if text.startswith(" ") and not self._user_transcription_buffer:
+            text = text.lstrip()
+
+        # Accumulate text in the buffer
+        self._user_transcription_buffer += text
+
+        # Check for complete sentences
+        while True:
+            eos_end_marker = match_endofsentence(self._user_transcription_buffer)
+            if not eos_end_marker:
+                break
+
+            # Extract the complete sentence
+            complete_sentence = self._user_transcription_buffer[:eos_end_marker]
+            # Keep the remainder for the next chunk
+            self._user_transcription_buffer = self._user_transcription_buffer[eos_end_marker:]
+
+            # Send a TranscriptionFrame with the complete sentence
+            logger.debug(f"[Transcription:user] [{complete_sentence}]")
+            await self.push_frame(
+                TranscriptionFrame(
+                    text=complete_sentence, user_id="", timestamp=time_now_iso8601()
+                ),
+                FrameDirection.UPSTREAM,
+            )
+
     async def _handle_evt_output_transcription(self, evt):
         if not evt.serverContent.outputTranscription:
             return