diff --git a/CHANGELOG.md b/CHANGELOG.md index 203276f1a..19dd9aab5 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/docs/frame.md b/docs/frame.md index 9026f30c1..7f50ffe62 100644 --- a/docs/frame.md +++ b/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. diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index efe4823fb..71aca70f3 100644 --- a/examples/canonical-metrics/bot.py +++ b/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, diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 708de522d..f020a9626 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/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): - if audiobuffer.has_audio(): - merged_audio = audiobuffer.merge_audio_buffers() +async def save_audio(audio: bytes, sample_rate: int, num_channels: int): + if len(audio) > 0: filename = f"conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav" - with wave.open(filename, "wb") as wf: - wf.setnchannels(2) - wf.setsampwidth(2) - wf.setframerate(audiobuffer._sample_rate) - wf.writeframes(merged_audio) + with io.BytesIO() as buffer: + with wave.open(buffer, "wb") as wf: + wf.setsampwidth(2) + wf.setnchannels(num_channels) + 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() diff --git a/examples/chatbot-audio-recording/requirements.txt b/examples/chatbot-audio-recording/requirements.txt index 9786b52de..4cafe3680 100644 --- a/examples/chatbot-audio-recording/requirements.txt +++ b/examples/chatbot-audio-recording/requirements.txt @@ -1,3 +1,4 @@ +aiofiles python-dotenv fastapi[all] uvicorn diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py index 5477d0691..64f85930b 100644 --- a/examples/foundational/05-sync-speech-and-image.py +++ b/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): diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index f2260c57b..057942e04 100644 --- a/src/pipecat/audio/utils.py +++ b/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)) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e057e97da..17059700a 100644 --- a/src/pipecat/frames/frames.py +++ b/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): - """A chunk of audio usually coming from an input transport.""" - - pass - - -@dataclass -class OutputAudioRawFrame(AudioRawFrame): - """A chunk of audio. Will be played by the output transport if the - transport's microphone has been enabled. +class OutputImageRawFrame(DataFrame, ImageRawFrame): + """An image that will be shown by the transport if the transport's camera is + 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 - transport's camera is enabled. + """An output image with an associated URL. These images are usually + 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 - and an LLMEndFrame. Note that the messages property on this class is - mutable, and will be be updated by various ResponseAggregator frame - processors. + service should run a chat completion and emit an LLMFullResponseStartFrame, + TextFrames and an LLMFullResponseStartFrame. Note that the `messages` + property in this class is mutable, and will be be updated by various + 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): - """A frame user to request an image from the given user.""" +class MetricsFrame(SystemFrame): + """Emitted by processor that can compute metrics like latencies.""" - user_id: str - context: Optional[Any] = None - - def __str__(self): - return f"{self.name}, user: {self.user_id}" + data: List[MetricsData] @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): - """Emitted by processor that can compute metrics like latencies.""" +class TranscriptionFrame(SystemFrame): + """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 diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 25a1f0237..488a251f0 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -4,11 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import wave -from io import BytesIO - +from pipecat.audio.utils import interleave_stereo_audio, mix_audio, resample_audio 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): - """ - Initialize the AudioBufferProcessor. + """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 + 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: - - audio_buffer: A bytearray to store incoming audio data. - - num_channels: The number of audio channels (initialized as None). - - sample_rate: The sample rate of the audio (initialized as None). + You can provide the desired output `sample_rate` and incoming audio frames + will resampled to match it. Also, you can provide the number of channels, 1 + for mono and 2 for stereo. With mono audio user and bot audio will be mixed, + 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._num_channels = None - self._sample_rate = None + self._bot_audio_buffer = bytearray() - def _buffer_has_audio(self, buffer: bytearray): - return buffer is not None and len(buffer) > 0 + self._register_event_handler("on_audio_data") - def has_audio(self): - return ( - self._buffer_has_audio(self._user_audio_buffer) - and self._buffer_has_audio(self._assistant_audio_buffer) - and self._sample_rate is not None + @property + def sample_rate(self) -> int: + return self._sample_rate + + @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() - - 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() + self._bot_audio_buffer = bytearray() 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) - # Sync the assistant's buffer to the user's buffer by adding silence if needed - if len(self._user_audio_buffer) > len(self._assistant_audio_buffer): - silence_length = len(self._user_audio_buffer) - len(self._assistant_audio_buffer) - silence = b"\x00" * silence_length - self._assistant_audio_buffer.extend(silence) + resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate) + self._user_audio_buffer.extend(resampled) + # Sync the bot's buffer to the user's buffer by adding silence if needed + if len(self._user_audio_buffer) > len(self._bot_audio_buffer): + silence = b"\x00" * len(resampled) + 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 isinstance(frame, OutputAudioRawFrame): - self._assistant_audio_buffer.extend(frame.audio) + if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size: + await self._call_on_audio_data_handler() - # do not push the user's audio frame, doing so will result in echo - if not isinstance(frame, InputAudioRawFrame): - await self.push_frame(frame, direction) + 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 diff --git a/src/pipecat/processors/filters/frame_filter.py b/src/pipecat/processors/filters/frame_filter.py index f4c4b0f61..11f2e601a 100644 --- a/src/pipecat/processors/filters/frame_filter.py +++ b/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 ( - isinstance(frame, AppFrame) - or isinstance(frame, ControlFrame) - or isinstance(frame, SystemFrame) - ) + return isinstance(frame, ControlFrame) or isinstance(frame, SystemFrame) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 93829c8fc..52066b4f4 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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) diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 4644e748f..0e1f3cf5c 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/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() diff --git a/src/pipecat/services/canonical.py b/src/pipecat/services/canonical.py index 048a6a4ee..265cc1b1b 100644 --- a/src/pipecat/services/canonical.py +++ b/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 - if pipeline.has_audio(): - os.makedirs(self._output_dir, exist_ok=True) - filename = self._get_output_filename() - wave_data = pipeline.merge_audio_buffers() + audio_buffer_processor = self._audio_buffer_processor + 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: - await self._multipart_upload(filename) - pipeline.reset_audio_buffer() - await aiofiles.os.remove(filename) - except FileNotFoundError: - pass - except Exception as e: - logger.error(f"Failed to upload recording: {e}") + try: + await self._multipart_upload(filename) + await aiofiles.os.remove(filename) + audio_buffer_processor.reset_audio_buffers() + except FileNotFoundError: + pass + except Exception as e: + logger.error(f"Failed to upload recording: {e}") def _get_output_filename(self): timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index afb93ebf8..c8aa5ef70 100644 --- a/src/pipecat/transports/base_output.py +++ b/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 diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 7b0cd6c00..a5936e80a 100644 --- a/src/pipecat/transports/services/daily.py +++ b/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) # diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 5d4fbdd6c..6c99f6a66 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/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):