wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'main' into snova-jorgep/sambanova-integration

Browse Source
This commit is contained in:
Jorge Piedrahita Ortiz
2025-06-24 16:00:28 -05:00
committed by GitHub
parent dfb0da32a9 f450da57e5
commit 851cf079c3
49 changed files with 3405 additions and 241 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
src/pipecat/audio/turn/base_turn_analyzer.py
View File

@@ -78,3 +78,8 @@ class BaseTurnAnalyzer(ABC):
EndOfTurnState: The result of the end of turn analysis.
"""
pass
@abstractmethod
def clear(self):
"""Reset the turn analyzer to its initial state."""
pass

3
src/pipecat/audio/turn/smart_turn/base_smart_turn.py
View File

@@ -98,6 +98,9 @@ class BaseSmartTurn(BaseTurnAnalyzer):
logger.debug(f"End of Turn result: {state}")
return state, result
def clear(self):
self._clear(EndOfTurnState.COMPLETE)
def _clear(self, turn_state: EndOfTurnState):
# If the state is still incomplete, keep the _speech_triggered as True
self._speech_triggered = turn_state == EndOfTurnState.INCOMPLETE

54
src/pipecat/frames/frames.py
View File

@@ -7,6 +7,7 @@
from dataclasses import dataclass, field
from enum import Enum
from typing import (
TYPE_CHECKING,
Any,
Awaitable,
Callable,
@@ -26,6 +27,9 @@ from pipecat.transcriptions.language import Language
from pipecat.utils.time import nanoseconds_to_str
from pipecat.utils.utils import obj_count, obj_id
if TYPE_CHECKING:
from pipecat.processors.frame_processor import FrameProcessor
class KeypadEntry(str, Enum):
"""DTMF entries."""
@@ -485,16 +489,6 @@ 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
@@ -529,25 +523,25 @@ class StopTaskFrame(SystemFrame):
@dataclass
class FrameProcessorPauseUrgentFrame(SystemFrame):
"""This processor is used to pause frame processing for the given processor
as fast as possible. Pausing frame processing will keep frames in the
internal queue which will then be processed when frame processing is resumed
with `FrameProcessorResumeFrame`.
"""This frame is used to pause frame processing for the given processor as
fast as possible. Pausing frame processing will keep frames in the internal
queue which will then be processed when frame processing is resumed with
`FrameProcessorResumeFrame`.
"""
processor: str
processor: "FrameProcessor"
@dataclass
class FrameProcessorResumeUrgentFrame(SystemFrame):
"""This processor is used to resume frame processing for the given processor
"""This frame is used to resume frame processing for the given processor
if it was previously paused as fast as possible. After resuming frame
processing all queued frames will be processed in the order received.
"""
processor: str
processor: "FrameProcessor"
@dataclass
@@ -877,25 +871,37 @@ class StopFrame(ControlFrame):
pass
@dataclass
class HeartbeatFrame(ControlFrame):
"""This frame is used by the pipeline task as a mechanism to know if the
pipeline is running properly.
"""
timestamp: int
@dataclass
class FrameProcessorPauseFrame(ControlFrame):
"""This processor is used to pause frame processing for the given
"""This frame is used to pause frame processing for the given
processor. Pausing frame processing will keep frames in the internal queue
which will then be processed when frame processing is resumed with
`FrameProcessorResumeFrame`."""
`FrameProcessorResumeFrame`.
processor: str
"""
processor: "FrameProcessor"
@dataclass
class FrameProcessorResumeFrame(ControlFrame):
"""This processor is used to resume frame processing for the given processor
if it was previously paused. After resuming frame processing all queued
frames will be processed in the order received.
"""This frame is used to resume frame processing for the given processor if
it was previously paused. After resuming frame processing all queued frames
will be processed in the order received.
"""
processor: str
processor: "FrameProcessor"
@dataclass

15
src/pipecat/pipeline/base_task.py
View File

@@ -6,18 +6,21 @@
import asyncio
from abc import abstractmethod
from dataclasses import dataclass
from typing import AsyncIterable, Iterable
from pipecat.frames.frames import Frame
from pipecat.utils.base_object import BaseObject
class BaseTask(BaseObject):
@abstractmethod
def set_event_loop(self, loop: asyncio.AbstractEventLoop):
"""Sets the event loop that this task will run on."""
pass
@dataclass
class PipelineTaskParams:
"""Specific configuration for the pipeline task."""
loop: asyncio.AbstractEventLoop
class BasePipelineTask(BaseObject):
@abstractmethod
def has_finished(self) -> bool:
"""Indicates whether the tasks has finished. That is, all processors
@@ -40,7 +43,7 @@ class BaseTask(BaseObject):
pass
@abstractmethod
async def run(self):
async def run(self, params: PipelineTaskParams):
"""Starts running the given pipeline."""
pass

6
src/pipecat/pipeline/parallel_pipeline.py
View File

@@ -202,14 +202,18 @@ class ParallelPipeline(BasePipeline):
async def _process_up_queue(self):
while True:
frame = await self._up_queue.get()
self.start_watchdog()
await self._parallel_push_frame(frame, FrameDirection.UPSTREAM)
self._up_queue.task_done()
self.reset_watchdog()
async def _process_down_queue(self):
running = True
while running:
frame = await self._down_queue.get()
self.start_watchdog()
endframe_counter = self._endframe_counter.get(frame.id, 0)
# If we have a counter, decrement it.
@@ -224,3 +228,5 @@ class ParallelPipeline(BasePipeline):
running = not (endframe_counter == 0 and isinstance(frame, EndFrame))
self._down_queue.task_done()
self.reset_watchdog()

5
src/pipecat/pipeline/runner.py
View File

@@ -11,6 +11,7 @@ from typing import Optional
from loguru import logger
from pipecat.pipeline.base_task import PipelineTaskParams
from pipecat.pipeline.task import PipelineTask
from pipecat.utils.base_object import BaseObject
@@ -37,8 +38,8 @@ class PipelineRunner(BaseObject):
async def run(self, task: PipelineTask):
logger.debug(f"Runner {self} started running {task}")
self._tasks[task.name] = task
task.set_event_loop(self._loop)
await task.run()
params = PipelineTaskParams(loop=self._loop)
await task.run(params)
del self._tasks[task.name]
# Cleanup base object.

119
src/pipecat/pipeline/task.py
View File

@@ -6,7 +6,8 @@
import asyncio
import time
from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Sequence, Tuple, Type
from collections import deque
from typing import Any, AsyncIterable, Deque, Dict, Iterable, List, Optional, Tuple, Type
from loguru import logger
from pydantic import BaseModel, ConfigDict, Field
@@ -23,6 +24,7 @@ from pipecat.frames.frames import (
ErrorFrame,
Frame,
HeartbeatFrame,
InputAudioRawFrame,
LLMFullResponseEndFrame,
MetricsFrame,
StartFrame,
@@ -33,19 +35,22 @@ from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData
from pipecat.observers.base_observer import BaseObserver
from pipecat.observers.turn_tracking_observer import TurnTrackingObserver
from pipecat.pipeline.base_pipeline import BasePipeline
from pipecat.pipeline.base_task import BaseTask
from pipecat.pipeline.base_task import BasePipelineTask, PipelineTaskParams
from pipecat.pipeline.task_observer import TaskObserver
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.utils.asyncio import BaseTaskManager, TaskManager
from pipecat.utils.asyncio import WATCHDOG_TIMEOUT, BaseTaskManager, TaskManager, TaskManagerParams
from pipecat.utils.tracing.setup import is_tracing_available
from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver
HEARTBEAT_SECONDS = 1.0
HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5
HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 10
class PipelineParams(BaseModel):
"""Configuration parameters for pipeline execution.
"""Configuration parameters for pipeline execution. These parameters are
usually passed to all frame processors using through `StartFrame`. For other
generic pipeline task parameters use `PipelineTask` constructor arguments
instead.
Attributes:
allow_interruptions: Whether to allow pipeline interruptions.
@@ -60,6 +65,7 @@ class PipelineParams(BaseModel):
send_initial_empty_metrics: Whether to send initial empty metrics.
start_metadata: Additional metadata for pipeline start.
interruption_strategies: Strategies for bot interruption behavior.
"""
model_config = ConfigDict(arbitrary_types_allowed=True)
@@ -71,11 +77,11 @@ class PipelineParams(BaseModel):
enable_metrics: bool = False
enable_usage_metrics: bool = False
heartbeats_period_secs: float = HEARTBEAT_SECONDS
interruption_strategies: List[BaseInterruptionStrategy] = Field(default_factory=list)
observers: List[BaseObserver] = Field(default_factory=list)
report_only_initial_ttfb: bool = False
send_initial_empty_metrics: bool = True
start_metadata: Dict[str, Any] = Field(default_factory=dict)
interruption_strategies: List[BaseInterruptionStrategy] = Field(default_factory=list)
class PipelineTaskSource(FrameProcessor):
@@ -125,7 +131,7 @@ class PipelineTaskSink(FrameProcessor):
await self._down_queue.put(frame)
class PipelineTask(BaseTask):
class PipelineTask(BasePipelineTask):
"""Manages the execution of a pipeline, handling frame processing and task lifecycle.
It has a couple of event handlers `on_frame_reached_upstream` and
@@ -172,21 +178,24 @@ class PipelineTask(BaseTask):
Args:
pipeline: The pipeline to execute.
params: Configuration parameters for the pipeline.
observers: List of observers for monitoring pipeline execution.
clock: Clock implementation for timing operations.
additional_span_attributes: Optional dictionary of attributes to propagate as
OpenTelemetry conversation span attributes.
cancel_on_idle_timeout: Whether the pipeline task should be cancelled if
the idle timeout is reached.
check_dangling_tasks: Whether to check for processors' tasks finishing properly.
clock: Clock implementation for timing operations.
conversation_id: Optional custom ID for the conversation.
enable_tracing: Whether to enable tracing.
enable_turn_tracking: Whether to enable turn tracking.
enable_watchdog_logging: Whether to print task processing times.
idle_timeout_frames: A tuple with the frames that should trigger an idle
timeout if not received withing `idle_timeout_seconds`.
idle_timeout_secs: Timeout (in seconds) to consider pipeline idle or
None. If a pipeline is idle the pipeline task will be cancelled
automatically.
idle_timeout_frames: A tuple with the frames that should trigger an idle
timeout if not received withing `idle_timeout_seconds`.
cancel_on_idle_timeout: Whether the pipeline task should be cancelled if
the idle timeout is reached.
enable_turn_tracking: Whether to enable turn tracking.
enable_turn_tracing: Whether to enable turn tracing.
conversation_id: Optional custom ID for the conversation.
additional_span_attributes: Optional dictionary of attributes to propagate as
OpenTelemetry conversation span attributes.
observers: List of observers for monitoring pipeline execution.
watchdog_timeout_secs: Watchdog timer timeout (in seconds). A warning
will be logged if the watchdog timer is not reset before this timeout.
"""
def __init__(
@@ -194,33 +203,37 @@ class PipelineTask(BaseTask):
pipeline: BasePipeline,
*,
params: Optional[PipelineParams] = None,
observers: Optional[List[BaseObserver]] = None,
clock: Optional[BaseClock] = None,
task_manager: Optional[BaseTaskManager] = None,
additional_span_attributes: Optional[dict] = None,
cancel_on_idle_timeout: bool = True,
check_dangling_tasks: bool = True,
idle_timeout_secs: Optional[float] = 300,
clock: Optional[BaseClock] = None,
conversation_id: Optional[str] = None,
enable_tracing: bool = False,
enable_turn_tracking: bool = True,
enable_watchdog_logging: bool = False,
idle_timeout_frames: Tuple[Type[Frame], ...] = (
BotSpeakingFrame,
LLMFullResponseEndFrame,
),
cancel_on_idle_timeout: bool = True,
enable_turn_tracking: bool = True,
enable_tracing: bool = False,
conversation_id: Optional[str] = None,
additional_span_attributes: Optional[dict] = None,
idle_timeout_secs: Optional[float] = 300,
observers: Optional[List[BaseObserver]] = None,
task_manager: Optional[BaseTaskManager] = None,
watchdog_timeout_secs: float = WATCHDOG_TIMEOUT,
):
super().__init__()
self._pipeline = pipeline
self._clock = clock or SystemClock()
self._params = params or PipelineParams()
self._check_dangling_tasks = check_dangling_tasks
self._idle_timeout_secs = idle_timeout_secs
self._idle_timeout_frames = idle_timeout_frames
self._cancel_on_idle_timeout = cancel_on_idle_timeout
self._enable_turn_tracking = enable_turn_tracking
self._enable_tracing = enable_tracing and is_tracing_available()
self._conversation_id = conversation_id
self._additional_span_attributes = additional_span_attributes or {}
self._cancel_on_idle_timeout = cancel_on_idle_timeout
self._check_dangling_tasks = check_dangling_tasks
self._clock = clock or SystemClock()
self._conversation_id = conversation_id
self._enable_tracing = enable_tracing and is_tracing_available()
self._enable_turn_tracking = enable_turn_tracking
self._enable_watchdog_logging = enable_watchdog_logging
self._idle_timeout_frames = idle_timeout_frames
self._idle_timeout_secs = idle_timeout_secs
self._watchdog_timeout_secs = watchdog_timeout_secs
if self._params.observers:
import warnings
@@ -322,9 +335,6 @@ class PipelineTask(BaseTask):
async def remove_observer(self, observer: BaseObserver):
await self._observer.remove_observer(observer)
def set_event_loop(self, loop: asyncio.AbstractEventLoop):
self._task_manager.set_event_loop(loop)
def set_reached_upstream_filter(self, types: Tuple[Type[Frame], ...]):
"""Sets which frames will be checked before calling the
on_frame_reached_upstream event handler.
@@ -358,14 +368,14 @@ class PipelineTask(BaseTask):
"""Stops the running pipeline immediately."""
await self._cancel()
async def run(self):
async def run(self, params: PipelineTaskParams):
"""Starts and manages the pipeline execution until completion or cancellation."""
if self.has_finished():
return
cleanup_pipeline = True
try:
# Setup processors.
await self._setup()
await self._setup(params)
# Create all main tasks and wait of the main push task. This is the
# task that pushes frames to the very beginning of our pipeline (our
@@ -485,7 +495,14 @@ class PipelineTask(BaseTask):
await self._pipeline_end_event.wait()
self._pipeline_end_event.clear()
async def _setup(self):
async def _setup(self, params: PipelineTaskParams):
mgr_params = TaskManagerParams(
loop=params.loop,
enable_watchdog_logging=self._enable_watchdog_logging,
watchdog_timeout=self._watchdog_timeout_secs,
)
self._task_manager.setup(mgr_params)
setup = FrameProcessorSetup(
clock=self._clock,
task_manager=self._task_manager,
@@ -509,6 +526,8 @@ class PipelineTask(BaseTask):
await self._pipeline.cleanup()
await self._sink.cleanup()
await self._task_manager.cleanup()
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
@@ -646,12 +665,17 @@ class PipelineTask(BaseTask):
"""
running = True
last_frame_time = 0
frame_buffer = deque(maxlen=10) # Store last 10 frames
while running:
try:
frame = await asyncio.wait_for(
self._idle_queue.get(), timeout=self._idle_timeout_secs
)
if not isinstance(frame, InputAudioRawFrame):
frame_buffer.append(frame)
if isinstance(frame, StartFrame) or isinstance(frame, self._idle_timeout_frames):
# If we find a StartFrame or one of the frames that prevents a
# time out we update the time.
@@ -662,7 +686,7 @@ class PipelineTask(BaseTask):
# valid frames.
diff_time = time.time() - last_frame_time
if diff_time >= self._idle_timeout_secs:
running = await self._idle_timeout_detected()
running = await self._idle_timeout_detected(frame_buffer)
# Reset `last_frame_time` so we don't trigger another
# immediate idle timeout if we are not cancelling. For
# example, we might want to force the bot to say goodbye
@@ -670,15 +694,20 @@ class PipelineTask(BaseTask):
last_frame_time = time.time()
self._idle_queue.task_done()
except asyncio.TimeoutError:
running = await self._idle_timeout_detected()
async def _idle_timeout_detected(self) -> bool:
except asyncio.TimeoutError:
running = await self._idle_timeout_detected(frame_buffer)
async def _idle_timeout_detected(self, last_frames: Deque[Frame]) -> bool:
"""Logic for when the pipeline is idle.
Returns:
bool: Whther the pipeline task is being cancelled or not.
"""
logger.warning("Idle timeout detected. Last 10 frames received:")
for i, frame in enumerate(last_frames, 1):
logger.warning(f"Frame {i}: {frame}")
await self._call_event_handler("on_idle_timeout")
if self._cancel_on_idle_timeout:
logger.warning(f"Idle pipeline detected, cancelling pipeline task...")

19
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -266,6 +266,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
self._user_speaking = False
self._bot_speaking = False
self._was_bot_speaking = False
self._emulating_vad = False
self._seen_interim_results = False
self._waiting_for_aggregation = False
@@ -275,6 +276,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
async def reset(self):
await super().reset()
self._was_bot_speaking = False
self._seen_interim_results = False
self._waiting_for_aggregation = False
[await s.reset() for s in self._interruption_strategies]
@@ -355,6 +357,20 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
else:
# No interruption config - normal behavior (always push aggregation)
await self._process_aggregation()
# Handles the case where both the user and the bot are not speaking,
# and the bot was previously speaking before the user interruption.
# Normally, when the user stops speaking, new text is expected,
# which triggers the bot to respond. However, if no new text
# is received, this safeguard ensures
# the bot doesn't hang indefinitely while waiting to speak again.
elif not self._seen_interim_results and self._was_bot_speaking and not self._bot_speaking:
logger.warning(
"User stopped speaking but no new aggregation received. Forcing aggregation processing to resume bot response."
)
# Resetting it so we don't trigger this twice
self._was_bot_speaking = False
# We are just pushing the same previous context to be processed again in this case
await self.push_frame(OpenAILLMContextFrame(self._context))
async def _should_interrupt_based_on_strategies(self) -> bool:
"""Check if interruption should occur based on configured strategies."""
@@ -381,6 +397,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
self._user_speaking = True
self._waiting_for_aggregation = True
self._was_bot_speaking = self._bot_speaking
# If we get a non-emulated UserStartedSpeakingFrame but we are in the
# middle of emulating VAD, let's stop emulating VAD (i.e. don't send the
@@ -393,7 +410,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
# We just stopped speaking. Let's see if there's some aggregation to
# push. If the last thing we saw is an interim transcription, let's wait
# pushing the aggregation as we will probably get a final transcription.
if not self._seen_interim_results:
if len(self._aggregation) > 0 and not self._seen_interim_results:
await self.push_aggregation()
async def _handle_bot_started_speaking(self, _: BotStartedSpeakingFrame):

2
src/pipecat/processors/consumer_processor.py
View File

@@ -61,5 +61,7 @@ class ConsumerProcessor(FrameProcessor):
async def _consumer_task_handler(self):
while True:
frame = await self._queue.get()
self.start_watchdog()
new_frame = await self._transformer(frame)
await self.push_frame(new_frame, self._direction)
self.reset_watchdog()

82
src/pipecat/processors/frame_processor.py
View File

@@ -51,6 +51,8 @@ class FrameProcessor(BaseObject):
*,
name: Optional[str] = None,
metrics: Optional[FrameProcessorMetrics] = None,
enable_watchdog_logging: Optional[bool] = None,
watchdog_timeout_secs: Optional[float] = None,
**kwargs,
):
super().__init__(name=name)
@@ -58,6 +60,12 @@ class FrameProcessor(BaseObject):
self._prev: Optional["FrameProcessor"] = None
self._next: Optional["FrameProcessor"] = None
# Enable watchdog logging for all tasks created by this frame processor.
self._enable_watchdog_logging = enable_watchdog_logging
# Allow this frame processor to control their tasks timeout.
self._watchdog_timeout = watchdog_timeout_secs
# Clock
self._clock: Optional[BaseClock] = None
@@ -171,24 +179,42 @@ class FrameProcessor(BaseObject):
await self.stop_ttfb_metrics()
await self.stop_processing_metrics()
def create_task(self, coroutine: Coroutine, name: Optional[str] = None) -> asyncio.Task:
if not self._task_manager:
raise Exception(f"{self} TaskManager is still not initialized.")
def create_task(
self,
coroutine: Coroutine,
name: Optional[str] = None,
*,
enable_watchdog_logging: Optional[bool] = None,
watchdog_timeout_secs: Optional[float] = None,
) -> asyncio.Task:
if name:
name = f"{self}::{name}"
else:
name = f"{self}::{coroutine.cr_code.co_name}"
return self._task_manager.create_task(coroutine, name)
return self.get_task_manager().create_task(
coroutine,
name,
enable_watchdog_logging=(
enable_watchdog_logging
if enable_watchdog_logging
else self._enable_watchdog_logging
),
watchdog_timeout=(
watchdog_timeout_secs if watchdog_timeout_secs else self._watchdog_timeout
),
)
async def cancel_task(self, task: asyncio.Task, timeout: Optional[float] = None):
if not self._task_manager:
raise Exception(f"{self} TaskManager is still not initialized.")
await self._task_manager.cancel_task(task, timeout)
await self.get_task_manager().cancel_task(task, timeout)
async def wait_for_task(self, task: asyncio.Task, timeout: Optional[float] = None):
if not self._task_manager:
raise Exception(f"{self} TaskManager is still not initialized.")
await self._task_manager.wait_for_task(task, timeout)
await self.get_task_manager().wait_for_task(task, timeout)
def start_watchdog(self):
self.get_task_manager().start_watchdog(asyncio.current_task())
def reset_watchdog(self):
self.get_task_manager().reset_watchdog(asyncio.current_task())
async def setup(self, setup: FrameProcessorSetup):
self._clock = setup.clock
@@ -206,9 +232,7 @@ class FrameProcessor(BaseObject):
logger.debug(f"Linking {self} -> {self._next}")
def get_event_loop(self) -> asyncio.AbstractEventLoop:
if not self._task_manager:
raise Exception(f"{self} TaskManager is still not initialized.")
return self._task_manager.get_event_loop()
return self.get_task_manager().get_event_loop()
def set_parent(self, parent: "FrameProcessor"):
self._parent = parent
@@ -296,11 +320,11 @@ class FrameProcessor(BaseObject):
await self.__cancel_push_task()
async def __pause(self, frame: FrameProcessorPauseFrame | FrameProcessorPauseUrgentFrame):
if frame.name == self.name:
if frame.processor.name == self.name:
await self.pause_processing_frames()
async def __resume(self, frame: FrameProcessorResumeFrame | FrameProcessorResumeUrgentFrame):
if frame.name == self.name:
if frame.processor.name == self.name:
await self.resume_processing_frames()
#
@@ -315,9 +339,8 @@ class FrameProcessor(BaseObject):
# Cancel the input task. This will stop processing queued frames.
await self.__cancel_input_task()
except Exception as e:
logger.exception(f"Uncaught exception in {self}: {e}")
logger.exception(f"Uncaught exception in {self} when handling _start_interruption: {e}")
await self.push_error(ErrorFrame(str(e)))
raise
# Create a new input queue and task.
self.__create_input_task()
@@ -360,7 +383,6 @@ class FrameProcessor(BaseObject):
except Exception as e:
logger.exception(f"Uncaught exception in {self}: {e}")
await self.push_error(ErrorFrame(str(e)))
raise
def _check_started(self, frame: Frame):
if not self.__started:
@@ -389,15 +411,19 @@ class FrameProcessor(BaseObject):
logger.trace(f"{self}: frame processing resumed")
(frame, direction, callback) = await self.__input_queue.get()
# Process the frame.
await self.process_frame(frame, direction)
# If this frame has an associated callback, call it now.
if callback:
await callback(self, frame, direction)
self.__input_queue.task_done()
try:
self.start_watchdog()
# Process the frame.
await self.process_frame(frame, direction)
# If this frame has an associated callback, call it now.
if callback:
await callback(self, frame, direction)
except Exception as e:
logger.exception(f"{self}: error processing frame: {e}")
await self.push_error(ErrorFrame(str(e)))
finally:
self.__input_queue.task_done()
self.reset_watchdog()
def __create_push_task(self):
if not self.__push_frame_task:
@@ -412,5 +438,7 @@ class FrameProcessor(BaseObject):
async def __push_frame_task_handler(self):
while True:
(frame, direction) = await self.__push_queue.get()
self.start_watchdog()
await self.__internal_push_frame(frame, direction)
self.__push_queue.task_done()
self.reset_watchdog()

4
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -783,14 +783,18 @@ class RTVIProcessor(FrameProcessor):
async def _action_task_handler(self):
while True:
frame = await self._action_queue.get()
self.start_watchdog()
await self._handle_action(frame.message_id, frame.rtvi_action_run)
self._action_queue.task_done()
self.reset_watchdog()
async def _message_task_handler(self):
while True:
message = await self._message_queue.get()
self.start_watchdog()
await self._handle_message(message)
self._message_queue.task_done()
self.reset_watchdog()
async def _handle_transport_message(self, frame: TransportMessageUrgentFrame):
try:

3
src/pipecat/services/assemblyai/stt.py
View File

@@ -190,6 +190,7 @@ class AssemblyAISTTService(STTService):
while self._connected:
try:
message = await self._websocket.recv()
self.start_watchdog()
data = json.loads(message)
await self._handle_message(data)
except websockets.exceptions.ConnectionClosedOK:
@@ -197,6 +198,8 @@ class AssemblyAISTTService(STTService):
except Exception as e:
logger.error(f"Error processing WebSocket message: {e}")
break
finally:
self.reset_watchdog()
except Exception as e:
logger.error(f"Fatal error in receive handler: {e}")

5
src/pipecat/services/aws/stt.py
View File

@@ -285,6 +285,9 @@ class AWSTranscribeSTTService(STTService):
try:
response = await self._ws_client.recv()
self.start_watchdog()
headers, payload = decode_event(response)
if headers.get(":message-type") == "event":
@@ -342,3 +345,5 @@ class AWSTranscribeSTTService(STTService):
except Exception as e:
logger.error(f"{self} Unexpected error in receive loop: {e}")
break
finally:
self.reset_watchdog()

4
src/pipecat/services/aws_nova_sonic/aws.py
View File

@@ -699,6 +699,8 @@ class AWSNovaSonicLLMService(LLMService):
output = await self._stream.await_output()
result = await output[1].receive()
self.start_watchdog()
if result.value and result.value.bytes_:
response_data = result.value.bytes_.decode("utf-8")
json_data = json.loads(response_data)
@@ -731,6 +733,8 @@ class AWSNovaSonicLLMService(LLMService):
logger.error(f"{self} error processing responses: {e}")
if self._wants_connection:
await self.reset_conversation()
finally:
self.reset_watchdog()
async def _handle_completion_start_event(self, event_json):
pass

55
src/pipecat/services/elevenlabs/tts.py
View File

@@ -284,7 +284,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
logger.trace(f"{self}: flushing audio")
msg = {"context_id": self._context_id, "flush": True}
await self._websocket.send(json.dumps(msg))
self._context_id = None
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
await super().push_frame(frame, direction)
@@ -380,6 +379,12 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
if self._context_id and self._websocket:
logger.trace(f"Closing context {self._context_id} due to interruption")
try:
# ElevenLabs requires that Pipecat manages the contexts and closes them
# when they're not longer in use. Since a StartInterruptionFrame is pushed
# every time the user speaks, we'll use this as a trigger to close the context
# and reset the state.
# Note: We do not need to call remove_audio_context here, as the context is
# automatically reset when super ()._handle_interruption is called.
await self._websocket.send(
json.dumps({"context_id": self._context_id, "close_context": True})
)
@@ -391,10 +396,20 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
async def _receive_messages(self):
async for message in self._get_websocket():
msg = json.loads(message)
# Check if this message belongs to the current context
received_ctx_id = msg.get("contextId")
# Handle final messages first, regardless of context availability
# At the moment, this message is received AFTER the close_context message is
# sent, so it doesn't serve any functional purpose. For now, we'll just log it.
if msg.get("isFinal") is True:
logger.trace(f"Received final message for context {received_ctx_id}")
continue
# Check if this message belongs to the current context.
# This should never happen, so warn about it.
if not self.audio_context_available(received_ctx_id):
logger.trace(f"Ignoring message from unavailable context: {received_ctx_id}")
logger.warning(f"Ignoring message from unavailable context: {received_ctx_id}")
continue
if msg.get("audio"):
@@ -408,21 +423,26 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
word_times = calculate_word_times(msg["alignment"], self._cumulative_time)
await self.add_word_timestamps(word_times)
self._cumulative_time = word_times[-1][1]
if msg.get("isFinal"):
logger.trace(f"Received final message for context {received_ctx_id}")
await self.remove_audio_context(received_ctx_id)
# Reset context tracking if this was our active context
if self._context_id == received_ctx_id:
self._context_id = None
self._started = False
async def _keepalive_task_handler(self):
while True:
await asyncio.sleep(10)
try:
# Send an empty message to keep the connection alive
if self._websocket and self._websocket.open:
await self._websocket.send(json.dumps({}))
if self._context_id:
# Send keepalive with context ID to keep the connection alive
keepalive_message = {
"text": "",
"context_id": self._context_id,
}
logger.trace(f"Sending keepalive for context {self._context_id}")
else:
# It's possible to have a user interruption which clears the context
# without generating a new TTS response. In this case, we'll just send
# an empty message to keep the connection alive.
keepalive_message = {"text": ""}
logger.trace("Sending keepalive without context")
await self._websocket.send(json.dumps(keepalive_message))
except websockets.ConnectionClosed as e:
logger.warning(f"{self} keepalive error: {e}")
break
@@ -441,14 +461,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
await self._connect()
try:
# Close previous context if there was one
if self._context_id and not self._started:
await self._websocket.send(
json.dumps({"context_id": self._context_id, "close_context": True})
)
await self.remove_audio_context(self._context_id)
self._context_id = None
if not self._started:
await self.start_ttfb_metrics()
yield TTSStartedFrame()
@@ -473,9 +485,6 @@ class ElevenLabsTTSService(AudioContextWordTTSService):
logger.error(f"{self} error sending message: {e}")
yield TTSStoppedFrame()
self._started = False
if self._context_id:
await self.remove_audio_context(self._context_id)
self._context_id = None
return
yield None
except Exception as e:

6
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

@@ -687,6 +687,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
async def _receive_task_handler(self):
async for message in self._websocket:
self.start_watchdog()
evt = events.parse_server_event(message)
# logger.debug(f"Received event: {message[:500]}")
# logger.debug(f"Received event: {evt}")
@@ -708,8 +710,8 @@ class GeminiMultimodalLiveLLMService(LLMService):
await self._handle_evt_error(evt)
# errors are fatal, so exit the receive loop
return
else:
pass
self.reset_watchdog()
#
#

6
src/pipecat/services/gladia/stt.py
View File

@@ -502,6 +502,8 @@ class GladiaSTTService(STTService):
async def _receive_task_handler(self):
try:
async for message in self._websocket:
self.start_watchdog()
content = json.loads(message)
# Handle audio chunk acknowledgments
@@ -559,11 +561,15 @@ class GladiaSTTService(STTService):
translation, "", time_now_iso8601(), translated_language
)
)
self.reset_watchdog()
except websockets.exceptions.ConnectionClosed:
# Expected when closing the connection
pass
except Exception as e:
logger.error(f"Error in Gladia WebSocket handler: {e}")
finally:
self.reset_watchdog()
async def _maybe_reconnect(self) -> bool:
"""Handle exponential backoff reconnection logic."""

19
src/pipecat/services/google/stt.py
View File

@@ -747,9 +747,12 @@ class GoogleSTTService(STTService):
try:
while True:
try:
self.start_watchdog()
if self._request_queue.empty():
# wait for 10ms in case we don't have audio
await asyncio.sleep(0.01)
self.reset_watchdog()
continue
# Start bi-directional streaming
@@ -760,12 +763,13 @@ class GoogleSTTService(STTService):
# Process responses
await self._process_responses(streaming_recognize)
self.reset_watchdog()
# If we're here, check if we need to reconnect
if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT:
logger.debug("Reconnecting stream after timeout")
# Reset stream start time
self._stream_start_time = int(time.time() * 1000)
continue
else:
# Normal stream end
break
@@ -775,7 +779,8 @@ class GoogleSTTService(STTService):
await asyncio.sleep(1) # Brief delay before reconnecting
self._stream_start_time = int(time.time() * 1000)
continue
finally:
self.reset_watchdog()
except Exception as e:
logger.error(f"Error in streaming task: {e}")
@@ -800,12 +805,16 @@ class GoogleSTTService(STTService):
"""Process streaming recognition responses."""
try:
async for response in streaming_recognize:
self.start_watchdog()
# Check streaming limit
if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT:
logger.debug("Stream timeout reached in response processing")
self.reset_watchdog()
break
if not response.results:
self.reset_watchdog()
continue
for result in response.results:
@@ -848,8 +857,10 @@ class GoogleSTTService(STTService):
)
)
self.reset_watchdog()
except Exception as e:
logger.error(f"Error processing Google STT responses: {e}")
# Re-raise the exception to let it propagate (e.g. in the case of a timeout, propagate to _stream_audio to reconnect)
self.reset_watchdog()
# Re-raise the exception to let it propagate (e.g. in the case of a
# timeout, propagate to _stream_audio to reconnect)
raise

27
src/pipecat/services/mcp_service.py
View File

