From d5ebc883b3571013a3262069ffe021434de397dd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sat, 17 May 2025 22:46:22 -0700 Subject: [PATCH] PipelineTask: add new started/stopped/ended/cancelled events --- CHANGELOG.md | 7 ++++- src/pipecat/pipeline/task.py | 35 ++++++++++++++++++++++-- tests/test_pipeline.py | 53 ++++++++++++++++++++++++++++++++++-- 3 files changed, 90 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index e01359589..a0bf5cc8f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,8 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added new `PipelineTask` event handlers `on_pipeline_started`, + `on_pipeline_stopped`, `on_pipeline_ended` and `on_pipeline_cancelled`, which + correspond to the `StartFrame`, `StopFrame`, `EndFrame` and `CancelFrame` + respectively. + - Added additional languages to `LmntTTSService`. Languages include: `hi`, `id`, - `it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`. + `it`, `ja`, `nl`, `pl`, `ru`, `sv`, `th`, `tr`, `uk`, `vi`. - Added a `model` parameter to the `LmntTTSService` constructor, allowing switching between LMNT models. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index b45999784..2b90d508d 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -144,7 +144,26 @@ class PipelineTask(BaseTask): `LLMFullResponseEndFrame` are received within `idle_timeout_secs`. @task.event_handler("on_idle_timeout") - async def on_idle_timeout(task): + async def on_pipeline_idle_timeout(task): + ... + + There are also events to know if a pipeline has been started, stopped, ended + or cancelled. + + @task.event_handler("on_pipeline_started") + async def on_pipeline_started(task, frame: StartFrame): + ... + + @task.event_handler("on_pipeline_stopped") + async def on_pipeline_stopped(task, frame: StopFrame): + ... + + @task.event_handler("on_pipeline_ended") + async def on_pipeline_ended(task, frame: EndFrame): + ... + + @task.event_handler("on_pipeline_cancelled") + async def on_pipeline_cancelled(task, frame: CancelFrame): ... Args: @@ -264,6 +283,10 @@ class PipelineTask(BaseTask): self._register_event_handler("on_frame_reached_upstream") self._register_event_handler("on_frame_reached_downstream") self._register_event_handler("on_idle_timeout") + self._register_event_handler("on_pipeline_started") + self._register_event_handler("on_pipeline_stopped") + self._register_event_handler("on_pipeline_ended") + self._register_event_handler("on_pipeline_cancelled") @property def params(self) -> PipelineParams: @@ -552,8 +575,16 @@ class PipelineTask(BaseTask): if isinstance(frame, self._reached_downstream_types): await self._call_event_handler("on_frame_reached_downstream", frame) - if isinstance(frame, (EndFrame, StopFrame)): + if isinstance(frame, StartFrame): + await self._call_event_handler("on_pipeline_started", frame) + 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) elif isinstance(frame, HeartbeatFrame): await self._heartbeat_queue.put(frame) self._down_queue.task_done() diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 46114b5b2..e02f4012d 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -8,7 +8,13 @@ import asyncio import time import unittest -from pipecat.frames.frames import EndFrame, HeartbeatFrame, StartFrame, TextFrame +from pipecat.frames.frames import ( + EndFrame, + HeartbeatFrame, + StartFrame, + StopFrame, + TextFrame, +) from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -95,7 +101,50 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): await task.run() assert task.has_finished() - async def test_task_event_handlers(self): + async def test_task_started_ended_event_handler(self): + start_received = False + end_received = False + + identity = IdentityFilter() + pipeline = Pipeline([identity]) + task = PipelineTask(pipeline) + task.set_event_loop(asyncio.get_event_loop()) + + @task.event_handler("on_pipeline_started") + async def on_pipeline_started(task, frame: StartFrame): + nonlocal start_received + start_received = True + + @task.event_handler("on_pipeline_ended") + async def on_pipeline_ended(task, frame: EndFrame): + nonlocal end_received + end_received = True + + await task.queue_frame(EndFrame()) + await task.run() + + assert start_received + assert end_received + + async def test_task_stopped_event_handler(self): + stop_received = False + + identity = IdentityFilter() + pipeline = Pipeline([identity]) + task = PipelineTask(pipeline) + task.set_event_loop(asyncio.get_event_loop()) + + @task.event_handler("on_pipeline_stopped") + async def on_pipeline_ended(task, frame: StopFrame): + nonlocal stop_received + stop_received = True + + await task.queue_frame(StopFrame()) + await task.run() + + assert stop_received + + async def test_task_frame_reached_event_handlers(self): upstream_received = False downstream_received = False