From 426d7ac2133d6c2076dc498abfb7b37783990d08 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 13:11:35 -0800 Subject: [PATCH 01/12] transports: some local audio and tk updates --- examples/foundational/01a-local-audio.py | 4 ++-- examples/foundational/13a-whisper-local.py | 4 ++-- examples/local-input-select-stt/bot.py | 4 ++-- src/pipecat/transports/local/audio.py | 18 ++++++++++-------- src/pipecat/transports/local/tk.py | 13 ++++++++++++- 5 files changed, 28 insertions(+), 15 deletions(-) diff --git a/examples/foundational/01a-local-audio.py b/examples/foundational/01a-local-audio.py index 633697684..20aaedfae 100644 --- a/examples/foundational/01a-local-audio.py +++ b/examples/foundational/01a-local-audio.py @@ -16,7 +16,7 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.services.cartesia import CartesiaTTSService -from pipecat.transports.local.audio import LocalAudioTransport, LocalTransportParams +from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams load_dotenv(override=True) @@ -25,7 +25,7 @@ logger.add(sys.stderr, level="DEBUG") async def main(): - transport = LocalAudioTransport(LocalTransportParams(audio_out_enabled=True)) + transport = LocalAudioTransport(LocalAudioTransportParams(audio_out_enabled=True)) tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), diff --git a/examples/foundational/13a-whisper-local.py b/examples/foundational/13a-whisper-local.py index 2d3cd5a57..c9bf16782 100644 --- a/examples/foundational/13a-whisper-local.py +++ b/examples/foundational/13a-whisper-local.py @@ -16,7 +16,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.whisper import WhisperSTTService -from pipecat.transports.local.audio import LocalAudioTransport, LocalTransportParams +from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams load_dotenv(override=True) @@ -33,7 +33,7 @@ class TranscriptionLogger(FrameProcessor): async def main(): - transport = LocalAudioTransport(LocalTransportParams(audio_in_enabled=True)) + transport = LocalAudioTransport(LocalAudioTransportParams(audio_in_enabled=True)) stt = WhisperSTTService() diff --git a/examples/local-input-select-stt/bot.py b/examples/local-input-select-stt/bot.py index 1cedbf96c..f472152b9 100644 --- a/examples/local-input-select-stt/bot.py +++ b/examples/local-input-select-stt/bot.py @@ -18,7 +18,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.whisper import Model, WhisperSTTService -from pipecat.transports.local.audio import LocalAudioTransport, LocalTransportParams +from pipecat.transports.local.audio import LocalAudioTransport, LocalAudioTransportParams load_dotenv(override=True) @@ -36,7 +36,7 @@ class TranscriptionLogger(FrameProcessor): async def main(input_device: int, output_device: int): transport = LocalAudioTransport( - LocalTransportParams( + LocalAudioTransportParams( audio_in_enabled=True, audio_out_enabled=False, input_device_index=input_device, diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index 30bdaaf79..e7b803f2b 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -26,17 +26,18 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class LocalTransportParams(TransportParams): - input_device_index: int = 0 - output_device_index: int = 0 +class LocalAudioTransportParams(TransportParams): + input_device_index: Optional[int] = None + output_device_index: Optional[int] = None class LocalAudioInputTransport(BaseInputTransport): - _params: LocalTransportParams + _params: LocalAudioTransportParams - def __init__(self, py_audio: pyaudio.PyAudio, params: LocalTransportParams): + def __init__(self, py_audio: pyaudio.PyAudio, params: LocalAudioTransportParams): super().__init__(params) self._py_audio = py_audio + self._in_stream = None self._sample_rate = 0 @@ -77,11 +78,12 @@ class LocalAudioInputTransport(BaseInputTransport): class LocalAudioOutputTransport(BaseOutputTransport): - _params: LocalTransportParams + _params: LocalAudioTransportParams - def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): + def __init__(self, py_audio: pyaudio.PyAudio, params: LocalAudioTransportParams): super().__init__(params) self._py_audio = py_audio + self._out_stream = None self._sample_rate = 0 @@ -117,7 +119,7 @@ class LocalAudioOutputTransport(BaseOutputTransport): class LocalAudioTransport(BaseTransport): - def __init__(self, params: LocalTransportParams): + def __init__(self, params: LocalAudioTransportParams): super().__init__() self._params = params self._pyaudio = pyaudio.PyAudio() diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 40147d39f..73777c573 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -34,8 +34,15 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") +class TkTransportParams(TransportParams): + audio_input_device_index: Optional[int] = None + audio_output_device_index: Optional[int] = None + + class TkInputTransport(BaseInputTransport): - def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): + _params: TkTransportParams + + def __init__(self, py_audio: pyaudio.PyAudio, params: TkTransportParams): super().__init__(params) self._py_audio = py_audio self._in_stream = None @@ -54,6 +61,7 @@ class TkInputTransport(BaseInputTransport): frames_per_buffer=num_frames, stream_callback=self._audio_in_callback, input=True, + input_device_index=self._params.audio_input_device_index, ) self._in_stream.start_stream() @@ -76,6 +84,8 @@ class TkInputTransport(BaseInputTransport): class TkOutputTransport(BaseOutputTransport): + _params: TkTransportParams + def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) self._py_audio = py_audio @@ -103,6 +113,7 @@ class TkOutputTransport(BaseOutputTransport): channels=self._params.audio_out_channels, rate=self._sample_rate, output=True, + output_device_index=self._params.audio_output_device_index, ) self._out_stream.start_stream() From 5126d4de92212e4f3addb5bcb205f5a1f5b8559b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 14:34:38 -0800 Subject: [PATCH 02/12] tts: handle incoming frames pausing/resuming from base TTSService class --- src/pipecat/services/ai_services.py | 21 +++++++++++++++++++++ src/pipecat/services/cartesia.py | 13 ------------- src/pipecat/services/elevenlabs.py | 13 ------------- src/pipecat/services/fish.py | 10 ---------- src/pipecat/services/playht.py | 13 ------------- src/pipecat/services/rime.py | 18 +----------------- 6 files changed, 22 insertions(+), 66 deletions(-) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 0fce2482a..7d49c5676 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -15,6 +15,7 @@ from loguru import logger from pipecat.audio.utils import calculate_audio_volume, exp_smoothing from pipecat.frames.frames import ( AudioRawFrame, + BotStoppedSpeakingFrame, CancelFrame, EndFrame, ErrorFrame, @@ -234,6 +235,7 @@ class TTSService(AIService): self._stop_frame_queue: asyncio.Queue = asyncio.Queue() self._current_sentence: str = "" + self._processing_text: bool = False @property def sample_rate(self) -> int: @@ -299,6 +301,7 @@ class TTSService(AIService): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) + if ( isinstance(frame, TextFrame) and not isinstance(frame, InterimTranscriptionFrame) @@ -308,8 +311,15 @@ class TTSService(AIService): elif isinstance(frame, StartInterruptionFrame): await self._handle_interruption(frame, direction) elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): + # We pause processing incoming frames if the LLM response included + # text (it might be that it's only a function calling response). We + # pause to avoid audio overlapping. + if self._processing_text: + await self.pause_processing_frames() + sentence = self._current_sentence self._current_sentence = "" + self._processing_text = False await self._push_tts_frames(sentence) if isinstance(frame, LLMFullResponseEndFrame): if self._push_text_frames: @@ -317,10 +327,16 @@ class TTSService(AIService): else: await self.push_frame(frame, direction) elif isinstance(frame, TTSSpeakFrame): + # We pause processing incoming frames because we are sending data to + # the TTS. We pause to avoid audio overlapping. + await self.pause_processing_frames() await self._push_tts_frames(frame.text) await self.flush_audio() + self._processing_text = False elif isinstance(frame, TTSUpdateSettingsFrame): await self._update_settings(frame.settings) + elif isinstance(frame, BotStoppedSpeakingFrame): + await self.resume_processing_frames() else: await self.push_frame(frame, direction) @@ -371,6 +387,11 @@ class TTSService(AIService): if not text.strip(): return + # This is just a flag that indicates if we sent something to the TTS + # service. It will be cleared if we sent text because of a TTSSpeakFrame + # or when we received an LLMFullResponseEndFrame + self._processing_text = True + await self.start_processing_metrics() if self._text_filter: self._text_filter.reset_interruption() diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 49bcac2f1..becd39f79 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -274,19 +274,6 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService): else: logger.error(f"{self} error, unknown message type: {msg}") - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # If we received a TTSSpeakFrame and the LLM response included text (it - # might be that it's only a function calling response) we pause - # processing more frames until we receive a BotStoppedSpeakingFrame. - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._context_id: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index b2047d34e..d8acedd6c 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -289,19 +289,6 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # If we received a TTSSpeakFrame and the LLM response included text (it - # might be that it's only a function calling response) we pause - # processing more frames until we receive a BotStoppedSpeakingFrame. - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._started: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def _connect(self): await self._connect_websocket() diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index d61514eb2..e36eaf497 100644 --- a/src/pipecat/services/fish.py +++ b/src/pipecat/services/fish.py @@ -166,16 +166,6 @@ class FishAudioTTSService(TTSService, WebsocketService): except Exception as e: logger.error(f"Error processing message: {e}") - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): await super()._handle_interruption(frame, direction) await self.stop_all_metrics() diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index bd9463f91..3196c306d 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -269,19 +269,6 @@ class PlayHTTTSService(TTSService, WebsocketService): except json.JSONDecodeError: logger.error(f"Invalid JSON message: {message}") - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - - # If we received a TTSSpeakFrame and the LLM response included text (it - # might be that it's only a function calling response) we pause - # processing more frames until we receive a BotStoppedSpeakingFrame. - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._request_id: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: logger.debug(f"Generating TTS: [{text}]") diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 1634578cd..79e557f48 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -126,7 +126,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): # State tracking self._context_id = None # Tracks current turn self._receive_task = None - self._started = False self._cumulative_time = 0 # Accumulates time across messages def can_generate_metrics(self) -> bool: @@ -200,7 +199,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): await self._websocket.send(json.dumps(self._build_eos_msg())) await self._websocket.close() self._websocket = None - self._started = False self._context_id = None except Exception as e: logger.error(f"{self} error closing websocket: {e}") @@ -217,7 +215,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): await self.stop_all_metrics() if self._context_id: await self._get_websocket().send(json.dumps(self._build_clear_msg())) - self._started = False self._context_id = None def _calculate_word_times(self, words: list, starts: list, ends: list) -> list: @@ -300,21 +297,9 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): """Push frame and handle end-of-turn conditions.""" await super().push_frame(frame, direction) if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): - self._started = False if isinstance(frame, TTSStoppedFrame): await self.add_word_timestamps([("LLMFullResponseEndFrame", 0), ("Reset", 0)]) - async def process_frame(self, frame: Frame, direction: FrameDirection): - """Process frames and manage turn state.""" - await super().process_frame(frame, direction) - - if isinstance(frame, TTSSpeakFrame): - await self.pause_processing_frames() - elif isinstance(frame, LLMFullResponseEndFrame) and self._started: - await self.pause_processing_frames() - elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Generate speech from text. @@ -330,10 +315,9 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): await self._connect() try: - if not self._started: + if not self._context_id: await self.start_ttfb_metrics() yield TTSStartedFrame() - self._started = True self._cumulative_time = 0 self._context_id = str(uuid.uuid4()) await self.create_audio_context(self._context_id) From 67da745bb3741db1305a9d04f4163c29e1ddd6e4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 14:45:35 -0800 Subject: [PATCH 03/12] tts: make frame pausing/resuming optional --- src/pipecat/services/ai_services.py | 24 +++++++++++++++++------- src/pipecat/services/cartesia.py | 1 + src/pipecat/services/elevenlabs.py | 1 + src/pipecat/services/fish.py | 2 +- src/pipecat/services/lmnt.py | 1 + src/pipecat/services/playht.py | 1 + src/pipecat/services/rime.py | 1 + 7 files changed, 23 insertions(+), 8 deletions(-) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 7d49c5676..99fd05481 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -76,13 +76,13 @@ class AIService(FrameProcessor): ) for key, value in settings.items(): - print("Update request for:", key, value) + logger.debug("Update request for:", key, value) if key in self._settings: logger.info(f"Updating LLM setting {key} to: [{value}]") self._settings[key] = value elif key in SessionProperties.model_fields: - print("Attempting to update", key, value) + logger.debug("Attempting to update", key, value) try: from pipecat.services.openai_realtime_beta.events import ( @@ -213,6 +213,8 @@ class TTSService(AIService): push_silence_after_stop: bool = False, # if push_silence_after_stop is True, send this amount of audio silence silence_time_s: float = 2.0, + # if True, we will pause processing frames while we are receiving audio + pause_frame_processing: bool = False, # TTS output sample rate sample_rate: Optional[int] = None, text_filter: Optional[BaseTextFilter] = None, @@ -225,6 +227,7 @@ class TTSService(AIService): self._stop_frame_timeout_s: float = stop_frame_timeout_s self._push_silence_after_stop: bool = push_silence_after_stop self._silence_time_s: float = silence_time_s + self._pause_frame_processing: bool = pause_frame_processing self._init_sample_rate = sample_rate self._sample_rate = 0 self._voice_id: str = "" @@ -314,8 +317,7 @@ class TTSService(AIService): # We pause processing incoming frames if the LLM response included # text (it might be that it's only a function calling response). We # pause to avoid audio overlapping. - if self._processing_text: - await self.pause_processing_frames() + await self._maybe_pause_frame_processing() sentence = self._current_sentence self._current_sentence = "" @@ -327,16 +329,16 @@ class TTSService(AIService): else: await self.push_frame(frame, direction) elif isinstance(frame, TTSSpeakFrame): + await self._push_tts_frames(frame.text) # We pause processing incoming frames because we are sending data to # the TTS. We pause to avoid audio overlapping. - await self.pause_processing_frames() - await self._push_tts_frames(frame.text) + await self._maybe_pause_frame_processing() await self.flush_audio() self._processing_text = False elif isinstance(frame, TTSUpdateSettingsFrame): await self._update_settings(frame.settings) elif isinstance(frame, BotStoppedSpeakingFrame): - await self.resume_processing_frames() + await self._maybe_resume_frame_processing() else: await self.push_frame(frame, direction) @@ -367,6 +369,14 @@ class TTSService(AIService): self._text_filter.handle_interruption() await self.push_frame(frame, direction) + async def _maybe_pause_frame_processing(self): + if self._processing_text and self._pause_frame_processing: + await self.pause_processing_frames() + + async def _maybe_resume_frame_processing(self): + if self._pause_frame_processing: + await self.resume_processing_frames() + async def _process_text_frame(self, frame: TextFrame): text: Optional[str] = None if not self._aggregate_sentences: diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index becd39f79..67fc67ebf 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -109,6 +109,7 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService): self, aggregate_sentences=True, push_text_frames=False, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index d8acedd6c..3c8cbb3ed 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -192,6 +192,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): push_text_frames=False, push_stop_frames=True, stop_frame_timeout_s=2.0, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index e36eaf497..94475aca7 100644 --- a/src/pipecat/services/fish.py +++ b/src/pipecat/services/fish.py @@ -60,7 +60,7 @@ class FishAudioTTSService(TTSService, WebsocketService): params: InputParams = InputParams(), **kwargs, ): - super().__init__(sample_rate=sample_rate, **kwargs) + super().__init__(pause_frame_processing=True, sample_rate=sample_rate, **kwargs) self._api_key = api_key self._base_url = "wss://api.fish.audio/v1/tts/live" diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py index 645494cc1..0272690e8 100644 --- a/src/pipecat/services/lmnt.py +++ b/src/pipecat/services/lmnt.py @@ -73,6 +73,7 @@ class LmntTTSService(TTSService, WebsocketService): TTSService.__init__( self, push_stop_frames=True, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 3196c306d..297e802d3 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -120,6 +120,7 @@ class PlayHTTTSService(TTSService, WebsocketService): ): TTSService.__init__( self, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 79e557f48..f87405a63 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -101,6 +101,7 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService): push_text_frames=False, push_stop_frames=True, stop_frame_timeout_s=2.0, + pause_frame_processing=True, sample_rate=sample_rate, **kwargs, ) From 633a4d4c58bcef03246be08ba92081bedd33c184 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 15:42:05 -0800 Subject: [PATCH 04/12] FalImageGenService: load image async to not block the event loop --- CHANGELOG.md | 3 +++ src/pipecat/services/fal.py | 14 +++++++++----- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 5542674ec..c8590e665 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -105,6 +105,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a `FalImageGenService` issue that was causing the event loop to be + blocked while loading the downloadded image. + - Fixed a `CartesiaTTSService` service issue that would cause audio overlapping in some cases. diff --git a/src/pipecat/services/fal.py b/src/pipecat/services/fal.py index d229861f1..7173861ab 100644 --- a/src/pipecat/services/fal.py +++ b/src/pipecat/services/fal.py @@ -4,6 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import asyncio import io import os from typing import AsyncGenerator, Dict, Optional, Union @@ -53,6 +54,11 @@ class FalImageGenService(ImageGenService): os.environ["FAL_KEY"] = key async def run_image_gen(self, prompt: str) -> AsyncGenerator[Frame, None]: + def load_image_bytes(encoded_image: bytes): + buffer = io.BytesIO(encoded_image) + image = Image.open(buffer) + return (image.tobytes(), image.size, image.format) + logger.debug(f"Generating image from prompt: {prompt}") response = await fal_client.run_async( @@ -73,10 +79,8 @@ class FalImageGenService(ImageGenService): logger.debug(f"Downloading image {image_url} ...") async with self._aiohttp_session.get(image_url) as response: logger.debug(f"Downloaded image {image_url}") - image_stream = io.BytesIO(await response.content.read()) - image = Image.open(image_stream) + encoded_image = await response.content.read() + (image_bytes, size, format) = await asyncio.to_thread(load_image_bytes, encoded_image) - frame = URLImageRawFrame( - url=image_url, image=image.tobytes(), size=image.size, format=image.format - ) + frame = URLImageRawFrame(url=image_url, image=image_bytes, size=size, format=format) yield frame From f6912c0f9a45efcb9e3d375985c11618aad36a18 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 15:44:13 -0800 Subject: [PATCH 05/12] utils: don't consider colon an end of sentence --- CHANGELOG.md | 2 ++ src/pipecat/utils/string.py | 4 ++-- tests/test_utils_string.py | 2 -- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c8590e665..0201fdc23 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,6 +56,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- We don't consider a colon `:` and end of sentence any more. + - Updated `DailyTransport` to respect the `audio_in_stream_on_start` field, ensuring it only starts receiving the audio input if it is enabled. diff --git a/src/pipecat/utils/string.py b/src/pipecat/utils/string.py index c269c5e57..916dfad41 100644 --- a/src/pipecat/utils/string.py +++ b/src/pipecat/utils/string.py @@ -13,8 +13,8 @@ ENDOFSENTENCE_PATTERN_STR = r""" (? Date: Fri, 14 Feb 2025 15:51:03 -0800 Subject: [PATCH 06/12] LLMAssistantResponseAggregator: initialize messages --- src/pipecat/processors/aggregators/llm_response.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 950c155e6..c12b0d90b 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -420,7 +420,7 @@ class LLMUserResponseAggregator(LLMUserContextAggregator): class LLMAssistantResponseAggregator(LLMAssistantContextAggregator): - def __init__(self, messages: List[dict], **kwargs): + def __init__(self, messages: List[dict] = [], **kwargs): super().__init__(context=OpenAILLMContext(messages), **kwargs) async def push_aggregation(self): From 63950912f01c2ffa79d53b6aa8c67aa52ac3043b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 15:55:22 -0800 Subject: [PATCH 07/12] LLMAssistantContextAggregator: add missing variable initialization --- src/pipecat/processors/aggregators/llm_response.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index c12b0d90b..ca73649b2 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -358,6 +358,8 @@ class LLMAssistantContextAggregator(LLMContextResponseAggregator): super().__init__(context=context, role="assistant", **kwargs) self._expect_stripped_words = expect_stripped_words + self._started = False + self.reset() async def process_frame(self, frame: Frame, direction: FrameDirection): From a107b1cb4b40c0d2ed6d5c0d35a8eae1ac6e139c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 15:57:18 -0800 Subject: [PATCH 08/12] examples(06a): use CartesiaTTSService --- examples/foundational/06a-image-sync.py | 4 ++-- src/pipecat/processors/aggregators/llm_response.py | 2 +- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py index 841c335f6..03a7fd3b8 100644 --- a/examples/foundational/06a-image-sync.py +++ b/examples/foundational/06a-image-sync.py @@ -27,7 +27,7 @@ from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.services.cartesia import CartesiaHttpTTSService +from pipecat.services.cartesia import CartesiaTTSService from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import DailyParams, DailyTransport @@ -91,7 +91,7 @@ async def main(): ), ) - tts = CartesiaHttpTTSService( + tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", # British Lady ) diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index ca73649b2..2de3ebae4 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -10,7 +10,6 @@ from abc import abstractmethod from typing import List from pipecat.frames.frames import ( - BotInterruptionFrame, CancelFrame, EmulateUserStartedSpeakingFrame, EmulateUserStoppedSpeakingFrame, @@ -281,6 +280,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): await self._cancel_aggregation_task() async def _handle_user_started_speaking(self, _: UserStartedSpeakingFrame): + self._last_user_speaking_time = time.time() self._user_speaking = True async def _handle_user_stopped_speaking(self, _: UserStoppedSpeakingFrame): From 1f5b790dd0f2bc42d24df90b3951a64946d071f7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 16:41:10 -0800 Subject: [PATCH 09/12] TTSService: reset processing text during interruptions --- src/pipecat/services/ai_services.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 99fd05481..0b67cd3ce 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -365,6 +365,7 @@ class TTSService(AIService): async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection): self._current_sentence = "" + self._processing_text = False if self._text_filter: self._text_filter.handle_interruption() await self.push_frame(frame, direction) From 883410d8ac4a3b89bc06f396c0fddd62f9b2fa01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 16:42:00 -0800 Subject: [PATCH 10/12] FrameProcessor: no need to create an input event every time --- src/pipecat/processors/frame_processor.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 3fb515a87..f27c6d6c2 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -73,10 +73,11 @@ class FrameProcessor: self._metrics.set_processor_name(self.name) # Processors have an input queue. The input queue will be processed - # immediately (default) or it will block if `pause_processing_frames()` is - # called. To resume processing frames we need to call - # `resume_processing_frames()`. + # immediately (default) or it will block if `pause_processing_frames()` + # is called. To resume processing frames we need to call + # `resume_processing_frames()` which will wake up the event. self.__should_block_frames = False + self.__input_event = asyncio.Event() self.__input_frame_task: Optional[asyncio.Task] = None # Every processor in Pipecat should only output frames from a single @@ -335,8 +336,8 @@ class FrameProcessor: def __create_input_task(self): if not self.__input_frame_task: self.__should_block_frames = False + self.__input_event.clear() self.__input_queue = asyncio.Queue() - self.__input_event = asyncio.Event() self.__input_frame_task = self.create_task(self.__input_frame_task_handler()) async def __cancel_input_task(self): From 329e89c1d938be4f297b5f84e6f10668b301f537 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 17:17:47 -0800 Subject: [PATCH 11/12] TTSService: push BotStoppedSpeakingFrame --- src/pipecat/services/ai_services.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 0b67cd3ce..c5adc89e0 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -313,6 +313,7 @@ class TTSService(AIService): await self._process_text_frame(frame) elif isinstance(frame, StartInterruptionFrame): await self._handle_interruption(frame, direction) + await self.push_frame(frame, direction) elif isinstance(frame, (LLMFullResponseEndFrame, EndFrame)): # We pause processing incoming frames if the LLM response included # text (it might be that it's only a function calling response). We @@ -339,6 +340,7 @@ class TTSService(AIService): await self._update_settings(frame.settings) elif isinstance(frame, BotStoppedSpeakingFrame): await self._maybe_resume_frame_processing() + await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) @@ -368,7 +370,6 @@ class TTSService(AIService): self._processing_text = False if self._text_filter: self._text_filter.handle_interruption() - await self.push_frame(frame, direction) async def _maybe_pause_frame_processing(self): if self._processing_text and self._pause_frame_processing: From 01c06c5cac5e4e53016a42e363af531f00d6afd1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 14 Feb 2025 14:12:02 -0800 Subject: [PATCH 12/12] update CHANGELOG for 0.0.57 --- CHANGELOG.md | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0201fdc23..a8731571f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,7 +5,7 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). -## [Unreleased] +## [0.0.57] - 2025-02-14 ### Added @@ -139,9 +139,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue[#1192] in 11labs where we are trying to reconnect/disconnect the websocket connection even when the connection is already closed. -- Fixed an issue where `has_regular_messages` condition was always been true in - `GoogleLLMContext` due to `Part` having `function_call` & `function_response` with - `None` values. +- Fixed an issue where `has_regular_messages` condition was always true in + `GoogleLLMContext` due to `Part` having `function_call` & `function_response` + with `None` values. + +### Other + +- Added new `instant-voice` example. This example showcases how to enable + instant voice communication as soon as a user connects. + +- Added new `local-input-select-stt` example. This examples allows you to play + with local audio inputs by slecting them through a nice text interface. ## [0.0.56] - 2025-02-06