Merge pull request #1215 from pipecat-ai/instant_voice_demo

Instant voice demo improvements - part 02

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/processors/frameworks/rtvi.py

@@ -5,6 +5,7 @@
 #
 
 import asyncio
+import base64
 from dataclasses import dataclass
 from typing import (
     Any,
@@ -31,6 +32,7 @@ from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
     FunctionCallResultFrame,
+    InputAudioRawFrame,
     InterimTranscriptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
@@ -58,7 +60,9 @@ from pipecat.processors.aggregators.openai_llm_context import (
     OpenAILLMContextFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.transports.base_input import BaseInputTransport
 from pipecat.transports.base_output import BaseOutputTransport
+from pipecat.transports.base_transport import BaseTransport
 from pipecat.utils.string import match_endofsentence
 
 RTVI_PROTOCOL_VERSION = "0.3.0"
@@ -819,6 +823,7 @@ class RTVIProcessor(FrameProcessor):
         self,
         *,
         config: RTVIConfig = RTVIConfig(config=[]),
+        transport: Optional[BaseTransport] = None,
         **kwargs,
     ):
         super().__init__(**kwargs)
@@ -844,6 +849,14 @@ class RTVIProcessor(FrameProcessor):
         self._register_event_handler("on_bot_started")
         self._register_event_handler("on_client_ready")
 
+        self._input_transport = None
+        self._transport = transport
+        if self._transport:
+            input_transport = self._transport.input()
+            if isinstance(input_transport, BaseInputTransport):
+                self._input_transport = input_transport
+                self._input_transport.enable_audio_in_stream_on_start(False)
+
     def observer(self) -> RTVIObserver:
         import warnings
 
@@ -1013,6 +1026,8 @@ class RTVIProcessor(FrameProcessor):
                 case "llm-function-call-result":
                     data = RTVILLMFunctionCallResultData.model_validate(message.data)
                     await self._handle_function_call_result(data)
+                case "raw-audio" | "raw-audio-batch":
+                    await self._handle_audio_buffer(message.data)
 
                 case _:
                     await self._send_error_response(message.id, f"Unsupported type {message.type}")
@@ -1025,9 +1040,34 @@ class RTVIProcessor(FrameProcessor):
             logger.warning(f"Exception processing message: {e}")
 
     async def _handle_client_ready(self, request_id: str):
+        logger.debug("Received client-ready")
+        if self._input_transport:
+            self._input_transport.start_audio_in_streaming()
+
         self._client_ready_id = request_id
         await self.set_client_ready()
 
+    async def _handle_audio_buffer(self, data):
+        if not self._input_transport:
+            return
+
+        # Extract audio batch ensuring it's a list
+        audio_list = data.get("base64AudioBatch") or [data.get("base64Audio")]
+
+        try:
+            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_transport.push_audio_frame(frame)
+
+        except (KeyError, TypeError, ValueError) as e:
+            # Handle missing keys, decoding errors, and invalid types
+            logger.error(f"Error processing audio buffer: {e}")
+
     async def _handle_describe_config(self, request_id: str):
         services = list(self._registered_services.values())
         message = RTVIDescribeConfig(id=request_id, data=RTVIDescribeConfigData(config=services))

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/network/fastapi_websocket.py

@@ -23,6 +23,8 @@ from pipecat.frames.frames import (
     OutputAudioRawFrame,
     StartFrame,
     StartInterruptionFrame,
+    TransportMessageFrame,
+    TransportMessageUrgentFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType
@@ -139,6 +141,9 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
             await self._write_frame(frame)
             self._next_send_time = 0
 
+    async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
+        await self._write_frame(frame)
+
     async def write_raw_audio_frames(self, frames: bytes):
         if self._websocket.client_state != WebSocketState.CONNECTED:
             # Simulate audio playback with a sleep.

src/pipecat/transports/network/websocket_server.py

@@ -193,10 +193,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
             self._next_send_time = 0
 
     async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame):
-        message_frame = TextFrame(
+        await self._write_frame(frame)
-            text=json.dumps(frame.message),
-        )
-        await self._write_frame(message_frame)
 
     async def write_raw_audio_frames(self, frames: bytes):
         if not self._websocket:

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
+        if self._params.audio_in_stream_on_start:
-        # internally for VAD processing.
+            self.start_audio_in_streaming()
-        if self._params.audio_in_enabled or self._params.vad_enabled:
-            self._audio_in_task = self.create_task(self._audio_in_task_handler())
 
     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"}:
+            await self._input.push_app_message(message, sender)
-                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._call_event_handler("on_app_message", message, sender)
 
     async def _on_call_state_updated(self, state: str):