diff --git a/CHANGELOG.md b/CHANGELOG.md index bef74dc06..9f9cd7e0c 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 186a26cf0..5b28846a2 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/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)) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index a44332a65..6bfe86001 100644 --- a/src/pipecat/transports/base_input.py +++ b/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 diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 436e78b94..e07b41c22 100644 --- a/src/pipecat/transports/base_transport.py +++ b/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 diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index 8da8396ff..9c4c170f2 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/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. diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 234f85119..ca5d07d2e 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/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( - text=json.dumps(frame.message), - ) - await self._write_frame(message_frame) + await self._write_frame(frame) async def write_raw_audio_frames(self, frames: bytes): if not self._websocket: diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 99071cf5c..2274b3e8f 100644 --- a/src/pipecat/transports/services/daily.py +++ b/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):