From e2b1b56e868ed12ac5b9e09fb0a17a68ff6885ae Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 31 Oct 2024 15:50:47 -0700 Subject: [PATCH 1/8] examples: don't require room token if using an STT --- examples/foundational/07c-interruptible-deepgram.py | 4 ++-- examples/foundational/07m-interruptible-aws.py | 4 ++-- examples/foundational/07n-interruptible-google.py | 4 ++-- 3 files changed, 6 insertions(+), 6 deletions(-) diff --git a/examples/foundational/07c-interruptible-deepgram.py b/examples/foundational/07c-interruptible-deepgram.py index 36bdb320a..e52ca370e 100644 --- a/examples/foundational/07c-interruptible-deepgram.py +++ b/examples/foundational/07c-interruptible-deepgram.py @@ -31,11 +31,11 @@ logger.add(sys.stderr, level="DEBUG") async def main(): async with aiohttp.ClientSession() as session: - (room_url, token) = await configure(session) + (room_url, _) = await configure(session) transport = DailyTransport( room_url, - token, + None, "Respond bot", DailyParams( audio_out_enabled=True, diff --git a/examples/foundational/07m-interruptible-aws.py b/examples/foundational/07m-interruptible-aws.py index 175f8864f..e64a9c6f8 100644 --- a/examples/foundational/07m-interruptible-aws.py +++ b/examples/foundational/07m-interruptible-aws.py @@ -32,11 +32,11 @@ logger.add(sys.stderr, level="DEBUG") async def main(): async with aiohttp.ClientSession() as session: - (room_url, token) = await configure(session) + (room_url, _) = await configure(session) transport = DailyTransport( room_url, - token, + None, "Respond bot", DailyParams( audio_out_enabled=True, diff --git a/examples/foundational/07n-interruptible-google.py b/examples/foundational/07n-interruptible-google.py index b5524b1fa..c0b29715e 100644 --- a/examples/foundational/07n-interruptible-google.py +++ b/examples/foundational/07n-interruptible-google.py @@ -32,11 +32,11 @@ logger.add(sys.stderr, level="DEBUG") async def main(): async with aiohttp.ClientSession() as session: - (room_url, token) = await configure(session) + (room_url, _) = await configure(session) transport = DailyTransport( room_url, - token, + None, "Respond bot", DailyParams( audio_out_enabled=True, From d2401a76c88a74b6741d7f09c130412e73328f63 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 31 Oct 2024 15:40:35 -0700 Subject: [PATCH 2/8] base_output: only generate bot speaking with TTS audio frames --- src/pipecat/transports/base_output.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index f3dd225c1..3398c48d7 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -28,6 +28,7 @@ from pipecat.frames.frames import ( StartInterruptionFrame, StopInterruptionFrame, SystemFrame, + TTSAudioRawFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) @@ -397,13 +398,15 @@ class BaseOutputTransport(FrameProcessor): frame = await asyncio.wait_for(self._audio_out_queue.get(), timeout=wait_time) # Notify the bot started speaking upstream if necessary. - await self._bot_started_speaking() + if isinstance(frame, TTSAudioRawFrame): + await self._bot_started_speaking() # Send audio. await self.write_raw_audio_frames(frame.audio) # Notify the bot is speaking upstream. - await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) + if isinstance(frame, TTSAudioRawFrame): + await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) # Push frame downstream in case anyone else needs it. await self.push_frame(frame) From 94062592efc46d5e2e54839fa80e918a65851a97 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 31 Oct 2024 15:41:01 -0700 Subject: [PATCH 3/8] base_output: generate smaller audio frames of the same incoming type --- src/pipecat/transports/base_output.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 3398c48d7..e604fd289 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -175,9 +175,10 @@ class BaseOutputTransport(FrameProcessor): if self._params.audio_out_is_live: await self._audio_out_queue.put(frame) else: + cls = type(frame) self._audio_buffer.extend(frame.audio) while len(self._audio_buffer) >= self._audio_chunk_size: - chunk = OutputAudioRawFrame( + chunk = cls( bytes(self._audio_buffer[: self._audio_chunk_size]), sample_rate=frame.sample_rate, num_channels=frame.num_channels, From 4455b2a42835e0e9b65acc4f2539baf5b6ed1a75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 31 Oct 2024 15:51:06 -0700 Subject: [PATCH 4/8] rtvi: create queues before tasks --- src/pipecat/processors/frameworks/rtvi.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 4ed63b723..af893dcd0 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -590,12 +590,12 @@ class RTVIProcessor(FrameProcessor): self._registered_services: Dict[str, RTVIService] = {} # A task to process incoming action frames. - self._action_task = self.get_event_loop().create_task(self._action_task_handler()) self._action_queue = asyncio.Queue() + self._action_task = self.get_event_loop().create_task(self._action_task_handler()) # A task to process incoming transport messages. - self._message_task = self.get_event_loop().create_task(self._message_task_handler()) self._message_queue = asyncio.Queue() + self._message_task = self.get_event_loop().create_task(self._message_task_handler()) self._register_event_handler("on_bot_ready") From 81c476dd4cd3d9c782080c1ecdbaef9c4f536045 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 1 Nov 2024 23:04:02 -0700 Subject: [PATCH 5/8] introduce output transport audio mixers --- CHANGELOG.md | 4 + pyproject.toml | 1 + src/pipecat/audio/mixers/__init__.py | 0 src/pipecat/audio/mixers/base_audio_mixer.py | 53 +++++++ src/pipecat/audio/mixers/soundfile_mixer.py | 140 +++++++++++++++++++ src/pipecat/frames/frames.py | 25 +++- src/pipecat/transports/base_output.py | 87 +++++++++--- src/pipecat/transports/base_transport.py | 3 + 8 files changed, 289 insertions(+), 24 deletions(-) create mode 100644 src/pipecat/audio/mixers/__init__.py create mode 100644 src/pipecat/audio/mixers/base_audio_mixer.py create mode 100644 src/pipecat/audio/mixers/soundfile_mixer.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 98cab2a05..dd56eb011 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Introduce output transport audio mixers. Output transport audio mixers can be + used, for example, to add background sounds or any other audio mixing + functionality before the output audio is actually written to the transport. + - Added `GatedOpenAILLMContextAggregator`. This aggregator keeps the last received OpenAI LLM context frame and it doesn't let it through until the notifier is notified. diff --git a/pyproject.toml b/pyproject.toml index d0422dacd..115822cbf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -60,6 +60,7 @@ openai = [ "openai~=1.50.2", "websockets~=13.1", "python-deepcompare~=1.0.1" ] openpipe = [ "openpipe~=4.24.0" ] playht = [ "pyht~=0.1.4", "websockets~=13.1" ] silero = [ "onnxruntime~=1.19.2" ] +soundfile = [ "soundfile~=0.12.1" ] together = [ "openai~=1.50.2" ] websocket = [ "websockets~=13.1", "fastapi~=0.115.0" ] whisper = [ "faster-whisper~=1.0.3" ] diff --git a/src/pipecat/audio/mixers/__init__.py b/src/pipecat/audio/mixers/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/audio/mixers/base_audio_mixer.py b/src/pipecat/audio/mixers/base_audio_mixer.py new file mode 100644 index 000000000..bd9d03b89 --- /dev/null +++ b/src/pipecat/audio/mixers/base_audio_mixer.py @@ -0,0 +1,53 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +from abc import ABC, abstractmethod + +from pipecat.frames.frames import Frame + + +class BaseAudioMixer(ABC): + """This is a base class for output transport audio mixers. If an audio mixer + is provided to the output transport it will be used to mix the audio frames + coming into to the transport with the audio generated from the mixer. There + are control frames to update mixer settings or to enable or disable the + mixer at runtime. + + """ + + @abstractmethod + async def start(self, sample_rate: int): + """This will be called from the output transport when the transport is + started. It can be used to initialize the mixer. The output transport + sample rate is provided so the mixer can adjust to that sample rate. + + """ + pass + + @abstractmethod + async def stop(self): + """This will be called from the output transport when the transport is + stopping. + + """ + pass + + @abstractmethod + async def process_frame(self, frame: Frame): + """This will be called when the output transport receives a + MixerControlFrame. + + """ + pass + + @abstractmethod + async def mix(self, audio: bytes) -> bytes: + """This is called with the audio that is about to be sent from the + output transport and that should be mixed with the mixer audio if the + mixer is enabled. + + """ + pass diff --git a/src/pipecat/audio/mixers/soundfile_mixer.py b/src/pipecat/audio/mixers/soundfile_mixer.py new file mode 100644 index 000000000..a9a17a106 --- /dev/null +++ b/src/pipecat/audio/mixers/soundfile_mixer.py @@ -0,0 +1,140 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio + +from typing import Any, Dict, Mapping + +import numpy as np + +from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer +from pipecat.audio.utils import resample_audio +from pipecat.frames.frames import Frame, MixerUpdateSettingsFrame, MixerEnableFrame + +from loguru import logger + +try: + import soundfile as sf +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use the soundfile mixer, you need to `pip install pipecat-ai[soundfile]`." + ) + raise Exception(f"Missing module: {e}") + + +class SoundfileMixer(BaseAudioMixer): + """This is an audio mixer that mixes incoming audio with audio from a + file. It uses the soundfile library to load files so it supports multiple + formats. The audio files need to only have one channel (mono) but they can + have any sample rate that will be resampled to the output transport sample + rate. + + Multiple files can be loaded, each with a different name. The + `MixerUpdateSettingsFrame` has the following settings available: `sound` + (str) and `volume` (float) to be able to update to a different sound file or + to change the volume at runtime. + + """ + + def __init__( + self, + sound_files: Mapping[str, str], + default_sound: str, + volume: float = 0.4, + **kwargs, + ): + super().__init__(**kwargs) + self._sound_files = sound_files + self._volume = volume + self._sample_rate = 0 + + self._sound_pos = 0 + self._sounds: Dict[str, Any] = {} + self._current_sound = default_sound + self._mixing = True + + async def start(self, sample_rate: int): + self._sample_rate = sample_rate + for sound_name, file_name in self._sound_files.items(): + await asyncio.to_thread(self._load_sound_file, sound_name, file_name) + + async def stop(self): + pass + + async def process_frame(self, frame: Frame): + if isinstance(frame, MixerUpdateSettingsFrame): + await self._update_settings(frame) + elif isinstance(frame, MixerEnableFrame): + await self._enable_mixing(frame.enable) + pass + + async def mix(self, audio: bytes) -> bytes: + return self._mix_with_sound(audio) + + async def _enable_mixing(self, enable: bool): + self._mixing = enable + + async def _update_settings(self, frame: MixerUpdateSettingsFrame): + for setting, value in frame.settings.items(): + match setting: + case "sound": + await self._change_sound(value) + case "volume": + await self._update_volume(value) + + async def _change_sound(self, sound: str): + if sound in self._sound_files: + self._current_sound = sound + self._sound_pos = 0 + else: + logger.error(f"Sound {sound} is not available") + + async def _update_volume(self, volume: float): + self._volume = volume + + def _load_sound_file(self, sound_name: str, file_name: str): + try: + logger.debug(f"Loading background sound from {file_name}") + sound, sample_rate = sf.read(file_name, dtype="int16") + + audio = sound.tobytes() + if sample_rate != self._sample_rate: + logger.debug(f"Resampling background sound to {self._sample_rate}") + audio = resample_audio(audio, sample_rate, self._sample_rate) + + # Convert from np to bytes again. + self._sounds[sound_name] = np.frombuffer(audio, dtype=np.int16) + except Exception as ex: + logger.error(f"Unable to open file {file_name}") + + def _mix_with_sound(self, audio: bytes): + """Mixes raw audio frames with chunks of the same length from the sound + file. + + """ + if not self._mixing: + return audio + + audio_np = np.frombuffer(audio, dtype=np.int16) + chunk_size = len(audio_np) + + # Sound currently playing. + sound = self._sounds[self._current_sound] + + # Go back to the beginning if we don't have enough data. + if self._sound_pos + chunk_size > len(sound): + self._sound_pos = 0 + + start_pos = self._sound_pos + end_pos = self._sound_pos + chunk_size + self._sound_pos = end_pos + + sound_np = sound[start_pos:end_pos] + + mixed_audio = np.clip(audio_np + sound_np * self._volume, -32768, 32767).astype(np.int16) + + return mixed_audio.astype(np.int16).tobytes() diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 517918a52..faa029635 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -5,7 +5,7 @@ # from dataclasses import dataclass, field -from typing import Any, Dict, List, Optional, Tuple +from typing import Any, List, Mapping, Optional, Tuple from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.clocks.base_clock import BaseClock @@ -557,7 +557,7 @@ class TTSStoppedFrame(ControlFrame): class ServiceUpdateSettingsFrame(ControlFrame): """A control frame containing a request to update service settings.""" - settings: Dict[str, Any] + settings: Mapping[str, Any] @dataclass @@ -582,3 +582,24 @@ class VADParamsUpdateFrame(ControlFrame): """ params: VADParams + + +@dataclass +class MixerControlFrame(ControlFrame): + """Base control frame for other mixer frames.""" + + pass + + +@dataclass +class MixerUpdateSettingsFrame(MixerControlFrame): + """Control frame to update mixer settings.""" + + settings: Mapping[str, Any] + + +@dataclass +class MixerEnableFrame(MixerControlFrame): + """Control frame to enable or disable the mixer at runtime.""" + + enable: bool diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index e604fd289..15091255d 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -8,19 +8,21 @@ import asyncio import itertools import sys import time -from typing import List +from typing import AsyncGenerator, List from loguru import logger from PIL import Image from pipecat.audio.vad.vad_analyzer import VAD_STOP_SECS from pipecat.frames.frames import ( + AudioRawFrame, BotSpeakingFrame, BotStartedSpeakingFrame, BotStoppedSpeakingFrame, CancelFrame, EndFrame, Frame, + MixerControlFrame, OutputAudioRawFrame, OutputImageRawFrame, SpriteFrame, @@ -73,11 +75,15 @@ class BaseOutputTransport(FrameProcessor): self._bot_speaking = False async def start(self, frame: StartFrame): + if self._params.audio_out_mixer: + await self._params.audio_out_mixer.start(self._params.audio_out_sample_rate) self._create_output_tasks() self._create_sink_tasks() async def stop(self, frame: EndFrame): await self._cancel_output_tasks() + if self._params.audio_out_mixer: + await self._params.audio_out_mixer.stop() async def cancel(self, frame: CancelFrame): # Since we are cancelling everything it doesn't matter if we cancel sink @@ -129,6 +135,8 @@ class BaseOutputTransport(FrameProcessor): await self.stop(frame) # We finally push EndFrame down so PipelineTask stops nicely. await self.push_frame(frame, direction) + elif isinstance(frame, MixerControlFrame) and self._params.audio_out_mixer: + await self._params.audio_out_mixer.process_frame(frame) # Other frames. elif isinstance(frame, OutputAudioRawFrame): await self._handle_audio(frame) @@ -386,35 +394,70 @@ class BaseOutputTransport(FrameProcessor): async def send_audio(self, frame: OutputAudioRawFrame): await self.process_frame(frame, FrameDirection.DOWNSTREAM) + def _next_audio_frame(self) -> AsyncGenerator[AudioRawFrame, None]: + async def without_mixer(vad_stop_secs: float) -> AsyncGenerator[AudioRawFrame, None]: + while True: + try: + frame = await asyncio.wait_for( + self._audio_out_queue.get(), timeout=vad_stop_secs + ) + yield frame + except asyncio.TimeoutError: + # Notify the bot stopped speaking upstream if necessary. + await self._bot_stopped_speaking() + + async def with_mixer(vad_stop_secs: float) -> AsyncGenerator[AudioRawFrame, None]: + last_frame_time = 0 + silence = b"\x00" * self._audio_chunk_size + while True: + try: + frame = self._audio_out_queue.get_nowait() + frame.audio = await self._params.audio_out_mixer.mix(frame.audio) + last_frame_time = time.time() + yield frame + except asyncio.QueueEmpty: + # Notify the bot stopped speaking upstream if necessary. + diff_time = time.time() - last_frame_time + if diff_time > vad_stop_secs: + await self._bot_stopped_speaking() + # Generate an audio frame with only the mixer's part. + frame = OutputAudioRawFrame( + audio=await self._params.audio_out_mixer.mix(silence), + sample_rate=self._params.audio_out_sample_rate, + num_channels=self._params.audio_out_channels, + ) + yield frame + + vad_stop_secs = ( + self._params.vad_analyzer.params.stop_secs + if self._params.vad_analyzer + else VAD_STOP_SECS + ) + if self._params.audio_out_mixer: + return with_mixer(vad_stop_secs) + else: + return without_mixer(vad_stop_secs) + async def _audio_out_task_handler(self): wait_time = ( self._params.vad_analyzer.params.stop_secs if self._params.vad_analyzer else VAD_STOP_SECS ) - while True: - try: - # If we don't have an audio frame for VAD stop secs we will - # consider the bot is not speaking. - frame = await asyncio.wait_for(self._audio_out_queue.get(), timeout=wait_time) - - # Notify the bot started speaking upstream if necessary. + try: + async for frame in self._next_audio_frame(): + # Notify the bot started speaking upstream if necessary and that + # it's actually speaking. if isinstance(frame, TTSAudioRawFrame): await self._bot_started_speaking() + await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) + + # Also, push frame downstream in case anyone else needs it. + await self.push_frame(frame) # Send audio. await self.write_raw_audio_frames(frame.audio) - - # Notify the bot is speaking upstream. - if isinstance(frame, TTSAudioRawFrame): - await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM) - - # Push frame downstream in case anyone else needs it. - await self.push_frame(frame) - except asyncio.TimeoutError: - # Notify the bot stopped speaking upstream if necessary. - await self._bot_stopped_speaking() - except asyncio.CancelledError: - break - except Exception as e: - logger.exception(f"{self} error writing to camera: {e}") + except asyncio.CancelledError: + pass + except Exception as e: + logger.exception(f"{self} error writing to microphone: {e}") diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 2ca2b6470..946a8fb90 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -8,10 +8,12 @@ import asyncio import inspect from abc import ABC, abstractmethod +from typing import Optional from pydantic import ConfigDict from pydantic.main import BaseModel +from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.processors.frame_processor import FrameProcessor @@ -33,6 +35,7 @@ class TransportParams(BaseModel): audio_out_sample_rate: int = 24000 audio_out_channels: int = 1 audio_out_bitrate: int = 96000 + audio_out_mixer: Optional[BaseAudioMixer] = None audio_in_enabled: bool = False audio_in_sample_rate: int = 16000 audio_in_channels: int = 1 From a9ef5ca95d1d36e8d6ed09a0ec72ca821ab6d683 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 31 Oct 2024 15:47:32 -0700 Subject: [PATCH 6/8] examples: add bot background sound example --- CHANGELOG.md | 4 +- .../foundational/23-bot-background-sound.py | 121 ++++++++++++++++++ 2 files changed, 124 insertions(+), 1 deletion(-) create mode 100644 examples/foundational/23-bot-background-sound.py diff --git a/CHANGELOG.md b/CHANGELOG.md index dd56eb011..eb344aa45 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -61,7 +61,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Other -- Added a new foundational example 22-natural-conversation.py. This examples +- Add `23-bot-background-sound.py` foundational example. + +- Added a new foundational example `22-natural-conversation.py`. This example shows how to achieve a more natural conversation detecting when the user ends statement. diff --git a/examples/foundational/23-bot-background-sound.py b/examples/foundational/23-bot-background-sound.py new file mode 100644 index 000000000..68dd700d5 --- /dev/null +++ b/examples/foundational/23-bot-background-sound.py @@ -0,0 +1,121 @@ +# +# Copyright (c) 2024, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import asyncio +import aiohttp +import os +import sys + +from pipecat.audio.mixers.soundfile_mixer import SoundfileMixer +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import LLMMessagesFrame, MixerUpdateSettingsFrame, MixerEnableFrame +from pipecat.pipeline.pipeline import Pipeline +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.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport + +from runner import configure_with_args + +from loguru import logger + +from dotenv import load_dotenv + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(): + async with aiohttp.ClientSession() as session: + parser = argparse.ArgumentParser(description="Bot Background Sound") + parser.add_argument("-i", "--input", type=str, required=True, help="Input audio file") + + (room_url, token, args) = await configure_with_args(session, parser) + + soundfile_mixer = SoundfileMixer( + sound_files={"office": args.input}, + default_sound="office", + volume=2.0, + ) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + audio_out_mixer=soundfile_mixer, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC 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.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + # Show how to use mixer control frames. + await asyncio.sleep(10.0) + await task.queue_frame(MixerUpdateSettingsFrame({"volume": 0.5})) + await asyncio.sleep(5.0) + await task.queue_frame(MixerEnableFrame(False)) + await asyncio.sleep(5.0) + await task.queue_frame(MixerEnableFrame(True)) + await asyncio.sleep(5.0) + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([LLMMessagesFrame(messages)]) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) From 0ac9e2dd3fca5598d101b62558eae42f4f8f6861 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 4 Nov 2024 13:04:18 -0800 Subject: [PATCH 7/8] transports(network): synchronize with time before sending data --- CHANGELOG.md | 3 + .../transports/network/fastapi_websocket.py | 62 +++++++++-------- .../transports/network/websocket_server.py | 69 ++++++++++++------- 3 files changed, 81 insertions(+), 53 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index eb344aa45..9b6cdcab1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -48,6 +48,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Websocket transports (FastAPI and Websocket) now synchronize with time before + sending data. This allows for interruptions to just work out of the box. + - Improved bot speaking detection for all TTS services by using actual bot audio. diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index dac162530..1bef71dc1 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -7,6 +7,7 @@ import asyncio import io +import time import wave from typing import Awaitable, Callable @@ -42,7 +43,6 @@ except ModuleNotFoundError as e: class FastAPIWebsocketParams(TransportParams): add_wav_header: bool = False - audio_frame_size: int = 6400 # 200ms serializer: FrameSerializer @@ -105,44 +105,52 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): self._websocket = websocket self._params = params - self._websocket_audio_buffer = bytes() + + self._send_interval = (self._audio_chunk_size / self._params.audio_out_sample_rate) / 2 + self._next_send_time = 0 async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, StartInterruptionFrame): await self._write_frame(frame) + self._next_send_time = 0 async def write_raw_audio_frames(self, frames: bytes): - self._websocket_audio_buffer += frames - while len(self._websocket_audio_buffer): - frame = AudioRawFrame( - audio=self._websocket_audio_buffer[: self._params.audio_frame_size], - sample_rate=self._params.audio_out_sample_rate, - num_channels=self._params.audio_out_channels, + frame = AudioRawFrame( + audio=frames, + sample_rate=self._params.audio_out_sample_rate, + num_channels=self._params.audio_out_channels, + ) + + if self._params.add_wav_header: + content = io.BytesIO() + ww = wave.open(content, "wb") + ww.setsampwidth(2) + ww.setnchannels(frame.num_channels) + ww.setframerate(frame.sample_rate) + ww.writeframes(frame.audio) + ww.close() + content.seek(0) + wav_frame = AudioRawFrame( + content.read(), sample_rate=frame.sample_rate, num_channels=frame.num_channels ) + frame = wav_frame - if self._params.add_wav_header: - content = io.BytesIO() - ww = wave.open(content, "wb") - ww.setsampwidth(2) - ww.setnchannels(frame.num_channels) - ww.setframerate(frame.sample_rate) - ww.writeframes(frame.audio) - ww.close() - content.seek(0) - wav_frame = AudioRawFrame( - content.read(), sample_rate=frame.sample_rate, num_channels=frame.num_channels - ) - frame = wav_frame + payload = self._params.serializer.serialize(frame) + if payload and self._websocket.client_state == WebSocketState.CONNECTED: + await self._websocket.send_text(payload) - payload = self._params.serializer.serialize(frame) - if payload and self._websocket.client_state == WebSocketState.CONNECTED: - await self._websocket.send_text(payload) + # Simulate a clock. + current_time = time.monotonic() + sleep_duration = max(0, self._next_send_time - current_time) + await asyncio.sleep(sleep_duration) + if sleep_duration == 0: + self._next_send_time = time.monotonic() + self._send_interval + else: + self._next_send_time += self._send_interval - self._websocket_audio_buffer = self._websocket_audio_buffer[ - self._params.audio_frame_size : - ] + self._websocket_audio_buffer = bytes() async def _write_frame(self, frame: Frame): payload = self._params.serializer.serialize(frame) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index b5d38f60e..c0c8595e9 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -6,6 +6,7 @@ import asyncio import io +import time import wave from typing import Awaitable, Callable @@ -15,9 +16,12 @@ from pipecat.frames.frames import ( AudioRawFrame, CancelFrame, EndFrame, + Frame, InputAudioRawFrame, StartFrame, + StartInterruptionFrame, ) +from pipecat.processors.frame_processor import FrameDirection from pipecat.serializers.base_serializer import FrameSerializer from pipecat.serializers.protobuf import ProtobufFrameSerializer from pipecat.transports.base_input import BaseInputTransport @@ -36,7 +40,6 @@ except ModuleNotFoundError as e: class WebsocketServerParams(TransportParams): add_wav_header: bool = False - audio_frame_size: int = 6400 # 200ms serializer: FrameSerializer = ProtobufFrameSerializer() @@ -132,45 +135,59 @@ class WebsocketServerOutputTransport(BaseOutputTransport): self._websocket_audio_buffer = bytes() + self._send_interval = (self._audio_chunk_size / self._params.audio_out_sample_rate) / 2 + self._next_send_time = 0 + async def set_client_connection(self, websocket: websockets.WebSocketServerProtocol | None): if self._websocket: await self._websocket.close() logger.warning("Only one client allowed, using new connection") self._websocket = websocket + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + if isinstance(frame, StartInterruptionFrame): + self._next_send_time = 0 + async def write_raw_audio_frames(self, frames: bytes): if not self._websocket: return - self._websocket_audio_buffer += frames - while len(self._websocket_audio_buffer) >= self._params.audio_frame_size: - frame = AudioRawFrame( - audio=self._websocket_audio_buffer[: self._params.audio_frame_size], - sample_rate=self._params.audio_out_sample_rate, - num_channels=self._params.audio_out_channels, + frame = AudioRawFrame( + audio=frames, + sample_rate=self._params.audio_out_sample_rate, + num_channels=self._params.audio_out_channels, + ) + + if self._params.add_wav_header: + content = io.BytesIO() + ww = wave.open(content, "wb") + ww.setsampwidth(2) + ww.setnchannels(frame.num_channels) + ww.setframerate(frame.sample_rate) + ww.writeframes(frame.audio) + ww.close() + content.seek(0) + wav_frame = AudioRawFrame( + content.read(), sample_rate=frame.sample_rate, num_channels=frame.num_channels ) + frame = wav_frame - if self._params.add_wav_header: - content = io.BytesIO() - ww = wave.open(content, "wb") - ww.setsampwidth(2) - ww.setnchannels(frame.num_channels) - ww.setframerate(frame.sample_rate) - ww.writeframes(frame.audio) - ww.close() - content.seek(0) - wav_frame = AudioRawFrame( - content.read(), sample_rate=frame.sample_rate, num_channels=frame.num_channels - ) - frame = wav_frame + proto = self._params.serializer.serialize(frame) + if proto: + await self._websocket.send(proto) - proto = self._params.serializer.serialize(frame) - if proto: - await self._websocket.send(proto) + # Simulate a clock. + current_time = time.monotonic() + sleep_duration = max(0, self._next_send_time - current_time) + await asyncio.sleep(sleep_duration) + if sleep_duration == 0: + self._next_send_time = time.monotonic() + self._send_interval + else: + self._next_send_time += self._send_interval - self._websocket_audio_buffer = self._websocket_audio_buffer[ - self._params.audio_frame_size : - ] + self._websocket_audio_buffer = bytes() class WebsocketServerTransport(BaseTransport): From 602915ae18b64f0c52a91da5c30e21169c1ed44c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 4 Nov 2024 13:05:02 -0800 Subject: [PATCH 8/8] examples(websocket-server): allow interruptions --- examples/websocket-server/bot.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/examples/websocket-server/bot.py b/examples/websocket-server/bot.py index 2faaea4ad..3f961de8d 100644 --- a/examples/websocket-server/bot.py +++ b/examples/websocket-server/bot.py @@ -12,7 +12,7 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +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.deepgram import DeepgramSTTService @@ -35,6 +35,7 @@ logger.add(sys.stderr, level="DEBUG") async def main(): transport = WebsocketServerTransport( params=WebsocketServerParams( + audio_out_sample_rate=16000, audio_out_enabled=True, add_wav_header=True, vad_enabled=True, @@ -50,6 +51,7 @@ async def main(): tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady + sample_rate=16000, ) messages = [ @@ -74,7 +76,7 @@ async def main(): ] ) - task = PipelineTask(pipeline) + task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) @transport.event_handler("on_client_connected") async def on_client_connected(transport, client):