Merge pull request #1592 from pipecat-ai/smart_turn

Smart turn

dot-env.template

@@ -92,4 +92,8 @@ ASSEMBLYAI_API_KEY=...
 OPENROUTER_API_KEY=...
 
 # Piper
-PIPER_BASE_URL=...
+PIPER_BASE_URL=...
+
+# Smart turn
+LOCAL_SMART_TURN_MODEL_PATH=
+REMOTE_SMART_TURN_URL=

examples/foundational/38-smart-turn.py

@@ -0,0 +1,111 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.turn.smart_turn import SmartTurnAnalyzer
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
+
+load_dotenv(override=True)
+
+
+async def run_bot(webrtc_connection: SmallWebRTCConnection):
+    logger.info(f"Starting bot")
+
+    remote_smart_turn_url = os.getenv("REMOTE_SMART_TURN_URL")
+
+    transport = SmallWebRTCTransport(
+        webrtc_connection=webrtc_connection,
+        params=TransportParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+            vad_audio_passthrough=True,
+            end_of_turn_analyzer=SmartTurnAnalyzer(url=remote_smart_turn_url),
+        ),
+    )
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),  # Assistant spoken responses
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            allow_interruptions=True,
+            enable_metrics=True,
+            enable_usage_metrics=True,
+            report_only_initial_ttfb=True,
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+
+    @transport.event_handler("on_client_closed")
+    async def on_client_closed(transport, client):
+        logger.info(f"Client closed connection")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=False)
+
+    await runner.run(task)
+
+
+if __name__ == "__main__":
+    from run import main
+
+    main()

examples/foundational/38a-local-smart-turn.py

@@ -0,0 +1,129 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.turn.base_smart_turn import SmartTurnParams
+from pipecat.audio.turn.local_smart_turn import LocalCoreMLSmartTurnAnalyzer
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
+from pipecat.services.cartesia.tts import CartesiaTTSService
+from pipecat.services.deepgram.stt import DeepgramSTTService
+from pipecat.services.openai.llm import OpenAILLMService
+from pipecat.transports.base_transport import TransportParams
+from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
+from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
+
+load_dotenv(override=True)
+
+
+async def run_bot(webrtc_connection: SmallWebRTCConnection):
+    logger.info(f"Starting bot")
+
+    # To use this locally, set the environment variable LOCAL_SMART_TURN_MODEL_PATH
+    # to the path where the smart-turn repo is cloned.
+    #
+    # Example setup:
+    #
+    #   # Git LFS (Large File Storage)
+    #   brew install git-lfs
+    #   # Hugging Face uses LFS to store large model files, including .mlpackage
+    #   git lfs install
+    #   # Clone the repo with the smart_turn_classifier.mlpackage
+    #   git clone https://huggingface.co/pipecat-ai/smart-turn
+    #
+    # Then set the env variable:
+    #   export LOCAL_SMART_TURN_MODEL_PATH=./smart-turn
+    # or add it to your .env file
+    smart_turn_model_path = os.getenv("LOCAL_SMART_TURN_MODEL_PATH")
+
+    transport = SmallWebRTCTransport(
+        webrtc_connection=webrtc_connection,
+        params=TransportParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            vad_enabled=True,
+            vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)),
+            vad_audio_passthrough=True,
+            end_of_turn_analyzer=LocalCoreMLSmartTurnAnalyzer(
+                smart_turn_model_path=smart_turn_model_path, params=SmartTurnParams()
+            ),
+        ),
+    )
+
+    stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY"))
+
+    tts = CartesiaTTSService(
+        api_key=os.getenv("CARTESIA_API_KEY"),
+        voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
+    )
+
+    llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"))
+
+    messages = [
+        {
+            "role": "system",
+            "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.",
+        },
+    ]
+
+    context = OpenAILLMContext(messages)
+    context_aggregator = llm.create_context_aggregator(context)
+
+    pipeline = Pipeline(
+        [
+            transport.input(),  # Transport user input
+            stt,
+            context_aggregator.user(),  # User responses
+            llm,  # LLM
+            tts,  # TTS
+            transport.output(),  # Transport bot output
+            context_aggregator.assistant(),  # Assistant spoken responses
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            allow_interruptions=True,
+            enable_metrics=True,
+            enable_usage_metrics=True,
+            report_only_initial_ttfb=True,
+        ),
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(transport, client):
+        logger.info(f"Client connected")
+        # Kick off the conversation.
+        messages.append({"role": "system", "content": "Please introduce yourself to the user."})
+        await task.queue_frames([context_aggregator.user().get_context_frame()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(transport, client):
+        logger.info(f"Client disconnected")
+
+    @transport.event_handler("on_client_closed")
+    async def on_client_closed(transport, client):
+        logger.info(f"Client closed connection")
+        await task.cancel()
+
+    runner = PipelineRunner(handle_sigint=False)
+
+    await runner.run(task)
+
+
+if __name__ == "__main__":
+    from run import main
+
+    main()

pyproject.toml

@@ -79,6 +79,8 @@ qwen = []
 rime = [ "websockets~=13.1" ]
 riva = [ "nvidia-riva-client~=2.19.0" ]
 sentry = [ "sentry-sdk~=2.23.1" ]
+local-smart-turn = [ "coremltools>=8.0", "transformers", "torch==2.5.0", "torchaudio==2.5.0" ]
+remote-smart-turn = []
 silero = [ "onnxruntime~=1.20.1" ]
 simli = [ "simli-ai~=0.1.10"]
 soundfile = [ "soundfile~=0.13.0" ]

src/pipecat/audio/turn/base_smart_turn.py

@@ -0,0 +1,182 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import time
+from abc import ABC, abstractmethod
+from enum import Enum
+from typing import Dict, Optional
+
+import numpy as np
+from loguru import logger
+from pydantic import BaseModel
+
+
+# Enum for end-of-turn detection states
+class EndOfTurnState(Enum):
+    COMPLETE = 1
+    INCOMPLETE = 2
+
+
+# Default timing parameters
+STOP_SECS = 3
+PRE_SPEECH_MS = 0
+MAX_DURATION_SECONDS = 8  # Max allowed segment duration
+USE_ONLY_LAST_VAD_SEGMENT = True
+
+
+class SmartTurnParams(BaseModel):
+    stop_secs: float = STOP_SECS
+    pre_speech_ms: float = PRE_SPEECH_MS
+    max_duration_secs: float = MAX_DURATION_SECONDS
+    # not exposing this for now yet until the model can handle it.
+    # use_only_last_vad_segment: bool = USE_ONLY_LAST_VAD_SEGMENT
+
+
+class BaseSmartTurn(ABC):
+    def __init__(
+        self, *, sample_rate: Optional[int] = None, params: SmartTurnParams = SmartTurnParams()
+    ):
+        self._init_sample_rate = sample_rate
+        self._params = params
+        # Configuration
+        self._sample_rate = 0
+        self._stop_ms = self._params.stop_secs * 1000  # silence threshold in ms
+        # Inference state
+        self._audio_buffer = []
+        self._speech_triggered = False
+        self._silence_ms = 0
+        self._speech_start_time = None
+
+    @property
+    def sample_rate(self) -> int:
+        return self._sample_rate
+
+    def set_sample_rate(self, sample_rate: int):
+        self._sample_rate = sample_rate
+
+    @property
+    def speech_triggered(self) -> bool:
+        return self._speech_triggered
+
+    def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState:
+        # Convert raw audio to float32 format and append to the buffer
+        audio_int16 = np.frombuffer(buffer, dtype=np.int16)
+        audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0
+        self._audio_buffer.append((time.time(), audio_float32))
+
+        state = EndOfTurnState.INCOMPLETE
+
+        if is_speech:
+            # Reset silence tracking on speech
+            self._silence_ms = 0
+            self._speech_triggered = True
+            if self._speech_start_time is None:
+                self._speech_start_time = time.time()
+                logger.debug(f"Speech started at {self._speech_start_time}")
+        else:
+            if self._speech_triggered:
+                chunk_duration_ms = len(audio_int16) / (self._sample_rate / 1000)
+                self._silence_ms += chunk_duration_ms
+                # If silence exceeds threshold, mark end of turn
+                if self._silence_ms >= self._stop_ms:
+                    logger.debug(
+                        f"End of Turn complete due to stop_secs. Silence in ms: {self._silence_ms}"
+                    )
+                    state = EndOfTurnState.COMPLETE
+                    self._clear(state)
+            else:
+                # Trim buffer to prevent unbounded growth before speech
+                max_buffer_time = (
+                    (self._params.pre_speech_ms / 1000)
+                    + self._params.stop_secs
+                    + self._params.max_duration_secs
+                )
+                while (
+                    self._audio_buffer and self._audio_buffer[0][0] < time.time() - max_buffer_time
+                ):
+                    self._audio_buffer.pop(0)
+
+        return state
+
+    def analyze_end_of_turn(self) -> EndOfTurnState:
+        logger.debug("Analyzing End of Turn...")
+        state = self._process_speech_segment(self._audio_buffer)
+        if state == EndOfTurnState.COMPLETE or USE_ONLY_LAST_VAD_SEGMENT:
+            self._clear(state)
+        logger.debug(f"End of Turn result: {state}")
+        return state
+
+    def _clear(self, turn_state: EndOfTurnState):
+        # Reset internal state for next turn
+        logger.debug("Clearing audio buffer...")
+        # If the state is still incomplete, keep the _speech_triggered as True
+        self._speech_triggered = turn_state == EndOfTurnState.INCOMPLETE
+        self._audio_buffer = []
+        self._speech_start_time = None
+        self._silence_ms = 0
+
+    def _process_speech_segment(self, audio_buffer) -> EndOfTurnState:
+        state = EndOfTurnState.INCOMPLETE
+
+        if not audio_buffer:
+            return state
+
+        # Extract recent audio segment for prediction
+        start_time = self._speech_start_time - (self._params.pre_speech_ms / 1000)
+        start_index = 0
+        for i, (t, _) in enumerate(audio_buffer):
+            if t >= start_time:
+                start_index = i
+                break
+
+        end_index = len(audio_buffer) - 1
+
+        # Extract the audio segment
+        segment_audio_chunks = [chunk for _, chunk in audio_buffer[start_index : end_index + 1]]
+        segment_audio = np.concatenate(segment_audio_chunks)
+
+        logger.debug(f"Segment audio chunks after start index: {len(segment_audio)}")
+
+        # Limit maximum duration
+        max_samples = int(self._params.max_duration_secs * self.sample_rate)
+        if len(segment_audio) > max_samples:
+            # slices the array to keep the last max_samples samples, discarding the earlier part.
+            segment_audio = segment_audio[-max_samples:]
+
+        logger.debug(f"Segment audio chunks after limiting duration: {len(segment_audio)}")
+
+        if len(segment_audio) > 0:
+            start_time = time.perf_counter()
+            result = self._predict_endpoint(segment_audio)
+            state = (
+                EndOfTurnState.COMPLETE if result["prediction"] == 1 else EndOfTurnState.INCOMPLETE
+            )
+            end_time = time.perf_counter()
+
+            logger.debug("--------")
+            logger.debug(f"Prediction: {'Complete' if result['prediction'] == 1 else 'Incomplete'}")
+            logger.debug(f"Probability of complete: {result['probability']:.4f}")
+            logger.debug(f"Prediction took {(end_time - start_time) * 1000:.2f}ms seconds")
+        else:
+            logger.debug(f"params: {self._params}, stop_ms: {self._stop_ms}")
+            logger.debug("Captured empty audio segment, skipping prediction.")
+
+        return state
+
+    @abstractmethod
+    def _predict_endpoint(self, buffer: np.ndarray) -> Dict[str, any]:
+        """
+        Abstract method to predict if a turn has ended based on audio.
+
+        Args:
+            buffer: Float32 numpy array of audio samples at 16kHz.
+
+        Returns:
+            Dictionary with:
+              - prediction: 1 if turn is complete, else 0
+              - probability: Confidence of the prediction
+        """
+        pass

src/pipecat/audio/turn/local_smart_turn.py

@@ -0,0 +1,65 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import os
+from typing import Dict
+
+import numpy as np
+import torch
+from loguru import logger
+
+from pipecat.audio.turn.base_smart_turn import BaseSmartTurn
+
+try:
+    import coremltools as ct
+    from transformers import AutoFeatureExtractor
+except ModuleNotFoundError as e:
+    logger.error(f"Exception: {e}")
+    logger.error(
+        "In order to use the LocalSmartTurnAnalyzer, you need to `pip install pipecat-ai[local-smart-turn]`."
+    )
+    raise Exception(f"Missing module: {e}")
+
+
+class LocalCoreMLSmartTurnAnalyzer(BaseSmartTurn):
+    def __init__(self, smart_turn_model_path: str, **kwargs):
+        super().__init__(**kwargs)
+
+        if not smart_turn_model_path:
+            logger.error("smart_turn_model_path is not set.")
+            raise Exception("smart_turn_model_path must be provided.")
+
+        core_ml_model_path = f"{smart_turn_model_path}/coreml/smart_turn_classifier.mlpackage"
+
+        logger.debug("Loading Local Smart Turn model...")
+        # Only load the processor, not the torch model
+        self._turn_processor = AutoFeatureExtractor.from_pretrained(smart_turn_model_path)
+        self._turn_model = ct.models.MLModel(core_ml_model_path)
+        logger.debug("Loaded Local Smart Turn")
+
+    def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, any]:
+        inputs = self._turn_processor(
+            audio_array,
+            sampling_rate=16000,
+            padding="max_length",
+            truncation=True,
+            max_length=800,  # Maximum length as specified in training
+            return_attention_mask=True,
+            return_tensors="pt",
+        )
+
+        output = self._turn_model.predict(dict(inputs))
+        logits = output["logits"]  # Core ML returns numpy array
+        logits_tensor = torch.tensor(logits)
+        probabilities = torch.nn.functional.softmax(logits_tensor, dim=1)
+        completion_prob = probabilities[0, 1].item()  # Probability of class 1 (Complete)
+        prediction = 1 if completion_prob > 0.5 else 0
+
+        return {
+            "prediction": prediction,
+            "probability": completion_prob,
+        }

src/pipecat/audio/turn/smart_turn.py

@@ -0,0 +1,75 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+
+import io
+import os
+from typing import Dict
+
+import numpy as np
+import requests
+from loguru import logger
+
+from pipecat.audio.turn.base_smart_turn import BaseSmartTurn
+
+
+class SmartTurnAnalyzer(BaseSmartTurn):
+    def __init__(self, url: str, **kwargs):
+        super().__init__(**kwargs)
+        self.remote_smart_turn_url = url
+
+        if not self.remote_smart_turn_url:
+            logger.error("remote_smart_turn_url is not set.")
+            raise Exception("remote_smart_turn_url must be provided.")
+
+        # Use a session to reuse connections (keep-alive)
+        self.session = requests.Session()
+        self.session.headers.update({"Connection": "keep-alive"})
+
+    def _serialize_array(self, audio_array: np.ndarray) -> bytes:
+        logger.trace("Serializing NumPy array to bytes...")
+        buffer = io.BytesIO()
+        np.save(buffer, audio_array)
+        serialized_bytes = buffer.getvalue()
+        logger.trace(f"Serialized size: {len(serialized_bytes)} bytes")
+        return serialized_bytes
+
+    def _send_raw_request(self, data_bytes: bytes):
+        headers = {"Content-Type": "application/octet-stream"}
+        logger.trace(
+            f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..."
+        )
+        try:
+            response = self.session.post(
+                self.remote_smart_turn_url,
+                data=data_bytes,
+                headers=headers,
+                timeout=60,
+            )
+
+            logger.trace("\n--- Response ---")
+            logger.trace(f"Status Code: {response.status_code}")
+
+            if response.ok:
+                try:
+                    logger.trace("Response JSON:")
+                    logger.trace(response.json())
+                    return response.json()
+                except requests.exceptions.JSONDecodeError:
+                    logger.trace("Response Content (non-JSON):")
+                    logger.trace(response.text)
+            else:
+                logger.trace("Response Content (Error):")
+                logger.trace(response.text)
+                response.raise_for_status()
+
+        except requests.exceptions.RequestException as e:
+            logger.error(f"Failed to send raw request to Daily Smart Turn: {e}")
+            raise Exception("Failed to send raw request to Daily Smart Turn.")
+
+    def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, any]:
+        serialized_array = self._serialize_array(audio_array)
+        return self._send_raw_request(serialized_array)

src/pipecat/transports/base_input.py

@@ -10,6 +10,7 @@ from typing import Optional
 
 from loguru import logger
 
+from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, EndOfTurnState
 from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState
 from pipecat.frames.frames import (
     BotInterruptionFrame,
@@ -64,12 +65,19 @@ class BaseInputTransport(FrameProcessor):
     def vad_analyzer(self) -> Optional[VADAnalyzer]:
         return self._params.vad_analyzer
 
+    @property
+    def end_of_turn_analyzer(self) -> Optional[BaseSmartTurn]:
+        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)
@@ -187,10 +195,18 @@ class BaseInputTransport(FrameProcessor):
             and new_vad_state != VADState.STOPPING
         ):
             frame = None
