Merge pull request #3431 from pipecat-ai/mb/update-realtime-examples-transcript-handler

Update GeminiLiveLLMService to push thought frames, update 26a for new transcript events

changelog/3431.changed.md

@@ -0,0 +1 @@
+- Updated `GeminiLiveLLMService` to push `LLMThoughtStartFrame`, `LLMThoughtTextFrame`, and `LLMThoughtEndFrame` when the model returns thought content.

examples/foundational/26a-gemini-live-transcription.py

@@ -12,13 +12,16 @@ from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
-from pipecat.frames.frames import LLMRunFrame, TranscriptionMessage
+from pipecat.frames.frames import LLMRunFrame
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.llm_context import LLMContext
-from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
+from pipecat.processors.aggregators.llm_response_universal import (
-from pipecat.processors.transcript_processor import TranscriptProcessor
+    AssistantTurnStoppedMessage,
+    LLMContextAggregatorPair,
+    UserTurnStoppedMessage,
+)
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService
@@ -93,17 +96,16 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     )
     context_aggregator = LLMContextAggregatorPair(context)
 
-    transcript = TranscriptProcessor()
+    user_aggregator = context_aggregator.user()
+    assistant_aggregator = context_aggregator.assistant()
 
     pipeline = Pipeline(
         [
             transport.input(),
-            context_aggregator.user(),
+            user_aggregator,
-            transcript.user(),
             llm,
             transport.output(),
-            transcript.assistant(),
+            assistant_aggregator,
-            context_aggregator.assistant(),
         ]
     )
 
@@ -127,14 +129,17 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         logger.info(f"Client disconnected")
         await task.cancel()
 
-    # Register event handler for transcript updates
+    @user_aggregator.event_handler("on_user_turn_stopped")
-    @transcript.event_handler("on_transcript_update")
+    async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMessage):
-    async def on_transcript_update(processor, frame):
+        timestamp = f"[{message.timestamp}] " if message.timestamp else ""
-        for msg in frame.messages:
+        line = f"{timestamp}user: {message.content}"
-            if isinstance(msg, TranscriptionMessage):
+        logger.info(f"Transcript: {line}")
-                timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+
-                line = f"{timestamp}{msg.role}: {msg.content}"
+    @assistant_aggregator.event_handler("on_assistant_turn_stopped")
-                logger.info(f"Transcript: {line}")
+    async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage):
+        timestamp = f"[{message.timestamp}] " if message.timestamp else ""
+        line = f"{timestamp}assistant: {message.content}"
+        logger.info(f"Transcript: {line}")
 
     runner = PipelineRunner(handle_sigint=runner_args.handle_sigint)
 

src/pipecat/services/google/gemini_live/llm.py

@@ -44,6 +44,9 @@ from pipecat.frames.frames import (
     LLMMessagesAppendFrame,
     LLMSetToolsFrame,
     LLMTextFrame,
+    LLMThoughtEndFrame,
+    LLMThoughtStartFrame,
+    LLMThoughtTextFrame,
     LLMUpdateSettingsFrame,
     StartFrame,
     TranscriptionFrame,
@@ -1455,10 +1458,19 @@ class GeminiLiveLLMService(LLMService):
                 await self._set_bot_is_responding(True)
                 await self.push_frame(LLMFullResponseStartFrame())
 
-            self._bot_text_buffer += text
+            # Check if this is a thought
-            self._search_result_buffer += text  # Also accumulate for grounding
+            if part.thought:
-            frame = LLMTextFrame(text=text)
+                # Gemini Live emits fully-formed thoughts rather than chunks,
-            await self.push_frame(frame)
+                # so bracket each thought in start/end frames
+                await self.push_frame(LLMThoughtStartFrame())
+                await self.push_frame(LLMThoughtTextFrame(text))
+                await self.push_frame(LLMThoughtEndFrame())
+            else:
+                # Regular text response
+                self._bot_text_buffer += text
+                self._search_result_buffer += text  # Also accumulate for grounding
+                frame = LLMTextFrame(text=text)
+                await self.push_frame(frame)
 
         # Check for grounding metadata in server content
         if msg.server_content and msg.server_content.grounding_metadata: