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

Merge pull request #2464 from pipecat-ai/aleix/frame-processor-updates

Browse Source 
various frame processor updates
This commit is contained in:
Aleix Conchillo Flaqué
2025-08-20 10:11:49 -07:00
committed by GitHub
parent 2f74a7e674 6831e63ec9
commit 5286591826
14 changed files with 436 additions and 540 deletions
Download Patch File Download Diff File Expand all files Collapse all files

32
CHANGELOG.md
View File

@@ -9,6 +9,22 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
### Added
- Allow passing custom pipeline sink and source processors to a
`Pipeline`. Pipeline source and sink processors are used to know and control
what's coming in and out of a `Pipeline` processor.
- Added `FrameProcessor.pause_processing_system_frames()` and
`FrameProcessor.resume_processing_system_frames()`. These allow to pause and
resume the processing of system frame.
- Added new `on_process_frame()` observer method which makes it possible to know
when a frame is being processed.
- Added new `FrameProcessor.entry_processor()` method. This allows you to access
the first non-compound processor in a pipeline.
- Added `FrameProcessor` properties `processors`, `next` and `previous`.
- `ElevenLabsTTSService` now supports additional runtime changes to the `model`,
`language`, and `voice_settings` parameters.
@@ -23,6 +39,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
`retry_timeout_secs` and `retry_on_timeout`. This feature is disabled by
default.
### Removed
- Removed unused `FrameProcessor.set_parent()` and
`FrameProcessor.get_parent()`.
### Fixed
- Fixed an `AudioBufferProcessor` issues that would cause audio overlap when
@@ -31,6 +52,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
- Fixed an issue where `AsyncAITTSService` had very high latency in responding
by adding `force=true` when sending the flush command.
### Performance
- Improve `PipelineTask` performance by using direct mode processors and by
removing unnecessary tasks.
- Improve `ParallelPipeline` performance by using direct mode, by not
creating a task for each frame and every sub-pipeline and also by removing
other unnecessary tasks.
- `Pipeline` performance improvements by using direct mode.
### Other
- Added `14w-function-calling-mistal.py` using `MistralLLMService`.

36
src/pipecat/observers/base_observer.py
View File

@@ -11,7 +11,6 @@ processors without modifying the pipeline structure. Observers can be used
for logging, debugging, analytics, and monitoring pipeline behavior.
"""
from abc import abstractmethod
from dataclasses import dataclass
from typing_extensions import TYPE_CHECKING
@@ -23,6 +22,28 @@ if TYPE_CHECKING:
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
@dataclass
class FrameProcessed:
"""Event data for frame processing in the pipeline.
Represents an event where a frame is being processed by a processor. This
data structure is typically used by observers to track the flow of frames
through the pipeline for logging, debugging, or analytics purposes.
Parameters:
processor: The processor processing the frame.
frame: The frame being processed.
direction: The direction of the frame (e.g., downstream or upstream).
timestamp: The time when the frame was pushed, based on the pipeline clock.
"""
processor: "FrameProcessor"
frame: Frame
direction: "FrameDirection"
timestamp: int
@dataclass
class FramePushed:
"""Event data for frame transfers between processors in the pipeline.
@@ -56,7 +77,18 @@ class BaseObserver(BaseObject):
performance analysis, and analytics collection.
"""
@abstractmethod
async def on_process_frame(self, data: FrameProcessed):
"""Handle the event when a frame is being processed by a processor.
This method should be implemented by subclasses to define specific
behavior (e.g., logging, monitoring, debugging) when a frame is
being processed by a processor.
Args:
data: The event data containing details about the frame processing.
"""
pass
async def on_push_frame(self, data: FramePushed):
"""Handle the event when a frame is pushed from one processor to another.

25
src/pipecat/pipeline/base_pipeline.py
View File

@@ -6,31 +6,12 @@
"""Base pipeline implementation for frame processing."""
from abc import abstractmethod
from typing import List
from pipecat.processors.frame_processor import FrameProcessor
class BasePipeline(FrameProcessor):
"""Base class for all pipeline implementations.
"""Base class for all pipeline implementations."""
Provides the foundation for pipeline processors that need to support
metrics collection from their contained processors.
"""
def __init__(self):
def __init__(self, **kwargs):
"""Initialize the base pipeline."""
super().__init__()
@abstractmethod
def processors_with_metrics(self) -> List[FrameProcessor]:
"""Return processors that can generate metrics.
Implementing classes should collect and return all processors within
their pipeline that support metrics generation.
Returns:
List of frame processors that support metrics collection.
"""
pass
super().__init__(**kwargs)

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

@@ -11,106 +11,15 @@ sub-pipelines concurrently, with coordination for system frames and proper
handling of pipeline lifecycle events.
"""
import asyncio
from itertools import chain
from typing import Awaitable, Callable, Dict, List
from typing import Dict, List
from loguru import logger
from pipecat.frames.frames import (
CancelFrame,
EndFrame,
Frame,
StartFrame,
StartInterruptionFrame,
SystemFrame,
)
from pipecat.frames.frames import EndFrame, Frame, StartFrame
from pipecat.pipeline.base_pipeline import BasePipeline
from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.pipeline import Pipeline, PipelineSink, PipelineSource
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
class ParallelPipelineSource(FrameProcessor):
"""Source processor for parallel pipeline branches.
Handles frame routing for parallel pipeline inputs, directing system frames
to the parent push function and other upstream frames to a queue for processing.
"""
def __init__(
self,
upstream_queue: asyncio.Queue,
push_frame_func: Callable[[Frame, FrameDirection], Awaitable[None]],
):
"""Initialize the parallel pipeline source.
Args:
upstream_queue: Queue for collecting upstream frames from this branch.
push_frame_func: Function to push frames to the parent parallel pipeline.
"""
super().__init__(enable_direct_mode=True)
self._up_queue = upstream_queue
self._push_frame_func = push_frame_func
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames with special handling for system frames.
Args:
frame: The frame to process.
direction: The direction of frame flow.
"""
await super().process_frame(frame, direction)
match direction:
case FrameDirection.UPSTREAM:
if isinstance(frame, SystemFrame):
await self._push_frame_func(frame, direction)
else:
await self._up_queue.put(frame)
case FrameDirection.DOWNSTREAM:
await self.push_frame(frame, direction)
class ParallelPipelineSink(FrameProcessor):
"""Sink processor for parallel pipeline branches.
Handles frame routing for parallel pipeline outputs, directing system frames
to the parent push function and other downstream frames to a queue for coordination.
"""
def __init__(
self,
downstream_queue: asyncio.Queue,
push_frame_func: Callable[[Frame, FrameDirection], Awaitable[None]],
):
"""Initialize the parallel pipeline sink.
Args:
downstream_queue: Queue for collecting downstream frames from this branch.
push_frame_func: Function to push frames to the parent parallel pipeline.
"""
super().__init__(enable_direct_mode=True)
self._down_queue = downstream_queue
self._push_frame_func = push_frame_func
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames with special handling for system frames.
Args:
frame: The frame to process.
direction: The direction of frame flow.
"""
await super().process_frame(frame, direction)
match direction:
case FrameDirection.UPSTREAM:
await self.push_frame(frame, direction)
case FrameDirection.DOWNSTREAM:
if isinstance(frame, SystemFrame):
await self._push_frame_func(frame, direction)
else:
await self._down_queue.put(frame)
class ParallelPipeline(BasePipeline):
@@ -132,28 +41,69 @@ class ParallelPipeline(BasePipeline):
Exception: If no processor lists are provided.
TypeError: If any argument is not a list of processors.
"""
# We don't set it to direct mode because we use frame pausing and that
# requires queues.
super().__init__()
if len(args) == 0:
raise Exception(f"ParallelPipeline needs at least one argument")
self._args = args
self._sources = []
self._sinks = []
self._pipelines = []
self._seen_ids = set()
self._endframe_counter: Dict[int, int] = {}
self._start_frame_counter: Dict[int, int] = {}
self._started = False
self._frame_counter: Dict[int, int] = {}
self._up_task = None
self._down_task = None
logger.debug(f"Creating {self} pipelines")
for processors in args:
if not isinstance(processors, list):
raise TypeError(f"ParallelPipeline argument {processors} is not a list")
num_pipelines = len(self._pipelines)
# We add a source before the pipeline and a sink after so we control
# the frames that are pushed upstream and downstream.
source = PipelineSource(
self._parallel_push_frame, name=f"{self}::Source{num_pipelines}"
)
sink = PipelineSink(self._pipeline_sink_push_frame, name=f"{self}::Sink{num_pipelines}")
# Create pipeline
pipeline = Pipeline(processors, source=source, sink=sink)
self._pipelines.append(pipeline)
logger.debug(f"Finished creating {self} pipelines")
#
# BasePipeline
# Frame processor
#
@property
def processors(self):
"""Return the list of sub-processors contained within this processor.
Only compound processors (e.g. pipelines and parallel pipelines) have
sub-processors. Non-compound processors will return an empty list.
Returns:
The list of sub-processors if this is a compound processor.
"""
return self._pipelines
@property
def entry_processors(self) -> List["FrameProcessor"]:
"""Return the list of entry processors for this processor.
Entry processors are the first processors in a compound processor
(e.g. pipelines, parallel pipelines). Note that pipelines can also be an
entry processor as pipelines are processors themselves. Non-compound
processors will simply return an empty list.
Returns:
The list of entry processors.
"""
return self._pipelines
def processors_with_metrics(self) -> List[FrameProcessor]:
"""Collect processors that can generate metrics from all parallel branches.
@@ -162,10 +112,6 @@ class ParallelPipeline(BasePipeline):
"""
return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines))
#
# Frame processor
#
async def setup(self, setup: FrameProcessorSetup):
"""Set up the parallel pipeline and all its branches.
@@ -176,39 +122,14 @@ class ParallelPipeline(BasePipeline):
TypeError: If any processor list argument is not actually a list.
"""
await super().setup(setup)
self._up_queue = WatchdogQueue(setup.task_manager)
self._down_queue = WatchdogQueue(setup.task_manager)
logger.debug(f"Creating {self} pipelines")
for processors in self._args:
if not isinstance(processors, list):
raise TypeError(f"ParallelPipeline argument {processors} is not a list")
# We will add a source before the pipeline and a sink after.
source = ParallelPipelineSource(self._up_queue, self._parallel_push_frame)
sink = ParallelPipelineSink(self._down_queue, self._pipeline_sink_push_frame)
self._sources.append(source)
self._sinks.append(sink)
# Create pipeline
pipeline = Pipeline(processors)
source.link(pipeline)
pipeline.link(sink)
self._pipelines.append(pipeline)
logger.debug(f"Finished creating {self} pipelines")
await asyncio.gather(*[s.setup(setup) for s in self._sources])
await asyncio.gather(*[p.setup(setup) for p in self._pipelines])
await asyncio.gather(*[s.setup(setup) for s in self._sinks])
for p in self._pipelines:
await p.setup(setup)
async def cleanup(self):
"""Clean up the parallel pipeline and all its branches."""
await super().cleanup()
await asyncio.gather(*[s.cleanup() for s in self._sources])
await asyncio.gather(*[p.cleanup() for p in self._pipelines])
await asyncio.gather(*[s.cleanup() for s in self._sinks])
for p in self._pipelines:
await p.cleanup()
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames through all parallel branches with lifecycle coordination.
@@ -219,79 +140,15 @@ class ParallelPipeline(BasePipeline):
"""
await super().process_frame(frame, direction)
if isinstance(frame, StartFrame):
self._start_frame_counter[frame.id] = len(self._pipelines)
elif isinstance(frame, EndFrame):
self._endframe_counter[frame.id] = len(self._pipelines)
elif isinstance(frame, CancelFrame):
await self._cancel()
# Parallel pipeline synchronized frames.
if isinstance(frame, (StartFrame, EndFrame)):
self._frame_counter[frame.id] = len(self._pipelines)
await self.pause_processing_system_frames()
await self.pause_processing_frames()
if direction == FrameDirection.UPSTREAM:
# If we get an upstream frame we process it in each sink.
await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sinks])
elif direction == FrameDirection.DOWNSTREAM:
# If we get a downstream frame we process it in each source.
await asyncio.gather(*[s.queue_frame(frame, direction) for s in self._sources])
# Handle interruptions after everything has been cancelled.
if isinstance(frame, StartInterruptionFrame):
await self._handle_interruption()
# Wait for tasks to finish.
elif isinstance(frame, EndFrame):
await self._stop()
async def _start(self, frame: StartFrame):
"""Start the parallel pipeline processing tasks."""
await self._create_tasks()
async def _stop(self):
"""Stop all parallel pipeline processing tasks."""
if self._up_task:
# The up task doesn't receive an EndFrame, so we just cancel it.
await self.cancel_task(self._up_task)
self._up_task = None
if self._down_task:
# The down tasks waits for the last EndFrame sent by the internal
# pipelines.
await self._down_task
self._down_task = None
async def _cancel(self):
"""Cancel all parallel pipeline processing tasks."""
if self._up_task:
self._up_queue.cancel()
await self.cancel_task(self._up_task)
self._up_task = None
if self._down_task:
self._down_queue.cancel()
await self.cancel_task(self._down_task)
self._down_task = None
async def _create_tasks(self):
"""Create upstream and downstream processing tasks if not already running."""
if not self._up_task:
self._up_task = self.create_task(self._process_up_queue())
if not self._down_task:
self._down_task = self.create_task(self._process_down_queue())
async def _drain_queue(self, queue: asyncio.Queue):
try:
while not queue.empty():
queue.get_nowait()
except asyncio.QueueEmpty:
logger.debug(f"Draining {self} queue already empty")
async def _drain_queues(self):
"""Drain all frames from upstream and downstream queues."""
await self._drain_queue(self._up_queue)
await self._drain_queue(self._down_queue)
async def _handle_interruption(self):
"""Handle interruption by cancelling tasks, draining queues, and restarting."""
await self._cancel()
await self._drain_queues()
await self._create_tasks()
# Process frames in each of the sub-pipelines.
for p in self._pipelines:
await p.queue_frame(frame, direction)
async def _parallel_push_frame(self, frame: Frame, direction: FrameDirection):
"""Push frames while avoiding duplicates using frame ID tracking."""
@@ -300,52 +157,18 @@ class ParallelPipeline(BasePipeline):
await self.push_frame(frame, direction)
async def _pipeline_sink_push_frame(self, frame: Frame, direction: FrameDirection):
if isinstance(frame, StartFrame):
# Decrement counter and check if all pipelines have processed the StartFrame
start_frame_counter = self._start_frame_counter.get(frame.id, 0)
if start_frame_counter > 0:
self._start_frame_counter[frame.id] -= 1
start_frame_counter = self._start_frame_counter[frame.id]
# Parallel pipeline synchronized frames.
if isinstance(frame, (StartFrame, EndFrame)):
# Decrement counter.
frame_counter = self._frame_counter.get(frame.id, 0)
if frame_counter > 0:
self._frame_counter[frame.id] -= 1
frame_counter = self._frame_counter[frame.id]
# Only push the StartFrame when all pipelines have processed it
if start_frame_counter == 0:
self._started = True
await self._start(frame)
# Only push the frame when all pipelines have processed it.
if frame_counter == 0:
await self._parallel_push_frame(frame, direction)
await self.resume_processing_system_frames()
await self.resume_processing_frames()
else:
if self._started:
await self._parallel_push_frame(frame, direction)
else:
await self._down_queue.put(frame)
async def _process_up_queue(self):
"""Process upstream frames from all parallel branches."""
while True:
frame = await self._up_queue.get()
await self._parallel_push_frame(frame, FrameDirection.UPSTREAM)
self._up_queue.task_done()
async def _process_down_queue(self):
"""Process downstream frames with EndFrame coordination.
Coordinates EndFrames to ensure they are only pushed upstream once
all parallel branches have completed processing them.
"""
running = True
while running:
frame = await self._down_queue.get()
endframe_counter = self._endframe_counter.get(frame.id, 0)
# If we have a counter, decrement it.
if endframe_counter > 0:
self._endframe_counter[frame.id] -= 1
endframe_counter = self._endframe_counter[frame.id]
# If we don't have a counter or we reached 0, push the frame.
if endframe_counter == 0:
await self._parallel_push_frame(frame, FrameDirection.DOWNSTREAM)
running = not (endframe_counter == 0 and isinstance(frame, EndFrame))
self._down_queue.task_done()
await self._parallel_push_frame(frame, direction)

71
src/pipecat/pipeline/pipeline.py
View File

@@ -11,7 +11,7 @@ in sequence and manages frame flow between them, along with helper classes
for pipeline source and sink operations.
"""
from typing import Callable, Coroutine, List
from typing import Callable, Coroutine, List, Optional
from pipecat.frames.frames import Frame
from pipecat.pipeline.base_pipeline import BasePipeline
@@ -26,13 +26,14 @@ class PipelineSource(FrameProcessor):
provided upstream handler function.
"""
def __init__(self, upstream_push_frame: Callable[[Frame, FrameDirection], Coroutine]):
def __init__(self, upstream_push_frame: Callable[[Frame, FrameDirection], Coroutine], **kwargs):
"""Initialize the pipeline source.
Args:
upstream_push_frame: Coroutine function to handle upstream frames.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(enable_direct_mode=True)
super().__init__(enable_direct_mode=True, **kwargs)
self._upstream_push_frame = upstream_push_frame
async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -59,13 +60,16 @@ class PipelineSink(FrameProcessor):
provided downstream handler function.
"""
def __init__(self, downstream_push_frame: Callable[[Frame, FrameDirection], Coroutine]):
def __init__(
self, downstream_push_frame: Callable[[Frame, FrameDirection], Coroutine], **kwargs
):
"""Initialize the pipeline sink.
Args:
downstream_push_frame: Coroutine function to handle downstream frames.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(enable_direct_mode=True)
super().__init__(enable_direct_mode=True, **kwargs)
self._downstream_push_frame = downstream_push_frame
async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -92,26 +96,60 @@ class Pipeline(BasePipeline):
provides metrics collection from contained processors.
"""
def __init__(self, processors: List[FrameProcessor]):
def __init__(
self,
processors: List[FrameProcessor],
*,
source: Optional[FrameProcessor] = None,
sink: Optional[FrameProcessor] = None,
):
"""Initialize the pipeline with a list of processors.
Args:
processors: List of frame processors to connect in sequence.
source: An optional pipeline source processor.
sink: An optional pipeline sink processor.
"""
super().__init__()
super().__init__(enable_direct_mode=True)
# Add a source and a sink queue so we can forward frames upstream and
# downstream outside of the pipeline.
self._source = PipelineSource(self.push_frame)
self._sink = PipelineSink(self.push_frame)
self._source = source or PipelineSource(self.push_frame, name=f"{self}::Source")
self._sink = sink or PipelineSink(self.push_frame, name=f"{self}::Sink")
self._processors: List[FrameProcessor] = [self._source] + processors + [self._sink]
self._link_processors()
#
# BasePipeline
# Frame processor
#
@property
def processors(self):
"""Return the list of sub-processors contained within this processor.
Only compound processors (e.g. pipelines and parallel pipelines) have
sub-processors. Non-compound processors will return an empty list.
Returns:
The list of sub-processors if this is a compound processor.
"""
return self._processors
@property
def entry_processors(self) -> List["FrameProcessor"]:
"""Return the list of entry processors for this processor.
Entry processors are the first processors in a compound processor
(e.g. pipelines, parallel pipelines). Note that pipelines can also be an
entry processor as pipelines are processors themselves. Non-compound
processors will simply return an empty list.
Returns:
The list of entry processors.
"""
return [self._source]
def processors_with_metrics(self):
"""Return processors that can generate metrics.
@@ -122,17 +160,12 @@ class Pipeline(BasePipeline):
List of frame processors that can generate metrics.
"""
services = []
for p in self._processors:
if isinstance(p, BasePipeline):
services.extend(p.processors_with_metrics())
elif p.can_generate_metrics():
for p in self.processors:
if p.can_generate_metrics():
services.append(p)
services.extend(p.processors_with_metrics())
return services
#
# Frame processor
#
async def setup(self, setup: FrameProcessorSetup):
"""Set up the pipeline and all contained processors.
@@ -175,7 +208,5 @@ class Pipeline(BasePipeline):
"""Link all processors in sequence and set their parent."""
prev = self._processors[0]
for curr in self._processors[1:]:
prev.set_parent(self)
prev.link(curr)
prev = curr
prev.set_parent(self)

