Merge pull request #770 from pipecat-ai/aleix/system-input-frames-and-audio-buffer-processor

system input frames and audio buffer processor fixes

CHANGELOG.md

@@ -9,19 +9,45 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- In order to obtain the audio stored by the `AudioBufferProcessor` you can now
+  also register an `on_audio_data` event handler. The `on_audio_data` handler
+  will be called every time `buffer_size` (a new constructor argument) is
+  reached. If `buffer_size` is 0 (default) you need to manually get the audio as
+  before using `AudioBufferProcessor.merge_audio_buffers()`.
+
+```
+@audiobuffer.event_handler("on_audio_data")
+async def on_audio_data(processor, audio, sample_rate, num_channels):
+    await save_audio(audio, sample_rate, num_channels)
+```
+
 - Added a new RTVI message called `disconnect-bot`, which when handled pushes
   an `EndFrame` to trigger the pipeline to stop.
 
 ### Changed
 
+- All input frames (text, audio, image, etc.) are now system frames. This means
+  they are processed immediately by all processors instead of being queued
+  internally.
+
 - Expanded the transcriptions.language module to support a superset of
   languages.
 
 - Updated STT and TTS services with language options that match the supported
   languages for each service.
 
+### Removed
+
+- Removed `AppFrame`. This was used as a special user custom frame, but there's
+  actually no use case for that.
+
 ### Fixed
 
+- Fixed an issue where other frames were being processed while a `CancelFrame`
+  was being pushed down the pipeline.
+
+- `AudioBufferProcessor` now handles interruptions properly.
+
 - Fixed a `WebsocketServerTransport` issue that would prevent interruptions with
   `TwilioSerializer` from working.
 

docs/frame.md

@@ -96,9 +96,6 @@ Notable control frames:
 
 ## 7. Special Purpose Frames
 
-### AppFrame
-Base class for application-specific custom frames.
-
 ### MetricsFrame
 Contains performance metrics data.
 

examples/canonical-metrics/bot.py

@@ -102,7 +102,6 @@ async def main():
             audio_buffer_processor=audio_buffer_processor,
             aiohttp_session=session,
             api_key=os.getenv("CANONICAL_API_KEY"),
-            api_url=os.getenv("CANONICAL_API_URL"),
             call_id=str(uuid.uuid4()),
             assistant="pipecat-chatbot",
             assistant_speaks_first=True,

examples/chatbot-audio-recording/bot.py

@@ -4,7 +4,9 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+import aiofiles
 import asyncio
+import io
 import os
 import sys
 
@@ -32,15 +34,17 @@ logger.remove(0)
 logger.add(sys.stderr, level="DEBUG")
 
 
-async def save_audio(audiobuffer):
+async def save_audio(audio: bytes, sample_rate: int, num_channels: int):
-    if audiobuffer.has_audio():
+    if len(audio) > 0:
-        merged_audio = audiobuffer.merge_audio_buffers()
         filename = f"conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav"
-        with wave.open(filename, "wb") as wf:
+        with io.BytesIO() as buffer:
-            wf.setnchannels(2)
+            with wave.open(buffer, "wb") as wf:
-            wf.setsampwidth(2)
+                wf.setsampwidth(2)
-            wf.setframerate(audiobuffer._sample_rate)
+                wf.setnchannels(num_channels)
-            wf.writeframes(merged_audio)
+                wf.setframerate(sample_rate)
+                wf.writeframes(audio)
+            async with aiofiles.open(filename, "wb") as file:
+                await file.write(buffer.getvalue())
         print(f"Merged audio saved to {filename}")
     else:
         print("No audio data to save")
@@ -106,7 +110,9 @@ async def main():
         context = OpenAILLMContext(messages)
         context_aggregator = llm.create_context_aggregator(context)
 
-        audiobuffer = AudioBufferProcessor()
+        # Save audio every 10 seconds.
+        audiobuffer = AudioBufferProcessor(buffer_size=480000)
+
         pipeline = Pipeline(
             [
                 transport.input(),  # microphone
@@ -121,6 +127,10 @@ async def main():
 
         task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True))
 
+        @audiobuffer.event_handler("on_audio_data")
+        async def on_audio_data(buffer, audio, sample_rate, num_channels):
+            await save_audio(audio, sample_rate, num_channels)
+
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
             await transport.capture_participant_transcription(participant["id"])
