From d42d02d809e4a437bfc3ddf7c3fe0653205934bd Mon Sep 17 00:00:00 2001 From: unknown Date: Thu, 22 May 2025 11:21:06 +0200 Subject: [PATCH 01/26] Add Google streaming TTS as a base TTS service. Rename non-streaming service to GoogleHttpTTSService. --- src/pipecat/services/google/tts.py | 147 ++++++++++++++++++++++++++++- 1 file changed, 146 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/google/tts.py b/src/pipecat/services/google/tts.py index 9725aa1c3..4a596ede9 100644 --- a/src/pipecat/services/google/tts.py +++ b/src/pipecat/services/google/tts.py @@ -202,7 +202,7 @@ def language_to_google_tts_language(language: Language) -> Optional[str]: return language_map.get(language) -class GoogleTTSService(TTSService): +class GoogleHttpTTSService(TTSService): class InputParams(BaseModel): pitch: Optional[str] = None rate: Optional[str] = None @@ -379,3 +379,148 @@ class GoogleTTSService(TTSService): logger.exception(f"{self} error generating TTS: {e}") error_message = f"TTS generation error: {str(e)}" yield ErrorFrame(error=error_message) + + +class GoogleTTSService(TTSService): + class InputParams(BaseModel): + pitch: Optional[str] = None + rate: Optional[str] = None + volume: Optional[str] = None + emphasis: Optional[Literal["strong", "moderate", "reduced", "none"]] = None + language: Optional[Language] = Language.EN + gender: Optional[Literal["male", "female", "neutral"]] = None + google_style: Optional[Literal["apologetic", "calm", "empathetic", "firm", "lively"]] = None + + def __init__( + self, + *, + credentials: Optional[str] = None, + credentials_path: Optional[str] = None, + voice_id: str = "en-US-Neural2-A", + sample_rate: Optional[int] = None, + params: InputParams = InputParams(), + **kwargs, + ): + super().__init__(sample_rate=sample_rate, **kwargs) + + self._settings = { + "pitch": params.pitch, + "rate": params.rate, + "volume": params.volume, + "emphasis": params.emphasis, + "language": self.language_to_service_language(params.language) + if params.language + else "en-US", + "gender": params.gender, + "google_style": params.google_style, + } + self.set_voice(voice_id) + self._client: texttospeech_v1.TextToSpeechAsyncClient = self._create_client( + credentials, credentials_path + ) + + def _create_client( + self, credentials: Optional[str], credentials_path: Optional[str] + ) -> texttospeech_v1.TextToSpeechAsyncClient: + creds: Optional[service_account.Credentials] = None + + # Create a Google Cloud service account for the Cloud Text-to-Speech API + # Using either the provided credentials JSON string or the path to a service account JSON + # file, create a Google Cloud service account and use it to authenticate with the API. + if credentials: + # Use provided credentials JSON string + json_account_info = json.loads(credentials) + creds = service_account.Credentials.from_service_account_info(json_account_info) + elif credentials_path: + # Use service account JSON file if provided + creds = service_account.Credentials.from_service_account_file(credentials_path) + else: + try: + creds, project_id = default( + scopes=["https://www.googleapis.com/auth/cloud-platform"] + ) + except GoogleAuthError: + pass + + if not creds: + raise ValueError("No valid credentials provided.") + + return texttospeech_v1.TextToSpeechAsyncClient(credentials=creds) + + def can_generate_metrics(self) -> bool: + return True + + def language_to_service_language(self, language: Language) -> Optional[str]: + return language_to_google_tts_language(language) + + @traced_tts + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + # Chirp3 and Journey voices only. Does not support SSML. + logger.debug(f"{self}: Generating TTS [{text}]") + + try: + await self.start_ttfb_metrics() + + voice = texttospeech_v1.VoiceSelectionParams( + language_code=self._settings["language"], name=self._voice_id + ) + + streaming_config = texttospeech_v1.StreamingSynthesizeConfig( + voice=voice, + streaming_audio_config=texttospeech_v1.StreamingAudioConfig( + audio_encoding=texttospeech_v1.AudioEncoding.PCM, + sample_rate_hertz=self.sample_rate, + #speaking_rate=self._settings["rate"], + ), + ) + config_request = texttospeech_v1.StreamingSynthesizeRequest( + streaming_config=streaming_config + ) + async def request_generator(): + yield config_request + yield texttospeech_v1.StreamingSynthesizeRequest( + input=texttospeech_v1.StreamingSynthesisInput(text=text) + ) + + streaming_responses = await self._client.streaming_synthesize(request_generator()) + await self.start_tts_usage_metrics(text) + + yield TTSStartedFrame() + + audio_buffer = b"" + CHUNK_SIZE = 1024 + first_chunk_for_ttfb = False + + async for response in streaming_responses: + chunk = response.audio_content + if not chunk: + continue + + if not first_chunk_for_ttfb: + await self.stop_ttfb_metrics() + first_chunk_for_ttfb = True + + audio_buffer += chunk + while len(audio_buffer) >= CHUNK_SIZE: + piece = audio_buffer[:CHUNK_SIZE] + audio_buffer = audio_buffer[CHUNK_SIZE:] + yield TTSAudioRawFrame(piece, self.sample_rate, 1) + await asyncio.sleep(0) + + if audio_buffer: + yield TTSAudioRawFrame(audio_buffer, self.sample_rate, 1) + await asyncio.sleep(0) + + # Add 1 second pause between sentences + silence = b"\x00" * self.sample_rate + yield TTSAudioRawFrame( + audio=silence, sample_rate=self.sample_rate, num_channels=1 + ) + await asyncio.sleep(0) + + yield TTSStoppedFrame() + + except Exception as e: + logger.exception(f"{self} error generating TTS: {e}") + error_message = f"TTS generation error: {str(e)}" + yield ErrorFrame(error=error_message) From 8d4894846d14401e8ff533a7bd67e79fda7af31b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 22 May 2025 15:47:28 -0700 Subject: [PATCH 02/26] pyproject: update to daily-python 0.19.0 --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index b8db368f9..4572e3ba7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -47,7 +47,7 @@ azure = [ "azure-cognitiveservices-speech~=1.42.0"] cartesia = [ "cartesia~=2.0.3", "websockets~=13.1" ] cerebras = [] deepseek = [] -daily = [ "daily-python~=0.18.2" ] +daily = [ "daily-python~=0.19.0" ] deepgram = [ "deepgram-sdk~=3.8.0" ] elevenlabs = [ "websockets~=13.1" ] fal = [ "fal-client~=0.5.9" ] From fcf49e79ccbb0c76937e53abf8e7a0ce39ab6ebc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 16 May 2025 19:03:53 -0700 Subject: [PATCH 03/26] DailyTransport: use participant audio renderers instead of virtual speaker --- CHANGELOG.md | 3 + src/pipecat/processors/frameworks/rtvi.py | 2 +- src/pipecat/transports/base_input.py | 2 +- src/pipecat/transports/services/daily.py | 131 +++++++++------------- 4 files changed, 56 insertions(+), 82 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 564650905..91605bc65 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -76,6 +76,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `DailyTransport` now captures audio from individual participants instead of + the whole room. This allows identifying audio frames per participant. + - Updated the default model for `AnthropicLLMService` to `claude-sonnet-4-20250514`. diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 449bf3e33..7bd54237d 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -843,7 +843,7 @@ class RTVIProcessor(FrameProcessor): 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() + await self._input_transport.start_audio_in_streaming() self._client_ready_id = request_id await self.set_client_ready() diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index f9a27a6d3..0dea8efdc 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -101,7 +101,7 @@ class BaseInputTransport(FrameProcessor): logger.debug(f"Enabling audio on start. {enabled}") self._params.audio_in_stream_on_start = enabled - def start_audio_in_streaming(self): + async def start_audio_in_streaming(self): pass @property diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 095084037..53619a0c0 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -14,14 +14,12 @@ import aiohttp from loguru import logger from pydantic import BaseModel -from pipecat.audio.utils import create_default_resampler from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams from pipecat.frames.frames import ( CancelFrame, EndFrame, ErrorFrame, Frame, - InputAudioRawFrame, InterimTranscriptionFrame, OutputAudioRawFrame, OutputImageRawFrame, @@ -51,7 +49,6 @@ try: VideoFrame, VirtualCameraDevice, VirtualMicrophoneDevice, - VirtualSpeakerDevice, ) except ModuleNotFoundError as e: logger.error(f"Exception: {e}") @@ -323,7 +320,6 @@ class DailyTransportClient(EventHandler): self._camera: Optional[VirtualCameraDevice] = None self._mic: Optional[VirtualMicrophoneDevice] = None - self._speaker: Optional[VirtualSpeakerDevice] = None self._audio_sources: Dict[str, CustomAudioSource] = {} def _camera_name(self): @@ -332,9 +328,6 @@ class DailyTransportClient(EventHandler): def _mic_name(self): return f"mic-{self}" - def _speaker_name(self): - return f"speaker-{self}" - @property def room_url(self) -> str: return self._room_url @@ -365,29 +358,6 @@ class DailyTransportClient(EventHandler): ) await future - async def read_next_audio_frame(self) -> Optional[InputAudioRawFrame]: - if not self._speaker: - return None - - sample_rate = self._in_sample_rate - num_channels = self._params.audio_in_channels - num_frames = int(sample_rate / 100) * 2 # 20ms of audio - - future = self._get_event_loop().create_future() - self._speaker.read_frames(num_frames, completion=completion_callback(future)) - audio = await future - - if len(audio) > 0: - return InputAudioRawFrame( - audio=audio, sample_rate=sample_rate, num_channels=num_channels - ) - else: - # If we don't read any audio it could be there's no participant - # connected. daily-python will return immediately if that's the - # case, so let's sleep for a little bit (i.e. busy wait). - await asyncio.sleep(0.01) - return None - async def register_audio_destination(self, destination: str): self._audio_sources[destination] = await self.add_custom_audio_track(destination) self._client.update_publishing({"customAudio": {destination: True}}) @@ -448,15 +418,6 @@ class DailyTransportClient(EventHandler): non_blocking=True, ) - if self._params.audio_in_enabled and not self._speaker: - self._speaker = Daily.create_speaker_device( - self._speaker_name(), - sample_rate=self._in_sample_rate, - channels=self._params.audio_in_channels, - non_blocking=True, - ) - Daily.select_speaker_device(self._speaker_name()) - async def join(self): # Transport already joined or joining, ignore. if self._joined or self._joining: @@ -694,6 +655,8 @@ class DailyTransportClient(EventHandler): participant_id: str, callback: Callable, audio_source: str = "microphone", + sample_rate: int = 16000, + callback_interval_ms: int = 20, ): # Only enable the desired audio source subscription on this participant. if audio_source in ("microphone", "screenAudio"): @@ -705,14 +668,14 @@ class DailyTransportClient(EventHandler): self._audio_renderers.setdefault(participant_id, {})[audio_source] = callback - logger.info( - f"Starting to capture audio from participant {participant_id} to {audio_source}" - ) + logger.info(f"Starting to capture [{audio_source}] audio from participant {participant_id}") self._client.set_audio_renderer( participant_id, self._audio_data_received, audio_source=audio_source, + sample_rate=sample_rate, + callback_interval_ms=callback_interval_ms, ) async def capture_participant_video( @@ -877,14 +840,24 @@ class DailyTransportClient(EventHandler): # def _audio_data_received(self, participant_id: str, audio_data: AudioData, audio_source: str): + # We don't need to use _call_async_callback() here because we don't care + # about ordering with other events. callback = self._audio_renderers[participant_id][audio_source] - self._call_async_callback(callback, participant_id, audio_data, audio_source) + future = asyncio.run_coroutine_threadsafe( + callback(participant_id, audio_data, audio_source), self._get_event_loop() + ) + future.result() def _video_frame_received( self, participant_id: str, video_frame: VideoFrame, video_source: str ): + # We don't need to use _call_async_callback() here because we don't care + # about ordering with other events. callback = self._video_renderers[participant_id][video_source] - self._call_async_callback(callback, participant_id, video_frame, video_source) + future = asyncio.run_coroutine_threadsafe( + callback(participant_id, video_frame, video_source), self._get_event_loop() + ) + future.result() def _call_async_callback(self, callback, *args): future = asyncio.run_coroutine_threadsafe( @@ -936,11 +909,12 @@ class DailyInputTransport(BaseInputTransport): # Whether we have seen a StartFrame already. self._initialized = False - # Task that gets audio data from a device or the network and queues it - # internally to be processed. - self._audio_in_task = None + # Whether we have started audio streaming. + self._streaming_started = False - self._resampler = create_default_resampler() + # Store the list of participants we should stream. This is necessary in + # case we don't start streaming right away. + self._capture_participant_audio = [] self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer @@ -948,12 +922,17 @@ 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 not self._audio_in_task and self._params.audio_in_enabled: - logger.debug(f"Start receiving audio") - self._audio_in_task = self.create_task(self._audio_in_task_handler()) + async def start_audio_in_streaming(self): + if not self._params.audio_in_enabled: + return + + logger.debug(f"Start receiving audio") + for participant_id, audio_source, sample_rate in self._capture_participant_audio: + await self._client.capture_participant_audio( + participant_id, self._on_participant_audio_data, audio_source, sample_rate + ) + + self._streaming_started = True async def setup(self, setup: FrameProcessorSetup): await super().setup(setup) @@ -983,27 +962,19 @@ class DailyInputTransport(BaseInputTransport): await self.set_transport_ready(frame) if self._params.audio_in_stream_on_start: - self.start_audio_in_streaming() + await self.start_audio_in_streaming() async def stop(self, frame: EndFrame): # Parent stop. await super().stop(frame) # Leave the room. await self._client.leave() - # Stop audio thread. - if self._audio_in_task and self._params.audio_in_enabled: - await self.cancel_task(self._audio_in_task) - self._audio_in_task = None async def cancel(self, frame: CancelFrame): # Parent stop. await super().cancel(frame) # Leave the room. await self._client.leave() - # Stop audio thread. - if self._audio_in_task and self._params.audio_in_enabled: - await self.cancel_task(self._audio_in_task) - self._audio_in_task = None # # FrameProcessor @@ -1034,32 +1005,26 @@ class DailyInputTransport(BaseInputTransport): self, participant_id: str, audio_source: str = "microphone", + sample_rate: int = 16000, ): - await self._client.capture_participant_audio( - participant_id, self._on_participant_audio_data, audio_source - ) + if self._streaming_started: + await self._client.capture_participant_audio( + participant_id, self._on_participant_audio_data, audio_source, sample_rate + ) + else: + self._capture_participant_audio.append((participant_id, audio_source, sample_rate)) async def _on_participant_audio_data( self, participant_id: str, audio: AudioData, audio_source: str ): - resampled = await self._resampler.resample( - audio.audio_frames, audio.sample_rate, self._client.out_sample_rate - ) - frame = UserAudioRawFrame( user_id=participant_id, - audio=resampled, - sample_rate=self._client.out_sample_rate, + audio=audio.audio_frames, + sample_rate=audio.sample_rate, num_channels=audio.num_channels, ) frame.transport_source = audio_source - await self.push_frame(frame) - - async def _audio_in_task_handler(self): - while True: - frame = await self._client.read_next_audio_frame() - if frame: - await self.push_audio_frame(frame) + await self.push_audio_frame(frame) # # Camera in @@ -1376,9 +1341,10 @@ class DailyTransport(BaseTransport): self, participant_id: str, audio_source: str = "microphone", + sample_rate: int = 16000, ): if self._input: - await self._input.capture_participant_audio(participant_id, audio_source) + await self._input.capture_participant_audio(participant_id, audio_source, sample_rate) async def capture_participant_video( self, @@ -1509,6 +1475,11 @@ class DailyTransport(BaseTransport): id = participant["id"] logger.info(f"Participant joined {id}") + if self._input and self._params.audio_in_enabled: + await self._input.capture_participant_audio( + id, "microphone", self._client.in_sample_rate + ) + if not self._other_participant_has_joined: self._other_participant_has_joined = True await self._call_event_handler("on_first_participant_joined", participant) From 69ac70eed86d76d99d04157bd91333887bf8c0ee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 21 May 2025 23:03:55 -0700 Subject: [PATCH 04/26] DailyTransport: replace virtual microphone with custom microphone track --- CHANGELOG.md | 3 + examples/foundational/04a-transports-daily.py | 2 +- src/pipecat/transports/services/daily.py | 68 +++++++++++-------- 3 files changed, 43 insertions(+), 30 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 91605bc65..1f79e8bc8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -76,6 +76,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- `DailyTransport` now uses custom microphone audio tracks instead of virtual + microphones. Now, multiple Daily transports can be used in the same process. + - `DailyTransport` now captures audio from individual participants instead of the whole room. This allows identifying audio frames per participant. diff --git a/examples/foundational/04a-transports-daily.py b/examples/foundational/04a-transports-daily.py index b8125c5da..98060dbab 100644 --- a/examples/foundational/04a-transports-daily.py +++ b/examples/foundational/04a-transports-daily.py @@ -37,9 +37,9 @@ async def main(): token, "Respond bot", DailyParams( + audio_in_enabled=True, audio_out_enabled=True, transcription_enabled=True, - vad_enabled=True, vad_analyzer=SileroVADAnalyzer(), ), ) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 53619a0c0..f48f4cd61 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -44,11 +44,11 @@ try: AudioData, CallClient, CustomAudioSource, + CustomAudioTrack, Daily, EventHandler, VideoFrame, VirtualCameraDevice, - VirtualMicrophoneDevice, ) except ModuleNotFoundError as e: logger.error(f"Exception: {e}") @@ -242,6 +242,12 @@ def completion_callback(future): return _callback +@dataclass +class DailyAudioTrack: + source: CustomAudioSource + track: CustomAudioTrack + + class DailyTransportClient(EventHandler): """Core client for interacting with Daily's API. @@ -319,15 +325,12 @@ class DailyTransportClient(EventHandler): self._out_sample_rate = 0 self._camera: Optional[VirtualCameraDevice] = None - self._mic: Optional[VirtualMicrophoneDevice] = None - self._audio_sources: Dict[str, CustomAudioSource] = {} + self._microphone_track: Optional[DailyAudioTrack] = None + self._custom_audio_tracks: Dict[str, DailyAudioTrack] = {} def _camera_name(self): return f"camera-{self}" - def _mic_name(self): - return f"mic-{self}" - @property def room_url(self) -> str: return self._room_url @@ -359,19 +362,25 @@ class DailyTransportClient(EventHandler): await future async def register_audio_destination(self, destination: str): - self._audio_sources[destination] = await self.add_custom_audio_track(destination) + self._custom_audio_tracks[destination] = await self.add_custom_audio_track(destination) self._client.update_publishing({"customAudio": {destination: True}}) async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None): future = self._get_event_loop().create_future() - if not destination and self._mic: - self._mic.write_frames(frames, completion=completion_callback(future)) - elif destination and destination in self._audio_sources: - source = self._audio_sources[destination] - source.write_frames(frames, completion=completion_callback(future)) + + audio_source: Optional[CustomAudioSource] = None + if not destination and self._microphone_track: + audio_source = self._microphone_track.source + elif destination and destination in self._custom_audio_tracks: + track = self._custom_audio_tracks[destination] + audio_source = track.source + + if audio_source: + audio_source.write_frames(frames, completion=completion_callback(future)) else: logger.warning(f"{self} unable to write audio frames to destination [{destination}]") future.set_result(None) + await future async def write_raw_video_frame( @@ -410,13 +419,10 @@ class DailyTransportClient(EventHandler): color_format=self._params.video_out_color_format, ) - if self._params.audio_out_enabled and not self._mic: - self._mic = Daily.create_microphone_device( - self._mic_name(), - sample_rate=self._out_sample_rate, - channels=self._params.audio_out_channels, - non_blocking=True, - ) + if self._params.audio_out_enabled and not self._microphone_track: + audio_source = CustomAudioSource(self._out_sample_rate, self._params.audio_out_channels) + audio_track = CustomAudioTrack(audio_source) + self._microphone_track = DailyAudioTrack(source=audio_source, track=audio_track) async def join(self): # Transport already joined or joining, ignore. @@ -501,12 +507,11 @@ class DailyTransportClient(EventHandler): "microphone": { "isEnabled": microphone_enabled, "settings": { - "deviceId": self._mic_name(), - "customConstraints": { - "autoGainControl": {"exact": False}, - "echoCancellation": {"exact": False}, - "noiseSuppression": {"exact": False}, - }, + "customTrack": { + "id": self._microphone_track.track.id + if self._microphone_track + else "no-microphone-track" + } }, }, }, @@ -553,7 +558,7 @@ class DailyTransportClient(EventHandler): await self._stop_transcription() # Remove any custom tracks, if any. - for track_name, _ in self._audio_sources.items(): + for track_name, _ in self._custom_audio_tracks.items(): await self.remove_custom_audio_track(track_name) try: @@ -703,19 +708,24 @@ class DailyTransportClient(EventHandler): color_format=color_format, ) - async def add_custom_audio_track(self, track_name: str) -> CustomAudioSource: + async def add_custom_audio_track(self, track_name: str) -> DailyAudioTrack: future = self._get_event_loop().create_future() audio_source = CustomAudioSource(self._out_sample_rate, 1) + + audio_track = CustomAudioTrack(audio_source) + self._client.add_custom_audio_track( track_name=track_name, - audio_source=audio_source, + audio_track=audio_track, completion=completion_callback(future), ) await future - return audio_source + track = DailyAudioTrack(source=audio_source, track=audio_track) + + return track async def remove_custom_audio_track(self, track_name: str): future = self._get_event_loop().create_future() From c3cfd1f0ceb967409972df0814c49056ea3ed5dd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 23 May 2025 00:38:40 -0700 Subject: [PATCH 05/26] DailyTransport: process audio, video and event callbacks in separate tasks --- CHANGELOG.md | 2 + src/pipecat/transports/services/daily.py | 134 +++++++++++++---------- 2 files changed, 80 insertions(+), 56 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1f79e8bc8..70ca7bd0e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -136,6 +136,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Performance +- `DailyTransport`: process audio, video and events in separate tasks. + - Don't create event handler tasks if no user event handlers have been registered. diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index f48f4cd61..d58737131 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -309,16 +309,21 @@ class DailyTransportClient(EventHandler): self._client: CallClient = CallClient(event_handler=self) - # We use a separate task to execute the callbacks, otherwise if we call - # a `CallClient` function and wait for its completion this will - # currently result in a deadlock. This is because `_call_async_callback` - # can be used inside `CallClient` event handlers which are holding the - # GIL in `daily-python`. So if the `callback` passed here makes a - # `CallClient` call and waits for it to finish using completions (and a - # future) we will deadlock because completions use event handlers (which - # are holding the GIL). - self._callback_queue = asyncio.Queue() - self._callback_task = None + # We use separate tasks to execute callbacks (events, audio or + # video). In the case of events, if we call a `CallClient` function + # inside the callback and wait for its completion this will result in a + # deadlock (because we haven't exited the event callback). The deadlocks + # occur because `daily-python` is holding the GIL when calling the + # callbacks. So, if our callback handler makes a `CallClient` call and + # waits for it to finish using completions (and a future) we will + # deadlock because completions use event handlers (which are holding the + # GIL). + self._event_queue = asyncio.Queue() + self._audio_queue = asyncio.Queue() + self._video_queue = asyncio.Queue() + self._event_task = None + self._audio_task = None + self._video_task = None # Input and ouput sample rates. They will be initialize on setup(). self._in_sample_rate = 0 @@ -394,15 +399,21 @@ class DailyTransportClient(EventHandler): return self._task_manager = setup.task_manager - self._callback_task = self._task_manager.create_task( - self._callback_task_handler(), - f"{self}::callback_task", + self._event_task = self._task_manager.create_task( + self._callback_task_handler(self._event_queue), + f"{self}::event_callback_task", ) async def cleanup(self): - if self._callback_task and self._task_manager: - await self._task_manager.cancel_task(self._callback_task) - self._callback_task = None + if self._event_task and self._task_manager: + await self._task_manager.cancel_task(self._event_task) + self._event_task = None + if self._audio_task and self._task_manager: + await self._task_manager.cancel_task(self._audio_task) + self._audio_task = None + if self._video_task and self._task_manager: + await self._task_manager.cancel_task(self._video_task) + self._video_task = None # Make sure we don't block the event loop in case `client.release()` # takes extra time. await self._get_event_loop().run_in_executor(self._executor, self._cleanup) @@ -411,18 +422,26 @@ class DailyTransportClient(EventHandler): self._in_sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate - if self._params.video_out_enabled and not self._camera: + if self._params.video_out_enabled and not self._camera and self._task_manager: self._camera = Daily.create_camera_device( self._camera_name(), width=self._params.video_out_width, height=self._params.video_out_height, color_format=self._params.video_out_color_format, ) + self._audio_task = self._task_manager.create_task( + self._callback_task_handler(self._video_queue), + f"{self}::video_callback_task", + ) - if self._params.audio_out_enabled and not self._microphone_track: + if self._params.audio_out_enabled and not self._microphone_track and self._task_manager: audio_source = CustomAudioSource(self._out_sample_rate, self._params.audio_out_channels) audio_track = CustomAudioTrack(audio_source) self._microphone_track = DailyAudioTrack(source=audio_source, track=audio_track) + self._audio_task = self._task_manager.create_task( + self._callback_task_handler(self._audio_queue), + f"{self}::audio_callback_task", + ) async def join(self): # Transport already joined or joining, ignore. @@ -772,57 +791,57 @@ class DailyTransportClient(EventHandler): # def on_active_speaker_changed(self, participant): - self._call_async_callback(self._callbacks.on_active_speaker_changed, participant) + self._call_event_callback(self._callbacks.on_active_speaker_changed, participant) def on_app_message(self, message: Any, sender: str): - self._call_async_callback(self._callbacks.on_app_message, message, sender) + self._call_event_callback(self._callbacks.on_app_message, message, sender) def on_call_state_updated(self, state: str): - self._call_async_callback(self._callbacks.on_call_state_updated, state) + self._call_event_callback(self._callbacks.on_call_state_updated, state) def on_dialin_connected(self, data: Any): - self._call_async_callback(self._callbacks.on_dialin_connected, data) + self._call_event_callback(self._callbacks.on_dialin_connected, data) def on_dialin_ready(self, sip_endpoint: str): - self._call_async_callback(self._callbacks.on_dialin_ready, sip_endpoint) + self._call_event_callback(self._callbacks.on_dialin_ready, sip_endpoint) def on_dialin_stopped(self, data: Any): - self._call_async_callback(self._callbacks.on_dialin_stopped, data) + self._call_event_callback(self._callbacks.on_dialin_stopped, data) def on_dialin_error(self, data: Any): - self._call_async_callback(self._callbacks.on_dialin_error, data) + self._call_event_callback(self._callbacks.on_dialin_error, data) def on_dialin_warning(self, data: Any): - self._call_async_callback(self._callbacks.on_dialin_warning, data) + self._call_event_callback(self._callbacks.on_dialin_warning, data) def on_dialout_answered(self, data: Any): - self._call_async_callback(self._callbacks.on_dialout_answered, data) + self._call_event_callback(self._callbacks.on_dialout_answered, data) def on_dialout_connected(self, data: Any): - self._call_async_callback(self._callbacks.on_dialout_connected, data) + self._call_event_callback(self._callbacks.on_dialout_connected, data) def on_dialout_stopped(self, data: Any): - self._call_async_callback(self._callbacks.on_dialout_stopped, data) + self._call_event_callback(self._callbacks.on_dialout_stopped, data) def on_dialout_error(self, data: Any): - self._call_async_callback(self._callbacks.on_dialout_error, data) + self._call_event_callback(self._callbacks.on_dialout_error, data) def on_dialout_warning(self, data: Any): - self._call_async_callback(self._callbacks.on_dialout_warning, data) + self._call_event_callback(self._callbacks.on_dialout_warning, data) def on_participant_joined(self, participant): - self._call_async_callback(self._callbacks.on_participant_joined, participant) + self._call_event_callback(self._callbacks.on_participant_joined, participant) def on_participant_left(self, participant, reason): - self._call_async_callback(self._callbacks.on_participant_left, participant, reason) + self._call_event_callback(self._callbacks.on_participant_left, participant, reason) def on_participant_updated(self, participant): - self._call_async_callback(self._callbacks.on_participant_updated, participant) + self._call_event_callback(self._callbacks.on_participant_updated, participant) def on_transcription_started(self, status): logger.debug(f"Transcription started: {status}") self._transcription_status = status - self._call_async_callback(self.update_transcription, self._transcription_ids) + self._call_event_callback(self.update_transcription, self._transcription_ids) def on_transcription_stopped(self, stopped_by, stopped_by_error): logger.debug("Transcription stopped") @@ -831,55 +850,58 @@ class DailyTransportClient(EventHandler): logger.error(f"Transcription error: {message}") def on_transcription_message(self, message): - self._call_async_callback(self._callbacks.on_transcription_message, message) + self._call_event_callback(self._callbacks.on_transcription_message, message) def on_recording_started(self, status): logger.debug(f"Recording started: {status}") - self._call_async_callback(self._callbacks.on_recording_started, status) + self._call_event_callback(self._callbacks.on_recording_started, status) def on_recording_stopped(self, stream_id): logger.debug(f"Recording stopped: {stream_id}") - self._call_async_callback(self._callbacks.on_recording_stopped, stream_id) + self._call_event_callback(self._callbacks.on_recording_stopped, stream_id) def on_recording_error(self, stream_id, message): logger.error(f"Recording error for {stream_id}: {message}") - self._call_async_callback(self._callbacks.on_recording_error, stream_id, message) + self._call_event_callback(self._callbacks.on_recording_error, stream_id, message) # # Daily (CallClient callbacks) # def _audio_data_received(self, participant_id: str, audio_data: AudioData, audio_source: str): - # We don't need to use _call_async_callback() here because we don't care - # about ordering with other events. callback = self._audio_renderers[participant_id][audio_source] - future = asyncio.run_coroutine_threadsafe( - callback(participant_id, audio_data, audio_source), self._get_event_loop() - ) - future.result() + self._call_audio_callback(callback, participant_id, audio_data, audio_source) def _video_frame_received( self, participant_id: str, video_frame: VideoFrame, video_source: str ): - # We don't need to use _call_async_callback() here because we don't care - # about ordering with other events. callback = self._video_renderers[participant_id][video_source] + self._call_video_callback(callback, participant_id, video_frame, video_source) + + # + # Queue callbacks handling + # + + def _call_audio_callback(self, callback, *args): + self._call_async_callback(self._audio_queue, callback, *args) + + def _call_video_callback(self, callback, *args): + self._call_async_callback(self._video_queue, callback, *args) + + def _call_event_callback(self, callback, *args): + self._call_async_callback(self._event_queue, callback, *args) + + def _call_async_callback(self, queue: asyncio.Queue, callback, *args): future = asyncio.run_coroutine_threadsafe( - callback(participant_id, video_frame, video_source), self._get_event_loop() + queue.put((callback, *args)), self._get_event_loop() ) future.result() - def _call_async_callback(self, callback, *args): - future = asyncio.run_coroutine_threadsafe( - self._callback_queue.put((callback, *args)), self._get_event_loop() - ) - future.result() - - async def _callback_task_handler(self): + async def _callback_task_handler(self, queue: asyncio.Queue): while True: # Wait to process any callback until we are joined. await self._joined_event.wait() - (callback, *args) = await self._callback_queue.get() + (callback, *args) = await queue.get() await callback(*args) def _get_event_loop(self) -> asyncio.AbstractEventLoop: From 30d67a78eb745b7ec24233dc64cdce38f06257e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 23 May 2025 00:41:22 -0700 Subject: [PATCH 06/26] examples(chatbot-audio-recording): use same sample rate to avoid downsampling --- examples/chatbot-audio-recording/bot.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 5428a0d2f..128c97c7e 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -128,7 +128,14 @@ async def main(): ] ) - task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=16000, + audio_out_sample_rate=16000, + allow_interruptions=True, + ), + ) @audiobuffer.event_handler("on_audio_data") async def on_audio_data(buffer, audio, sample_rate, num_channels): From 7f74c2465c77c8dcac1f8c68af0a0f26c3414f55 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 23 May 2025 01:26:15 -0700 Subject: [PATCH 07/26] SileroVADAnalyzer: improve non-matching sample rate log --- src/pipecat/audio/vad/silero.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/pipecat/audio/vad/silero.py b/src/pipecat/audio/vad/silero.py index 26d9368a8..cb1dc7631 100644 --- a/src/pipecat/audio/vad/silero.py +++ b/src/pipecat/audio/vad/silero.py @@ -138,7 +138,9 @@ class SileroVADAnalyzer(VADAnalyzer): def set_sample_rate(self, sample_rate: int): if sample_rate != 16000 and sample_rate != 8000: - raise ValueError("Silero VAD sample rate needs to be 16000 or 8000") + raise ValueError( + f"Silero VAD sample rate needs to be 16000 or 8000 (sample rate: {sample_rate})" + ) super().set_sample_rate(sample_rate) From bfe9952c9a49ae915d434a02f971849688707834 Mon Sep 17 00:00:00 2001 From: unknown Date: Fri, 23 May 2025 12:11:08 +0200 Subject: [PATCH 08/26] Remove sleep(0), add doc string etc. --- src/pipecat/services/google/tts.py | 84 ++++++++++++++++++------------ 1 file changed, 52 insertions(+), 32 deletions(-) diff --git a/src/pipecat/services/google/tts.py b/src/pipecat/services/google/tts.py index 4a596ede9..ce56913cd 100644 --- a/src/pipecat/services/google/tts.py +++ b/src/pipecat/services/google/tts.py @@ -210,21 +210,23 @@ class GoogleHttpTTSService(TTSService): emphasis: Optional[Literal["strong", "moderate", "reduced", "none"]] = None language: Optional[Language] = Language.EN gender: Optional[Literal["male", "female", "neutral"]] = None - google_style: Optional[Literal["apologetic", "calm", "empathetic", "firm", "lively"]] = None + google_style: Optional[ + Literal["apologetic", "calm", "empathetic", "firm", "lively"] + ] = None def __init__( self, *, credentials: Optional[str] = None, credentials_path: Optional[str] = None, - voice_id: str = "en-US-Neural2-A", + voice_id: str = "en-US-Chirp3-HD-Charon", sample_rate: Optional[int] = None, params: Optional[InputParams] = None, **kwargs, ): super().__init__(sample_rate=sample_rate, **kwargs) - params = params or GoogleTTSService.InputParams() + params = params or GoogleHttpTTSService.InputParams() self._settings = { "pitch": params.pitch, @@ -253,10 +255,14 @@ class GoogleHttpTTSService(TTSService): if credentials: # Use provided credentials JSON string json_account_info = json.loads(credentials) - creds = service_account.Credentials.from_service_account_info(json_account_info) + creds = service_account.Credentials.from_service_account_info( + json_account_info + ) elif credentials_path: # Use service account JSON file if provided - creds = service_account.Credentials.from_service_account_file(credentials_path) + creds = service_account.Credentials.from_service_account_file( + credentials_path + ) else: try: creds, project_id = default( @@ -371,7 +377,6 @@ class GoogleHttpTTSService(TTSService): await self.stop_ttfb_metrics() frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) yield frame - await asyncio.sleep(0) # Allow other tasks to run yield TTSStoppedFrame() @@ -382,14 +387,37 @@ class GoogleHttpTTSService(TTSService): class GoogleTTSService(TTSService): + """Text-to-Speech service using Google Cloud Text-to-Speech API. + + Converts text to speech using Google's TTS models with streaming synthesis + for low latency. Supports multiple languages and voices. + + Args: + credentials: JSON string containing Google Cloud service account credentials. + credentials_path: Path to Google Cloud service account JSON file. + voice_id: Google TTS voice identifier (e.g., "en-US-Chirp3-HD-Charon"). + sample_rate: Audio sample rate in Hz. + params: Language only. + + Notes: + Requires Google Cloud credentials via service account JSON, file path, or + default application credentials (GOOGLE_APPLICATION_CREDENTIALS env var). + Only Chirp 3 HD and Journey voices are supported. Use GoogleHttpTTSService for other voices. + + Example: + ```python + tts = GoogleTTSService( + credentials_path="/path/to/service-account.json", + voice_id="en-US-Chirp3-HD-Charon", + params=GoogleTTSService.InputParams( + language=Language.EN_US, + ) + ) + ``` + """ + class InputParams(BaseModel): - pitch: Optional[str] = None - rate: Optional[str] = None - volume: Optional[str] = None - emphasis: Optional[Literal["strong", "moderate", "reduced", "none"]] = None language: Optional[Language] = Language.EN - gender: Optional[Literal["male", "female", "neutral"]] = None - google_style: Optional[Literal["apologetic", "calm", "empathetic", "firm", "lively"]] = None def __init__( self, @@ -403,16 +431,12 @@ class GoogleTTSService(TTSService): ): super().__init__(sample_rate=sample_rate, **kwargs) + params = params or GoogleTTSService.InputParams() + self._settings = { - "pitch": params.pitch, - "rate": params.rate, - "volume": params.volume, - "emphasis": params.emphasis, "language": self.language_to_service_language(params.language) if params.language else "en-US", - "gender": params.gender, - "google_style": params.google_style, } self.set_voice(voice_id) self._client: texttospeech_v1.TextToSpeechAsyncClient = self._create_client( @@ -430,10 +454,14 @@ class GoogleTTSService(TTSService): if credentials: # Use provided credentials JSON string json_account_info = json.loads(credentials) - creds = service_account.Credentials.from_service_account_info(json_account_info) + creds = service_account.Credentials.from_service_account_info( + json_account_info + ) elif credentials_path: # Use service account JSON file if provided - creds = service_account.Credentials.from_service_account_file(credentials_path) + creds = service_account.Credentials.from_service_account_file( + credentials_path + ) else: try: creds, project_id = default( @@ -455,7 +483,6 @@ class GoogleTTSService(TTSService): @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - # Chirp3 and Journey voices only. Does not support SSML. logger.debug(f"{self}: Generating TTS [{text}]") try: @@ -470,19 +497,21 @@ class GoogleTTSService(TTSService): streaming_audio_config=texttospeech_v1.StreamingAudioConfig( audio_encoding=texttospeech_v1.AudioEncoding.PCM, sample_rate_hertz=self.sample_rate, - #speaking_rate=self._settings["rate"], ), ) config_request = texttospeech_v1.StreamingSynthesizeRequest( streaming_config=streaming_config ) + async def request_generator(): yield config_request yield texttospeech_v1.StreamingSynthesizeRequest( input=texttospeech_v1.StreamingSynthesisInput(text=text) ) - streaming_responses = await self._client.streaming_synthesize(request_generator()) + streaming_responses = await self._client.streaming_synthesize( + request_generator() + ) await self.start_tts_usage_metrics(text) yield TTSStartedFrame() @@ -505,18 +534,9 @@ class GoogleTTSService(TTSService): piece = audio_buffer[:CHUNK_SIZE] audio_buffer = audio_buffer[CHUNK_SIZE:] yield TTSAudioRawFrame(piece, self.sample_rate, 1) - await asyncio.sleep(0) if audio_buffer: yield TTSAudioRawFrame(audio_buffer, self.sample_rate, 1) - await asyncio.sleep(0) - - # Add 1 second pause between sentences - silence = b"\x00" * self.sample_rate - yield TTSAudioRawFrame( - audio=silence, sample_rate=self.sample_rate, num_channels=1 - ) - await asyncio.sleep(0) yield TTSStoppedFrame() From 88e8fcdaca2cd70d9c49a812acc148c38fba8f46 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 21 May 2025 15:56:38 -0700 Subject: [PATCH 09/26] observers: added UserBotLatencyLogObserver --- CHANGELOG.md | 4 ++ .../foundational/29-turn-tracking-observer.py | 2 + .../loggers/user_bot_latency_log_observer.py | 50 +++++++++++++++++++ 3 files changed, 56 insertions(+) create mode 100644 src/pipecat/observers/loggers/user_bot_latency_log_observer.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 564650905..28e85beb2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `UserBotLatencyLogObserver`. This is an observer that logs the latency + between when the user stops speaking and when the bot starts speaking. This + gives you an initial idea on how quickly the AI services respond. + - Added `SarvamTTSService`, which implements Sarvam AI's TTS API: https://docs.sarvam.ai/api-reference-docs/text-to-speech/convert. diff --git a/examples/foundational/29-turn-tracking-observer.py b/examples/foundational/29-turn-tracking-observer.py index 0e9c7acec..08c39fe55 100644 --- a/examples/foundational/29-turn-tracking-observer.py +++ b/examples/foundational/29-turn-tracking-observer.py @@ -11,6 +11,7 @@ from dotenv import load_dotenv from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.observers.loggers.user_bot_latency_log_observer import UserBotLatencyLogObserver from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -76,6 +77,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac enable_usage_metrics=True, report_only_initial_ttfb=True, ), + observers=[UserBotLatencyLogObserver()], ) turn_observer = task.turn_tracking_observer diff --git a/src/pipecat/observers/loggers/user_bot_latency_log_observer.py b/src/pipecat/observers/loggers/user_bot_latency_log_observer.py new file mode 100644 index 000000000..f601a9e9e --- /dev/null +++ b/src/pipecat/observers/loggers/user_bot_latency_log_observer.py @@ -0,0 +1,50 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import time + +from loguru import logger + +from pipecat.frames.frames import ( + BotStartedSpeakingFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, +) +from pipecat.observers.base_observer import BaseObserver, FramePushed +from pipecat.processors.frame_processor import FrameDirection + + +class UserBotLatencyLogObserver(BaseObserver): + """Observer that logs the latency between when the user stops speaking and + when the bot starts speaking. + + This helps measure how quickly the AI services respond. + + """ + + def __init__(self): + super().__init__() + self._processed_frames = set() + self._user_stopped_time = 0 + + async def on_push_frame(self, data: FramePushed): + # Only process downstream frames + if data.direction != FrameDirection.DOWNSTREAM: + return + + # Skip already processed frames + if data.frame.id in self._processed_frames: + return + + self._processed_frames.add(data.frame.id) + + if isinstance(data.frame, UserStartedSpeakingFrame): + self._user_stopped_time = 0 + elif isinstance(data.frame, UserStoppedSpeakingFrame): + self._user_stopped_time = time.time() + elif isinstance(data.frame, BotStartedSpeakingFrame) and self._user_stopped_time: + latency = time.time() - self._user_stopped_time + logger.debug(f"⏱️ LATENCY FROM USER STOPPED SPEAKING TO BOT STARTED SPEAKING: {latency}") From 50f024c6f9feada2e8b30766100c71be60c7c9e1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 23 May 2025 11:25:54 -0700 Subject: [PATCH 10/26] LiveKitTransport: use UserAudioRawFrame instead of InputAudioRawFrame --- src/pipecat/transports/services/livekit.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 53a9b1789..143163ed3 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -17,13 +17,13 @@ from pipecat.frames.frames import ( AudioRawFrame, CancelFrame, EndFrame, - InputAudioRawFrame, OutputAudioRawFrame, StartFrame, TransportMessageFrame, TransportMessageUrgentFrame, + UserAudioRawFrame, ) -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup +from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams @@ -411,7 +411,8 @@ class LiveKitInputTransport(BaseInputTransport): pipecat_audio_frame = await self._convert_livekit_audio_to_pipecat( audio_frame_event ) - input_audio_frame = InputAudioRawFrame( + input_audio_frame = UserAudioRawFrame( + user_id=participant_id, audio=pipecat_audio_frame.audio, sample_rate=pipecat_audio_frame.sample_rate, num_channels=pipecat_audio_frame.num_channels, From 6ca6ff37c92be65f035387c61028f7cc51ecc2a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 23 May 2025 17:54:25 -0700 Subject: [PATCH 11/26] DailyTransport: fix video task variable --- src/pipecat/transports/services/daily.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index d58737131..aa35ac279 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -429,7 +429,7 @@ class DailyTransportClient(EventHandler): height=self._params.video_out_height, color_format=self._params.video_out_color_format, ) - self._audio_task = self._task_manager.create_task( + self._video_task = self._task_manager.create_task( self._callback_task_handler(self._video_queue), f"{self}::video_callback_task", ) From 4db5b186944cd13cf6ba20de0fe8cdece7172851 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 23 May 2025 18:28:56 -0700 Subject: [PATCH 12/26] BaseOutputTransport: don't block event loop during image resize --- CHANGELOG.md | 3 +++ src/pipecat/transports/base_output.py | 33 ++++++++++++++++++--------- 2 files changed, 25 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6414c4d61..35b173c3d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -127,6 +127,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a `DailyTransport` issue that would cause images needing resize to block + the event loop. + - Fixed an issue with `ElevenLabsTTSService` where changing the model or voice while the service is running wasn't working. diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 81492b84d..f26602ed7 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -8,6 +8,7 @@ import asyncio import itertools import sys import time +from concurrent.futures import ThreadPoolExecutor from typing import Any, AsyncGenerator, Dict, List, Mapping, Optional from loguru import logger @@ -234,6 +235,9 @@ class BaseOutputTransport(FrameProcessor): self._audio_chunk_size = audio_chunk_size self._params = params + # This is to resize images. We only need to resize one image at a time. + self._executor = ThreadPoolExecutor(max_workers=1) + # Buffer to keep track of incoming audio. self._audio_buffer = bytearray() @@ -558,18 +562,25 @@ class BaseOutputTransport(FrameProcessor): self._video_queue.task_done() async def _draw_image(self, frame: OutputImageRawFrame): - desired_size = (self._params.video_out_width, self._params.video_out_height) + def resize_frame(frame: OutputImageRawFrame) -> OutputImageRawFrame: + desired_size = (self._params.video_out_width, self._params.video_out_height) - # TODO: we should refactor in the future to support dynamic resolutions - # which is kind of what happens in P2P connections. - # We need to add support for that inside the DailyTransport - if frame.size != desired_size: - image = Image.frombytes(frame.format, frame.size, frame.image) - resized_image = image.resize(desired_size) - # logger.warning(f"{frame} does not have the expected size {desired_size}, resizing") - frame = OutputImageRawFrame( - resized_image.tobytes(), resized_image.size, resized_image.format - ) + # TODO: we should refactor in the future to support dynamic resolutions + # which is kind of what happens in P2P connections. + # We need to add support for that inside the DailyTransport + if frame.size != desired_size: + image = Image.frombytes(frame.format, frame.size, frame.image) + resized_image = image.resize(desired_size) + # logger.warning(f"{frame} does not have the expected size {desired_size}, resizing") + frame = OutputImageRawFrame( + resized_image.tobytes(), resized_image.size, resized_image.format + ) + + return frame + + frame = await self._transport.get_event_loop().run_in_executor( + self._executor, resize_frame, frame + ) await self._transport.write_raw_video_frame(frame, self._destination) From 86e684156932ee619d080024f22dbc740e61ef56 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 23 May 2025 23:02:37 -0300 Subject: [PATCH 13/26] Creating TavusTransport and TavusTransportClient. --- src/pipecat/transports/services/tavus.py | 532 +++++++++++++++++++++++ 1 file changed, 532 insertions(+) create mode 100644 src/pipecat/transports/services/tavus.py diff --git a/src/pipecat/transports/services/tavus.py b/src/pipecat/transports/services/tavus.py new file mode 100644 index 000000000..8d704a242 --- /dev/null +++ b/src/pipecat/transports/services/tavus.py @@ -0,0 +1,532 @@ +import asyncio +import base64 +import time +from functools import partial +from typing import Any, Awaitable, Callable, Mapping, Optional + +import aiohttp +from daily.daily import AudioData +from loguru import logger +from pydantic import BaseModel + +from pipecat.audio.utils import create_default_resampler +from pipecat.frames.frames import ( + CancelFrame, + EndFrame, + Frame, + InputAudioRawFrame, + OutputImageRawFrame, + StartFrame, + StartInterruptionFrame, + TransportMessageFrame, + TransportMessageUrgentFrame, + TTSStartedFrame, + TTSStoppedFrame, +) +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup +from pipecat.transports.base_input import BaseInputTransport +from pipecat.transports.base_output import BaseOutputTransport +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.services.daily import ( + DailyCallbacks, + DailyParams, + DailyTransportClient, +) + + +class TavusApi: + """ + A helper class for interacting with the Tavus API (v2). + """ + + BASE_URL = "https://tavusapi.com/v2" + + def __init__(self, api_key: str, session: aiohttp.ClientSession): + """ + Initialize the TavusApi client. + + Args: + api_key (str): Tavus API key. + session (aiohttp.ClientSession): An aiohttp session for making HTTP requests. + """ + self._api_key = api_key + self._session = session + self._headers = {"Content-Type": "application/json", "x-api-key": self._api_key} + + async def create_conversation(self, replica_id: str, persona_id: str) -> dict: + logger.debug(f"Creating Tavus conversation: replica={replica_id}, persona={persona_id}") + url = f"{self.BASE_URL}/conversations" + payload = { + "replica_id": replica_id, + "persona_id": persona_id, + } + async with self._session.post(url, headers=self._headers, json=payload) as r: + r.raise_for_status() + response = await r.json() + logger.debug(f"Created Tavus conversation: {response}") + return response + + async def end_conversation(self, conversation_id: str): + if conversation_id is None: + return + + url = f"{self.BASE_URL}/conversations/{conversation_id}/end" + async with self._session.post(url, headers=self._headers) as r: + r.raise_for_status() + logger.debug(f"Ended Tavus conversation {conversation_id}") + + async def get_persona_name(self, persona_id: str) -> str: + url = f"{self.BASE_URL}/personas/{persona_id}" + async with self._session.get(url, headers=self._headers) as r: + r.raise_for_status() + response = await r.json() + logger.debug(f"Fetched Tavus persona: {response}") + return response["persona_name"] + + +class TavusCallbacks(BaseModel): + """Callback handlers for the Tavus events. + + Attributes: + on_participant_joined: Called when a participant joins. + on_participant_left: Called when a participant leaves. + """ + + on_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]] + on_participant_left: Callable[[Mapping[str, Any], str], Awaitable[None]] + + +class TavusParams(DailyParams): + """Configuration parameters for the Tavus transport.""" + + audio_in_enabled: bool = True + audio_out_enabled: bool = True + microphone_out_enabled: bool = False + + +class TavusTransportClient: + """ + A transport client that integrates a Pipecat Bot with the Tavus platform by managing + conversation sessions using the Tavus API. + + This client uses `TavusApi` to interact with the Tavus backend services. When a conversation + is started via `TavusApi`, Tavus provides a `roomURL` that can be used to connect the Pipecat Bot + into the same virtual room where the TavusBot is operating. + + Args: + bot_name (str): The name of the Pipecat bot instance. + params (TavusParams): Optional parameters for Tavus operation. Defaults to `TavusParams()`. + 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 + 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. + """ + + def __init__( + self, + *, + bot_name: str, + params: TavusParams = TavusParams(), + 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 + session: aiohttp.ClientSession, + ) -> None: + self._bot_name = bot_name + self._api = TavusApi(api_key, session) + 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 + + async def _initialize(self) -> str: + response = await self._api.create_conversation(self._replica_id, self._persona_id) + self._conversation_id = response["conversation_id"] + return response["conversation_url"] + + async def setup(self, setup: FrameProcessorSetup): + if self._conversation_id is not None: + return + try: + room_url = await self._initialize() + daily_callbacks = DailyCallbacks( + on_active_speaker_changed=partial( + self._on_handle_callback, "on_active_speaker_changed" + ), + on_joined=self._on_joined, + on_left=self._on_left, + on_error=partial(self._on_handle_callback, "on_error"), + on_app_message=partial(self._on_handle_callback, "on_app_message"), + on_call_state_updated=partial(self._on_handle_callback, "on_call_state_updated"), + on_client_connected=partial(self._on_handle_callback, "on_client_connected"), + on_client_disconnected=partial(self._on_handle_callback, "on_client_disconnected"), + on_dialin_connected=partial(self._on_handle_callback, "on_dialin_connected"), + on_dialin_ready=partial(self._on_handle_callback, "on_dialin_ready"), + on_dialin_stopped=partial(self._on_handle_callback, "on_dialin_stopped"), + on_dialin_error=partial(self._on_handle_callback, "on_dialin_error"), + on_dialin_warning=partial(self._on_handle_callback, "on_dialin_warning"), + on_dialout_answered=partial(self._on_handle_callback, "on_dialout_answered"), + on_dialout_connected=partial(self._on_handle_callback, "on_dialout_connected"), + on_dialout_stopped=partial(self._on_handle_callback, "on_dialout_stopped"), + on_dialout_error=partial(self._on_handle_callback, "on_dialout_error"), + on_dialout_warning=partial(self._on_handle_callback, "on_dialout_warning"), + on_participant_joined=self._callbacks.on_participant_joined, + on_participant_left=self._callbacks.on_participant_left, + on_participant_updated=partial(self._on_handle_callback, "on_participant_updated"), + on_transcription_message=partial( + self._on_handle_callback, "on_transcription_message" + ), + on_recording_started=partial(self._on_handle_callback, "on_recording_started"), + on_recording_stopped=partial(self._on_handle_callback, "on_recording_stopped"), + on_recording_error=partial(self._on_handle_callback, "on_recording_error"), + ) + self._client = DailyTransportClient( + room_url, None, "Pipecat", self._params, daily_callbacks, self._bot_name + ) + await self._client.setup(setup) + except Exception as e: + logger.error(f"Failed to setup TavusTransportClient: {e}") + await self._api.end_conversation(self._conversation_id) + + async def cleanup(self): + if self._client is None: + return + await self._client.cleanup() + self._client = None + + async def _on_joined(self, data): + logger.debug("TavusTransportClient joined!") + + async def _on_left(self): + logger.debug("TavusTransportClient left!") + + async def _on_handle_callback(self, event_name, *args, **kwargs): + logger.trace(f"[Callback] {event_name} called with args={args}, kwargs={kwargs}") + + async def get_persona_name(self) -> str: + return await self._api.get_persona_name(self._persona_id) + + async def start(self, frame: StartFrame): + logger.debug("TavusTransportClient start invoked!") + await self._client.start(frame) + await self._client.join() + + async def stop(self): + await self._client.leave() + await self._api.end_conversation(self._conversation_id) + + async def capture_participant_video( + self, + participant_id: str, + callback: Callable, + framerate: int = 30, + video_source: str = "camera", + color_format: str = "RGB", + ): + await self._client.capture_participant_video( + participant_id, callback, framerate, video_source, color_format + ) + + async def capture_participant_audio( + self, + participant_id: str, + callback: Callable, + audio_source: str = "microphone", + sample_rate: int = 16000, + callback_interval_ms: int = 20, + ): + await self._client.capture_participant_audio( + participant_id, callback, audio_source, sample_rate, callback_interval_ms + ) + + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + await self._client.send_message(frame) + + @property + def out_sample_rate(self) -> int: + return self._client.out_sample_rate + + @property + 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={ + "message_type": "conversation", + "event_type": "conversation.interrupt", + "conversation_id": self._conversation_id, + } + ) + 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): + await self._client.update_subscriptions( + participant_settings=participant_settings, profile_settings=profile_settings + ) + + async def write_raw_audio_frames(self, frames: bytes, destination: Optional[str] = None): + await self._client.write_raw_audio_frames(frames, destination) + + +class TavusInputTransport(BaseInputTransport): + def __init__( + self, + client: TavusTransportClient, + params: TransportParams, + **kwargs, + ): + super().__init__(params, **kwargs) + self._client = client + self._params = params + self._resampler = create_default_resampler() + + async def setup(self, setup: FrameProcessorSetup): + await super().setup(setup) + await self._client.setup(setup) + + async def cleanup(self): + await super().cleanup() + await self._client.cleanup() + + async def start(self, frame: StartFrame): + await super().start(frame) + await self._client.start(frame) + await self.set_transport_ready(frame) + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._client.stop() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._client.stop() + + async def start_capturing_audio(self, participant): + if self._params.audio_in_enabled: + logger.info( + f"TavusTransportClient start capturing audio for participant {participant['id']}" + ) + await self._client.capture_participant_audio( + participant_id=participant["id"], + callback=self._on_participant_audio_data, + sample_rate=self._client.in_sample_rate, + ) + + async def _on_participant_audio_data( + self, participant_id: str, audio: AudioData, audio_source: str + ): + frame = InputAudioRawFrame( + audio=audio.audio_frames, + sample_rate=audio.audio_frames, + num_channels=audio.num_channels, + ) + frame.transport_source = audio_source + await self.push_audio_frame(frame) + + +class TavusOutputTransport(BaseOutputTransport): + def __init__( + self, + client: TavusTransportClient, + params: TransportParams, + **kwargs, + ): + super().__init__(params, **kwargs) + self._client = client + self._params = params + self._samples_sent = 0 + self._start_time = time.time() + + async def setup(self, setup: FrameProcessorSetup): + await super().setup(setup) + await self._client.setup(setup) + + async def cleanup(self): + await super().cleanup() + await self._client.cleanup() + + 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) + + async def stop(self, frame: EndFrame): + await super().stop(frame) + await self._client.stop() + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + await self._client.stop() + + async def send_message(self, frame: TransportMessageFrame | TransportMessageUrgentFrame): + logger.info(f"TavusOutputTransport sending message {frame}") + await self._client.send_message(frame) + + 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() + if wait > 0: + await asyncio.sleep(wait) + + await self._client.encode_audio_and_send(frames, False, self._current_idx_str) + + # Update timestamp based on number of samples sent + self._samples_sent += len(frames) // 2 # 2 bytes per sample (16-bit) + + async def write_raw_video_frame( + self, frame: OutputImageRawFrame, destination: Optional[str] = None + ): + pass + + +class TavusTransport(BaseTransport): + """ + Transport implementation for Tavus video calls. + + When used, the Pipecat bot joins the same virtual room as the Tavus Avatar and the user. + This is achieved by using `TavusTransportClient`, which initiates the conversation via + `TavusApi` and obtains a room URL that all participants connect to. + + Args: + bot_name (str): The name of the Pipecat bot. + 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. + 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. + """ + + def __init__( + self, + bot_name: str, + 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 + params: TavusParams = TavusParams(), + input_name: Optional[str] = None, + output_name: Optional[str] = None, + ): + 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, + ) + self._client = TavusTransportClient( + bot_name="Pipecat", + callbacks=callbacks, + api_key=api_key, + replica_id=replica_id, + persona_id=persona_id, + session=session, + params=params, + ) + self._input: Optional[TavusInputTransport] = None + self._output: Optional[TavusOutputTransport] = None + self._tavus_participant_id = None + + # Register supported handlers. The user will only be able to register + # these handlers. + self._register_event_handler("on_client_connected") + self._register_event_handler("on_client_disconnected") + + async def _on_participant_left(self, participant, reason): + persona_name = await self._client.get_persona_name() + if participant.get("info", {}).get("userName", "") != persona_name: + await self._on_client_disconnected(participant) + + 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"] + else: + await self._on_client_connected(participant) + if self._tavus_participant_id: + logger.debug(f"Ignoring {self._tavus_participant_id}'s microphone") + await self.update_subscriptions( + participant_settings={ + self._tavus_participant_id: { + "media": {"microphone": "unsubscribed"}, + } + } + ) + if self._input: + await self._input.start_capturing_audio(participant) + + async def update_subscriptions(self, participant_settings=None, profile_settings=None): + await self._client.update_subscriptions( + participant_settings=participant_settings, + profile_settings=profile_settings, + ) + + def input(self) -> FrameProcessor: + if not self._input: + self._input = TavusInputTransport(client=self._client, params=self._params) + return self._input + + def output(self) -> FrameProcessor: + if not self._output: + self._output = TavusOutputTransport(client=self._client, params=self._params) + return self._output + + async def _on_client_connected(self, participant: Any): + await self._call_event_handler("on_client_connected", participant) + + async def _on_client_disconnected(self, participant: Any): + await self._call_event_handler("on_client_disconnected", participant) From e0d1381f8777eee8fd0e6a25661874f5ae9a90a3 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 23 May 2025 23:02:49 -0300 Subject: [PATCH 14/26] Refactoring the TavusVideoService to allow to work with any transport. --- src/pipecat/services/tavus/video.py | 200 ++++++++++++++++++---------- 1 file changed, 128 insertions(+), 72 deletions(-) diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index 3699ba512..2d7ba6cd4 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -7,10 +7,11 @@ """This module implements Tavus as a sink transport layer""" import asyncio -import base64 +import time from typing import Optional import aiohttp +from daily.daily import AudioData, VideoFrame from loguru import logger from pipecat.audio.utils import create_default_resampler @@ -18,19 +19,38 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, + OutputAudioRawFrame, + OutputImageRawFrame, StartFrame, StartInterruptionFrame, - TransportMessageUrgentFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ) -from pipecat.processors.frame_processor import FrameDirection +from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup from pipecat.services.ai_service import AIService +from pipecat.transports.services.tavus import TavusCallbacks, TavusParams, TavusTransportClient class TavusVideoService(AIService): - """Class to send base64 encoded audio to Tavus""" + """ + Service class that proxies audio to Tavus and receives both audio and video in return. + + It uses the `TavusTransportClient` to manage the session and handle communication. When + audio is sent, Tavus responds with both audio and video streams, which are then routed + through Pipecat’s media pipeline. + + In use cases such as with `DailyTransport`, this results in two distinct virtual rooms: + - **Tavus room**: Contains the Tavus Avatar and the Pipecat Bot. + - **User room**: Contains the Pipecat Bot and the user. + + 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. + session (aiohttp.ClientSession): Async HTTP session used for communication with Tavus. + **kwargs: Additional arguments passed to the parent `AIService` class. + """ def __init__( self, @@ -39,54 +59,101 @@ class TavusVideoService(AIService): replica_id: str, persona_id: str = "pipecat0", # Use `pipecat0` so that your TTS voice is used in place of the Tavus persona session: aiohttp.ClientSession, - sample_rate: int = 16000, **kwargs, ) -> None: super().__init__(**kwargs) self._api_key = api_key + self._session = session self._replica_id = replica_id self._persona_id = persona_id - self._session = session - self._sample_rate = sample_rate + + self._other_participant_has_joined = False + self._client: Optional[TavusTransportClient] = None self._conversation_id: str - self._resampler = create_default_resampler() self._audio_buffer = bytearray() self._queue = asyncio.Queue() self._send_task: Optional[asyncio.Task] = None - async def initialize(self) -> str: - url = "https://tavusapi.com/v2/conversations" - headers = {"Content-Type": "application/json", "x-api-key": self._api_key} - payload = { - "replica_id": self._replica_id, - "persona_id": self._persona_id, - } - async with self._session.post(url, headers=headers, json=payload) as r: - r.raise_for_status() - response_json = await r.json() + async def setup(self, setup: FrameProcessorSetup): + await super().setup(setup) + callbacks = TavusCallbacks( + on_participant_joined=self._on_participant_joined, + on_participant_left=self._on_participant_left, + ) + self._client = TavusTransportClient( + bot_name="Pipecat", + callbacks=callbacks, + api_key=self._api_key, + replica_id=self._replica_id, + persona_id=self._persona_id, + session=self._session, + params=TavusParams( + audio_out_enabled=True, + microphone_out_enabled=False, + audio_in_enabled=True, + video_in_enabled=True, + video_out_enabled=True, + ), + ) + await self._client.setup(setup) - logger.debug(f"TavusVideoService joined {response_json['conversation_url']}") - self._conversation_id = response_json["conversation_id"] - return response_json["conversation_url"] + async def cleanup(self): + await super().cleanup() + await self._client.cleanup() + self._client = None + + async def _on_participant_left(self, participant, reason): + participant_id = participant["id"] + logger.info(f"Participant left {participant_id}, reason: {reason}") + + async def _on_participant_joined(self, participant): + participant_id = participant["id"] + logger.info(f"Participant joined {participant_id}") + if not self._other_participant_has_joined: + self._other_participant_has_joined = True + await self._client.capture_participant_video( + participant_id, self._on_participant_video_frame, 30 + ) + await self._client.capture_participant_audio( + participant_id=participant_id, + callback=self._on_participant_audio_data, + sample_rate=self._client.out_sample_rate, + ) + + async def _on_participant_video_frame( + self, participant_id: str, video_frame: VideoFrame, video_source: str + ): + frame = OutputImageRawFrame( + image=video_frame.buffer, + size=(video_frame.width, video_frame.height), + format=video_frame.color_format, + ) + frame.transport_source = video_source + await self.push_frame(frame) + + async def _on_participant_audio_data( + self, participant_id: str, audio: AudioData, audio_source: str + ): + frame = OutputAudioRawFrame( + audio=audio.audio_frames, + sample_rate=audio.sample_rate, + num_channels=audio.num_channels, + ) + frame.transport_source = audio_source + await self.push_frame(frame) def can_generate_metrics(self) -> bool: return True async def get_persona_name(self) -> str: - url = f"https://tavusapi.com/v2/personas/{self._persona_id}" - headers = {"Content-Type": "application/json", "x-api-key": self._api_key} - async with self._session.get(url, headers=headers) as r: - r.raise_for_status() - response_json = await r.json() - - logger.debug(f"TavusVideoService persona grabbed {response_json}") - return response_json["persona_name"] + return await self._client.get_persona_name() async def start(self, frame: StartFrame): await super().start(frame) + await self._client.start(frame) await self._create_send_task() async def stop(self, frame: EndFrame): @@ -112,7 +179,7 @@ class TavusVideoService(AIService): elif isinstance(frame, TTSAudioRawFrame): await self._queue_audio(frame.audio, frame.sample_rate, done=False) elif isinstance(frame, TTSStoppedFrame): - await self._queue_audio(b"\x00\x00", self._sample_rate, done=True) + await self._queue_audio(b"\x00\x00", self._client.in_sample_rate, done=True) await self.stop_ttfb_metrics() await self.stop_processing_metrics() else: @@ -121,13 +188,11 @@ class TavusVideoService(AIService): async def _handle_interruptions(self): await self._cancel_send_task() await self._create_send_task() - await self._send_interrupt_message() + await self._client.send_interrupt_message() async def _end_conversation(self): - url = f"https://tavusapi.com/v2/conversations/{self._conversation_id}/end" - headers = {"Content-Type": "application/json", "x-api-key": self._api_key} - async with self._session.post(url, headers=headers) as r: - r.raise_for_status() + 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)) @@ -142,6 +207,15 @@ 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 @@ -149,57 +223,39 @@ class TavusVideoService(AIService): # 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. - MAX_CHUNK_SIZE = int((self._sample_rate * 2) / 20) - SLEEP_TIME = 1 / 20 + sample_rate = self._client.out_sample_rate + MAX_CHUNK_SIZE = int((sample_rate * 2) / 20) audio_buffer = bytearray() + samples_sent = 0 + start_time = time.time() + while True: (audio, in_rate, done) = await self._queue.get() if done: # Send any remaining audio. if len(audio_buffer) > 0: - await self._encode_audio_and_send(bytes(audio_buffer), done) - await self._encode_audio_and_send(audio, done) + await self._client.encode_audio_and_send( + bytes(audio_buffer), done, self._current_idx_str + ) + await self._client.encode_audio_and_send(audio, done, self._current_idx_str) audio_buffer.clear() else: - audio = await self._resampler.resample(audio, in_rate, self._sample_rate) + audio = await self._resampler.resample(audio, in_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:] - await self._encode_audio_and_send(bytes(chunk), done) - await asyncio.sleep(SLEEP_TIME) - async def _encode_audio_and_send(self, audio: bytes, done: bool): - """Encodes audio to base64 and sends it to Tavus""" - audio_base64 = base64.b64encode(audio).decode("utf-8") - logger.trace(f"{self}: sending {len(audio)} bytes") - await self._send_audio_message(audio_base64, done=done) + # Compute wait time for synchronization + wait = start_time + (samples_sent / sample_rate) - time.time() + if wait > 0: + await asyncio.sleep(wait) - async def _send_interrupt_message(self) -> None: - transport_frame = TransportMessageUrgentFrame( - message={ - "message_type": "conversation", - "event_type": "conversation.interrupt", - "conversation_id": self._conversation_id, - } - ) - await self.push_frame(transport_frame) + await self._client.encode_audio_and_send( + bytes(chunk), done, self._current_idx_str + ) - async def _send_audio_message(self, audio_base64: str, done: bool): - transport_frame = TransportMessageUrgentFrame( - message={ - "message_type": "conversation", - "event_type": "conversation.echo", - "conversation_id": self._conversation_id, - "properties": { - "modality": "audio", - "inference_id": self._current_idx_str, - "audio": audio_base64, - "done": done, - "sample_rate": self._sample_rate, - }, - } - ) - await self.push_frame(transport_frame) + # Update timestamp based on number of samples sent + samples_sent += len(chunk) // 2 # 2 bytes per sample (16-bit) From 1ebca353132adebd7ebe87a80d02402029683d52 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 23 May 2025 23:03:04 -0300 Subject: [PATCH 15/26] Queuing the app messages if the SmallWebrtcTransport is not connected yet. --- src/pipecat/transports/network/webrtc_connection.py | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/src/pipecat/transports/network/webrtc_connection.py b/src/pipecat/transports/network/webrtc_connection.py index 3981460e7..49aa2b1da 100644 --- a/src/pipecat/transports/network/webrtc_connection.py +++ b/src/pipecat/transports/network/webrtc_connection.py @@ -144,6 +144,7 @@ class SmallWebRTCConnection(BaseObject): self._renegotiation_in_progress = False self._last_received_time = None self._message_queue = [] + self._pending_app_messages = [] def _setup_listeners(self): @self._pc.on("datachannel") @@ -170,7 +171,11 @@ class SmallWebRTCConnection(BaseObject): if json_message["type"] == SIGNALLING_TYPE and json_message.get("message"): self._handle_signalling_message(json_message["message"]) else: - await self._call_event_handler("app-message", json_message) + if self.is_connected(): + await self._call_event_handler("app-message", json_message) + else: + logger.debug("Client not connected. Queuing app-message.") + self._pending_app_messages.append(json_message) except Exception as e: logger.exception(f"Error parsing JSON message {message}, {e}") @@ -225,6 +230,9 @@ class SmallWebRTCConnection(BaseObject): # If we already connected, trigger again the connected event if self.is_connected(): await self._call_event_handler("connected") + logger.debug("Flushing pending app-messages") + for message in self._pending_app_messages: + await self._call_event_handler("app-message", message) # We are renegotiating here, because likely we have loose the first video frames # and aiortc does not handle that pretty well. video_input_track = self.video_input_track() @@ -293,6 +301,7 @@ class SmallWebRTCConnection(BaseObject): if self._pc: await self._pc.close() self._message_queue.clear() + self._pending_app_messages.clear() self._track_map = {} def get_answer(self): From 4a3404883fac1dc47126d500b073a8e8523b7e6f Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 23 May 2025 23:03:16 -0300 Subject: [PATCH 16/26] Creating a Tavus example using the new TavusTransport. --- .../21-tavus-layer-tavus-transport.py | 112 ++++++++++++++++++ 1 file changed, 112 insertions(+) create mode 100644 examples/foundational/21-tavus-layer-tavus-transport.py diff --git a/examples/foundational/21-tavus-layer-tavus-transport.py b/examples/foundational/21-tavus-layer-tavus-transport.py new file mode 100644 index 000000000..c9bcd2501 --- /dev/null +++ b/examples/foundational/21-tavus-layer-tavus-transport.py @@ -0,0 +1,112 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import os +import sys + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.google.llm import GoogleLLMService +from pipecat.transports.services.tavus import TavusParams, TavusTransport + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(): + async with aiohttp.ClientSession() as session: + transport = TavusTransport( + bot_name="Pipecat bot", + api_key=os.getenv("TAVUS_API_KEY"), + replica_id=os.getenv("TAVUS_REPLICA_ID"), + session=session, + params=TavusParams( + audio_in_enabled=True, + audio_out_enabled=True, + microphone_out_enabled=False, + vad_analyzer=SileroVADAnalyzer(), + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab", + ) + + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=16000, + audio_out_sample_rate=24000, + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, participant): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": "Start by greeting the user and ask how you can help.", + } + ) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, participant): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) From 6346ca1a84d534d02b0e21129557003613e97bf9 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 23 May 2025 23:03:24 -0300 Subject: [PATCH 17/26] Creating a Tavus example using the SmallWebRTCTransport. --- .../21a-tavus-layer-small-webrtc.py | 125 ++++++++++++++++++ 1 file changed, 125 insertions(+) create mode 100644 examples/foundational/21a-tavus-layer-small-webrtc.py diff --git a/examples/foundational/21a-tavus-layer-small-webrtc.py b/examples/foundational/21a-tavus-layer-small-webrtc.py new file mode 100644 index 000000000..2f557b5cd --- /dev/null +++ b/examples/foundational/21a-tavus-layer-small-webrtc.py @@ -0,0 +1,125 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import argparse +import os + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.cartesia.tts import CartesiaTTSService +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.google.llm import GoogleLLMService +from pipecat.services.tavus.video import TavusVideoService +from pipecat.transports.base_transport import TransportParams +from pipecat.transports.network.small_webrtc import SmallWebRTCTransport +from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection + +load_dotenv(override=True) + + +async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespace): + logger.info(f"Starting bot") + async with aiohttp.ClientSession() as session: + transport = SmallWebRTCTransport( + webrtc_connection=webrtc_connection, + params=TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + video_out_enabled=True, + video_out_is_live=True, + vad_analyzer=SileroVADAnalyzer(), + video_out_width=1280, + video_out_height=720, + ), + ) + + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab", + ) + + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + + tavus = TavusVideoService( + api_key=os.getenv("TAVUS_API_KEY"), + replica_id=os.getenv("TAVUS_REPLICA_ID"), + session=session, + ) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + tavus, # Tavus output layer + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + audio_in_sample_rate=16000, + audio_out_sample_rate=24000, + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": "Start by greeting the user and ask how you can help.", + } + ) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + + @transport.event_handler("on_client_closed") + async def on_client_closed(transport, client): + logger.info(f"Client closed connection") + await task.cancel() + + runner = PipelineRunner(handle_sigint=False) + + await runner.run(task) + + +if __name__ == "__main__": + from run import main + + main() From 8221dd594ee318cd32d6cc3b18a119961e8a076c Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 23 May 2025 23:03:40 -0300 Subject: [PATCH 18/26] Creating a Tavus example using the DailyTransport. --- ....py => 21b-tavus-layer-daily-transport.py} | 72 ++++++++----------- 1 file changed, 31 insertions(+), 41 deletions(-) rename examples/foundational/{21-tavus-layer.py => 21b-tavus-layer-daily-transport.py} (64%) diff --git a/examples/foundational/21-tavus-layer.py b/examples/foundational/21b-tavus-layer-daily-transport.py similarity index 64% rename from examples/foundational/21-tavus-layer.py rename to examples/foundational/21b-tavus-layer-daily-transport.py index ffe95e074..564828136 100644 --- a/examples/foundational/21-tavus-layer.py +++ b/examples/foundational/21b-tavus-layer-daily-transport.py @@ -7,9 +7,9 @@ import asyncio import os import sys -from typing import Any, Mapping import aiohttp +from daily_runner import configure from dotenv import load_dotenv from loguru import logger @@ -20,7 +20,7 @@ from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.cartesia.tts import CartesiaTTSService from pipecat.services.deepgram.stt import DeepgramSTTService -from pipecat.services.openai.llm import OpenAILLMService +from pipecat.services.google.llm import GoogleLLMService from pipecat.services.tavus.video import TavusVideoService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -32,23 +32,20 @@ logger.add(sys.stderr, level="DEBUG") async def main(): async with aiohttp.ClientSession() as session: - tavus = TavusVideoService( - api_key=os.getenv("TAVUS_API_KEY"), - replica_id=os.getenv("TAVUS_REPLICA_ID"), - session=session, - ) - - # get persona, look up persona_name, set this as the bot name to ignore - persona_name = await tavus.get_persona_name() - room_url = await tavus.initialize() + (room_url, token) = await configure(session) transport = DailyTransport( - room_url=room_url, - token=None, - bot_name="Pipecat bot", - params=DailyParams( + room_url, + token, + "Pipecat bot", + DailyParams( audio_in_enabled=True, + audio_out_enabled=True, + video_out_enabled=True, + video_out_is_live=True, vad_analyzer=SileroVADAnalyzer(), + video_out_width=1280, + video_out_height=720, ), ) @@ -59,7 +56,13 @@ async def main(): voice_id="a167e0f3-df7e-4d52-a9c3-f949145efdab", ) - llm = OpenAILLMService(model="gpt-4o-mini") + llm = GoogleLLMService(api_key=os.getenv("GOOGLE_API_KEY")) + + tavus = TavusVideoService( + api_key=os.getenv("TAVUS_API_KEY"), + replica_id=os.getenv("TAVUS_REPLICA_ID"), + session=session, + ) messages = [ { @@ -87,10 +90,8 @@ async def main(): task = PipelineTask( pipeline, params=PipelineParams( - # We just use 16000 because that's what Tavus is expecting and - # we avoid resampling. audio_in_sample_rate=16000, - audio_out_sample_rate=16000, + audio_out_sample_rate=24000, allow_interruptions=True, enable_metrics=True, enable_usage_metrics=True, @@ -98,33 +99,22 @@ async def main(): ), ) - @transport.event_handler("on_participant_joined") - async def on_participant_joined( - transport: DailyTransport, participant: Mapping[str, Any] - ) -> None: - # Ignore the Tavus replica's microphone - if participant.get("info", {}).get("userName", "") == persona_name: - logger.debug(f"Ignoring {participant['id']}'s microphone") - await transport.update_subscriptions( - participant_settings={ - participant["id"]: { - "media": {"microphone": "unsubscribed"}, - } - } - ) - - if participant.get("info", {}).get("userName", "") != persona_name: - # Kick off the conversation. - messages.append( - {"role": "system", "content": "Please introduce yourself to the user."} - ) - await task.queue_frames([context_aggregator.user().get_context_frame()]) + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + # Kick off the conversation. + messages.append( + { + "role": "system", + "content": "Start by greeting the user and ask how you can help.", + } + ) + await task.queue_frames([context_aggregator.user().get_context_frame()]) @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): await task.cancel() - runner = PipelineRunner() + runner = PipelineRunner(handle_sigint=False) await runner.run(task) From 31492831ccbeafc39610e49e5dbb2981545045e1 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Fri, 23 May 2025 23:04:04 -0300 Subject: [PATCH 19/26] Updating the changelog and readme to reflect the Tavus changes. --- CHANGELOG.md | 10 ++++++++++ examples/foundational/README.md | 2 +- 2 files changed, 11 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 35b173c3d..5746a04b7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `TavusTransport`, a new transport implementation compatible with any + Pipecat pipeline. When using the `TavusTransport`the Pipecat bot will + connect in the same room as the Tavus Avatar and the user. + - Added `UserBotLatencyLogObserver`. This is an observer that logs the latency between when the user stops speaking and when the bot starts speaking. This gives you an initial idea on how quickly the AI services respond. @@ -80,6 +84,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- ⚠️Refactored the `TavusVideoService`, so it acts like a proxy, sending audio to + Tavus and receiving both audio and video. This will make `TavusVideoService` usable + with any Pipecat pipeline and with any transport. This is a **breaking change**, + check the `examples/foundational/21a-tavus-layer-small-webrtc.py` to see how to + use it. + - `DailyTransport` now uses custom microphone audio tracks instead of virtual microphones. Now, multiple Daily transports can be used in the same process. diff --git a/examples/foundational/README.md b/examples/foundational/README.md index 14323d189..3c77a7bef 100644 --- a/examples/foundational/README.md +++ b/examples/foundational/README.md @@ -95,7 +95,7 @@ Depending on what you're trying to build, these learning paths will guide you th - **[18-gstreamer-filesrc.py](./18-gstreamer-filesrc.py)**: GStreamer video streaming (Video processing) - **[19-openai-realtime-beta.py](./19-openai-realtime-beta.py)**: OpenAI Speech-to-Speech (Direct S2S, Function calls) -- **[21-tavus-layer.py](./21-tavus-layer.py)**: Tavus digital twin (Avatar integration) +- **[21-tavus-layer-tavus-transport.py](./21-tavus-layer-tavus-transport.py)**: Tavus digital twin (Avatar integration) - **[27-simli-layer.py](./27-simli-layer.py)**: Simli avatar integration (Video synchronization) ### Performance & Optimization From 85c096df0b6cf3337500cbfddf41ec3efb1f372a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 23 May 2025 19:26:58 -0700 Subject: [PATCH 20/26] DailyTransport: create audio/video input tasks when input flag is enabled --- src/pipecat/transports/services/daily.py | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index aa35ac279..7fe3ed0dd 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -422,26 +422,29 @@ class DailyTransportClient(EventHandler): self._in_sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate - if self._params.video_out_enabled and not self._camera and self._task_manager: + if self._params.audio_in_enabled and not self._audio_task and self._task_manager: + self._audio_task = self._task_manager.create_task( + self._callback_task_handler(self._audio_queue), + f"{self}::audio_callback_task", + ) + + if self._params.video_in_enabled and not self._video_task and self._task_manager: + self._video_task = self._task_manager.create_task( + self._callback_task_handler(self._video_queue), + f"{self}::video_callback_task", + ) + if self._params.video_out_enabled and not self._camera: self._camera = Daily.create_camera_device( self._camera_name(), width=self._params.video_out_width, height=self._params.video_out_height, color_format=self._params.video_out_color_format, ) - self._video_task = self._task_manager.create_task( - self._callback_task_handler(self._video_queue), - f"{self}::video_callback_task", - ) - if self._params.audio_out_enabled and not self._microphone_track and self._task_manager: + if self._params.audio_out_enabled and not self._microphone_track: audio_source = CustomAudioSource(self._out_sample_rate, self._params.audio_out_channels) audio_track = CustomAudioTrack(audio_source) self._microphone_track = DailyAudioTrack(source=audio_source, track=audio_track) - self._audio_task = self._task_manager.create_task( - self._callback_task_handler(self._audio_queue), - f"{self}::audio_callback_task", - ) async def join(self): # Transport already joined or joining, ignore. From 940926b5ec2c5569f8b983919b9a5643fbabee4d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 23 May 2025 19:29:34 -0700 Subject: [PATCH 21/26] TavusVideoService: no need to enable audio/video outputs --- src/pipecat/services/tavus/video.py | 3 --- 1 file changed, 3 deletions(-) diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index 2d7ba6cd4..8fa30db8a 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -91,11 +91,8 @@ class TavusVideoService(AIService): persona_id=self._persona_id, session=self._session, params=TavusParams( - audio_out_enabled=True, - microphone_out_enabled=False, audio_in_enabled=True, video_in_enabled=True, - video_out_enabled=True, ), ) await self._client.setup(setup) From cd03b9111509e310bab9963b995288ce628ece26 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 24 May 2025 07:23:59 -0400 Subject: [PATCH 22/26] Update README: Add OTel, Add serializers --- README.md | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index ae6c3cc53..3966e8d65 100644 --- a/README.md +++ b/README.md @@ -58,11 +58,12 @@ You can connect to Pipecat from any platform using our official SDKs: | Text-to-Speech | [AWS](https://docs.pipecat.ai/server/services/tts/aws), [Azure](https://docs.pipecat.ai/server/services/tts/azure), [Cartesia](https://docs.pipecat.ai/server/services/tts/cartesia), [Deepgram](https://docs.pipecat.ai/server/services/tts/deepgram), [ElevenLabs](https://docs.pipecat.ai/server/services/tts/elevenlabs), [FastPitch (NVIDIA)](https://docs.pipecat.ai/server/services/tts/fastpitch), [Fish](https://docs.pipecat.ai/server/services/tts/fish), [Google](https://docs.pipecat.ai/server/services/tts/google), [LMNT](https://docs.pipecat.ai/server/services/tts/lmnt), [MiniMax](https://docs.pipecat.ai/server/services/tts/minimax), [Neuphonic](https://docs.pipecat.ai/server/services/tts/neuphonic), [OpenAI](https://docs.pipecat.ai/server/services/tts/openai), [Piper](https://docs.pipecat.ai/server/services/tts/piper), [PlayHT](https://docs.pipecat.ai/server/services/tts/playht), [Rime](https://docs.pipecat.ai/server/services/tts/rime), [Sarvam](https://docs.pipecat.ai/server/services/tts/sarvam), [XTTS](https://docs.pipecat.ai/server/services/tts/xtts) | | Speech-to-Speech | [AWS Nova Sonic](https://docs.pipecat.ai/server/services/s2s/aws), [Gemini Multimodal Live](https://docs.pipecat.ai/server/services/s2s/gemini), [OpenAI Realtime](https://docs.pipecat.ai/server/services/s2s/openai) | | Transport | [Daily (WebRTC)](https://docs.pipecat.ai/server/services/transport/daily), [FastAPI Websocket](https://docs.pipecat.ai/server/services/transport/fastapi-websocket), [SmallWebRTCTransport](https://docs.pipecat.ai/server/services/transport/small-webrtc), [WebSocket Server](https://docs.pipecat.ai/server/services/transport/websocket-server), Local | +| Serializers | [Plivo](https://docs.pipecat.ai/server/utilities/serializers/plivo), [Twilio](https://docs.pipecat.ai/server/utilities/serializers/twilio), [Telnyx](https://docs.pipecat.ai/server/utilities/serializers/telnyx) | | Video | [Tavus](https://docs.pipecat.ai/server/services/video/tavus), [Simli](https://docs.pipecat.ai/server/services/video/simli) | | Memory | [mem0](https://docs.pipecat.ai/server/services/memory/mem0) | | Vision & Image | [fal](https://docs.pipecat.ai/server/services/image-generation/fal), [Google Imagen](https://docs.pipecat.ai/server/services/image-generation/fal), [Moondream](https://docs.pipecat.ai/server/services/vision/moondream) | | Audio Processing | [Silero VAD](https://docs.pipecat.ai/server/utilities/audio/silero-vad-analyzer), [Krisp](https://docs.pipecat.ai/server/utilities/audio/krisp-filter), [Koala](https://docs.pipecat.ai/server/utilities/audio/koala-filter), [Noisereduce](https://docs.pipecat.ai/server/utilities/audio/noisereduce-filter) | -| Analytics & Metrics | [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) | +| Analytics & Metrics | [OpenTelemetry](https://docs.pipecat.ai/server/utilities/opentelemetry), [Sentry](https://docs.pipecat.ai/server/services/analytics/sentry) | 📚 [View full services documentation →](https://docs.pipecat.ai/server/services/supported-services) From b4cd7d79413113f25a2456b763221709190266d5 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sat, 24 May 2025 08:16:33 -0400 Subject: [PATCH 23/26] Langfuse OTel env.example improvements --- examples/open-telemetry/jaeger/README.md | 2 ++ examples/open-telemetry/langfuse/README.md | 2 ++ examples/open-telemetry/langfuse/env.example | 13 ++++++++++--- 3 files changed, 14 insertions(+), 3 deletions(-) diff --git a/examples/open-telemetry/jaeger/README.md b/examples/open-telemetry/jaeger/README.md index a19d3ee9d..57a9bb7d3 100644 --- a/examples/open-telemetry/jaeger/README.md +++ b/examples/open-telemetry/jaeger/README.md @@ -38,6 +38,8 @@ OPENAI_API_KEY=your_key_here pip install -r requirements.txt ``` +> Install only the grpc exporter. If you have a conflict, uninstall the http exporter. + ### 4. Run the Demo ```bash diff --git a/examples/open-telemetry/langfuse/README.md b/examples/open-telemetry/langfuse/README.md index 002a3f64a..c17de8394 100644 --- a/examples/open-telemetry/langfuse/README.md +++ b/examples/open-telemetry/langfuse/README.md @@ -43,6 +43,8 @@ OPENAI_API_KEY=your_key_here pip install -r requirements.txt ``` +> Install only the http exporter. If you have a conflict, uninstall the grpc exporter. + ### 4. Run the Demo ```bash diff --git a/examples/open-telemetry/langfuse/env.example b/examples/open-telemetry/langfuse/env.example index c86485674..622854799 100644 --- a/examples/open-telemetry/langfuse/env.example +++ b/examples/open-telemetry/langfuse/env.example @@ -4,8 +4,15 @@ OPENAI_API_KEY=your_openai_key # Set to any value to enable tracing ENABLE_TRACING=true -# OTLP endpoint (change to us.cloud.langfuse.com if you use the US data region) -OTEL_EXPORTER_OTLP_ENDPOINT=http://cloud.langfuse.com/api/public/otel -OTEL_EXPORTER_OTLP_HEADERS=Authorization=Basic%20 + +# 🇪🇺 EU data region +OTEL_EXPORTER_OTLP_ENDPOINT="https://cloud.langfuse.com/api/public/otel" +# 🇺🇸 US data region +# OTEL_EXPORTER_OTLP_ENDPOINT="https://us.cloud.langfuse.com/api/public/otel" +# 🏠 Local deployment (>= v3.22.0) +# OTEL_EXPORTER_OTLP_ENDPOINT="http://localhost:3000/api/public/otel" + +OTEL_EXPORTER_OTLP_HEADERS="Authorization=Basic " + # Set to any value to enable console output for debugging # OTEL_CONSOLE_EXPORT=true \ No newline at end of file From d3e2a9e5c06938e30bdf077d4b7f88b502a04d4b Mon Sep 17 00:00:00 2001 From: unknown Date: Sat, 24 May 2025 15:14:39 +0200 Subject: [PATCH 24/26] Change default voice and fix formatting --- CHANGELOG.md | 4 ++++ src/pipecat/services/google/tts.py | 26 +++++++------------------- 2 files changed, 11 insertions(+), 19 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6dde0bae5..8208329a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `GoogleHttpTTSService` which uses Google's HTTP TTS API. + - Added `PipelineTask.add_observer()` and `PipelineTask.remove_observer()` to allow mangaging observers at runtime. This is useful for cases where the task is passed around to other code components that might want to observe the @@ -73,6 +75,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Updated `GoogleTTSService` to use Google's streaming TTS API. The default voice also updated to `en-US-Chirp3-HD-Charon`. + - `BaseTextFilter` methods `filter()`, `update_settings()`, `handle_interruption()` and `reset_interruption()` are now async. diff --git a/src/pipecat/services/google/tts.py b/src/pipecat/services/google/tts.py index ce56913cd..e28f9fadb 100644 --- a/src/pipecat/services/google/tts.py +++ b/src/pipecat/services/google/tts.py @@ -210,9 +210,7 @@ class GoogleHttpTTSService(TTSService): emphasis: Optional[Literal["strong", "moderate", "reduced", "none"]] = None language: Optional[Language] = Language.EN gender: Optional[Literal["male", "female", "neutral"]] = None - google_style: Optional[ - Literal["apologetic", "calm", "empathetic", "firm", "lively"] - ] = None + google_style: Optional[Literal["apologetic", "calm", "empathetic", "firm", "lively"]] = None def __init__( self, @@ -255,14 +253,10 @@ class GoogleHttpTTSService(TTSService): if credentials: # Use provided credentials JSON string json_account_info = json.loads(credentials) - creds = service_account.Credentials.from_service_account_info( - json_account_info - ) + creds = service_account.Credentials.from_service_account_info(json_account_info) elif credentials_path: # Use service account JSON file if provided - creds = service_account.Credentials.from_service_account_file( - credentials_path - ) + creds = service_account.Credentials.from_service_account_file(credentials_path) else: try: creds, project_id = default( @@ -424,7 +418,7 @@ class GoogleTTSService(TTSService): *, credentials: Optional[str] = None, credentials_path: Optional[str] = None, - voice_id: str = "en-US-Neural2-A", + voice_id: str = "en-US-Chirp3-HD-Charon", sample_rate: Optional[int] = None, params: InputParams = InputParams(), **kwargs, @@ -454,14 +448,10 @@ class GoogleTTSService(TTSService): if credentials: # Use provided credentials JSON string json_account_info = json.loads(credentials) - creds = service_account.Credentials.from_service_account_info( - json_account_info - ) + creds = service_account.Credentials.from_service_account_info(json_account_info) elif credentials_path: # Use service account JSON file if provided - creds = service_account.Credentials.from_service_account_file( - credentials_path - ) + creds = service_account.Credentials.from_service_account_file(credentials_path) else: try: creds, project_id = default( @@ -509,9 +499,7 @@ class GoogleTTSService(TTSService): input=texttospeech_v1.StreamingSynthesisInput(text=text) ) - streaming_responses = await self._client.streaming_synthesize( - request_generator() - ) + streaming_responses = await self._client.streaming_synthesize(request_generator()) await self.start_tts_usage_metrics(text) yield TTSStartedFrame() From 514716042be06fbe30eab2cd2a32c36b3da40917 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 26 May 2025 08:25:03 -0400 Subject: [PATCH 25/26] Fix mismatched html tag in websocket client example --- examples/websocket-server/index.html | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/websocket-server/index.html b/examples/websocket-server/index.html index ac93c62d7..6f0bd1ada 100644 --- a/examples/websocket-server/index.html +++ b/examples/websocket-server/index.html @@ -10,7 +10,7 @@

Pipecat WebSocket Client Example

-

Loading, wait...

+

Loading, wait...