diff --git a/CHANGELOG.md b/CHANGELOG.md index 14645859d..4c928028f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Include OpenAI-based LLM services cached tokens to `MetricsFrame`. +## Fixed + +- Fixed an issue where local SmartTurn was not being ran in a separate thread. + ## [0.0.86] - 2025-09-24 ### Added diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py index f64655fb6..929c2ac46 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/src/pipecat/audio/turn/base_turn_analyzer.py @@ -14,6 +14,8 @@ from abc import ABC, abstractmethod from enum import Enum from typing import Optional, Tuple +from pydantic import BaseModel + from pipecat.metrics.metrics import MetricsData @@ -29,6 +31,12 @@ class EndOfTurnState(Enum): INCOMPLETE = 2 +class BaseTurnParams(BaseModel): + """Base class for turn analyzer parameters.""" + + pass + + class BaseTurnAnalyzer(ABC): """Abstract base class for analyzing user end of turn. @@ -78,7 +86,7 @@ class BaseTurnAnalyzer(ABC): @property @abstractmethod - def params(self): + def params(self) -> BaseTurnParams: """Get the current turn analyzer parameters. Returns: diff --git a/src/pipecat/audio/turn/smart_turn/base_smart_turn.py b/src/pipecat/audio/turn/smart_turn/base_smart_turn.py index f72ac7d84..4fba0f703 100644 --- a/src/pipecat/audio/turn/smart_turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn/base_smart_turn.py @@ -11,15 +11,17 @@ machine learning models to determine when a user has finished speaking, going beyond simple silence-based detection. """ +import asyncio import time from abc import abstractmethod +from concurrent.futures import ThreadPoolExecutor from typing import Any, Dict, Optional, Tuple import numpy as np from loguru import logger from pydantic import BaseModel -from pipecat.audio.turn.base_turn_analyzer import BaseTurnAnalyzer, EndOfTurnState +from pipecat.audio.turn.base_turn_analyzer import BaseTurnAnalyzer, BaseTurnParams, EndOfTurnState from pipecat.metrics.metrics import MetricsData, SmartTurnMetricsData # Default timing parameters @@ -29,7 +31,7 @@ MAX_DURATION_SECONDS = 8 # Max allowed segment duration USE_ONLY_LAST_VAD_SEGMENT = True -class SmartTurnParams(BaseModel): +class SmartTurnParams(BaseTurnParams): """Configuration parameters for smart turn analysis. Parameters: @@ -77,6 +79,9 @@ class BaseSmartTurn(BaseTurnAnalyzer): self._speech_triggered = False self._silence_ms = 0 self._speech_start_time = 0 + # Thread executor that will run the model. We only need one thread per + # analyzer because one analyzer just handles one audio stream. + self._executor = ThreadPoolExecutor(max_workers=1) @property def speech_triggered(self) -> bool: @@ -151,7 +156,10 @@ class BaseSmartTurn(BaseTurnAnalyzer): Tuple containing the end-of-turn state and optional metrics data from the ML model analysis. """ - state, result = await self._process_speech_segment(self._audio_buffer) + loop = asyncio.get_running_loop() + state, result = await loop.run_in_executor( + self._executor, 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}") @@ -169,9 +177,7 @@ class BaseSmartTurn(BaseTurnAnalyzer): self._speech_start_time = 0 self._silence_ms = 0 - async def _process_speech_segment( - self, audio_buffer - ) -> Tuple[EndOfTurnState, Optional[MetricsData]]: + def _process_speech_segment(self, audio_buffer) -> Tuple[EndOfTurnState, Optional[MetricsData]]: """Process accumulated audio segment using ML model.""" state = EndOfTurnState.INCOMPLETE @@ -203,7 +209,7 @@ class BaseSmartTurn(BaseTurnAnalyzer): if len(segment_audio) > 0: start_time = time.perf_counter() try: - result = await self._predict_endpoint(segment_audio) + result = self._predict_endpoint(segment_audio) state = ( EndOfTurnState.COMPLETE if result["prediction"] == 1 @@ -249,6 +255,6 @@ class BaseSmartTurn(BaseTurnAnalyzer): return state, result_data @abstractmethod - async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]: + def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]: """Predict end-of-turn using ML model from audio data.""" pass diff --git a/src/pipecat/audio/turn/smart_turn/http_smart_turn.py b/src/pipecat/audio/turn/smart_turn/http_smart_turn.py index c28727f78..878c0620c 100644 --- a/src/pipecat/audio/turn/smart_turn/http_smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn/http_smart_turn.py @@ -104,11 +104,15 @@ class HttpSmartTurnAnalyzer(BaseSmartTurn): 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]: + def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]: """Predict end-of-turn using remote HTTP ML service.""" try: serialized_array = self._serialize_array(audio_array) - return await self._send_raw_request(serialized_array) + loop = asyncio.get_running_loop() + future = asyncio.run_coroutine_threadsafe( + self._send_raw_request(serialized_array), loop + ) + return future.result() except Exception as e: logger.error(f"Smart turn prediction failed: {str(e)}") # Return an incomplete prediction when a failure occurs diff --git a/src/pipecat/audio/turn/smart_turn/local_smart_turn.py b/src/pipecat/audio/turn/smart_turn/local_smart_turn.py index ed67dad12..e52994f1d 100644 --- a/src/pipecat/audio/turn/smart_turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn/local_smart_turn.py @@ -64,7 +64,7 @@ class LocalSmartTurnAnalyzer(BaseSmartTurn): self._turn_model.eval() logger.debug("Loaded Local Smart Turn") - async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]: + def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]: """Predict end-of-turn using local PyTorch model.""" inputs = self._turn_processor( audio_array, diff --git a/src/pipecat/audio/turn/smart_turn/local_smart_turn_v2.py b/src/pipecat/audio/turn/smart_turn/local_smart_turn_v2.py index 06263c699..2a1cecda1 100644 --- a/src/pipecat/audio/turn/smart_turn/local_smart_turn_v2.py +++ b/src/pipecat/audio/turn/smart_turn/local_smart_turn_v2.py @@ -73,7 +73,7 @@ class LocalSmartTurnAnalyzerV2(BaseSmartTurn): self._turn_model.eval() logger.debug("Loaded Local Smart Turn v2") - async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]: + def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]: """Predict end-of-turn using local PyTorch model.""" inputs = self._turn_processor( audio_array, diff --git a/src/pipecat/audio/turn/smart_turn/local_smart_turn_v3.py b/src/pipecat/audio/turn/smart_turn/local_smart_turn_v3.py index 07e710da6..ec60c293a 100644 --- a/src/pipecat/audio/turn/smart_turn/local_smart_turn_v3.py +++ b/src/pipecat/audio/turn/smart_turn/local_smart_turn_v3.py @@ -77,7 +77,7 @@ class LocalSmartTurnAnalyzerV3(BaseSmartTurn): logger.debug("Loaded Local Smart Turn v3") - async def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]: + def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, Any]: """Predict end-of-turn using local ONNX model.""" def truncate_audio_to_last_n_seconds(audio_array, n_seconds=8, sample_rate=16000):