From dcf317f2faefb44b8272719aa8364f894be560fb Mon Sep 17 00:00:00 2001 From: Maxim Makatchev Date: Thu, 16 Jan 2025 17:43:12 +0900 Subject: [PATCH 01/26] Twilio serializer reading dtmf websocket messages and generating InputDTMFFrame containing the corresponding value of KeypadEntry --- src/pipecat/frames/frames.py | 25 +++++++++++++++++++++++++ src/pipecat/serializers/twilio.py | 23 +++++++++++++++++++---- 2 files changed, 44 insertions(+), 4 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e2ac25e7a..6f0861a37 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -5,6 +5,7 @@ # from dataclasses import dataclass, field +from enum import Enum from typing import TYPE_CHECKING, Any, Awaitable, Callable, List, Literal, Mapping, Optional, Tuple from pipecat.audio.vad.vad_analyzer import VADParams @@ -18,6 +19,23 @@ if TYPE_CHECKING: from pipecat.observers.base_observer import BaseObserver +class KeypadEntry(str, Enum): + """DTMF entries.""" + + ONE = "1" + TWO = "2" + THREE = "3" + FOUR = "4" + FIVE = "5" + SIX = "6" + SEVEN = "7" + EIGHT = "8" + NINE = "9" + ZERO = "0" + POUND = "#" + STAR = "*" + + def format_pts(pts: int | None): return nanoseconds_to_str(pts) if pts else None @@ -615,6 +633,13 @@ class VisionImageRawFrame(InputImageRawFrame): return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})" +@dataclass +class InputDTMFFrame(Frame): + """A DTMF button input""" + + button: KeypadEntry + + # # Control frames # diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index 558a68046..298e8a6cb 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -10,7 +10,14 @@ import json from pydantic import BaseModel from pipecat.audio.utils import pcm_to_ulaw, ulaw_to_pcm -from pipecat.frames.frames import AudioRawFrame, Frame, InputAudioRawFrame, StartInterruptionFrame +from pipecat.frames.frames import ( + AudioRawFrame, + Frame, + InputAudioRawFrame, + InputDTMFFrame, + KeypadEntry, + StartInterruptionFrame, +) from pipecat.serializers.base_serializer import FrameSerializer, FrameSerializerType @@ -48,9 +55,7 @@ class TwilioFrameSerializer(FrameSerializer): def deserialize(self, data: str | bytes) -> Frame | None: message = json.loads(data) - if message["event"] != "media": - return None - else: + if message["event"] == "media": payload_base64 = message["media"]["payload"] payload = base64.b64decode(payload_base64) @@ -61,3 +66,13 @@ class TwilioFrameSerializer(FrameSerializer): audio=deserialized_data, num_channels=1, sample_rate=self._params.sample_rate ) return audio_frame + elif message["event"] == "dtmf": + digit = message.get("dtmf", {}).get("digit") + + try: + return InputDTMFFrame(KeypadEntry(digit)) + except ValueError as e: + # Handle case where string doesn't match any enum value + return None + else: + return None From 7ee6e7193d119bb89e9c73f45502af03b88029fb Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Thu, 16 Jan 2025 21:23:56 +0530 Subject: [PATCH 02/26] adding metric generation without deepgram VAD --- CHANGELOG.md | 2 ++ src/pipecat/services/deepgram.py | 15 +++++++++++---- 2 files changed, 13 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 232bf0f4f..127695c36 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -74,6 +74,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `UserStoppedSpeakingFrame`. This helps in faster transcriptions and clearing the `Deepgram` audio buffer. +- Changed `DeepgramSTTService` to generate metrics using pipeline VAD. + ### Fixed - Fixed an issue where websocket based TTS services could incorrectly terminate diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index bba2e72ae..f13928af1 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -21,6 +21,7 @@ from pipecat.frames.frames import ( TTSStartedFrame, TTSStoppedFrame, UserStoppedSpeakingFrame, + UserStartedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import STTService, TTSService @@ -169,7 +170,7 @@ class DeepgramSTTService(STTService): return self._settings["vad_events"] def can_generate_metrics(self) -> bool: - return self.vad_enabled + return True async def set_model(self, model: str): await super().set_model(model) @@ -210,9 +211,12 @@ class DeepgramSTTService(STTService): logger.debug("Disconnecting from Deepgram") await self._connection.finish() - async def _on_speech_started(self, *args, **kwargs): + async def start_metrics(self): await self.start_ttfb_metrics() await self.start_processing_metrics() + + async def _on_speech_started(self, *args, **kwargs): + await self.start_metrics() await self._call_event_handler("on_speech_started", *args, **kwargs) async def _on_utterance_end(self, *args, **kwargs): @@ -243,7 +247,10 @@ class DeepgramSTTService(STTService): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - if isinstance(frame, UserStoppedSpeakingFrame): + if isinstance(frame, UserStartedSpeakingFrame) and not self.vad_enabled: + # Start metrics if Deepgram VAD is disabled & pipeline VAD has detected speech + await self.start_metrics() + elif isinstance(frame, UserStoppedSpeakingFrame): # https://developers.deepgram.com/docs/finalize await self._connection.finalize() - logger.debug(f"Triggering finalize event on: {frame.name=}, {direction=}") + logger.trace(f"Triggering finalize event on: {frame.name=}, {direction=}") From 923d33eeffa28f64035f3e45707a2db819251af0 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Thu, 16 Jan 2025 21:32:48 +0530 Subject: [PATCH 03/26] fixing ruff --- src/pipecat/services/deepgram.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index f13928af1..bd54c4245 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -20,8 +20,8 @@ from pipecat.frames.frames import ( TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, - UserStoppedSpeakingFrame, UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import STTService, TTSService From 85e7d62f94e9ed4af9536b09ece92cc64ae06ea2 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Thu, 16 Jan 2025 21:36:51 +0530 Subject: [PATCH 04/26] fixing log text --- src/pipecat/services/deepgram.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index bd54c4245..a5d36370f 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -253,4 +253,4 @@ class DeepgramSTTService(STTService): elif isinstance(frame, UserStoppedSpeakingFrame): # https://developers.deepgram.com/docs/finalize await self._connection.finalize() - logger.trace(f"Triggering finalize event on: {frame.name=}, {direction=}") + logger.trace(f"Triggered finalize event on: {frame.name=}, {direction=}") From c5edbf4b7520a0c279a10787b6a67846cf599748 Mon Sep 17 00:00:00 2001 From: Maxim Makatchev Date: Fri, 17 Jan 2025 12:27:04 +0900 Subject: [PATCH 05/26] Made InputDTMFFrame a DataFrame and moved up to data frames --- src/pipecat/frames/frames.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 6f0861a37..93ef18c42 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -393,6 +393,13 @@ class TransportMessageFrame(DataFrame): return f"{self.name}(message: {self.message})" +@dataclass +class InputDTMFFrame(DataFrame): + """A DTMF button input""" + + button: KeypadEntry + + # # System frames # @@ -633,13 +640,6 @@ class VisionImageRawFrame(InputImageRawFrame): return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})" -@dataclass -class InputDTMFFrame(Frame): - """A DTMF button input""" - - button: KeypadEntry - - # # Control frames # From ff8aa6894238b7e5a1d3ec8c98237f7b6eb10061 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 16 Jan 2025 18:56:01 -0800 Subject: [PATCH 06/26] introduce heartbeat frames --- CHANGELOG.md | 7 ++ src/pipecat/frames/frames.py | 10 ++ src/pipecat/pipeline/task.py | 184 ++++++++++++++++++++++++++++------- 3 files changed, 165 insertions(+), 36 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a47979a85..5246be593 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 pipeline. This can be useful, for example, to implement frame loggers or debuggers among other things. +- Introduced heartbeat frames. The pipeline task can now push periodic + heartbeats down the pipeline when `enable_heartbeats=True`. Heartbeats are + system frames that are supposed to make it all the way to the end of the + pipeline. When a heartbeat frame is received the traversing time (i.e. the time + it took to go through the whole pipeline) will be displayed (with TRACE + logging) otherwise a warning will be shown. + - Added `30-observer.py` to show how to add an Observer to a pipeline for debugging. diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e2ac25e7a..3df7d0a33 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -424,6 +424,16 @@ class FatalErrorFrame(ErrorFrame): fatal: bool = field(default=True, init=False) +@dataclass +class HeartbeatFrame(SystemFrame): + """This frame is used by the pipeline task as a mechanism to know if the + pipeline is running properly. + + """ + + timestamp: int + + @dataclass class EndTaskFrame(SystemFrame): """This is used to notify the pipeline task that the pipeline should be diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index fc404f75b..e04e0d285 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -19,6 +19,7 @@ from pipecat.frames.frames import ( EndTaskFrame, ErrorFrame, Frame, + HeartbeatFrame, MetricsFrame, StartFrame, StopTaskFrame, @@ -29,11 +30,14 @@ from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.utils import obj_count, obj_id +HEARTBEAT_SECONDS = 1.0 + class PipelineParams(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) allow_interruptions: bool = False + enable_heartbeats: bool = False enable_metrics: bool = False enable_usage_metrics: bool = False send_initial_empty_metrics: bool = True @@ -58,25 +62,10 @@ class Source(FrameProcessor): match direction: case FrameDirection.UPSTREAM: - await self._handle_upstream_frame(frame) + await self._up_queue.put(frame) case FrameDirection.DOWNSTREAM: await self.push_frame(frame, direction) - async def _handle_upstream_frame(self, frame: Frame): - if isinstance(frame, EndTaskFrame): - # Tell the task we should end nicely. - await self._up_queue.put(EndTaskFrame()) - elif isinstance(frame, CancelTaskFrame): - # Tell the task we should end right away. - await self._up_queue.put(CancelTaskFrame()) - elif isinstance(frame, ErrorFrame): - logger.error(f"Error running app: {frame}") - if frame.fatal: - # Cancel all tasks downstream. - await self.push_frame(CancelFrame()) - # Tell the task we should stop. - await self._up_queue.put(StopTaskFrame()) - class Sink(FrameProcessor): """This is the sink processor that is linked at the end of the pipeline @@ -91,10 +80,7 @@ class Sink(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - - # We really just want to know when the EndFrame reached the sink. - if isinstance(frame, EndFrame): - await self._down_queue.put(frame) + await self._down_queue.put(frame) class Observer(BaseObserver): @@ -135,9 +121,18 @@ class PipelineTask: self._params = params self._finished = False + # This queue receives frames coming from the pipeline upstream. self._up_queue = asyncio.Queue() + # This queue receives frames coming from the pipeline downstream. self._down_queue = asyncio.Queue() + # This queue is the queue used to push frames to the pipeline. self._push_queue = asyncio.Queue() + # This is the heartbeat queue. When a heartbeat frame is received in the + # down queue we add it to the heartbeat queue for processing. + self._heartbeat_queue = asyncio.Queue() + # This event is used to indicate an EndFrame has been received in the + # down queue. + self._endframe_event = asyncio.Event() self._source = Source(self._up_queue) self._source.link(pipeline) @@ -148,33 +143,49 @@ class PipelineTask: self._observer = Observer(params.observers) def has_finished(self): + """Indicates whether the tasks has finished. That is, all processors + have stopped. + + """ return self._finished async def stop_when_done(self): + """This is a helper function that sends an EndFrame to the pipeline in + order to stop the task after everything in it has been processed. + + """ logger.debug(f"Task {self} scheduled to stop when done") await self.queue_frame(EndFrame()) async def cancel(self): + """ + Stops the running pipeline immediately. + """ logger.debug(f"Canceling pipeline task {self}") # Make sure everything is cleaned up downstream. This is sent # out-of-band from the main streaming task which is what we want since # we want to cancel right away. await self._source.push_frame(CancelFrame()) - self._process_push_task.cancel() - self._process_up_task.cancel() - await self._process_push_task - await self._process_up_task + await self._cancel_tasks(True) async def run(self): - self._process_up_task = asyncio.create_task(self._process_up_queue()) - self._process_push_task = asyncio.create_task(self._process_push_queue()) - await asyncio.gather(self._process_up_task, self._process_push_task) + """ + Starts running the given pipeline. + """ + tasks = self._create_tasks() + await asyncio.gather(*tasks) self._finished = True async def queue_frame(self, frame: Frame): + """ + Queue a frame to be pushed down the pipeline. + """ await self._push_queue.put(frame) async def queue_frames(self, frames: Iterable[Frame] | AsyncIterable[Frame]): + """ + Queues multiple frames to be pushed down the pipeline. + """ if isinstance(frames, AsyncIterable): async for frame in frames: await self.queue_frame(frame) @@ -182,6 +193,39 @@ class PipelineTask: for frame in frames: await self.queue_frame(frame) + def _create_tasks(self): + tasks = [] + self._process_up_task = asyncio.create_task(self._process_up_queue()) + self._process_down_task = asyncio.create_task(self._process_down_queue()) + self._process_push_task = asyncio.create_task(self._process_push_queue()) + + tasks = [self._process_up_task, self._process_down_task, self._process_push_task] + + if self._params.enable_heartbeats: + self._heartbeat_push_task = asyncio.create_task(self._heartbeat_push_handler()) + self._heartbeat_monitor_task = asyncio.create_task(self._heartbeat_monitor_handler()) + tasks.append(self._heartbeat_push_task) + tasks.append(self._heartbeat_monitor_task) + + return tasks + + async def _cancel_tasks(self, cancel_push: bool): + if cancel_push: + self._process_push_task.cancel() + await self._process_push_task + + self._process_up_task.cancel() + await self._process_up_task + + self._process_down_task.cancel() + await self._process_down_task + + if self._params.enable_heartbeats: + self._heartbeat_push_task.cancel() + await self._heartbeat_push_task + self._heartbeat_monitor_task.cancel() + await self._heartbeat_monitor_task + def _initial_metrics_frame(self) -> MetricsFrame: processors = self._pipeline.processors_with_metrics() data = [] @@ -190,7 +234,16 @@ class PipelineTask: data.append(ProcessingMetricsData(processor=p.name, value=0.0)) return MetricsFrame(data=data) + async def _wait_for_endframe(self): + await self._endframe_event.wait() + self._endframe_event.clear() + async def _process_push_queue(self): + """This is the task that runs the pipeline for the first time by sending + a StartFrame and by pushing any other frames queued by the user. It runs + until the tasks is canceled or stopped (e.g. with an EndFrame). + + """ self._clock.start() start_frame = StartFrame( @@ -224,29 +277,88 @@ class PipelineTask: await self._source.cleanup() await self._pipeline.cleanup() await self._sink.cleanup() - # We just enqueue None to terminate the task gracefully. - self._process_up_task.cancel() - await self._process_up_task - - async def _wait_for_endframe(self): - # NOTE(aleix): the Sink element just pushes EndFrames to the down queue, - # so just wait for it. In the future we might do something else here, - # but for now this is fine. - await self._down_queue.get() + # Finally, cancel internal tasks. We don't cancel the push tasks because + # that's us. + await self._cancel_tasks(False) async def _process_up_queue(self): + """This is the task that processes frames coming upstream from the + pipeline. These frames might indicate, for example, that we want the + pipeline to be stopped (e.g. EndTaskFrame) in which case we would send + an EndFrame down the pipeline. + + """ while True: try: frame = await self._up_queue.get() if isinstance(frame, EndTaskFrame): + # Tell the task we should end nicely. await self.queue_frame(EndFrame()) elif isinstance(frame, CancelTaskFrame): + # Tell the task we should end right away. await self.queue_frame(CancelFrame()) elif isinstance(frame, StopTaskFrame): await self.queue_frame(StopTaskFrame()) + elif isinstance(frame, ErrorFrame): + logger.error(f"Error running app: {frame}") + if frame.fatal: + # Cancel all tasks downstream. + await self.queue_frame(CancelFrame()) + # Tell the task we should stop. + await self.queue_frame(StopTaskFrame()) self._up_queue.task_done() except asyncio.CancelledError: break + async def _process_down_queue(self): + """This tasks process frames coming downstream from the pipeline. For + example, heartbeat frames or an EndFrame which would indicate all + processors have handled the EndFrame and therefore we can exit the task + cleanly. + + """ + while True: + try: + frame = await self._down_queue.get() + if isinstance(frame, EndFrame): + self._endframe_event.set() + elif isinstance(frame, HeartbeatFrame): + await self._heartbeat_queue.put(frame) + self._down_queue.task_done() + except asyncio.CancelledError: + break + + async def _heartbeat_push_handler(self): + """ + This tasks pushes a heartbeat frame every HEARTBEAT_SECONDS. + """ + while True: + try: + await self.queue_frame(HeartbeatFrame(timestamp=self._clock.get_time())) + await asyncio.sleep(HEARTBEAT_SECONDS) + except asyncio.CancelledError: + break + + async def _heartbeat_monitor_handler(self): + """This tasks monitors heartbeat frames. If a heartbeat frame has not + been received for a long period a warning will be logged. It also logs + the time that a heartbeat frame takes to processes, that is how long it + takes for the heartbeat frame to traverse all the pipeline. + + """ + wait_time = HEARTBEAT_SECONDS * 2 + while True: + try: + frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time) + process_time = (self._clock.get_time() - frame.timestamp) / 1_000_000_000 + logger.trace(f"{self}: heartbeat frame processed in {process_time} seconds") + self._heartbeat_queue.task_done() + except asyncio.TimeoutError: + logger.warning( + f"{self}: heartbeat frame not received for more than {wait_time} seconds" + ) + except asyncio.CancelledError: + break + def __str__(self): return self.name From 2503f761078fedc14dd1b655f44efb918440db26 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 16 Jan 2025 18:56:21 -0800 Subject: [PATCH 07/26] examples: add 31-heartbeats.py --- CHANGELOG.md | 15 ++++----- examples/foundational/31-heartbeats.py | 43 ++++++++++++++++++++++++++ 2 files changed, 51 insertions(+), 7 deletions(-) create mode 100644 examples/foundational/31-heartbeats.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 5246be593..27a620e5d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,17 +12,18 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Introduced pipeline frame observers. Observers can view all the frames that go through the pipeline without the need to inject processors in the pipeline. This can be useful, for example, to implement frame loggers or - debuggers among other things. + debuggers among other things. The example + `examples/foundational/30-observer.py` shows how to add an observer to a + pipeline for debugging. - Introduced heartbeat frames. The pipeline task can now push periodic heartbeats down the pipeline when `enable_heartbeats=True`. Heartbeats are system frames that are supposed to make it all the way to the end of the - pipeline. When a heartbeat frame is received the traversing time (i.e. the time - it took to go through the whole pipeline) will be displayed (with TRACE - logging) otherwise a warning will be shown. - -- Added `30-observer.py` to show how to add an Observer to a pipeline for - debugging. + pipeline. When a heartbeat frame is received the traversing time (i.e. the + time it took to go through the whole pipeline) will be displayed (with TRACE + logging) otherwise a warning will be shown. The example + `examples/foundational/31-heartbeats.py` shows how to enable heartbeats and + forces warnings to be displayed. - Added `OpenRouter` for OpenRouter integration with an OpenAI-compatible interface. Added foundational example `14m-function-calling-openrouter.py`. diff --git a/examples/foundational/31-heartbeats.py b/examples/foundational/31-heartbeats.py new file mode 100644 index 000000000..fbb959519 --- /dev/null +++ b/examples/foundational/31-heartbeats.py @@ -0,0 +1,43 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +import sys + +from loguru import logger + +from pipecat.frames.frames import Frame +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +class NullProcessor(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + +async def main(): + """This test shows heartbeat monitoring by displaying a warning when + heartbeats are not received. + + """ + + pipeline = Pipeline([NullProcessor()]) + + task = PipelineTask(pipeline, PipelineParams(enable_heartbeats=True)) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) From 225b65c3d2848bb58bb911d8f0813f232e03a3eb Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Jan 2025 22:39:36 -0500 Subject: [PATCH 08/26] Add ElevenLabsHttpTTSService --- .../07d-interruptible-elevenlabs-http.py | 103 ++++++++++++++++ src/pipecat/services/elevenlabs.py | 115 ++++++++++++++++++ 2 files changed, 218 insertions(+) create mode 100644 examples/foundational/07d-interruptible-elevenlabs-http.py diff --git a/examples/foundational/07d-interruptible-elevenlabs-http.py b/examples/foundational/07d-interruptible-elevenlabs-http.py new file mode 100644 index 000000000..1f09680d2 --- /dev/null +++ b/examples/foundational/07d-interruptible-elevenlabs-http.py @@ -0,0 +1,103 @@ +# +# 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 runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import EndFrame +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.elevenlabs import ElevenLabsHttpTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport + +load_dotenv(override=True) + +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Respond bot", + DailyParams( + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + ) + + tts = ElevenLabsHttpTTSService( + api_key=os.getenv("ELEVENLABS_API_KEY", ""), + voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + + 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 + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + PipelineParams( + allow_interruptions=True, + enable_metrics=True, + enable_usage_metrics=True, + report_only_initial_ttfb=True, + ), + ) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + # 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_participant_left") + async def on_participant_left(transport, participant, reason): + await task.queue_frame(EndFrame()) + + runner = PipelineRunner() + + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index c1d326dfc..6c4c5469e 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -6,9 +6,11 @@ import asyncio import base64 +import io import json from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, Tuple +import aiohttp from loguru import logger from pydantic import BaseModel, model_validator @@ -33,6 +35,8 @@ from pipecat.transcriptions.language import Language # See .env.example for ElevenLabs configuration needed try: import websockets + from elevenlabs import Voice, VoiceSettings + from elevenlabs.client import ElevenLabs except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( @@ -418,3 +422,114 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): yield None except Exception as e: logger.error(f"{self} exception: {e}") + + +class ElevenLabsHttpTTSService(WordTTSService): + class InputParams(BaseModel): + language: Optional[Language] = Language.EN + optimize_streaming_latency: Optional[str] = None + stability: Optional[float] = None + similarity_boost: Optional[float] = None + style: Optional[float] = None + use_speaker_boost: Optional[bool] = None + + def __init__( + self, + *, + api_key: str, + voice_id: str, + model: str = "eleven_flash_v2_5", + output_format: ElevenLabsOutputFormat = "pcm_24000", + params: InputParams = InputParams(), + **kwargs, + ): + sample_rate = self._sample_rate_from_output_format(output_format) + super().__init__( + aggregate_sentences=True, + push_text_frames=False, + push_stop_frames=True, + stop_frame_timeout_s=2.0, + sample_rate=sample_rate, + **kwargs, + ) + + self._client = ElevenLabs(api_key=api_key) + self._voice_id = voice_id + self._model = model + self._output_format = output_format + + # Create voice settings if provided + self._voice_settings = None + if params.stability is not None and params.similarity_boost is not None: + self._voice_settings = VoiceSettings( + stability=params.stability, + similarity_boost=params.similarity_boost, + style=params.style or 0.0, + use_speaker_boost=params.use_speaker_boost or False, + ) + + logger.debug(f"Initialized with sample rate: {sample_rate}") + + @staticmethod + def _sample_rate_from_output_format(output_format: str) -> int: + return { + "pcm_16000": 16000, + "pcm_22050": 22050, + "pcm_24000": 24000, + "pcm_44100": 44100, + }[output_format] + + async def start(self, frame: StartFrame): + await super().start(frame) + + async def stop(self, frame: EndFrame): + await super().stop(frame) + + async def cancel(self, frame: CancelFrame): + await super().cancel(frame) + + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + def read_audio_stream(**kwargs): + # Run the streaming in a separate thread + audio_chunks = [] + stream = self._client.text_to_speech.convert_as_stream(**kwargs) + for chunk in stream: + if chunk: + audio_chunks.append(chunk) + return b"".join(audio_chunks) + + try: + yield TTSStartedFrame() + + # Prepare parameters + params = { + "text": text, + "voice_id": self._voice_id, + "model_id": self._model, + "output_format": self._output_format, + "voice_settings": self._voice_settings, + "optimize_streaming_latency": 4, # Maximum optimization + disabled text normalizer + } + + # Get audio data in a separate thread + audio_data = await asyncio.to_thread(read_audio_stream, **params) + + if not audio_data: + logger.error(f"{self} No audio data returned") + yield None + return + + # Stream the audio data in chunks + chunk_size = 4096 # Adjust this value as needed + for i in range(0, len(audio_data), chunk_size): + chunk = audio_data[i : i + chunk_size] + if len(chunk) > 0: + yield TTSAudioRawFrame( + chunk, self._sample_rate_from_output_format(self._output_format), 1 + ) + + yield TTSStoppedFrame() + + except Exception as e: + logger.error(f"Error in run_tts: {e}") + yield TTSStoppedFrame() From 740d2743df16149b031186fdf1e89c2bfee297c0 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Jan 2025 22:56:25 -0500 Subject: [PATCH 09/26] Add TTFB metrics --- CHANGELOG.md | 3 +++ pyproject.toml | 2 +- src/pipecat/services/elevenlabs.py | 33 ++++++++++++++++++++++-------- 3 files changed, 28 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a47979a85..e2870e9a0 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 `ElevenLabsHttpTTSService` and the + `07d-interruptible-elevenlabs-http.py` foundational example. + - Introduced pipeline frame observers. Observers can view all the frames that go through the pipeline without the need to inject processors in the pipeline. This can be useful, for example, to implement frame loggers or diff --git a/pyproject.toml b/pyproject.toml index ee842a533..b50f9d180 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,7 +51,7 @@ cerebras = [ "openai~=1.59.6" ] deepseek = [ "openai~=1.59.6" ] daily = [ "daily-python~=0.14.2" ] deepgram = [ "deepgram-sdk~=3.8.0" ] -elevenlabs = [ "websockets~=13.1" ] +elevenlabs = [ "elevenlabs~=1.50.3","websockets~=13.1" ] fal = [ "fal-client~=0.5.6" ] fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ] gladia = [ "websockets~=13.1" ] diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 6c4c5469e..cca421052 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -6,11 +6,9 @@ import asyncio import base64 -import io import json from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, Tuple -import aiohttp from loguru import logger from pydantic import BaseModel, model_validator @@ -18,6 +16,7 @@ from pipecat.frames.frames import ( BotStoppedSpeakingFrame, CancelFrame, EndFrame, + ErrorFrame, Frame, LLMFullResponseEndFrame, StartFrame, @@ -28,14 +27,14 @@ from pipecat.frames.frames import ( TTSStoppedFrame, ) from pipecat.processors.frame_processor import FrameDirection -from pipecat.services.ai_services import WordTTSService +from pipecat.services.ai_services import TTSService, WordTTSService from pipecat.services.websocket_service import WebsocketService from pipecat.transcriptions.language import Language # See .env.example for ElevenLabs configuration needed try: import websockets - from elevenlabs import Voice, VoiceSettings + from elevenlabs import VoiceSettings from elevenlabs.client import ElevenLabs except ModuleNotFoundError as e: logger.error(f"Exception: {e}") @@ -424,7 +423,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): logger.error(f"{self} exception: {e}") -class ElevenLabsHttpTTSService(WordTTSService): +class ElevenLabsHttpTTSService(TTSService): class InputParams(BaseModel): language: Optional[Language] = Language.EN optimize_streaming_latency: Optional[str] = None @@ -479,6 +478,9 @@ class ElevenLabsHttpTTSService(WordTTSService): "pcm_44100": 44100, }[output_format] + def can_generate_metrics(self) -> bool: + return True + async def start(self, frame: StartFrame): await super().start(frame) @@ -490,7 +492,6 @@ class ElevenLabsHttpTTSService(WordTTSService): async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: def read_audio_stream(**kwargs): - # Run the streaming in a separate thread audio_chunks = [] stream = self._client.text_to_speech.convert_as_stream(**kwargs) for chunk in stream: @@ -498,8 +499,11 @@ class ElevenLabsHttpTTSService(WordTTSService): audio_chunks.append(chunk) return b"".join(audio_chunks) + logger.debug(f"Generating TTS: [{text}]") + try: - yield TTSStartedFrame() + # Start TTFB metrics before any processing + await self.start_ttfb_metrics() # Prepare parameters params = { @@ -508,7 +512,7 @@ class ElevenLabsHttpTTSService(WordTTSService): "model_id": self._model, "output_format": self._output_format, "voice_settings": self._voice_settings, - "optimize_streaming_latency": 4, # Maximum optimization + disabled text normalizer + "optimize_streaming_latency": 4, } # Get audio data in a separate thread @@ -519,11 +523,19 @@ class ElevenLabsHttpTTSService(WordTTSService): yield None return + # Start usage metrics before sending any frames + await self.start_tts_usage_metrics(text) + + yield TTSStartedFrame() + # Stream the audio data in chunks - chunk_size = 4096 # Adjust this value as needed + chunk_size = 4096 for i in range(0, len(audio_data), chunk_size): chunk = audio_data[i : i + chunk_size] if len(chunk) > 0: + # Stop TTFB metrics on first chunk + await self.stop_ttfb_metrics() + yield TTSAudioRawFrame( chunk, self._sample_rate_from_output_format(self._output_format), 1 ) @@ -532,4 +544,7 @@ class ElevenLabsHttpTTSService(WordTTSService): except Exception as e: logger.error(f"Error in run_tts: {e}") + yield ErrorFrame(error=str(e)) + + finally: yield TTSStoppedFrame() From d51893f61ccec3f0828f74478be8da51ac50a69a Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 16 Jan 2025 23:49:53 -0500 Subject: [PATCH 10/26] Refactor for aiohttp, correct use of settings --- .../07d-interruptible-elevenlabs-http.py | 1 + pyproject.toml | 2 +- src/pipecat/services/elevenlabs.py | 188 +++++++++++------- 3 files changed, 114 insertions(+), 77 deletions(-) diff --git a/examples/foundational/07d-interruptible-elevenlabs-http.py b/examples/foundational/07d-interruptible-elevenlabs-http.py index 1f09680d2..4b8aefd98 100644 --- a/examples/foundational/07d-interruptible-elevenlabs-http.py +++ b/examples/foundational/07d-interruptible-elevenlabs-http.py @@ -48,6 +48,7 @@ async def main(): tts = ElevenLabsHttpTTSService( api_key=os.getenv("ELEVENLABS_API_KEY", ""), voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), + # params=ElevenLabsHttpTTSService.InputParams(language="en"), ) llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") diff --git a/pyproject.toml b/pyproject.toml index b50f9d180..ee842a533 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,7 +51,7 @@ cerebras = [ "openai~=1.59.6" ] deepseek = [ "openai~=1.59.6" ] daily = [ "daily-python~=0.14.2" ] deepgram = [ "deepgram-sdk~=3.8.0" ] -elevenlabs = [ "elevenlabs~=1.50.3","websockets~=13.1" ] +elevenlabs = [ "websockets~=13.1" ] fal = [ "fal-client~=0.5.6" ] fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ] gladia = [ "websockets~=13.1" ] diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index cca421052..0ad542821 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -7,8 +7,9 @@ import asyncio import base64 import json -from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, Tuple +from typing import Any, AsyncGenerator, Dict, List, Literal, Mapping, Optional, Tuple, Union +import aiohttp from loguru import logger from pydantic import BaseModel, model_validator @@ -424,9 +425,20 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService): class ElevenLabsHttpTTSService(TTSService): + """ElevenLabs Text-to-Speech service using HTTP streaming. + + Args: + api_key: ElevenLabs API key + voice_id: ID of the voice to use + model: Model ID (default: "eleven_flash_v2_5" for low latency) + base_url: API base URL + output_format: Audio output format (PCM) + params: Additional parameters for voice configuration + """ + class InputParams(BaseModel): language: Optional[Language] = Language.EN - optimize_streaming_latency: Optional[str] = None + optimize_streaming_latency: Optional[int] = None stability: Optional[float] = None similarity_boost: Optional[float] = None style: Optional[float] = None @@ -438,109 +450,133 @@ class ElevenLabsHttpTTSService(TTSService): api_key: str, voice_id: str, model: str = "eleven_flash_v2_5", + base_url: str = "https://api.elevenlabs.io", output_format: ElevenLabsOutputFormat = "pcm_24000", params: InputParams = InputParams(), **kwargs, ): - sample_rate = self._sample_rate_from_output_format(output_format) - super().__init__( - aggregate_sentences=True, - push_text_frames=False, - push_stop_frames=True, - stop_frame_timeout_s=2.0, - sample_rate=sample_rate, - **kwargs, - ) + sample_rate = sample_rate_from_output_format(output_format) + super().__init__(sample_rate=sample_rate, **kwargs) - self._client = ElevenLabs(api_key=api_key) - self._voice_id = voice_id - self._model = model + self._api_key = api_key + self._base_url = base_url self._output_format = output_format + self._params = params + self._session: Optional[aiohttp.ClientSession] = None - # Create voice settings if provided - self._voice_settings = None - if params.stability is not None and params.similarity_boost is not None: - self._voice_settings = VoiceSettings( - stability=params.stability, - similarity_boost=params.similarity_boost, - style=params.style or 0.0, - use_speaker_boost=params.use_speaker_boost or False, - ) - - logger.debug(f"Initialized with sample rate: {sample_rate}") - - @staticmethod - def _sample_rate_from_output_format(output_format: str) -> int: - return { - "pcm_16000": 16000, - "pcm_22050": 22050, - "pcm_24000": 24000, - "pcm_44100": 44100, - }[output_format] + self._settings = { + "sample_rate": sample_rate_from_output_format(output_format), + "language": self.language_to_service_language(params.language) + if params.language + else "en", + "output_format": output_format, + "optimize_streaming_latency": params.optimize_streaming_latency, + "stability": params.stability, + "similarity_boost": params.similarity_boost, + "style": params.style, + "use_speaker_boost": params.use_speaker_boost, + } + self.set_model_name(model) + self.set_voice(voice_id) + self._voice_settings = self._set_voice_settings() def can_generate_metrics(self) -> bool: return True + def _set_voice_settings(self) -> Optional[Dict[str, Union[float, bool]]]: + voice_settings: Dict[str, Union[float, bool]] = {} + if ( + self._settings["stability"] is not None + and self._settings["similarity_boost"] is not None + ): + voice_settings["stability"] = float(self._settings["stability"]) + voice_settings["similarity_boost"] = float(self._settings["similarity_boost"]) + if self._settings["style"] is not None: + voice_settings["style"] = float(self._settings["style"]) + if self._settings["use_speaker_boost"] is not None: + voice_settings["use_speaker_boost"] = bool(self._settings["use_speaker_boost"]) + else: + if self._settings["style"] is not None: + logger.warning( + "'style' is set but will not be applied because 'stability' and 'similarity_boost' are not both set." + ) + if self._settings["use_speaker_boost"] is not None: + logger.warning( + "'use_speaker_boost' is set but will not be applied because 'stability' and 'similarity_boost' are not both set." + ) + + return voice_settings or None + async def start(self, frame: StartFrame): await super().start(frame) + self._session = aiohttp.ClientSession() async def stop(self, frame: EndFrame): await super().stop(frame) + if self._session: + await self._session.close() + self._session = None async def cancel(self, frame: CancelFrame): await super().cancel(frame) + if self._session: + await self._session.close() + self._session = None async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - def read_audio_stream(**kwargs): - audio_chunks = [] - stream = self._client.text_to_speech.convert_as_stream(**kwargs) - for chunk in stream: - if chunk: - audio_chunks.append(chunk) - return b"".join(audio_chunks) - logger.debug(f"Generating TTS: [{text}]") + if not self._session: + self._session = aiohttp.ClientSession() + + url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream" + + payload = { + "text": text, + "model_id": self._model_name, + } + + if self._voice_settings: + payload["voice_settings"] = json.dumps(self._voice_settings) + + if self._settings["language"]: + payload["language_code"] = self._settings["language"] + + headers = { + "xi-api-key": self._api_key, + "Content-Type": "application/json", + } + + # Build query parameters + params = { + "output_format": self._output_format, + } + if self._settings["optimize_streaming_latency"] is not None: + params["optimize_streaming_latency"] = self._settings["optimize_streaming_latency"] + + logger.debug(f"ElevenLabs request - payload: {payload}, params: {params}") + try: - # Start TTFB metrics before any processing await self.start_ttfb_metrics() - # Prepare parameters - params = { - "text": text, - "voice_id": self._voice_id, - "model_id": self._model, - "output_format": self._output_format, - "voice_settings": self._voice_settings, - "optimize_streaming_latency": 4, - } + async with self._session.post( + url, json=payload, headers=headers, params=params + ) as response: + if response.status != 200: + error_text = await response.text() + logger.error(f"{self} error: {error_text}") + yield ErrorFrame(error=f"ElevenLabs API error: {error_text}") + return - # Get audio data in a separate thread - audio_data = await asyncio.to_thread(read_audio_stream, **params) + await self.start_tts_usage_metrics(text) + yield TTSStartedFrame() - if not audio_data: - logger.error(f"{self} No audio data returned") - yield None - return + async for chunk in response.content: + if chunk: + await self.stop_ttfb_metrics() + yield TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1) - # Start usage metrics before sending any frames - await self.start_tts_usage_metrics(text) - - yield TTSStartedFrame() - - # Stream the audio data in chunks - chunk_size = 4096 - for i in range(0, len(audio_data), chunk_size): - chunk = audio_data[i : i + chunk_size] - if len(chunk) > 0: - # Stop TTFB metrics on first chunk - await self.stop_ttfb_metrics() - - yield TTSAudioRawFrame( - chunk, self._sample_rate_from_output_format(self._output_format), 1 - ) - - yield TTSStoppedFrame() + yield TTSStoppedFrame() except Exception as e: logger.error(f"Error in run_tts: {e}") From c44455796590202fc45a1fcc0b362825c0994144 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Fri, 17 Jan 2025 19:50:53 +0530 Subject: [PATCH 11/26] fixing IdleFrameProcessor and UserIdleProcessor init logic --- CHANGELOG.md | 3 +++ src/pipecat/processors/idle_frame_processor.py | 7 ++++--- src/pipecat/processors/user_idle_processor.py | 6 ++++-- 3 files changed, 11 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a47979a85..be58657a0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -95,6 +95,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue where setting the voice and model for `RimeHttpTTSService` wasn't working. +- Fixed an issue where `IdleFrameProcessor` and `UserIdleProcessor` were getting + initialized before the start of the pipeline. + ## [0.0.52] - 2024-12-24 ### Added diff --git a/src/pipecat/processors/idle_frame_processor.py b/src/pipecat/processors/idle_frame_processor.py index 3ed52c354..3f5f51e45 100644 --- a/src/pipecat/processors/idle_frame_processor.py +++ b/src/pipecat/processors/idle_frame_processor.py @@ -7,7 +7,7 @@ import asyncio from typing import Awaitable, Callable, List -from pipecat.frames.frames import Frame +from pipecat.frames.frames import Frame, StartFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -31,11 +31,12 @@ class IdleFrameProcessor(FrameProcessor): self._timeout = timeout self._types = types - self._create_idle_task() - async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) + if isinstance(frame, StartFrame): + self._create_idle_task() + await self.push_frame(frame, direction) # If we are not waiting for any specific frame set the event, otherwise diff --git a/src/pipecat/processors/user_idle_processor.py b/src/pipecat/processors/user_idle_processor.py index 44027787a..754bbca20 100644 --- a/src/pipecat/processors/user_idle_processor.py +++ b/src/pipecat/processors/user_idle_processor.py @@ -14,6 +14,7 @@ from pipecat.frames.frames import ( Frame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, + StartFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -36,7 +37,6 @@ class UserIdleProcessor(FrameProcessor): self._callback = callback self._timeout = timeout self._interrupted = False - self._create_idle_task() async def _stop(self): self._idle_task.cancel() @@ -46,7 +46,9 @@ class UserIdleProcessor(FrameProcessor): await super().process_frame(frame, direction) # Check for end frames before processing - if isinstance(frame, (EndFrame, CancelFrame)): + if isinstance(frame, StartFrame): + self._create_idle_task() + elif isinstance(frame, (EndFrame, CancelFrame)): await self._stop() await self.push_frame(frame, direction) From 80779c48d619f5a192cfeccfb329cac90fe252ca Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Fri, 17 Jan 2025 20:07:25 +0530 Subject: [PATCH 12/26] sort fix --- src/pipecat/processors/user_idle_processor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/processors/user_idle_processor.py b/src/pipecat/processors/user_idle_processor.py index 754bbca20..8f40fd52e 100644 --- a/src/pipecat/processors/user_idle_processor.py +++ b/src/pipecat/processors/user_idle_processor.py @@ -12,9 +12,9 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, + StartFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, - StartFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor From 85f4663a41ce37e91f0e7ae449833d929f1ffcb9 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 17 Jan 2025 12:54:17 -0500 Subject: [PATCH 13/26] Start UserIdleProcessor on speaking frame, fix bug not pushing EndFrame --- CHANGELOG.md | 7 ++ src/pipecat/processors/user_idle_processor.py | 93 ++++++++++++++----- 2 files changed, 76 insertions(+), 24 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index be58657a0..07696a739 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -56,6 +56,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Modified `UserIdleProcessor` to start monitoring only after first + conversation activity (`UserStartedSpeakingFrame` or + `BotStartedSpeakingFrame`) instead of immediately. + - Modified `OpenAIAssistantContextAggregator` to support controlled completions and to emit context update callbacks via `FunctionCallResultProperties`. @@ -79,6 +83,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed `UserIdleProcessor` not properly propagating `EndFrame`s through the + pipeline. + - Fixed an issue where websocket based TTS services could incorrectly terminate their connection due to a retry counter not resetting. diff --git a/src/pipecat/processors/user_idle_processor.py b/src/pipecat/processors/user_idle_processor.py index 8f40fd52e..3a7202c80 100644 --- a/src/pipecat/processors/user_idle_processor.py +++ b/src/pipecat/processors/user_idle_processor.py @@ -12,7 +12,6 @@ from pipecat.frames.frames import ( CancelFrame, EndFrame, Frame, - StartFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, ) @@ -20,10 +19,24 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class UserIdleProcessor(FrameProcessor): - """This class is useful to check if the user is interacting with the bot - within a given timeout. If the timeout is reached before any interaction - occurred the provided callback will be called. + """Monitors user inactivity and triggers callbacks after timeout periods. + Starts monitoring only after the first conversation activity (UserStartedSpeaking + or BotSpeaking). + + Args: + callback: Function to call when user is idle + timeout: Seconds to wait before considering user idle + **kwargs: Additional arguments passed to FrameProcessor + + Example: + async def handle_idle(processor: "UserIdleProcessor") -> None: + await send_reminder("Are you still there?") + + processor = UserIdleProcessor( + callback=handle_idle, + timeout=5.0 + ) """ def __init__( @@ -37,40 +50,72 @@ class UserIdleProcessor(FrameProcessor): self._callback = callback self._timeout = timeout self._interrupted = False + self._conversation_started = False + self._idle_task = None + self._idle_event = asyncio.Event() + + def _create_idle_task(self): + """Create the idle task if it hasn't been created yet.""" + if self._idle_task is None: + self._idle_task = self.get_event_loop().create_task(self._idle_task_handler()) async def _stop(self): - self._idle_task.cancel() - await self._idle_task + """Stops and cleans up the idle monitoring task.""" + if self._idle_task is not None: + self._idle_task.cancel() + try: + await self._idle_task + except asyncio.CancelledError: + pass # Expected when task is cancelled + self._idle_task = None async def process_frame(self, frame: Frame, direction: FrameDirection): + """Processes incoming frames and manages idle monitoring state. + + Args: + frame: The frame to process + direction: Direction of the frame flow + """ await super().process_frame(frame, direction) # Check for end frames before processing - if isinstance(frame, StartFrame): - self._create_idle_task() - elif isinstance(frame, (EndFrame, CancelFrame)): - await self._stop() + if isinstance(frame, (EndFrame, CancelFrame)): + await self.push_frame(frame, direction) # Push the frame down the pipeline + if self._idle_task: + await self._stop() # Stop the idle task, if it exists + return await self.push_frame(frame, direction) - # We shouldn't call the idle callback if the user or the bot are speaking - if isinstance(frame, UserStartedSpeakingFrame): - self._interrupted = True - self._idle_event.set() - elif isinstance(frame, UserStoppedSpeakingFrame): - self._interrupted = False - self._idle_event.set() - elif isinstance(frame, BotSpeakingFrame): - self._idle_event.set() + # Start monitoring on first conversation activity + if not self._conversation_started and isinstance( + frame, (UserStartedSpeakingFrame, BotSpeakingFrame) + ): + self._conversation_started = True + self._create_idle_task() + + # Only process these events if conversation has started + if self._conversation_started: + # We shouldn't call the idle callback if the user or the bot are speaking + if isinstance(frame, UserStartedSpeakingFrame): + self._interrupted = True + self._idle_event.set() + elif isinstance(frame, UserStoppedSpeakingFrame): + self._interrupted = False + self._idle_event.set() + elif isinstance(frame, BotSpeakingFrame): + self._idle_event.set() async def cleanup(self): - await self._stop() - - def _create_idle_task(self): - self._idle_event = asyncio.Event() - self._idle_task = self.get_event_loop().create_task(self._idle_task_handler()) + """Cleans up resources when processor is shutting down.""" + if self._idle_task: # Only stop if task exists + await self._stop() async def _idle_task_handler(self): + """Monitors for idle timeout and triggers callbacks. + + Runs in a loop until cancelled. + """ while True: try: await asyncio.wait_for(self._idle_event.wait(), timeout=self._timeout) From f22a00570d96480146093203d12a526e859a99db Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Jan 2025 10:03:13 -0800 Subject: [PATCH 14/26] task: start heartbeats task when push task starts --- src/pipecat/pipeline/task.py | 12 ++++++++---- 1 file changed, 8 insertions(+), 4 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index e04e0d285..f0e2089e4 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -201,15 +201,16 @@ class PipelineTask: tasks = [self._process_up_task, self._process_down_task, self._process_push_task] + return tasks + + def _maybe_start_heartbeat_tasks(self): if self._params.enable_heartbeats: self._heartbeat_push_task = asyncio.create_task(self._heartbeat_push_handler()) self._heartbeat_monitor_task = asyncio.create_task(self._heartbeat_monitor_handler()) - tasks.append(self._heartbeat_push_task) - tasks.append(self._heartbeat_monitor_task) - - return tasks async def _cancel_tasks(self, cancel_push: bool): + await self._maybe_cancel_heartbeat_tasks() + if cancel_push: self._process_push_task.cancel() await self._process_push_task @@ -220,6 +221,7 @@ class PipelineTask: self._process_down_task.cancel() await self._process_down_task + async def _maybe_cancel_heartbeat_tasks(self): if self._params.enable_heartbeats: self._heartbeat_push_task.cancel() await self._heartbeat_push_task @@ -246,6 +248,8 @@ class PipelineTask: """ self._clock.start() + self._maybe_start_heartbeat_tasks() + start_frame = StartFrame( allow_interruptions=self._params.allow_interruptions, enable_metrics=self._params.enable_metrics, From da0c4cfd99584269bde678e470b3f0abed757382 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Jan 2025 10:04:05 -0800 Subject: [PATCH 15/26] task: increase heartbeat monitoring to 5 seconds --- src/pipecat/pipeline/task.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index f0e2089e4..a77997d08 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -350,7 +350,7 @@ class PipelineTask: takes for the heartbeat frame to traverse all the pipeline. """ - wait_time = HEARTBEAT_SECONDS * 2 + wait_time = HEARTBEAT_SECONDS * 5 while True: try: frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time) From 4b3c776f582ce708ddb2e2b93fb772a499649dd2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Jan 2025 10:04:24 -0800 Subject: [PATCH 16/26] task: don't use push queue to send a heartbeat This is because we might be waiting for the EndFrame. Currently, if we push an EndFrame to the task, the task will block until the EndFrame traverses all the pipeline. --- src/pipecat/pipeline/task.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index a77997d08..9a6d344ea 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -338,7 +338,10 @@ class PipelineTask: """ while True: try: - await self.queue_frame(HeartbeatFrame(timestamp=self._clock.get_time())) + # Don't use `queue_frame()` because if an EndFrame is queued the + # task will just stop waiting for the pipeline to finish not + # allowing more frames to be pushed. + await self._source.queue_frame(HeartbeatFrame(timestamp=self._clock.get_time())) await asyncio.sleep(HEARTBEAT_SECONDS) except asyncio.CancelledError: break From 477d0d154b0ef7a7e29918212c04f7877eca95b3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Jan 2025 10:05:23 -0800 Subject: [PATCH 17/26] frame_processor: make sure clock is initialized --- src/pipecat/processors/frame_processor.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 47863017e..ae830c52c 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -260,7 +260,7 @@ class FrameProcessor: async def __internal_push_frame(self, frame: Frame, direction: FrameDirection): try: - timestamp = self._clock.get_time() + timestamp = self._clock.get_time() if self._clock else 0 if direction == FrameDirection.DOWNSTREAM and self._next: logger.trace(f"Pushing {frame} from {self} to {self._next}") if self._observer: From 45cbad5b3e2238aae51fc859bd0a23d686de917a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Jan 2025 10:11:28 -0800 Subject: [PATCH 18/26] task: add HEARTBEAT_MONITOR_SECONDS --- src/pipecat/pipeline/task.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9a6d344ea..2fb13f627 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -31,6 +31,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.utils import obj_count, obj_id HEARTBEAT_SECONDS = 1.0 +HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5 class PipelineParams(BaseModel): @@ -353,7 +354,7 @@ class PipelineTask: takes for the heartbeat frame to traverse all the pipeline. """ - wait_time = HEARTBEAT_SECONDS * 5 + wait_time = HEARTBEAT_MONITOR_SECONDS while True: try: frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time) From 3e4020cdbaae3f3f8d8e7bd1efec65cbe5119342 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Jan 2025 11:13:33 -0800 Subject: [PATCH 19/26] task: add TaskObserver and avoid pipeline blocking Observers now process frames in separate tasks. This avoids blocking the pipeline while the observer is processing the frame. --- src/pipecat/pipeline/task.py | 28 ++------ src/pipecat/pipeline/task_observer.py | 97 +++++++++++++++++++++++++++ 2 files changed, 101 insertions(+), 24 deletions(-) create mode 100644 src/pipecat/pipeline/task_observer.py diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 2fb13f627..9263352e4 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -27,6 +27,7 @@ from pipecat.frames.frames import ( from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData from pipecat.observers.base_observer import BaseObserver from pipecat.pipeline.base_pipeline import BasePipeline +from pipecat.pipeline.task_observer import TaskObserver from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.utils import obj_count, obj_id @@ -84,29 +85,6 @@ class Sink(FrameProcessor): await self._down_queue.put(frame) -class Observer(BaseObserver): - """This is a pipeline frame observer that is used as a proxy to the user - provided observers. That is, this is the only observer passed to the frame - processors. Then, every time a frame is pushed this observer will call all - the observers registered to the pipeline task. - - """ - - def __init__(self, observers: List[BaseObserver] = []): - self._observers = observers - - async def on_push_frame( - self, - src: FrameProcessor, - dst: FrameProcessor, - frame: Frame, - direction: FrameDirection, - timestamp: int, - ): - for observer in self._observers: - await observer.on_push_frame(src, dst, frame, direction, timestamp) - - class PipelineTask: def __init__( self, @@ -141,7 +119,7 @@ class PipelineTask: self._sink = Sink(self._down_queue) pipeline.link(self._sink) - self._observer = Observer(params.observers) + self._observer = TaskObserver(params.observers) def has_finished(self): """Indicates whether the tasks has finished. That is, all processors @@ -222,6 +200,8 @@ class PipelineTask: self._process_down_task.cancel() await self._process_down_task + await self._observer.stop() + async def _maybe_cancel_heartbeat_tasks(self): if self._params.enable_heartbeats: self._heartbeat_push_task.cancel() diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py new file mode 100644 index 000000000..2fd13f517 --- /dev/null +++ b/src/pipecat/pipeline/task_observer.py @@ -0,0 +1,97 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +import asyncio +from typing import List + +from attr import dataclass + +from pipecat.frames.frames import Frame +from pipecat.observers.base_observer import BaseObserver +from pipecat.processors.frame_processor import FrameDirection, FrameProcessor + + +@dataclass +class Proxy: + """This is the data we receive from the main observer and that we put into + a queue for later processing. + + """ + + queue: asyncio.Queue + task: asyncio.Task + observer: BaseObserver + + +@dataclass +class ObserverData: + """This is the data we receive from the main observer and that we put into a + proxy queue for later processing. + + """ + + src: FrameProcessor + dst: FrameProcessor + frame: Frame + direction: FrameDirection + timestamp: int + + +class TaskObserver(BaseObserver): + """This is a pipeline frame observer that is meant to be used as a proxy to + the user provided observers. That is, this is the observer that should be + passed to the frame processors. Then, every time a frame is pushed this + observer will call all the observers registered to the pipeline task. + + This observer makes sure that passing frames to observers doesn't block the + pipeline by creating a queue and a task for each user observer. When a frame + is received, it will be put in a queue for efficiency and later processed by + each task. + + """ + + def __init__(self, observers: List[BaseObserver] = []): + self._proxies: List[Proxy] = self._create_proxies(observers) + + async def stop(self): + """Stops all proxy observer tasks.""" + for proxy in self._proxies: + proxy.task.cancel() + await proxy.task + + async def on_push_frame( + self, + src: FrameProcessor, + dst: FrameProcessor, + frame: Frame, + direction: FrameDirection, + timestamp: int, + ): + for proxy in self._proxies: + await proxy.queue.put( + ObserverData( + src=src, dst=dst, frame=frame, direction=direction, timestamp=timestamp + ) + ) + + def _create_proxies(self, observers) -> List[Proxy]: + proxies = [] + for observer in observers: + queue = asyncio.Queue() + task = asyncio.create_task(self._proxy_task_handler(queue, observer)) + proxy = Proxy(queue=queue, task=task, observer=observer) + proxies.append(proxy) + return proxies + + async def _proxy_task_handler(self, queue: asyncio.Queue, observer: BaseObserver): + while True: + try: + data = await queue.get() + await observer.on_push_frame( + data.src, data.dst, data.frame, data.direction, data.timestamp + ) + except asyncio.CancelledError: + break From 65fa77dfa5b484664a7b140439903c1223cf7573 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Jan 2025 15:24:41 -0800 Subject: [PATCH 20/26] audio: use resample_audio to resample ulaw bytes --- src/pipecat/audio/utils.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index 5706c2c5f..143f05c28 100644 --- a/src/pipecat/audio/utils.py +++ b/src/pipecat/audio/utils.py @@ -80,14 +80,14 @@ def ulaw_to_pcm(ulaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int): in_pcm_bytes = audioop.ulaw2lin(ulaw_bytes, 2) # Resample - out_pcm_bytes = audioop.ratecv(in_pcm_bytes, 2, 1, in_sample_rate, out_sample_rate, None)[0] + out_pcm_bytes = resample_audio(in_pcm_bytes, in_sample_rate, out_sample_rate) return out_pcm_bytes def pcm_to_ulaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int): # Resample - in_pcm_bytes = audioop.ratecv(pcm_bytes, 2, 1, in_sample_rate, out_sample_rate, None)[0] + in_pcm_bytes = resample_audio(pcm_bytes, in_sample_rate, out_sample_rate) # Convert PCM to μ-law ulaw_bytes = audioop.lin2ulaw(in_pcm_bytes, 2) From b81323d6765c997d33d17c30b8930894ba3f533b Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Fri, 17 Jan 2025 20:11:16 -0500 Subject: [PATCH 21/26] Code review fixes + docstrings --- .../07d-interruptible-elevenlabs-http.py | 104 ------------------ src/pipecat/services/elevenlabs.py | 39 +++---- 2 files changed, 17 insertions(+), 126 deletions(-) delete mode 100644 examples/foundational/07d-interruptible-elevenlabs-http.py diff --git a/examples/foundational/07d-interruptible-elevenlabs-http.py b/examples/foundational/07d-interruptible-elevenlabs-http.py deleted file mode 100644 index 4b8aefd98..000000000 --- a/examples/foundational/07d-interruptible-elevenlabs-http.py +++ /dev/null @@ -1,104 +0,0 @@ -# -# 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 runner import configure - -from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame -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.elevenlabs import ElevenLabsHttpTTSService -from pipecat.services.openai import OpenAILLMService -from pipecat.transports.services.daily import DailyParams, DailyTransport - -load_dotenv(override=True) - -logger.remove(0) -logger.add(sys.stderr, level="DEBUG") - - -async def main(): - async with aiohttp.ClientSession() as session: - (room_url, token) = await configure(session) - - transport = DailyTransport( - room_url, - token, - "Respond bot", - DailyParams( - audio_out_enabled=True, - transcription_enabled=True, - vad_enabled=True, - vad_analyzer=SileroVADAnalyzer(), - ), - ) - - tts = ElevenLabsHttpTTSService( - api_key=os.getenv("ELEVENLABS_API_KEY", ""), - voice_id=os.getenv("ELEVENLABS_VOICE_ID", ""), - # params=ElevenLabsHttpTTSService.InputParams(language="en"), - ) - - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") - - 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 - context_aggregator.user(), # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - context_aggregator.assistant(), # Assistant spoken responses - ] - ) - - task = PipelineTask( - pipeline, - PipelineParams( - allow_interruptions=True, - enable_metrics=True, - enable_usage_metrics=True, - report_only_initial_ttfb=True, - ), - ) - - @transport.event_handler("on_first_participant_joined") - async def on_first_participant_joined(transport, participant): - await transport.capture_participant_transcription(participant["id"]) - # 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_participant_left") - async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) - - runner = PipelineRunner() - - await runner.run(task) - - -if __name__ == "__main__": - asyncio.run(main()) diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 0ad542821..188c5a9c9 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -430,6 +430,7 @@ class ElevenLabsHttpTTSService(TTSService): Args: api_key: ElevenLabs API key voice_id: ID of the voice to use + aiohttp_session: aiohttp ClientSession model: Model ID (default: "eleven_flash_v2_5" for low latency) base_url: API base URL output_format: Audio output format (PCM) @@ -449,20 +450,20 @@ class ElevenLabsHttpTTSService(TTSService): *, api_key: str, voice_id: str, + aiohttp_session: aiohttp.ClientSession, model: str = "eleven_flash_v2_5", base_url: str = "https://api.elevenlabs.io", output_format: ElevenLabsOutputFormat = "pcm_24000", params: InputParams = InputParams(), **kwargs, ): - sample_rate = sample_rate_from_output_format(output_format) - super().__init__(sample_rate=sample_rate, **kwargs) + super().__init__(sample_rate=sample_rate_from_output_format(output_format), **kwargs) self._api_key = api_key self._base_url = base_url self._output_format = output_format self._params = params - self._session: Optional[aiohttp.ClientSession] = None + self._session = aiohttp_session self._settings = { "sample_rate": sample_rate_from_output_format(output_format), @@ -484,6 +485,11 @@ class ElevenLabsHttpTTSService(TTSService): return True def _set_voice_settings(self) -> Optional[Dict[str, Union[float, bool]]]: + """Configure voice settings if stability and similarity_boost are provided. + + Returns: + Dictionary of voice settings or None if required parameters are missing. + """ voice_settings: Dict[str, Union[float, bool]] = {} if ( self._settings["stability"] is not None @@ -507,27 +513,16 @@ class ElevenLabsHttpTTSService(TTSService): return voice_settings or None - async def start(self, frame: StartFrame): - await super().start(frame) - self._session = aiohttp.ClientSession() - - async def stop(self, frame: EndFrame): - await super().stop(frame) - if self._session: - await self._session.close() - self._session = None - - async def cancel(self, frame: CancelFrame): - await super().cancel(frame) - if self._session: - await self._session.close() - self._session = None - async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: - logger.debug(f"Generating TTS: [{text}]") + """Generate speech from text using ElevenLabs streaming API. - if not self._session: - self._session = aiohttp.ClientSession() + Args: + text: The text to convert to speech + + Yields: + Frames containing audio data and status information + """ + logger.debug(f"Generating TTS: [{text}]") url = f"{self._base_url}/v1/text-to-speech/{self._voice_id}/stream" From e049ae470df511b40c358c8c39d60fbc9a757153 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 15 Jan 2025 22:01:06 -0500 Subject: [PATCH 22/26] Register the on_error handler --- src/pipecat/transports/services/daily.py | 16 +++++++++++++--- 1 file changed, 13 insertions(+), 3 deletions(-) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index d3a56d5a7..7dd8fecaa 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -27,6 +27,7 @@ from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADParams from pipecat.frames.frames import ( CancelFrame, EndFrame, + ErrorFrame, Frame, InputAudioRawFrame, InterimTranscriptionFrame, @@ -917,6 +918,7 @@ class DailyTransport(BaseTransport): # these handlers. self._register_event_handler("on_joined") self._register_event_handler("on_left") + self._register_event_handler("on_error") self._register_event_handler("on_app_message") self._register_event_handler("on_call_state_updated") self._register_event_handler("on_dialin_connected") @@ -1031,9 +1033,17 @@ class DailyTransport(BaseTransport): await self._call_event_handler("on_left") async def _on_error(self, error): - # TODO(aleix): Report error to input/output transports. The one managing - # the client should report the error. - pass + await self._call_event_handler("on_error", error) + # Push error frame to notify the pipeline + error_frame = ErrorFrame(error) + + if self._input: + await self._input.push_error(error_frame) + elif self._output: + await self._output.push_error(error_frame) + else: + logger.error("Both input and output are None while trying to push error") + raise RuntimeError("No valid input or output channel to push error") async def _on_app_message(self, message: Any, sender: str): if self._input: From 8a87e92b2b42a2e076f45141948ebdecc3b63ba8 Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Sat, 18 Jan 2025 10:48:57 +0530 Subject: [PATCH 23/26] adding missing 11labs package --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index ee842a533..462d1c9b6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,7 +51,7 @@ cerebras = [ "openai~=1.59.6" ] deepseek = [ "openai~=1.59.6" ] daily = [ "daily-python~=0.14.2" ] deepgram = [ "deepgram-sdk~=3.8.0" ] -elevenlabs = [ "websockets~=13.1" ] +elevenlabs = [ "websockets~=13.1", "elevenlabs~=1.8.2" ] fal = [ "fal-client~=0.5.6" ] fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ] gladia = [ "websockets~=13.1" ] From a9c7dbbc0541ecf9ada68f86a71dd2c3b433809e Mon Sep 17 00:00:00 2001 From: Vaibhav159 Date: Sat, 18 Jan 2025 10:58:07 +0530 Subject: [PATCH 24/26] removing unused code --- pyproject.toml | 2 +- src/pipecat/services/elevenlabs.py | 2 -- 2 files changed, 1 insertion(+), 3 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 462d1c9b6..ee842a533 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -51,7 +51,7 @@ cerebras = [ "openai~=1.59.6" ] deepseek = [ "openai~=1.59.6" ] daily = [ "daily-python~=0.14.2" ] deepgram = [ "deepgram-sdk~=3.8.0" ] -elevenlabs = [ "websockets~=13.1", "elevenlabs~=1.8.2" ] +elevenlabs = [ "websockets~=13.1" ] fal = [ "fal-client~=0.5.6" ] fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ] gladia = [ "websockets~=13.1" ] diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 188c5a9c9..d80a7e9c4 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -35,8 +35,6 @@ from pipecat.transcriptions.language import Language # See .env.example for ElevenLabs configuration needed try: import websockets - from elevenlabs import VoiceSettings - from elevenlabs.client import ElevenLabs except ModuleNotFoundError as e: logger.error(f"Exception: {e}") logger.error( From e0011a3996262b7ae95223945d0b24e00ab75b78 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sat, 18 Jan 2025 14:27:21 -0800 Subject: [PATCH 25/26] services(fish): FishAudioTTSService to use WebsocketService --- pyproject.toml | 5 ++--- src/pipecat/services/fish.py | 33 +++++---------------------------- 2 files changed, 7 insertions(+), 31 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ee842a533..38bf6d902 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -32,8 +32,7 @@ dependencies = [ "protobuf~=5.29.3", "pydantic~=2.10.5", "pyloudnorm~=0.1.1", - "resampy~=0.4.3", - "tenacity~=9.0.0" + "resampy~=0.4.3" ] [project.urls] @@ -63,7 +62,7 @@ fireworks = [ "openai~=1.59.6" ] krisp = [ "pipecat-ai-krisp~=0.3.0" ] koala = [ "pvkoala~=2.0.3" ] langchain = [ "langchain~=0.3.14", "langchain-community~=0.3.14", "langchain-openai~=0.3.0" ] -livekit = [ "livekit~=0.19.1", "livekit-api~=0.8.1" ] +livekit = [ "livekit~=0.19.1", "livekit-api~=0.8.1", "tenacity~=9.0.0" ] lmnt = [ "websockets~=13.1" ] local = [ "pyaudio~=0.2.14" ] moondream = [ "einops~=0.8.0", "timm~=1.0.13", "transformers~=4.48.0" ] diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index 245497b1d..710ab04ec 100644 --- a/src/pipecat/services/fish.py +++ b/src/pipecat/services/fish.py @@ -4,13 +4,11 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import asyncio import uuid from typing import AsyncGenerator, Literal, Optional from loguru import logger from pydantic import BaseModel -from tenacity import AsyncRetrying, RetryCallState, stop_after_attempt, wait_exponential from pipecat.frames.frames import ( BotStoppedSpeakingFrame, @@ -28,6 +26,7 @@ from pipecat.frames.frames import ( ) from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import TTSService +from pipecat.services.websocket_service import WebsocketService from pipecat.transcriptions.language import Language try: @@ -44,7 +43,7 @@ except ModuleNotFoundError as e: FishAudioOutputFormat = Literal["opus", "mp3", "pcm", "wav"] -class FishAudioTTSService(TTSService): +class FishAudioTTSService(TTSService, WebsocketService): class InputParams(BaseModel): language: Optional[Language] = Language.EN latency: Optional[str] = "normal" # "normal" or "balanced" @@ -105,7 +104,9 @@ class FishAudioTTSService(TTSService): async def _connect(self): await self._connect_websocket() - self._receive_task = self.get_event_loop().create_task(self._receive_task_handler()) + self._receive_task = self.get_event_loop().create_task( + self._receive_task_handler(self.push_error) + ) async def _disconnect(self): await self._disconnect_websocket() @@ -169,30 +170,6 @@ class FishAudioTTSService(TTSService): except Exception as e: logger.error(f"Error processing message: {e}") - async def _reconnect_websocket(self, retry_state: RetryCallState): - logger.warning(f"Fish Audio reconnecting (attempt: {retry_state.attempt_number})") - await self._disconnect_websocket() - await self._connect_websocket() - - async def _receive_task_handler(self): - while True: - try: - async for attempt in AsyncRetrying( - stop=stop_after_attempt(3), - wait=wait_exponential(multiplier=1, min=4, max=10), - before_sleep=self._reconnect_websocket, - reraise=True, - ): - with attempt: - await self._receive_messages() - except asyncio.CancelledError: - break - except Exception as e: - message = f"Fish Audio error receiving messages: {e}" - logger.error(message) - await self.push_error(ErrorFrame(message, fatal=True)) - break - async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) From c6d643d4ec1f0668027f685422076a6fd54f543d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 17 Jan 2025 20:40:11 -0800 Subject: [PATCH 26/26] update CHANGELOG for 0.0.53 --- CHANGELOG.md | 10 +++++++--- README.md | 6 ++++++ 2 files changed, 13 insertions(+), 3 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c5f5d0c8c..37b927ecd 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.53] - 2025-01-18 ### Added @@ -28,6 +28,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `examples/foundational/31-heartbeats.py` shows how to enable heartbeats and forces warnings to be displayed. +- Added `LLMTextFrame` and `TTSTextFrame` which should be pushed by LLM and TTS + services respectively instead of `TextFrame`s. + - Added `OpenRouter` for OpenRouter integration with an OpenAI-compatible interface. Added foundational example `14m-function-calling-openrouter.py`. @@ -92,10 +95,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `UserStoppedSpeakingFrame`. This helps in faster transcriptions and clearing the `Deepgram` audio buffer. -- Changed `DeepgramSTTService` to generate metrics using pipeline VAD. - ### Fixed +- Fixed an issue where `DeepgramSTTService` was not generating metrics using + pipeline's VAD. + - Fixed `UserIdleProcessor` not properly propagating `EndFrame`s through the pipeline. diff --git a/README.md b/README.md index 2c16f24c5..52c6f831b 100644 --- a/README.md +++ b/README.md @@ -53,6 +53,12 @@ To keep things lightweight, only the core framework is included by default. If y pip install "pipecat-ai[option,...]" ``` +Or you can install all of them with: + +```shell +pip install "pipecat-ai[all]" +``` + Available options include: | Category | Services | Install Command Example |