Merge pull request #687 from pipecat-ai/aleix/transport-audio-mixers

introduce transport audio mixers

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.
@@ -44,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.
 
@@ -57,7 +64,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.
 

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,

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,

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,

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

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):

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" ]

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

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

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

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")
 

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,
@@ -28,6 +30,7 @@ from pipecat.frames.frames import (
     StartInterruptionFrame,
     StopInterruptionFrame,
     SystemFrame,
+    TTSAudioRawFrame,
     TransportMessageFrame,
     TransportMessageUrgentFrame,
 )
@@ -72,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
@@ -128,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)
@@ -174,9 +183,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,
@@ -384,33 +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:
-            try:
+            async for frame in self._next_audio_frame():
-                # If we don't have an audio frame for VAD stop secs we will
+                # Notify the bot started speaking upstream if necessary and that
-                # consider the bot is not speaking.
+                # it's actually speaking.
-                frame = await asyncio.wait_for(self._audio_out_queue.get(), timeout=wait_time)
+                if isinstance(frame, TTSAudioRawFrame):
+                    await self._bot_started_speaking()
+                    await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
 
-                # Notify the bot started speaking upstream if necessary.
+                # Also, push frame downstream in case anyone else needs it.
-                await self._bot_started_speaking()
+                await self.push_frame(frame)
 
                 # Send audio.
                 await self.write_raw_audio_frames(frame.audio)
-
+        except asyncio.CancelledError:
-                # Notify the bot is speaking upstream.
+            pass
-                await self.push_frame(BotSpeakingFrame(), FrameDirection.UPSTREAM)
+        except Exception as e:
-
+            logger.exception(f"{self} error writing to microphone: {e}")
-                # 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}")

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

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
+        frame = AudioRawFrame(
-        while len(self._websocket_audio_buffer):
+            audio=frames,
-            frame = AudioRawFrame(
+            sample_rate=self._params.audio_out_sample_rate,
-                audio=self._websocket_audio_buffer[: self._params.audio_frame_size],
+            num_channels=self._params.audio_out_channels,
-                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:
+        payload = self._params.serializer.serialize(frame)
-                content = io.BytesIO()
+        if payload and self._websocket.client_state == WebSocketState.CONNECTED:
-                ww = wave.open(content, "wb")
+            await self._websocket.send_text(payload)
-                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)
+        # Simulate a clock.
-            if payload and self._websocket.client_state == WebSocketState.CONNECTED:
+        current_time = time.monotonic()
-                await self._websocket.send_text(payload)
+        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._websocket_audio_buffer = bytes()
-                self._params.audio_frame_size :
-            ]
 
     async def _write_frame(self, frame: Frame):
         payload = self._params.serializer.serialize(frame)

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
+        frame = AudioRawFrame(
-        while len(self._websocket_audio_buffer) >= self._params.audio_frame_size:
+            audio=frames,
-            frame = AudioRawFrame(
+            sample_rate=self._params.audio_out_sample_rate,
-                audio=self._websocket_audio_buffer[: self._params.audio_frame_size],
+            num_channels=self._params.audio_out_channels,
-                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:
+        proto = self._params.serializer.serialize(frame)
-                content = io.BytesIO()
+        if proto:
-                ww = wave.open(content, "wb")
+            await self._websocket.send(proto)
-                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)
+        # Simulate a clock.
-            if proto:
+        current_time = time.monotonic()
-                await self._websocket.send(proto)
+        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._websocket_audio_buffer = bytes()
-                self._params.audio_frame_size :
-            ]
 
 
 class WebsocketServerTransport(BaseTransport):