diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index e32e4c870..74b40cb90 100644 --- a/src/pipecat/services/deepgram.py +++ b/src/pipecat/services/deepgram.py @@ -19,7 +19,6 @@ from pipecat.frames.frames import ( InterimTranscriptionFrame, StartFrame, StartInterruptionFrame, - StopInterruptionFrame, SystemFrame, TranscriptionFrame) from pipecat.processors.frame_processor import FrameDirection @@ -118,16 +117,12 @@ class DeepgramSTTService(AIService): self._connection = self._client.listen.asynclive.v("1") self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message) - # This event will be used to ignore out-of-band transcriptions while we - # are itnerrupted. - self._is_interrupted_event = asyncio.Event() - self._create_push_task() async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - if isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame): + if isinstance(frame, StartInterruptionFrame): await self._handle_interruptions(frame) elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) @@ -153,21 +148,14 @@ class DeepgramSTTService(AIService): await self._push_frame_task async def _handle_interruptions(self, frame: Frame): - if isinstance(frame, StartInterruptionFrame): - # Indicate we are interrupted, we should ignore any out-of-band - # transcriptions. - self._is_interrupted_event.set() - # Cancel the task. This will stop pushing frames downstream. - self._push_frame_task.cancel() - await self._push_frame_task - # Push an out-of-band frame (i.e. not using the ordered push - # frame task). - await self.push_frame(frame) - # Create a new queue and task. - self._create_push_task() - elif isinstance(frame, StopInterruptionFrame): - # We should now be able to receive transcriptions again. - self._is_interrupted_event.clear() + # Cancel the task. This will stop pushing frames downstream. + self._push_frame_task.cancel() + await self._push_frame_task + # Push an out-of-band frame (i.e. not using the ordered push + # frame task). + await self.push_frame(frame) + # Create a new queue and task. + self._create_push_task() def _create_push_task(self): self._push_queue = asyncio.Queue() @@ -184,9 +172,6 @@ class DeepgramSTTService(AIService): break async def _on_message(self, *args, **kwargs): - if self._is_interrupted_event.is_set(): - return - result = kwargs["result"] is_final = result.is_final transcript = result.channel.alternatives[0].transcript