From 5a583574298ecb5cf49525cc19750243d89f593b Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Mon, 26 May 2025 18:34:52 -0300 Subject: [PATCH 1/6] Fixing the issue when using the TavusTransport with some TTS services. --- src/pipecat/transports/services/tavus.py | 37 +++++++++++++++++------- 1 file changed, 26 insertions(+), 11 deletions(-) diff --git a/src/pipecat/transports/services/tavus.py b/src/pipecat/transports/services/tavus.py index 8d704a242..221947875 100644 --- a/src/pipecat/transports/services/tavus.py +++ b/src/pipecat/transports/services/tavus.py @@ -11,6 +11,8 @@ from pydantic import BaseModel from pipecat.audio.utils import create_default_resampler from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, CancelFrame, EndFrame, Frame, @@ -20,8 +22,6 @@ from pipecat.frames.frames import ( StartInterruptionFrame, TransportMessageFrame, TransportMessageUrgentFrame, - TTSStartedFrame, - TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup from pipecat.transports.base_input import BaseInputTransport @@ -365,7 +365,8 @@ class TavusOutputTransport(BaseOutputTransport): self._client = client self._params = params self._samples_sent = 0 - self._start_time = time.time() + self._start_time = None + self._current_idx_str: Optional[str] = None async def setup(self, setup: FrameProcessorSetup): await super().setup(setup) @@ -377,8 +378,6 @@ class TavusOutputTransport(BaseOutputTransport): async def start(self, frame: StartFrame): await super().start(frame) - self._samples_sent = 0 - self._start_time = time.time() await self._client.start(frame) await self.set_transport_ready(frame) @@ -394,25 +393,41 @@ class TavusOutputTransport(BaseOutputTransport): logger.info(f"TavusOutputTransport sending message {frame}") await self._client.send_message(frame) + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + # The BotStartedSpeakingFrame and BotStoppedSpeakingFrame are created inside BaseOutputTransport + # so TavusOutputTransport never receives these frames. + # This is a workaround, so we can more reliably be aware when the bot has started or stopped speaking + if direction == FrameDirection.DOWNSTREAM: + if isinstance(frame, BotStartedSpeakingFrame): + if self._current_idx_str is not None: + logger.warning("TavusOutputTransport self._current_idx_str is already defined!") + self._current_idx_str = str(frame.id) + self._start_time = time.time() + self._samples_sent = 0 + elif isinstance(frame, BotStoppedSpeakingFrame): + await self._client.encode_audio_and_send(b"\x00\x00", True, self._current_idx_str) + self._current_idx_str = None + await super().push_frame(frame, direction) + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) if isinstance(frame, StartInterruptionFrame): await self._handle_interruptions() - elif isinstance(frame, TTSStartedFrame): - self._current_idx_str = str(frame.id) - elif isinstance(frame, TTSStoppedFrame): - logger.debug(f"TAVUS: {self}: stopped speaking") - await self._client.encode_audio_and_send(b"\x00\x00", True, self._current_idx_str) async def _handle_interruptions(self): await self._client.send_interrupt_message() async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None): # Compute wait time for synchronization - wait = self._start_time + (self._samples_sent / self._sample_rate) - time.time() + wait = self._start_time + (self._samples_sent / self.sample_rate) - time.time() if wait > 0: + logger.trace(f"TavusOutputTransport write_raw_audio_frames wait: {wait}") await asyncio.sleep(wait) + if self._current_idx_str is None: + logger.warning("TavusOutputTransport self._current_idx_str not defined yet!") + return + await self._client.encode_audio_and_send(frames, False, self._current_idx_str) # Update timestamp based on number of samples sent From 2b3d2cb34244a65291076a34c93d5751a72461d4 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Mon, 26 May 2025 22:55:20 -0300 Subject: [PATCH 2/6] Fixing the issue when using the TavusVideoService with some TTS services. --- src/pipecat/services/tavus/video.py | 49 +++++++++++++---------------- 1 file changed, 21 insertions(+), 28 deletions(-) diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index 8fa30db8a..aa3af9c50 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -170,15 +170,11 @@ class TavusVideoService(AIService): await self._handle_interruptions() await self.push_frame(frame, direction) elif isinstance(frame, TTSStartedFrame): - await self.start_processing_metrics() - await self.start_ttfb_metrics() - self._current_idx_str = str(frame.id) + await self._queue.put(frame) elif isinstance(frame, TTSAudioRawFrame): - await self._queue_audio(frame.audio, frame.sample_rate, done=False) + await self._queue.put(frame) elif isinstance(frame, TTSStoppedFrame): - await self._queue_audio(b"\x00\x00", self._client.in_sample_rate, done=True) - await self.stop_ttfb_metrics() - await self.stop_processing_metrics() + await self._queue.put(frame) else: await self.push_frame(frame, direction) @@ -191,9 +187,6 @@ class TavusVideoService(AIService): await self._client.stop() self._other_participant_has_joined = False - async def _queue_audio(self, audio: bytes, in_rate: int, done: bool): - await self._queue.put((audio, in_rate, done)) - async def _create_send_task(self): if not self._send_task: self._queue = asyncio.Queue() @@ -224,35 +217,35 @@ class TavusVideoService(AIService): MAX_CHUNK_SIZE = int((sample_rate * 2) / 20) audio_buffer = bytearray() - samples_sent = 0 - start_time = time.time() + current_idx_str = None + silence = b"\x00\x00" while True: - (audio, in_rate, done) = await self._queue.get() - - if done: + frame = await self._queue.get() + if isinstance(frame, TTSStartedFrame): + if current_idx_str is not None: + continue + current_idx_str = str(frame.id) + elif isinstance(frame, TTSStoppedFrame): # Send any remaining audio. if len(audio_buffer) > 0: await self._client.encode_audio_and_send( - bytes(audio_buffer), done, self._current_idx_str + bytes(audio_buffer), True, current_idx_str ) - await self._client.encode_audio_and_send(audio, done, self._current_idx_str) + await self._client.encode_audio_and_send(silence, True, current_idx_str) audio_buffer.clear() - else: - audio = await self._resampler.resample(audio, in_rate, sample_rate) + current_idx_str = None + elif isinstance(frame, TTSAudioRawFrame): + if current_idx_str is None: + continue + audio = await self._resampler.resample(frame.audio, frame.sample_rate, sample_rate) audio_buffer.extend(audio) while len(audio_buffer) >= MAX_CHUNK_SIZE: chunk = audio_buffer[:MAX_CHUNK_SIZE] audio_buffer = audio_buffer[MAX_CHUNK_SIZE:] # Compute wait time for synchronization - wait = start_time + (samples_sent / sample_rate) - time.time() - if wait > 0: - await asyncio.sleep(wait) + wait = 1 / 20 # 50ms + await asyncio.sleep(wait) - await self._client.encode_audio_and_send( - bytes(chunk), done, self._current_idx_str - ) - - # Update timestamp based on number of samples sent - samples_sent += len(chunk) // 2 # 2 bytes per sample (16-bit) + await self._client.encode_audio_and_send(bytes(chunk), False, current_idx_str) From 3039a1444e04554f1148c4efcc8abbebbeb7261c Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 27 May 2025 10:26:41 -0300 Subject: [PATCH 3/6] Refactoring the TavusVideoService to match the same the behavior of the bot started speaking and bot stopped speaking. --- src/pipecat/services/tavus/video.py | 69 +++++++++++++++-------------- 1 file changed, 36 insertions(+), 33 deletions(-) diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index aa3af9c50..d42b2a1c1 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -31,6 +31,9 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSet from pipecat.services.ai_service import AIService from pipecat.transports.services.tavus import TavusCallbacks, TavusParams, TavusTransportClient +# Using the same values that we do in the BaseOutputTransport +BOT_VAD_STOP_SECS = 0.35 + class TavusVideoService(AIService): """ @@ -169,12 +172,8 @@ class TavusVideoService(AIService): if isinstance(frame, StartInterruptionFrame): await self._handle_interruptions() await self.push_frame(frame, direction) - elif isinstance(frame, TTSStartedFrame): - await self._queue.put(frame) elif isinstance(frame, TTSAudioRawFrame): await self._queue.put(frame) - elif isinstance(frame, TTSStoppedFrame): - await self._queue.put(frame) else: await self.push_frame(frame, direction) @@ -197,15 +196,6 @@ class TavusVideoService(AIService): await self.cancel_task(self._send_task) self._send_task = None - # TODO (Filipi): this should be all that is needed use this Microphone Echo mode - # https://docs.tavus.io/sections/conversational-video-interface/layers-and-modes-overview#microphone-echo - # This would allow us to send an audio stream for the replica to repeat - # Checking with Tavus what is the right way to create the Persona to make it work - # async def _send_task_handler(self): - # while True: - # (audio, in_rate, done) = await self._queue.get() - # await self._client.write_raw_audio_frames(audio) - async def _send_task_handler(self): # Daily app-messages have a 4kb limit and also a rate limit of 20 # messages per second. Below, we only consider the rate limit because 1 @@ -219,14 +209,41 @@ class TavusVideoService(AIService): audio_buffer = bytearray() current_idx_str = None silence = b"\x00\x00" + samples_sent = 0 + start_time = None while True: - frame = await self._queue.get() - if isinstance(frame, TTSStartedFrame): - if current_idx_str is not None: - continue - current_idx_str = str(frame.id) - elif isinstance(frame, TTSStoppedFrame): + try: + frame = await asyncio.wait_for(self._queue.get(), timeout=BOT_VAD_STOP_SECS) + if isinstance(frame, TTSAudioRawFrame): + # starting the new inference + if current_idx_str is None: + current_idx_str = str(frame.id) + samples_sent = 0 + start_time = time.time() + + audio = await self._resampler.resample( + frame.audio, frame.sample_rate, sample_rate + ) + audio_buffer.extend(audio) + while len(audio_buffer) >= MAX_CHUNK_SIZE: + chunk = audio_buffer[:MAX_CHUNK_SIZE] + audio_buffer = audio_buffer[MAX_CHUNK_SIZE:] + + # Compute wait time for synchronization + wait = start_time + (samples_sent / sample_rate) - time.time() + if wait > 0: + logger.trace(f"TavusVideoService _send_task_handler wait: {wait}") + await asyncio.sleep(wait) + + await self._client.encode_audio_and_send( + bytes(chunk), False, current_idx_str + ) + + # Update timestamp based on number of samples sent + samples_sent += len(chunk) // 2 # 2 bytes per sample (16-bit) + except asyncio.TimeoutError: + # Bot has stopped speaking # Send any remaining audio. if len(audio_buffer) > 0: await self._client.encode_audio_and_send( @@ -235,17 +252,3 @@ class TavusVideoService(AIService): await self._client.encode_audio_and_send(silence, True, current_idx_str) audio_buffer.clear() current_idx_str = None - elif isinstance(frame, TTSAudioRawFrame): - if current_idx_str is None: - continue - audio = await self._resampler.resample(frame.audio, frame.sample_rate, sample_rate) - audio_buffer.extend(audio) - while len(audio_buffer) >= MAX_CHUNK_SIZE: - chunk = audio_buffer[:MAX_CHUNK_SIZE] - audio_buffer = audio_buffer[MAX_CHUNK_SIZE:] - - # Compute wait time for synchronization - wait = 1 / 20 # 50ms - await asyncio.sleep(wait) - - await self._client.encode_audio_and_send(bytes(chunk), False, current_idx_str) From 798b935ff6b2fcc325a54b9dfd98391733ddfc49 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 27 May 2025 10:28:40 -0300 Subject: [PATCH 4/6] The remaining audio should not be sent as done. --- src/pipecat/services/tavus/video.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index d42b2a1c1..f08e5b6d0 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -247,7 +247,7 @@ class TavusVideoService(AIService): # Send any remaining audio. if len(audio_buffer) > 0: await self._client.encode_audio_and_send( - bytes(audio_buffer), True, current_idx_str + bytes(audio_buffer), False, current_idx_str ) await self._client.encode_audio_and_send(silence, True, current_idx_str) audio_buffer.clear() From 3d442620f94c595ac30f8743bdc0ee746c6b15e5 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 27 May 2025 10:42:06 -0300 Subject: [PATCH 5/6] Removing not used imports. --- src/pipecat/services/tavus/video.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index f08e5b6d0..d354014d5 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -24,8 +24,6 @@ from pipecat.frames.frames import ( StartFrame, StartInterruptionFrame, TTSAudioRawFrame, - TTSStartedFrame, - TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup from pipecat.services.ai_service import AIService From acce9489d76ef34df23f672e6314bfb0fc40a9f6 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 27 May 2025 11:26:34 -0300 Subject: [PATCH 6/6] Creating the silence based on the chunk size. --- src/pipecat/services/tavus/video.py | 3 ++- src/pipecat/transports/services/tavus.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index d354014d5..0b59514e7 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -202,11 +202,12 @@ class TavusVideoService(AIService): # limit (even including base64 encoding). For a sample rate of 16000, # that would be 32000 / 20 = 1600. sample_rate = self._client.out_sample_rate + # 50 ms of audio MAX_CHUNK_SIZE = int((sample_rate * 2) / 20) audio_buffer = bytearray() current_idx_str = None - silence = b"\x00\x00" + silence = b"\x00" * MAX_CHUNK_SIZE samples_sent = 0 start_time = None diff --git a/src/pipecat/transports/services/tavus.py b/src/pipecat/transports/services/tavus.py index 221947875..0ab59f03b 100644 --- a/src/pipecat/transports/services/tavus.py +++ b/src/pipecat/transports/services/tavus.py @@ -405,7 +405,8 @@ class TavusOutputTransport(BaseOutputTransport): self._start_time = time.time() self._samples_sent = 0 elif isinstance(frame, BotStoppedSpeakingFrame): - await self._client.encode_audio_and_send(b"\x00\x00", True, self._current_idx_str) + silence = b"\x00" * self.audio_chunk_size + await self._client.encode_audio_and_send(silence, True, self._current_idx_str) self._current_idx_str = None await super().push_frame(frame, direction)