From cff62650ee16277840744b92e91a5363ba1560e7 Mon Sep 17 00:00:00 2001 From: daniil Date: Wed, 2 Oct 2024 12:51:17 +0300 Subject: [PATCH] add background_noise service and example --- .../11a-sound-effects-background-twilio.py | 100 +++++++++++++++ src/pipecat/services/background_noise.py | 120 ++++++++++++++++++ 2 files changed, 220 insertions(+) create mode 100644 examples/foundational/11a-sound-effects-background-twilio.py create mode 100644 src/pipecat/services/background_noise.py diff --git a/examples/foundational/11a-sound-effects-background-twilio.py b/examples/foundational/11a-sound-effects-background-twilio.py new file mode 100644 index 000000000..79e7aea2c --- /dev/null +++ b/examples/foundational/11a-sound-effects-background-twilio.py @@ -0,0 +1,100 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import os +import sys + +from pipecat.frames.frames import EndFrame, LLMMessagesFrame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.llm_response import ( + LLMAssistantResponseAggregator, + LLMUserResponseAggregator, +) +from pipecat.services.background_noise import BackgroundNoiseEffect +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.services.deepgram import DeepgramSTTService +from pipecat.transports.network.fastapi_websocket import ( + FastAPIWebsocketTransport, + FastAPIWebsocketParams, +) +from pipecat.vad.silero import SileroVADAnalyzer +from pipecat.serializers.twilio import TwilioFrameSerializer + +from loguru import logger + +from dotenv import load_dotenv + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def run_bot(websocket_client, stream_sid): + transport = FastAPIWebsocketTransport( + websocket=websocket_client, + params=FastAPIWebsocketParams( + audio_out_enabled=True, + add_wav_header=False, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, + serializer=TwilioFrameSerializer(stream_sid), + ), + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in an audio call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + tma_in = LLMUserResponseAggregator(messages) + tma_out = LLMAssistantResponseAggregator(messages) + background_noise = BackgroundNoiseEffect(websocket_client, stream_sid, "your_path_to_audio_in_format_pcm16000") + + pipeline = Pipeline( + [ + transport.input(), # Websocket input from client + stt, # Speech-To-Text + tma_in, # User responses + llm, # LLM + tts, # Text-To-Speech + background_noise, + transport.output(), # Websocket output to client + tma_out, # LLM responses + ] + ) + + task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + await task.queue_frames([EndFrame()]) + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) + diff --git a/src/pipecat/services/background_noise.py b/src/pipecat/services/background_noise.py new file mode 100644 index 000000000..8f502fec4 --- /dev/null +++ b/src/pipecat/services/background_noise.py @@ -0,0 +1,120 @@ +import asyncio +import json +import time +from asyncio import sleep +from io import BytesIO + +import loguru + +from pipecat.processors.frame_processor import FrameProcessor, FrameDirection +from pydub import AudioSegment +from pipecat.frames.frames import AudioRawFrame, OutputAudioRawFrame, Frame, BotStartedSpeakingFrame, \ + BotStoppedSpeakingFrame, EndFrame + + +class BackgroundNoiseEffect(FrameProcessor): + def __init__(self, websocket_client, stream_sid, music_path): + super().__init__(sync=False) + self._speaking = True + self._audio_task = self.get_event_loop().create_task(self._audio_task_handler()) + self._audio_queue = asyncio.Queue() + self._stop = False + self.stream_sid = stream_sid + self.websocket_client = websocket_client + self.music_path = music_path + self.get_music_part_gen = self._get_music_part() + self.emptied = False + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, BotStartedSpeakingFrame): + self._speaking = True + + if isinstance(frame, BotStoppedSpeakingFrame): + self._speaking = False + self.emptied = False + + if isinstance(frame, AudioRawFrame) and self._speaking: + if not self.emptied: + self.emptied = True + buffer_clear_message = {"event": "clear", "streamSid": self.stream_sid} + await self.websocket_client.send_text(json.dumps(buffer_clear_message)) + + frame.audio = self._combine_with_music(frame) + + if isinstance(frame, EndFrame): + self._stop = True + + await self.push_frame(frame, direction) + + def _combine_with_music(self, frame: AudioRawFrame): + """ + Combines small raw audio segments from the frame with chunks of a music file. + """ + small_audio_bytes = frame.audio + music_audio = AudioSegment.from_wav(self.music_path) + music_audio = music_audio - 15 + + music_position = 0 + small_audio = AudioSegment( + data=small_audio_bytes, + sample_width=2, + frame_rate=16000, + channels=1 + ) + + small_audio_length = len(small_audio) + music_chunk = music_audio[music_position:music_position + small_audio_length] + + if len(music_chunk) < small_audio_length: + music_position = 0 + music_chunk += music_audio[:small_audio_length - len(music_chunk)] + + combined_audio = music_chunk.overlay(small_audio) + music_position += small_audio_length + + output_buffer = BytesIO() + try: + combined_audio.export(output_buffer, format="raw") + return output_buffer.getvalue() + finally: + output_buffer.close() + + def _get_music_part(self): + """ + Generator that yields chunks of background music audio. + """ + music_audio = AudioSegment.from_wav(self.music_path) + music_audio = music_audio - 15 + + music_position = 0 + small_audio_length = 6400 + + while True: + if music_position + small_audio_length > len(music_audio): + music_chunk = music_audio[music_position:] + music_audio[ + :(music_position + small_audio_length) % len(music_audio)] + music_position = (music_position + small_audio_length) % len(music_audio) + else: + music_chunk = music_audio[music_position:music_position + small_audio_length] + music_position += small_audio_length + + output_buffer = BytesIO() + try: + music_chunk.export(output_buffer, format="raw") + frame = OutputAudioRawFrame(audio=output_buffer.getvalue(), sample_rate=16000, num_channels=1) + yield frame + finally: + output_buffer.close() + + async def _audio_task_handler(self): + while True: + await sleep(0.005) + if self._stop: + break + + if not self._speaking: + frame = next(self.get_music_part_gen) + await self.push_frame(frame, FrameDirection.DOWNSTREAM) +