@@ -8,8 +8,8 @@ from pipecat.adapters.schemas.tools_schema import ToolsSchema
from pipecat.utils.base_object import BaseObject
try:
from mcp import ClientSession, StdioServerParameters, types
from mcp.client.session import ClientSession
from mcp import ClientSession, StdioServerParameters
from mcp.client.session_group import SseServerParameters
from mcp.client.sse import sse_client
from mcp.client.stdio import stdio_client
except ModuleNotFoundError as e:
@@ -21,7 +21,7 @@ except ModuleNotFoundError as e:
class MCPClient(BaseObject):
def __init__(
self,
server_params: Union[StdioServerParameters, str],
server_params: Union[StdioServerParameters, SseServerParameters],
**kwargs,
):
super().__init__(**kwargs)
@@ -30,12 +30,12 @@ class MCPClient(BaseObject):
if isinstance(server_params, StdioServerParameters):
self._client = stdio_client
self._register_tools = self._stdio_register_tools
elif isinstance(server_params, str):
elif isinstance(server_params, SseServerParameters):
self._client = sse_client
self._register_tools = self._sse_register_tools
else:
raise TypeError(
f"{self} invalid argument type: `server_params` must be either StdioServerParameters or an SSE server url string."
f"{self} invalid argument type: `server_params` must be either StdioServerParameters or SseServerParameters."
)
async def register_tools(self, llm) -> ToolsSchema:
@@ -90,7 +90,12 @@ class MCPClient(BaseObject):
logger.debug(f"Executing tool '{function_name}' with call ID: {tool_call_id}")
logger.trace(f"Tool arguments: {json.dumps(arguments, indent=2)}")
try:
async with self._client(self._server_params) as (read, write):
async with self._client(
url=self._server_params.url,
headers=self._server_params.headers,
timeout=self._server_params.timeout,
sse_read_timeout=self._server_params.sse_read_timeout,
) as (read, write):
async with self._session(read, write) as session:
await session.initialize()
await self._call_tool(session, function_name, arguments, result_callback)
@@ -100,10 +105,14 @@ class MCPClient(BaseObject):
logger.exception("Full exception details:")
await result_callback(error_msg)
logger.debug("Starting registration of mcp.run tools")
tool_schemas: List[FunctionSchema] = []
logger.debug(f"SSE server parameters: {self._server_params}")
async with self._client(self._server_params) as (read, write):
async with self._client(
url=self._server_params.url,
headers=self._server_params.headers,
timeout=self._server_params.timeout,
sse_read_timeout=self._server_params.sse_read_timeout,
) as (read, write):
async with self._session(read, write) as session:
await session.initialize()
tools_schema = await self._list_tools(session, mcp_tool_wrapper, llm)

9
src/pipecat/services/openai_realtime_beta/events.py
View File

@@ -36,10 +36,6 @@ class InputAudioTranscription(BaseModel):
prompt: Optional[str] = None,
):
super().__init__(model=model, language=language, prompt=prompt)
if self.model != "gpt-4o-transcribe" and (self.language or self.prompt):
raise ValueError(
"Fields 'language' and 'prompt' are only supported when model is 'gpt-4o-transcribe'"
)
class TurnDetection(BaseModel):
@@ -207,12 +203,11 @@ class ResponseCancelEvent(ClientEvent):
class ServerEvent(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
event_id: str
type: str
class Config:
arbitrary_types_allowed = True
class SessionCreatedEvent(ServerEvent):
type: Literal["session.created"]

4
src/pipecat/services/openai_realtime_beta/openai.py
View File

@@ -86,7 +86,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
self,
*,
api_key: str,
model: str = "gpt-4o-realtime-preview-2024-12-17",
model: str = "gpt-4o-realtime-preview-2025-06-03",
base_url: str = "wss://api.openai.com/v1/realtime",
session_properties: Optional[events.SessionProperties] = None,
start_audio_paused: bool = False,
@@ -370,6 +370,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
async def _receive_task_handler(self):
async for message in self._websocket:
self.start_watchdog()
evt = events.parse_server_event(message)
if evt.type == "session.created":
await self._handle_evt_session_created(evt)
@@ -400,6 +401,7 @@ class OpenAIRealtimeBetaLLMService(LLMService):
await self._handle_evt_error(evt)
# errors are fatal, so exit the receive loop
return
self.reset_watchdog()
@traced_openai_realtime(operation="llm_setup")
async def _handle_evt_session_created(self, evt):

4
src/pipecat/services/riva/stt.py
View File

@@ -224,11 +224,13 @@ class RivaSTTService(STTService):
streaming_config=self._config,
)
for response in responses:
self.start_watchdog()
if not response.results:
continue
asyncio.run_coroutine_threadsafe(
self._response_queue.put(response), self.get_event_loop()
)
self.reset_watchdog()
async def _thread_task_handler(self):
try:
@@ -283,7 +285,9 @@ class RivaSTTService(STTService):
async def _response_task_handler(self):
while True:
response = await self._response_queue.get()
self.start_watchdog()
await self._handle_response(response)
self.reset_watchdog()
async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
await self.start_ttfb_metrics()

4
src/pipecat/services/simli/video.py
View File

@@ -62,6 +62,7 @@ class SimliVideoService(FrameProcessor):
async def _consume_and_process_audio(self):
await self._pipecat_resampler_event.wait()
async for audio_frame in self._simli_client.getAudioStreamIterator():
self.start_watchdog()
resampled_frames = self._pipecat_resampler.resample(audio_frame)
for resampled_frame in resampled_frames:
audio_array = resampled_frame.to_ndarray()
@@ -74,10 +75,12 @@ class SimliVideoService(FrameProcessor):
num_channels=1,
),
)
self.reset_watchdog()
async def _consume_and_process_video(self):
await self._pipecat_resampler_event.wait()
async for video_frame in self._simli_client.getVideoStreamIterator(targetFormat="rgb24"):
self.start_watchdog()
# Process the video frame
convertedFrame: OutputImageRawFrame = OutputImageRawFrame(
image=video_frame.to_rgb().to_image().tobytes(),
@@ -86,6 +89,7 @@ class SimliVideoService(FrameProcessor):
)
convertedFrame.pts = video_frame.pts
await self.push_frame(convertedFrame)
self.reset_watchdog()
async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction)

4
src/pipecat/services/tavus/video.py
View File

@@ -217,5 +217,7 @@ class TavusVideoService(AIService):
async def _send_task_handler(self):
while True:
frame = await self._queue.get()
if isinstance(frame, OutputAudioRawFrame):
self.start_watchdog()
if isinstance(frame, OutputAudioRawFrame) and self._client:
await self._client.write_audio_frame(frame)
self.reset_watchdog()

54
src/pipecat/transports/base_input.py
View File

