Merge branch 'pipecat-ai:main' into snova-jorgep/sambanova-integration

src/pipecat/examples/run.py

@@ -64,7 +64,7 @@ async def maybe_capture_participant_screen(
 def run_example_daily(
     run_example: Callable,
     args: argparse.Namespace,
-    params: DailyParams,
+    transport_params: Mapping[str, Callable] = {},
 ):
     logger.info("Running example with DailyTransport...")
 
@@ -75,6 +75,7 @@ def run_example_daily(
             (room_url, token) = await configure(session)
 
             # Run example function with DailyTransport transport arguments.
+            params: DailyParams = transport_params[args.transport]()
             transport = DailyTransport(room_url, token, "Pipecat", params=params)
             await run_example(transport, args, True)
 
@@ -84,7 +85,7 @@ def run_example_daily(
 def run_example_webrtc(
     run_example: Callable,
     args: argparse.Namespace,
-    params: TransportParams,
+    transport_params: Mapping[str, Callable] = {},
 ):
     logger.info("Running example with SmallWebRTCTransport...")
 
@@ -130,6 +131,7 @@ def run_example_webrtc(
                 pcs_map.pop(webrtc_connection.pc_id, None)
 
             # Run example function with SmallWebRTC transport arguments.
+            params: TransportParams = transport_params[args.transport]()
             transport = SmallWebRTCTransport(params=params, webrtc_connection=pipecat_connection)
             background_tasks.add_task(run_example, transport, args, False)
 
@@ -152,7 +154,7 @@ def run_example_webrtc(
 def run_example_twilio(
     run_example: Callable,
     args: argparse.Namespace,
-    params: FastAPIWebsocketParams,
+    transport_params: Mapping[str, Callable] = {},
 ):
     logger.info("Running example with FastAPIWebsocketTransport (Twilio)...")
 
@@ -195,6 +197,7 @@ def run_example_twilio(
         call_sid = call_data["start"]["callSid"]
 
         # Create websocket transport and update params.
+        params: FastAPIWebsocketParams = transport_params[args.transport]()
         params.add_wav_header = False
         params.serializer = TwilioFrameSerializer(
             stream_sid=stream_sid,
@@ -217,14 +220,13 @@ def run_main(
         logger.error(f"Transport '{args.transport}' not supported by this example")
         return
 
-    params = transport_params[args.transport]()
     match args.transport:
         case "daily":
-            run_example_daily(run_example, args, params)
+            run_example_daily(run_example, args, transport_params)
         case "webrtc":
-            run_example_webrtc(run_example, args, params)
+            run_example_webrtc(run_example, args, transport_params)
         case "twilio":
-            run_example_twilio(run_example, args, params)
+            run_example_twilio(run_example, args, transport_params)
 
 
 def main(

src/pipecat/observers/turn_tracking_observer.py

@@ -12,6 +12,8 @@ from loguru import logger
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
+    CancelFrame,
+    EndFrame,
     StartFrame,
     UserStartedSpeakingFrame,
 )
@@ -73,6 +75,8 @@ class TurnTrackingObserver(BaseObserver):
         # We only want to end the turn if the bot was previously speaking
         elif isinstance(data.frame, BotStoppedSpeakingFrame) and self._is_bot_speaking:
             await self._handle_bot_stopped_speaking(data)
+        elif isinstance(data.frame, (EndFrame, CancelFrame)):
+            await self._handle_pipeline_end(data)
 
     def _schedule_turn_end(self, data: FramePushed):
         """Schedule turn end with a timeout."""
@@ -134,6 +138,14 @@ class TurnTrackingObserver(BaseObserver):
         # This can happen with HTTP TTS services or function calls
         self._schedule_turn_end(data)
 
+    async def _handle_pipeline_end(self, data: FramePushed):
+        """Handle pipeline end or cancellation by flushing any active turn."""
+        if self._is_turn_active:
+            # Cancel any pending turn end timer
+            self._cancel_turn_end_timer()
+            # End the current turn
+            await self._end_turn(data, was_interrupted=True)
+
     async def _start_turn(self, data: FramePushed):
         """Start a new turn."""
         self._is_turn_active = True

src/pipecat/pipeline/task.py

@@ -64,7 +64,7 @@ class PipelineParams(BaseModel):
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
 
-    allow_interruptions: bool = False
+    allow_interruptions: bool = True
     audio_in_sample_rate: int = 16000
     audio_out_sample_rate: int = 24000
     enable_heartbeats: bool = False
@@ -663,6 +663,11 @@ class PipelineTask(BaseTask):
                     diff_time = time.time() - last_frame_time
                     if diff_time >= self._idle_timeout_secs:
                         running = await self._idle_timeout_detected()
+                        # Reset `last_frame_time` so we don't trigger another
+                        # immediate idle timeout if we are not cancelling. For
+                        # example, we might want to force the bot to say goodbye
+                        # and then clean nicely with an `EndFrame`.
+                        last_frame_time = time.time()
 
                 self._idle_queue.task_done()
             except asyncio.TimeoutError:

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -41,7 +41,6 @@ class AudioBufferProcessor(FrameProcessor):
         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:
@@ -50,10 +49,6 @@ class AudioBufferProcessor(FrameProcessor):
         - Automatic resampling of incoming audio to match desired sample_rate
         - Silence insertion for non-continuous audio streams
         - Buffer synchronization between user and bot audio
-
-    Note:
-        When user_continuous_stream is False, the processor expects only speech
-        segments and will handle silence insertion between segments automatically.
     """
 
     def __init__(
@@ -62,7 +57,7 @@ class AudioBufferProcessor(FrameProcessor):
         sample_rate: Optional[int] = None,
         num_channels: int = 1,
         buffer_size: int = 0,
-        user_continuous_stream: bool = True,
+        user_continuous_stream: Optional[bool] = None,
         enable_turn_audio: bool = False,
         **kwargs,
     ):
@@ -72,9 +67,18 @@ class AudioBufferProcessor(FrameProcessor):
         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
 
+        if user_continuous_stream is not None:
+            import warnings
+
+            with warnings.catch_warnings():
+                warnings.simplefilter("always")
+                warnings.warn(
+                    "Parameter `user_continuous_stream` is deprecated.",
+                    DeprecationWarning,
+                )
+
         self._user_audio_buffer = bytearray()
         self._bot_audio_buffer = bytearray()
 
@@ -181,10 +185,24 @@ class AudioBufferProcessor(FrameProcessor):
         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 isinstance(frame, InputAudioRawFrame):
+            # Add silence if we need to.
+            silence = self._compute_silence(self._last_user_frame_at)
+            self._user_audio_buffer.extend(silence)
+            # Add user audio.
+            resampled = await self._resample_audio(frame)
+            self._user_audio_buffer.extend(resampled)
+            # Save time of frame so we can compute silence.
+            self._last_user_frame_at = time.time()
+        elif self._recording and isinstance(frame, OutputAudioRawFrame):
+            # Add silence if we need to.
+            silence = self._compute_silence(self._last_bot_frame_at)
+            self._bot_audio_buffer.extend(silence)
+            # Add bot audio.
+            resampled = await self._resample_audio(frame)
+            self._bot_audio_buffer.extend(resampled)
+            # Save time of frame so we can compute silence.
+            self._last_bot_frame_at = time.time()
 
         if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size:
             await self._call_on_audio_data_handler()
@@ -223,41 +241,6 @@ class AudioBufferProcessor(FrameProcessor):
             resampled = await self._resample_audio(frame)
             self._bot_turn_audio_buffer += resampled
 
-    async def _handle_continuous_stream(self, frame: Frame):
-        if isinstance(frame, InputAudioRawFrame):
-            # Add user audio.
-            resampled = await self._resample_audio(frame)
-            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_size = len(self._user_audio_buffer) - len(self._bot_audio_buffer)
-                silence = b"\x00" * silence_size
-                self._bot_audio_buffer.extend(silence)
-        elif self._recording and isinstance(frame, OutputAudioRawFrame):
-            # Add bot audio.
-            resampled = await self._resample_audio(frame)
-            self._bot_audio_buffer.extend(resampled)
-
-    async def _handle_intermittent_stream(self, frame: Frame):
-        if isinstance(frame, InputAudioRawFrame):
-            # Add silence if we need to.
-            silence = self._compute_silence(self._last_user_frame_at)
-            self._user_audio_buffer.extend(silence)
-            # Add user audio.
-            resampled = await self._resample_audio(frame)
-            self._user_audio_buffer.extend(resampled)
-            # Save time of frame so we can compute silence.
-            self._last_user_frame_at = time.time()
-        elif self._recording and isinstance(frame, OutputAudioRawFrame):
-            # Add silence if we need to.
-            silence = self._compute_silence(self._last_bot_frame_at)
-            self._bot_audio_buffer.extend(silence)
-            # Add bot audio.
-            resampled = await self._resample_audio(frame)
-            self._bot_audio_buffer.extend(resampled)
-            # Save time of frame so we can compute silence.
-            self._last_bot_frame_at = time.time()
-
     async def _call_on_audio_data_handler(self):
         if not self.has_audio() or not self._recording:
             return

src/pipecat/services/aws/tts.py

@@ -6,7 +6,7 @@
 
 import asyncio
 import os
-from typing import AsyncGenerator, Optional
+from typing import AsyncGenerator, List, Optional
 
 from loguru import logger
 from pydantic import BaseModel
@@ -115,6 +115,7 @@ class AWSPollyTTSService(TTSService):
         pitch: Optional[str] = None
         rate: Optional[str] = None
         volume: Optional[str] = None
+        lexicon_names: Optional[List[str]] = None
 
     def __init__(
         self,
@@ -147,6 +148,7 @@ class AWSPollyTTSService(TTSService):
             "pitch": params.pitch,
             "rate": params.rate,
             "volume": params.volume,
+            "lexicon_names": params.lexicon_names,
         }
 
         self._resampler = create_default_resampler()
@@ -235,6 +237,7 @@ class AWSPollyTTSService(TTSService):
                 "Engine": self._settings["engine"],
                 # AWS only supports 8000 and 16000 for PCM. We select 16000.
                 "SampleRate": "16000",
+                "LexiconNames": self._settings["lexicon_names"],
             }
 
             # Filter out None values

src/pipecat/services/aws_nova_sonic/aws.py

@@ -25,6 +25,7 @@ from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
     Frame,
+    FunctionCallFromLLM,
     InputAudioRawFrame,
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
@@ -804,12 +805,16 @@ class AWSNovaSonicLLMService(LLMService):
         # Call tool function
         if self.has_function(function_name):
             if function_name in self._functions.keys() or None in self._functions.keys():
-                await self.call_function(
-                    context=self._context,
-                    tool_call_id=tool_call_id,
-                    function_name=function_name,
-                    arguments=arguments,
-                )
+                function_calls_llm = [
+                    FunctionCallFromLLM(
+                        context=self._context,
+                        tool_call_id=tool_call_id,
+                        function_name=function_name,
+                        arguments=arguments,
+                    )
+                ]
+
+                await self.run_function_calls(function_calls_llm)
         else:
             raise AWSNovaSonicUnhandledFunctionException(
                 f"The LLM tried to call a function named '{function_name}', but there isn't a callback registered for that function."

src/pipecat/services/elevenlabs/tts.py

@@ -428,26 +428,9 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                 break
 
     async def _send_text(self, text: str):
-        if self._websocket:
-            if not self._context_id:
-                # First message for a new context - need a space to initialize
-                msg = {"text": " ", "context_id": str(uuid.uuid4())}
-
-                # Add voice settings only in first message for a context
-                if self._voice_settings:
-                    msg["voice_settings"] = self._voice_settings
-
-                await self._websocket.send(json.dumps(msg))
-                self._context_id = msg["context_id"]
-                logger.trace(f"Created new context {self._context_id}")
-
-                # Now send the actual text content
-                msg = {"text": text, "context_id": self._context_id}
-                await self._websocket.send(json.dumps(msg))
-            else:
-                # Continuing with an existing context
-                msg = {"text": text, "context_id": self._context_id}
-                await self._websocket.send(json.dumps(msg))
+        if self._websocket and self._context_id:
+            msg = {"text": text, "context_id": self._context_id}
+            await self._websocket.send(json.dumps(msg))
 
     @traced_tts
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
@@ -475,8 +458,17 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
                     self._context_id = str(uuid.uuid4())
                     await self.create_audio_context(self._context_id)
 
-                await self._send_text(text)
-                await self.start_tts_usage_metrics(text)
+                    # Initialize context with voice settings
+                    msg = {"text": " ", "context_id": self._context_id}
+                    if self._voice_settings:
+                        msg["voice_settings"] = self._voice_settings
+                    await self._websocket.send(json.dumps(msg))
+                    logger.trace(f"Created new context {self._context_id} with voice settings")
+
+                    await self._send_text(text)
+                    await self.start_tts_usage_metrics(text)
+                else:
+                    await self._send_text(text)
             except Exception as e:
                 logger.error(f"{self} error sending message: {e}")
                 yield TTSStoppedFrame()

src/pipecat/services/gladia/config.py

@@ -74,12 +74,18 @@ class TranslationConfig(BaseModel):
         target_languages: List of target language codes for translation
         model: Translation model to use ("base" or "enhanced")
         match_original_utterances: Whether to align translations with original utterances
+        lipsync: Whether to enable lip-sync optimization for translations
+        context_adaptation: Whether to enable context-aware translation adaptation
+        context: Additional context to help with translation accuracy
         informal: Force informal language forms when available
     """
 
     target_languages: Optional[List[str]] = None
     model: Optional[str] = None
     match_original_utterances: Optional[bool] = None
+    lipsync: Optional[bool] = None
+    context_adaptation: Optional[bool] = None
+    context: Optional[str] = None
     informal: Optional[bool] = None
 
 

src/pipecat/services/gladia/stt.py

@@ -195,6 +195,9 @@ class GladiaSTTService(STTService):
         sample_rate: Optional[int] = None,
         model: str = "solaria-1",
         params: Optional[GladiaInputParams] = None,
+        max_reconnection_attempts: int = 5,
+        reconnection_delay: float = 1.0,
+        max_buffer_size: int = 1024 * 1024 * 20,  # 20MB default buffer
         **kwargs,
     ):
         """Initialize the Gladia STT service.
@@ -204,9 +207,11 @@ class GladiaSTTService(STTService):
             url: Gladia API URL
             confidence: Minimum confidence threshold for transcriptions
             sample_rate: Audio sample rate in Hz
-            model: Model to use ("solaria-1", "solaria-mini-1", "fast",
-                or "accurate")
+            model: Model to use ("solaria-1")
             params: Additional configuration parameters
+            max_reconnection_attempts: Maximum number of reconnection attempts
+            reconnection_delay: Initial delay between reconnection attempts (exponential backoff)
+            max_buffer_size: Maximum size of audio buffer in bytes
             **kwargs: Additional arguments passed to the STTService
         """
         super().__init__(sample_rate=sample_rate, **kwargs)
@@ -232,6 +237,23 @@ class GladiaSTTService(STTService):
         self._keepalive_task = None
         self._settings = {}
 
+        # Reconnection settings
+        self._max_reconnection_attempts = max_reconnection_attempts
+        self._reconnection_delay = reconnection_delay
+        self._reconnection_attempts = 0
+        self._session_url = None
+        self._connection_active = False
+
+        # Audio buffer management
+        self._audio_buffer = bytearray()
+        self._bytes_sent = 0
+        self._max_buffer_size = max_buffer_size
+        self._buffer_lock = asyncio.Lock()
+
+        # Connection management
+        self._connection_task = None
+        self._should_reconnect = True
+
     def can_generate_metrics(self) -> bool:
         return True
 
@@ -293,36 +315,116 @@ class GladiaSTTService(STTService):
     async def start(self, frame: StartFrame):
         """Start the Gladia STT websocket connection."""
         await super().start(frame)
-        if self._websocket:
+        if self._connection_task:
             return
-        settings = self._prepare_settings()
-        response = await self._setup_gladia(settings)
-        self._websocket = await websockets.connect(response["url"])
-        if self._websocket and not self._receive_task:
-            self._receive_task = self.create_task(self._receive_task_handler())
-        if self._websocket and not self._keepalive_task:
-            self._keepalive_task = self.create_task(self._keepalive_task_handler())
+
+        self._should_reconnect = True
+        self._connection_task = self.create_task(self._connection_handler())
 
     async def stop(self, frame: EndFrame):
         """Stop the Gladia STT websocket connection."""
         await super().stop(frame)
+        self._should_reconnect = False
         await self._send_stop_recording()
 
-        if self._keepalive_task:
-            await self.cancel_task(self._keepalive_task)
-            self._keepalive_task = None
+        if self._connection_task:
+            await self.cancel_task(self._connection_task)
+            self._connection_task = None
 
-        if self._websocket:
-            await self._websocket.close()
-            self._websocket = None
-
-        if self._receive_task:
-            await self.wait_for_task(self._receive_task)
-            self._receive_task = None
+        await self._cleanup_connection()
 
     async def cancel(self, frame: CancelFrame):
         """Cancel the Gladia STT websocket connection."""
         await super().cancel(frame)
+        self._should_reconnect = False
+
+        if self._connection_task:
+            await self.cancel_task(self._connection_task)
+            self._connection_task = None
+
+        await self._cleanup_connection()
+
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
+        """Run speech-to-text on audio data."""
+        await self.start_ttfb_metrics()
+        await self.start_processing_metrics()
+
+        # Add audio to buffer
+        async with self._buffer_lock:
+            self._audio_buffer.extend(audio)
+            # Trim buffer if it exceeds max size
+            if len(self._audio_buffer) > self._max_buffer_size:
+                trim_size = len(self._audio_buffer) - self._max_buffer_size
+                self._audio_buffer = self._audio_buffer[trim_size:]
+                self._bytes_sent = max(0, self._bytes_sent - trim_size)
+                logger.warning(f"Audio buffer exceeded max size, trimmed {trim_size} bytes")
+
+        # Send audio if connected
+        if self._connection_active and self._websocket and not self._websocket.closed:
+            try:
+                await self._send_audio(audio)
+            except websockets.exceptions.ConnectionClosed as e:
+                logger.warning(f"Websocket closed while sending audio chunk: {e}")
+                self._connection_active = False
+
+        yield None
+
+    async def _connection_handler(self):
+        """Handle WebSocket connection with automatic reconnection."""
+        while self._should_reconnect:
+            try:
+                # Initialize session if needed
+                if not self._session_url:
+                    settings = self._prepare_settings()
+                    response = await self._setup_gladia(settings)
+                    self._session_url = response["url"]
+                    self._reconnection_attempts = 0
+
+                # Connect with automatic reconnection
+                async with websockets.connect(self._session_url) as websocket:
+                    try:
+                        self._websocket = websocket
+                        self._connection_active = True
+                        logger.info("Connected to Gladia WebSocket")
+
+                        # Send buffered audio if any
+                        await self._send_buffered_audio()
+
+                        # Start tasks
+                        self._receive_task = asyncio.create_task(self._receive_task_handler())
+                        self._keepalive_task = asyncio.create_task(self._keepalive_task_handler())
+
+                        # Wait for tasks to complete
+                        await asyncio.gather(self._receive_task, self._keepalive_task)
+
+                    except websockets.exceptions.ConnectionClosed as e:
+                        logger.warning(f"WebSocket connection closed: {e}")
+                        self._connection_active = False
+
+                        # Clean up tasks
+                        if self._receive_task:
+                            self._receive_task.cancel()
+                        if self._keepalive_task:
+                            self._keepalive_task.cancel()
+
+                        # Attempt reconnect using helper
+                        if not await self._maybe_reconnect():
+                            break
+
+            except Exception as e:
+                logger.error(f"Error in connection handler: {e}")
+                self._connection_active = False
+
+                if not self._should_reconnect:
+                    break
+
+                # Reset session URL to get a new one
+                self._session_url = None
+                await asyncio.sleep(self._reconnection_delay)
+
+    async def _cleanup_connection(self):
+        """Clean up connection resources."""
+        self._connection_active = False
 
         if self._keepalive_task:
             await self.cancel_task(self._keepalive_task)
@@ -336,13 +438,6 @@ class GladiaSTTService(STTService):
             await self.cancel_task(self._receive_task)
             self._receive_task = None
 
-    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
-        """Run speech-to-text on audio data."""
-        await self.start_ttfb_metrics()
-        await self.start_processing_metrics()
-        await self._send_audio(audio)
-        yield None
-
     async def _setup_gladia(self, settings: Dict[str, Any]):
         async with aiohttp.ClientSession() as session:
             async with session.post(
@@ -369,9 +464,18 @@ class GladiaSTTService(STTService):
         await self.stop_processing_metrics()
 
     async def _send_audio(self, audio: bytes):
-        data = base64.b64encode(audio).decode("utf-8")
-        message = {"type": "audio_chunk", "data": {"chunk": data}}
-        await self._websocket.send(json.dumps(message))
+        """Send audio chunk with proper message format."""
+        if self._websocket and not self._websocket.closed:
+            data = base64.b64encode(audio).decode("utf-8")
+            message = {"type": "audio_chunk", "data": {"chunk": data}}
+            await self._websocket.send(json.dumps(message))
+
+    async def _send_buffered_audio(self):
+        """Send any buffered audio after reconnection."""
+        async with self._buffer_lock:
+            if self._audio_buffer:
+                logger.info(f"Sending {len(self._audio_buffer)} bytes of buffered audio")
+                await self._send_audio(bytes(self._audio_buffer))
 
     async def _send_stop_recording(self):
         if self._websocket and not self._websocket.closed:
@@ -380,7 +484,7 @@ class GladiaSTTService(STTService):
     async def _keepalive_task_handler(self):
         """Send periodic empty audio chunks to keep the connection alive."""
         try:
-            while True:
+            while self._connection_active:
                 # Send keepalive every 20 seconds (Gladia times out after 30 seconds)
                 await asyncio.sleep(20)
                 if self._websocket and not self._websocket.closed:
@@ -399,7 +503,19 @@ class GladiaSTTService(STTService):
         try:
             async for message in self._websocket:
                 content = json.loads(message)
-                if content["type"] == "transcript":
+
+                # Handle audio chunk acknowledgments
+                if content["type"] == "audio_chunk" and content.get("acknowledged"):
+                    byte_range = content["data"]["byte_range"]
+                    async with self._buffer_lock:
+                        # Update bytes sent and trim acknowledged data from buffer
+                        end_byte = byte_range[1]
+                        if end_byte > self._bytes_sent:
+                            trim_size = end_byte - self._bytes_sent
+                            self._audio_buffer = self._audio_buffer[trim_size:]
+                            self._bytes_sent = end_byte
+
+                elif content["type"] == "transcript":
                     utterance = content["data"]["utterance"]
                     confidence = utterance.get("confidence", 0)
                     language = utterance["language"]
@@ -448,3 +564,19 @@ class GladiaSTTService(STTService):
             pass
         except Exception as e:
             logger.error(f"Error in Gladia WebSocket handler: {e}")
+
+    async def _maybe_reconnect(self) -> bool:
+        """Handle exponential backoff reconnection logic."""
+        if not self._should_reconnect:
+            return False
+        self._reconnection_attempts += 1
+        if self._reconnection_attempts > self._max_reconnection_attempts:
+            logger.error(f"Max reconnection attempts ({self._max_reconnection_attempts}) reached")
+            self._should_reconnect = False
+            return False
+        delay = self._reconnection_delay * (2 ** (self._reconnection_attempts - 1))
+        logger.info(
+            f"Reconnecting in {delay} seconds (attempt {self._reconnection_attempts}/{self._max_reconnection_attempts})"
+        )
+        await asyncio.sleep(delay)
+        return True

src/pipecat/services/tavus/video.py

@@ -7,7 +7,6 @@
 """This module implements Tavus as a sink transport layer"""
 
 import asyncio
-import time
 from typing import Optional
 
 import aiohttp
@@ -29,9 +28,6 @@ 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):
     """
@@ -48,7 +44,7 @@ class TavusVideoService(AIService):
     Args:
         api_key (str): Tavus API key used for authentication.
         replica_id (str): ID of the Tavus voice replica to use for speech synthesis.
-        persona_id (str): ID of the Tavus persona. Defaults to "pipecat0" to use the Pipecat TTS voice.
+        persona_id (str): ID of the Tavus persona. Defaults to "pipecat-stream" to use the Pipecat TTS voice.
         session (aiohttp.ClientSession): Async HTTP session used for communication with Tavus.
         **kwargs: Additional arguments passed to the parent `AIService` class.
     """
@@ -58,7 +54,7 @@ class TavusVideoService(AIService):
         *,
         api_key: str,
         replica_id: str,
-        persona_id: str = "pipecat0",  # Use `pipecat0` so that your TTS voice is used in place of the Tavus persona
+        persona_id: str = "pipecat-stream",
         session: aiohttp.ClientSession,
         **kwargs,
     ) -> None:
@@ -77,6 +73,8 @@ class TavusVideoService(AIService):
         self._audio_buffer = bytearray()
         self._queue = asyncio.Queue()
         self._send_task: Optional[asyncio.Task] = None
+        # This is the custom track destination expected by Tavus
+        self._transport_destination: Optional[str] = "stream"
 
     async def setup(self, setup: FrameProcessorSetup):
         await super().setup(setup)
@@ -94,6 +92,8 @@ class TavusVideoService(AIService):
             params=TavusParams(
                 audio_in_enabled=True,
                 video_in_enabled=True,
+                audio_out_enabled=True,
+                microphone_out_enabled=False,
             ),
         )
         await self._client.setup(setup)
@@ -152,6 +152,8 @@ class TavusVideoService(AIService):
     async def start(self, frame: StartFrame):
         await super().start(frame)
         await self._client.start(frame)
+        if self._transport_destination:
+            await self._client.register_audio_destination(self._transport_destination)
         await self._create_send_task()
 
     async def stop(self, frame: EndFrame):
@@ -171,7 +173,7 @@ class TavusVideoService(AIService):
             await self._handle_interruptions()
             await self.push_frame(frame, direction)
         elif isinstance(frame, TTSAudioRawFrame):
-            await self._queue.put(frame)
+            await self._handle_audio_frame(frame)
         else:
             await self.push_frame(frame, direction)
 
@@ -194,60 +196,26 @@ class TavusVideoService(AIService):
             await self.cancel_task(self._send_task)
             self._send_task = None
 
-    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
-        # second of a 24000 sample rate would be 48000 bytes (16-bit samples and
-        # 1 channel). So, that is 48000 / 20 = 2400, which is below the 4kb
-        # limit (even including base64 encoding). For a sample rate of 16000,
-        # that would be 32000 / 20 = 1600.
+    async def _handle_audio_frame(self, frame: OutputAudioRawFrame):
         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" * MAX_CHUNK_SIZE
-        samples_sent = 0
-        start_time = None
+        # 40 ms of audio
+        chunk_size = int((sample_rate * 2) / 25)
+        # We might need to resample if incoming audio doesn't match the
+        # transport sample rate.
+        resampled = await self._resampler.resample(frame.audio, frame.sample_rate, sample_rate)
+        self._audio_buffer.extend(resampled)
+        while len(self._audio_buffer) >= chunk_size:
+            chunk = OutputAudioRawFrame(
+                bytes(self._audio_buffer[:chunk_size]),
+                sample_rate=sample_rate,
+                num_channels=frame.num_channels,
+            )
+            chunk.transport_destination = self._transport_destination
+            await self._queue.put(chunk)
+            self._audio_buffer = self._audio_buffer[chunk_size:]
 
+    async def _send_task_handler(self):
         while True:
-            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(
-                        bytes(audio_buffer), False, current_idx_str
-                    )
-                await self._client.encode_audio_and_send(silence, True, current_idx_str)
-                audio_buffer.clear()
-                current_idx_str = None
+            frame = await self._queue.get()
+            if isinstance(frame, OutputAudioRawFrame):
+                await self._client.write_audio_frame(frame)

src/pipecat/transports/services/daily.py

@@ -767,6 +767,7 @@ class DailyTransportClient(EventHandler):
         self._client.add_custom_audio_track(
             track_name=track_name,
             audio_track=audio_track,
+            ignore_audio_level=True,
             completion=completion_callback(future),
         )
 

src/pipecat/transports/services/helpers/daily_rest.py

@@ -300,6 +300,7 @@ class DailyRESTHelper:
         Args:
             room_url: Daily room URL
             expiry_time: Token validity duration in seconds (default: 1 hour)
+            eject_at_token_exp: Whether to eject user when token expires
             owner: Whether token has owner privileges
             params: Optional additional token properties. Note that room_name,
                 exp, and is_owner will be set based on the other function

src/pipecat/transports/services/tavus.py

@@ -1,6 +1,4 @@
-import asyncio
-import base64
-import time
+import os
 from functools import partial
 from typing import Any, Awaitable, Callable, Mapping, Optional
 
@@ -11,8 +9,6 @@ from pydantic import BaseModel
 
 from pipecat.audio.utils import create_default_resampler
 from pipecat.frames.frames import (
-    BotStartedSpeakingFrame,
-    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     Frame,
@@ -40,6 +36,8 @@ class TavusApi:
     """
 
     BASE_URL = "https://tavusapi.com/v2"
+    MOCK_CONVERSATION_ID = "dev-conversation"
+    MOCK_PERSONA_NAME = "TestTavusTransport"
 
     def __init__(self, api_key: str, session: aiohttp.ClientSession):
         """
@@ -52,8 +50,16 @@ class TavusApi:
         self._api_key = api_key
         self._session = session
         self._headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
+        # 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:
+        if self._dev_room_url:
+            return {
+                "conversation_id": self.MOCK_CONVERSATION_ID,
+                "conversation_url": self._dev_room_url,
+            }
+
         logger.debug(f"Creating Tavus conversation: replica={replica_id}, persona={persona_id}")
         url = f"{self.BASE_URL}/conversations"
         payload = {
@@ -67,7 +73,7 @@ class TavusApi:
             return response
 
     async def end_conversation(self, conversation_id: str):
-        if conversation_id is None:
+        if conversation_id is None or conversation_id == self.MOCK_CONVERSATION_ID:
             return
 
         url = f"{self.BASE_URL}/conversations/{conversation_id}/end"
@@ -76,6 +82,9 @@ class TavusApi:
             logger.debug(f"Ended Tavus conversation {conversation_id}")
 
     async def get_persona_name(self, persona_id: str) -> str:
+        if self._dev_room_url is not None:
+            return self.MOCK_PERSONA_NAME
+
         url = f"{self.BASE_URL}/personas/{persona_id}"
         async with self._session.get(url, headers=self._headers) as r:
             r.raise_for_status()
@@ -119,7 +128,7 @@ class TavusTransportClient:
         callbacks (TavusCallbacks): Callback handlers for Tavus-related events.
         api_key (str): API key for authenticating with Tavus API.
         replica_id (str): ID of the replica to use in the Tavus conversation.
-        persona_id (str): ID of the Tavus persona. Defaults to "pipecat0", which signals Tavus to use
+        persona_id (str): ID of the Tavus persona. Defaults to "pipecat-stream", which signals Tavus to use
                           the TTS voice of the Pipecat bot instead of a Tavus persona voice.
         session (aiohttp.ClientSession): The aiohttp session for making async HTTP requests.
         sample_rate: Audio sample rate to be used by the client.
@@ -133,7 +142,7 @@ class TavusTransportClient:
         callbacks: TavusCallbacks,
         api_key: str,
         replica_id: str,
-        persona_id: str = "pipecat0",  # Use `pipecat0` so that your TTS voice is used in place of the Tavus persona
+        persona_id: str = "pipecat-stream",
         session: aiohttp.ClientSession,
     ) -> None:
         self._bot_name = bot_name
@@ -141,7 +150,6 @@ class TavusTransportClient:
         self._replica_id = replica_id
         self._persona_id = persona_id
         self._conversation_id: Optional[str] = None
-        self._other_participant_has_joined = False
         self._client: Optional[DailyTransportClient] = None
         self._callbacks = callbacks
         self._params = params
@@ -153,6 +161,7 @@ class TavusTransportClient:
 
     async def setup(self, setup: FrameProcessorSetup):
         if self._conversation_id is not None:
+            logger.debug(f"Conversation ID already defined: {self._conversation_id}")
             return
         try:
             room_url = await self._initialize()
@@ -194,12 +203,13 @@ class TavusTransportClient:
         except Exception as e:
             logger.error(f"Failed to setup TavusTransportClient: {e}")
             await self._api.end_conversation(self._conversation_id)
+            self._conversation_id = None
 
     async def cleanup(self):
-        if self._client is None:
-            return
-        await self._client.cleanup()
-        self._client = None
+        try:
+            await self._client.cleanup()
+        except Exception as e:
+            logger.exception(f"Exception during cleanup: {e}")
 
     async def _on_joined(self, data):
         logger.debug("TavusTransportClient joined!")
@@ -221,6 +231,7 @@ class TavusTransportClient:
     async def stop(self):
         await self._client.leave()
         await self._api.end_conversation(self._conversation_id)
+        self._conversation_id = None
 
     async def capture_participant_video(
         self,
@@ -257,11 +268,6 @@ class TavusTransportClient:
     def in_sample_rate(self) -> int:
         return self._client.in_sample_rate
 
-    async def encode_audio_and_send(self, audio: bytes, done: bool, inference_id: str):
-        """Encodes audio to base64 and sends it to Tavus"""
-        audio_base64 = base64.b64encode(audio).decode("utf-8")
-        await self._send_audio_message(audio_base64, done=done, inference_id=inference_id)
-
     async def send_interrupt_message(self) -> None:
         transport_frame = TransportMessageUrgentFrame(
             message={
@@ -272,23 +278,6 @@ class TavusTransportClient:
         )
         await self.send_message(transport_frame)
 
-    async def _send_audio_message(self, audio_base64: str, done: bool, inference_id: str):
-        transport_frame = TransportMessageUrgentFrame(
-            message={
-                "message_type": "conversation",
-                "event_type": "conversation.echo",
-                "conversation_id": self._conversation_id,
-                "properties": {
-                    "modality": "audio",
-                    "inference_id": inference_id,
-                    "audio": audio_base64,
-                    "done": done,
-                    "sample_rate": self.out_sample_rate,
-                },
-            }
-        )
-        await self.send_message(transport_frame)
-
     async def update_subscriptions(self, participant_settings=None, profile_settings=None):
         if not self._client:
             return
@@ -300,9 +289,14 @@ class TavusTransportClient:
     async def write_audio_frame(self, frame: OutputAudioRawFrame):
         if not self._client:
             return
-
         await self._client.write_audio_frame(frame)
 
+    async def register_audio_destination(self, destination: str):
+        if not self._client:
+            return
+
+        await self._client.register_audio_destination(destination)
+
 
 class TavusInputTransport(BaseInputTransport):
     def __init__(
@@ -379,12 +373,11 @@ class TavusOutputTransport(BaseOutputTransport):
         super().__init__(params, **kwargs)
         self._client = client
         self._params = params
-        self._samples_sent = 0
-        self._start_time = None
-        self._current_idx_str: Optional[str] = None
 
         # Whether we have seen a StartFrame already.
         self._initialized = False
+        # This is the custom track destination expected by Tavus
+        self._transport_destination: Optional[str] = "stream"
 
     async def setup(self, setup: FrameProcessorSetup):
         await super().setup(setup)
@@ -403,6 +396,10 @@ class TavusOutputTransport(BaseOutputTransport):
         self._initialized = True
 
         await self._client.start(frame)
+
+        if self._transport_destination:
+            await self._client.register_audio_destination(self._transport_destination)
+
         await self.set_transport_ready(frame)
 
     async def stop(self, frame: EndFrame):
@@ -417,23 +414,6 @@ 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):
-                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)
-
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
         if isinstance(frame, StartInterruptionFrame):
@@ -443,20 +423,12 @@ class TavusOutputTransport(BaseOutputTransport):
         await self._client.send_interrupt_message()
 
     async def write_audio_frame(self, frame: OutputAudioRawFrame):
-        # Compute wait time for synchronization
-        wait = self._start_time + (self._samples_sent / self.sample_rate) - time.time()
-        if wait > 0:
-            logger.trace(f"TavusOutputTransport write_audio_frame wait: {wait}")
-            await asyncio.sleep(wait)
+        # This is the custom track destination expected by Tavus
+        frame.transport_destination = self._transport_destination
+        await self._client.write_audio_frame(frame)
 
-        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(frame.audio, False, self._current_idx_str)
-
-        # Update timestamp based on number of samples sent
-        self._samples_sent += len(frame.audio) // 2  # 2 bytes per sample (16-bit)
+    async def register_audio_destination(self, destination: str):
+        await self._client.register_audio_destination(destination)
 
 
 class TavusTransport(BaseTransport):
@@ -472,7 +444,7 @@ class TavusTransport(BaseTransport):
         session (aiohttp.ClientSession): aiohttp session used for async HTTP requests.
         api_key (str): Tavus API key for authentication.
         replica_id (str): ID of the replica model used for voice generation.
-        persona_id (str): ID of the Tavus persona. Defaults to "pipecat0" to use the Pipecat TTS voice.
+        persona_id (str): ID of the Tavus persona. Defaults to "pipecat-stream" to use the Pipecat TTS voice.
         params (TavusParams): Optional Tavus-specific configuration parameters.
         input_name (Optional[str]): Optional name for the input transport.
         output_name (Optional[str]): Optional name for the output transport.
@@ -484,7 +456,7 @@ class TavusTransport(BaseTransport):
         session: aiohttp.ClientSession,
         api_key: str,
         replica_id: str,
-        persona_id: str = "pipecat0",  # Use `pipecat0` so that your TTS voice is used in place of the Tavus persona
+        persona_id: str = "pipecat-stream",
         params: TavusParams = TavusParams(),
         input_name: Optional[str] = None,
         output_name: Optional[str] = None,
@@ -492,11 +464,6 @@ class TavusTransport(BaseTransport):
         super().__init__(input_name=input_name, output_name=output_name)
         self._params = params
 
-        # TODO: Filipi - We can remove this if we stop sending the audio through app messages
-        # Limiting this so we don't go over 20 messages per second
-        # each message is going to have 50ms of audio
-        self._params.audio_out_10ms_chunks = 5
-
         callbacks = TavusCallbacks(
             on_participant_joined=self._on_participant_joined,
             on_participant_left=self._on_participant_left,
@@ -527,6 +494,7 @@ class TavusTransport(BaseTransport):
     async def _on_participant_joined(self, participant):
         # get persona, look up persona_name, set this as the bot name to ignore
         persona_name = await self._client.get_persona_name()
+
         # Ignore the Tavus replica's microphone
         if participant.get("info", {}).get("userName", "") == persona_name:
             self._tavus_participant_id = participant["id"]