wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

services(deepgram): allow transcriptions during interruptions

Browse Source 
If the user interrupts we can't just discard transcriptions because the user is
actually interrupting and talking.
This commit is contained in:
Aleix Conchillo Flaqué
2024-06-25 21:32:21 -07:00
parent e3b407db28
commit d302cbb114
1 changed files with 9 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/pipecat/services/deepgram.py
View File

@@ -19,7 +19,6 @@ from pipecat.frames.frames import (
InterimTranscriptionFrame, InterimTranscriptionFrame,
StartFrame, StartFrame,
StartInterruptionFrame, StartInterruptionFrame,
StopInterruptionFrame,
SystemFrame, SystemFrame,
TranscriptionFrame) TranscriptionFrame)
from pipecat.processors.frame_processor import FrameDirection from pipecat.processors.frame_processor import FrameDirection
@@ -118,16 +117,12 @@ class DeepgramSTTService(AIService):
self._connection = self._client.listen.asynclive.v("1") self._connection = self._client.listen.asynclive.v("1")
self._connection.on(LiveTranscriptionEvents.Transcript, self._on_message) 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() self._create_push_task()
async def process_frame(self, frame: Frame, direction: FrameDirection): async def process_frame(self, frame: Frame, direction: FrameDirection):
await super().process_frame(frame, direction) await super().process_frame(frame, direction)
if isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame): if isinstance(frame, StartInterruptionFrame):
await self._handle_interruptions(frame) await self._handle_interruptions(frame)
elif isinstance(frame, SystemFrame): elif isinstance(frame, SystemFrame):
await self.push_frame(frame, direction) await self.push_frame(frame, direction)
@@ -153,21 +148,14 @@ class DeepgramSTTService(AIService):
await self._push_frame_task await self._push_frame_task
async def _handle_interruptions(self, frame: Frame): async def _handle_interruptions(self, frame: Frame):
if isinstance(frame, StartInterruptionFrame): # Cancel the task. This will stop pushing frames downstream.
# Indicate we are interrupted, we should ignore any out-of-band self._push_frame_task.cancel()
# transcriptions. await self._push_frame_task
self._is_interrupted_event.set() # Push an out-of-band frame (i.e. not using the ordered push
# Cancel the task. This will stop pushing frames downstream. # frame task).
self._push_frame_task.cancel() await self.push_frame(frame)
await self._push_frame_task # Create a new queue and task.
# Push an out-of-band frame (i.e. not using the ordered push self._create_push_task()
# 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()
def _create_push_task(self): def _create_push_task(self):
self._push_queue = asyncio.Queue() self._push_queue = asyncio.Queue()
@@ -184,9 +172,6 @@ class DeepgramSTTService(AIService):
break break
async def _on_message(self, *args, **kwargs): async def _on_message(self, *args, **kwargs):
if self._is_interrupted_event.is_set():
return
result = kwargs["result"] result = kwargs["result"]
is_final = result.is_final is_final = result.is_final
transcript = result.channel.alternatives[0].transcript transcript = result.channel.alternatives[0].transcript