Merge pull request #2183 from pipecat-ai/filipi/parallel_pipeline_issue

Fixed an issue in ParallelPipeline that caused errors when attempting to drain the queues.

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue in ParallelPipeline that caused errors when attempting to drain
+  the queues.
+
 - Fix a pipeline freeze when using AWS Nova Sonic, which would occur if the
   user started early, while the bot was still working through
   `trigger_assistant_response()`.

src/pipecat/pipeline/parallel_pipeline.py

@@ -273,12 +273,17 @@ class ParallelPipeline(BasePipeline):
         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."""
-        while not self._up_queue.empty:
+        await self._drain_queue(self._up_queue)
-            await self._up_queue.get()
+        await self._drain_queue(self._down_queue)
-        while not self._down_queue.empty:
-            await self._down_queue.get()
 
     async def _handle_interruption(self):
         """Handle interruption by cancelling tasks, draining queues, and restarting."""