Merge pull request #1631 from pipecat-ai/smart_turn_timeout

Returning the turn as complete if the request don’t return a result within SmartTurnParams stop_secs

CHANGELOG.md

@@ -21,9 +21,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   `GoogleSTTService`, `GoogleTTSService`, and `GoogleVertexLLMService`.
 
 - Added support for Smart Turn Detection via the `turn_analyzer` transport
-  parameter. You can now choose between `SmartTurnAnalyzer()` for remote
+  parameter. You can now choose between `HttpSmartTurnAnalyzer()` or
-  inference or `LocalCoreMLSmartTurnAnalyzer()` for on-device inference using
+  `FalSmartTurnAnalyzer()` for remote inference or
-  Core ML.
+  `LocalCoreMLSmartTurnAnalyzer()` for on-device inference using Core ML.
 
 - `DeepgramTTSService` accepts `base_url` argument again, allowing you to
   connect to an on-prem service.

dot-env.template

@@ -96,7 +96,7 @@ PIPER_BASE_URL=...
 
 # Smart turn
 LOCAL_SMART_TURN_MODEL_PATH=
-REMOTE_SMART_TURN_URL=
+FAL_SMART_TURN_API_KEY=...
 
 # Twilio
 TWILIO_ACCOUNT_SID=

examples/foundational/38-smart-turn-fal.py

@@ -0,0 +1,113 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import os
+
+import aiohttp
+from dotenv import load_dotenv
+from loguru import logger
+
+from pipecat.audio.turn.smart_turn.fal_smart_turn import FalSmartTurnAnalyzer
+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")
+
+    async with aiohttp.ClientSession() as session:
+        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,
+                turn_analyzer=FalSmartTurnAnalyzer(
+                    api_key=os.getenv("FAL_SMART_TURN_API_KEY"), aiohttp_session=session
+                ),
+            ),
+        )
+
+        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/38-smart-turn.py

@@ -1,111 +0,0 @@
-#
-# 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,
-            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-smart-turn-local-coreml.py

@@ -9,8 +9,8 @@ import os
 from dotenv import load_dotenv
 from loguru import logger
 
-from pipecat.audio.turn.base_smart_turn import SmartTurnParams
+from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams
-from pipecat.audio.turn.local_smart_turn import LocalCoreMLSmartTurnAnalyzer
+from pipecat.audio.turn.smart_turn.local_coreml_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

src/pipecat/audio/turn/base_turn_analyzer.py

@@ -71,7 +71,7 @@ class BaseTurnAnalyzer(ABC):
         pass
 
     @abstractmethod
-    def analyze_end_of_turn(self) -> Tuple[EndOfTurnState, Optional[MetricsData]]:
+    async def analyze_end_of_turn(self) -> Tuple[EndOfTurnState, Optional[MetricsData]]:
         """Analyzes if an end of turn has occurred based on the audio input.
 
         Returns:

src/pipecat/audio/turn/smart_turn.py

@@ -1,75 +0,0 @@
-#
-# 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/audio/turn/smart_turn/base_smart_turn.py

@@ -30,6 +30,10 @@ class SmartTurnParams(BaseModel):
     # use_only_last_vad_segment: bool = USE_ONLY_LAST_VAD_SEGMENT
 
 
+class SmartTurnTimeoutException(Exception):
+    pass
+
+
 class BaseSmartTurn(BaseTurnAnalyzer):
     def __init__(
         self, *, sample_rate: Optional[int] = None, params: SmartTurnParams = SmartTurnParams()
@@ -42,7 +46,7 @@ class BaseSmartTurn(BaseTurnAnalyzer):
         self._audio_buffer = []
         self._speech_triggered = False
         self._silence_ms = 0
-        self._speech_start_time = None
+        self._speech_start_time = 0
 
     @property
     def speech_triggered(self) -> bool:
@@ -60,7 +64,7 @@ class BaseSmartTurn(BaseTurnAnalyzer):
             # Reset silence tracking on speech
             self._silence_ms = 0
             self._speech_triggered = True
-            if self._speech_start_time is None:
+            if self._speech_start_time == 0:
                 self._speech_start_time = time.time()
         else:
             if self._speech_triggered:
@@ -87,8 +91,8 @@ class BaseSmartTurn(BaseTurnAnalyzer):
 
         return state
 
-    def analyze_end_of_turn(self) -> Tuple[EndOfTurnState, Optional[MetricsData]]:
+    async def analyze_end_of_turn(self) -> Tuple[EndOfTurnState, Optional[MetricsData]]:
-        state, result = self._process_speech_segment(self._audio_buffer)
+        state, result = await 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}")
@@ -98,10 +102,12 @@ class BaseSmartTurn(BaseTurnAnalyzer):
         # 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._speech_start_time = 0
         self._silence_ms = 0
 
-    def _process_speech_segment(self, audio_buffer) -> Tuple[EndOfTurnState, Optional[MetricsData]]:
+    async def _process_speech_segment(
+        self, audio_buffer
+    ) -> Tuple[EndOfTurnState, Optional[MetricsData]]:
         state = EndOfTurnState.INCOMPLETE
 
         if not audio_buffer:
@@ -131,30 +137,41 @@ class BaseSmartTurn(BaseTurnAnalyzer):
 
         if len(segment_audio) > 0:
             start_time = time.perf_counter()
-            result = self._predict_endpoint(segment_audio)
+            try:
-            state = (
+                result = await self._predict_endpoint(segment_audio)
-                EndOfTurnState.COMPLETE if result["prediction"] == 1 else EndOfTurnState.INCOMPLETE
+                state = (
-            )
+                    EndOfTurnState.COMPLETE
-            end_time = time.perf_counter()
+                    if result["prediction"] == 1
+                    else EndOfTurnState.INCOMPLETE
+                )
+                end_time = time.perf_counter()
 
-            # Calculate processing time
+                # Calculate processing time
-            e2e_processing_time_ms = (end_time - start_time) * 1000
+                e2e_processing_time_ms = (end_time - start_time) * 1000
 
-            # Prepare the result data
+                # Prepare the result data
-            result_data = SmartTurnMetricsData(
+                result_data = SmartTurnMetricsData(
-                processor="BaseSmartTurn",
+                    processor="BaseSmartTurn",
-                is_complete=result["prediction"] == 1,
+                    is_complete=result["prediction"] == 1,
-                probability=result["probability"],
+                    probability=result["probability"],
-                inference_time_ms=result.get("inference_time", 0) * 1000,
+                    inference_time_ms=result.get("inference_time", 0) * 1000,
-                server_total_time_ms=result.get("total_time", 0) * 1000,
+                    server_total_time_ms=result.get("total_time", 0) * 1000,
-                e2e_processing_time_ms=e2e_processing_time_ms,
+                    e2e_processing_time_ms=e2e_processing_time_ms,
-            )
+                )
+
+                logger.trace(
+                    f"Prediction: {'Complete' if result_data.is_complete else 'Incomplete'}"
+                )
+                logger.trace(f"Probability of complete: {result_data.probability:.4f}")
+                logger.trace(f"Inference time: {result_data.inference_time_ms:.2f}ms")
+                logger.trace(f"Server total time: {result_data.server_total_time_ms:.2f}ms")
+                logger.trace(f"E2E processing time: {result_data.e2e_processing_time_ms:.2f}ms")
+            except SmartTurnTimeoutException:
+                logger.debug(
+                    f"End of Turn complete due to stop_secs. Silence in ms: {self._silence_ms}"
+                )
+                state = EndOfTurnState.COMPLETE
 
-            logger.trace(f"Prediction: {'Complete' if result_data.is_complete else 'Incomplete'}")
-            logger.trace(f"Probability of complete: {result_data.probability:.4f}")
-            logger.trace(f"Inference time: {result_data.inference_time_ms:.2f}ms")
-            logger.trace(f"Server total time: {result_data.server_total_time_ms:.2f}ms")
-            logger.trace(f"E2E processing time: {result_data.e2e_processing_time_ms:.2f}ms")
         else:
             logger.trace(f"params: {self._params}, stop_ms: {self._stop_ms}")
             logger.trace("Captured empty audio segment, skipping prediction.")
@@ -162,11 +179,11 @@ class BaseSmartTurn(BaseTurnAnalyzer):
         return state, result_data
 
     @abstractmethod
-    def _predict_endpoint(self, buffer: np.ndarray) -> Dict[str, Any]:
+    async def _predict_endpoint(self, audio_array: 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.
+            audio_array: Float32 numpy array of audio samples at 16kHz.
 
         Returns:
             Dictionary with:

src/pipecat/audio/turn/smart_turn/fal_smart_turn.py

@@ -0,0 +1,26 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from typing import Optional
+
+import aiohttp
+
+from pipecat.audio.turn.smart_turn.http_smart_turn import HttpSmartTurnAnalyzer
+
+
+class FalSmartTurnAnalyzer(HttpSmartTurnAnalyzer):
+    def __init__(
+        self,
+        *,
+        aiohttp_session: aiohttp.ClientSession,
+        url: str = "https://fal.run/fal-ai/smart-turn/raw",
+        api_key: Optional[str] = None,
+        **kwargs,
+    ):
+        headers = {}
+        if api_key:
+            headers = {"Authorization": f"Key {api_key}"}
+        super().__init__(url=url, aiohttp_session=aiohttp_session, headers=headers, **kwargs)

src/pipecat/audio/turn/smart_turn/http_smart_turn.py

@@ -0,0 +1,80 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import asyncio
+import io
+from typing import Any, Dict
+
+import aiohttp
+import numpy as np
+from loguru import logger
+
+from pipecat.audio.turn.smart_turn.base_smart_turn import BaseSmartTurn, SmartTurnTimeoutException
+
+
+class HttpSmartTurnAnalyzer(BaseSmartTurn):
+    def __init__(
+        self,
+        *,
+        url: str,
+        aiohttp_session: aiohttp.ClientSession,
+        headers: Dict[str, str] = {},
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self._url = url
+        self._headers = headers
+        self._aiohttp_session = aiohttp_session
+
+    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
+
+    async def _send_raw_request(self, data_bytes: bytes) -> Dict[str, Any]:
+        headers = {"Content-Type": "application/octet-stream"}
+        headers.update(self._headers)
+        logger.trace(f"Sending {len(data_bytes)} bytes as raw body to {self._url}...")
+        try:
+            timeout = aiohttp.ClientTimeout(total=self._params.stop_secs)
+
+            async with self._aiohttp_session.post(
+                self._url, data=data_bytes, headers=headers, timeout=timeout
+            ) as response:
+                logger.trace("\n--- Response ---")
+                logger.trace(f"Status Code: {response.status}")
+
+                if response.status == 200:
+                    try:
+                        json_data = await response.json()
+                        logger.trace("Response JSON:")
+                        logger.trace(json_data)
+                        return json_data
+                    except aiohttp.ContentTypeError:
+                        # Non-JSON response
+                        text = await response.text()
+                        logger.trace("Response Content (non-JSON):")
+                        logger.trace(text)
+                        raise Exception(f"Non-JSON response: {text}")
+                else:
+                    error_text = await response.text()
+                    logger.trace("Response Content (Error):")
+                    logger.trace(error_text)
+                    response.raise_for_status()
+
+        except asyncio.TimeoutError:
+            logger.error(f"Request timed out after {self._params.stop_secs} seconds")
+            raise SmartTurnTimeoutException(f"Request exceeded {self._params.stop_secs} seconds.")
+        except aiohttp.ClientError 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.")
+
+    async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
+        serialized_array = self._serialize_array(audio_array)
+        return await self._send_raw_request(serialized_array)

src/pipecat/audio/turn/smart_turn/local_coreml_smart_turn.py

@@ -5,17 +5,16 @@
 #
 
 
-import os
+from typing import Any, Dict
-from typing import Dict
 
 import numpy as np
-import torch
 from loguru import logger
 
-from pipecat.audio.turn.base_smart_turn import BaseSmartTurn
+from pipecat.audio.turn.smart_turn.base_smart_turn import BaseSmartTurn
 
 try:
     import coremltools as ct
+    import torch
     from transformers import AutoFeatureExtractor
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
@@ -26,7 +25,7 @@ except ModuleNotFoundError as e:
 
 
 class LocalCoreMLSmartTurnAnalyzer(BaseSmartTurn):
-    def __init__(self, smart_turn_model_path: str, **kwargs):
+    def __init__(self, *, smart_turn_model_path: str, **kwargs):
         super().__init__(**kwargs)
 
         if not smart_turn_model_path:
@@ -41,7 +40,7 @@ class LocalCoreMLSmartTurnAnalyzer(BaseSmartTurn):
         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]:
+    async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]:
         inputs = self._turn_processor(
             audio_array,
             sampling_rate=16000,

src/pipecat/transports/base_input.py

@@ -222,12 +222,8 @@ class BaseInputTransport(FrameProcessor):
 
     async def _handle_end_of_turn(self):
         if self.turn_analyzer:
-            state, prediction = await self.get_event_loop().run_in_executor(
+            state, prediction = await self.turn_analyzer.analyze_end_of_turn()
-                self._executor, self.turn_analyzer.analyze_end_of_turn
-            )
-
             await self._handle_prediction_result(prediction)
-
             await self._handle_end_of_turn_complete(state)
 
     async def _handle_end_of_turn_complete(self, state: EndOfTurnState):