Triggering to check if the turn is complete based on the maximum timeout

src/pipecat/audio/turn/base_turn_analyzer.py

@@ -36,7 +36,7 @@ class BaseEndOfTurnAnalyzer(ABC):
         self._chunk_size_ms = chunk_size_ms
 
     @abstractmethod
-    def append_audio(self, buffer: bytes, is_speech: bool):
+    def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
         pass
 
     @abstractmethod

src/pipecat/audio/turn/local_smart_turn.py

@@ -68,11 +68,12 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer):
         self._silence_frames = 0
         self._speech_start_time = None
 
-    def append_audio(self, buffer: bytes, is_speech: bool):
+    def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
         audio_int16 = np.frombuffer(buffer, dtype=np.int16)
         # Divide by 32768 because we have signed 16-bit data.
         audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0
 
+        state = EndOfTurnState.INCOMPLETE
         if is_speech:
             if not self._speech_triggered:
                 self._silence_frames = 0
@@ -84,6 +85,10 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer):
             if self._speech_triggered:
                 self._audio_buffer.append((time.time(), audio_float32))
                 self._silence_frames += 1
+                if self._silence_frames * self._chunk_size_ms >= STOP_MS:
+                    logger.debug("End of Turn complete due to STOP_MS.")
+                    state = EndOfTurnState.COMPLETE
+                    self._clear()
             else:
                 # Keep buffering some silence before potential speech starts
                 self._audio_buffer.append((time.time(), audio_float32))
@@ -96,22 +101,22 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer):
                 ):
                     self._audio_buffer.pop(0)
 
+        return state
+
     def analyze_end_of_turn(self) -> EndOfTurnState:
         logger.debug("Analyzing End of Turn...")
-        if self._silence_frames * self._chunk_size_ms >= STOP_MS:
+        state = self._process_speech_segment(self._audio_buffer)
-            logger.debug("End of Turn complete due to STOP_MS.")
-            state = EndOfTurnState.COMPLETE
-        else:
-            state = self._process_speech_segment(self._audio_buffer)
-
         if state == EndOfTurnState.COMPLETE:
-            self._speech_triggered = False
+            self._clear()
-            self._audio_buffer = []
-            self._speech_start_time = None
 
         logger.debug(f"End of Turn result: {state}")
         return state
 
+    def _clear(self):
+        self._speech_triggered = False
+        self._audio_buffer = []
+        self._speech_start_time = None
+
     def _process_speech_segment(self, audio_buffer) -> EndOfTurnState:
         state = EndOfTurnState.INCOMPLETE
 

src/pipecat/transports/base_input.py

@@ -175,16 +175,9 @@ class BaseInputTransport(FrameProcessor):
         elif isinstance(frame, UserStoppedSpeakingFrame):
             logger.debug("User stopped speaking")
             await self.push_frame(frame)
-
-            # TODO check, we probably should change here as well.
-            # if the end of turn is enabled, we should only stop interruption after this point
             if self.interruptions_allowed:
                 await self._stop_interruption()
                 await self.push_frame(StopInterruptionFrame())
-        elif isinstance(frame, UserEndOfTurnFrame):
-            logger.debug("User end of turn")
-            await self.push_frame(frame)
-            # TODO: implement to handle interruptions
 
     #
     # Audio input
@@ -208,7 +201,9 @@ class BaseInputTransport(FrameProcessor):
             frame = None
             if new_vad_state == VADState.SPEAKING:
                 frame = UserStartedSpeakingFrame()
-            elif new_vad_state == VADState.QUIET:
+            # TODO: need to double check if this is the expected behavior
+            # Not triggering the UserStoppedSpeakingFrame if the turn analyser is enabled
+            elif new_vad_state == VADState.QUIET and not self.end_of_turn_analyzer:
                 frame = UserStoppedSpeakingFrame()
 
             if frame:
@@ -217,19 +212,20 @@ class BaseInputTransport(FrameProcessor):
             vad_state = new_vad_state
         return vad_state
 
-    async def _handle_end_of_turn(self, end_of_turn_state: EndOfTurnState):
+    async def _handle_end_of_turn(self):
-        state = end_of_turn_state
         if self.end_of_turn_analyzer:
-            new_state = await self.get_event_loop().run_in_executor(
+            state = await self.get_event_loop().run_in_executor(
                 self._executor, self.end_of_turn_analyzer.analyze_end_of_turn
             )
-            if new_state != state and new_state == EndOfTurnState.COMPLETE:
+            await self._handle_end_of_turn_complete(state)
-                await self._handle_user_interruption(UserEndOfTurnFrame())
+
-        return state
+    async def _handle_end_of_turn_complete(self, state: EndOfTurnState):
+        if state == EndOfTurnState.COMPLETE:
+            await self.push_frame(UserEndOfTurnFrame())
+            await self._handle_user_interruption(UserStoppedSpeakingFrame())
 
     async def _audio_task_handler(self):
         vad_state: VADState = VADState.QUIET
-        end_of_turn_state: EndOfTurnState = EndOfTurnState.INCOMPLETE
         while True:
             frame: InputAudioRawFrame = await self._audio_in_queue.get()
 
@@ -248,9 +244,13 @@ class BaseInputTransport(FrameProcessor):
 
             if self._params.end_of_turn_analyzer:
                 is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING
-                self._params.end_of_turn_analyzer.append_audio(frame.audio, is_speech)
+                end_of_turn_state = self._params.end_of_turn_analyzer.append_audio(
-                if vad_state == VADState.QUIET and vad_state != previous_vad_state:
+                    frame.audio, is_speech
-                    end_of_turn_state = await self._handle_end_of_turn(end_of_turn_state)
+                )
+                if end_of_turn_state == EndOfTurnState.COMPLETE:
+                    await self._handle_end_of_turn_complete(end_of_turn_state)
+                elif vad_state == VADState.QUIET and vad_state != previous_vad_state:
+                    await self._handle_end_of_turn()
 
             # Push audio downstream if passthrough.
             if audio_passthrough: