From 7358bc6428ad5390aad8e62e69c43a145e000e08 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 22 Apr 2025 10:35:14 -0300 Subject: [PATCH 1/5] =?UTF-8?q?Returning=20the=20turn=20as=20complete=20if?= =?UTF-8?q?=20the=20request=20don=E2=80=99t=20return=20a=20result=20within?= =?UTF-8?q?=20SmartTurnParams=20stop=5Fsecs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/pipecat/audio/turn/base_smart_turn.py | 67 ++++++++++++-------- src/pipecat/audio/turn/base_turn_analyzer.py | 2 +- src/pipecat/audio/turn/local_smart_turn.py | 2 +- src/pipecat/audio/turn/smart_turn.py | 37 ++++++----- src/pipecat/transports/base_input.py | 6 +- 5 files changed, 65 insertions(+), 49 deletions(-) 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): From cc9901a82f9d108c21b5d9172a498acf09e3c890 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Tue, 22 Apr 2025 17:14:19 -0400 Subject: [PATCH 2/5] Replace httpx with aiohttp --- examples/foundational/38-smart-turn.py | 124 +++++++++++++------------ src/pipecat/audio/turn/smart_turn.py | 58 ++++++------ 2 files changed, 92 insertions(+), 90 deletions(-) diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn.py index 6bace018b..929a22e68 100644 --- a/examples/foundational/38-smart-turn.py +++ b/examples/foundational/38-smart-turn.py @@ -6,6 +6,7 @@ import os +import aiohttp from dotenv import load_dotenv from loguru import logger @@ -31,78 +32,79 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): 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), - ), - ) + 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=SmartTurnAnalyzer(url=remote_smart_turn_url, aiohttp_session=session), + ), + ) - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + 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 - ) + 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")) + 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 + 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.", + }, ] - ) - task = PipelineTask( - pipeline, - params=PipelineParams( - allow_interruptions=True, - enable_metrics=True, - enable_usage_metrics=True, - report_only_initial_ttfb=True, - ), - ) + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) - @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()]) + 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 + ] + ) - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") + 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_closed") - async def on_client_closed(transport, client): - logger.info(f"Client closed connection") - await task.cancel() + @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()]) - runner = PipelineRunner(handle_sigint=False) + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") - await runner.run(task) + @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__": diff --git a/src/pipecat/audio/turn/smart_turn.py b/src/pipecat/audio/turn/smart_turn.py index b540fb187..8ac165943 100644 --- a/src/pipecat/audio/turn/smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn.py @@ -5,10 +5,11 @@ # +import asyncio import io from typing import Dict -import httpx +import aiohttp import numpy as np from loguru import logger @@ -16,19 +17,15 @@ from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, SmartTurnTimeoutEx class SmartTurnAnalyzer(BaseSmartTurn): - def __init__(self, url: str, **kwargs): + def __init__(self, url: str, aiohttp_session: aiohttp.ClientSession, **kwargs): super().__init__(**kwargs) self.remote_smart_turn_url = url + self._aiohttp_session = aiohttp_session 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.") - 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...") buffer = io.BytesIO() @@ -43,33 +40,36 @@ class SmartTurnAnalyzer(BaseSmartTurn): f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..." ) try: - response = await self.client.post( - self.remote_smart_turn_url, - content=data_bytes, - headers=headers, - ) + timeout = aiohttp.ClientTimeout(total=self._params.stop_secs) - logger.trace("\n--- Response ---") - logger.trace(f"Status Code: {response.status_code}") + async with self._aiohttp_session.post( + self.remote_smart_turn_url, data=data_bytes, headers=headers, timeout=timeout + ) as response: + logger.trace("\n--- Response ---") + logger.trace(f"Status Code: {response.status}") - if response.is_success: - try: - json_data = response.json() - logger.trace("Response JSON:") - logger.trace(json_data) - return json_data - except httpx.DecodingError: - logger.trace("Response Content (non-JSON):") - logger.trace(response.text) - else: - logger.trace("Response Content (Error):") - logger.trace(response.text) - response.raise_for_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 httpx.TimeoutException: + 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 httpx.RequestError as e: + 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.") From 50e8d82ecee732c312656ea43383f1afe07d8067 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 22 Apr 2025 14:39:02 -0700 Subject: [PATCH 3/5] SmartTurn: some linting cleanup --- src/pipecat/audio/turn/base_smart_turn.py | 10 +++++----- src/pipecat/audio/turn/local_smart_turn.py | 7 +++---- src/pipecat/audio/turn/smart_turn.py | 18 ++++++------------ 3 files changed, 14 insertions(+), 21 deletions(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index 704d41806..a45ac87e6 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -46,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: @@ -64,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: @@ -102,7 +102,7 @@ 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 async def _process_speech_segment( @@ -179,11 +179,11 @@ class BaseSmartTurn(BaseTurnAnalyzer): return state, result_data @abstractmethod - async 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/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index 923eacb76..5e59f474a 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_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 try: import coremltools as ct + import torch from transformers import AutoFeatureExtractor except ModuleNotFoundError as e: logger.error(f"Exception: {e}") @@ -41,7 +40,7 @@ class LocalCoreMLSmartTurnAnalyzer(BaseSmartTurn): self._turn_model = ct.models.MLModel(core_ml_model_path) logger.debug("Loaded Local Smart Turn") - async 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 8ac165943..2af94e139 100644 --- a/src/pipecat/audio/turn/smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn.py @@ -7,7 +7,7 @@ import asyncio import io -from typing import Dict +from typing import Any, Dict import aiohttp import numpy as np @@ -19,13 +19,9 @@ from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, SmartTurnTimeoutEx class SmartTurnAnalyzer(BaseSmartTurn): def __init__(self, url: str, aiohttp_session: aiohttp.ClientSession, **kwargs): super().__init__(**kwargs) - self.remote_smart_turn_url = url + self._url = url self._aiohttp_session = aiohttp_session - 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.") - def _serialize_array(self, audio_array: np.ndarray) -> bytes: logger.trace("Serializing NumPy array to bytes...") buffer = io.BytesIO() @@ -34,16 +30,14 @@ class SmartTurnAnalyzer(BaseSmartTurn): logger.trace(f"Serialized size: {len(serialized_bytes)} bytes") return serialized_bytes - async def _send_raw_request(self, data_bytes: bytes): + async def _send_raw_request(self, data_bytes: bytes) -> Dict[str, Any]: headers = {"Content-Type": "application/octet-stream"} - logger.trace( - f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..." - ) + 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.remote_smart_turn_url, data=data_bytes, headers=headers, timeout=timeout + self._url, data=data_bytes, headers=headers, timeout=timeout ) as response: logger.trace("\n--- Response ---") logger.trace(f"Status Code: {response.status}") @@ -73,6 +67,6 @@ class SmartTurnAnalyzer(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]: + 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) From ae60d42016ac6c5592af50b649ad23059dbda095 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 22 Apr 2025 15:13:12 -0700 Subject: [PATCH 4/5] s/SmartTurnAnalyzer/HttpSmartTurnAnalyzer/ and add FalSmartTurnAnalyzer --- examples/foundational/38-smart-turn.py | 6 +++-- examples/foundational/38a-local-smart-turn.py | 2 +- src/pipecat/audio/turn/fal_smart_turn.py | 26 +++++++++++++++++++ .../{smart_turn.py => http_smart_turn.py} | 14 +++++++--- ...art_turn.py => local_coreml_smart_turn.py} | 0 5 files changed, 42 insertions(+), 6 deletions(-) create mode 100644 src/pipecat/audio/turn/fal_smart_turn.py rename src/pipecat/audio/turn/{smart_turn.py => http_smart_turn.py} (90%) rename src/pipecat/audio/turn/{local_smart_turn.py => local_coreml_smart_turn.py} (100%) diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn.py index 929a22e68..ea32c69ed 100644 --- a/examples/foundational/38-smart-turn.py +++ b/examples/foundational/38-smart-turn.py @@ -10,7 +10,7 @@ import aiohttp from dotenv import load_dotenv from loguru import logger -from pipecat.audio.turn.smart_turn import SmartTurnAnalyzer +from pipecat.audio.turn.http_smart_turn import HttpSmartTurnAnalyzer from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.pipeline.pipeline import Pipeline @@ -41,7 +41,9 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), vad_audio_passthrough=True, - turn_analyzer=SmartTurnAnalyzer(url=remote_smart_turn_url, aiohttp_session=session), + turn_analyzer=HttpSmartTurnAnalyzer( + url=remote_smart_turn_url, aiohttp_session=session + ), ), ) diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py index c1260c248..856cc69b6 100644 --- a/examples/foundational/38a-local-smart-turn.py +++ b/examples/foundational/38a-local-smart-turn.py @@ -10,7 +10,7 @@ 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.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/fal_smart_turn.py b/src/pipecat/audio/turn/fal_smart_turn.py new file mode 100644 index 000000000..7e7974ed1 --- /dev/null +++ b/src/pipecat/audio/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.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.py b/src/pipecat/audio/turn/http_smart_turn.py similarity index 90% rename from src/pipecat/audio/turn/smart_turn.py rename to src/pipecat/audio/turn/http_smart_turn.py index 2af94e139..4c1fb8491 100644 --- a/src/pipecat/audio/turn/smart_turn.py +++ b/src/pipecat/audio/turn/http_smart_turn.py @@ -4,7 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # - import asyncio import io from typing import Any, Dict @@ -16,10 +15,18 @@ from loguru import logger from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, SmartTurnTimeoutException -class SmartTurnAnalyzer(BaseSmartTurn): - def __init__(self, url: str, aiohttp_session: aiohttp.ClientSession, **kwargs): +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: @@ -32,6 +39,7 @@ class SmartTurnAnalyzer(BaseSmartTurn): 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) diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_coreml_smart_turn.py similarity index 100% rename from src/pipecat/audio/turn/local_smart_turn.py rename to src/pipecat/audio/turn/local_coreml_smart_turn.py From e7da08dab12c5e1cd27c23862eeb8dffeb6ab484 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 22 Apr 2025 15:25:02 -0700 Subject: [PATCH 5/5] move smart turn files to audio.turn.smart_turn package --- CHANGELOG.md | 6 +++--- dot-env.template | 4 ++-- .../{38-smart-turn.py => 38-smart-turn-fal.py} | 8 +++----- ...local-smart-turn.py => 38a-smart-turn-local-coreml.py} | 4 ++-- src/pipecat/audio/turn/smart_turn/__init__.py | 0 .../audio/turn/{ => smart_turn}/base_smart_turn.py | 0 src/pipecat/audio/turn/{ => smart_turn}/fal_smart_turn.py | 2 +- .../audio/turn/{ => smart_turn}/http_smart_turn.py | 2 +- .../turn/{ => smart_turn}/local_coreml_smart_turn.py | 4 ++-- 9 files changed, 14 insertions(+), 16 deletions(-) rename examples/foundational/{38-smart-turn.py => 38-smart-turn-fal.py} (93%) rename examples/foundational/{38a-local-smart-turn.py => 38a-smart-turn-local-coreml.py} (96%) create mode 100644 src/pipecat/audio/turn/smart_turn/__init__.py rename src/pipecat/audio/turn/{ => smart_turn}/base_smart_turn.py (100%) rename src/pipecat/audio/turn/{ => smart_turn}/fal_smart_turn.py (88%) rename src/pipecat/audio/turn/{ => smart_turn}/http_smart_turn.py (96%) rename src/pipecat/audio/turn/{ => smart_turn}/local_coreml_smart_turn.py (93%) diff --git a/CHANGELOG.md b/CHANGELOG.md index 870550b9c..e974f5eee 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,9 +17,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 18ddbee0a..c5a34d708 100644 --- a/dot-env.template +++ b/dot-env.template @@ -95,5 +95,5 @@ OPENROUTER_API_KEY=... PIPER_BASE_URL=... # Smart turn -LOCAL_SMART_TURN_MODEL_PATH= -REMOTE_SMART_TURN_URL= \ No newline at end of file +LOCAL_SMART_TURN_MODEL_PATH=... +FAL_SMART_TURN_API_KEY=... diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn-fal.py similarity index 93% rename from examples/foundational/38-smart-turn.py rename to examples/foundational/38-smart-turn-fal.py index ea32c69ed..9b22c63f6 100644 --- a/examples/foundational/38-smart-turn.py +++ b/examples/foundational/38-smart-turn-fal.py @@ -10,7 +10,7 @@ import aiohttp from dotenv import load_dotenv from loguru import logger -from pipecat.audio.turn.http_smart_turn import HttpSmartTurnAnalyzer +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 @@ -30,8 +30,6 @@ 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") - async with aiohttp.ClientSession() as session: transport = SmallWebRTCTransport( webrtc_connection=webrtc_connection, @@ -41,8 +39,8 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), vad_audio_passthrough=True, - turn_analyzer=HttpSmartTurnAnalyzer( - url=remote_smart_turn_url, aiohttp_session=session + turn_analyzer=FalSmartTurnAnalyzer( + api_key=os.getenv("FAL_SMART_TURN_API_KEY"), aiohttp_session=session ), ), ) 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 856cc69b6..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_coreml_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/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 100% rename from src/pipecat/audio/turn/base_smart_turn.py rename to src/pipecat/audio/turn/smart_turn/base_smart_turn.py diff --git a/src/pipecat/audio/turn/fal_smart_turn.py b/src/pipecat/audio/turn/smart_turn/fal_smart_turn.py similarity index 88% rename from src/pipecat/audio/turn/fal_smart_turn.py rename to src/pipecat/audio/turn/smart_turn/fal_smart_turn.py index 7e7974ed1..9e3a85b56 100644 --- a/src/pipecat/audio/turn/fal_smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn/fal_smart_turn.py @@ -8,7 +8,7 @@ from typing import Optional import aiohttp -from pipecat.audio.turn.http_smart_turn import HttpSmartTurnAnalyzer +from pipecat.audio.turn.smart_turn.http_smart_turn import HttpSmartTurnAnalyzer class FalSmartTurnAnalyzer(HttpSmartTurnAnalyzer): diff --git a/src/pipecat/audio/turn/http_smart_turn.py b/src/pipecat/audio/turn/smart_turn/http_smart_turn.py similarity index 96% rename from src/pipecat/audio/turn/http_smart_turn.py rename to src/pipecat/audio/turn/smart_turn/http_smart_turn.py index 4c1fb8491..4f542f81d 100644 --- a/src/pipecat/audio/turn/http_smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn/http_smart_turn.py @@ -12,7 +12,7 @@ import aiohttp import numpy as np from loguru import logger -from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, SmartTurnTimeoutException +from pipecat.audio.turn.smart_turn.base_smart_turn import BaseSmartTurn, SmartTurnTimeoutException class HttpSmartTurnAnalyzer(BaseSmartTurn): diff --git a/src/pipecat/audio/turn/local_coreml_smart_turn.py b/src/pipecat/audio/turn/smart_turn/local_coreml_smart_turn.py similarity index 93% rename from src/pipecat/audio/turn/local_coreml_smart_turn.py rename to src/pipecat/audio/turn/smart_turn/local_coreml_smart_turn.py index 5e59f474a..88d6530bd 100644 --- a/src/pipecat/audio/turn/local_coreml_smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn/local_coreml_smart_turn.py @@ -10,7 +10,7 @@ from typing import Any, Dict import numpy as np 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 @@ -25,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: