Merge pull request #1100 from pipecat-ai/aleix/use-task-cancel-on-lef…

…t-disconnected

use `task.cancel()` when participant leaves/disconnects

CHANGELOG.md

@@ -12,6 +12,21 @@ 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.
+
+- 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`
+  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

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)

examples/canonical-metrics/bot.py

@@ -130,11 +130,13 @@ async def on_first_participant_joined(transport, participant):
         @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()

examples/chatbot-audio-recording/bot.py

@@ -18,7 +18,6 @@
 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 on_first_participant_joined(transport, participant):
         @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()
 

examples/deployment/flyio-example/bot.py

@@ -79,11 +79,13 @@ async def on_first_participant_joined(transport, participant):
 
     @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()

examples/deployment/modal-example/bot.py

@@ -5,23 +5,22 @@
 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)
 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 on_first_participant_joined(transport, participant):
 
     @transport.event_handler("on_participant_left")
     async def on_participant_left(transport, participant, reason):
-        await task.queue_frame(EndFrame())
+        await task.cancel()
 
     runner = PipelineRunner()
 

examples/foundational/03-still-frame.py

@@ -13,7 +13,7 @@
 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 on_first_participant_joined(transport, participant):
 
         @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)
 

examples/foundational/06-listen-and-respond.py

@@ -14,7 +14,7 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/06a-image-sync.py

@@ -18,7 +18,6 @@
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
-    EndFrame,
     Frame,
     OutputImageRawFrame,
     TextFrame,
@@ -144,7 +143,7 @@ async def on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07-interruptible-vad.py

@@ -13,7 +13,6 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07-interruptible.py

@@ -14,7 +14,6 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07a-interruptible-anthropic.py

@@ -14,7 +14,6 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07b-interruptible-langchain.py

@@ -19,7 +19,7 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07c-interruptible-deepgram-vad.py

@@ -16,7 +16,6 @@
 
 from pipecat.frames.frames import (
     BotInterruptionFrame,
-    EndFrame,
     StopInterruptionFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
@@ -106,7 +105,7 @@ async def on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07c-interruptible-deepgram.py

@@ -14,7 +14,6 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07d-interruptible-elevenlabs.py

@@ -14,7 +14,6 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07e-interruptible-playht-http.py

@@ -14,14 +14,12 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07e-interruptible-playht.py

@@ -14,7 +14,6 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07f-interruptible-azure.py

@@ -14,7 +14,6 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07g-interruptible-openai-tts.py

@@ -14,7 +14,6 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()
 

examples/foundational/07h-interruptible-openpipe.py

@@ -15,7 +15,6 @@
 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 on_first_participant_joined(transport, participant):
 
         @transport.event_handler("on_participant_left")
         async def on_participant_left(transport, participant, reason):
-            await task.queue_frame(EndFrame())
+            await task.cancel()
 
         runner = PipelineRunner()