diff --git a/CHANGELOG.md b/CHANGELOG.md
index a77671228..d3cb773f0 100644
--- a/CHANGELOG.md
+++ b/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()`.
diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py
index 250a3e989..9538e9ab4 100644
--- a/src/pipecat/pipeline/parallel_pipeline.py
+++ b/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._up_queue.get()
-        while not self._down_queue.empty:
-            await self._down_queue.get()
+        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."""