diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index 8d7cd8647..704d41806 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/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]]: - state, result = self._process_speech_segment(self._audio_buffer) + async def analyze_end_of_turn(self) -> Tuple[EndOfTurnState, Optional[MetricsData]]: + 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) - state = ( - EndOfTurnState.COMPLETE if result["prediction"] == 1 else EndOfTurnState.INCOMPLETE - ) - end_time = time.perf_counter() + try: + result = await self._predict_endpoint(segment_audio) + state = ( + EndOfTurnState.COMPLETE + if result["prediction"] == 1 + else EndOfTurnState.INCOMPLETE + ) + end_time = time.perf_counter() - # Calculate processing time - e2e_processing_time_ms = (end_time - start_time) * 1000 + # Calculate processing time + e2e_processing_time_ms = (end_time - start_time) * 1000 - # Prepare the result data - result_data = SmartTurnMetricsData( - processor="BaseSmartTurn", - is_complete=result["prediction"] == 1, - probability=result["probability"], - inference_time_ms=result.get("inference_time", 0) * 1000, - server_total_time_ms=result.get("total_time", 0) * 1000, - e2e_processing_time_ms=e2e_processing_time_ms, - ) + # Prepare the result data + result_data = SmartTurnMetricsData( + processor="BaseSmartTurn", + is_complete=result["prediction"] == 1, + probability=result["probability"], + inference_time_ms=result.get("inference_time", 0) * 1000, + server_total_time_ms=result.get("total_time", 0) * 1000, + 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: diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py index b35630c1f..fd4f18d66 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/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: diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index 665e4b64d..923eacb76 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/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, diff --git a/src/pipecat/audio/turn/smart_turn.py b/src/pipecat/audio/turn/smart_turn.py index 5378e4b5b..b540fb187 100644 --- a/src/pipecat/audio/turn/smart_turn.py +++ b/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.session = requests.Session() - self.session.headers.update({"Connection": "keep-alive"}) + self.client = httpx.AsyncClient( + headers={"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()) - return response.json() - except requests.exceptions.JSONDecodeError: + logger.trace(json_data) + return json_data + 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) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 64d6f679d..3e3760be1 100644 --- a/src/pipecat/transports/base_input.py +++ b/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( - self._executor, self.turn_analyzer.analyze_end_of_turn - ) - + state, prediction = await 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):