diff --git a/examples/foundational/25-google-audio-in.py b/examples/foundational/25-google-audio-in.py new file mode 100644 index 000000000..49654e673 --- /dev/null +++ b/examples/foundational/25-google-audio-in.py @@ -0,0 +1,263 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import aiohttp +import asyncio +import os +import sys + +import google.ai.generativelanguage as glm + +from dataclasses import dataclass +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.parallel_pipeline import ParallelPipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import ( + OpenAILLMContext, +) +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.google import GoogleLLMService +from pipecat.processors.frame_processor import FrameProcessor +from pipecat.transports.services.daily import DailyParams, DailyTransport +from pipecat.frames.frames import ( + Frame, + InputAudioRawFrame, + TextFrame, + LLMFullResponseEndFrame, + TranscriptionFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + +conversation_system_message = """ +You are a helpful LLM in a WebRTC call. Your goals are to be helpful and brief in your responses. Respond with one or two sentences at most, unless you are asked to +respond at more length. Your output will be converted to audio so don't include special characters in your answers. +""" + +transcriber_system_message = """ +You are an audio transcriber. You are receiving audio from a user. Your job is to +transcribe the input audio to text exactly as it was said by the user.. + +You will receive the full conversation history before the audio input, to help with context. Use the full history only to help improve the accuracy of your transcription. + +Rules: + - Respond with an exact transcription of the audio input. + - Do not include any text other than the transcription. + - Do not explain or add to your response. + - Transcribe the audio input simply and precisely. + - If the audio is not clear, emit the special string "EMPTY". + - No response other than exact transcription, or "EMPTY", is allowed. +""" + + +class UserAudioCollector(FrameProcessor): + def __init__(self, context, user_context_aggregator): + super().__init__() + self._context = context + self._user_context_aggregator = user_context_aggregator + self._audio_frames = [] + self._start_secs = 0.2 # this should match VAD start_secs (hardcoding for now) + self._user_speaking = False + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + + if isinstance(frame, TranscriptionFrame): + # We could gracefully handle both audio input and text/transcription input ... + # but let's leave that as an exercise to the reader. :-) + return + if isinstance(frame, UserStartedSpeakingFrame): + self._user_speaking = True + elif isinstance(frame, UserStoppedSpeakingFrame): + self._user_speaking = False + self._context.add_audio_frames_message(audio_frames=self._audio_frames) + await self._user_context_aggregator.push_frame( + self._user_context_aggregator.get_context_frame() + ) + elif isinstance(frame, InputAudioRawFrame): + if self._user_speaking: + self._audio_frames.append(frame) + else: + # Append the audio frame to our buffer. Treat the buffer as a ring buffer, dropping the oldest + # frames as necessary. Assume all audio frames have the same duration. + self._audio_frames.append(frame) + frame_duration = len(frame.audio) / 16 * frame.num_channels / frame.sample_rate + buffer_duration = frame_duration * len(self._audio_frames) + while buffer_duration > self._start_secs: + self._audio_frames.pop(0) + buffer_duration -= frame_duration + + await self.push_frame(frame, direction) + + +class InputTranscriptionContextFilter(FrameProcessor): + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + # todo: make sure the most recent context message is audio input. + + +@dataclass +class MagicDemoTranscriptionFrame(Frame): + text: str + + +class InputTranscriptionFrameEmitter(FrameProcessor): + def __init__(self): + super().__init__() + self._aggregation = "" + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + + if isinstance(frame, TextFrame): + self._aggregation += frame.text + elif isinstance(frame, LLMFullResponseEndFrame): + logger.debug(f"TRANSCRIPTION: {self._aggregation}") + await self.push_frame(MagicDemoTranscriptionFrame(text=self._aggregation.strip())) + self._aggregation = "" + + +class TranscriptionContextFixup(FrameProcessor): + def __init__(self, context): + super().__init__() + self._context = context + self._transcript = "THIS IS A TRANSCRIPT" + + def swap_user_audio(self): + if not self._transcript: + return + message = self._context.messages[-2] + last_part = message.parts[-1] + if ( + message.role == "user" + and last_part.inline_data + and last_part.inline_data.mime_type == "audio/wav" + ): + self._context.messages[-2] = glm.Content( + role="user", parts=[glm.Part(text=self._transcript)] + ) + + async def process_frame(self, frame, direction): + await super().process_frame(frame, direction) + + if isinstance(frame, TranscriptionFrame): + # Assume for this demo that we are only transcribing a single user's input, and that + # all transcription arrives in a single frame for each turn. + self._transcript = frame.text + self.swap_user_audio() + self._transcript = "" + + await self.push_frame(frame, direction) + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + # No transcription at all. just audio input to Gemini! + # transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + ), + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + conversation_llm = GoogleLLMService( + model="gemini-1.5-flash-latest", + # model="gemini-exp-1114", + api_key=os.getenv("GOOGLE_API_KEY"), + # we can give the GoogleLLMService a system instruction to use directly + # in the GenerativeModel constructor. Let's do that rather than put + # our system message in the messages list. + system_instruction=conversation_system_message, + ) + + input_transcription_llm = GoogleLLMService( + model="gemini-1.5-flash-latest", + # model="gemini-exp-1114", + api_key=os.getenv("GOOGLE_API_KEY"), + system_instruction=transcriber_system_message, + ) + + messages = [ + { + "role": "user", + "content": "Start by saying hello.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = conversation_llm.create_context_aggregator(context) + audio_collector = UserAudioCollector(context, context_aggregator.user()) + transcription_frames_emitter = InputTranscriptionFrameEmitter() + fixup_context_messages = TranscriptionContextFixup(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + audio_collector, + context_aggregator.user(), # User responses + ParallelPipeline( + [ # transcribe + # input_transcription_context_filter, + input_transcription_llm, + transcription_frames_emitter, + ], + [ # conversation inference + conversation_llm, + ], + ), + tts, + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + # fixup_context_messages, + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + # Kick off the conversation. + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index be8d80d1f..86111beb0 100644 --- a/src/pipecat/services/google.py +++ b/src/pipecat/services/google.py @@ -568,10 +568,12 @@ class GoogleLLMService(LLMService): async def _process_context(self, context: OpenAILLMContext): await self.push_frame(LLMFullResponseStartFrame()) try: - logger.debug(f"Generating chat: {context.get_messages_for_logging()}") + logger.debug( + f"Generating chat: {self._system_instruction} | {context.get_messages_for_logging()}" + ) messages = context.messages - if self._system_instruction != context.system_message: + if context.system_message and self._system_instruction != context.system_message: logger.debug(f"System instruction changed: {context.system_message}") self._system_instruction = context.system_message self._create_client()