From f024476b08e5cf9368724678c12e0a0a42132f0e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 28 Jan 2025 11:22:01 -0800 Subject: [PATCH 1/4] transports(daily): queue events until join completes --- CHANGELOG.md | 7 +++++ src/pipecat/transports/services/daily.py | 33 ++++++++++++------------ 2 files changed, 24 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6afd15aa7..6886f37d7 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,13 @@ All notable changes to **Pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Fixed + +- Fixed a `DailyTransport` issue that would cause events to be triggered before + join finished. + ## [0.0.54] - 2025-01-27 ### Added diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 6d32786d9..099b78aa8 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -140,7 +140,6 @@ class DailyCallbacks(BaseModel): on_dialout_stopped: Callable[[Any], Awaitable[None]] on_dialout_error: Callable[[Any], Awaitable[None]] on_dialout_warning: Callable[[Any], Awaitable[None]] - on_first_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]] on_participant_joined: Callable[[Mapping[str, Any]], Awaitable[None]] on_participant_left: Callable[[Mapping[str, Any], str], Awaitable[None]] on_participant_updated: Callable[[Mapping[str, Any]], Awaitable[None]] @@ -201,9 +200,9 @@ class DailyTransportClient(EventHandler): self._video_renderers = {} self._transcription_ids = [] self._transcription_status = None - self._other_participant_has_joined = False self._joined = False + self._joined_event = asyncio.Event() self._leave_counter = 0 # We use the executor to cleanup the client. We just do it from one @@ -353,6 +352,8 @@ class DailyTransportClient(EventHandler): await self._start_transcription() await self._callbacks.on_joined(data) + + self._joined_event.set() else: error_msg = f"Error joining {self._room_url}: {error}" logger.error(error_msg) @@ -441,6 +442,7 @@ class DailyTransportClient(EventHandler): return self._joined = False + self._joined_event.clear() logger.info(f"Leaving {self._room_url}") @@ -629,19 +631,9 @@ class DailyTransportClient(EventHandler): self._call_async_callback(self._callbacks.on_dialout_warning, data) def on_participant_joined(self, participant): - id = participant["id"] - logger.info(f"Participant joined {id}") - - if not self._other_participant_has_joined: - self._other_participant_has_joined = True - self._call_async_callback(self._callbacks.on_first_participant_joined, participant) - self._call_async_callback(self._callbacks.on_participant_joined, participant) def on_participant_left(self, participant, reason): - id = participant["id"] - logger.info(f"Participant left {id}") - self._call_async_callback(self._callbacks.on_participant_left, participant, reason) def on_participant_updated(self, participant): @@ -695,6 +687,8 @@ class DailyTransportClient(EventHandler): async def _callback_task_handler(self): while True: + # Wait to process any callback until we are joined. + await self._joined_event.wait() (callback, *args) = await self._callback_queue.get() await callback(*args) @@ -901,7 +895,6 @@ class DailyTransport(BaseTransport): on_dialout_stopped=self._on_dialout_stopped, on_dialout_error=self._on_dialout_error, on_dialout_warning=self._on_dialout_warning, - on_first_participant_joined=self._on_first_participant_joined, on_participant_joined=self._on_participant_joined, on_participant_left=self._on_participant_left, on_participant_updated=self._on_participant_updated, @@ -918,6 +911,8 @@ class DailyTransport(BaseTransport): self._input: DailyInputTransport | None = None self._output: DailyOutputTransport | None = None + self._other_participant_has_joined = False + # Register supported handlers. The user will only be able to register # these handlers. self._register_event_handler("on_joined") @@ -1124,17 +1119,23 @@ class DailyTransport(BaseTransport): await self._call_event_handler("on_dialout_warning", data) async def _on_participant_joined(self, participant): + id = participant["id"] + logger.info(f"Participant joined {id}") + + if not self._other_participant_has_joined: + self._other_participant_has_joined = True + await self._call_event_handler("on_first_participant_joined", participant) + await self._call_event_handler("on_participant_joined", participant) async def _on_participant_left(self, participant, reason): + id = participant["id"] + logger.info(f"Participant left {id}") await self._call_event_handler("on_participant_left", participant, reason) async def _on_participant_updated(self, participant): await self._call_event_handler("on_participant_updated", participant) - async def _on_first_participant_joined(self, participant): - await self._call_event_handler("on_first_participant_joined", participant) - async def _on_transcription_message(self, message): await self._call_event_handler("on_transcription_message", message) From ba358a4f0acd98f51e3ecb5ef473ddaae3b8dc91 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 28 Jan 2025 15:01:20 -0800 Subject: [PATCH 2/4] task: cleanup processors after task finishes running --- CHANGELOG.md | 3 +++ src/pipecat/pipeline/task.py | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6886f37d7..8c26602d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed a `DailyTransport` issue that would cause events to be triggered before join finished. +- Fixed a `PipelineTask` issue that was preventing processors to be cleaned up + after cancelling the task. + ## [0.0.54] - 2025-01-27 ### Added diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 45596980f..9f59e8177 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -160,7 +160,6 @@ class PipelineTask(BaseTask): await self._source.push_frame(CancelFrame()) # Only cancel the push task. Everything else will be cancelled in run(). await cancel_task(self._process_push_task) - await self._cleanup() async def run(self): """ @@ -176,6 +175,7 @@ class PipelineTask(BaseTask): # awaiting a task. pass await self._cancel_tasks() + await self._cleanup() self._finished = True async def queue_frame(self, frame: Frame): From 3fe212431418fc2dae1a9f862d35e7390b4ddd71 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 28 Jan 2025 15:02:37 -0800 Subject: [PATCH 3/4] examples: use task.cancel() when participant leaves or disconnects --- CHANGELOG.md | 8 +++++++ README.md | 4 ++-- examples/canonical-metrics/bot.py | 4 +++- examples/chatbot-audio-recording/bot.py | 3 +-- examples/deployment/flyio-example/bot.py | 4 +++- examples/deployment/modal-example/bot.py | 21 +++++++++---------- examples/foundational/03-still-frame.py | 4 ++-- .../foundational/06-listen-and-respond.py | 4 ++-- examples/foundational/06a-image-sync.py | 3 +-- examples/foundational/07-interruptible-vad.py | 3 +-- examples/foundational/07-interruptible.py | 3 +-- .../07a-interruptible-anthropic.py | 3 +-- .../07b-interruptible-langchain.py | 4 ++-- .../07c-interruptible-deepgram-vad.py | 3 +-- .../07c-interruptible-deepgram.py | 3 +-- .../07d-interruptible-elevenlabs.py | 3 +-- .../07e-interruptible-playht-http.py | 4 +--- .../foundational/07e-interruptible-playht.py | 3 +-- .../foundational/07f-interruptible-azure.py | 3 +-- .../07g-interruptible-openai-tts.py | 3 +-- .../07h-interruptible-openpipe.py | 3 +-- .../foundational/07i-interruptible-xtts.py | 3 +-- .../foundational/07j-interruptible-gladia.py | 3 +-- .../foundational/07k-interruptible-lmnt.py | 3 +-- .../07l-interruptible-together.py | 3 +-- .../foundational/07m-interruptible-polly.py | 3 +-- .../foundational/07n-interruptible-google.py | 3 +-- .../07o-interruptible-assemblyai.py | 3 +-- .../foundational/07p-interruptible-krisp.py | 3 +-- .../foundational/07q-interruptible-rime.py | 3 +-- .../07r-interruptible-riva-nim.py | 3 +-- .../07s-interruptible-google-audio-in.py | 3 +-- .../foundational/07t-interruptible-fish.py | 3 +-- examples/foundational/21-tavus-layer.py | 3 +-- .../26d-gemini-multimodal-live-text.py | 3 +-- .../28a-transcription-processor-openai.py | 8 ++----- .../28b-transcript-processor-anthropic.py | 8 ++----- .../28c-transcription-processor-gemini.py | 8 ++----- examples/foundational/30-observer.py | 3 +-- examples/news-chatbot/server/news_bot.py | 6 +++--- examples/phone-chatbot/bot_daily.py | 3 +-- examples/phone-chatbot/bot_twilio.py | 3 +-- examples/simple-chatbot/server/bot-gemini.py | 3 +-- examples/simple-chatbot/server/bot-openai.py | 3 +-- examples/storytelling-chatbot/src/bot.py | 14 +++++-------- examples/studypal/studypal.py | 3 +-- examples/translation-chatbot/bot.py | 4 +--- examples/twilio-chatbot/bot.py | 3 +-- 48 files changed, 81 insertions(+), 123 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8c26602d8..46ac58a8c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,14 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed a `PipelineTask` issue that was preventing processors to be cleaned up after cancelling the task. +### Other + +- Updated all examples to use `task.cancel()` instead of pushing an `EndFrame` + when a participant leaves/disconnects. If you push an `EndFrame` this will + cause the bot to run through everything that is internally queued (which could + take seconds). Instead, if a participant disconnects there is nothing else to + be sent and therefore we should stop immediately. + ## [0.0.54] - 2025-01-27 ### Added diff --git a/README.md b/README.md index 44cab0367..5a915041d 100644 --- a/README.md +++ b/README.md @@ -81,7 +81,7 @@ Here is a very basic Pipecat bot that greets a user when they join a real-time s ```python import asyncio -from pipecat.frames.frames import EndFrame, TextFrame +from pipecat.frames.frames import TextFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineTask from pipecat.pipeline.runner import PipelineRunner @@ -122,7 +122,7 @@ async def main(): # Register an event handler to exit the application when the user leaves. @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() # Run the pipeline task await runner.run(task) diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index 33a82b4fd..945792b25 100644 --- a/examples/canonical-metrics/bot.py +++ b/examples/canonical-metrics/bot.py @@ -130,11 +130,13 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): print(f"Participant left: {participant}") - await task.queue_frame(EndFrame()) + await task.cancel() @transport.event_handler("on_call_state_updated") async def on_call_state_updated(transport, state): if state == "left": + # Here we don't want to cancel, we just want to finish sending + # whatever is queued, so we use an EndFrame(). await task.queue_frame(EndFrame()) runner = PipelineRunner() diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 2614fcf6b..7cf750c62 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -18,7 +18,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -139,7 +138,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): print(f"Participant left: {participant}") - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/deployment/flyio-example/bot.py b/examples/deployment/flyio-example/bot.py index 57f64889e..b74e01231 100644 --- a/examples/deployment/flyio-example/bot.py +++ b/examples/deployment/flyio-example/bot.py @@ -79,11 +79,13 @@ async def main(room_url: str, token: str): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() @transport.event_handler("on_call_state_updated") async def on_call_state_updated(transport, state): if state == "left": + # Here we don't want to cancel, we just want to finish sending + # whatever is queued, so we use an EndFrame(). await task.queue_frame(EndFrame()) runner = PipelineRunner() diff --git a/examples/deployment/modal-example/bot.py b/examples/deployment/modal-example/bot.py index 39d7c9573..7855caeb8 100644 --- a/examples/deployment/modal-example/bot.py +++ b/examples/deployment/modal-example/bot.py @@ -5,6 +5,15 @@ import sys from dotenv import load_dotenv from loguru import logger +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport + load_dotenv(override=True) logger.remove(0) @@ -12,16 +21,6 @@ logger.add(sys.stderr, level="DEBUG") async def main(room_url: str, token: str): - from pipecat.audio.vad.silero import SileroVADAnalyzer - from pipecat.frames.frames import EndFrame - from pipecat.pipeline.pipeline import Pipeline - from pipecat.pipeline.runner import PipelineRunner - from pipecat.pipeline.task import PipelineParams, PipelineTask - from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext - from pipecat.services.cartesia import CartesiaTTSService - from pipecat.services.openai import OpenAILLMService - from pipecat.transports.services.daily import DailyParams, DailyTransport - transport = DailyTransport( room_url, token, @@ -79,7 +78,7 @@ async def main(room_url: str, token: str): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/03-still-frame.py b/examples/foundational/03-still-frame.py index 830d4c6bf..455371ab4 100644 --- a/examples/foundational/03-still-frame.py +++ b/examples/foundational/03-still-frame.py @@ -13,7 +13,7 @@ from dotenv import load_dotenv from loguru import logger from runner import configure -from pipecat.frames.frames import EndFrame, TextFrame +from pipecat.frames.frames import TextFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask @@ -53,7 +53,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() await runner.run(task) diff --git a/examples/foundational/06-listen-and-respond.py b/examples/foundational/06-listen-and-respond.py index 58a763c09..60550880a 100644 --- a/examples/foundational/06-listen-and-respond.py +++ b/examples/foundational/06-listen-and-respond.py @@ -14,7 +14,7 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame, Frame, MetricsFrame +from pipecat.frames.frames import Frame, MetricsFrame from pipecat.metrics.metrics import ( LLMUsageMetricsData, ProcessingMetricsData, @@ -115,7 +115,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/06a-image-sync.py b/examples/foundational/06a-image-sync.py index 2cd56ebc9..841c335f6 100644 --- a/examples/foundational/06a-image-sync.py +++ b/examples/foundational/06a-image-sync.py @@ -18,7 +18,6 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, - EndFrame, Frame, OutputImageRawFrame, TextFrame, @@ -144,7 +143,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07-interruptible-vad.py b/examples/foundational/07-interruptible-vad.py index 3dc82d870..1aee662c3 100644 --- a/examples/foundational/07-interruptible-vad.py +++ b/examples/foundational/07-interruptible-vad.py @@ -13,7 +13,6 @@ from dotenv import load_dotenv from loguru import logger from runner import configure -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -94,7 +93,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07-interruptible.py b/examples/foundational/07-interruptible.py index 05c467257..bb3965159 100644 --- a/examples/foundational/07-interruptible.py +++ b/examples/foundational/07-interruptible.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -92,7 +91,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07a-interruptible-anthropic.py b/examples/foundational/07a-interruptible-anthropic.py index 18d654ff1..62e1186b6 100644 --- a/examples/foundational/07a-interruptible-anthropic.py +++ b/examples/foundational/07a-interruptible-anthropic.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -96,7 +95,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07b-interruptible-langchain.py b/examples/foundational/07b-interruptible-langchain.py index 9f31a831e..9c093f233 100644 --- a/examples/foundational/07b-interruptible-langchain.py +++ b/examples/foundational/07b-interruptible-langchain.py @@ -19,7 +19,7 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame, LLMMessagesFrame +from pipecat.frames.frames import LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -124,7 +124,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07c-interruptible-deepgram-vad.py b/examples/foundational/07c-interruptible-deepgram-vad.py index c2b6498ed..831accfd9 100644 --- a/examples/foundational/07c-interruptible-deepgram-vad.py +++ b/examples/foundational/07c-interruptible-deepgram-vad.py @@ -16,7 +16,6 @@ from runner import configure from pipecat.frames.frames import ( BotInterruptionFrame, - EndFrame, StopInterruptionFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, @@ -106,7 +105,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07c-interruptible-deepgram.py b/examples/foundational/07c-interruptible-deepgram.py index d85fb4e0c..fc805ed56 100644 --- a/examples/foundational/07c-interruptible-deepgram.py +++ b/examples/foundational/07c-interruptible-deepgram.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -91,7 +90,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07d-interruptible-elevenlabs.py b/examples/foundational/07d-interruptible-elevenlabs.py index d6526e55f..dd2e4735f 100644 --- a/examples/foundational/07d-interruptible-elevenlabs.py +++ b/examples/foundational/07d-interruptible-elevenlabs.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -92,7 +91,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07e-interruptible-playht-http.py b/examples/foundational/07e-interruptible-playht-http.py index 986377c1b..17c49d1be 100644 --- a/examples/foundational/07e-interruptible-playht-http.py +++ b/examples/foundational/07e-interruptible-playht-http.py @@ -14,14 +14,12 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.openai import OpenAILLMService from pipecat.services.playht import PlayHTHttpTTSService -from pipecat.transcriptions.language import Language from pipecat.transports.services.daily import DailyParams, DailyTransport load_dotenv(override=True) @@ -94,7 +92,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07e-interruptible-playht.py b/examples/foundational/07e-interruptible-playht.py index 057b9b973..fa4c5d6d4 100644 --- a/examples/foundational/07e-interruptible-playht.py +++ b/examples/foundational/07e-interruptible-playht.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -95,7 +94,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07f-interruptible-azure.py b/examples/foundational/07f-interruptible-azure.py index 439aa348b..9f66a2811 100644 --- a/examples/foundational/07f-interruptible-azure.py +++ b/examples/foundational/07f-interruptible-azure.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -101,7 +100,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07g-interruptible-openai-tts.py b/examples/foundational/07g-interruptible-openai-tts.py index 69952c394..78a23b65f 100644 --- a/examples/foundational/07g-interruptible-openai-tts.py +++ b/examples/foundational/07g-interruptible-openai-tts.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -89,7 +88,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07h-interruptible-openpipe.py b/examples/foundational/07h-interruptible-openpipe.py index 8729f87fa..0084f9b64 100644 --- a/examples/foundational/07h-interruptible-openpipe.py +++ b/examples/foundational/07h-interruptible-openpipe.py @@ -15,7 +15,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -99,7 +98,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07i-interruptible-xtts.py b/examples/foundational/07i-interruptible-xtts.py index cc74723d9..18980afe9 100644 --- a/examples/foundational/07i-interruptible-xtts.py +++ b/examples/foundational/07i-interruptible-xtts.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -93,7 +92,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07j-interruptible-gladia.py b/examples/foundational/07j-interruptible-gladia.py index 0aa5f410d..b4405ebf7 100644 --- a/examples/foundational/07j-interruptible-gladia.py +++ b/examples/foundational/07j-interruptible-gladia.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -99,7 +98,7 @@ async def main(): # Register an event handler to exit the application when the user leaves. @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07k-interruptible-lmnt.py b/examples/foundational/07k-interruptible-lmnt.py index 2a7311c7e..4cfd76f24 100644 --- a/examples/foundational/07k-interruptible-lmnt.py +++ b/examples/foundational/07k-interruptible-lmnt.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -90,7 +89,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07l-interruptible-together.py b/examples/foundational/07l-interruptible-together.py index 0b92ff99a..14bcb703c 100644 --- a/examples/foundational/07l-interruptible-together.py +++ b/examples/foundational/07l-interruptible-together.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -105,7 +104,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07m-interruptible-polly.py b/examples/foundational/07m-interruptible-polly.py index 2cb0a7302..15035a35f 100644 --- a/examples/foundational/07m-interruptible-polly.py +++ b/examples/foundational/07m-interruptible-polly.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -99,7 +98,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07n-interruptible-google.py b/examples/foundational/07n-interruptible-google.py index 442de6dd8..6c83825aa 100644 --- a/examples/foundational/07n-interruptible-google.py +++ b/examples/foundational/07n-interruptible-google.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -98,7 +97,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07o-interruptible-assemblyai.py b/examples/foundational/07o-interruptible-assemblyai.py index 258d960bb..6e3ac5d5c 100644 --- a/examples/foundational/07o-interruptible-assemblyai.py +++ b/examples/foundational/07o-interruptible-assemblyai.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -98,7 +97,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07p-interruptible-krisp.py b/examples/foundational/07p-interruptible-krisp.py index 8c2316ca5..0773acad6 100644 --- a/examples/foundational/07p-interruptible-krisp.py +++ b/examples/foundational/07p-interruptible-krisp.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.filters.krisp_filter import KrispFilter -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -93,7 +92,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07q-interruptible-rime.py b/examples/foundational/07q-interruptible-rime.py index ccf630e68..2465e2596 100644 --- a/examples/foundational/07q-interruptible-rime.py +++ b/examples/foundational/07q-interruptible-rime.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -93,7 +92,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07r-interruptible-riva-nim.py b/examples/foundational/07r-interruptible-riva-nim.py index 972bb121f..d54ae4c19 100644 --- a/examples/foundational/07r-interruptible-riva-nim.py +++ b/examples/foundational/07r-interruptible-riva-nim.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -85,7 +84,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07s-interruptible-google-audio-in.py b/examples/foundational/07s-interruptible-google-audio-in.py index 980c4640d..57e1a5c98 100644 --- a/examples/foundational/07s-interruptible-google-audio-in.py +++ b/examples/foundational/07s-interruptible-google-audio-in.py @@ -17,7 +17,6 @@ from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( - EndFrame, Frame, InputAudioRawFrame, LLMFullResponseEndFrame, @@ -271,7 +270,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/07t-interruptible-fish.py b/examples/foundational/07t-interruptible-fish.py index 4a5d1395d..f7521ac68 100644 --- a/examples/foundational/07t-interruptible-fish.py +++ b/examples/foundational/07t-interruptible-fish.py @@ -14,7 +14,6 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -92,7 +91,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/21-tavus-layer.py b/examples/foundational/21-tavus-layer.py index 2a1ca0fbc..158c485d2 100644 --- a/examples/foundational/21-tavus-layer.py +++ b/examples/foundational/21-tavus-layer.py @@ -14,7 +14,6 @@ from dotenv import load_dotenv from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -120,7 +119,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/26d-gemini-multimodal-live-text.py b/examples/foundational/26d-gemini-multimodal-live-text.py index 7586764bd..aa72cae53 100644 --- a/examples/foundational/26d-gemini-multimodal-live-text.py +++ b/examples/foundational/26d-gemini-multimodal-live-text.py @@ -15,7 +15,6 @@ from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -124,7 +123,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): print(f"Participant left: {participant}") - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/28a-transcription-processor-openai.py b/examples/foundational/28a-transcription-processor-openai.py index 7f1d3dc25..d432c35f2 100644 --- a/examples/foundational/28a-transcription-processor-openai.py +++ b/examples/foundational/28a-transcription-processor-openai.py @@ -15,11 +15,7 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import ( - CancelFrame, - TranscriptionMessage, - TranscriptionUpdateFrame, -) +from pipecat.frames.frames import TranscriptionMessage, TranscriptionUpdateFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -170,7 +166,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): # Stop the pipeline immediately when the participant leaves - await task.queue_frame(CancelFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/28b-transcript-processor-anthropic.py b/examples/foundational/28b-transcript-processor-anthropic.py index b9f95db66..5b45a9c3d 100644 --- a/examples/foundational/28b-transcript-processor-anthropic.py +++ b/examples/foundational/28b-transcript-processor-anthropic.py @@ -15,11 +15,7 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import ( - CancelFrame, - TranscriptionMessage, - TranscriptionUpdateFrame, -) +from pipecat.frames.frames import TranscriptionMessage, TranscriptionUpdateFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -170,7 +166,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): # Stop the pipeline immediately when the participant leaves - await task.queue_frame(CancelFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/28c-transcription-processor-gemini.py b/examples/foundational/28c-transcription-processor-gemini.py index 377274ae1..05e1870e9 100644 --- a/examples/foundational/28c-transcription-processor-gemini.py +++ b/examples/foundational/28c-transcription-processor-gemini.py @@ -15,11 +15,7 @@ from loguru import logger from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import ( - CancelFrame, - TranscriptionMessage, - TranscriptionUpdateFrame, -) +from pipecat.frames.frames import TranscriptionMessage, TranscriptionUpdateFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -180,7 +176,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): # Stop the pipeline immediately when the participant leaves - await task.queue_frame(CancelFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/foundational/30-observer.py b/examples/foundational/30-observer.py index 7b6c64e47..4b672ba0e 100644 --- a/examples/foundational/30-observer.py +++ b/examples/foundational/30-observer.py @@ -17,7 +17,6 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, - EndFrame, Frame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, @@ -170,7 +169,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/news-chatbot/server/news_bot.py b/examples/news-chatbot/server/news_bot.py index f3d28cab5..f1c83c2fd 100644 --- a/examples/news-chatbot/server/news_bot.py +++ b/examples/news-chatbot/server/news_bot.py @@ -1,5 +1,5 @@ # -# Copyright (c) 2024, Daily +# Copyright (c) 2024-2025, Daily # # SPDX-License-Identifier: BSD 2-Clause License # @@ -14,7 +14,7 @@ from dotenv import load_dotenv from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame, Frame +from pipecat.frames.frames import Frame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -156,7 +156,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): print(f"Participant left: {participant}") - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() await runner.run(task) diff --git a/examples/phone-chatbot/bot_daily.py b/examples/phone-chatbot/bot_daily.py index 81913e87a..b02794bb3 100644 --- a/examples/phone-chatbot/bot_daily.py +++ b/examples/phone-chatbot/bot_daily.py @@ -7,7 +7,6 @@ from dotenv import load_dotenv from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -115,7 +114,7 @@ async def main(room_url: str, token: str, callId: str, callDomain: str, dialout_ @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/phone-chatbot/bot_twilio.py b/examples/phone-chatbot/bot_twilio.py index b2b21bb01..9da28dfa1 100644 --- a/examples/phone-chatbot/bot_twilio.py +++ b/examples/phone-chatbot/bot_twilio.py @@ -8,7 +8,6 @@ from loguru import logger from twilio.rest import Client from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -87,7 +86,7 @@ async def main(room_url: str, token: str, callId: str, sipUri: str): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() @transport.event_handler("on_dialin_ready") async def on_dialin_ready(transport, cdata): diff --git a/examples/simple-chatbot/server/bot-gemini.py b/examples/simple-chatbot/server/bot-gemini.py index 16e3f8fcd..7999b1bfc 100644 --- a/examples/simple-chatbot/server/bot-gemini.py +++ b/examples/simple-chatbot/server/bot-gemini.py @@ -31,7 +31,6 @@ from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, - EndFrame, Frame, OutputImageRawFrame, SpriteFrame, @@ -196,7 +195,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): print(f"Participant left: {participant}") - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/simple-chatbot/server/bot-openai.py b/examples/simple-chatbot/server/bot-openai.py index d23d254bf..0d30a5d87 100644 --- a/examples/simple-chatbot/server/bot-openai.py +++ b/examples/simple-chatbot/server/bot-openai.py @@ -31,7 +31,6 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( BotStartedSpeakingFrame, BotStoppedSpeakingFrame, - EndFrame, Frame, OutputImageRawFrame, SpriteFrame, @@ -220,7 +219,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): print(f"Participant left: {participant}") - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/storytelling-chatbot/src/bot.py b/examples/storytelling-chatbot/src/bot.py index 8be1e785d..8c896dc86 100644 --- a/examples/storytelling-chatbot/src/bot.py +++ b/examples/storytelling-chatbot/src/bot.py @@ -17,20 +17,14 @@ from prompts import CUE_USER_TURN, LLM_BASE_PROMPT from utils.helpers import load_images, load_sounds from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame, StopTaskFrame +from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.openai_llm_context import ( - OpenAILLMContext, - OpenAILLMContextFrame, -) -from pipecat.processors.logger import FrameLogger -from pipecat.services.cartesia import CartesiaHttpTTSService, CartesiaTTSService +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.services.elevenlabs import ElevenLabsTTSService from pipecat.services.fal import FalImageGenService from pipecat.services.google import GoogleLLMService -from pipecat.services.openai import OpenAILLMService from pipecat.transports.services.daily import ( DailyParams, DailyTransport, @@ -145,11 +139,13 @@ async def main(room_url, token=None): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await main_task.queue_frame(EndFrame()) + await main_task.cancel() @transport.event_handler("on_call_state_updated") async def on_call_state_updated(transport, state): if state == "left": + # Here we don't want to cancel, we just want to finish sending + # whatever is queued, so we use an EndFrame(). await main_task.queue_frame(EndFrame()) await runner.run(main_task) diff --git a/examples/studypal/studypal.py b/examples/studypal/studypal.py index 8e45b609f..b4691fc5f 100644 --- a/examples/studypal/studypal.py +++ b/examples/studypal/studypal.py @@ -12,7 +12,6 @@ from pypdf import PdfReader from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -171,7 +170,7 @@ Your task is to help the user understand and learn from this article in 2 senten @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/translation-chatbot/bot.py b/examples/translation-chatbot/bot.py index cf3d891b2..d35cf7af2 100644 --- a/examples/translation-chatbot/bot.py +++ b/examples/translation-chatbot/bot.py @@ -16,7 +16,6 @@ from runner import configure from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.frames.frames import ( - EndFrame, Frame, LLMMessagesFrame, TranscriptionFrame, @@ -26,7 +25,6 @@ from pipecat.frames.frames import ( from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask -from pipecat.processors.aggregators.llm_response import LLMFullResponseAggregator from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.processors.transcript_processor import TranscriptProcessor @@ -190,7 +188,7 @@ async def main(): @transport.event_handler("on_participant_left") async def on_participant_left(transport, participant, reason): - await task.queue_frame(EndFrame()) + await task.cancel() runner = PipelineRunner() diff --git a/examples/twilio-chatbot/bot.py b/examples/twilio-chatbot/bot.py index aed7041f8..e715c02e1 100644 --- a/examples/twilio-chatbot/bot.py +++ b/examples/twilio-chatbot/bot.py @@ -11,7 +11,6 @@ from dotenv import load_dotenv from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.frames.frames import EndFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -85,7 +84,7 @@ async def run_bot(websocket_client, stream_sid): @transport.event_handler("on_client_disconnected") async def on_client_disconnected(transport, client): - await task.queue_frames([EndFrame()]) + await task.cancel() runner = PipelineRunner(handle_sigint=False) From 0547a1569583b390a7151827d36f32dee41aeefa Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 28 Jan 2025 15:51:22 -0800 Subject: [PATCH 4/4] task: allow queuing a CancelFrame to cancel the task --- CHANGELOG.md | 4 ++++ src/pipecat/pipeline/task.py | 2 +- 2 files changed, 5 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 46ac58a8c..0e324c463 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed a `PipelineTask` issue that was preventing processors to be cleaned up after cancelling the task. +- Fixed an issue where queuing a `CancelFrame` to a pipeline task would not + cause the task to finish. However, using `PipelineTask.cancel()` is still the + recommended way to cancel a task. + ### Other - Updated all examples to use `task.cancel()` instead of pushing an `EndFrame` diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9f59e8177..b862583ef 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -279,7 +279,7 @@ class PipelineTask(BaseTask): await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM) if isinstance(frame, EndFrame): await self._wait_for_endframe() - running = not isinstance(frame, (StopTaskFrame, EndFrame)) + running = not isinstance(frame, (CancelFrame, EndFrame, StopTaskFrame)) should_cleanup = not isinstance(frame, StopTaskFrame) self._push_queue.task_done() # Cleanup only if we need to.