Update 26a to use context aggregator transcription 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

@@ -1464,7 +1464,7 @@ class GeminiLiveLLMService(LLMService):
                 # so bracket each thought in start/end frames
                 await self.push_frame(LLMThoughtStartFrame())
                 await self.push_frame(LLMThoughtTextFrame(text))
-                await self.push_frame(LLMThoughtEndFrame(signature=part.thought_signature))
+                await self.push_frame(LLMThoughtEndFrame())
             else:
                 # Regular text response
                 self._bot_text_buffer += text