From a674b43243f8d70a57c2e2103623e958992c7394 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 3 Jun 2024 20:54:55 -0700 Subject: [PATCH 01/12] transport: remove redundant camera thread and switch audio pull for push --- src/pipecat/transports/base_input.py | 41 ++++++++-------- src/pipecat/transports/base_output.py | 21 +++++---- src/pipecat/transports/local/audio.py | 26 ++++++---- src/pipecat/transports/local/tk.py | 27 ++++++----- src/pipecat/transports/services/daily.py | 60 ++++++++++++------------ 5 files changed, 92 insertions(+), 83 deletions(-) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index ba4f8ae4e..6fb1e2ee8 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -36,11 +36,7 @@ class BaseInputTransport(FrameProcessor): self._running = False self._allow_interruptions = False - self._in_executor = ThreadPoolExecutor(max_workers=5) - - # Create audio input queue if needed. - if self._params.audio_in_enabled or self._params.vad_enabled: - self._audio_in_queue = queue.Queue() + self._executor = ThreadPoolExecutor(max_workers=5) # Create push frame task. This is the task that will push frames in # order. We also guarantee that all frames are pushed in the same task. @@ -57,9 +53,11 @@ class BaseInputTransport(FrameProcessor): self._running = True + # Create audio input queue and thread if needed. if self._params.audio_in_enabled or self._params.vad_enabled: - loop = self.get_event_loop() - self._audio_thread = loop.run_in_executor(self._in_executor, self._audio_thread_handler) + self._audio_in_queue = queue.Queue() + self._audio_thread = self._loop.run_in_executor( + self._executor, self._audio_thread_handler) async def stop(self): if not self._running: @@ -77,8 +75,8 @@ class BaseInputTransport(FrameProcessor): def vad_analyzer(self) -> VADAnalyzer | None: return self._params.vad_analyzer - def read_next_audio_frame(self) -> AudioRawFrame | None: - pass + def push_audio_frame(self, frame: AudioRawFrame): + self._audio_in_queue.put_nowait(frame) # # Frame processor @@ -174,21 +172,22 @@ class BaseInputTransport(FrameProcessor): vad_state: VADState = VADState.QUIET while self._running: try: - frame = self.read_next_audio_frame() + frame: AudioRawFrame = self._audio_in_queue.get(timeout=1) - if frame: - audio_passthrough = True + audio_passthrough = True - # Check VAD and push event if necessary. We just care about - # changes from QUIET to SPEAKING and vice versa. - if self._params.vad_enabled: - vad_state = self._handle_vad(frame.audio, vad_state) - audio_passthrough = self._params.vad_audio_passthrough + # Check VAD and push event if necessary. We just care about + # changes from QUIET to SPEAKING and vice versa. + if self._params.vad_enabled: + vad_state = self._handle_vad(frame.audio, vad_state) + audio_passthrough = self._params.vad_audio_passthrough # Push audio downstream if passthrough. - if audio_passthrough: - future = asyncio.run_coroutine_threadsafe( - self._internal_push_frame(frame), self.get_event_loop()) - future.result() + if audio_passthrough: + future = asyncio.run_coroutine_threadsafe( + self._internal_push_frame(frame), self._loop) + future.result() + except queue.Empty: + pass except BaseException as e: logger.error(f"Error reading audio frames: {e}") diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 146c58496..9e99ce736 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -43,7 +43,7 @@ class BaseOutputTransport(FrameProcessor): self._running = False self._allow_interruptions = False - self._out_executor = ThreadPoolExecutor(max_workers=5) + self._executor = ThreadPoolExecutor(max_workers=5) # These are the images that we should send to the camera at our desired # framerate. @@ -57,6 +57,10 @@ class BaseOutputTransport(FrameProcessor): self._stopped_event = asyncio.Event() self._is_interrupted = threading.Event() + # Create push frame task. This is the task that will push frames in + # order. We also guarantee that all frames are pushed in the same task. + self._create_push_task() + async def start(self, frame: StartFrame): # Make sure we have the latest params. Note that this transport might # have been started on another task that might not need interruptions, @@ -70,15 +74,12 @@ class BaseOutputTransport(FrameProcessor): loop = self.get_event_loop() + # Create queues and threads. if self._params.camera_out_enabled: self._camera_out_thread = loop.run_in_executor( - self._out_executor, self._camera_out_thread_handler) + self._executor, self._camera_out_thread_handler) - self._sink_thread = loop.run_in_executor(self._out_executor, self._sink_thread_handler) - - # Create push frame task. This is the task that will push frames in - # order. We also guarantee that all frames are pushed in the same task. - self._create_push_task() + self._sink_thread = loop.run_in_executor(self._executor, self._sink_thread_handler) async def stop(self): if not self._running: @@ -117,7 +118,7 @@ class BaseOutputTransport(FrameProcessor): # if isinstance(frame, StartFrame): await self.start(frame) - self._sink_queue.put(frame) + self._sink_queue.put_nowait(frame) # EndFrame is managed in the queue handler. elif isinstance(frame, CancelFrame): await self.stop() @@ -126,7 +127,7 @@ class BaseOutputTransport(FrameProcessor): await self._handle_interruptions(frame) await self.push_frame(frame, direction) else: - self._sink_queue.put(frame) + self._sink_queue.put_nowait(frame) # If we are finishing, wait here until we have stopped, otherwise we might # close things too early upstream. We need this event because we don't @@ -233,7 +234,7 @@ class BaseOutputTransport(FrameProcessor): def _set_camera_image(self, image: ImageRawFrame): if self._params.camera_out_is_live: - self._camera_out_queue.put(image) + self._camera_out_queue.put_nowait(image) else: self._camera_images = itertools.cycle([image]) diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index 14b8bd5d3..6ac461b3b 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -28,22 +28,17 @@ class LocalAudioInputTransport(BaseInputTransport): def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) + sample_rate = self._params.audio_in_sample_rate + num_frames = int(sample_rate / 100) # 10ms of audio + self._in_stream = py_audio.open( format=py_audio.get_format_from_width(2), channels=params.audio_in_channels, rate=params.audio_in_sample_rate, - frames_per_buffer=params.audio_in_sample_rate, + frames_per_buffer=num_frames, + stream_callback=self._audio_in_callback, input=True) - def read_next_audio_frame(self) -> AudioRawFrame | None: - sample_rate = self._params.audio_in_sample_rate - num_channels = self._params.audio_in_channels - num_frames = int(sample_rate / 100) # 10ms of audio - - audio = self._in_stream.read(num_frames, exception_on_overflow=False) - - return AudioRawFrame(audio=audio, sample_rate=sample_rate, num_channels=num_channels) - async def start(self, frame: StartFrame): await super().start(frame) self._in_stream.start_stream() @@ -60,6 +55,17 @@ class LocalAudioInputTransport(BaseInputTransport): await super().cleanup() + def _audio_in_callback(self, in_data, frame_count, time_info, status): + if not self._running: + return (None, pyaudio.paAbort) + + frame = AudioRawFrame(audio=in_data, + sample_rate=self._params.audio_in_sample_rate, + num_channels=self._params.audio_in_channels) + self.push_audio_frame(frame) + + return (None, pyaudio.paContinue) + class LocalAudioOutputTransport(BaseOutputTransport): diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 808837998..b8de9ce71 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -38,22 +38,17 @@ class TkInputTransport(BaseInputTransport): def __init__(self, py_audio: pyaudio.PyAudio, params: TransportParams): super().__init__(params) + sample_rate = self._params.audio_in_sample_rate + num_frames = int(sample_rate / 100) # 10ms of audio + self._in_stream = py_audio.open( format=py_audio.get_format_from_width(2), channels=params.audio_in_channels, rate=params.audio_in_sample_rate, - frames_per_buffer=params.audio_in_sample_rate, + frames_per_buffer=num_frames, + stream_callback=self._audio_in_callback, input=True) - def read_next_audio_frame(self) -> AudioRawFrame | None: - sample_rate = self._params.audio_in_sample_rate - num_channels = self._params.audio_in_channels - num_frames = int(sample_rate / 100) # 10ms of audio - - audio = self._in_stream.read(num_frames, exception_on_overflow=False) - - return AudioRawFrame(audio=audio, sample_rate=sample_rate, num_channels=num_channels) - async def start(self, frame: StartFrame): await super().start(frame) self._in_stream.start_stream() @@ -63,12 +58,22 @@ class TkInputTransport(BaseInputTransport): self._in_stream.stop_stream() async def cleanup(self): + await super().cleanup() # This is not very pretty (taken from PyAudio docs). while self._in_stream.is_active(): await asyncio.sleep(0.1) self._in_stream.close() - await super().cleanup() + def _audio_in_callback(self, in_data, frame_count, time_info, status): + if not self._running: + return (None, pyaudio.paAbort) + + frame = AudioRawFrame(audio=in_data, + sample_rate=self._params.audio_in_sample_rate, + num_channels=self._params.audio_in_channels) + self.push_audio_frame(frame) + + return (None, pyaudio.paContinue) class TkOutputTransport(BaseOutputTransport): diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index f20eaa4dc..5ce4a6c84 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -37,7 +37,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.vad.vad_analyzer import VADAnalyzer, VADParams, VADState +from pipecat.vad.vad_analyzer import VADAnalyzer, VADParams from loguru import logger @@ -193,7 +193,7 @@ class DailyTransportClient(EventHandler): num_channels = self._params.audio_in_channels if self._other_participant_has_joined: - num_frames = int(sample_rate / 100) # 10ms of audio + num_frames = int(sample_rate / 100) * 2 # 20ms of audio audio = self._speaker.read_frames(num_frames) @@ -472,7 +472,6 @@ class DailyInputTransport(BaseInputTransport): self._client = client self._video_renderers = {} - self._camera_in_queue = queue.Queue() self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer if params.vad_enabled and not params.vad_analyzer: @@ -483,23 +482,26 @@ class DailyInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): if self._running: return + # Parent start. + await super().start(frame) # Join the room. await self._client.join() - # This will set _running=True - await super().start(frame) - # Create camera in thread (runs if _running is true). - self._camera_in_thread = self._loop.run_in_executor( - self._in_executor, self._camera_in_thread_handler) + # Create audio task. It reads audio frames from Daily and push them + # internally for VAD processing. + if self._params.audio_in_enabled or self._params.vad_enabled: + self._audio_in_thread = self._loop.run_in_executor( + self._executor, self._audio_in_thread_handler) async def stop(self): if not self._running: return + # Parent stop. This will set _running to False. + await super().stop() # Leave the room. await self._client.leave() - # This will set _running=False - await super().stop() - # The thread will stop. - await self._camera_in_thread + # Stop audio thread. + if self._params.audio_in_enabled or self._params.vad_enabled: + await self._audio_in_thread async def cleanup(self): await super().cleanup() @@ -508,9 +510,6 @@ class DailyInputTransport(BaseInputTransport): def vad_analyzer(self) -> VADAnalyzer | None: return self._vad_analyzer - def read_next_audio_frame(self) -> AudioRawFrame | None: - return self._client.read_next_audio_frame() - # # FrameProcessor # @@ -536,6 +535,16 @@ class DailyInputTransport(BaseInputTransport): self._internal_push_frame(frame), self.get_event_loop()) future.result() + # + # Audio in + # + + def _audio_in_thread_handler(self): + while self._running: + frame = self._client.read_next_audio_frame() + if frame: + self.push_audio_frame(frame) + # # Camera in # @@ -584,23 +593,12 @@ class DailyInputTransport(BaseInputTransport): image=buffer, size=size, format=format) - self._camera_in_queue.put(frame) + future = asyncio.run_coroutine_threadsafe( + self._internal_push_frame(frame), self.get_event_loop()) + future.result() self._video_renderers[participant_id]["timestamp"] = curr_time - def _camera_in_thread_handler(self): - while self._running: - try: - frame = self._camera_in_queue.get(timeout=1) - future = asyncio.run_coroutine_threadsafe( - self._internal_push_frame(frame), self.get_event_loop()) - future.result() - self._camera_in_queue.task_done() - except queue.Empty: - pass - except BaseException as e: - logger.error(f"Error capturing video: {e}") - class DailyOutputTransport(BaseOutputTransport): @@ -612,7 +610,7 @@ class DailyOutputTransport(BaseOutputTransport): async def start(self, frame: StartFrame): if self._running: return - # This will set _running=True + # Parent start. await super().start(frame) # Join the room. await self._client.join() @@ -620,7 +618,7 @@ class DailyOutputTransport(BaseOutputTransport): async def stop(self): if not self._running: return - # This will set _running=False + # Parent stop. This will set _running to False. await super().stop() # Leave the room. await self._client.leave() From 5cdb8a79a1e930840330ee2e07f8b5b5a897bce4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 00:52:13 -0700 Subject: [PATCH 02/12] examples: use camera_out_is_live for live video --- examples/foundational/09-mirror.py | 1 + examples/foundational/09a-local-mirror.py | 1 + examples/foundational/13a-whisper-local.py | 7 ++----- 3 files changed, 4 insertions(+), 5 deletions(-) diff --git a/examples/foundational/09-mirror.py b/examples/foundational/09-mirror.py index 4fecd2be6..cf36a6220 100644 --- a/examples/foundational/09-mirror.py +++ b/examples/foundational/09-mirror.py @@ -30,6 +30,7 @@ async def main(room_url, token): audio_in_enabled=True, audio_out_enabled=True, camera_out_enabled=True, + camera_out_is_live=True, camera_out_width=1280, camera_out_height=720 ) diff --git a/examples/foundational/09a-local-mirror.py b/examples/foundational/09a-local-mirror.py index d8da14343..78a2fd97d 100644 --- a/examples/foundational/09a-local-mirror.py +++ b/examples/foundational/09a-local-mirror.py @@ -38,6 +38,7 @@ async def main(room_url, token): TransportParams( audio_out_enabled=True, camera_out_enabled=True, + camera_out_is_live=True, camera_out_width=1280, camera_out_height=720)) diff --git a/examples/foundational/13a-whisper-local.py b/examples/foundational/13a-whisper-local.py index 1c3e3dd22..366cd2cf5 100644 --- a/examples/foundational/13a-whisper-local.py +++ b/examples/foundational/13a-whisper-local.py @@ -16,8 +16,6 @@ from pipecat.services.whisper import WhisperSTTService from pipecat.transports.base_transport import TransportParams from pipecat.transports.local.audio import LocalAudioTransport -from runner import configure - from loguru import logger from dotenv import load_dotenv @@ -34,7 +32,7 @@ class TranscriptionLogger(FrameProcessor): print(f"Transcription: {frame.text}") -async def main(room_url: str): +async def main(): transport = LocalAudioTransport(TransportParams(audio_in_enabled=True)) stt = WhisperSTTService() @@ -51,5 +49,4 @@ async def main(room_url: str): if __name__ == "__main__": - (url, token) = configure() - asyncio.run(main(url)) + asyncio.run(main()) From 4057fbbcfde831c3836d9d9c06fe6990f98934a7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 12:11:07 -0700 Subject: [PATCH 03/12] transports(tk): fix pyaudio output stream cleanup --- src/pipecat/transports/local/audio.py | 16 ++-------------- src/pipecat/transports/local/tk.py | 16 ++-------------- 2 files changed, 4 insertions(+), 28 deletions(-) diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index 6ac461b3b..33d6062aa 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -29,7 +29,7 @@ class LocalAudioInputTransport(BaseInputTransport): super().__init__(params) sample_rate = self._params.audio_in_sample_rate - num_frames = int(sample_rate / 100) # 10ms of audio + num_frames = int(sample_rate / 100) * 2 # 20ms of audio self._in_stream = py_audio.open( format=py_audio.get_format_from_width(2), @@ -81,21 +81,9 @@ class LocalAudioOutputTransport(BaseOutputTransport): def write_raw_audio_frames(self, frames: bytes): self._out_stream.write(frames) - async def start(self, frame: StartFrame): - await super().start(frame) - self._out_stream.start_stream() - - async def stop(self): - await super().stop() - self._out_stream.stop_stream() - async def cleanup(self): - # This is not very pretty (taken from PyAudio docs). - while self._out_stream.is_active(): - await asyncio.sleep(0.1) - self._out_stream.close() - await super().cleanup() + self._out_stream.close() class LocalAudioTransport(BaseTransport): diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index b8de9ce71..6370e8d39 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -39,7 +39,7 @@ class TkInputTransport(BaseInputTransport): super().__init__(params) sample_rate = self._params.audio_in_sample_rate - num_frames = int(sample_rate / 100) # 10ms of audio + num_frames = int(sample_rate / 100) * 2 # 20ms of audio self._in_stream = py_audio.open( format=py_audio.get_format_from_width(2), @@ -100,21 +100,9 @@ class TkOutputTransport(BaseOutputTransport): def write_frame_to_camera(self, frame: ImageRawFrame): self.get_event_loop().call_soon(self._write_frame_to_tk, frame) - async def start(self, frame: StartFrame): - await super().start(frame) - self._out_stream.start_stream() - - async def stop(self): - await super().stop() - self._out_stream.stop_stream() - async def cleanup(self): - # This is not very pretty (taken from PyAudio docs). - while self._out_stream.is_active(): - await asyncio.sleep(0.1) - self._out_stream.close() - await super().cleanup() + self._out_stream.close() def _write_frame_to_tk(self, frame: ImageRawFrame): width = frame.size[0] From af202d4fe57fff9e362caf647b7796ae3eab2331 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 12:12:19 -0700 Subject: [PATCH 04/12] pipeline(task): introduce has_finished() --- examples/foundational/05a-local-sync-speech-and-image.py | 2 +- examples/foundational/09a-local-mirror.py | 6 +++--- src/pipecat/pipeline/runner.py | 6 ------ src/pipecat/pipeline/task.py | 5 +++++ 4 files changed, 9 insertions(+), 10 deletions(-) diff --git a/examples/foundational/05a-local-sync-speech-and-image.py b/examples/foundational/05a-local-sync-speech-and-image.py index d476754fb..a377ebe98 100644 --- a/examples/foundational/05a-local-sync-speech-and-image.py +++ b/examples/foundational/05a-local-sync-speech-and-image.py @@ -156,7 +156,7 @@ async def main(): await runner.stop_when_done() async def run_tk(): - while True: + while not task.has_finished(): tk_root.update() tk_root.update_idletasks() await asyncio.sleep(0.1) diff --git a/examples/foundational/09a-local-mirror.py b/examples/foundational/09a-local-mirror.py index 78a2fd97d..92e244d3f 100644 --- a/examples/foundational/09a-local-mirror.py +++ b/examples/foundational/09a-local-mirror.py @@ -48,15 +48,15 @@ async def main(room_url, token): pipeline = Pipeline([daily_transport.input(), tk_transport.output()]) - runner = PipelineRunner() + task = PipelineTask(pipeline) async def run_tk(): - while runner.is_active(): + while not task.has_finished(): tk_root.update() tk_root.update_idletasks() await asyncio.sleep(0.1) - task = PipelineTask(pipeline) + runner = PipelineRunner() await asyncio.gather(runner.run(task), run_tk()) diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index 2d1b8100a..47754d82f 100644 --- a/src/pipecat/pipeline/runner.py +++ b/src/pipecat/pipeline/runner.py @@ -20,18 +20,15 @@ class PipelineRunner: self.name: str = name or f"{self.__class__.__name__}#{obj_count(self)}" self._tasks = {} - self._running = True if handle_sigint: self._setup_sigint() async def run(self, task: PipelineTask): logger.debug(f"Runner {self} started running {task}") - self._running = True self._tasks[task.name] = task await task.run() del self._tasks[task.name] - self._running = False logger.debug(f"Runner {self} finished running {task}") async def stop_when_done(self): @@ -42,9 +39,6 @@ class PipelineRunner: logger.debug(f"Canceling runner {self}") await asyncio.gather(*[t.cancel() for t in self._tasks.values()]) - def is_active(self): - return self._running - def _setup_sigint(self): loop = asyncio.get_running_loop() loop.add_signal_handler( diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index f81281820..e3e7f7e36 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -43,6 +43,7 @@ class PipelineTask: self._pipeline = pipeline self._params = params + self._finished = False self._down_queue = asyncio.Queue() self._up_queue = asyncio.Queue() @@ -50,6 +51,9 @@ class PipelineTask: self._source = Source(self._up_queue) self._source.link(pipeline) + def has_finished(self): + return self._finished + async def stop_when_done(self): logger.debug(f"Task {self} scheduled to stop when done") await self.queue_frame(EndFrame()) @@ -67,6 +71,7 @@ class PipelineTask: self._process_up_task = asyncio.create_task(self._process_up_queue()) self._process_down_task = asyncio.create_task(self._process_down_queue()) await asyncio.gather(self._process_up_task, self._process_down_task) + self._finished = True async def queue_frame(self, frame: Frame): await self._down_queue.put(frame) From 571e10f83e4f7bf367e597dcddc9fc8d48238d1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 12:13:29 -0700 Subject: [PATCH 05/12] services(anthropic): fix interruptions with anthropic --- src/pipecat/services/anthropic.py | 39 ++++++++++++------------------- src/pipecat/services/cartesia.py | 4 ++-- src/pipecat/services/deepgram.py | 5 ++-- 3 files changed, 20 insertions(+), 28 deletions(-) diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index f1368f6fa..3e774a6bd 100644 --- a/src/pipecat/services/anthropic.py +++ b/src/pipecat/services/anthropic.py @@ -4,8 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import os -import asyncio import time import base64 @@ -80,8 +78,20 @@ class AnthropicLLMService(LLMService): }] }) else: - # text frame - anthropic_messages.append({"role": role, "content": content}) + # Text frame. Anthropic needs the roles to alternate. This will + # cause an issue with interruptions. So, if we detect we are the + # ones asking again it probably means we were interrupted. + if role == "user" and len(anthropic_messages) > 1: + last_message = anthropic_messages[-1] + if last_message["role"] == "user": + anthropic_messages = anthropic_messages[:-1] + content = last_message["content"] + anthropic_messages.append( + {"role": "user", "content": f"Sorry, I just asked you about [{content}] but now I would like to know [{text}]."}) + else: + anthropic_messages.append({"role": role, "content": text}) + else: + anthropic_messages.append({"role": role, "content": text}) return anthropic_messages @@ -107,7 +117,7 @@ class AnthropicLLMService(LLMService): await self.push_frame(LLMResponseEndFrame()) except Exception as e: - logger.error(f"Exception: {e}") + logger.error(f"Anthrophic exception: {e}") finally: await self.push_frame(LLMFullResponseEndFrame()) @@ -125,22 +135,3 @@ class AnthropicLLMService(LLMService): if context: await self._process_context(context) - - async def x_process_frame(self, frame: Frame, direction: FrameDirection): - if isinstance(frame, LLMMessagesFrame): - stream = await self.client.messages.create( - max_tokens=self.max_tokens, - messages=[ - { - "role": "user", - "content": "Hello, Claude", - } - ], - model=self.model, - stream=True, - ) - async for event in stream: - if event.type == "content_block_delta": - await self.push_frame(TextFrame(event.delta.text)) - else: - await self.push_frame(frame, direction) diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 581b0d5c7..86baa7a3e 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -36,7 +36,7 @@ class CartesiaTTSService(TTSService): logger.error(f"Cartesia initialization error: {e}") async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Transcribing text: [{text}]") + logger.debug(f"Generating TTS: [{text}]") try: chunk_generator = await self._client.generate( @@ -50,4 +50,4 @@ class CartesiaTTSService(TTSService): async for chunk in chunk_generator: yield AudioRawFrame(chunk['audio'], 16000, 1) except Exception as e: - logger.error(f"Cartesia error: {e}") + logger.error(f"Cartesia exception: {e}") diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index 5a80f19f0..b5901825e 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -30,7 +30,8 @@ class DeepgramTTSService(TTSService): self._aiohttp_session = aiohttp_session async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.info(f"Running Deepgram TTS for {text}") + logger.debug(f"Generating TTS: [{text}]") + base_url = "https://api.deepgram.com/v1/speak" request_url = f"{base_url}?model={self._voice}&encoding=linear16&container=none&sample_rate=16000" headers = {"authorization": f"token {self._api_key}"} @@ -48,4 +49,4 @@ class DeepgramTTSService(TTSService): frame = AudioRawFrame(audio=data, sample_rate=16000, num_channels=1) yield frame except Exception as e: - logger.error(f"Exception {e}") + logger.error(f"Deepgram exception: {e}") From 7eb9dfde384e656b1ef16de8d73c8446685c8610 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 12:14:25 -0700 Subject: [PATCH 06/12] pyproject: include langchain-community and langchain-openai --- .../07b-interruptible-langchain.py | 27 ++-- linux-py3.10-requirements.txt | 133 +++++++++++++++--- pyproject.toml | 4 +- 3 files changed, 129 insertions(+), 35 deletions(-) diff --git a/examples/foundational/07b-interruptible-langchain.py b/examples/foundational/07b-interruptible-langchain.py index bda1d92fc..767877dff 100644 --- a/examples/foundational/07b-interruptible-langchain.py +++ b/examples/foundational/07b-interruptible-langchain.py @@ -4,15 +4,11 @@ # 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 runner import configure from pipecat.frames.frames import LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline @@ -25,20 +21,19 @@ from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.transports.services.daily import DailyParams, DailyTransport from pipecat.vad.silero import SileroVADAnalyzer +from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder +from langchain_community.chat_message_histories import ChatMessageHistory +from langchain_core.chat_history import BaseChatMessageHistory +from langchain_core.runnables.history import RunnableWithMessageHistory +from langchain_openai import ChatOpenAI + +from loguru import logger + +from runner import configure + +from dotenv import load_dotenv load_dotenv(override=True) -try: - from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder - from langchain_community.chat_message_histories import ChatMessageHistory - from langchain_core.chat_history import BaseChatMessageHistory - from langchain_core.runnables.history import RunnableWithMessageHistory - from langchain_openai import ChatOpenAI - -except ModuleNotFoundError as e: - logger.exception( - "In order to run this example you need to `pip install pipecat-ai[langchain] langchain-community langchain-openai. Also, be sure to set `OPENAI_API_KEY` in the environment variable." - ) - raise Exception(f"Missing module: {e}") logger.remove(0) logger.add(sys.stderr, level="DEBUG") diff --git a/linux-py3.10-requirements.txt b/linux-py3.10-requirements.txt index d3687c17e..6a6bf2d92 100644 --- a/linux-py3.10-requirements.txt +++ b/linux-py3.10-requirements.txt @@ -5,7 +5,11 @@ # pip-compile --all-extras pyproject.toml # aiohttp==3.9.5 - # via pipecat-ai (pyproject.toml) + # via + # cartesia + # langchain + # langchain-community + # pipecat-ai (pyproject.toml) aiosignal==1.3.1 # via aiohttp annotated-types==0.7.0 @@ -18,10 +22,12 @@ anyio==4.4.0 # httpx # openai async-timeout==4.0.3 - # via aiohttp + # via + # aiohttp + # langchain attrs==23.2.0 # via aiohttp -av==12.0.0 +av==12.1.0 # via faster-whisper azure-cognitiveservices-speech==1.37.0 # via pipecat-ai (pyproject.toml) @@ -29,11 +35,15 @@ blinker==1.8.2 # via flask cachetools==5.3.3 # via google-auth -certifi==2024.2.2 +cartesia==0.1.1 + # via pipecat-ai (pyproject.toml) +certifi==2024.6.2 # via # httpcore # httpx # requests +cffi==1.16.0 + # via sounddevice charset-normalizer==3.3.2 # via requests click==8.1.7 @@ -44,6 +54,8 @@ ctranslate2==4.2.1 # via faster-whisper daily-python==0.9.1 # via pipecat-ai (pyproject.toml) +dataclasses-json==0.6.6 + # via langchain-community distro==1.9.0 # via # anthropic @@ -51,7 +63,9 @@ distro==1.9.0 einops==0.8.0 # via pipecat-ai (pyproject.toml) exceptiongroup==1.2.1 - # via anyio + # via + # anyio + # pytest fal-client==0.4.0 # via pipecat-ai (pyproject.toml) faster-whisper==1.0.2 @@ -75,7 +89,7 @@ frozenlist==1.4.1 # via # aiohttp # aiosignal -fsspec==2024.5.0 +fsspec==2024.6.0 # via # huggingface-hub # torch @@ -88,7 +102,7 @@ google-api-core[grpc]==2.19.0 # google-ai-generativelanguage # google-api-python-client # google-generativeai -google-api-python-client==2.131.0 +google-api-python-client==2.132.0 # via google-generativeai google-auth==2.29.0 # via @@ -101,11 +115,13 @@ google-auth-httplib2==0.2.0 # via google-api-python-client google-generativeai==0.5.4 # via pipecat-ai (pyproject.toml) -googleapis-common-protos==1.63.0 +googleapis-common-protos==1.63.1 # via # google-api-core # grpcio-status -grpcio==1.64.0 +greenlet==3.0.3 + # via sqlalchemy +grpcio==1.64.1 # via # google-api-core # grpcio-status @@ -123,6 +139,7 @@ httplib2==0.22.0 httpx==0.27.0 # via # anthropic + # cartesia # fal-client # openai httpx-sse==0.4.0 @@ -141,29 +158,62 @@ idna==3.7 # httpx # requests # yarl +iniconfig==2.0.0 + # via pytest itsdangerous==2.2.0 # via flask jinja2==3.1.4 # via # flask # torch +jsonpatch==1.33 + # via langchain-core +jsonpointer==2.4 + # via jsonpatch +langchain==0.2.1 + # via + # langchain-community + # pipecat-ai (pyproject.toml) +langchain-community==0.2.1 + # via pipecat-ai (pyproject.toml) +langchain-core==0.2.3 + # via + # langchain + # langchain-community + # langchain-openai + # langchain-text-splitters +langchain-openai==0.1.8 + # via pipecat-ai (pyproject.toml) +langchain-text-splitters==0.2.0 + # via langchain +langsmith==0.1.69 + # via + # langchain + # langchain-community + # langchain-core loguru==0.7.2 # via pipecat-ai (pyproject.toml) markupsafe==2.1.5 # via # jinja2 # werkzeug +marshmallow==3.21.2 + # via dataclasses-json mpmath==1.3.0 # via sympy multidict==6.0.5 # via # aiohttp # yarl +mypy-extensions==1.0.0 + # via typing-inspect networkx==3.3 # via torch numpy==1.26.4 # via # ctranslate2 + # langchain + # langchain-community # onnxruntime # pipecat-ai (pyproject.toml) # pyloudnorm @@ -204,16 +254,25 @@ nvidia-nvtx-cu12==12.1.105 onnxruntime==1.18.0 # via faster-whisper openai==1.26.0 - # via pipecat-ai (pyproject.toml) -packaging==24.0 + # via + # langchain-openai + # pipecat-ai (pyproject.toml) +orjson==3.10.3 + # via langsmith +packaging==23.2 # via # huggingface-hub + # langchain-core + # marshmallow # onnxruntime + # pytest # transformers pillow==10.3.0 # via # pipecat-ai (pyproject.toml) # torchvision +pluggy==1.5.0 + # via pytest proto-plus==1.23.0 # via # google-ai-generativelanguage @@ -237,12 +296,17 @@ pyasn1-modules==0.4.0 # via google-auth pyaudio==0.2.14 # via pipecat-ai (pyproject.toml) -pydantic==2.7.2 +pycparser==2.22 + # via cffi +pydantic==2.7.3 # via # anthropic # google-generativeai + # langchain + # langchain-core + # langsmith # openai -pydantic-core==2.18.3 +pydantic-core==2.18.4 # via pydantic pyht==0.0.28 # via pipecat-ai (pyproject.toml) @@ -250,21 +314,35 @@ pyloudnorm==0.1.1 # via pipecat-ai (pyproject.toml) pyparsing==3.1.2 # via httplib2 +pytest==8.2.2 + # via pytest-asyncio +pytest-asyncio==0.23.7 + # via cartesia python-dotenv==1.0.1 # via pipecat-ai (pyproject.toml) pyyaml==6.0.1 # via # ctranslate2 # huggingface-hub + # langchain + # langchain-community + # langchain-core # timm # transformers regex==2024.5.15 - # via transformers + # via + # tiktoken + # transformers requests==2.32.3 # via + # cartesia # google-api-core # huggingface-hub + # langchain + # langchain-community + # langsmith # pyht + # tiktoken # transformers rsa==4.9 # via google-auth @@ -280,10 +358,23 @@ sniffio==1.3.1 # anyio # httpx # openai -sympy==1.12 +sounddevice==0.4.7 + # via pipecat-ai (pyproject.toml) +sqlalchemy==2.0.30 + # via + # langchain + # langchain-community +sympy==1.12.1 # via # onnxruntime # torch +tenacity==8.3.0 + # via + # langchain + # langchain-community + # langchain-core +tiktoken==0.7.0 + # via langchain-openai timm==0.9.16 # via pipecat-ai (pyproject.toml) tokenizers==0.19.1 @@ -291,6 +382,8 @@ tokenizers==0.19.1 # anthropic # faster-whisper # transformers +tomli==2.0.1 + # via pytest torch==2.3.0 # via # pipecat-ai (pyproject.toml) @@ -311,7 +404,7 @@ transformers==4.40.2 # via pipecat-ai (pyproject.toml) triton==2.3.0 # via torch -typing-extensions==4.11.0 +typing-extensions==4.12.1 # via # anthropic # anyio @@ -321,13 +414,19 @@ typing-extensions==4.11.0 # pipecat-ai (pyproject.toml) # pydantic # pydantic-core + # sqlalchemy # torch + # typing-inspect +typing-inspect==0.9.0 + # via dataclasses-json uritemplate==4.1.1 # via google-api-python-client urllib3==2.2.1 # via requests websockets==12.0 - # via pipecat-ai (pyproject.toml) + # via + # cartesia + # pipecat-ai (pyproject.toml) werkzeug==3.0.3 # via flask yarl==1.9.4 diff --git a/pyproject.toml b/pyproject.toml index aa3558f87..31121d537 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,7 +26,7 @@ dependencies = [ "Pillow~=10.3.0", "protobuf~=4.25.3", "pyloudnorm~=0.1.1", - "typing-extensions~=4.11.0", + "typing-extensions~=4.12.1", ] [project.urls] @@ -42,7 +42,7 @@ examples = [ "python-dotenv~=1.0.0", "flask~=3.0.3", "flask_cors~=4.0.1" ] fal = [ "fal-client~=0.4.0" ] google = [ "google-generativeai~=0.5.3" ] fireworks = [ "openai~=1.26.0" ] -langchain = [ "langchain~=0.2.1" ] +langchain = [ "langchain~=0.2.1", "langchain-community~=0.2.1", "langchain-openai~=0.1.8" ] local = [ "pyaudio~=0.2.0" ] moondream = [ "einops~=0.8.0", "timm~=0.9.16", "transformers~=4.40.2" ] openai = [ "openai~=1.26.0" ] From d56a4cce1b87b72894e798bbaf31b9d74423a832 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 13:06:05 -0700 Subject: [PATCH 07/12] update CHANGELOG with latest changes --- CHANGELOG.md | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index f3c5c6df9..9eb6617ed 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,12 +7,32 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- `PipelineTask` now has a `has_finished()` method to indicate if the task has + completed. If a task is never ran `has_finished()` will return False. + ### Fixed +- Fixed an error closing local audio transports. + - Fixed an issue with Deepgram TTS that was introduced in the previous release. +- Fixed `AnthropicLLMService` interruptions. If an interruption occurred, a + `user` message could be appended after the previous `user` message. Anthropic + does not allow that because it requires alternate `user` and `assistant` + messages. + ### Performance +- The `BaseInputTransport` does not pull audio frames from sub-classes any + more. Instead, sub-classes now push audio frames into a queue in the base + class. Also, `DailyInputTransport` now pushes audio frames every 20ms instead + of 10ms. + +- Remove redundant camera input thread from `DailyInputTransport`. This should + improve performance a little bit when processing participant videos. + - Load Cartesia voice on startup. ## [0.0.25] - 2024-05-31 From 489060881dda329ae6ef4da5ef5edf96b33ff57d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 13:10:28 -0700 Subject: [PATCH 08/12] update macos-py3.10-requirements --- macos-py3.10-requirements.txt | 98 ++++++++++++++++++++++++++++++----- 1 file changed, 84 insertions(+), 14 deletions(-) diff --git a/macos-py3.10-requirements.txt b/macos-py3.10-requirements.txt index e6f91d0f9..d76387aa7 100644 --- a/macos-py3.10-requirements.txt +++ b/macos-py3.10-requirements.txt @@ -7,6 +7,8 @@ aiohttp==3.9.5 # via # cartesia + # langchain + # langchain-community # pipecat-ai (pyproject.toml) aiosignal==1.3.1 # via aiohttp @@ -20,7 +22,9 @@ anyio==4.4.0 # httpx # openai async-timeout==4.0.3 - # via aiohttp + # via + # aiohttp + # langchain attrs==23.2.0 # via aiohttp av==12.1.0 @@ -31,9 +35,9 @@ blinker==1.8.2 # via flask cachetools==5.3.3 # via google-auth -cartesia==0.1.0 +cartesia==0.1.1 # via pipecat-ai (pyproject.toml) -certifi==2024.2.2 +certifi==2024.6.2 # via # httpcore # httpx @@ -50,6 +54,8 @@ ctranslate2==4.2.1 # via faster-whisper daily-python==0.9.1 # via pipecat-ai (pyproject.toml) +dataclasses-json==0.6.6 + # via langchain-community distro==1.9.0 # via # anthropic @@ -82,7 +88,7 @@ frozenlist==1.4.1 # via # aiohttp # aiosignal -fsspec==2024.5.0 +fsspec==2024.6.0 # via # huggingface-hub # torch @@ -95,7 +101,7 @@ google-api-core[grpc]==2.19.0 # google-ai-generativelanguage # google-api-python-client # google-generativeai -google-api-python-client==2.131.0 +google-api-python-client==2.132.0 # via google-generativeai google-auth==2.29.0 # via @@ -108,11 +114,11 @@ google-auth-httplib2==0.2.0 # via google-api-python-client google-generativeai==0.5.4 # via pipecat-ai (pyproject.toml) -googleapis-common-protos==1.63.0 +googleapis-common-protos==1.63.1 # via # google-api-core # grpcio-status -grpcio==1.64.0 +grpcio==1.64.1 # via # google-api-core # grpcio-status @@ -157,23 +163,54 @@ jinja2==3.1.4 # via # flask # torch +jsonpatch==1.33 + # via langchain-core +jsonpointer==2.4 + # via jsonpatch +langchain==0.2.2 + # via + # langchain-community + # pipecat-ai (pyproject.toml) +langchain-community==0.2.2 + # via pipecat-ai (pyproject.toml) +langchain-core==0.2.4 + # via + # langchain + # langchain-community + # langchain-openai + # langchain-text-splitters +langchain-openai==0.1.8 + # via pipecat-ai (pyproject.toml) +langchain-text-splitters==0.2.1 + # via langchain +langsmith==0.1.69 + # via + # langchain + # langchain-community + # langchain-core loguru==0.7.2 # via pipecat-ai (pyproject.toml) markupsafe==2.1.5 # via # jinja2 # werkzeug +marshmallow==3.21.2 + # via dataclasses-json mpmath==1.3.0 # via sympy multidict==6.0.5 # via # aiohttp # yarl +mypy-extensions==1.0.0 + # via typing-inspect networkx==3.3 # via torch numpy==1.26.4 # via # ctranslate2 + # langchain + # langchain-community # onnxruntime # pipecat-ai (pyproject.toml) # pyloudnorm @@ -183,10 +220,16 @@ numpy==1.26.4 onnxruntime==1.18.0 # via faster-whisper openai==1.26.0 - # via pipecat-ai (pyproject.toml) -packaging==24.0 + # via + # langchain-openai + # pipecat-ai (pyproject.toml) +orjson==3.10.3 + # via langsmith +packaging==23.2 # via # huggingface-hub + # langchain-core + # marshmallow # onnxruntime # pytest # transformers @@ -221,12 +264,15 @@ pyaudio==0.2.14 # via pipecat-ai (pyproject.toml) pycparser==2.22 # via cffi -pydantic==2.7.2 +pydantic==2.7.3 # via # anthropic # google-generativeai + # langchain + # langchain-core + # langsmith # openai -pydantic-core==2.18.3 +pydantic-core==2.18.4 # via pydantic pyht==0.0.28 # via pipecat-ai (pyproject.toml) @@ -234,7 +280,7 @@ pyloudnorm==0.1.1 # via pipecat-ai (pyproject.toml) pyparsing==3.1.2 # via httplib2 -pytest==8.2.1 +pytest==8.2.2 # via pytest-asyncio pytest-asyncio==0.23.7 # via cartesia @@ -244,16 +290,25 @@ pyyaml==6.0.1 # via # ctranslate2 # huggingface-hub + # langchain + # langchain-community + # langchain-core # timm # transformers regex==2024.5.15 - # via transformers + # via + # tiktoken + # transformers requests==2.32.3 # via # cartesia # google-api-core # huggingface-hub + # langchain + # langchain-community + # langsmith # pyht + # tiktoken # transformers rsa==4.9 # via google-auth @@ -271,10 +326,21 @@ sniffio==1.3.1 # openai sounddevice==0.4.7 # via pipecat-ai (pyproject.toml) +sqlalchemy==2.0.30 + # via + # langchain + # langchain-community sympy==1.12.1 # via # onnxruntime # torch +tenacity==8.3.0 + # via + # langchain + # langchain-community + # langchain-core +tiktoken==0.7.0 + # via langchain-openai timm==0.9.16 # via pipecat-ai (pyproject.toml) tokenizers==0.19.1 @@ -302,7 +368,7 @@ tqdm==4.66.4 # transformers transformers==4.40.2 # via pipecat-ai (pyproject.toml) -typing-extensions==4.11.0 +typing-extensions==4.12.1 # via # anthropic # anyio @@ -312,7 +378,11 @@ typing-extensions==4.11.0 # pipecat-ai (pyproject.toml) # pydantic # pydantic-core + # sqlalchemy # torch + # typing-inspect +typing-inspect==0.9.0 + # via dataclasses-json uritemplate==4.1.1 # via google-api-python-client urllib3==2.2.1 From c8d37a7227e23f7d21e3b9d0ae7a3c7bbd1fc159 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 15:08:11 -0700 Subject: [PATCH 09/12] pipeline(runner): add support for SIGTERM --- CHANGELOG.md | 3 +++ src/pipecat/pipeline/runner.py | 10 +++++++--- 2 files changed, 10 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9eb6617ed..1c73a1832 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `PipelineTask` now has a `has_finished()` method to indicate if the task has completed. If a task is never ran `has_finished()` will return False. +- `PipelineRunner` now supports SIGTERM. If received, the runner will be + canceled. + ### Fixed - Fixed an error closing local audio transports. diff --git a/src/pipecat/pipeline/runner.py b/src/pipecat/pipeline/runner.py index 47754d82f..df480648e 100644 --- a/src/pipecat/pipeline/runner.py +++ b/src/pipecat/pipeline/runner.py @@ -43,11 +43,15 @@ class PipelineRunner: loop = asyncio.get_running_loop() loop.add_signal_handler( signal.SIGINT, - lambda *args: asyncio.create_task(self._sigint_handler()) + lambda *args: asyncio.create_task(self._sig_handler()) + ) + loop.add_signal_handler( + signal.SIGTERM, + lambda *args: asyncio.create_task(self._sig_handler()) ) - async def _sigint_handler(self): - logger.warning(f"Ctrl-C detected. Canceling runner {self}") + async def _sig_handler(self): + logger.warning(f"Interruption detected. Canceling runner {self}") await self.cancel() def __str__(self): From 891b7b22ea9353b067390a532f41fb131e943c51 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 15:11:04 -0700 Subject: [PATCH 10/12] transports: push EndFrame/CancelFrame before stopping push task --- CHANGELOG.md | 3 +++ src/pipecat/transports/base_input.py | 4 ++-- src/pipecat/transports/base_output.py | 4 ++-- 3 files changed, 7 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1c73a1832..f05d6c94f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,6 +17,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue where `BaseInputTransport` and `BaseOutputTransport` where + stopping push tasks before pushing `EndFrame` frames. + - Fixed an error closing local audio transports. - Fixed an issue with Deepgram TTS that was introduced in the previous release. diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 6fb1e2ee8..531269bec 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -87,16 +87,16 @@ class BaseInputTransport(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, CancelFrame): - await self.stop() # We don't queue a CancelFrame since we want to stop ASAP. await self.push_frame(frame, direction) + await self.stop() elif isinstance(frame, StartFrame): self._allow_interruption = frame.allow_interruptions await self.start(frame) await self._internal_push_frame(frame, direction) elif isinstance(frame, EndFrame): - await self.stop() await self._internal_push_frame(frame, direction) + await self.stop() else: await self._internal_push_frame(frame, direction) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 9e99ce736..d167919f6 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -121,11 +121,11 @@ class BaseOutputTransport(FrameProcessor): self._sink_queue.put_nowait(frame) # EndFrame is managed in the queue handler. elif isinstance(frame, CancelFrame): + await self.push_frame(frame, direction) await self.stop() - await self.push_frame(frame, direction) elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame): - await self._handle_interruptions(frame) await self.push_frame(frame, direction) + await self._handle_interruptions(frame) else: self._sink_queue.put_nowait(frame) From 854ffb0323db739187df4200d6fe2cf3f10fb01b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 15:45:17 -0700 Subject: [PATCH 11/12] update CHANGELOG for DailyRESTHelper --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index f05d6c94f..74b7aafcf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `DailyRESTHelper` which helps you create Daily rooms and tokens in an + easy way. + - `PipelineTask` now has a `has_finished()` method to indicate if the task has completed. If a task is never ran `has_finished()` will return False. From b515c28417795737124492028305e63d60ffa0c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 4 Jun 2024 19:24:33 -0700 Subject: [PATCH 12/12] services(cartesia): allow output_format and model_id --- CHANGELOG.md | 3 +++ .../foundational/07d-interruptible-cartesia.py | 4 +++- src/pipecat/services/cartesia.py | 16 ++++++++++------ 3 files changed, 16 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 74b7aafcf..d82f66bd5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Allow passing `output_format` and `model_id` to `CartesiaTTSService` to change + audio sample format and the model to use. + - Added `DailyRESTHelper` which helps you create Daily rooms and tokens in an easy way. diff --git a/examples/foundational/07d-interruptible-cartesia.py b/examples/foundational/07d-interruptible-cartesia.py index d9e5128d5..39a77492b 100644 --- a/examples/foundational/07d-interruptible-cartesia.py +++ b/examples/foundational/07d-interruptible-cartesia.py @@ -39,6 +39,7 @@ async def main(room_url: str, token): "Respond bot", DailyParams( audio_out_enabled=True, + audio_out_sample_rate=44100, transcription_enabled=True, vad_enabled=True, vad_analyzer=SileroVADAnalyzer() @@ -47,7 +48,8 @@ async def main(room_url: str, token): tts = CartesiaTTSService( api_key=os.getenv("CARTESIA_API_KEY"), - voice_name="Barbershop Man" + voice_name="British Lady", + output_format="pcm_44100" ) llm = OpenAILLMService( diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 86baa7a3e..f2d8c9b14 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -21,11 +21,15 @@ class CartesiaTTSService(TTSService): *, api_key: str, voice_name: str, + model_id: str = "upbeat-moon", + output_format: str = "pcm_16000", **kwargs): super().__init__(**kwargs) self._api_key = api_key self._voice_name = voice_name + self._model_id = model_id + self._output_format = output_format try: self._client = AsyncCartesiaTTS(api_key=self._api_key) @@ -40,14 +44,14 @@ class CartesiaTTSService(TTSService): try: chunk_generator = await self._client.generate( - transcript=text, voice=self._voice, stream=True, - model_id="upbeat-moon", data_rtype='array', output_format='pcm_16000', - # a chunk_time of 0.1 seems to be the default. there are small audio pops/gaps which - # we need to debug - chunk_time=0.1 + stream=True, + transcript=text, + voice=self._voice, + model_id=self._model_id, + output_format=self._output_format, ) async for chunk in chunk_generator: - yield AudioRawFrame(chunk['audio'], 16000, 1) + yield AudioRawFrame(chunk["audio"], chunk["sampling_rate"], 1) except Exception as e: logger.error(f"Cartesia exception: {e}")