From 5bb64098e7c35f3c9ec2736148b331937a6ed278 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 24 Jul 2024 14:33:09 -0700 Subject: [PATCH] transports(inputs): don't queue incoming system frames --- CHANGELOG.md | 7 +++++++ src/pipecat/transports/base_input.py | 11 +++++++++-- 2 files changed, 16 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 902426bff..3ef4a75c3 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 `BaseInputTransport` issue that was causing incoming system frames to + be queued instead of being pushed immediately. + ## [0.0.39] - 2024-07-23 ### Fixed diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index c0a13d1ce..9f2683e7b 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -18,6 +18,7 @@ from pipecat.frames.frames import ( Frame, StartInterruptionFrame, StopInterruptionFrame, + SystemFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame) from pipecat.transports.base_transport import TransportParams @@ -69,18 +70,24 @@ class BaseInputTransport(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) + # Specific system frames if isinstance(frame, CancelFrame): await self.stop() # We don't queue a CancelFrame since we want to stop ASAP. await self.push_frame(frame, direction) + elif isinstance(frame, BotInterruptionFrame): + await self._handle_interruptions(frame, False) + # All other system frames + elif isinstance(frame, SystemFrame): + await self.push_frame(frame, direction) + # Control frames elif isinstance(frame, StartFrame): await self.start(frame) await self._internal_push_frame(frame, direction) elif isinstance(frame, EndFrame): await self._internal_push_frame(frame, direction) await self.stop() - elif isinstance(frame, BotInterruptionFrame): - await self._handle_interruptions(frame, False) + # Other frames else: await self._internal_push_frame(frame, direction)