@@ -43,6 +43,8 @@ from pipecat.metrics.metrics import MetricsData
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
from pipecat.transports.base_transport import TransportParams
AUDIO_INPUT_TIMEOUT_SECS = 0.5
class BaseInputTransport(FrameProcessor):
def __init__(self, params: TransportParams, **kwargs):
@@ -56,6 +58,9 @@ class BaseInputTransport(FrameProcessor):
# Track bot speaking state for interruption logic
self._bot_speaking = False
# Track user speaking state for interruption logic
self._user_speaking = False
# We read audio from a single queue one at a time and we then run VAD in
# a thread. Therefore, only one thread should be necessary.
self._executor = ThreadPoolExecutor(max_workers=1)
@@ -130,6 +135,7 @@ class BaseInputTransport(FrameProcessor):
async def start(self, frame: StartFrame):
self._paused = False
self._user_speaking = False
self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
@@ -240,6 +246,7 @@ class BaseInputTransport(FrameProcessor):
async def _handle_user_interruption(self, frame: Frame):
if isinstance(frame, UserStartedSpeakingFrame):
logger.debug("User started speaking")
self._user_speaking = True
await self.push_frame(frame)
# Only push StartInterruptionFrame if:
@@ -263,6 +270,7 @@ class BaseInputTransport(FrameProcessor):
)
elif isinstance(frame, UserStoppedSpeakingFrame):
logger.debug("User stopped speaking")
self._user_speaking = False
await self.push_frame(frame)
if self.interruptions_allowed:
await self._stop_interruption()
@@ -355,26 +363,42 @@ class BaseInputTransport(FrameProcessor):
async def _audio_task_handler(self):
vad_state: VADState = VADState.QUIET
while True:
frame: InputAudioRawFrame = await self._audio_in_queue.get()
try:
frame: InputAudioRawFrame = await asyncio.wait_for(
self._audio_in_queue.get(), timeout=AUDIO_INPUT_TIMEOUT_SECS
)
# If an audio filter is available, run it before VAD.
if self._params.audio_in_filter:
frame.audio = await self._params.audio_in_filter.filter(frame.audio)
self.start_watchdog()
# Check VAD and push event if necessary. We just care about
# changes from QUIET to SPEAKING and vice versa.
previous_vad_state = vad_state
if self._params.vad_analyzer:
vad_state = await self._handle_vad(frame, vad_state)
# If an audio filter is available, run it before VAD.
if self._params.audio_in_filter:
frame.audio = await self._params.audio_in_filter.filter(frame.audio)
if self._params.turn_analyzer:
await self._run_turn_analyzer(frame, vad_state, previous_vad_state)
# Check VAD and push event if necessary. We just care about
# changes from QUIET to SPEAKING and vice versa.
previous_vad_state = vad_state
if self._params.vad_analyzer:
vad_state = await self._handle_vad(frame, vad_state)
# Push audio downstream if passthrough is set.
if self._params.audio_in_passthrough:
await self.push_frame(frame)
if self._params.turn_analyzer:
await self._run_turn_analyzer(frame, vad_state, previous_vad_state)
self._audio_in_queue.task_done()
# Push audio downstream if passthrough is set.
if self._params.audio_in_passthrough:
await self.push_frame(frame)
self._audio_in_queue.task_done()
except asyncio.TimeoutError:
if self._user_speaking:
logger.warning(
"Forcing user stopped speaking due to timeout receiving audio frame!"
)
vad_state = VADState.QUIET
if self._params.turn_analyzer:
self._params.turn_analyzer.clear()
await self._handle_user_interruption(UserStoppedSpeakingFrame())
finally:
self.reset_watchdog()
async def _handle_prediction_result(self, result: MetricsData):
"""Handle a prediction result event from the turn analyzer.

27
src/pipecat/transports/network/fastapi_websocket.py
View File

@@ -70,11 +70,22 @@ class FastAPIWebsocketClient:
return self._websocket.iter_bytes() if self._is_binary else self._websocket.iter_text()
async def send(self, data: str | bytes):
if self._can_send():
if self._is_binary:
await self._websocket.send_bytes(data)
else:
await self._websocket.send_text(data)
try:
if self._can_send():
if self._is_binary:
await self._websocket.send_bytes(data)
else:
await self._websocket.send_text(data)
except Exception as e:
logger.error(
f"{self} exception sending data: {e.__class__.__name__} ({e}), application_state: {self._websocket.application_state}"
)
# For some reason the websocket is disconnected, and we are not able to send data
# So let's properly handle it and disconnect the transport
if self._websocket.application_state == WebSocketState.DISCONNECTED:
logger.warning("Closing already disconnected websocket!")
self._closing = True
await self.trigger_client_disconnected()
async def disconnect(self):
self._leave_counter -= 1
@@ -171,6 +182,8 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
if not self._params.serializer:
continue
self.start_watchdog()
frame = await self._params.serializer.deserialize(message)
if not frame:
@@ -180,9 +193,13 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
await self.push_audio_frame(frame)
else:
await self.push_frame(frame)
self.reset_watchdog()
except Exception as e:
logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})")
self.reset_watchdog()
await self._client.trigger_client_disconnected()
async def _monitor_websocket(self):

4
src/pipecat/transports/network/small_webrtc.py
View File

@@ -423,8 +423,10 @@ class SmallWebRTCInputTransport(BaseInputTransport):
async def _receive_audio(self):
try:
async for audio_frame in self._client.read_audio_frame():
self.start_watchdog()
if audio_frame:
await self.push_audio_frame(audio_frame)
self.reset_watchdog()
except Exception as e:
logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})")
@@ -432,6 +434,7 @@ class SmallWebRTCInputTransport(BaseInputTransport):
async def _receive_video(self):
try:
async for video_frame in self._client.read_video_frame():
self.start_watchdog()
if video_frame:
await self.push_video_frame(video_frame)
@@ -450,6 +453,7 @@ class SmallWebRTCInputTransport(BaseInputTransport):
await self.push_video_frame(image_frame)
# Remove from pending requests
del self._image_requests[req_id]
self.reset_watchdog()
except Exception as e:
logger.error(f"{self} exception receiving data: {e.__class__.__name__} ({e})")

2
src/pipecat/transports/services/livekit.py
View File

@@ -415,6 +415,7 @@ class LiveKitInputTransport(BaseInputTransport):
logger.info("Audio input task started")
while True:
audio_data = await self._client.get_next_audio_frame()
self.start_watchdog()
if audio_data:
audio_frame_event, participant_id = audio_data
pipecat_audio_frame = await self._convert_livekit_audio_to_pipecat(
@@ -427,6 +428,7 @@ class LiveKitInputTransport(BaseInputTransport):
num_channels=pipecat_audio_frame.num_channels,
)
await self.push_audio_frame(input_audio_frame)
self.reset_watchdog()
async def _convert_livekit_audio_to_pipecat(
self, audio_frame_event: rtc.AudioFrameEvent

186
src/pipecat/utils/asyncio.py
View File

@@ -5,15 +5,30 @@
#
import asyncio
import time
from abc import ABC, abstractmethod
from typing import Coroutine, Dict, Optional, Sequence, Set
from dataclasses import dataclass
from typing import Coroutine, Dict, List, Optional, Sequence
from loguru import logger
WATCHDOG_TIMEOUT = 5.0
@dataclass
class TaskManagerParams:
loop: asyncio.AbstractEventLoop
enable_watchdog_logging: bool = False
watchdog_timeout: float = WATCHDOG_TIMEOUT
class BaseTaskManager(ABC):
@abstractmethod
def set_event_loop(self, loop: asyncio.AbstractEventLoop):
def setup(self, params: TaskManagerParams):
pass
@abstractmethod
async def cleanup(self):
pass
@abstractmethod
@@ -21,7 +36,14 @@ class BaseTaskManager(ABC):
pass
@abstractmethod
def create_task(self, coroutine: Coroutine, name: str) -> asyncio.Task:
def create_task(
self,
coroutine: Coroutine,
name: str,
*,
enable_watchdog_logging: Optional[bool] = None,
watchdog_timeout: Optional[float] = None,
) -> asyncio.Task:
"""
Creates and schedules a new asyncio Task that runs the given coroutine.
@@ -31,6 +53,8 @@ class BaseTaskManager(ABC):
loop (asyncio.AbstractEventLoop): The event loop to use for creating the task.
coroutine (Coroutine): The coroutine to be executed within the task.
name (str): The name to assign to the task for identification.
enable_watchdog_logging(bool): whether this task should log watchdog processing times.
watchdog_timeout(float): watchdog timer timeout for this task.
Returns:
asyncio.Task: The created task object.
@@ -73,21 +97,64 @@ class BaseTaskManager(ABC):
"""Returns the list of currently created/registered tasks."""
pass
@abstractmethod
def start_watchdog(self, task: asyncio.Task):
"""Starts the given task watchdog timer. If not reset, a warning will be
logged indicating the task is stalling.
"""
pass
@abstractmethod
def reset_watchdog(self, task: asyncio.Task):
"""Resets the given task watchdog timer. If not reset, a warning will be
logged indicating the task is stalling.
"""
pass
@dataclass
class TaskData:
task: asyncio.Task
watchdog_start: asyncio.Event
watchdog_timer: asyncio.Event
enable_watchdog_logging: bool
watchdog_timeout: float
class TaskManager(BaseTaskManager):
def __init__(self) -> None:
self._tasks: Dict[str, asyncio.Task] = {}
self._loop: Optional[asyncio.AbstractEventLoop] = None
self._tasks: Dict[str, TaskData] = {}
self._params: Optional[TaskManagerParams] = None
self._watchdog_tasks: List[asyncio.Task] = []
def set_event_loop(self, loop: asyncio.AbstractEventLoop):
self._loop = loop
def setup(self, params: TaskManagerParams):
if not self._params:
self._params = params
async def cleanup(self):
for task in self._watchdog_tasks:
try:
task.cancel()
await task
except asyncio.CancelledError:
# This is expected, no need to re-raise.
pass
def get_event_loop(self) -> asyncio.AbstractEventLoop:
if not self._loop:
raise Exception("TaskManager missing event loop, use TaskManager.set_event_loop().")
return self._loop
if not self._params:
raise Exception("TaskManager is not setup: unable to get event loop")
return self._params.loop
def create_task(self, coroutine: Coroutine, name: str) -> asyncio.Task:
def create_task(
self,
coroutine: Coroutine,
name: str,
*,
enable_watchdog_logging: Optional[bool] = None,
watchdog_timeout: Optional[float] = None,
) -> asyncio.Task:
"""
Creates and schedules a new asyncio Task that runs the given coroutine.
@@ -97,6 +164,8 @@ class TaskManager(BaseTaskManager):
loop (asyncio.AbstractEventLoop): The event loop to use for creating the task.
coroutine (Coroutine): The coroutine to be executed within the task.
name (str): The name to assign to the task for identification.
enable_watchdog_logging(bool): whether this task should log watchdog processing time.
watchdog_timeout(float): watchdog timer timeout for this task.
Returns:
asyncio.Task: The created task object.
@@ -112,12 +181,26 @@ class TaskManager(BaseTaskManager):
except Exception as e:
logger.exception(f"{name}: unexpected exception: {e}")
if not self._loop:
raise Exception("TaskManager missing event loop, use TaskManager.set_event_loop().")
if not self._params:
raise Exception("TaskManager is not setup: unable to get event loop")
task = self._loop.create_task(run_coroutine())
task = self._params.loop.create_task(run_coroutine())
task.set_name(name)
self._add_task(task)
self._add_task(
TaskData(
task=task,
watchdog_start=asyncio.Event(),
watchdog_timer=asyncio.Event(),
enable_watchdog_logging=(
enable_watchdog_logging
if enable_watchdog_logging
else self._params.enable_watchdog_logging
),
watchdog_timeout=(
watchdog_timeout if watchdog_timeout else self._params.watchdog_timeout
),
)
)
logger.trace(f"{name}: task created")
return task
@@ -165,6 +248,8 @@ class TaskManager(BaseTaskManager):
name = task.get_name()
task.cancel()
try:
# Make sure to reset watchdog if a task is cancelled.
self.reset_watchdog(task)
if timeout:
await asyncio.wait_for(task, timeout=timeout)
else:
@@ -176,16 +261,51 @@ class TaskManager(BaseTaskManager):
pass
except Exception as e:
logger.exception(f"{name}: unexpected exception while cancelling task: {e}")
except BaseException as e:
logger.critical(f"{name}: fatal base exception while cancelling task: {e}")
raise
finally:
self._remove_task(task)
def current_tasks(self) -> Sequence[asyncio.Task]:
"""Returns the list of currently created/registered tasks."""
return list(self._tasks.values())
return [data.task for data in self._tasks.values()]
def _add_task(self, task: asyncio.Task):
def start_watchdog(self, task: asyncio.Task):
"""Starts the given task watchdog timer. If not reset, a warning will be
logged indicating the task is stalling. If the timer was already started
a warning will be logged.
"""
name = task.get_name()
self._tasks[name] = task
if name in self._tasks:
if self._tasks[name].watchdog_start.is_set():
logger.warning(f"Watchdog timer for task {name} already started")
else:
self._tasks[name].watchdog_timer.clear()
self._tasks[name].watchdog_start.set()
else:
logger.warning(f"Unable to start watchdog timer: task {name} does not exist")
def reset_watchdog(self, task: asyncio.Task):
"""Resets the given task watchdog timer. If not reset, a warning will be
logged indicating the task is stalling.
"""
name = task.get_name()
if name in self._tasks:
self._tasks[name].watchdog_start.clear()
self._tasks[name].watchdog_timer.set()
else:
logger.warning(f"Unable to reset watchdog timer: task {name} does not exist")
def _add_task(self, task_data: TaskData):
name = task_data.task.get_name()
self._tasks[name] = task_data
watchdog_task = self.get_event_loop().create_task(
self._watchdog_task_handler(self._tasks[name])
)
self._watchdog_tasks.append(watchdog_task)
def _remove_task(self, task: asyncio.Task):
name = task.get_name()
@@ -193,3 +313,33 @@ class TaskManager(BaseTaskManager):
del self._tasks[name]
except KeyError as e:
logger.trace(f"{name}: unable to remove task (already removed?): {e}")
async def _watchdog_task_handler(self, task_data: TaskData):
name = task_data.task.get_name()
start = task_data.watchdog_start
timer = task_data.watchdog_timer
enable_watchdog_logging = task_data.enable_watchdog_logging
watchdog_timeout = task_data.watchdog_timeout
async def wait_for_reset():
waiting = True
while waiting:
try:
start_time = time.time()
await asyncio.wait_for(timer.wait(), timeout=watchdog_timeout)
total_time = time.time() - start_time
if enable_watchdog_logging:
logger.debug(f"{name} task processing time: {total_time:.20f}")
waiting = False
except asyncio.TimeoutError:
logger.warning(
f"{name}: task is taking too long {WATCHDOG_TIMEOUT} second(s) (forgot to reset watchdog?)"
)
finally:
timer.clear()
while True:
# Wait for the user to start the watchdog timer.
await start.wait()
# Now, waiting for the task to finish.
await wait_for_reset()