From 57b39c084fddd893799c7a536fc1527ba03fecb7 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 15 Apr 2025 20:42:41 -0300 Subject: [PATCH] Triggering to check if the turn is complete based on the maximum timeout --- src/pipecat/audio/turn/base_turn_analyzer.py | 2 +- src/pipecat/audio/turn/local_smart_turn.py | 25 ++++++++------ src/pipecat/transports/base_input.py | 36 ++++++++++---------- 3 files changed, 34 insertions(+), 29 deletions(-) diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py index 1e6a964f9..76777bc65 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/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 diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index b831fffeb..57f7584e0 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/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: - logger.debug("End of Turn complete due to STOP_MS.") - state = EndOfTurnState.COMPLETE - else: - state = self._process_speech_segment(self._audio_buffer) - + state = self._process_speech_segment(self._audio_buffer) if state == EndOfTurnState.COMPLETE: - self._speech_triggered = False - self._audio_buffer = [] - self._speech_start_time = None + self._clear() 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 diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 296d0a453..7755e9670 100644 --- a/src/pipecat/transports/base_input.py +++ b/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): - state = end_of_turn_state + async def _handle_end_of_turn(self): 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_user_interruption(UserEndOfTurnFrame()) - return state + await self._handle_end_of_turn_complete(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) - if vad_state == VADState.QUIET and vad_state != previous_vad_state: - end_of_turn_state = await self._handle_end_of_turn(end_of_turn_state) + end_of_turn_state = self._params.end_of_turn_analyzer.append_audio( + frame.audio, is_speech + ) + 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: