Merge branch 'pipecat-ai:main' into main

CHANGELOG.md

@@ -5,17 +5,31 @@ 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
 
+- 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
-  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.
 
-- Added `30-observer.py` to show how to add an Observer to a pipeline for
+- Introduced heartbeat frames. The pipeline task can now push periodic
-  debugging.
+  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. The example
+  `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`.
@@ -56,6 +70,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 +97,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue where `DeepgramSTTService` was not generating metrics using
+  pipeline's VAD.
+
+- 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.
 
@@ -95,6 +119,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

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                 |

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())

pyproject.toml

@@ -32,8 +32,7 @@ dependencies = [
     "protobuf~=5.29.3",
     "pydantic~=2.10.5",
     "pyloudnorm~=0.1.1",
-    "resampy~=0.4.3",
+    "resampy~=0.4.3"
-    "tenacity~=9.0.0"
 ]
 
 [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" ]

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)

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
 
@@ -375,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
 #
@@ -424,6 +449,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

src/pipecat/pipeline/task.py

@@ -19,6 +19,7 @@ from pipecat.frames.frames import (
     EndTaskFrame,
     ErrorFrame,
     Frame,
+    HeartbeatFrame,
     MetricsFrame,
     StartFrame,
     StopTaskFrame,
@@ -26,14 +27,19 @@ 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
 
+HEARTBEAT_SECONDS = 1.0
+HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5
+
 
 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 +64,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,33 +82,7 @@ class Sink(FrameProcessor):
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
-
+        await self._down_queue.put(frame)
-        # We really just want to know when the EndFrame reached the sink.
-        if isinstance(frame, EndFrame):
-            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:
@@ -135,9 +100,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)
@@ -145,36 +119,52 @@ 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
+        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()
+        await self._cancel_tasks(True)
-        self._process_up_task.cancel()
-        await self._process_push_task
-        await self._process_up_task
 
     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())
+        Starts running the given pipeline.
-        await asyncio.gather(self._process_up_task, self._process_push_task)
+        """
+        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 +172,43 @@ 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]
+
+        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())
+
+    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
+
+        self._process_up_task.cancel()
+        await self._process_up_task
+
+        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()
+            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,9 +217,20 @@ 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()
 
+        self._maybe_start_heartbeat_tasks()
+
         start_frame = StartFrame(
             allow_interruptions=self._params.allow_interruptions,
             enable_metrics=self._params.enable_metrics,
@@ -224,29 +262,91 @@ class PipelineTask:
             await self._source.cleanup()
             await self._pipeline.cleanup()
             await self._sink.cleanup()
-        # We just enqueue None to terminate the task gracefully.
+        # Finally, cancel internal tasks. We don't cancel the push tasks because
-        self._process_up_task.cancel()
+        # that's us.
-        await self._process_up_task
+        await self._cancel_tasks(False)
-
-    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()
 
     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:
+                # 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
+
+    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_MONITOR_SECONDS
+        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

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

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:

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

src/pipecat/processors/user_idle_processor.py

@@ -19,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
+    """Monitors user inactivity and triggers callbacks after timeout periods.
-    within a given timeout. If the timeout is reached before any interaction
-    occurred the provided callback will be called.
 
+    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__(
@@ -36,39 +50,72 @@ class UserIdleProcessor(FrameProcessor):
         self._callback = callback
         self._timeout = timeout
         self._interrupted = False
-        self._create_idle_task()
+        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()
+        """Stops and cleans up the idle monitoring task."""
-        await self._idle_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, (EndFrame, CancelFrame)):
-            await self._stop()
+            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
+        # Start monitoring on first conversation activity
-        if isinstance(frame, UserStartedSpeakingFrame):
+        if not self._conversation_started and isinstance(
-            self._interrupted = True
+            frame, (UserStartedSpeakingFrame, BotSpeakingFrame)
-            self._idle_event.set()
+        ):
-        elif isinstance(frame, UserStoppedSpeakingFrame):
+            self._conversation_started = True
-            self._interrupted = False
+            self._create_idle_task()
-            self._idle_event.set()
+
-        elif isinstance(frame, BotSpeakingFrame):
+        # Only process these events if conversation has started
-            self._idle_event.set()
+        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()
+        """Cleans up resources when processor is shutting down."""
-
+        if self._idle_task:  # Only stop if task exists
-    def _create_idle_task(self):
+            await self._stop()
-        self._idle_event = asyncio.Event()
-        self._idle_task = self.get_event_loop().create_task(self._idle_task_handler())
 
     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)

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":
+        if message["event"] == "media":
-            return None
-        else:
             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

src/pipecat/services/deepgram.py

@@ -20,6 +20,7 @@ from pipecat.frames.frames import (
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
+    UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
@@ -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"Triggered finalize event on: {frame.name=}, {direction=}")

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
 
@@ -16,6 +17,7 @@ from pipecat.frames.frames import (
     BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
+    ErrorFrame,
     Frame,
     LLMFullResponseEndFrame,
     StartFrame,
@@ -26,7 +28,7 @@ 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
 
@@ -418,3 +420,160 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
             yield None
         except Exception as e:
             logger.error(f"{self} exception: {e}")
+
+
+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
+        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)
+        params: Additional parameters for voice configuration
+    """
+
+    class InputParams(BaseModel):
+        language: Optional[Language] = Language.EN
+        optimize_streaming_latency: Optional[int] = 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,
+        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,
+    ):
+        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 = aiohttp_session
+
+        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]]]:
+        """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
+            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 run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
+        """Generate speech from text using ElevenLabs streaming API.
+
+        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"
+
+        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:
+            await self.start_ttfb_metrics()
+
+            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
+
+                await self.start_tts_usage_metrics(text)
+                yield TTSStartedFrame()
+
+                async for chunk in response.content:
+                    if chunk:
+                        await self.stop_ttfb_metrics()
+                        yield TTSAudioRawFrame(chunk, self._settings["sample_rate"], 1)
+
+                yield TTSStoppedFrame()
+
+        except Exception as e:
+            logger.error(f"Error in run_tts: {e}")
+            yield ErrorFrame(error=str(e))
+
+        finally:
+            yield TTSStoppedFrame()

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)
 

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,
@@ -921,6 +922,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")
@@ -1035,9 +1037,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
+        await self._call_event_handler("on_error", error)
-        # the client should report the error.
+        # Push error frame to notify the pipeline
-        pass
+        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: