Push GeminiMultimodalLiveLLMService TranscriptionFrame Upstream, remove direct context addition

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

@@ -12,10 +12,12 @@ from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import TranscriptionMessage
 from pipecat.pipeline.pipeline import Pipeline
 from pipecat.pipeline.runner import PipelineRunner
 from pipecat.pipeline.task import PipelineParams, PipelineTask
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.processors.transcript_processor import TranscriptProcessor
 from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService
 from pipecat.transports.base_transport import TransportParams
 from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
@@ -69,12 +71,16 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
     )
     context_aggregator = llm.create_context_aggregator(context)
 
+    transcript = TranscriptProcessor()
+
     pipeline = Pipeline(
         [
             transport.input(),
             context_aggregator.user(),
+            transcript.user(),
             llm,
             transport.output(),
+            transcript.assistant(),
             context_aggregator.assistant(),
         ]
     )
@@ -103,6 +109,15 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
         logger.info(f"Client closed connection")
         await task.cancel()
 
+    # Register event handler for transcript updates
+    @transcript.event_handler("on_transcript_update")
+    async def on_transcript_update(processor, frame):
+        for msg in frame.messages:
+            if isinstance(msg, TranscriptionMessage):
+                timestamp = f"[{msg.timestamp}] " if msg.timestamp else ""
+                line = f"{timestamp}{msg.role}: {msg.content}"
+                logger.info(f"Transcript: {line}")
+
     runner = PipelineRunner(handle_sigint=False)
 
     await runner.run(task)

src/pipecat/services/gemini_multimodal_live/gemini.py

@@ -227,10 +227,8 @@ class GeminiMultimodalLiveAssistantContextAggregator(OpenAIAssistantContextAggre
     # but the GeminiMultimodalLiveAssistantContextAggregator pushes LLMTextFrames and TTSTextFrames. We
     # need to override this proces_frame for LLMTextFrame, so that only the TTSTextFrames
     # are process. This ensures that the context gets only one set of messages.
-    # GeminiMultimodalLiveLLMService also pushes TranscriptionFrames, so we need to
-    # ignore pushing those as well, as they're also TextFrames.
     async def process_frame(self, frame: Frame, direction: FrameDirection):
-        if not isinstance(frame, (LLMTextFrame, TranscriptionFrame)):
+        if not isinstance(frame, LLMTextFrame):
             await super().process_frame(frame, direction)
 
     async def handle_user_image_frame(self, frame: UserImageRawFrame):
@@ -464,9 +462,9 @@ class GeminiMultimodalLiveLLMService(LLMService):
         # Sometimes the transcription contains newlines; we want to remove them.
         cleaned_text = text.rstrip("\n")
         logger.debug(f"[Transcription:user] {cleaned_text}")
-        context.add_message({"role": "user", "content": [{"type": "text", "text": cleaned_text}]})
         await self.push_frame(
-            TranscriptionFrame(text=cleaned_text, user_id="user", timestamp=time_now_iso8601())
+            TranscriptionFrame(text=cleaned_text, user_id="user", timestamp=time_now_iso8601()),
+            FrameDirection.UPSTREAM,
         )
 
     async def _transcribe_audio(self, audio, context):