From 3ebef9346fce6ffeae0e3a386a31edb6e5141fce Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 06:33:42 -0300 Subject: [PATCH] Adding support for RemoteSmartTurn --- dot-env.template | 5 +- examples/foundational/38-smart-turn.py | 109 +++++++++++++++++++ src/pipecat/audio/turn/base_turn_analyzer.py | 11 ++ src/pipecat/audio/turn/local_smart_turn.py | 12 -- src/pipecat/audio/turn/remote_smart_turn.py | 73 +++++++++++++ 5 files changed, 196 insertions(+), 14 deletions(-) create mode 100644 examples/foundational/38-smart-turn.py create mode 100644 src/pipecat/audio/turn/remote_smart_turn.py diff --git a/dot-env.template b/dot-env.template index ae0df90e5..18ddbee0a 100644 --- a/dot-env.template +++ b/dot-env.template @@ -94,5 +94,6 @@ OPENROUTER_API_KEY=... # Piper PIPER_BASE_URL=... -# Local Smart turn -LOCAL_SMART_TURN_MODEL_PATH= \ No newline at end of file +# Smart turn +LOCAL_SMART_TURN_MODEL_PATH= +REMOTE_SMART_TURN_URL= \ No newline at end of file diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn.py new file mode 100644 index 000000000..251a13445 --- /dev/null +++ b/examples/foundational/38-smart-turn.py @@ -0,0 +1,109 @@ +# +# 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.remote_smart_turn import RemoteSmartTurnAnalyzer +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") + + 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, + end_of_turn_analyzer=RemoteSmartTurnAnalyzer(), + ), + ) + + 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"), model="gpt-4o") + + 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/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py index bb10c7df3..91b38bb00 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/src/pipecat/audio/turn/base_turn_analyzer.py @@ -154,4 +154,15 @@ class BaseEndOfTurnAnalyzer(ABC): @abstractmethod def _predict_endpoint(self, buffer: np.ndarray) -> Dict[str, any]: + """ + Predict whether an audio segment is complete (turn ended) or incomplete. + + Args: + audio_array: Numpy array containing audio samples at 16kHz + + Returns: + Dictionary containing prediction results: + - prediction: 1 for complete, 0 for incomplete + - probability: Probability of completion class + """ pass diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index efa6c81d1..a84bab1e0 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -60,18 +60,6 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): logger.debug("Loaded Local Smart Turn") def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, any]: - """ - Predict whether an audio segment is complete (turn ended) or incomplete. - - Args: - audio_array: Numpy array containing audio samples at 16kHz - - Returns: - Dictionary containing prediction results: - - prediction: 1 for complete, 0 for incomplete - - probability: Probability of completion class - """ - inputs = self._turn_processor( audio_array, sampling_rate=16000, diff --git a/src/pipecat/audio/turn/remote_smart_turn.py b/src/pipecat/audio/turn/remote_smart_turn.py new file mode 100644 index 000000000..a971dac37 --- /dev/null +++ b/src/pipecat/audio/turn/remote_smart_turn.py @@ -0,0 +1,73 @@ +# +# 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_turn_analyzer import ( + BaseEndOfTurnAnalyzer, +) + + +class RemoteSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): + def __init__(self): + super().__init__() + self.remote_smart_turn_url = os.getenv("REMOTE_SMART_TURN_URL") + + if not self.remote_smart_turn_url: + logger.error("REMOTE_SMART_TURN_URL is not set.") + raise Exception("REMOTE_SMART_TURN_URL environment variable must be provided.") + + def _serialize_array(self, audio_array: np.ndarray) -> bytes: + """Serializes a NumPy array into bytes using np.save.""" + logger.debug("Serializing NumPy array to bytes...") + buffer = io.BytesIO() + np.save(buffer, audio_array) # Saves in npy format + serialized_bytes = buffer.getvalue() + logger.debug(f"Serialized size: {len(serialized_bytes)} bytes") + return serialized_bytes + + def _send_raw_request(self, data_bytes: bytes): + """Sends the bytes as the raw request body.""" + headers = {"Content-Type": "application/octet-stream"} + logger.debug( + f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..." + ) + try: + response = requests.post( + self.remote_smart_turn_url, data=data_bytes, headers=headers, timeout=60 + ) # Added timeout + + logger.debug("\n--- Response ---") + logger.debug(f"Status Code: {response.status_code}") + + # Try to logger.debug JSON if successful, otherwise logger.debug text + if response.ok: + try: + logger.debug("Response JSON:") + logger.debug(response.json()) + return response.json() + except requests.exceptions.JSONDecodeError: + logger.debug("Response Content (non-JSON):") + logger.debug(response.text) + else: + logger.debug("Response Content (Error):") + logger.debug(response.text) + response.raise_for_status() # Raise an exception for bad status codes + + except requests.exceptions.RequestException as e: + logger.debug(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)