# # 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())