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

src/pipecat/audio/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()
@@ -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}")
@@ -101,7 +105,9 @@ class BaseSmartTurn(BaseTurnAnalyzer):
         self._speech_start_time = None
         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,7 +179,7 @@ class BaseSmartTurn(BaseTurnAnalyzer):
         return state, result_data
 
     @abstractmethod
-    def _predict_endpoint(self, buffer: np.ndarray) -> Dict[str, Any]:
+    async def _predict_endpoint(self, buffer: np.ndarray) -> Dict[str, Any]:
         """Abstract method to predict if a turn has ended based on audio.
 
         Args:

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/local_smart_turn.py

@@ -41,7 +41,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/audio/turn/smart_turn.py

@@ -6,14 +6,13 @@
 
 
 import io
-import os
 from typing import Dict
 
+import httpx
 import numpy as np
-import requests
 from loguru import logger
 
-from pipecat.audio.turn.base_smart_turn import BaseSmartTurn
+from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, SmartTurnTimeoutException
 
 
 class SmartTurnAnalyzer(BaseSmartTurn):
@@ -25,9 +24,10 @@ class SmartTurnAnalyzer(BaseSmartTurn):
             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.client = httpx.AsyncClient(
-        self.session = requests.Session()
+            headers={"Connection": "keep-alive"},
-        self.session.headers.update({"Connection": "keep-alive"})
+            timeout=httpx.Timeout(self._params.stop_secs),
+        )
 
     def _serialize_array(self, audio_array: np.ndarray) -> bytes:
         logger.trace("Serializing NumPy array to bytes...")
@@ -37,28 +37,28 @@ class SmartTurnAnalyzer(BaseSmartTurn):
         logger.trace(f"Serialized size: {len(serialized_bytes)} bytes")
         return serialized_bytes
 
-    def _send_raw_request(self, data_bytes: bytes):
+    async 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(
+            response = await self.client.post(
                 self.remote_smart_turn_url,
-                data=data_bytes,
+                content=data_bytes,
                 headers=headers,
-                timeout=60,
             )
 
             logger.trace("\n--- Response ---")
             logger.trace(f"Status Code: {response.status_code}")
 
-            if response.ok:
+            if response.is_success:
                 try:
+                    json_data = response.json()
                     logger.trace("Response JSON:")
-                    logger.trace(response.json())
+                    logger.trace(json_data)
-                    return response.json()
+                    return json_data
-                except requests.exceptions.JSONDecodeError:
+                except httpx.DecodingError:
                     logger.trace("Response Content (non-JSON):")
                     logger.trace(response.text)
             else:
@@ -66,10 +66,13 @@ class SmartTurnAnalyzer(BaseSmartTurn):
                 logger.trace(response.text)
                 response.raise_for_status()
 
-        except requests.exceptions.RequestException as e:
+        except httpx.TimeoutException:
+            logger.error(f"Request timed out after {self._params.stop_secs} seconds")
+            raise SmartTurnTimeoutException(f"Request exceeded {self._params.stop_secs} seconds.")
+        except httpx.RequestError 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]:
+    async 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)
+        return await self._send_raw_request(serialized_array)

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):