wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #1334 from pipecat-ai/aleix/user-and-bot-turn-audio

Browse Source 
add support for user and bot turn audio
This commit is contained in:
Mark Backman
2025-03-06 18:35:56 -05:00
committed by GitHub
parent 710eebab09 a91c26785f
commit 61f534ca34
3 changed files with 88 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGELOG.md
View File

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- Added `on_user_turn_audio_data` and `on_bot_turn_audio_data` to
`AudioBufferProcessor`. This gives the ability to grab the audio of only that
turn for both the user and the bot.
- Added new base class `BaseObject` which is now the base class of
`FrameProcessor`, `PipelineRunner`, `PipelineTask` and `BaseTransport`. The
new `BaseObject` adds supports for event handlers.

22
examples/chatbot-audio-recording/bot.py
View File

@@ -32,10 +32,16 @@ load_dotenv(override=True)
logger.remove(0)
logger.add(sys.stderr, level="DEBUG")
# Create the recordings directory if it doesn't exist
os.makedirs("recordings", exist_ok=True)
async def save_audio(audio: bytes, sample_rate: int, num_channels: int):
async def save_audio(audio: bytes, sample_rate: int, num_channels: int, name: str):
if len(audio) > 0:
filename = f"conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav"
filename = os.path.join(
"recordings",
f"{name}_conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav",
)
with io.BytesIO() as buffer:
with wave.open(buffer, "wb") as wf:
wf.setsampwidth(2)
@@ -110,7 +116,7 @@ async def main():
# NOTE: Watch out! This will save all the conversation in memory. You
# can pass `buffer_size` to get periodic callbacks.
audiobuffer = AudioBufferProcessor()
audiobuffer = AudioBufferProcessor(enable_turn_audio=True)
pipeline = Pipeline(
[
@@ -128,7 +134,15 @@ async def main():
@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)
await save_audio(audio, sample_rate, num_channels, "full")
@audiobuffer.event_handler("on_user_turn_audio_data")
async def on_user_turn_audio_data(buffer, audio, sample_rate, num_channels):
await save_audio(audio, sample_rate, num_channels, "user")
@audiobuffer.event_handler("on_bot_turn_audio_data")
async def on_bot_turn_audio_data(buffer, audio, sample_rate, num_channels):
await save_audio(audio, sample_rate, num_channels, "bot")
@transport.event_handler("on_first_participant_joined")
async def on_first_participant_joined(transport, participant):

73
src/pipecat/processors/audio/audio_buffer_processor.py
View File

@@ -10,12 +10,16 @@ from typing import Optional
from pipecat.audio.utils import create_default_resampler, interleave_stereo_audio, mix_audio
from pipecat.frames.frames import (
AudioRawFrame,
BotStartedSpeakingFrame,
BotStoppedSpeakingFrame,
CancelFrame,
EndFrame,
Frame,
InputAudioRawFrame,
OutputAudioRawFrame,
StartFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@@ -30,12 +34,15 @@ class AudioBufferProcessor(FrameProcessor):
Events:
on_audio_data: Triggered when buffer_size is reached, providing merged audio
on_track_audio_data: Triggered when buffer_size is reached, providing separate tracks
on_user_turn_audio_data: Triggered when user turn has ended, providing that user turn's audio
on_bot_turn_audio_data: Triggered when bot turn has ended, providing that bot turn's audio
Args:
sample_rate (Optional[int]): Desired output sample rate. If None, uses source rate
num_channels (int): Number of channels (1 for mono, 2 for stereo). Defaults to 1
buffer_size (int): Size of buffer before triggering events. 0 for no buffering
user_continuous_stream (bool): Whether user audio is continuous or speech-only
enable_turn_audio (bool): Whether turn audio event handlers should be triggered
Audio handling:
- Mono output (num_channels=1): User and bot audio are mixed
@@ -56,18 +63,26 @@ class AudioBufferProcessor(FrameProcessor):
num_channels: int = 1,
buffer_size: int = 0,
user_continuous_stream: bool = True,
enable_turn_audio: bool = False,
**kwargs,
):
super().__init__(**kwargs)
self._init_sample_rate = sample_rate
self._sample_rate = 0
self._audio_buffer_size_1s = 0
self._num_channels = num_channels
self._buffer_size = buffer_size
self._user_continuous_stream = user_continuous_stream
self._enable_turn_audio = enable_turn_audio
self._user_audio_buffer = bytearray()
self._bot_audio_buffer = bytearray()
self._user_speaking = False
self._bot_speaking = False
self._user_turn_audio_buffer = bytearray()
self._bot_turn_audio_buffer = bytearray()
# Intermittent (non continous user stream variables)
self._last_user_frame_at = 0
self._last_bot_frame_at = 0
@@ -78,6 +93,8 @@ class AudioBufferProcessor(FrameProcessor):
self._register_event_handler("on_audio_data")
self._register_event_handler("on_track_audio_data")
self._register_event_handler("on_user_turn_audio_data")
self._register_event_handler("on_bot_turn_audio_data")
@property
def sample_rate(self) -> int:
@@ -150,13 +167,9 @@ class AudioBufferProcessor(FrameProcessor):
self._update_sample_rate(frame)
if self._recording:
if self._user_continuous_stream:
await self._handle_continuous_stream(frame)
else:
await self._handle_intermittent_stream(frame)
if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size:
await self._call_on_audio_data_handler()
await self._process_recording(frame)
if self._enable_turn_audio:
await self._process_turn_recording(frame)
if isinstance(frame, (CancelFrame, EndFrame)):
await self.stop_recording()
@@ -165,6 +178,50 @@ class AudioBufferProcessor(FrameProcessor):
def _update_sample_rate(self, frame: StartFrame):
self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate
self._audio_buffer_size_1s = self._sample_rate * 2
async def _process_recording(self, frame: Frame):
if self._user_continuous_stream:
await self._handle_continuous_stream(frame)
else:
await self._handle_intermittent_stream(frame)
if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size:
await self._call_on_audio_data_handler()
async def _process_turn_recording(self, frame: Frame):
if isinstance(frame, UserStartedSpeakingFrame):
self._user_speaking = True
elif isinstance(frame, UserStoppedSpeakingFrame):
await self._call_event_handler(
"on_user_turn_audio_data", self._user_turn_audio_buffer, self.sample_rate, 1
)
self._user_speaking = False
self._user_turn_audio_buffer = bytearray()
elif isinstance(frame, BotStartedSpeakingFrame):
self._bot_speaking = True
elif isinstance(frame, BotStoppedSpeakingFrame):
await self._call_event_handler(
"on_bot_turn_audio_data", self._bot_turn_audio_buffer, self.sample_rate, 1
)
self._bot_speaking = False
self._bot_turn_audio_buffer = bytearray()
if isinstance(frame, InputAudioRawFrame):
resampled = await self._resample_audio(frame)
self._user_turn_audio_buffer += resampled
# In the case of the user, we need to keep a short buffer of audio
# since VAD notification of when the user starts speaking comes
# later.
if (
not self._user_speaking
and len(self._user_turn_audio_buffer) > self._audio_buffer_size_1s
):
discarded = len(self._user_turn_audio_buffer) - self._audio_buffer_size_1s
self._user_turn_audio_buffer = self._user_turn_audio_buffer[discarded:]
elif self._bot_speaking and isinstance(frame, OutputAudioRawFrame):
resampled = await self._resample_audio(frame)
self._bot_turn_audio_buffer += resampled
async def _handle_continuous_stream(self, frame: Frame):
if isinstance(frame, InputAudioRawFrame):
@@ -233,6 +290,8 @@ class AudioBufferProcessor(FrameProcessor):
def _reset_audio_buffers(self):
self._user_audio_buffer = bytearray()
self._bot_audio_buffer = bytearray()
self._user_turn_audio_buffer = bytearray()
self._bot_turn_audio_buffer = bytearray()
async def _resample_audio(self, frame: AudioRawFrame) -> bytes:
return await self._resampler.resample(frame.audio, frame.sample_rate, self._sample_rate)