Update another "natural conversation" example to use universal LLMContext. Note that this one had to also be fixed in various ways, as it wasn't working.

examples/foundational/22d-natural-conversation-gemini-audio.py

@@ -9,7 +9,6 @@ import os
 import time
 
 from dotenv import load_dotenv
-from google.genai.types import Content, Part
 from loguru import logger
 
 from pipecat.audio.vad.silero import SileroVADAnalyzer
@@ -21,6 +20,7 @@ from pipecat.frames.frames import (
     FunctionCallResultFrame,
     InputAudioRawFrame,
     InterruptionFrame,
+    LLMContextFrame,
     LLMFullResponseStartFrame,
     LLMRunFrame,
     StartFrame,
@@ -34,20 +34,18 @@ from pipecat.pipeline.parallel_pipeline import ParallelPipeline
 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 import (
     LLMAssistantAggregatorParams,
     LLMAssistantResponseAggregator,
 )
-from pipecat.processors.aggregators.openai_llm_context import (
+from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair
-    OpenAILLMContext,
-    OpenAILLMContextFrame,
-)
 from pipecat.processors.filters.function_filter import FunctionFilter
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 from pipecat.runner.types import RunnerArguments
 from pipecat.runner.utils import create_transport
 from pipecat.services.cartesia.tts import CartesiaTTSService
-from pipecat.services.google.llm import GoogleLLMContext, GoogleLLMService
+from pipecat.services.google.llm import GoogleLLMService
 from pipecat.services.llm_service import LLMService
 from pipecat.sync.base_notifier import BaseNotifier
 from pipecat.sync.event_notifier import EventNotifier
@@ -375,7 +373,7 @@ class AudioAccumulator(FrameProcessor):
         await super().process_frame(frame, direction)
 
         # ignore context frame
-        if isinstance(frame, OpenAILLMContextFrame):
+        if isinstance(frame, LLMContextFrame):
             return
 
         if isinstance(frame, TranscriptionFrame):
@@ -392,9 +390,9 @@ class AudioAccumulator(FrameProcessor):
                 f"Processing audio buffer seconds: ({len(self._audio_frames)}) ({len(data)}) {len(data) / 2 / 16000}"
             )
             self._user_speaking = False
-            context = GoogleLLMContext()
+            context = LLMContext()
             context.add_audio_frames_message(audio_frames=self._audio_frames)
-            await self.push_frame(OpenAILLMContextFrame(context=context))
+            await self.push_frame(LLMContextFrame(context=context))
         elif isinstance(frame, InputAudioRawFrame):
             # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest
             # frames as necessary.
@@ -513,7 +511,7 @@ class LLMAggregatorBuffer(LLMAssistantResponseAggregator):
 class ConversationAudioContextAssembler(FrameProcessor):
     """Takes the single-message context generated by the AudioAccumulator and adds it to the conversation LLM's context."""
 
-    def __init__(self, context: OpenAILLMContext, **kwargs):
+    def __init__(self, context: LLMContext, **kwargs):
         super().__init__(**kwargs)
         self._context = context
 
@@ -525,11 +523,10 @@ class ConversationAudioContextAssembler(FrameProcessor):
             await self.push_frame(frame, direction)
             return
 
-        if isinstance(frame, OpenAILLMContextFrame):
+        if isinstance(frame, LLMContextFrame):
-            GoogleLLMContext.upgrade_to_google(self._context)
+            last_message = frame.context.get_messages()[-1]
-            last_message = frame.context.messages[-1]
             self._context._messages.append(last_message)
-            await self.push_frame(OpenAILLMContextFrame(context=self._context))
+            await self.push_frame(LLMContextFrame(context=self._context))
 
 
 class OutputGate(FrameProcessor):
@@ -543,7 +540,7 @@ class OutputGate(FrameProcessor):
     def __init__(
         self,
         notifier: BaseNotifier,
-        context: OpenAILLMContext,
+        context: LLMContext,
         llm_transcription_buffer: LLMAggregatorBuffer,
         **kwargs,
     ):
@@ -610,19 +607,23 @@ class OutputGate(FrameProcessor):
             self._gate_task = None
 
     async def _gate_task_handler(self):
-        await self._notifier.wait()
+        while True:
+            try:
+                await self._notifier.wait()
 
-        transcription = await self._transcription_buffer.wait_for_transcription() or "-"
+                transcription = await self._transcription_buffer.wait_for_transcription() or "-"
-        self._context.add_message(Content(role="user", parts=[Part(text=transcription)]))
+                self._context.add_message({"role": "user", "content": transcription})
 
-        self.open_gate()
+                self.open_gate()
-        for frame, direction in self._frames_buffer:
+                for frame, direction in self._frames_buffer:
-            await self.push_frame(frame, direction)
+                    await self.push_frame(frame, direction)
-        self._frames_buffer = []
+                self._frames_buffer = []
+            except asyncio.CancelledError:
+                break
 
 
 class TurnDetectionLLM(Pipeline):
-    def __init__(self, llm: LLMService, context: OpenAILLMContext):
+    def __init__(self, llm: LLMService, context: LLMContext):
         # This is the LLM that will transcribe user speech.
         tx_llm = GoogleLLMService(
             name="Transcriber",
@@ -648,10 +649,10 @@ class TurnDetectionLLM(Pipeline):
         # as complete or incomplete.
         # statement_judge_context_filter = StatementJudgeAudioContextAccumulator(notifier=notifier)
 
-        audio_accumulater = AudioAccumulator()
+        audio_accumulator = AudioAccumulator()
         # This sends a UserStoppedSpeakingFrame and triggers the notifier event
         completeness_check = CompletenessCheck(
-            notifier=notifier, audio_accumulator=audio_accumulater
+            notifier=notifier, audio_accumulator=audio_accumulator
         )
 
         async def block_user_stopped_speaking(frame):
@@ -667,7 +668,7 @@ class TurnDetectionLLM(Pipeline):
 
         super().__init__(
             [
-                audio_accumulater,
+                audio_accumulator,
                 ParallelPipeline(
                     [
                         # Pass everything except UserStoppedSpeaking to the elements after
@@ -734,8 +735,8 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
         system_instruction=conversation_system_instruction,
     )
 
-    context = OpenAILLMContext()
+    context = LLMContext()
-    context_aggregator = conversation_llm.create_context_aggregator(context)
+    context_aggregator = LLMContextAggregatorPair(context)
 
     llm = TurnDetectionLLM(conversation_llm, context)
 
@@ -761,12 +762,10 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments):
     @transport.event_handler("on_client_connected")
     async def on_client_connected(transport, client):
         logger.info(f"Client connected")
-        # Kick off the conversation.
-        await task.queue_frames([LLMRunFrame()])
 
     @transport.event_handler("on_app_message")
     async def on_app_message(transport, message, sender):
-        logger.debug(f"Received app message: {message}")
+        logger.debug(f"Received app message: {message}, sender: {sender}")  # TODO: revert
         if "message" not in message:
             return