diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index 2dbdac377..5819c9cb9 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/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 ( - OpenAILLMContext, - OpenAILLMContextFrame, -) +from pipecat.processors.aggregators.llm_response_universal import LLMContextAggregatorPair 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): - GoogleLLMContext.upgrade_to_google(self._context) - last_message = frame.context.messages[-1] + if isinstance(frame, LLMContextFrame): + last_message = frame.context.get_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 "-" - self._context.add_message(Content(role="user", parts=[Part(text=transcription)])) + transcription = await self._transcription_buffer.wait_for_transcription() or "-" + self._context.add_message({"role": "user", "content": transcription}) - self.open_gate() - for frame, direction in self._frames_buffer: - await self.push_frame(frame, direction) - self._frames_buffer = [] + self.open_gate() + for frame, direction in self._frames_buffer: + await self.push_frame(frame, direction) + 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_aggregator = conversation_llm.create_context_aggregator(context) + context = LLMContext() + 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