46
src/pipecat/pipeline/sync_parallel_pipeline.py
View File

@@ -134,9 +134,35 @@ class SyncParallelPipeline(BasePipeline):
self._pipelines = []
#
# BasePipeline
# Frame processor
#
@property
def processors(self):
"""Return the list of sub-processors contained within this processor.
Only compound processors (e.g. pipelines and parallel pipelines) have
sub-processors. Non-compound processors will return an empty list.
Returns:
The list of sub-processors if this is a compound processor.
"""
return self._pipelines
@property
def entry_processors(self) -> List["FrameProcessor"]:
"""Return the list of entry processors for this processor.
Entry processors are the first processors in a compound processor
(e.g. pipelines, parallel pipelines). Note that pipelines can also be an
entry processor as pipelines are processors themselves. Non-compound
processors will simply return an empty list.
Returns:
The list of entry processors.
"""
return self._sources
def processors_with_metrics(self) -> List[FrameProcessor]:
"""Collect processors that can generate metrics from all parallel pipelines.
@@ -145,10 +171,6 @@ class SyncParallelPipeline(BasePipeline):
"""
return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines))
#
# Frame processor
#
async def setup(self, setup: FrameProcessorSetup):
"""Set up the parallel pipeline and all contained processors.
@@ -171,29 +193,23 @@ class SyncParallelPipeline(BasePipeline):
source = SyncParallelPipelineSource(up_queue)
sink = SyncParallelPipelineSink(down_queue)
# Create pipeline
pipeline = Pipeline(processors)
source.link(pipeline)
pipeline.link(sink)
self._pipelines.append(pipeline)
# Keep track of sources and sinks. We also keep the output queue of
# the source and the sinks so we can use it later.
self._sources.append({"processor": source, "queue": down_queue})
self._sinks.append({"processor": sink, "queue": up_queue})
# Create pipeline
pipeline = Pipeline(processors, source=source, sink=sink)
self._pipelines.append(pipeline)
logger.debug(f"Finished creating {self} pipelines")
await asyncio.gather(*[s["processor"].setup(setup) for s in self._sources])
await asyncio.gather(*[p.setup(setup) for p in self._pipelines])
await asyncio.gather(*[s["processor"].setup(setup) for s in self._sinks])
async def cleanup(self):
"""Clean up the parallel pipeline and all contained processors."""
await super().cleanup()
await asyncio.gather(*[s["processor"].cleanup() for s in self._sources])
await asyncio.gather(*[p.cleanup() for p in self._pipelines])
await asyncio.gather(*[s["processor"].cleanup() for s in self._sinks])
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames through all parallel pipelines with synchronization.

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

@@ -41,8 +41,8 @@ from pipecat.frames.frames import (
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 BasePipelineTask, PipelineTaskParams
from pipecat.pipeline.pipeline import Pipeline, PipelineSink, PipelineSource
from pipecat.pipeline.task_observer import TaskObserver
from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup
from pipecat.utils.asyncio.task_manager import (
@@ -101,70 +101,6 @@ class PipelineParams(BaseModel):
start_metadata: Dict[str, Any] = Field(default_factory=dict)
class PipelineTaskSource(FrameProcessor):
"""Source processor for pipeline tasks that handles frame routing.
This is the source processor that is linked at the beginning of the
pipeline given to the pipeline task. It allows us to easily push frames
downstream to the pipeline and also receive upstream frames coming from the
pipeline.
"""
def __init__(self, up_queue: asyncio.Queue):
"""Initialize the pipeline task source.
Args:
up_queue: Queue for upstream frame processing.
**kwargs: Additional arguments passed to the parent class.
"""
super().__init__(enable_direct_mode=True)
self._up_queue = up_queue
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames and route them based on direction.
Args:
frame: The frame to process.
direction: The direction of frame flow.
"""
await super().process_frame(frame, direction)
match direction:
case FrameDirection.UPSTREAM:
await self._up_queue.put(frame)
case FrameDirection.DOWNSTREAM:
await self.push_frame(frame, direction)
class PipelineTaskSink(FrameProcessor):
"""Sink processor for pipeline tasks that handles final frame processing.
This is the sink processor that is linked at the end of the pipeline
given to the pipeline task. It allows us to receive downstream frames and
act on them, for example, waiting to receive an EndFrame.
"""
def __init__(self, down_queue: asyncio.Queue):
"""Initialize the pipeline task sink.
Args:
down_queue: Queue for downstream frame processing.
**kwargs: Additional arguments passed to the parent class.
"""
super().__init__(enable_direct_mode=True)
self._down_queue = down_queue
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames and route them to the downstream queue.
Args:
frame: The frame to process.
direction: The direction of frame flow.
"""
await super().process_frame(frame, direction)
await self._down_queue.put(frame)
class PipelineTask(BasePipelineTask):
"""Manages the execution of a pipeline, handling frame processing and task lifecycle.
@@ -196,7 +132,7 @@ class PipelineTask(BasePipelineTask):
def __init__(
self,
pipeline: BasePipeline,
pipeline: FrameProcessor,
*,
params: Optional[PipelineParams] = None,
additional_span_attributes: Optional[dict] = None,
@@ -244,7 +180,6 @@ class PipelineTask(BasePipelineTask):
will be logged if the watchdog timer is not reset before this timeout.
"""
super().__init__()
self._pipeline = pipeline
self._params = params or PipelineParams()
self._additional_span_attributes = additional_span_attributes or {}
self._cancel_on_idle_timeout = cancel_on_idle_timeout
@@ -288,12 +223,6 @@ class PipelineTask(BasePipelineTask):
# PipelineTask and its frame processors.
self._task_manager = task_manager or TaskManager()
# This queue receives frames coming from the pipeline upstream.
self._up_queue = WatchdogQueue(self._task_manager)
self._process_up_task: Optional[asyncio.Task] = None
# This queue receives frames coming from the pipeline downstream.
self._down_queue = WatchdogQueue(self._task_manager)
self._process_down_task: Optional[asyncio.Task] = None
# This queue is the queue used to push frames to the pipeline.
self._push_queue = WatchdogQueue(self._task_manager)
self._process_push_task: Optional[asyncio.Task] = None
@@ -311,17 +240,13 @@ class PipelineTask(BasePipelineTask):
# StopFrame) has been received in the down queue.
self._pipeline_end_event = asyncio.Event()
# This is a source processor that we connect to the provided
# pipeline. This source processor allows up to receive and react to
# upstream frames.
self._source = PipelineTaskSource(self._up_queue)
self._source.link(pipeline)
# This is a sink processor that we connect to the provided
# pipeline. This sink processor allows up to receive and react to
# downstream frames.
self._sink = PipelineTaskSink(self._down_queue)
pipeline.link(self._sink)
# This is the final pipeline. It is composed of a source processor,
# followed by the user pipeline, and ending with a sink processor. The
# source allows us to receive and react to upstream frames, and the sink
# allows us to receive and react to downstream frames.
source = PipelineSource(self._source_push_frame, name=f"{self}::Source")
sink = PipelineSink(self._sink_push_frame, name=f"{self}::Sink")
self._pipeline = Pipeline([pipeline], source=source, sink=sink)
# The task observer acts as a proxy to the provided observers. This way,
# we only need to pass a single observer (using the StartFrame) which
@@ -494,7 +419,7 @@ class PipelineTask(BasePipelineTask):
# 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())
await self._pipeline.queue_frame(CancelFrame())
# Wait for CancelFrame to make it throught the pipeline.
await self._wait_for_pipeline_end()
# Only cancel the push task, we don't want to be able to process any
@@ -506,12 +431,6 @@ class PipelineTask(BasePipelineTask):
async def _create_tasks(self):
"""Create and start all pipeline processing tasks."""
self._process_up_task = self._task_manager.create_task(
self._process_up_queue(), f"{self}::_process_up_queue"
)
self._process_down_task = self._task_manager.create_task(
self._process_down_queue(), f"{self}::_process_down_queue"
)
self._process_push_task = self._task_manager.create_task(
self._process_push_queue(), f"{self}::_process_push_queue"
)
@@ -545,14 +464,6 @@ class PipelineTask(BasePipelineTask):
await self._task_manager.cancel_task(self._process_push_task)
self._process_push_task = None
if self._process_up_task:
await self._task_manager.cancel_task(self._process_up_task)
self._process_up_task = None
if self._process_down_task:
await self._task_manager.cancel_task(self._process_down_task)
self._process_down_task = None
await self._maybe_cancel_heartbeat_tasks()
await self._maybe_cancel_idle_task()
@@ -606,9 +517,7 @@ class PipelineTask(BasePipelineTask):
observer=self._observer,
watchdog_timers_enabled=self._enable_watchdog_timers,
)
await self._source.setup(setup)
await self._pipeline.setup(setup)
await self._sink.setup(setup)
async def _cleanup(self, cleanup_pipeline: bool):
"""Clean up the pipeline task and processors."""
@@ -620,10 +529,8 @@ class PipelineTask(BasePipelineTask):
self._turn_trace_observer.end_conversation_tracing()
# Cleanup pipeline processors.
await self._source.cleanup()
if cleanup_pipeline:
await self._pipeline.cleanup()
await self._sink.cleanup()
async def _process_push_queue(self):
"""Process frames from the push queue and send them through the pipeline.
@@ -647,16 +554,16 @@ class PipelineTask(BasePipelineTask):
interruption_strategies=self._params.interruption_strategies,
)
start_frame.metadata = self._params.start_metadata
await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM)
await self._pipeline.queue_frame(start_frame)
if self._params.enable_metrics and self._params.send_initial_empty_metrics:
await self._source.queue_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM)
await self._pipeline.queue_frame(self._initial_metrics_frame())
running = True
cleanup_pipeline = True
while running:
frame = await self._push_queue.get()
await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM)
await self._pipeline.queue_frame(frame)
if isinstance(frame, (CancelFrame, EndFrame, StopFrame)):
await self._wait_for_pipeline_end()
running = not isinstance(frame, (CancelFrame, EndFrame, StopFrame))
@@ -664,7 +571,7 @@ class PipelineTask(BasePipelineTask):
self._push_queue.task_done()
await self._cleanup(cleanup_pipeline)
async def _process_up_queue(self):
async def _source_push_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames coming upstream from the pipeline.
This is the task that processes frames coming upstream from the
@@ -672,33 +579,29 @@ class PipelineTask(BasePipelineTask):
pipeline to be stopped (e.g. EndTaskFrame) in which case we would send
an EndFrame down the pipeline.
"""
while True:
frame = await self._up_queue.get()
if isinstance(frame, self._reached_upstream_types):
await self._call_event_handler("on_frame_reached_upstream", frame)
if isinstance(frame, self._reached_upstream_types):
await self._call_event_handler("on_frame_reached_upstream", frame)
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.
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):
# Tell the task we should stop nicely.
await self.queue_frame(StopFrame())
elif isinstance(frame, ErrorFrame):
if frame.fatal:
logger.error(f"A fatal error occurred: {frame}")
# Cancel all tasks downstream.
await self.queue_frame(CancelFrame())
elif isinstance(frame, StopTaskFrame):
# Tell the task we should stop nicely.
await self.queue_frame(StopFrame())
elif isinstance(frame, ErrorFrame):
if frame.fatal:
logger.error(f"A fatal error occurred: {frame}")
# Cancel all tasks downstream.
await self.queue_frame(CancelFrame())
# Tell the task we should stop.
await self.queue_frame(StopTaskFrame())
else:
logger.warning(f"Something went wrong: {frame}")
self._up_queue.task_done()
# Tell the task we should stop.
await self.queue_frame(StopTaskFrame())
else:
logger.warning(f"Something went wrong: {frame}")
async def _process_down_queue(self):
async def _sink_push_frame(self, frame: Frame, direction: FrameDirection):
"""Process frames coming downstream from the pipeline.
This tasks process frames coming downstream from the pipeline. For
@@ -706,34 +609,30 @@ class PipelineTask(BasePipelineTask):
processors have handled the EndFrame and therefore we can exit the task
cleanly.
"""
while True:
frame = await self._down_queue.get()
# Queue received frame to the idle queue so we can monitor idle
# pipelines.
await self._idle_queue.put(frame)
# Queue received frame to the idle queue so we can monitor idle
# pipelines.
await self._idle_queue.put(frame)
if isinstance(frame, self._reached_downstream_types):
await self._call_event_handler("on_frame_reached_downstream", frame)
if isinstance(frame, self._reached_downstream_types):
await self._call_event_handler("on_frame_reached_downstream", frame)
if isinstance(frame, StartFrame):
await self._call_event_handler("on_pipeline_started", frame)
if isinstance(frame, StartFrame):
await self._call_event_handler("on_pipeline_started", frame)
# Start heartbeat tasks now that StartFrame has been processed
# by all processors in the pipeline
self._maybe_start_heartbeat_tasks()
elif isinstance(frame, EndFrame):
await self._call_event_handler("on_pipeline_ended", frame)
self._pipeline_end_event.set()
elif isinstance(frame, StopFrame):
await self._call_event_handler("on_pipeline_stopped", frame)
self._pipeline_end_event.set()
elif isinstance(frame, CancelFrame):
await self._call_event_handler("on_pipeline_cancelled", frame)
self._pipeline_end_event.set()
elif isinstance(frame, HeartbeatFrame):
await self._heartbeat_queue.put(frame)
self._down_queue.task_done()
# Start heartbeat tasks now that StartFrame has been processed
# by all processors in the pipeline
self._maybe_start_heartbeat_tasks()
elif isinstance(frame, EndFrame):
await self._call_event_handler("on_pipeline_ended", frame)
self._pipeline_end_event.set()
elif isinstance(frame, StopFrame):
await self._call_event_handler("on_pipeline_stopped", frame)
self._pipeline_end_event.set()
elif isinstance(frame, CancelFrame):
await self._call_event_handler("on_pipeline_cancelled", frame)
self._pipeline_end_event.set()
elif isinstance(frame, HeartbeatFrame):
await self._heartbeat_queue.put(frame)
async def _heartbeat_push_handler(self):
"""Push heartbeat frames at regular intervals."""
@@ -741,7 +640,7 @@ class PipelineTask(BasePipelineTask):
# 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 self._pipeline.queue_frame(HeartbeatFrame(timestamp=self._clock.get_time()))
await asyncio.sleep(self._params.heartbeats_period_secs)
async def _heartbeat_monitor_handler(self):

35
src/pipecat/pipeline/task_observer.py
View File

@@ -13,11 +13,11 @@ the main pipeline execution.
import asyncio
import inspect
from typing import Dict, List, Optional
from typing import Any, Dict, List, Optional
from attr import dataclass
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
@@ -120,14 +120,21 @@ class TaskObserver(BaseObserver):
for proxy in self._proxies.values():
await self._task_manager.cancel_task(proxy.task)
async def on_process_frame(self, data: FramePushed):
"""Queue frame data for all managed observers.
Args:
data: The frame push event data to distribute to observers.
"""
await self._send_to_proxy(data)
async def on_push_frame(self, data: FramePushed):
"""Queue frame data for all managed observers.
Args:
data: The frame push event data to distribute to observers.
"""
for proxy in self._proxies.values():
await proxy.queue.put(data)
await self._send_to_proxy(data)
def _started(self) -> bool:
"""Check if the task observer has been started."""
@@ -151,6 +158,10 @@ class TaskObserver(BaseObserver):
proxies[observer] = proxy
return proxies
async def _send_to_proxy(self, data: Any):
for proxy in self._proxies.values():
await proxy.queue.put(data)
async def _proxy_task_handler(self, queue: asyncio.Queue, observer: BaseObserver):
"""Handle frame processing for a single observer."""
on_push_frame_deprecated = False
@@ -169,11 +180,15 @@ class TaskObserver(BaseObserver):
while True:
data = await queue.get()
if on_push_frame_deprecated:
await observer.on_push_frame(
data.src, data.dst, data.frame, data.direction, data.timestamp
)
else:
await observer.on_push_frame(data)
if isinstance(data, FramePushed):
if on_push_frame_deprecated:
await observer.on_push_frame(
data.src, data.dst, data.frame, data.direction, data.timestamp
)
else:
await observer.on_push_frame(data)
elif isinstance(data, FrameProcessed):
await observer.on_process_frame(data)
queue.task_done()

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

@@ -34,14 +34,11 @@ from pipecat.frames.frames import (
SystemFrame,
)
from pipecat.metrics.metrics import LLMTokenUsage, MetricsData
from pipecat.observers.base_observer import BaseObserver, FramePushed
from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.asyncio.watchdog_event import WatchdogEvent
from pipecat.utils.asyncio.watchdog_priority_queue import (
WatchdogPriorityCancelSentinel,
WatchdogPriorityQueue,
)
from pipecat.utils.asyncio.watchdog_priority_queue import WatchdogPriorityQueue
from pipecat.utils.asyncio.watchdog_queue import WatchdogQueue
from pipecat.utils.base_object import BaseObject
@@ -168,8 +165,7 @@ class FrameProcessor(BaseObject):
watchdog_timeout_secs: Timeout in seconds for watchdog operations.
**kwargs: Additional arguments passed to parent class.
"""
super().__init__(name=name)
self._parent: Optional["FrameProcessor"] = None
super().__init__(name=name, **kwargs)
self._prev: Optional["FrameProcessor"] = None
self._next: Optional["FrameProcessor"] = None
@@ -226,6 +222,8 @@ class FrameProcessor(BaseObject):
# The input task that handles all types of frames. It processes system
# frames right away and queues non-system frames for later processing.
self.__should_block_system_frames = False
self.__input_event: Optional[asyncio.Event] = None
self.__input_frame_task: Optional[asyncio.Task] = None
# The process task processes non-system frames. Non-system frames will
@@ -234,6 +232,7 @@ class FrameProcessor(BaseObject):
# called. To resume processing frames we need to call
# `resume_processing_frames()` which will wake up the event.
self.__should_block_frames = False
self.__process_event: Optional[asyncio.Event] = None
self.__process_frame_task: Optional[asyncio.Task] = None
@property
@@ -254,6 +253,50 @@ class FrameProcessor(BaseObject):
"""
return self._name
@property
def processors(self) -> List["FrameProcessor"]:
"""Return the list of sub-processors contained within this processor.
Only compound processors (e.g. pipelines and parallel pipelines) have
sub-processors. Non-compound processors will return an empty list.
Returns:
The list of sub-processors if this is a compound processor.
"""
return []
@property
def entry_processors(self) -> List["FrameProcessor"]:
"""Return the list of entry processors for this processor.
Entry processors are the first processors in a compound processor
(e.g. pipelines, parallel pipelines). Note that pipelines can also be an
entry processor as pipelines are processors themselves. Non-compound
processors will simply return an empty list.
Returns:
The list of entry processors.
"""
return []
@property
def next(self) -> Optional["FrameProcessor"]:
"""Get the next processor.
Returns:
The next processor, or None if there's no next processor.
"""
return self._next
@property
def previous(self) -> Optional["FrameProcessor"]:
"""Get the previous processor.
Returns:
The previous processor, or None if there's no previous processor.
"""
return self._prev
@property
def interruptions_allowed(self):
"""Check if interruptions are allowed for this processor.
@@ -313,6 +356,17 @@ class FrameProcessor(BaseObject):
raise Exception(f"{self} TaskManager is still not initialized.")
return self._task_manager
def processors_with_metrics(self):
"""Return processors that can generate metrics.
Recursively collects all processors that support metrics generation,
including those from nested processors.
Returns:
List of frame processors that can generate metrics.
"""
return []
def can_generate_metrics(self) -> bool:
"""Check if this processor can generate metrics.
@@ -482,30 +536,6 @@ class FrameProcessor(BaseObject):
processor._prev = self
logger.debug(f"Linking {self} -> {self._next}")
def get_event_loop(self) -> asyncio.AbstractEventLoop:
"""Get the event loop used by this processor.
Returns:
The asyncio event loop.
"""
return self.task_manager.get_event_loop()
def set_parent(self, parent: "FrameProcessor"):
"""Set the parent processor for this processor.
Args:
parent: The parent processor.
"""
self._parent = parent
def get_parent(self) -> Optional["FrameProcessor"]:
"""Get the parent processor.
Returns:
The parent processor, or None if no parent is set.
"""
return self._parent
def get_clock(self) -> BaseClock:
"""Get the clock used by this processor.
@@ -519,6 +549,14 @@ class FrameProcessor(BaseObject):
raise Exception(f"{self} Clock is still not initialized.")
return self._clock
def get_event_loop(self) -> asyncio.AbstractEventLoop:
"""Get the event loop used by this processor.
Returns:
The asyncio event loop.
"""
return self.task_manager.get_event_loop()
async def queue_frame(
self,
frame: Frame,
@@ -546,12 +584,23 @@ class FrameProcessor(BaseObject):
logger.trace(f"{self}: pausing frame processing")
self.__should_block_frames = True
async def pause_processing_system_frames(self):
"""Pause processing of queued system frames."""
logger.trace(f"{self}: pausing system frame processing")
self.__should_block_system_frames = True
async def resume_processing_frames(self):
"""Resume processing of queued frames."""
logger.trace(f"{self}: resuming frame processing")
if self.__process_event:
self.__process_event.set()
async def resume_processing_system_frames(self):
"""Resume processing of queued system frames."""
logger.trace(f"{self}: resuming system frame processing")
if self.__input_event:
self.__input_event.set()
async def process_frame(self, frame: Frame, direction: FrameDirection):
"""Process a frame.
@@ -559,6 +608,16 @@ class FrameProcessor(BaseObject):
frame: The frame to process.
direction: The direction of frame flow.
"""
if self._observer:
timestamp = self._clock.get_time() if self._clock else 0
data = FrameProcessed(
processor=self,
frame=frame,
direction=direction,
timestamp=timestamp,
)
await self._observer.on_process_frame(data)
if isinstance(frame, StartFrame):
await self.__start(frame)
elif isinstance(frame, StartInterruptionFrame):
@@ -715,6 +774,7 @@ class FrameProcessor(BaseObject):
return
if not self.__input_frame_task:
self.__input_event = WatchdogEvent(self.task_manager)
self.__input_queue = FrameProcessorQueue(self.task_manager)
self.__input_frame_task = self.create_task(self.__input_frame_task_handler())
@@ -764,6 +824,13 @@ class FrameProcessor(BaseObject):
"""
while True:
if self.__should_block_system_frames and self.__input_event:
logger.trace(f"{self}: system frame processing paused")
await self.__input_event.wait()
self.__input_event.clear()
self.__should_block_system_frames = False
logger.trace(f"{self}: system frame processing resumed")
(frame, direction, callback) = await self.__input_queue.get()
if isinstance(frame, SystemFrame):
@@ -780,7 +847,7 @@ class FrameProcessor(BaseObject):
async def __process_frame_task_handler(self):
"""Handle non-system frames from the process queue."""
while True:
if self.__should_block_frames:
if self.__should_block_frames and self.__process_event:
logger.trace(f"{self}: frame processing paused")
await self.__process_event.wait()
self.__process_event.clear()

4
src/pipecat/tests/utils.py
View File

@@ -36,7 +36,7 @@ class SleepFrame(SystemFrame):
sleep: Duration to sleep in seconds before processing the next frame.
"""
sleep: float = 0.1
sleep: float = 0.2
class HeartbeatsObserver(BaseObserver):
@@ -100,7 +100,7 @@ class QueuedFrameProcessor(FrameProcessor):
queue_direction: The direction of frames to capture (UPSTREAM or DOWNSTREAM).
ignore_start: Whether to ignore StartFrames when capturing.
"""
super().__init__()
super().__init__(enable_direct_mode=True)
self._queue = queue
self._queue_direction = queue_direction
self._ignore_start = ignore_start

8
tests/test_context_aggregators.py
View File

@@ -102,8 +102,8 @@ class BaseTestUserContextAggregator:
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
*self.EXPECTED_CONTEXT_FRAMES,
UserStoppedSpeakingFrame,
]
await run_test(
aggregator,
@@ -127,8 +127,8 @@ class BaseTestUserContextAggregator:
]
expected_down_frames = [
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
*self.EXPECTED_CONTEXT_FRAMES,
UserStoppedSpeakingFrame,
]
await run_test(
aggregator,
@@ -158,8 +158,8 @@ class BaseTestUserContextAggregator:
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
*self.EXPECTED_CONTEXT_FRAMES,
UserStoppedSpeakingFrame,
]
await run_test(
aggregator,
@@ -298,8 +298,8 @@ class BaseTestUserContextAggregator:
expected_down_frames = [
SpeechControlParamsFrame,
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
*self.EXPECTED_CONTEXT_FRAMES,
UserStoppedSpeakingFrame,
*self.EXPECTED_CONTEXT_FRAMES,
]
await run_test(

10
tests/test_dtmf_aggregator.py
View File

@@ -53,9 +53,9 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
InputDTMFFrame(button=KeypadEntry.ONE),
InputDTMFFrame(button=KeypadEntry.TWO),
SleepFrame(sleep=0.2), # This should trigger timeout
SleepFrame(), # This should trigger timeout
InputDTMFFrame(button=KeypadEntry.THREE),
SleepFrame(sleep=0.2), # This should trigger another timeout
SleepFrame(), # This should trigger another timeout
]
expected_down_frames = [
InputDTMFFrame,
@@ -86,10 +86,10 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
InputDTMFFrame(button=KeypadEntry.ONE),
InputDTMFFrame(button=KeypadEntry.TWO),
InputDTMFFrame(button=KeypadEntry.POUND), # First sequence
SleepFrame(sleep=0.1),
SleepFrame(),
InputDTMFFrame(button=KeypadEntry.FOUR),
InputDTMFFrame(button=KeypadEntry.FIVE),
SleepFrame(sleep=0.3), # Second sequence via timeout
SleepFrame(), # Second sequence via timeout
]
expected_down_frames = [
InputDTMFFrame,
@@ -120,7 +120,7 @@ class TestDTMFAggregator(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
InputDTMFFrame(button=KeypadEntry.ONE),
InputDTMFFrame(button=KeypadEntry.TWO),
SleepFrame(sleep=0.1), # Allow time for aggregation
SleepFrame(), # Allow time for aggregation
EndFrame(),
]
expected_down_frames = [

4
tests/test_stt_mute_filter.py
View File

@@ -124,14 +124,14 @@ class TestSTTMuteFilter(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
# Bot speaking - should mute
BotStartedSpeakingFrame(),
SleepFrame(sleep=0.1), # Wait for StartedSpeaking to process
SleepFrame(), # Wait for StartedSpeaking to process
InterimTranscriptionFrame(
user_id="user1", text="This should be suppressed", timestamp="1234567890"
),
TranscriptionFrame(
user_id="user1", text="This should be suppressed", timestamp="1234567890"
),
SleepFrame(sleep=0.1), # Wait for transcription frames to queue
SleepFrame(), # Wait for transcription frames to queue
BotStoppedSpeakingFrame(),
# Bot not speaking - should pass through
InterimTranscriptionFrame(

30
tests/test_transcript_processor.py
View File

@@ -129,13 +129,13 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
# Create test frames simulating bot speaking multiple text chunks
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(sleep=0.1), # Wait for StartedSpeaking to process
SleepFrame(), # Wait for StartedSpeaking to process
TTSTextFrame(text="Hello"),
TTSTextFrame(text="world!"),
TTSTextFrame(text="How"),
TTSTextFrame(text="are"),
TTSTextFrame(text="you?"),
SleepFrame(sleep=0.1), # Wait for text frames to queue
SleepFrame(), # Wait for text frames to queue
BotStoppedSpeakingFrame(),
]
@@ -151,8 +151,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
TTSTextFrame,
TTSTextFrame,
TTSTextFrame,
BotStoppedSpeakingFrame,
TranscriptionUpdateFrame,
BotStoppedSpeakingFrame,
]
# Run test
@@ -179,7 +179,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
self.assertIsNotNone(message.timestamp)
# All frames should be passed through in order, with update at end
downstream_update = cast(TranscriptionUpdateFrame, received_frames[-1])
downstream_update = cast(TranscriptionUpdateFrame, received_frames[-2])
self.assertEqual(downstream_update.messages[0].content, "Hello world! How are you?")
async def test_empty_text_handling(self):
@@ -194,7 +194,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(sleep=0.1),
SleepFrame(),
TTSTextFrame(text=""), # Empty text
TTSTextFrame(text=" "), # Just whitespace
TTSTextFrame(text="\n"), # Just newline
@@ -234,16 +234,16 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
# Simulate bot being interrupted mid-sentence
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(sleep=0.1),
SleepFrame(),
TTSTextFrame(text="Hello"),
TTSTextFrame(text="world!"),
SleepFrame(sleep=0.1),
SleepFrame(),
StartInterruptionFrame(), # User interrupts here
SleepFrame(sleep=0.1),
SleepFrame(),
BotStartedSpeakingFrame(),
TTSTextFrame(text="New"),
TTSTextFrame(text="response"),
SleepFrame(sleep=0.1),
SleepFrame(),
BotStoppedSpeakingFrame(),
]
@@ -257,8 +257,8 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
BotStartedSpeakingFrame,
TTSTextFrame, # "New"
TTSTextFrame, # "response"
BotStoppedSpeakingFrame,
TranscriptionUpdateFrame, # Second message
BotStoppedSpeakingFrame,
]
# Run test
@@ -298,7 +298,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(sleep=0.1),
SleepFrame(),
TTSTextFrame(text="Hello"),
TTSTextFrame(text="world"),
# Pipeline ends here; run_test will automatically send EndFrame
@@ -337,10 +337,10 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(sleep=0.1),
SleepFrame(),
TTSTextFrame(text="Hello"),
TTSTextFrame(text="world"),
SleepFrame(sleep=0.1), # Ensure messages are processed
SleepFrame(), # Ensure messages are processed
CancelFrame(),
]
@@ -400,7 +400,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
# Test assistant processor
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(sleep=0.1),
SleepFrame(),
TTSTextFrame(text="Assistant"),
TTSTextFrame(text="message"),
BotStoppedSpeakingFrame(),
@@ -440,7 +440,7 @@ class TestUserTranscriptProcessor(unittest.IsolatedAsyncioTestCase):
# Test the specific pattern shared
frames_to_send = [
BotStartedSpeakingFrame(),
SleepFrame(sleep=0.1),
SleepFrame(),
TTSTextFrame(text="Hello"),
TTSTextFrame(text=" there"),
TTSTextFrame(text="!"),