-            if new_vad_state == VADState.SPEAKING:
+            # If the turn analyser is enabled, this will prevent:
-                frame = UserStartedSpeakingFrame()
+            # - Creating the UserStoppedSpeakingFrame
-            elif new_vad_state == VADState.QUIET:
+            # - Creating the UserStartedSpeakingFrame multiple times
-                frame = UserStoppedSpeakingFrame()
+            can_create_user_frames = (
+                self._params.end_of_turn_analyzer is None
+                or not self._params.end_of_turn_analyzer.speech_triggered
+            )
+            if can_create_user_frames:
+                if new_vad_state == VADState.SPEAKING:
+                    frame = UserStartedSpeakingFrame()
+                elif new_vad_state == VADState.QUIET:
+                    frame = UserStoppedSpeakingFrame()
 
             if frame:
                 await self._handle_user_interruption(frame)
@@ -198,6 +214,29 @@ class BaseInputTransport(FrameProcessor):
             vad_state = new_vad_state
         return vad_state
 
+    async def _handle_end_of_turn(self):
+        if self.end_of_turn_analyzer:
+            state = await self.get_event_loop().run_in_executor(
+                self._executor, self.end_of_turn_analyzer.analyze_end_of_turn
+            )
+            await self._handle_end_of_turn_complete(state)
+
+    async def _handle_end_of_turn_complete(self, state: EndOfTurnState):
+        if state == EndOfTurnState.COMPLETE:
+            await self._handle_user_interruption(UserStoppedSpeakingFrame())
+
+    async def _run_turn_analyzer(
+        self, frame: InputAudioRawFrame, vad_state: VADState, previous_vad_state: VADState
+    ):
+        is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING
+        # If silence exceeds threshold, we are going to receive EndOfTurnState.COMPLETE
+        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)
+        # Otherwise we are going to trigger to check if the turn is completed based on the VAD
+        elif vad_state == VADState.QUIET and vad_state != previous_vad_state:
+            await self._handle_end_of_turn()
+
     async def _audio_task_handler(self):
         vad_state: VADState = VADState.QUIET
         while True:
@@ -211,10 +250,14 @@ class BaseInputTransport(FrameProcessor):
 
             # Check VAD and push event if necessary. We just care about
             # changes from QUIET to SPEAKING and vice versa.
+            previous_vad_state = vad_state
             if self._params.vad_enabled:
                 vad_state = await self._handle_vad(frame, vad_state)
                 audio_passthrough = self._params.vad_audio_passthrough
 
+            if self._params.end_of_turn_analyzer:
+                await self._run_turn_analyzer(frame, vad_state, previous_vad_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_smart_turn import BaseSmartTurn
 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[BaseSmartTurn] = None
 
 
 class BaseTransport(BaseObject):