From 996aa461ac6ed7ff9f89632c34539c084fd08bd7 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Wed, 20 May 2026 12:03:24 -0300 Subject: [PATCH] Sending audio faster than realtime. --- scripts/daily/test_tavus_transport.py | 74 ++++++--- src/pipecat/transports/tavus/transport.py | 176 ++++++++++++++-------- 2 files changed, 167 insertions(+), 83 deletions(-) diff --git a/scripts/daily/test_tavus_transport.py b/scripts/daily/test_tavus_transport.py index ffb6b8ee6..b14e7ec99 100644 --- a/scripts/daily/test_tavus_transport.py +++ b/scripts/daily/test_tavus_transport.py @@ -1,6 +1,7 @@ import asyncio import os import signal +import time from daily import ( AudioData, @@ -15,6 +16,14 @@ from loguru import logger load_dotenv(override=True) +# Pipecat sends audio at this true content rate but declares it as +# DECLARED_SAMPLE_RATE to write_frames(), which makes delivery faster than +# real-time. We receive at the declared rate (no resampling) and play back at +# the true rate so the avatar consumes audio at normal speed. +TRUE_SAMPLE_RATE = 24000 +DECLARED_SAMPLE_RATE = 48000 +SPEEDUP = DECLARED_SAMPLE_RATE // TRUE_SAMPLE_RATE + def completion_callback(future): def _callback(*args): @@ -37,19 +46,21 @@ class DailyProxyApp(EventHandler): def __new__(cls, *args, **kwargs): return super().__new__(cls) - def __init__(self, sample_rate: int): + def __init__(self): super().__init__() - self._sample_rate = sample_rate self._loop = asyncio.new_event_loop() - self._audio_queue: asyncio.Queue = asyncio.Queue() + # Raw PCM buffer — filled at DECLARED_SAMPLE_RATE speed, drained at TRUE_SAMPLE_RATE speed. + self._buffer = bytearray() self._audio_task: asyncio.Task | None = None + self._receive_start_time: float | None = None self._client: CallClient = CallClient(event_handler=self) self._client.update_subscription_profiles( {"base": {"camera": "unsubscribed", "microphone": "subscribed"}} ) - self._audio_source = CustomAudioSource(self._sample_rate, 1) + # Playback source declared at TRUE_SAMPLE_RATE — consumes audio at real-time speed. + self._audio_source = CustomAudioSource(TRUE_SAMPLE_RATE, 1) self._audio_track = CustomAudioTrack(self._audio_source) def on_joined(self, data, error): @@ -113,7 +124,6 @@ class DailyProxyApp(EventHandler): if self._audio_task: self._audio_task.cancel() try: - # Waits for it to finish await self._audio_task except asyncio.CancelledError: pass @@ -121,36 +131,60 @@ class DailyProxyApp(EventHandler): async def capture_participant_audio(self, participant_id: str): logger.info(f"Capturing participant audio: {participant_id}") - # Receiving from this custom track - # audio_source: str = "microphone" audio_source: str = "stream" media = {"media": {"customAudio": {audio_source: "subscribed"}}} await self.update_subscriptions(participant_settings={participant_id: media}) + # Must match the declared rate Pipecat used so WebRTC skips resampling — + # every original byte arrives intact. self._client.set_audio_renderer( participant_id, self._audio_data_received, audio_source=audio_source, - sample_rate=self._sample_rate, + sample_rate=DECLARED_SAMPLE_RATE, callback_interval_ms=20, ) + logger.info( + f"Receiving at declared_rate={DECLARED_SAMPLE_RATE} Hz " + f"(true content: {TRUE_SAMPLE_RATE} Hz, ~{SPEEDUP}x faster than real-time)" + ) - async def send_audio(self, audio: AudioData): - future = asyncio.get_running_loop().create_future() - self._audio_source.write_frames(audio.audio_frames, completion=completion_callback(future)) - await future + async def _buffer_audio(self, audio_data: AudioData): + """Append received bytes to the buffer and log the fill rate.""" + new_bytes = audio_data.audio_frames + if self._receive_start_time is None: + self._receive_start_time = time.monotonic() - async def queue_audio(self, audio: AudioData): - await self._audio_queue.put(audio) + self._buffer.extend(new_bytes) def _audio_data_received(self, participant_id: str, audio_data: AudioData, audio_source: str): - # logger.info(f"Received audio data for {participant_id}, audio_source: {audio_source}") - asyncio.run_coroutine_threadsafe(self.queue_audio(audio_data), self._loop) + asyncio.run_coroutine_threadsafe(self._buffer_audio(audio_data), self._loop) async def _audio_task_handler(self): + """Drain the buffer at TRUE_SAMPLE_RATE speed (real-time playback).""" + chunk_frames = int(TRUE_SAMPLE_RATE * 20 / 1000) # 20 ms chunks + chunk_bytes = chunk_frames * 2 # 16-bit mono + last_log_time = self._loop.time() + while True: - audio = await self._audio_queue.get() - await self.send_audio(audio) + if len(self._buffer) >= chunk_bytes: + chunk = bytes(self._buffer[:chunk_bytes]) + del self._buffer[:chunk_bytes] + + future = asyncio.get_running_loop().create_future() + self._audio_source.write_frames(chunk, completion=completion_callback(future)) + await future + else: + await asyncio.sleep(0.001) + + now = self._loop.time() + if now - last_log_time >= 1.0: + buffer_seconds = len(self._buffer) / (TRUE_SAMPLE_RATE * 2) + if buffer_seconds > 0: + logger.info( + f"Buffer status: {len(self._buffer)}B ({buffer_seconds:.3f}s buffered)" + ) + last_log_time = now # # Daily (EventHandler) @@ -160,7 +194,7 @@ class DailyProxyApp(EventHandler): participant_name = participant["info"]["userName"] logger.info(f"Participant {participant_name} joined") if participant_name != "Pipecat": - # We are only subscribing for audios from Pipecat. + # We are only subscribing for audio from Pipecat. return asyncio.run_coroutine_threadsafe( self.capture_participant_audio(participant_id=participant["id"]), self._loop @@ -173,7 +207,7 @@ class DailyProxyApp(EventHandler): def main(): Daily.init() room_url = os.environ["TAVUS_SAMPLE_ROOM_URL"] - app = DailyProxyApp(sample_rate=24000) + app = DailyProxyApp() app.run(room_url) diff --git a/src/pipecat/transports/tavus/transport.py b/src/pipecat/transports/tavus/transport.py index 698d0dc95..dc901681a 100644 --- a/src/pipecat/transports/tavus/transport.py +++ b/src/pipecat/transports/tavus/transport.py @@ -40,10 +40,16 @@ from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.daily.transport import ( DailyCallbacks, + DailyCustomAudioTrackParams, DailyParams, DailyTransportClient, ) +# Opus codec maximum. When the Tavus server supports fast audio delivery it +# returns this as stream_declared_sample_rate so that write_frames() blocks for +# n/48000 s instead of n/true_rate s, delivering audio faster than real-time. +_STREAM_DECLARED_SAMPLE_RATE = 48000 + class TavusApi: """Helper class for interacting with the Tavus API (v2). @@ -69,20 +75,30 @@ class TavusApi: # Only for development self._dev_room_url = os.getenv("TAVUS_SAMPLE_ROOM_URL") - async def create_conversation(self, replica_id: str, persona_id: str) -> dict: + async def create_conversation( + self, replica_id: str, persona_id: str, sample_rate: int + ) -> dict: """Create a new conversation with the specified replica and persona. Args: replica_id: ID of the replica to use in the conversation. persona_id: ID of the persona to use in the conversation. + sample_rate: True audio sample rate of the pipeline's output. Sent + to Tavus so the server can negotiate fast audio delivery. When + the server supports it, the response includes + ``stream_declared_sample_rate`` — the rate Pipecat should + declare to the ``CustomAudioSource`` for faster-than-realtime + delivery. Returns: - Dictionary containing conversation_id and conversation_url. + Dictionary containing conversation_id, conversation_url, and + optionally stream_declared_sample_rate. """ if self._dev_room_url: return { "conversation_id": self.MOCK_CONVERSATION_ID, "conversation_url": self._dev_room_url, + "stream_declared_sample_rate": _STREAM_DECLARED_SAMPLE_RATE, } logger.debug(f"Creating Tavus conversation: replica={replica_id}, persona={persona_id}") @@ -90,6 +106,8 @@ class TavusApi: payload = { "replica_id": replica_id, "persona_id": persona_id, + # TODO: start to send it when Tavus start to support it. + # "sample_rate": sample_rate, } async with self._session.post(url, headers=self._headers, json=payload) as r: r.raise_for_status() @@ -202,76 +220,64 @@ class TavusTransportClient: self._client: DailyTransportClient | None = None self._callbacks = callbacks self._params = params + self._setup: FrameProcessorSetup | None = None + self._initialized = False - async def _initialize(self) -> str: - """Initialize the conversation and return the room URL.""" - response = await self._api.create_conversation(self._replica_id, self._persona_id) - self._conversation_id = response["conversation_id"] - return response["conversation_url"] + def _build_daily_callbacks(self) -> DailyCallbacks: + """Build the DailyCallbacks object.""" + return DailyCallbacks( + on_active_speaker_changed=partial( + self._on_handle_callback, "on_active_speaker_changed" + ), + on_joined=self._on_joined, + on_left=self._on_left, + on_before_leave=partial(self._on_handle_callback, "on_before_leave"), + on_error=partial(self._on_handle_callback, "on_error"), + on_app_message=partial(self._on_handle_callback, "on_app_message"), + on_call_state_updated=partial(self._on_handle_callback, "on_call_state_updated"), + on_client_connected=partial(self._on_handle_callback, "on_client_connected"), + on_client_disconnected=partial(self._on_handle_callback, "on_client_disconnected"), + on_dialin_connected=partial(self._on_handle_callback, "on_dialin_connected"), + on_dialin_ready=partial(self._on_handle_callback, "on_dialin_ready"), + on_dialin_stopped=partial(self._on_handle_callback, "on_dialin_stopped"), + on_dialin_error=partial(self._on_handle_callback, "on_dialin_error"), + on_dialin_warning=partial(self._on_handle_callback, "on_dialin_warning"), + on_dialout_answered=partial(self._on_handle_callback, "on_dialout_answered"), + on_dialout_connected=partial(self._on_handle_callback, "on_dialout_connected"), + on_dialout_stopped=partial(self._on_handle_callback, "on_dialout_stopped"), + on_dialout_error=partial(self._on_handle_callback, "on_dialout_error"), + on_dialout_warning=partial(self._on_handle_callback, "on_dialout_warning"), + on_dtmf_event=partial(self._on_handle_callback, "on_dtmf_event"), + on_participant_joined=self._callbacks.on_participant_joined, + on_participant_left=self._callbacks.on_participant_left, + on_participant_updated=partial(self._on_handle_callback, "on_participant_updated"), + on_transcription_message=partial( + self._on_handle_callback, "on_transcription_message" + ), + on_recording_started=partial(self._on_handle_callback, "on_recording_started"), + on_recording_stopped=partial(self._on_handle_callback, "on_recording_stopped"), + on_recording_error=partial(self._on_handle_callback, "on_recording_error"), + on_transcription_stopped=partial( + self._on_handle_callback, "on_transcription_stopped" + ), + on_transcription_error=partial(self._on_handle_callback, "on_transcription_error"), + ) async def setup(self, setup: FrameProcessorSetup): - """Setup the client and initialize the conversation. + """Save setup context for later use in start(). Args: setup: The frame processor setup configuration. """ - if self._conversation_id is not None: - logger.debug(f"Conversation ID already defined: {self._conversation_id}") - return - try: - room_url = await self._initialize() - daily_callbacks = DailyCallbacks( - on_active_speaker_changed=partial( - self._on_handle_callback, "on_active_speaker_changed" - ), - on_joined=self._on_joined, - on_left=self._on_left, - on_before_leave=partial(self._on_handle_callback, "on_before_leave"), - on_error=partial(self._on_handle_callback, "on_error"), - on_app_message=partial(self._on_handle_callback, "on_app_message"), - on_call_state_updated=partial(self._on_handle_callback, "on_call_state_updated"), - on_client_connected=partial(self._on_handle_callback, "on_client_connected"), - on_client_disconnected=partial(self._on_handle_callback, "on_client_disconnected"), - on_dialin_connected=partial(self._on_handle_callback, "on_dialin_connected"), - on_dialin_ready=partial(self._on_handle_callback, "on_dialin_ready"), - on_dialin_stopped=partial(self._on_handle_callback, "on_dialin_stopped"), - on_dialin_error=partial(self._on_handle_callback, "on_dialin_error"), - on_dialin_warning=partial(self._on_handle_callback, "on_dialin_warning"), - on_dialout_answered=partial(self._on_handle_callback, "on_dialout_answered"), - on_dialout_connected=partial(self._on_handle_callback, "on_dialout_connected"), - on_dialout_stopped=partial(self._on_handle_callback, "on_dialout_stopped"), - on_dialout_error=partial(self._on_handle_callback, "on_dialout_error"), - on_dialout_warning=partial(self._on_handle_callback, "on_dialout_warning"), - on_dtmf_event=partial(self._on_handle_callback, "on_dtmf_event"), - on_participant_joined=self._callbacks.on_participant_joined, - on_participant_left=self._callbacks.on_participant_left, - on_participant_updated=partial(self._on_handle_callback, "on_participant_updated"), - on_transcription_message=partial( - self._on_handle_callback, "on_transcription_message" - ), - on_recording_started=partial(self._on_handle_callback, "on_recording_started"), - on_recording_stopped=partial(self._on_handle_callback, "on_recording_stopped"), - on_recording_error=partial(self._on_handle_callback, "on_recording_error"), - on_transcription_stopped=partial( - self._on_handle_callback, "on_transcription_stopped" - ), - on_transcription_error=partial(self._on_handle_callback, "on_transcription_error"), - ) - self._client = DailyTransportClient( - room_url, None, "Pipecat", self._params, daily_callbacks, self._bot_name - ) - await self._client.setup(setup) - except Exception as e: - logger.error(f"Failed to setup TavusTransportClient: {e}") - await self._api.end_conversation(self._conversation_id) - self._conversation_id = None + self._setup = setup async def cleanup(self): """Cleanup client resources.""" - try: - await self._client.cleanup() - except Exception as e: - logger.error(f"Exception during cleanup: {e}") + if self._client: + try: + await self._client.cleanup() + except Exception as e: + logger.error(f"Exception during cleanup: {e}") async def _on_joined(self, data): """Handle joined event.""" @@ -295,12 +301,56 @@ class TavusTransportClient: return await self._api.get_persona_name(self._persona_id) async def start(self, frame: StartFrame): - """Start the client and join the room. + """Create the conversation, build the Daily client, and join the room. Args: frame: The start frame containing initialization parameters. """ + if self._initialized: + return + self._initialized = True + logger.debug("TavusTransportClient start invoked!") + try: + sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate + response = await self._api.create_conversation( + self._replica_id, self._persona_id, sample_rate + ) + self._conversation_id = response["conversation_id"] + room_url = response["conversation_url"] + + params = self._params + stream_declared_sample_rate = response.get("stream_declared_sample_rate") + if stream_declared_sample_rate: + # Tavus supports fast audio delivery: we write true-rate PCM bytes into a + # CustomAudioSource declared at stream_declared_sample_rate (e.g. 48 kHz). + # write_frames() blocks for n/declared_rate seconds instead of n/true_rate + # seconds, so audio is delivered faster than real-time. The receiver must + # also request the same declared rate so WebRTC skips resampling and every + # original byte arrives intact. + # We always override sample_rate here even if the user already provided + # "stream" params, because the declared rate must match what the server + # negotiated — other fields (channels, send_settings) are preserved. + logger.debug( + f"Tavus fast audio: true_rate={sample_rate} declared_rate={stream_declared_sample_rate}" + ) + existing = dict(params.custom_audio_track_params or {}) + existing["stream"] = ( + existing.get("stream") or DailyCustomAudioTrackParams() + ).model_copy(update={"sample_rate": stream_declared_sample_rate}) + params = params.model_copy(update={"custom_audio_track_params": existing}) + + self._client = DailyTransportClient( + room_url, None, "Pipecat", params, self._build_daily_callbacks(), self._bot_name + ) + await self._client.setup(self._setup) + except Exception as e: + logger.error(f"Failed to start TavusTransportClient: {e}") + await self._api.end_conversation(self._conversation_id) + self._conversation_id = None + self._initialized = False + return + await self._client.start(frame) await self._client.join()