- Added a new audio_in_stream_on_start field to TransportParams.

- Added a new method `start_audio_in_streaming` in the `BaseInputTransport`.
- Updated `DailyTransport` to respect the `audio_in_stream_on_start` field, ensuring it only starts receiving the audio input if it is enabled.

CHANGELOG.md

@@ -9,6 +9,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added a new `audio_in_stream_on_start` field to `TransportParams`.
+ 
+- Added a new method `start_audio_in_streaming` in the `BaseInputTransport`. 
+  - This method should be used to start receiving the input audio in case the field `audio_in_stream_on_start` is set to `false`.
+
+- Added support for the `RTVIProcessor` to handle buffered audio in `base64` format, converting it into InputAudioRawFrame for transport.
+
+- Added support for the `RTVIProcessor` to trigger `start_audio_in_streaming` only after the `client-ready` message.
+
 - Added new `MUTE_UNTIL_FIRST_BOT_COMPLETE` strategy to `STTMuteStrategy`. This
   strategy starts muted and remains muted until the first bot speech completes,
   ensuring the bot's first response cannot be interrupted. This complements the
@@ -36,6 +45,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- Updated `DailyTransport` to respect the `audio_in_stream_on_start` field, ensuring it only starts receiving the audio input if it is enabled.
+
+- Updated `FastAPIWebsocketOutputTransport` to send `TransportMessageFrame` and `TransportMessageUrgentFrame` to the serializer.
+
+- Updated `WebsocketServerOutputTransport` to send `TransportMessageFrame` and `TransportMessageUrgentFrame` to the serializer.
+
 - Enhanced `STTMuteConfig` to validate strategy combinations, preventing
   `MUTE_UNTIL_FIRST_BOT_COMPLETE` and `FIRST_SPEECH` from being used together
   as they handle first bot speech differently.

src/pipecat/transports/base_input.py

@@ -47,6 +47,13 @@ class BaseInputTransport(FrameProcessor):
         # if passthrough is enabled.
         self._audio_task = None
 
+    def enable_audio_in_stream_on_start(self, enabled: bool) -> None:
+        logger.debug(f"Enabling audio on start. {enabled}")
+        self._params.audio_in_stream_on_start = enabled
+
+    def start_audio_in_streaming(self):
+        pass
+
     @property
     def sample_rate(self) -> int:
         return self._sample_rate

src/pipecat/transports/base_transport.py

@@ -39,6 +39,7 @@ class TransportParams(BaseModel):
     audio_in_sample_rate: Optional[int] = None
     audio_in_channels: int = 1
     audio_in_filter: Optional[BaseAudioFilter] = None
+    audio_in_stream_on_start: bool = True
     vad_enabled: bool = False
     vad_audio_passthrough: bool = False
     vad_analyzer: Optional[VADAnalyzer] = None

src/pipecat/transports/services/daily.py

@@ -5,7 +5,6 @@
 #
 
 import asyncio
-import base64
 import time
 import warnings
 from concurrent.futures import ThreadPoolExecutor
@@ -839,6 +838,13 @@ class DailyInputTransport(BaseInputTransport):
     def vad_analyzer(self) -> Optional[VADAnalyzer]:
         return self._vad_analyzer
 
+    def start_audio_in_streaming(self):
+        # Create audio task. It reads audio frames from Daily and push them
+        # internally for VAD processing.
+        if self._params.audio_in_enabled or self._params.vad_enabled:
+            logger.debug(f"Start receiving audio")
+            self._audio_in_task = self.create_task(self._audio_in_task_handler())
+
     async def start(self, frame: StartFrame):
         # Parent start.
         await super().start(frame)
@@ -849,10 +855,8 @@ class DailyInputTransport(BaseInputTransport):
         # Inialize WebRTC VAD if needed.
         if self._params.vad_enabled and not self._params.vad_analyzer:
             self._vad_analyzer = WebRTCVADAnalyzer(sample_rate=self.sample_rate)
-        # Create audio task. It reads audio frames from Daily and push them
-        # internally for VAD processing.
-        if self._params.audio_in_enabled or self._params.vad_enabled:
-            self._audio_in_task = self.create_task(self._audio_in_task_handler())
+        if self._params.audio_in_stream_on_start:
+            self.start_audio_in_streaming()
 
     async def stop(self, frame: EndFrame):
         # Parent stop.
@@ -1200,22 +1204,7 @@ class DailyTransport(BaseTransport):
 
     async def _on_app_message(self, message: Any, sender: str):
         if self._input:
-            if message["type"] in {"raw-audio", "raw-audio-batch"}:
-                data = message["data"]
-                audio_list = data.get(
-                    "base64AudioBatch", [data.get("base64Audio")]
-                )  # Ensure a list
-
-                for base64_audio in filter(None, audio_list):  # Filter out None values
-                    pcm_bytes = base64.b64decode(base64_audio)
-                    frame = InputAudioRawFrame(
-                        audio=pcm_bytes,
-                        sample_rate=data["sampleRate"],
-                        num_channels=data["numChannels"],
-                    )
-                    await self._input.push_audio_frame(frame)
-            else:
-                await self._input.push_app_message(message, sender)
+            await self._input.push_app_message(message, sender)
         await self._call_event_handler("on_app_message", message, sender)
 
     async def _on_call_state_updated(self, state: str):