@@ -130,7 +140,6 @@ async def main():
         async def on_participant_left(transport, participant, reason):
             print(f"Participant left: {participant}")
             await task.queue_frame(EndFrame())
-            await save_audio(audiobuffer)
 
         runner = PipelineRunner()
 

examples/chatbot-audio-recording/requirements.txt

@@ -1,3 +1,4 @@
+aiofiles
 python-dotenv
 fastapi[all]
 uvicorn

examples/foundational/05-sync-speech-and-image.py

@@ -12,7 +12,7 @@ import sys
 from dataclasses import dataclass
 
 from pipecat.frames.frames import (
-    AppFrame,
+    DataFrame,
     Frame,
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
@@ -42,7 +42,7 @@ logger.add(sys.stderr, level="DEBUG")
 
 
 @dataclass
-class MonthFrame(AppFrame):
+class MonthFrame(DataFrame):
     month: str
 
     def __str__(self):

src/pipecat/audio/utils.py

@@ -18,6 +18,37 @@ def resample_audio(audio: bytes, original_rate: int, target_rate: int) -> bytes:
     return resampled_audio.astype(np.int16).tobytes()
 
 
+def mix_audio(audio1: bytes, audio2: bytes) -> bytes:
+    data1 = np.frombuffer(audio1, dtype=np.int16)
+    data2 = np.frombuffer(audio2, dtype=np.int16)
+
+    # Max length
+    max_length = max(len(data1), len(data2))
+
+    # Zero-pad the arrays to the same length
+    padded1 = np.pad(data1, (0, max_length - len(data1)), mode="constant")
+    padded2 = np.pad(data2, (0, max_length - len(data2)), mode="constant")
+
+    # Mix the arrays
+    mixed_audio = padded1.astype(np.int32) + padded2.astype(np.int32)
+    mixed_audio = np.clip(mixed_audio, -32768, 32767).astype(np.int16)
+
+    return mixed_audio.astype(np.int16).tobytes()
+
+
+def interleave_stereo_audio(left_audio: bytes, right_audio: bytes) -> bytes:
+    left = np.frombuffer(left_audio, dtype=np.int16)
+    right = np.frombuffer(right_audio, dtype=np.int16)
+
+    min_length = min(len(left), len(right))
+    left = left[:min_length]
+    right = right[:min_length]
+
+    stereo = np.column_stack((left, right))
+
+    return stereo.astype(np.int16).tobytes()
+
+
 def normalize_value(value, min_value, max_value):
     normalized = (value - min_value) / (max_value - min_value)
     normalized_clamped = max(0, min(1, normalized))

src/pipecat/frames/frames.py

@@ -21,6 +21,8 @@ def format_pts(pts: int | None):
 
 @dataclass
 class Frame:
+    """Base frame class."""
+
     id: int = field(init=False)
     name: str = field(init=False)
     pts: Optional[int] = field(init=False)
@@ -35,17 +37,71 @@ class Frame:
 
 
 @dataclass
-class DataFrame(Frame):
+class SystemFrame(Frame):
+    """System frames are frames that are not internally queued by any of the
+    frame processors and should be processed immediately.
+
+    """
+
     pass
 
 
 @dataclass
-class AudioRawFrame(DataFrame):
+class DataFrame(Frame):
+    """Data frames are frames that will be processed in order and usually
+    contain data such as LLM context, text, audio or images.
+
+    """
+
+    pass
+
+
+@dataclass
+class ControlFrame(Frame):
+    """Control frames are frames that, similar to data frames, will be processed
+    in order and usually contain control information such as frames to update
+    settings or to end the pipeline.
+
+    """
+
+    pass
+
+
+#
+# Mixins
+#
+
+
+@dataclass
+class AudioRawFrame:
     """A chunk of audio."""
 
     audio: bytes
     sample_rate: int
     num_channels: int
+    num_frames: int = field(init=False)
+
+
+@dataclass
+class ImageRawFrame:
+    """A raw image."""
+
+    image: bytes
+    size: Tuple[int, int]
+    format: str | None
+
+
+#
+# Data frames.
+#
+
+
+@dataclass
+class OutputAudioRawFrame(DataFrame, AudioRawFrame):
+    """A chunk of audio. Will be played by the output transport if the
+    transport's microphone has been enabled.
+
+    """
 
     def __post_init__(self):
         super().__post_init__()
@@ -57,20 +113,15 @@ class AudioRawFrame(DataFrame):
 
 
 @dataclass
-class InputAudioRawFrame(AudioRawFrame):
+class OutputImageRawFrame(DataFrame, ImageRawFrame):
-    """A chunk of audio usually coming from an input transport."""
+    """An image that will be shown by the transport if the transport's camera is
-
+    enabled.
-    pass
-
-
-@dataclass
-class OutputAudioRawFrame(AudioRawFrame):
-    """A chunk of audio. Will be played by the output transport if the
-    transport's microphone has been enabled.
 
     """
 
-    pass
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})"
 
 
 @dataclass
@@ -80,64 +131,10 @@ class TTSAudioRawFrame(OutputAudioRawFrame):
     pass
 
 
-@dataclass
-class ImageRawFrame(DataFrame):
-    """An image. Will be shown by the transport if the transport's camera is
-    enabled.
-
-    """
-
-    image: bytes
-    size: Tuple[int, int]
-    format: str | None
-
-    def __str__(self):
-        pts = format_pts(self.pts)
-        return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})"
-
-
-@dataclass
-class InputImageRawFrame(ImageRawFrame):
-    pass
-
-
-@dataclass
-class OutputImageRawFrame(ImageRawFrame):
-    pass
-
-
-@dataclass
-class UserImageRawFrame(InputImageRawFrame):
-    """An image associated to a user. Will be shown by the transport if the
-    transport's camera is enabled.
-
-    """
-
-    user_id: str
-
-    def __str__(self):
-        pts = format_pts(self.pts)
-        return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})"
-
-
-@dataclass
-class VisionImageRawFrame(InputImageRawFrame):
-    """An image with an associated text to ask for a description of it. Will be
-    shown by the transport if the transport's camera is enabled.
-
-    """
-
-    text: str | None
-
-    def __str__(self):
-        pts = format_pts(self.pts)
-        return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})"
-
-
 @dataclass
 class URLImageRawFrame(OutputImageRawFrame):
-    """An image with an associated URL. Will be shown by the transport if the
+    """An output image with an associated URL. These images are usually
-    transport's camera is enabled.
+    generated by third-party services that provide a URL to download the image.
 
     """
 
@@ -149,14 +146,14 @@ class URLImageRawFrame(OutputImageRawFrame):
 
 
 @dataclass
-class SpriteFrame(Frame):
+class SpriteFrame(DataFrame):
     """An animated sprite. Will be shown by the transport if the transport's
     camera is enabled. Will play at the framerate specified in the transport's
     `camera_out_framerate` constructor parameter.
 
     """
 
-    images: List[ImageRawFrame]
+    images: List[OutputImageRawFrame]
 
     def __str__(self):
         pts = format_pts(self.pts)
@@ -166,7 +163,7 @@ class SpriteFrame(Frame):
 @dataclass
 class TextFrame(DataFrame):
     """A chunk of text. Emitted by LLM services, consumed by TTS services, can
-    be used to send text through pipelines.
+    be used to send text through processors.
 
     """
 
@@ -177,41 +174,13 @@ class TextFrame(DataFrame):
         return f"{self.name}(pts: {pts}, text: [{self.text}])"
 
 
-@dataclass
-class TranscriptionFrame(TextFrame):
-    """A text frame with transcription-specific data. Will be placed in the
-    transport's receive queue when a participant speaks.
-
-    """
-
-    user_id: str
-    timestamp: str
-    language: Language | None = None
-
-    def __str__(self):
-        return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})"
-
-
-@dataclass
-class InterimTranscriptionFrame(TextFrame):
-    """A text frame with interim transcription-specific data. Will be placed in
-    the transport's receive queue when a participant speaks."""
-
-    user_id: str
-    timestamp: str
-    language: Language | None = None
-
-    def __str__(self):
-        return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})"
-
-
 @dataclass
 class LLMMessagesFrame(DataFrame):
     """A frame containing a list of LLM messages. Used to signal that an LLM
-    service should run a chat completion and emit an LLMStartFrames, TextFrames
+    service should run a chat completion and emit an LLMFullResponseStartFrame,
-    and an LLMEndFrame. Note that the messages property on this class is
+    TextFrames and an LLMFullResponseStartFrame. Note that the `messages`
-    mutable, and will be be updated by various ResponseAggregator frame
+    property in this class is mutable, and will be be updated by various
-    processors.
+    aggregators.
 
     """
 
@@ -220,7 +189,7 @@ class LLMMessagesFrame(DataFrame):
 
 @dataclass
 class LLMMessagesAppendFrame(DataFrame):
-    """A frame containing a list of LLM messages that neeed to be added to the
+    """A frame containing a list of LLM messages that need to be added to the
     current context.
 
     """
@@ -274,37 +243,11 @@ class TransportMessageFrame(DataFrame):
         return f"{self.name}(message: {self.message})"
 
 
-@dataclass
-class FunctionCallResultFrame(DataFrame):
-    """A frame containing the result of an LLM function (tool) call."""
-
-    function_name: str
-    tool_call_id: str
-    arguments: str
-    result: Any
-    run_llm: bool = True
-
-
-#
-# App frames. Application user-defined frames.
-#
-
-
-@dataclass
-class AppFrame(Frame):
-    pass
-
-
 #
 # System frames
 #
 
 
-@dataclass
-class SystemFrame(Frame):
-    pass
-
-
 @dataclass
 class StartFrame(SystemFrame):
     """This is the first frame that should be pushed down a pipeline."""
@@ -461,14 +404,10 @@ class BotSpeakingFrame(SystemFrame):
 
 
 @dataclass
-class UserImageRequestFrame(SystemFrame):
+class MetricsFrame(SystemFrame):
-    """A frame user to request an image from the given user."""
+    """Emitted by processor that can compute metrics like latencies."""
 
-    user_id: str
+    data: List[MetricsData]
-    context: Optional[Any] = None
-
-    def __str__(self):
-        return f"{self.name}, user: {self.user_id}"
 
 
 @dataclass
@@ -480,6 +419,17 @@ class FunctionCallInProgressFrame(SystemFrame):
     arguments: str
 
 
+@dataclass
+class FunctionCallResultFrame(SystemFrame):
+    """A frame containing the result of an LLM function (tool) call."""
+
+    function_name: str
+    tool_call_id: str
+    arguments: str
+    result: Any
+    run_llm: bool = True
+
+
 @dataclass
 class TransportMessageUrgentFrame(SystemFrame):
     message: Any
@@ -489,10 +439,88 @@ class TransportMessageUrgentFrame(SystemFrame):
 
 
 @dataclass
-class MetricsFrame(SystemFrame):
+class TranscriptionFrame(SystemFrame):
-    """Emitted by processor that can compute metrics like latencies."""
+    """A text frame with transcription-specific data. Will be placed in the
+    transport's receive queue when a participant speaks.
 
