diff --git a/CHANGELOG.md b/CHANGELOG.md index a22fd9c5b..04f9be3fa 100644 --- a/CHANGELOG.md +++ b/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 - inference or `LocalCoreMLSmartTurnAnalyzer()` for on-device inference using - Core ML. + parameter. You can now choose between `HttpSmartTurnAnalyzer()` or + `FalSmartTurnAnalyzer()` for remote inference or + `LocalCoreMLSmartTurnAnalyzer()` for on-device inference using Core ML. - `DeepgramTTSService` accepts `base_url` argument again, allowing you to connect to an on-prem service. diff --git a/dot-env.template b/dot-env.template index eb87366aa..8033bdfba 100644 --- a/dot-env.template +++ b/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= diff --git a/examples/foundational/38-smart-turn-fal.py b/examples/foundational/38-smart-turn-fal.py new file mode 100644 index 000000000..9b22c63f6 --- /dev/null +++ b/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() diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn.py deleted file mode 100644 index 6bace018b..000000000 --- a/examples/foundational/38-smart-turn.py +++ /dev/null @@ -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() diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-smart-turn-local-coreml.py similarity index 96% rename from examples/foundational/38a-local-smart-turn.py rename to examples/foundational/38a-smart-turn-local-coreml.py index c1260c248..4fa889656 100644 --- a/examples/foundational/38a-local-smart-turn.py +++ b/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.local_smart_turn import LocalCoreMLSmartTurnAnalyzer +from pipecat.audio.turn.smart_turn.base_smart_turn import SmartTurnParams +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 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/smart_turn.py b/src/pipecat/audio/turn/smart_turn.py deleted file mode 100644 index 5378e4b5b..000000000 --- a/src/pipecat/audio/turn/smart_turn.py +++ /dev/null @@ -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) diff --git a/src/pipecat/audio/turn/smart_turn/__init__.py b/src/pipecat/audio/turn/smart_turn/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/smart_turn/base_smart_turn.py similarity index 69% rename from src/pipecat/audio/turn/base_smart_turn.py rename to src/pipecat/audio/turn/smart_turn/base_smart_turn.py index 8d7cd8647..a45ac87e6 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/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]]: - 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}") @@ -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) - 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,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: diff --git a/src/pipecat/audio/turn/smart_turn/fal_smart_turn.py b/src/pipecat/audio/turn/smart_turn/fal_smart_turn.py new file mode 100644 index 000000000..9e3a85b56 --- /dev/null +++ b/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) diff --git a/src/pipecat/audio/turn/smart_turn/http_smart_turn.py b/src/pipecat/audio/turn/smart_turn/http_smart_turn.py new file mode 100644 index 000000000..4f542f81d --- /dev/null +++ b/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) diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/smart_turn/local_coreml_smart_turn.py similarity index 88% rename from src/pipecat/audio/turn/local_smart_turn.py rename to src/pipecat/audio/turn/smart_turn/local_coreml_smart_turn.py index 665e4b64d..88d6530bd 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn/local_coreml_smart_turn.py @@ -5,17 +5,16 @@ # -import os -from typing import Dict +from typing import Any, 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, 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):