Bringing Aleix initial implementation for the smart turn.

src/pipecat/audio/turn/base_turn_analyzer.py

@@ -0,0 +1,32 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+from abc import ABC, abstractmethod
+from enum import Enum
+from typing import Optional
+
+
+class EndOfTurnState(Enum):
+    COMPLETE = 1
+    INCOMPLETE = 2
+
+
+class BaseEndOfTurnAnalyzer(ABC):
+    def __init__(self, *, sample_rate: Optional[int] = None):
+        self._init_sample_rate = sample_rate
+        self._sample_rate = 0
+
+    @property
+    def sample_rate(self) -> int:
+        return self._sample_rate
+
+    def set_sample_rate(self, sample_rate: int):
+        self._sample_rate = self._init_sample_rate or sample_rate
+
+    @abstractmethod
+    def analyze_audio(self, buffer: bytes) -> EndOfTurnState:
+        pass

src/pipecat/frames/frames.py

@@ -571,6 +571,15 @@ class EmulateUserStoppedSpeakingFrame(SystemFrame):
     pass
 
 
+@dataclass
+class UserEndOfTurnFrame(SystemFrame):
+    """Emitted based on the Smart Turn model to indicate that the user has
+    completed their turn/sentence.
+    """
+
+    pass
+
+
 @dataclass
 class BotInterruptionFrame(SystemFrame):
     """Emitted by when the bot should be interrupted. This will mainly cause the

src/pipecat/transports/base_input.py

@@ -10,6 +10,7 @@ from typing import Optional
 
 from loguru import logger
 
+from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer, EndOfTurnState
 from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState
 from pipecat.frames.frames import (
     BotInterruptionFrame,
@@ -24,6 +25,7 @@ from pipecat.frames.frames import (
     StartInterruptionFrame,
     StopInterruptionFrame,
     SystemFrame,
+    UserEndOfTurnFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
     VADParamsUpdateFrame,
@@ -64,12 +66,19 @@ class BaseInputTransport(FrameProcessor):
     def vad_analyzer(self) -> Optional[VADAnalyzer]:
         return self._params.vad_analyzer
 
+    @property
+    def end_of_turn_analyzer(self) -> Optional[BaseEndOfTurnAnalyzer]:
+        return self._params.end_of_turn_analyzer
+
     async def start(self, frame: StartFrame):
         self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate
 
         # Configure VAD analyzer.
         if self._params.vad_enabled and self._params.vad_analyzer:
             self._params.vad_analyzer.set_sample_rate(self._sample_rate)
+        # Configure End of turn analyzer.
+        if self._params.end_of_turn_analyzer:
+            self._params.end_of_turn_analyzer.set_sample_rate(self._sample_rate)
         # Start audio filter.
         if self._params.audio_in_filter:
             await self._params.audio_in_filter.start(self._sample_rate)
@@ -198,8 +207,25 @@ class BaseInputTransport(FrameProcessor):
             vad_state = new_vad_state
         return vad_state
 
+    async def _end_of_turn_analyze(self, audio_frame: InputAudioRawFrame) -> EndOfTurnState:
+        state = EndOfTurnState.INCOMPLETE
+        if self.end_of_turn_analyzer:
+            state = await self.get_event_loop().run_in_executor(
+                self._executor, self.end_of_turn_analyzer.analyze_audio, audio_frame.audio
+            )
+        return state
+
+    async def _handle_end_of_turn(
+        self, audio_frame: InputAudioRawFrame, end_of_turn_state: EndOfTurnState
+    ):
+        new_eot_state = await self._end_of_turn_analyze(audio_frame)
+        if new_eot_state != end_of_turn_state:
+            await self.push_frame(UserEndOfTurnFrame())
+        return new_eot_state
+
     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()
 
@@ -215,6 +241,9 @@ class BaseInputTransport(FrameProcessor):
                 vad_state = await self._handle_vad(frame, vad_state)
                 audio_passthrough = self._params.vad_audio_passthrough
 
+            if self._params.end_of_turn_analyzer:
+                end_of_turn_state = await self._handle_end_of_turn(frame, end_of_turn_state)
+
             # Push audio downstream if passthrough.
             if audio_passthrough:
                 await self.push_frame(frame)

src/pipecat/transports/base_transport.py

@@ -11,6 +11,7 @@ from pydantic import BaseModel, ConfigDict
 
 from pipecat.audio.filters.base_audio_filter import BaseAudioFilter
 from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer
+from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADAnalyzer
 from pipecat.processors.frame_processor import FrameProcessor
 from pipecat.utils.base_object import BaseObject
@@ -41,6 +42,7 @@ class TransportParams(BaseModel):
     vad_enabled: bool = False
     vad_audio_passthrough: bool = False
     vad_analyzer: Optional[VADAnalyzer] = None
+    end_of_turn_analyzer: Optional[BaseEndOfTurnAnalyzer] = None
 
 
 class BaseTransport(BaseObject):