-    data: List[MetricsData]
+    """
+
+    text: str
+    user_id: str
+    timestamp: str
+    language: Language | None = None
+
+    def __str__(self):
+        return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})"
+
+
+@dataclass
+class InterimTranscriptionFrame(SystemFrame):
+    """A text frame with interim transcription-specific data. Will be placed in
+    the transport's receive queue when a participant speaks."""
+
+    text: str
+    user_id: str
+    timestamp: str
+    language: Language | None = None
+
+    def __str__(self):
+        return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})"
+
+
+@dataclass
+class UserImageRequestFrame(SystemFrame):
+    """A frame user to request an image from the given user."""
+
+    user_id: str
+    context: Optional[Any] = None
+
+    def __str__(self):
+        return f"{self.name}, user: {self.user_id}"
+
+
+@dataclass
+class InputAudioRawFrame(SystemFrame, AudioRawFrame):
+    """A chunk of audio usually coming from an input transport."""
+
+    def __post_init__(self):
+        super().__post_init__()
+        self.num_frames = int(len(self.audio) / (self.num_channels * 2))
+
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return f"{self.name}(pts: {pts}, size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})"
+
+
+@dataclass
+class InputImageRawFrame(SystemFrame, ImageRawFrame):
+    """An image usually coming from an input transport."""
+
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})"
+
+
+@dataclass
+class UserImageRawFrame(InputImageRawFrame):
+    """An image associated to a user."""
+
+    user_id: str
+
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})"
+
+
+@dataclass
+class VisionImageRawFrame(InputImageRawFrame):
+    """An image with an associated text to ask for a description of it."""
+
+    text: str | None
+
+    def __str__(self):
+        pts = format_pts(self.pts)
+        return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})"
 
 
 #
@@ -500,11 +528,6 @@ class MetricsFrame(SystemFrame):
 #
 
 
-@dataclass
-class ControlFrame(Frame):
-    pass
-
-
 @dataclass
 class EndFrame(ControlFrame):
     """Indicates that a pipeline has ended and frame processors and pipelines

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -4,11 +4,8 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-import wave
+from pipecat.audio.utils import interleave_stereo_audio, mix_audio, resample_audio
-from io import BytesIO
-
 from pipecat.frames.frames import (
-    AudioRawFrame,
     Frame,
     InputAudioRawFrame,
     OutputAudioRawFrame,
@@ -17,84 +14,89 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 
 class AudioBufferProcessor(FrameProcessor):
-    def __init__(self, **kwargs):
+    """This processor buffers audio raw frames (input and output). The mixed
-        """
+    audio can be obtained by calling `get_audio()` (if `buffer_size` is 0) or by
-        Initialize the AudioBufferProcessor.
+    registering an "on_audio_data" event handler. The event handler will be
+    called every time `buffer_size` is reached.
 
-        This constructor sets up the initial state for audio processing:
+    You can provide the desired output `sample_rate` and incoming audio frames
-        - audio_buffer: A bytearray to store incoming audio data.
+    will resampled to match it. Also, you can provide the number of channels, 1
-        - num_channels: The number of audio channels (initialized as None).
+    for mono and 2 for stereo. With mono audio user and bot audio will be mixed,
-        - sample_rate: The sample rate of the audio (initialized as None).
+    in the case of stereo the left channel will be used for the user's audio and
+    the right channel for the bot.
 
-        The num_channels and sample_rate are set to None initially and will be
+    """
-        populated when the first audio frame is processed.
+
-        """
+    def __init__(
+        self, *, sample_rate: int = 24000, num_channels: int = 1, buffer_size: int = 0, **kwargs
+    ):
         super().__init__(**kwargs)
+        self._sample_rate = sample_rate
+        self._num_channels = num_channels
+        self._buffer_size = buffer_size
+
         self._user_audio_buffer = bytearray()
-        self._assistant_audio_buffer = bytearray()
+        self._bot_audio_buffer = bytearray()
-        self._num_channels = None
-        self._sample_rate = None
 
-    def _buffer_has_audio(self, buffer: bytearray):
+        self._register_event_handler("on_audio_data")
-        return buffer is not None and len(buffer) > 0
 
-    def has_audio(self):
+    @property
-        return (
+    def sample_rate(self) -> int:
-            self._buffer_has_audio(self._user_audio_buffer)
+        return self._sample_rate
-            and self._buffer_has_audio(self._assistant_audio_buffer)
+
-            and self._sample_rate is not None
+    @property
+    def num_channels(self) -> int:
+        return self._num_channels
+
+    def has_audio(self) -> bool:
+        return self._buffer_has_audio(self._user_audio_buffer) and self._buffer_has_audio(
+            self._bot_audio_buffer
         )
 
-    def reset_audio_buffer(self):
+    def merge_audio_buffers(self) -> bytes:
+        if self._num_channels == 1:
+            return mix_audio(bytes(self._user_audio_buffer), bytes(self._bot_audio_buffer))
+        elif self._num_channels == 2:
+            return interleave_stereo_audio(
+                bytes(self._user_audio_buffer), bytes(self._bot_audio_buffer)
+            )
+        else:
+            return b""
+
+    def reset_audio_buffers(self):
         self._user_audio_buffer = bytearray()
-        self._assistant_audio_buffer = bytearray()
+        self._bot_audio_buffer = bytearray()
-
-    def merge_audio_buffers(self):
-        with BytesIO() as buffer:
-            with wave.open(buffer, "wb") as wf:
-                wf.setnchannels(2)
-                wf.setsampwidth(2)
-                wf.setframerate(self._sample_rate)
-                # Interleave the two audio streams
-                max_length = max(len(self._user_audio_buffer), len(self._assistant_audio_buffer))
-                interleaved = bytearray(max_length * 2)
-
-                for i in range(0, max_length, 2):
-                    if i < len(self._user_audio_buffer):
-                        interleaved[i * 2] = self._user_audio_buffer[i]
-                        interleaved[i * 2 + 1] = self._user_audio_buffer[i + 1]
-                    else:
-                        interleaved[i * 2] = 0
-                        interleaved[i * 2 + 1] = 0
-
-                    if i < len(self._assistant_audio_buffer):
-                        interleaved[i * 2 + 2] = self._assistant_audio_buffer[i]
-                        interleaved[i * 2 + 3] = self._assistant_audio_buffer[i + 1]
-                    else:
-                        interleaved[i * 2 + 2] = 0
-                        interleaved[i * 2 + 3] = 0
-
-                wf.writeframes(interleaved)
-            return buffer.getvalue()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
-        if isinstance(frame, AudioRawFrame) and self._sample_rate is None:
-            self._sample_rate = frame.sample_rate
 
-        # include all audio from the user
+        # Include all audio from the user.
         if isinstance(frame, InputAudioRawFrame):
-            self._user_audio_buffer.extend(frame.audio)
+            resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate)
-            # Sync the assistant's buffer to the user's buffer by adding silence if needed
+            self._user_audio_buffer.extend(resampled)
-            if len(self._user_audio_buffer) > len(self._assistant_audio_buffer):
+            # Sync the bot's buffer to the user's buffer by adding silence if needed
-                silence_length = len(self._user_audio_buffer) - len(self._assistant_audio_buffer)
+            if len(self._user_audio_buffer) > len(self._bot_audio_buffer):
-                silence = b"\x00" * silence_length
+                silence = b"\x00" * len(resampled)
-                self._assistant_audio_buffer.extend(silence)
+                self._bot_audio_buffer.extend(silence)
+        # If the bot is speaking, include all audio from the bot.
+        elif isinstance(frame, OutputAudioRawFrame):
+            resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate)
+            self._bot_audio_buffer.extend(resampled)
 
-        # if the assistant is speaking, include all audio from the assistant,
+        if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size:
-        if isinstance(frame, OutputAudioRawFrame):
+            await self._call_on_audio_data_handler()
-            self._assistant_audio_buffer.extend(frame.audio)
 
-        # do not push the user's audio frame, doing so will result in echo
+        await self.push_frame(frame, direction)
-        if not isinstance(frame, InputAudioRawFrame):
+
-            await self.push_frame(frame, direction)
+    async def _call_on_audio_data_handler(self):
+        if not self.has_audio():
+            return
+
+        merged_audio = self.merge_audio_buffers()
+        await self._call_event_handler(
+            "on_audio_data", merged_audio, self._sample_rate, self._num_channels
+        )
+        self.reset_audio_buffers()
+
+    def _buffer_has_audio(self, buffer: bytearray) -> bool:
+        return buffer is not None and len(buffer) > 0

src/pipecat/processors/filters/frame_filter.py

@@ -6,7 +6,7 @@
 
 from typing import Tuple, Type
 
-from pipecat.frames.frames import AppFrame, ControlFrame, Frame, SystemFrame
+from pipecat.frames.frames import ControlFrame, Frame, SystemFrame
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 
@@ -23,11 +23,7 @@ class FrameFilter(FrameProcessor):
         if isinstance(frame, self._types):
             return True
 
-        return (
+        return isinstance(frame, ControlFrame) or isinstance(frame, SystemFrame)
-            isinstance(frame, AppFrame)
-            or isinstance(frame, ControlFrame)
-            or isinstance(frame, SystemFrame)
-        )
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)

src/pipecat/processors/frame_processor.py

@@ -13,6 +13,7 @@ from loguru import logger
 
 from pipecat.clocks.base_clock import BaseClock
 from pipecat.frames.frames import (
+    CancelFrame,
     EndFrame,
     ErrorFrame,
     Frame,
@@ -58,6 +59,13 @@ class FrameProcessor:
         self._enable_usage_metrics = False
         self._report_only_initial_ttfb = False
 
+        # Cancellation is done through CancelFrame (a system frame). This could
+        # cause other events being triggered (e.g. closing a transport) which
+        # could also cause other frames to be pushed from other tasks
+        # (e.g. EndFrame). So, when we are cancelling we don't want anything
+        # else to be pushed.
+        self._cancelling = False
+
         # Metrics
         self._metrics = metrics or FrameProcessorMetrics()
         self._metrics.set_processor_name(self.name)
@@ -161,6 +169,10 @@ class FrameProcessor:
             Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]]
         ] = None,
     ):
+        # If we are cancelling we don't want to process any other frame.
+        if self._cancelling:
+            return
+
         if isinstance(frame, SystemFrame):
             # We don't want to queue system frames.
             await self.process_frame(frame, direction)
@@ -187,6 +199,8 @@ class FrameProcessor:
             await self.stop_all_metrics()
         elif isinstance(frame, StopInterruptionFrame):
             self._should_report_ttfb = True
+        elif isinstance(frame, CancelFrame):
+            self._cancelling = True
 
     async def push_error(self, error: ErrorFrame):
         await self.push_frame(error, FrameDirection.UPSTREAM)

src/pipecat/processors/frameworks/rtvi.py

@@ -676,6 +676,7 @@ class RTVIProcessor(FrameProcessor):
             await self.push_frame(frame, direction)
 
     async def cleanup(self):
+        await super().cleanup()
         if self._pipeline:
             await self._pipeline.cleanup()
 

src/pipecat/services/canonical.py

@@ -5,13 +5,16 @@
 #
 
 import aiohttp
+import io
 import os
 import uuid
+import wave
 
 from datetime import datetime
 from typing import Dict, List, Tuple
 
 from pipecat.frames.frames import CancelFrame, EndFrame, Frame
+from pipecat.processors.audio import audio_buffer_processor
 from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import AIService
@@ -81,9 +84,11 @@ class CanonicalMetricsService(AIService):
         self._output_dir = output_dir
 
     async def stop(self, frame: EndFrame):
+        await super().stop(frame)
         await self._process_audio()
 
     async def cancel(self, frame: CancelFrame):
+        await super().cancel(frame)
         await self._process_audio()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -91,23 +96,32 @@ class CanonicalMetricsService(AIService):
         await self.push_frame(frame, direction)
 
     async def _process_audio(self):
-        pipeline = self._audio_buffer_processor
+        audio_buffer_processor = self._audio_buffer_processor
-        if pipeline.has_audio():
-            os.makedirs(self._output_dir, exist_ok=True)
-            filename = self._get_output_filename()
-            wave_data = pipeline.merge_audio_buffers()
 
+        if not audio_buffer_processor.has_audio():
+            return
+
+        os.makedirs(self._output_dir, exist_ok=True)
+        filename = self._get_output_filename()
+        audio = audio_buffer_processor.merge_audio_buffers()
+
+        with io.BytesIO() as buffer:
+            with wave.open(buffer, "wb") as wf:
+                wf.setsampwidth(2)
+                wf.setnchannels(audio_buffer_processor.num_channels)
+                wf.setframerate(audio_buffer_processor.sample_rate)
+                wf.writeframes(audio)
             async with aiofiles.open(filename, "wb") as file:
-                await file.write(wave_data)
+                await file.write(buffer.getvalue())
 
-            try:
+        try:
-                await self._multipart_upload(filename)
+            await self._multipart_upload(filename)
-                pipeline.reset_audio_buffer()
+            await aiofiles.os.remove(filename)
-                await aiofiles.os.remove(filename)
+            audio_buffer_processor.reset_audio_buffers()
-            except FileNotFoundError:
+        except FileNotFoundError:
-                pass
+            pass
-            except Exception as e:
+        except Exception as e:
-                logger.error(f"Failed to upload recording: {e}")
+            logger.error(f"Failed to upload recording: {e}")
 
     def _get_output_filename(self):
         timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")

src/pipecat/transports/base_output.py

@@ -443,11 +443,6 @@ class BaseOutputTransport(FrameProcessor):
             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
-        )
         try:
             async for frame in self._next_audio_frame():
                 # Notify the bot started speaking upstream if necessary and that

src/pipecat/transports/services/daily.py

@@ -723,7 +723,7 @@ class DailyInputTransport(BaseInputTransport):
         await self.push_frame(frame)
 
     async def push_app_message(self, message: Any, sender: str):
-        frame = DailyTransportMessageFrame(message=message, participant_id=sender)
+        frame = DailyTransportMessageUrgentFrame(message=message, participant_id=sender)
         await self.push_frame(frame)
 
     #

src/pipecat/transports/services/livekit.py

@@ -342,7 +342,7 @@ class LiveKitInputTransport(BaseInputTransport):
         return self._vad_analyzer
 
     async def push_app_message(self, message: Any, sender: str):
-        frame = LiveKitTransportMessageFrame(message=message, participant_id=sender)
+        frame = LiveKitTransportMessageUrgentFrame(message=message, participant_id=sender)
         await self.push_frame(frame)
 
     async def _audio_in_task_handler(self):