From 821e3032496d786ccfbc9152e5362e3270e5a63c Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 15 Apr 2025 10:21:40 -0300 Subject: [PATCH 01/34] Bringing Aleix initial implementation for the smart turn. --- src/pipecat/audio/turn/__init__.py | 0 src/pipecat/audio/turn/base_turn_analyzer.py | 32 ++++++++++++++++++++ src/pipecat/frames/frames.py | 9 ++++++ src/pipecat/transports/base_input.py | 29 ++++++++++++++++++ src/pipecat/transports/base_transport.py | 2 ++ 5 files changed, 72 insertions(+) create mode 100644 src/pipecat/audio/turn/__init__.py create mode 100644 src/pipecat/audio/turn/base_turn_analyzer.py diff --git a/src/pipecat/audio/turn/__init__.py b/src/pipecat/audio/turn/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py new file mode 100644 index 000000000..1dffa563b --- /dev/null +++ b/src/pipecat/audio/turn/base_turn_analyzer.py @@ -0,0 +1,32 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +from abc import ABC, abstractmethod +from enum import Enum +from typing import Optional + + +class EndOfTurnState(Enum): + COMPLETE = 1 + INCOMPLETE = 2 + + +class BaseEndOfTurnAnalyzer(ABC): + def __init__(self, *, sample_rate: Optional[int] = None): + self._init_sample_rate = sample_rate + self._sample_rate = 0 + + @property + def sample_rate(self) -> int: + return self._sample_rate + + def set_sample_rate(self, sample_rate: int): + self._sample_rate = self._init_sample_rate or sample_rate + + @abstractmethod + def analyze_audio(self, buffer: bytes) -> EndOfTurnState: + pass diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 72acf1a2a..4301c5c1c 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -571,6 +571,15 @@ class EmulateUserStoppedSpeakingFrame(SystemFrame): pass +@dataclass +class UserEndOfTurnFrame(SystemFrame): + """Emitted based on the Smart Turn model to indicate that the user has + completed their turn/sentence. + """ + + pass + + @dataclass class BotInterruptionFrame(SystemFrame): """Emitted by when the bot should be interrupted. This will mainly cause the diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 26f386576..153c2d65b 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -10,6 +10,7 @@ from typing import Optional from loguru import logger +from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer, EndOfTurnState from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState from pipecat.frames.frames import ( BotInterruptionFrame, @@ -24,6 +25,7 @@ from pipecat.frames.frames import ( StartInterruptionFrame, StopInterruptionFrame, SystemFrame, + UserEndOfTurnFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, VADParamsUpdateFrame, @@ -64,12 +66,19 @@ class BaseInputTransport(FrameProcessor): def vad_analyzer(self) -> Optional[VADAnalyzer]: return self._params.vad_analyzer + @property + def end_of_turn_analyzer(self) -> Optional[BaseEndOfTurnAnalyzer]: + return self._params.end_of_turn_analyzer + async def start(self, frame: StartFrame): self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate # Configure VAD analyzer. if self._params.vad_enabled and self._params.vad_analyzer: self._params.vad_analyzer.set_sample_rate(self._sample_rate) + # Configure End of turn analyzer. + if self._params.end_of_turn_analyzer: + self._params.end_of_turn_analyzer.set_sample_rate(self._sample_rate) # Start audio filter. if self._params.audio_in_filter: await self._params.audio_in_filter.start(self._sample_rate) @@ -198,8 +207,25 @@ class BaseInputTransport(FrameProcessor): vad_state = new_vad_state return vad_state + async def _end_of_turn_analyze(self, audio_frame: InputAudioRawFrame) -> EndOfTurnState: + state = EndOfTurnState.INCOMPLETE + if self.end_of_turn_analyzer: + state = await self.get_event_loop().run_in_executor( + self._executor, self.end_of_turn_analyzer.analyze_audio, audio_frame.audio + ) + return state + + async def _handle_end_of_turn( + self, audio_frame: InputAudioRawFrame, end_of_turn_state: EndOfTurnState + ): + new_eot_state = await self._end_of_turn_analyze(audio_frame) + if new_eot_state != end_of_turn_state: + await self.push_frame(UserEndOfTurnFrame()) + return new_eot_state + async def _audio_task_handler(self): vad_state: VADState = VADState.QUIET + end_of_turn_state: EndOfTurnState = EndOfTurnState.INCOMPLETE while True: frame: InputAudioRawFrame = await self._audio_in_queue.get() @@ -215,6 +241,9 @@ class BaseInputTransport(FrameProcessor): vad_state = await self._handle_vad(frame, vad_state) audio_passthrough = self._params.vad_audio_passthrough + if self._params.end_of_turn_analyzer: + end_of_turn_state = await self._handle_end_of_turn(frame, end_of_turn_state) + # Push audio downstream if passthrough. if audio_passthrough: await self.push_frame(frame) diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 411456071..6045609f7 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -11,6 +11,7 @@ from pydantic import BaseModel, ConfigDict from pipecat.audio.filters.base_audio_filter import BaseAudioFilter from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer +from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.processors.frame_processor import FrameProcessor from pipecat.utils.base_object import BaseObject @@ -41,6 +42,7 @@ class TransportParams(BaseModel): vad_enabled: bool = False vad_audio_passthrough: bool = False vad_analyzer: Optional[VADAnalyzer] = None + end_of_turn_analyzer: Optional[BaseEndOfTurnAnalyzer] = None class BaseTransport(BaseObject): From 6ab9a8ad7f2269e1f1af65e9b0fc17f6cb0feb20 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 15 Apr 2025 11:24:39 -0300 Subject: [PATCH 02/34] Starting to create a local smart turn --- examples/foundational/38a-local-smart-turn.py | 108 ++++++++++++++++++ src/pipecat/audio/turn/local_smart_turn.py | 51 +++++++++ src/pipecat/transports/base_input.py | 15 ++- 3 files changed, 172 insertions(+), 2 deletions(-) create mode 100644 examples/foundational/38a-local-smart-turn.py create mode 100644 src/pipecat/audio/turn/local_smart_turn.py diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py new file mode 100644 index 000000000..a9bf8e293 --- /dev/null +++ b/examples/foundational/38a-local-smart-turn.py @@ -0,0 +1,108 @@ +# +# 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.local_smart_turn import LocalSmartTurnAnalyzer +from pipecat.audio.vad.silero import SileroVADAnalyzer +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(), + vad_audio_passthrough=True, + end_of_turn_analyzer=LocalSmartTurnAnalyzer(), + ), + ) + + 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/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py new file mode 100644 index 000000000..c511ae2ba --- /dev/null +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -0,0 +1,51 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import numpy as np +from loguru import logger + +from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer, EndOfTurnState + + +class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): + def __init__(self): + super().__init__() + self._audio_buffer = bytearray() + + logger.debug("Loading Local Smart Turn model...") + + # TODO: implement it + + logger.debug("Loaded Local Smart Turn") + + def analyze_audio(self, buffer: bytes) -> EndOfTurnState: + self._audio_buffer += buffer + + # TODO: we probably don't need this + # Checking if we have at least 6 seconds of audio + # if len(self._audio_buffer) < 16000 * 2 * 6: + # return EndOfTurnState.INCOMPLETE + + audio_int16 = np.frombuffer(self._audio_buffer, dtype=np.int16) + + # Divide by 32768 because we have signed 16-bit data. + audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 + + # TODO: implement to use the smart turn + # for now it is always returning as complete only for testing it + prediction = 1 + + state = EndOfTurnState.COMPLETE if prediction == 1 else EndOfTurnState.INCOMPLETE + + if state == EndOfTurnState.COMPLETE: + # clears the buffer completely + self._audio_buffer = bytearray() + else: + # TODO: implement it + pass + + return state diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 153c2d65b..946feb063 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -172,9 +172,16 @@ class BaseInputTransport(FrameProcessor): elif isinstance(frame, UserStoppedSpeakingFrame): logger.debug("User stopped speaking") await self.push_frame(frame) + + # TODO check, we probably should change here as well. + # if the end of turn is enabled, we should only stop interruption after this point if self.interruptions_allowed: await self._stop_interruption() await self.push_frame(StopInterruptionFrame()) + elif isinstance(frame, UserEndOfTurnFrame): + logger.debug("User end of turn") + await self.push_frame(frame) + # TODO: implement to handle interruptions # # Audio input @@ -220,7 +227,7 @@ class BaseInputTransport(FrameProcessor): ): new_eot_state = await self._end_of_turn_analyze(audio_frame) if new_eot_state != end_of_turn_state: - await self.push_frame(UserEndOfTurnFrame()) + await self._handle_user_interruption(UserEndOfTurnFrame()) return new_eot_state async def _audio_task_handler(self): @@ -239,9 +246,13 @@ class BaseInputTransport(FrameProcessor): # changes from QUIET to SPEAKING and vice versa. if self._params.vad_enabled: vad_state = await self._handle_vad(frame, vad_state) + # TODO: need to check if we need to keep it later + if vad_state == VADState.QUIET: + end_of_turn_state = EndOfTurnState.INCOMPLETE audio_passthrough = self._params.vad_audio_passthrough - if self._params.end_of_turn_analyzer: + # We only need to check for completion if the user is speaking + if self._params.end_of_turn_analyzer and VADState.QUIET != vad_state: end_of_turn_state = await self._handle_end_of_turn(frame, end_of_turn_state) # Push audio downstream if passthrough. From 73874f6ec06cf81c743c746675a83c769fca7019 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 15 Apr 2025 12:11:06 -0300 Subject: [PATCH 03/34] Loading the smart turn model. --- dot-env.template | 5 ++- pyproject.toml | 1 + src/pipecat/audio/turn/local_smart_turn.py | 39 +++++++++++++++++++++- 3 files changed, 43 insertions(+), 2 deletions(-) diff --git a/dot-env.template b/dot-env.template index f0b5bdc0f..ae0df90e5 100644 --- a/dot-env.template +++ b/dot-env.template @@ -92,4 +92,7 @@ ASSEMBLYAI_API_KEY=... OPENROUTER_API_KEY=... # Piper -PIPER_BASE_URL=... \ No newline at end of file +PIPER_BASE_URL=... + +# Local Smart turn +LOCAL_SMART_TURN_MODEL_PATH= \ No newline at end of file diff --git a/pyproject.toml b/pyproject.toml index 8b7c8546a..10b44508f 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -79,6 +79,7 @@ qwen = [] rime = [ "websockets~=13.1" ] riva = [ "nvidia-riva-client~=2.19.0" ] sentry = [ "sentry-sdk~=2.23.1" ] +local-smart-turn = [ "coremltools>=8.0", "transformers" ] silero = [ "onnxruntime~=1.20.1" ] simli = [ "simli-ai~=0.1.10"] soundfile = [ "soundfile~=0.13.0" ] diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index c511ae2ba..497487e06 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -5,11 +5,23 @@ # +import os + import numpy as np from loguru import logger from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer, EndOfTurnState +try: + import coremltools as ct + from transformers import AutoFeatureExtractor +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use the LocalSmartTurnAnalyzer, you need to `pip install pipecat-ai[local-smart-turn]`." + ) + raise Exception(f"Missing module: {e}") + class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): def __init__(self): @@ -18,7 +30,32 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): logger.debug("Loading Local Smart Turn model...") - # TODO: implement it + # To use this locally, set the environment variable LOCAL_SMART_TURN_MODEL_PATH + # to the path where the smart-turn repo is cloned. + # + # Example setup: + # + # # Git LFS (Large File Storage) + # brew install git-lfs + # # Hugging Face uses LFS to store large model files, including .mlpackage + # git lfs install + # # Clone the repo with the smart_turn_classifier.mlpackage + # git clone https://huggingface.co/pipecat-ai/smart-turn + # + # Then set the env variable: + # export LOCAL_SMART_TURN_MODEL_PATH=./smart-turn + # or add it to your .env file + smart_turn_model_path = os.getenv("LOCAL_SMART_TURN_MODEL_PATH") + + if not smart_turn_model_path: + logger.error("LOCAL_SMART_TURN_MODEL_PATH is not set.") + raise Exception("LOCAL_SMART_TURN_MODEL_PATH environment variable must be provided.") + + core_ml_model_path = f"{smart_turn_model_path}/coreml/smart_turn_classifier.mlpackage" + + # Only load the processor, not the torch model + processor = AutoFeatureExtractor.from_pretrained(smart_turn_model_path) + model = ct.models.MLModel(core_ml_model_path) logger.debug("Loaded Local Smart Turn") From 3588b067187cc5f85f010f6c3c01879dd7c64fda Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 15 Apr 2025 12:28:36 -0300 Subject: [PATCH 04/34] Adding missing torch dependency. --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 10b44508f..2cb292761 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -79,7 +79,7 @@ qwen = [] rime = [ "websockets~=13.1" ] riva = [ "nvidia-riva-client~=2.19.0" ] sentry = [ "sentry-sdk~=2.23.1" ] -local-smart-turn = [ "coremltools>=8.0", "transformers" ] +local-smart-turn = [ "coremltools>=8.0", "transformers", "torch==2.5.0", "torchaudio==2.5.0" ] silero = [ "onnxruntime~=1.20.1" ] simli = [ "simli-ai~=0.1.10"] soundfile = [ "soundfile~=0.13.0" ] From e6325a8229d2fe2ade92790af5f2864f25e76ff7 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 15 Apr 2025 16:01:09 -0300 Subject: [PATCH 05/34] Integrating with the smart turn model to predict --- src/pipecat/audio/turn/base_turn_analyzer.py | 10 +- src/pipecat/audio/turn/local_smart_turn.py | 163 ++++++++++++++++--- src/pipecat/transports/base_input.py | 27 +-- 3 files changed, 164 insertions(+), 36 deletions(-) diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py index 1dffa563b..eeb9bfeb8 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/src/pipecat/audio/turn/base_turn_analyzer.py @@ -19,6 +19,7 @@ class BaseEndOfTurnAnalyzer(ABC): def __init__(self, *, sample_rate: Optional[int] = None): self._init_sample_rate = sample_rate self._sample_rate = 0 + self._chunk_size_ms = 0 @property def sample_rate(self) -> int: @@ -27,6 +28,13 @@ class BaseEndOfTurnAnalyzer(ABC): def set_sample_rate(self, sample_rate: int): self._sample_rate = self._init_sample_rate or sample_rate + @property + def chunk_size_ms(self) -> int: + return self._chunk_size_ms + + def set_chunk_size_ms(self, chunk_size_ms: int): + self._chunk_size_ms = chunk_size_ms + @abstractmethod - def analyze_audio(self, buffer: bytes) -> EndOfTurnState: + def analyze_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: pass diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index 497487e06..d7733adc3 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -6,8 +6,10 @@ import os +import time import numpy as np +import torch from loguru import logger from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer, EndOfTurnState @@ -23,13 +25,15 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") +# TODO: we should convert all this to params +STOP_MS = 1000 +PRE_SPEECH_MS = 200 +MAX_DURATION_SECONDS = 16 # Maximum duration for the smart turn model + + class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): def __init__(self): super().__init__() - self._audio_buffer = bytearray() - - logger.debug("Loading Local Smart Turn model...") - # To use this locally, set the environment variable LOCAL_SMART_TURN_MODEL_PATH # to the path where the smart-turn repo is cloned. # @@ -53,36 +57,145 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): core_ml_model_path = f"{smart_turn_model_path}/coreml/smart_turn_classifier.mlpackage" + logger.debug("Loading Local Smart Turn model...") # Only load the processor, not the torch model - processor = AutoFeatureExtractor.from_pretrained(smart_turn_model_path) - model = ct.models.MLModel(core_ml_model_path) - + self._turn_processor = AutoFeatureExtractor.from_pretrained(smart_turn_model_path) + self._turn_model = ct.models.MLModel(core_ml_model_path) logger.debug("Loaded Local Smart Turn") - def analyze_audio(self, buffer: bytes) -> EndOfTurnState: - self._audio_buffer += buffer + self._audio_buffer = [] + self._speech_triggered = False + self._silence_frames = 0 + self._speech_start_time = None - # TODO: we probably don't need this - # Checking if we have at least 6 seconds of audio - # if len(self._audio_buffer) < 16000 * 2 * 6: - # return EndOfTurnState.INCOMPLETE + def analyze_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: + state = EndOfTurnState.INCOMPLETE - audio_int16 = np.frombuffer(self._audio_buffer, dtype=np.int16) + audio_int16 = np.frombuffer(buffer, dtype=np.int16) # Divide by 32768 because we have signed 16-bit data. audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 - # TODO: implement to use the smart turn - # for now it is always returning as complete only for testing it - prediction = 1 - - state = EndOfTurnState.COMPLETE if prediction == 1 else EndOfTurnState.INCOMPLETE - - if state == EndOfTurnState.COMPLETE: - # clears the buffer completely - self._audio_buffer = bytearray() + if is_speech: + if not self._speech_triggered: + self._silence_frames = 0 + self._speech_triggered = True + if self._speech_start_time is None: + self._speech_start_time = time.time() + self._audio_buffer.append((time.time(), audio_float32)) else: - # TODO: implement it - pass + if self._speech_triggered: + self._audio_buffer.append((time.time(), audio_float32)) + self._silence_frames += 1 + if self._silence_frames * self._chunk_size_ms >= STOP_MS: + self._speech_triggered = False + + # TODO: do we need to stop or do something to prevent ?? + + state = self._process_speech_segment( + self._audio_buffer, self._speech_start_time + ) + self._audio_buffer = [] + self._speech_start_time = None + + # TODO: same here for restart + else: + # Keep buffering some silence before potential speech starts + self._audio_buffer.append((time.time(), audio_float32)) + # Keep the buffer size reasonable, assuming CHUNK is small + max_buffer_time = ( + PRE_SPEECH_MS + STOP_MS + ) / 1000 + MAX_DURATION_SECONDS # Some extra buffer + while ( + self._audio_buffer and self._audio_buffer[0][0] < time.time() - max_buffer_time + ): + self._audio_buffer.pop(0) return state + + def _process_speech_segment(self, audio_buffer, speech_start_time) -> EndOfTurnState: + state = EndOfTurnState.INCOMPLETE + + if not audio_buffer: + return state + + # Find start and end indices for the segment + start_time = speech_start_time - (PRE_SPEECH_MS / 1000) + start_index = 0 + for i, (t, _) in enumerate(audio_buffer): + if t >= start_time: + start_index = i + break + + end_index = len(audio_buffer) - 1 + + # Extract the audio segment + segment_audio_chunks = [chunk for _, chunk in audio_buffer[start_index : end_index + 1]] + segment_audio = np.concatenate(segment_audio_chunks) + + # Remove (STOP_MS - 200)ms from the end of the segment + samples_to_remove = int((STOP_MS - 200) / 1000 * self.sample_rate) + segment_audio = segment_audio[:-samples_to_remove] + + # Limit maximum duration + if len(segment_audio) / self.sample_rate > MAX_DURATION_SECONDS: + segment_audio = segment_audio[: int(MAX_DURATION_SECONDS * self.sample_rate)] + + # No resampling needed as both recording and prediction use 16000 Hz + segment_audio_resampled = segment_audio + + if len(segment_audio_resampled) > 0: + # Call the new predict_endpoint function with the audio data + start_time = time.perf_counter() + + result = self._predict_endpoint(segment_audio_resampled) + + state = ( + EndOfTurnState.COMPLETE if result["prediction"] == 1 else EndOfTurnState.INCOMPLETE + ) + + end_time = time.perf_counter() + + logger.debug("--------") + logger.debug(f"Prediction: {'Complete' if result['prediction'] == 1 else 'Incomplete'}") + logger.debug(f"Probability of complete: {result['probability']:.4f}") + logger.debug(f"Prediction took {(end_time - start_time) * 1000:.2f}ms seconds") + else: + logger.debug("Captured empty audio segment, skipping prediction.") + + return state + + def _predict_endpoint(self, audio_array): + """ + 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, + padding="max_length", + truncation=True, + max_length=800, # Maximum length as specified in training + return_attention_mask=True, + return_tensors="pt", + ) + + output = self._turn_model.predict(dict(inputs)) + logits = output["logits"] # Core ML returns numpy array + logits_tensor = torch.tensor(logits) + probabilities = torch.nn.functional.softmax(logits_tensor, dim=1) + completion_prob = probabilities[0, 1].item() # Probability of class 1 (Complete) + prediction = 1 if completion_prob > 0.5 else 0 + + return { + "prediction": prediction, + "probability": completion_prob, + } diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 946feb063..4221b2f65 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -79,6 +79,9 @@ class BaseInputTransport(FrameProcessor): # Configure End of turn analyzer. if self._params.end_of_turn_analyzer: self._params.end_of_turn_analyzer.set_sample_rate(self._sample_rate) + self._params.end_of_turn_analyzer.set_chunk_size_ms( + self._params.audio_out_10ms_chunks * 10 + ) # Start audio filter. if self._params.audio_in_filter: await self._params.audio_in_filter.start(self._sample_rate) @@ -214,18 +217,23 @@ class BaseInputTransport(FrameProcessor): vad_state = new_vad_state return vad_state - async def _end_of_turn_analyze(self, audio_frame: InputAudioRawFrame) -> EndOfTurnState: + async def _end_of_turn_analyze( + self, audio_frame: InputAudioRawFrame, is_speech: bool + ) -> EndOfTurnState: state = EndOfTurnState.INCOMPLETE if self.end_of_turn_analyzer: state = await self.get_event_loop().run_in_executor( - self._executor, self.end_of_turn_analyzer.analyze_audio, audio_frame.audio + self._executor, + self.end_of_turn_analyzer.analyze_audio, + audio_frame.audio, + is_speech, ) return state async def _handle_end_of_turn( - self, audio_frame: InputAudioRawFrame, end_of_turn_state: EndOfTurnState + self, audio_frame: InputAudioRawFrame, end_of_turn_state: EndOfTurnState, is_speech: bool ): - new_eot_state = await self._end_of_turn_analyze(audio_frame) + new_eot_state = await self._end_of_turn_analyze(audio_frame, is_speech) if new_eot_state != end_of_turn_state: await self._handle_user_interruption(UserEndOfTurnFrame()) return new_eot_state @@ -246,14 +254,13 @@ class BaseInputTransport(FrameProcessor): # changes from QUIET to SPEAKING and vice versa. if self._params.vad_enabled: vad_state = await self._handle_vad(frame, vad_state) - # TODO: need to check if we need to keep it later - if vad_state == VADState.QUIET: - end_of_turn_state = EndOfTurnState.INCOMPLETE audio_passthrough = self._params.vad_audio_passthrough - # We only need to check for completion if the user is speaking - if self._params.end_of_turn_analyzer and VADState.QUIET != vad_state: - end_of_turn_state = await self._handle_end_of_turn(frame, end_of_turn_state) + if self._params.end_of_turn_analyzer: + is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING + end_of_turn_state = await self._handle_end_of_turn( + frame, end_of_turn_state, is_speech + ) # Push audio downstream if passthrough. if audio_passthrough: From 11b6de0900f552533a00d800a05c3d66602af9d2 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 15 Apr 2025 17:28:00 -0300 Subject: [PATCH 06/34] Triggering to check if the turn is complete each time the user stops speaking based on the vad --- examples/foundational/38a-local-smart-turn.py | 3 +- src/pipecat/audio/turn/base_turn_analyzer.py | 6 ++- src/pipecat/audio/turn/local_smart_turn.py | 37 +++++++++---------- src/pipecat/transports/base_input.py | 30 +++++---------- 4 files changed, 35 insertions(+), 41 deletions(-) diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py index a9bf8e293..490a26684 100644 --- a/examples/foundational/38a-local-smart-turn.py +++ b/examples/foundational/38a-local-smart-turn.py @@ -11,6 +11,7 @@ from loguru import logger from pipecat.audio.turn.local_smart_turn import LocalSmartTurnAnalyzer 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 @@ -34,7 +35,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): audio_in_enabled=True, audio_out_enabled=True, vad_enabled=True, - vad_analyzer=SileroVADAnalyzer(), + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), vad_audio_passthrough=True, end_of_turn_analyzer=LocalSmartTurnAnalyzer(), ), diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py index eeb9bfeb8..1e6a964f9 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/src/pipecat/audio/turn/base_turn_analyzer.py @@ -36,5 +36,9 @@ class BaseEndOfTurnAnalyzer(ABC): self._chunk_size_ms = chunk_size_ms @abstractmethod - def analyze_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: + def append_audio(self, buffer: bytes, is_speech: bool): + pass + + @abstractmethod + def analyze_end_of_turn(self) -> EndOfTurnState: pass diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index d7733adc3..b831fffeb 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -28,7 +28,7 @@ except ModuleNotFoundError as e: # TODO: we should convert all this to params STOP_MS = 1000 PRE_SPEECH_MS = 200 -MAX_DURATION_SECONDS = 16 # Maximum duration for the smart turn model +MAX_DURATION_SECONDS = 8 # Maximum duration for the smart turn model class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): @@ -68,11 +68,8 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): self._silence_frames = 0 self._speech_start_time = None - def analyze_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: - state = EndOfTurnState.INCOMPLETE - + def append_audio(self, buffer: bytes, is_speech: bool): audio_int16 = np.frombuffer(buffer, dtype=np.int16) - # Divide by 32768 because we have signed 16-bit data. audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 @@ -87,18 +84,6 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): if self._speech_triggered: self._audio_buffer.append((time.time(), audio_float32)) self._silence_frames += 1 - if self._silence_frames * self._chunk_size_ms >= STOP_MS: - self._speech_triggered = False - - # TODO: do we need to stop or do something to prevent ?? - - state = self._process_speech_segment( - self._audio_buffer, self._speech_start_time - ) - self._audio_buffer = [] - self._speech_start_time = None - - # TODO: same here for restart else: # Keep buffering some silence before potential speech starts self._audio_buffer.append((time.time(), audio_float32)) @@ -111,16 +96,30 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): ): self._audio_buffer.pop(0) + def analyze_end_of_turn(self) -> EndOfTurnState: + logger.debug("Analyzing End of Turn...") + if self._silence_frames * self._chunk_size_ms >= STOP_MS: + logger.debug("End of Turn complete due to STOP_MS.") + state = EndOfTurnState.COMPLETE + else: + state = self._process_speech_segment(self._audio_buffer) + + if state == EndOfTurnState.COMPLETE: + self._speech_triggered = False + self._audio_buffer = [] + self._speech_start_time = None + + logger.debug(f"End of Turn result: {state}") return state - def _process_speech_segment(self, audio_buffer, speech_start_time) -> EndOfTurnState: + def _process_speech_segment(self, audio_buffer) -> EndOfTurnState: state = EndOfTurnState.INCOMPLETE if not audio_buffer: return state # Find start and end indices for the segment - start_time = speech_start_time - (PRE_SPEECH_MS / 1000) + start_time = self._speech_start_time - (PRE_SPEECH_MS / 1000) start_index = 0 for i, (t, _) in enumerate(audio_buffer): if t >= start_time: diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 4221b2f65..296d0a453 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -217,27 +217,16 @@ class BaseInputTransport(FrameProcessor): vad_state = new_vad_state return vad_state - async def _end_of_turn_analyze( - self, audio_frame: InputAudioRawFrame, is_speech: bool - ) -> EndOfTurnState: - state = EndOfTurnState.INCOMPLETE + async def _handle_end_of_turn(self, end_of_turn_state: EndOfTurnState): + state = end_of_turn_state if self.end_of_turn_analyzer: - state = await self.get_event_loop().run_in_executor( - self._executor, - self.end_of_turn_analyzer.analyze_audio, - audio_frame.audio, - is_speech, + new_state = await self.get_event_loop().run_in_executor( + self._executor, self.end_of_turn_analyzer.analyze_end_of_turn ) + if new_state != state and new_state == EndOfTurnState.COMPLETE: + await self._handle_user_interruption(UserEndOfTurnFrame()) return state - async def _handle_end_of_turn( - self, audio_frame: InputAudioRawFrame, end_of_turn_state: EndOfTurnState, is_speech: bool - ): - new_eot_state = await self._end_of_turn_analyze(audio_frame, is_speech) - if new_eot_state != end_of_turn_state: - await self._handle_user_interruption(UserEndOfTurnFrame()) - return new_eot_state - async def _audio_task_handler(self): vad_state: VADState = VADState.QUIET end_of_turn_state: EndOfTurnState = EndOfTurnState.INCOMPLETE @@ -252,15 +241,16 @@ class BaseInputTransport(FrameProcessor): # Check VAD and push event if necessary. We just care about # changes from QUIET to SPEAKING and vice versa. + previous_vad_state = vad_state if self._params.vad_enabled: vad_state = await self._handle_vad(frame, vad_state) audio_passthrough = self._params.vad_audio_passthrough if self._params.end_of_turn_analyzer: is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING - end_of_turn_state = await self._handle_end_of_turn( - frame, end_of_turn_state, is_speech - ) + self._params.end_of_turn_analyzer.append_audio(frame.audio, is_speech) + if vad_state == VADState.QUIET and vad_state != previous_vad_state: + end_of_turn_state = await self._handle_end_of_turn(end_of_turn_state) # Push audio downstream if passthrough. if audio_passthrough: From 57b39c084fddd893799c7a536fc1527ba03fecb7 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 15 Apr 2025 20:42:41 -0300 Subject: [PATCH 07/34] Triggering to check if the turn is complete based on the maximum timeout --- src/pipecat/audio/turn/base_turn_analyzer.py | 2 +- src/pipecat/audio/turn/local_smart_turn.py | 25 ++++++++------ src/pipecat/transports/base_input.py | 36 ++++++++++---------- 3 files changed, 34 insertions(+), 29 deletions(-) diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py index 1e6a964f9..76777bc65 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/src/pipecat/audio/turn/base_turn_analyzer.py @@ -36,7 +36,7 @@ class BaseEndOfTurnAnalyzer(ABC): self._chunk_size_ms = chunk_size_ms @abstractmethod - def append_audio(self, buffer: bytes, is_speech: bool): + def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: pass @abstractmethod diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index b831fffeb..57f7584e0 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -68,11 +68,12 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): self._silence_frames = 0 self._speech_start_time = None - def append_audio(self, buffer: bytes, is_speech: bool): + def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: audio_int16 = np.frombuffer(buffer, dtype=np.int16) # Divide by 32768 because we have signed 16-bit data. audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 + state = EndOfTurnState.INCOMPLETE if is_speech: if not self._speech_triggered: self._silence_frames = 0 @@ -84,6 +85,10 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): if self._speech_triggered: self._audio_buffer.append((time.time(), audio_float32)) self._silence_frames += 1 + if self._silence_frames * self._chunk_size_ms >= STOP_MS: + logger.debug("End of Turn complete due to STOP_MS.") + state = EndOfTurnState.COMPLETE + self._clear() else: # Keep buffering some silence before potential speech starts self._audio_buffer.append((time.time(), audio_float32)) @@ -96,22 +101,22 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): ): self._audio_buffer.pop(0) + return state + def analyze_end_of_turn(self) -> EndOfTurnState: logger.debug("Analyzing End of Turn...") - if self._silence_frames * self._chunk_size_ms >= STOP_MS: - logger.debug("End of Turn complete due to STOP_MS.") - state = EndOfTurnState.COMPLETE - else: - state = self._process_speech_segment(self._audio_buffer) - + state = self._process_speech_segment(self._audio_buffer) if state == EndOfTurnState.COMPLETE: - self._speech_triggered = False - self._audio_buffer = [] - self._speech_start_time = None + self._clear() logger.debug(f"End of Turn result: {state}") return state + def _clear(self): + self._speech_triggered = False + self._audio_buffer = [] + self._speech_start_time = None + def _process_speech_segment(self, audio_buffer) -> EndOfTurnState: state = EndOfTurnState.INCOMPLETE diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 296d0a453..7755e9670 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -175,16 +175,9 @@ class BaseInputTransport(FrameProcessor): elif isinstance(frame, UserStoppedSpeakingFrame): logger.debug("User stopped speaking") await self.push_frame(frame) - - # TODO check, we probably should change here as well. - # if the end of turn is enabled, we should only stop interruption after this point if self.interruptions_allowed: await self._stop_interruption() await self.push_frame(StopInterruptionFrame()) - elif isinstance(frame, UserEndOfTurnFrame): - logger.debug("User end of turn") - await self.push_frame(frame) - # TODO: implement to handle interruptions # # Audio input @@ -208,7 +201,9 @@ class BaseInputTransport(FrameProcessor): frame = None if new_vad_state == VADState.SPEAKING: frame = UserStartedSpeakingFrame() - elif new_vad_state == VADState.QUIET: + # TODO: need to double check if this is the expected behavior + # Not triggering the UserStoppedSpeakingFrame if the turn analyser is enabled + elif new_vad_state == VADState.QUIET and not self.end_of_turn_analyzer: frame = UserStoppedSpeakingFrame() if frame: @@ -217,19 +212,20 @@ class BaseInputTransport(FrameProcessor): vad_state = new_vad_state return vad_state - async def _handle_end_of_turn(self, end_of_turn_state: EndOfTurnState): - state = end_of_turn_state + async def _handle_end_of_turn(self): if self.end_of_turn_analyzer: - new_state = await self.get_event_loop().run_in_executor( + state = await self.get_event_loop().run_in_executor( self._executor, self.end_of_turn_analyzer.analyze_end_of_turn ) - if new_state != state and new_state == EndOfTurnState.COMPLETE: - await self._handle_user_interruption(UserEndOfTurnFrame()) - return state + await self._handle_end_of_turn_complete(state) + + async def _handle_end_of_turn_complete(self, state: EndOfTurnState): + if state == EndOfTurnState.COMPLETE: + await self.push_frame(UserEndOfTurnFrame()) + await self._handle_user_interruption(UserStoppedSpeakingFrame()) async def _audio_task_handler(self): vad_state: VADState = VADState.QUIET - end_of_turn_state: EndOfTurnState = EndOfTurnState.INCOMPLETE while True: frame: InputAudioRawFrame = await self._audio_in_queue.get() @@ -248,9 +244,13 @@ class BaseInputTransport(FrameProcessor): if self._params.end_of_turn_analyzer: is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING - self._params.end_of_turn_analyzer.append_audio(frame.audio, is_speech) - if vad_state == VADState.QUIET and vad_state != previous_vad_state: - end_of_turn_state = await self._handle_end_of_turn(end_of_turn_state) + end_of_turn_state = self._params.end_of_turn_analyzer.append_audio( + frame.audio, is_speech + ) + if end_of_turn_state == EndOfTurnState.COMPLETE: + await self._handle_end_of_turn_complete(end_of_turn_state) + elif vad_state == VADState.QUIET and vad_state != previous_vad_state: + await self._handle_end_of_turn() # Push audio downstream if passthrough. if audio_passthrough: From cfefcac35ff861286fbf11dc13948d1c50431324 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Tue, 15 Apr 2025 20:51:36 -0300 Subject: [PATCH 08/34] Resetting the silence frames when the user speaks. --- src/pipecat/audio/turn/local_smart_turn.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index 57f7584e0..ebc9250a9 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -75,8 +75,7 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): state = EndOfTurnState.INCOMPLETE if is_speech: - if not self._speech_triggered: - self._silence_frames = 0 + self._silence_frames = 0 self._speech_triggered = True if self._speech_start_time is None: self._speech_start_time = time.time() @@ -116,6 +115,7 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): self._speech_triggered = False self._audio_buffer = [] self._speech_start_time = None + self._silence_frames = 0 def _process_speech_segment(self, audio_buffer) -> EndOfTurnState: state = EndOfTurnState.INCOMPLETE From 3e2d21779fa1c1d191eace0319e32f28ab695b98 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 06:11:56 -0300 Subject: [PATCH 09/34] Refactoring the BaseEndOfTurnAnalyzer to include most of the logic --- src/pipecat/audio/turn/base_turn_analyzer.py | 121 ++++++++++++++++++- src/pipecat/audio/turn/local_smart_turn.py | 120 +----------------- 2 files changed, 122 insertions(+), 119 deletions(-) diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_turn_analyzer.py index 76777bc65..bb10c7df3 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/src/pipecat/audio/turn/base_turn_analyzer.py @@ -5,9 +5,13 @@ # +import time from abc import ABC, abstractmethod from enum import Enum -from typing import Optional +from typing import Dict, Optional + +import numpy as np +from loguru import logger class EndOfTurnState(Enum): @@ -15,11 +19,23 @@ class EndOfTurnState(Enum): INCOMPLETE = 2 +# TODO: we should convert all this to params +STOP_MS = 1000 +PRE_SPEECH_MS = 200 +MAX_DURATION_SECONDS = 8 # Maximum duration for the smart turn model + + class BaseEndOfTurnAnalyzer(ABC): def __init__(self, *, sample_rate: Optional[int] = None): self._init_sample_rate = sample_rate + # settings variables self._sample_rate = 0 self._chunk_size_ms = 0 + # inference variables + self._audio_buffer = [] + self._speech_triggered = False + self._silence_frames = 0 + self._speech_start_time = None @property def sample_rate(self) -> int: @@ -35,10 +51,107 @@ class BaseEndOfTurnAnalyzer(ABC): def set_chunk_size_ms(self, chunk_size_ms: int): self._chunk_size_ms = chunk_size_ms - @abstractmethod def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: - pass + audio_int16 = np.frombuffer(buffer, dtype=np.int16) + # Divide by 32768 because we have signed 16-bit data. + audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 + + state = EndOfTurnState.INCOMPLETE + if is_speech: + self._silence_frames = 0 + self._speech_triggered = True + if self._speech_start_time is None: + self._speech_start_time = time.time() + self._audio_buffer.append((time.time(), audio_float32)) + else: + if self._speech_triggered: + self._audio_buffer.append((time.time(), audio_float32)) + self._silence_frames += 1 + if self._silence_frames * self._chunk_size_ms >= STOP_MS: + logger.debug("End of Turn complete due to STOP_MS.") + state = EndOfTurnState.COMPLETE + self._clear() + else: + # Keep buffering some silence before potential speech starts + self._audio_buffer.append((time.time(), audio_float32)) + # Keep the buffer size reasonable, assuming CHUNK is small + max_buffer_time = ( + PRE_SPEECH_MS + STOP_MS + ) / 1000 + MAX_DURATION_SECONDS # Some extra buffer + while ( + self._audio_buffer and self._audio_buffer[0][0] < time.time() - max_buffer_time + ): + self._audio_buffer.pop(0) + + return state + + def analyze_end_of_turn(self) -> EndOfTurnState: + logger.debug("Analyzing End of Turn...") + state = self._process_speech_segment(self._audio_buffer) + if state == EndOfTurnState.COMPLETE: + self._clear() + + logger.debug(f"End of Turn result: {state}") + return state + + def _clear(self): + self._speech_triggered = False + self._audio_buffer = [] + self._speech_start_time = None + self._silence_frames = 0 + + def _process_speech_segment(self, audio_buffer) -> EndOfTurnState: + state = EndOfTurnState.INCOMPLETE + + if not audio_buffer: + return state + + # Find start and end indices for the segment + start_time = self._speech_start_time - (PRE_SPEECH_MS / 1000) + start_index = 0 + for i, (t, _) in enumerate(audio_buffer): + if t >= start_time: + start_index = i + break + + end_index = len(audio_buffer) - 1 + + # Extract the audio segment + segment_audio_chunks = [chunk for _, chunk in audio_buffer[start_index : end_index + 1]] + segment_audio = np.concatenate(segment_audio_chunks) + + # Remove (STOP_MS - 200)ms from the end of the segment + samples_to_remove = int((STOP_MS - 200) / 1000 * self.sample_rate) + segment_audio = segment_audio[:-samples_to_remove] + + # Limit maximum duration + if len(segment_audio) / self.sample_rate > MAX_DURATION_SECONDS: + segment_audio = segment_audio[: int(MAX_DURATION_SECONDS * self.sample_rate)] + + # No resampling needed as both recording and prediction use 16000 Hz + segment_audio_resampled = segment_audio + + if len(segment_audio_resampled) > 0: + # Call the new predict_endpoint function with the audio data + start_time = time.perf_counter() + + result = self._predict_endpoint(segment_audio_resampled) + + state = ( + EndOfTurnState.COMPLETE if result["prediction"] == 1 else EndOfTurnState.INCOMPLETE + ) + + end_time = time.perf_counter() + + logger.debug("--------") + logger.debug(f"Prediction: {'Complete' if result['prediction'] == 1 else 'Incomplete'}") + logger.debug(f"Probability of complete: {result['probability']:.4f}") + logger.debug(f"Prediction took {(end_time - start_time) * 1000:.2f}ms seconds") + else: + logger.debug("Captured empty audio segment, skipping prediction.") + + return state @abstractmethod - def analyze_end_of_turn(self) -> EndOfTurnState: + def _predict_endpoint(self, buffer: np.ndarray) -> Dict[str, any]: pass diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index ebc9250a9..efa6c81d1 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -6,13 +6,15 @@ import os -import time +from typing import Dict import numpy as np import torch from loguru import logger -from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer, EndOfTurnState +from pipecat.audio.turn.base_turn_analyzer import ( + BaseEndOfTurnAnalyzer, +) try: import coremltools as ct @@ -25,12 +27,6 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -# TODO: we should convert all this to params -STOP_MS = 1000 -PRE_SPEECH_MS = 200 -MAX_DURATION_SECONDS = 8 # Maximum duration for the smart turn model - - class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): def __init__(self): super().__init__() @@ -63,113 +59,7 @@ class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): self._turn_model = ct.models.MLModel(core_ml_model_path) logger.debug("Loaded Local Smart Turn") - self._audio_buffer = [] - self._speech_triggered = False - self._silence_frames = 0 - self._speech_start_time = None - - def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: - audio_int16 = np.frombuffer(buffer, dtype=np.int16) - # Divide by 32768 because we have signed 16-bit data. - audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 - - state = EndOfTurnState.INCOMPLETE - if is_speech: - self._silence_frames = 0 - self._speech_triggered = True - if self._speech_start_time is None: - self._speech_start_time = time.time() - self._audio_buffer.append((time.time(), audio_float32)) - else: - if self._speech_triggered: - self._audio_buffer.append((time.time(), audio_float32)) - self._silence_frames += 1 - if self._silence_frames * self._chunk_size_ms >= STOP_MS: - logger.debug("End of Turn complete due to STOP_MS.") - state = EndOfTurnState.COMPLETE - self._clear() - else: - # Keep buffering some silence before potential speech starts - self._audio_buffer.append((time.time(), audio_float32)) - # Keep the buffer size reasonable, assuming CHUNK is small - max_buffer_time = ( - PRE_SPEECH_MS + STOP_MS - ) / 1000 + MAX_DURATION_SECONDS # Some extra buffer - while ( - self._audio_buffer and self._audio_buffer[0][0] < time.time() - max_buffer_time - ): - self._audio_buffer.pop(0) - - return state - - def analyze_end_of_turn(self) -> EndOfTurnState: - logger.debug("Analyzing End of Turn...") - state = self._process_speech_segment(self._audio_buffer) - if state == EndOfTurnState.COMPLETE: - self._clear() - - logger.debug(f"End of Turn result: {state}") - return state - - def _clear(self): - self._speech_triggered = False - self._audio_buffer = [] - self._speech_start_time = None - self._silence_frames = 0 - - def _process_speech_segment(self, audio_buffer) -> EndOfTurnState: - state = EndOfTurnState.INCOMPLETE - - if not audio_buffer: - return state - - # Find start and end indices for the segment - start_time = self._speech_start_time - (PRE_SPEECH_MS / 1000) - start_index = 0 - for i, (t, _) in enumerate(audio_buffer): - if t >= start_time: - start_index = i - break - - end_index = len(audio_buffer) - 1 - - # Extract the audio segment - segment_audio_chunks = [chunk for _, chunk in audio_buffer[start_index : end_index + 1]] - segment_audio = np.concatenate(segment_audio_chunks) - - # Remove (STOP_MS - 200)ms from the end of the segment - samples_to_remove = int((STOP_MS - 200) / 1000 * self.sample_rate) - segment_audio = segment_audio[:-samples_to_remove] - - # Limit maximum duration - if len(segment_audio) / self.sample_rate > MAX_DURATION_SECONDS: - segment_audio = segment_audio[: int(MAX_DURATION_SECONDS * self.sample_rate)] - - # No resampling needed as both recording and prediction use 16000 Hz - segment_audio_resampled = segment_audio - - if len(segment_audio_resampled) > 0: - # Call the new predict_endpoint function with the audio data - start_time = time.perf_counter() - - result = self._predict_endpoint(segment_audio_resampled) - - state = ( - EndOfTurnState.COMPLETE if result["prediction"] == 1 else EndOfTurnState.INCOMPLETE - ) - - end_time = time.perf_counter() - - logger.debug("--------") - logger.debug(f"Prediction: {'Complete' if result['prediction'] == 1 else 'Incomplete'}") - logger.debug(f"Probability of complete: {result['probability']:.4f}") - logger.debug(f"Prediction took {(end_time - start_time) * 1000:.2f}ms seconds") - else: - logger.debug("Captured empty audio segment, skipping prediction.") - - return state - - def _predict_endpoint(self, audio_array): + def _predict_endpoint(self, audio_array: np.ndarray) -> Dict[str, any]: """ Predict whether an audio segment is complete (turn ended) or incomplete. From 3ebef9346fce6ffeae0e3a386a31edb6e5141fce Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 06:33:42 -0300 Subject: [PATCH 10/34] 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) From 0e4115049b137a040bc56e6c49436e06c0427592 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 06:44:57 -0300 Subject: [PATCH 11/34] Refactoring to use keep alive sessions. --- src/pipecat/audio/turn/remote_smart_turn.py | 24 +++++++++++---------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/src/pipecat/audio/turn/remote_smart_turn.py b/src/pipecat/audio/turn/remote_smart_turn.py index a971dac37..f54eb2372 100644 --- a/src/pipecat/audio/turn/remote_smart_turn.py +++ b/src/pipecat/audio/turn/remote_smart_turn.py @@ -13,9 +13,7 @@ import numpy as np import requests from loguru import logger -from pipecat.audio.turn.base_turn_analyzer import ( - BaseEndOfTurnAnalyzer, -) +from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer class RemoteSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): @@ -27,30 +25,34 @@ class RemoteSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): logger.error("REMOTE_SMART_TURN_URL is not set.") raise Exception("REMOTE_SMART_TURN_URL environment variable 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: - """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 + np.save(buffer, audio_array) 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 + response = self.session.post( + self.remote_smart_turn_url, + data=data_bytes, + headers=headers, + timeout=60, + ) 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:") @@ -62,7 +64,7 @@ class RemoteSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): else: logger.debug("Response Content (Error):") logger.debug(response.text) - response.raise_for_status() # Raise an exception for bad status codes + response.raise_for_status() except requests.exceptions.RequestException as e: logger.debug(f"Failed to send raw request to Daily Smart Turn: {e}") From 2627cb6bf2d5ce52f1494013aabcb7688ca07202 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 07:13:13 -0300 Subject: [PATCH 12/34] Allowing to define SmartTurnParams --- ...se_turn_analyzer.py => base_smart_turn.py} | 34 ++++++++++++------- src/pipecat/audio/turn/local_smart_turn.py | 10 +++--- src/pipecat/audio/turn/remote_smart_turn.py | 8 ++--- src/pipecat/transports/base_input.py | 4 +-- src/pipecat/transports/base_transport.py | 4 +-- 5 files changed, 33 insertions(+), 27 deletions(-) rename src/pipecat/audio/turn/{base_turn_analyzer.py => base_smart_turn.py} (82%) diff --git a/src/pipecat/audio/turn/base_turn_analyzer.py b/src/pipecat/audio/turn/base_smart_turn.py similarity index 82% rename from src/pipecat/audio/turn/base_turn_analyzer.py rename to src/pipecat/audio/turn/base_smart_turn.py index 91b38bb00..e83336ec1 100644 --- a/src/pipecat/audio/turn/base_turn_analyzer.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -12,6 +12,7 @@ from typing import Dict, Optional import numpy as np from loguru import logger +from pydantic import BaseModel class EndOfTurnState(Enum): @@ -19,18 +20,25 @@ class EndOfTurnState(Enum): INCOMPLETE = 2 -# TODO: we should convert all this to params -STOP_MS = 1000 +STOP_SECS = 1 PRE_SPEECH_MS = 200 MAX_DURATION_SECONDS = 8 # Maximum duration for the smart turn model -class BaseEndOfTurnAnalyzer(ABC): - def __init__(self, *, sample_rate: Optional[int] = None): +class SmartTurnParams(BaseModel): + stop_secs: float = STOP_SECS + pre_speech_ms: float = PRE_SPEECH_MS + max_duration_secs: float = MAX_DURATION_SECONDS + + +class BaseSmartTurn(ABC): + def __init__(self, *, sample_rate: Optional[int] = None, params: SmartTurnParams = SmartTurnParams()): self._init_sample_rate = sample_rate + self._params = params # settings variables self._sample_rate = 0 self._chunk_size_ms = 0 + self._stop_ms = self._params.stop_secs * 1000 # inference variables self._audio_buffer = [] self._speech_triggered = False @@ -67,8 +75,8 @@ class BaseEndOfTurnAnalyzer(ABC): if self._speech_triggered: self._audio_buffer.append((time.time(), audio_float32)) self._silence_frames += 1 - if self._silence_frames * self._chunk_size_ms >= STOP_MS: - logger.debug("End of Turn complete due to STOP_MS.") + if self._silence_frames * self._chunk_size_ms >= self._stop_ms: + logger.debug("End of Turn complete due to stop_secs.") state = EndOfTurnState.COMPLETE self._clear() else: @@ -76,8 +84,8 @@ class BaseEndOfTurnAnalyzer(ABC): self._audio_buffer.append((time.time(), audio_float32)) # Keep the buffer size reasonable, assuming CHUNK is small max_buffer_time = ( - PRE_SPEECH_MS + STOP_MS - ) / 1000 + MAX_DURATION_SECONDS # Some extra buffer + self._params.pre_speech_ms + self._stop_ms + ) / 1000 + self._params.max_duration_secs # Some extra buffer while ( self._audio_buffer and self._audio_buffer[0][0] < time.time() - max_buffer_time ): @@ -107,7 +115,7 @@ class BaseEndOfTurnAnalyzer(ABC): return state # Find start and end indices for the segment - start_time = self._speech_start_time - (PRE_SPEECH_MS / 1000) + start_time = self._speech_start_time - (self._params.pre_speech_ms / 1000) start_index = 0 for i, (t, _) in enumerate(audio_buffer): if t >= start_time: @@ -120,13 +128,13 @@ class BaseEndOfTurnAnalyzer(ABC): segment_audio_chunks = [chunk for _, chunk in audio_buffer[start_index : end_index + 1]] segment_audio = np.concatenate(segment_audio_chunks) - # Remove (STOP_MS - 200)ms from the end of the segment - samples_to_remove = int((STOP_MS - 200) / 1000 * self.sample_rate) + # Remove (self._stop_ms - 200)ms from the end of the segment + samples_to_remove = int((self._stop_ms - 200) / 1000 * self.sample_rate) segment_audio = segment_audio[:-samples_to_remove] # Limit maximum duration - if len(segment_audio) / self.sample_rate > MAX_DURATION_SECONDS: - segment_audio = segment_audio[: int(MAX_DURATION_SECONDS * self.sample_rate)] + if len(segment_audio) / self.sample_rate > self._params.max_duration_secs: + segment_audio = segment_audio[: int(self._params.max_duration_secs * self.sample_rate)] # No resampling needed as both recording and prediction use 16000 Hz segment_audio_resampled = segment_audio diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index a84bab1e0..5143e5f3b 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -12,9 +12,7 @@ import numpy as np import torch from loguru import logger -from pipecat.audio.turn.base_turn_analyzer import ( - BaseEndOfTurnAnalyzer, -) +from pipecat.audio.turn.base_smart_turn import BaseSmartTurn try: import coremltools as ct @@ -27,9 +25,9 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class LocalSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): - def __init__(self): - super().__init__() +class LocalSmartTurnAnalyzer(BaseSmartTurn): + def __init__(self, **kwargs): + super().__init__(**kwargs) # To use this locally, set the environment variable LOCAL_SMART_TURN_MODEL_PATH # to the path where the smart-turn repo is cloned. # diff --git a/src/pipecat/audio/turn/remote_smart_turn.py b/src/pipecat/audio/turn/remote_smart_turn.py index f54eb2372..e24609607 100644 --- a/src/pipecat/audio/turn/remote_smart_turn.py +++ b/src/pipecat/audio/turn/remote_smart_turn.py @@ -13,12 +13,12 @@ import numpy as np import requests from loguru import logger -from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer +from pipecat.audio.turn.base_smart_turn import BaseSmartTurn -class RemoteSmartTurnAnalyzer(BaseEndOfTurnAnalyzer): - def __init__(self): - super().__init__() +class RemoteSmartTurnAnalyzer(BaseSmartTurn): + def __init__(self, **kwargs): + super().__init__(**kwargs) self.remote_smart_turn_url = os.getenv("REMOTE_SMART_TURN_URL") if not self.remote_smart_turn_url: diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 7755e9670..d0b2c8c2c 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -10,7 +10,7 @@ from typing import Optional from loguru import logger -from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer, EndOfTurnState +from pipecat.audio.turn.base_smart_turn import BaseSmartTurn, EndOfTurnState from pipecat.audio.vad.vad_analyzer import VADAnalyzer, VADState from pipecat.frames.frames import ( BotInterruptionFrame, @@ -67,7 +67,7 @@ class BaseInputTransport(FrameProcessor): return self._params.vad_analyzer @property - def end_of_turn_analyzer(self) -> Optional[BaseEndOfTurnAnalyzer]: + def end_of_turn_analyzer(self) -> Optional[BaseSmartTurn]: return self._params.end_of_turn_analyzer async def start(self, frame: StartFrame): diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 6045609f7..79c876fc4 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -11,7 +11,7 @@ from pydantic import BaseModel, ConfigDict from pipecat.audio.filters.base_audio_filter import BaseAudioFilter from pipecat.audio.mixers.base_audio_mixer import BaseAudioMixer -from pipecat.audio.turn.base_turn_analyzer import BaseEndOfTurnAnalyzer +from pipecat.audio.turn.base_smart_turn import BaseSmartTurn from pipecat.audio.vad.vad_analyzer import VADAnalyzer from pipecat.processors.frame_processor import FrameProcessor from pipecat.utils.base_object import BaseObject @@ -42,7 +42,7 @@ class TransportParams(BaseModel): vad_enabled: bool = False vad_audio_passthrough: bool = False vad_analyzer: Optional[VADAnalyzer] = None - end_of_turn_analyzer: Optional[BaseEndOfTurnAnalyzer] = None + end_of_turn_analyzer: Optional[BaseSmartTurn] = None class BaseTransport(BaseObject): From 650d4d9ee2e03dc8ba76fd855e52a851e7c59c18 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 07:55:20 -0300 Subject: [PATCH 13/34] Changing the start speech time and adding logs. --- src/pipecat/audio/turn/base_smart_turn.py | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index e83336ec1..a4fb64651 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -21,7 +21,7 @@ class EndOfTurnState(Enum): STOP_SECS = 1 -PRE_SPEECH_MS = 200 +PRE_SPEECH_MS = 0 MAX_DURATION_SECONDS = 8 # Maximum duration for the smart turn model @@ -32,7 +32,9 @@ class SmartTurnParams(BaseModel): class BaseSmartTurn(ABC): - def __init__(self, *, sample_rate: Optional[int] = None, params: SmartTurnParams = SmartTurnParams()): + def __init__( + self, *, sample_rate: Optional[int] = None, params: SmartTurnParams = SmartTurnParams() + ): self._init_sample_rate = sample_rate self._params = params # settings variables @@ -63,6 +65,7 @@ class BaseSmartTurn(ABC): audio_int16 = np.frombuffer(buffer, dtype=np.int16) # Divide by 32768 because we have signed 16-bit data. audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 + self._audio_buffer.append((time.time(), audio_float32)) state = EndOfTurnState.INCOMPLETE if is_speech: @@ -70,18 +73,15 @@ class BaseSmartTurn(ABC): self._speech_triggered = True if self._speech_start_time is None: self._speech_start_time = time.time() - self._audio_buffer.append((time.time(), audio_float32)) + logger.debug(f"Speech started at {self._speech_start_time}") else: if self._speech_triggered: - self._audio_buffer.append((time.time(), audio_float32)) self._silence_frames += 1 if self._silence_frames * self._chunk_size_ms >= self._stop_ms: logger.debug("End of Turn complete due to stop_secs.") state = EndOfTurnState.COMPLETE self._clear() else: - # Keep buffering some silence before potential speech starts - self._audio_buffer.append((time.time(), audio_float32)) # Keep the buffer size reasonable, assuming CHUNK is small max_buffer_time = ( self._params.pre_speech_ms + self._stop_ms @@ -103,6 +103,7 @@ class BaseSmartTurn(ABC): return state def _clear(self): + logger.debug("Clearing audio buffer...") self._speech_triggered = False self._audio_buffer = [] self._speech_start_time = None @@ -156,6 +157,7 @@ class BaseSmartTurn(ABC): logger.debug(f"Probability of complete: {result['probability']:.4f}") logger.debug(f"Prediction took {(end_time - start_time) * 1000:.2f}ms seconds") else: + logger.debug(f"params: {self._params}, stop_ms: {self._stop_ms}") logger.debug("Captured empty audio segment, skipping prediction.") return state From 616961b4879109efcf5f17422032b654bd813450 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 08:04:38 -0300 Subject: [PATCH 14/34] Stop removing segments from the end --- src/pipecat/audio/turn/base_smart_turn.py | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index a4fb64651..bcd028cab 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -129,22 +129,20 @@ class BaseSmartTurn(ABC): segment_audio_chunks = [chunk for _, chunk in audio_buffer[start_index : end_index + 1]] segment_audio = np.concatenate(segment_audio_chunks) - # Remove (self._stop_ms - 200)ms from the end of the segment - samples_to_remove = int((self._stop_ms - 200) / 1000 * self.sample_rate) - segment_audio = segment_audio[:-samples_to_remove] + logger.debug(f"Segment audio chunks after start index: {len(segment_audio)}") # Limit maximum duration if len(segment_audio) / self.sample_rate > self._params.max_duration_secs: segment_audio = segment_audio[: int(self._params.max_duration_secs * self.sample_rate)] - # No resampling needed as both recording and prediction use 16000 Hz - segment_audio_resampled = segment_audio + logger.debug(f"Segment audio chunks after limiting duration: {len(segment_audio)}") - if len(segment_audio_resampled) > 0: + # No resampling needed as both recording and prediction use 16000 Hz + if len(segment_audio) > 0: # Call the new predict_endpoint function with the audio data start_time = time.perf_counter() - result = self._predict_endpoint(segment_audio_resampled) + result = self._predict_endpoint(segment_audio) state = ( EndOfTurnState.COMPLETE if result["prediction"] == 1 else EndOfTurnState.INCOMPLETE From 5fa47b7a5c938eb8b0bb76ba57d70ef9add945f6 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 08:45:01 -0300 Subject: [PATCH 15/34] Adding the dependencies for the remote smart turn --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 2cb292761..fdabdb199 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -80,6 +80,7 @@ rime = [ "websockets~=13.1" ] riva = [ "nvidia-riva-client~=2.19.0" ] sentry = [ "sentry-sdk~=2.23.1" ] local-smart-turn = [ "coremltools>=8.0", "transformers", "torch==2.5.0", "torchaudio==2.5.0" ] +remote-smart-turn = [] silero = [ "onnxruntime~=1.20.1" ] simli = [ "simli-ai~=0.1.10"] soundfile = [ "soundfile~=0.13.0" ] From cd8bd7f487649b24f252cbfb6ee33f9127c19ade Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 08:58:40 -0300 Subject: [PATCH 16/34] Adding some comments to the code. --- src/pipecat/audio/turn/base_smart_turn.py | 45 ++++++++++++----------- 1 file changed, 23 insertions(+), 22 deletions(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index bcd028cab..d0f566393 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -4,7 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # - import time from abc import ABC, abstractmethod from enum import Enum @@ -15,14 +14,16 @@ from loguru import logger from pydantic import BaseModel +# Enum for end-of-turn detection states class EndOfTurnState(Enum): COMPLETE = 1 INCOMPLETE = 2 +# Default timing parameters STOP_SECS = 1 PRE_SPEECH_MS = 0 -MAX_DURATION_SECONDS = 8 # Maximum duration for the smart turn model +MAX_DURATION_SECONDS = 8 # Max allowed segment duration class SmartTurnParams(BaseModel): @@ -37,11 +38,11 @@ class BaseSmartTurn(ABC): ): self._init_sample_rate = sample_rate self._params = params - # settings variables + # Configuration self._sample_rate = 0 self._chunk_size_ms = 0 - self._stop_ms = self._params.stop_secs * 1000 - # inference variables + self._stop_ms = self._params.stop_secs * 1000 # silence threshold in ms + # Inference state self._audio_buffer = [] self._speech_triggered = False self._silence_frames = 0 @@ -52,7 +53,7 @@ class BaseSmartTurn(ABC): return self._sample_rate def set_sample_rate(self, sample_rate: int): - self._sample_rate = self._init_sample_rate or sample_rate + self._sample_rate = sample_rate @property def chunk_size_ms(self) -> int: @@ -62,13 +63,15 @@ class BaseSmartTurn(ABC): self._chunk_size_ms = chunk_size_ms def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: + # Convert raw audio to float32 format and append to the buffer audio_int16 = np.frombuffer(buffer, dtype=np.int16) - # Divide by 32768 because we have signed 16-bit data. audio_float32 = np.frombuffer(audio_int16, dtype=np.int16).astype(np.float32) / 32768.0 self._audio_buffer.append((time.time(), audio_float32)) state = EndOfTurnState.INCOMPLETE + if is_speech: + # Reset silence tracking on speech self._silence_frames = 0 self._speech_triggered = True if self._speech_start_time is None: @@ -77,15 +80,18 @@ class BaseSmartTurn(ABC): else: if self._speech_triggered: self._silence_frames += 1 + # If silence exceeds threshold, mark end of turn if self._silence_frames * self._chunk_size_ms >= self._stop_ms: logger.debug("End of Turn complete due to stop_secs.") state = EndOfTurnState.COMPLETE self._clear() else: - # Keep the buffer size reasonable, assuming CHUNK is small + # Trim buffer to prevent unbounded growth before speech max_buffer_time = ( - self._params.pre_speech_ms + self._stop_ms - ) / 1000 + self._params.max_duration_secs # Some extra buffer + (self._params.pre_speech_ms / 1000) + + self._params.stop_secs + + self._params.max_duration_secs + ) while ( self._audio_buffer and self._audio_buffer[0][0] < time.time() - max_buffer_time ): @@ -98,11 +104,11 @@ class BaseSmartTurn(ABC): state = self._process_speech_segment(self._audio_buffer) if state == EndOfTurnState.COMPLETE: self._clear() - logger.debug(f"End of Turn result: {state}") return state def _clear(self): + # Reset internal state for next turn logger.debug("Clearing audio buffer...") self._speech_triggered = False self._audio_buffer = [] @@ -115,7 +121,7 @@ class BaseSmartTurn(ABC): if not audio_buffer: return state - # Find start and end indices for the segment + # Extract recent audio segment for prediction start_time = self._speech_start_time - (self._params.pre_speech_ms / 1000) start_index = 0 for i, (t, _) in enumerate(audio_buffer): @@ -137,17 +143,12 @@ class BaseSmartTurn(ABC): logger.debug(f"Segment audio chunks after limiting duration: {len(segment_audio)}") - # No resampling needed as both recording and prediction use 16000 Hz if len(segment_audio) > 0: - # Call the new predict_endpoint function with the audio data 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() logger.debug("--------") @@ -163,14 +164,14 @@ class BaseSmartTurn(ABC): @abstractmethod def _predict_endpoint(self, buffer: np.ndarray) -> Dict[str, any]: """ - Predict whether an audio segment is complete (turn ended) or incomplete. + Abstract method to predict if a turn has ended based on audio. Args: - audio_array: Numpy array containing audio samples at 16kHz + buffer: Float32 numpy array of audio samples at 16kHz. Returns: - Dictionary containing prediction results: - - prediction: 1 for complete, 0 for incomplete - - probability: Probability of completion class + Dictionary with: + - prediction: 1 if turn is complete, else 0 + - probability: Confidence of the prediction """ pass From 8e36bdbed76633eae5210010b0c834d9c6eebcc7 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 16 Apr 2025 09:11:27 -0300 Subject: [PATCH 17/34] Adding some comments to the code. --- src/pipecat/transports/base_input.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index d0b2c8c2c..fadc3629e 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -244,11 +244,13 @@ class BaseInputTransport(FrameProcessor): if self._params.end_of_turn_analyzer: is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING + # If silence exceeds threshold, we are going to receive EndOfTurnState.COMPLETE end_of_turn_state = self._params.end_of_turn_analyzer.append_audio( frame.audio, is_speech ) if end_of_turn_state == EndOfTurnState.COMPLETE: await self._handle_end_of_turn_complete(end_of_turn_state) + # Otherwise we are going to trigger to check if the turn is completed based on the VAD elif vad_state == VADState.QUIET and vad_state != previous_vad_state: await self._handle_end_of_turn() From 3f0688aefa83c469ce975b9eae4451612d917e07 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 09:36:03 -0300 Subject: [PATCH 18/34] Testing smart turn using stop_secs as 5 seconds --- examples/foundational/38a-local-smart-turn.py | 3 ++- src/pipecat/audio/turn/base_smart_turn.py | 4 +++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py index 490a26684..2be547cd0 100644 --- a/examples/foundational/38a-local-smart-turn.py +++ b/examples/foundational/38a-local-smart-turn.py @@ -9,6 +9,7 @@ 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 LocalSmartTurnAnalyzer from pipecat.audio.vad.silero import SileroVADAnalyzer from pipecat.audio.vad.vad_analyzer import VADParams @@ -37,7 +38,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), vad_audio_passthrough=True, - end_of_turn_analyzer=LocalSmartTurnAnalyzer(), + end_of_turn_analyzer=LocalSmartTurnAnalyzer(params=SmartTurnParams(stop_secs=5)), ), ) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index d0f566393..6bf5cf655 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -82,7 +82,9 @@ class BaseSmartTurn(ABC): self._silence_frames += 1 # If silence exceeds threshold, mark end of turn if self._silence_frames * self._chunk_size_ms >= self._stop_ms: - logger.debug("End of Turn complete due to stop_secs.") + logger.debug( + f"End of Turn complete due to stop_secs. Silence: {self._silence_frames}, chunk_size_ms: {self._chunk_size_ms}" + ) state = EndOfTurnState.COMPLETE self._clear() else: From 266537c3f47103897ba2ffe44f2ec55d2b8387e2 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 10:07:08 -0300 Subject: [PATCH 19/34] Fixing to respect the stop_secs. --- src/pipecat/audio/turn/base_smart_turn.py | 21 +++++++-------------- src/pipecat/transports/base_input.py | 3 --- 2 files changed, 7 insertions(+), 17 deletions(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index 6bf5cf655..9fc439fb2 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -40,12 +40,11 @@ class BaseSmartTurn(ABC): self._params = params # Configuration self._sample_rate = 0 - self._chunk_size_ms = 0 self._stop_ms = self._params.stop_secs * 1000 # silence threshold in ms # Inference state self._audio_buffer = [] self._speech_triggered = False - self._silence_frames = 0 + self._silence_ms = 0 self._speech_start_time = None @property @@ -55,13 +54,6 @@ class BaseSmartTurn(ABC): def set_sample_rate(self, sample_rate: int): self._sample_rate = sample_rate - @property - def chunk_size_ms(self) -> int: - return self._chunk_size_ms - - def set_chunk_size_ms(self, chunk_size_ms: int): - self._chunk_size_ms = chunk_size_ms - def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: # Convert raw audio to float32 format and append to the buffer audio_int16 = np.frombuffer(buffer, dtype=np.int16) @@ -72,18 +64,19 @@ class BaseSmartTurn(ABC): if is_speech: # Reset silence tracking on speech - self._silence_frames = 0 + self._silence_ms = 0 self._speech_triggered = True if self._speech_start_time is None: self._speech_start_time = time.time() logger.debug(f"Speech started at {self._speech_start_time}") else: if self._speech_triggered: - self._silence_frames += 1 + chunk_duration_ms = len(audio_int16) / (self._sample_rate / 1000) + self._silence_ms += chunk_duration_ms # If silence exceeds threshold, mark end of turn - if self._silence_frames * self._chunk_size_ms >= self._stop_ms: + if self._silence_ms >= self._stop_ms: logger.debug( - f"End of Turn complete due to stop_secs. Silence: {self._silence_frames}, chunk_size_ms: {self._chunk_size_ms}" + f"End of Turn complete due to stop_secs. Silence in ms: {self._silence_ms}" ) state = EndOfTurnState.COMPLETE self._clear() @@ -115,7 +108,7 @@ class BaseSmartTurn(ABC): self._speech_triggered = False self._audio_buffer = [] self._speech_start_time = None - self._silence_frames = 0 + self._silence_ms = 0 def _process_speech_segment(self, audio_buffer) -> EndOfTurnState: state = EndOfTurnState.INCOMPLETE diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index fadc3629e..7ae6d3e64 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -79,9 +79,6 @@ class BaseInputTransport(FrameProcessor): # Configure End of turn analyzer. if self._params.end_of_turn_analyzer: self._params.end_of_turn_analyzer.set_sample_rate(self._sample_rate) - self._params.end_of_turn_analyzer.set_chunk_size_ms( - self._params.audio_out_10ms_chunks * 10 - ) # Start audio filter. if self._params.audio_in_filter: await self._params.audio_in_filter.start(self._sample_rate) From 88ce117e845b556a62afd83d44107fa5cb25e76d Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 10:35:13 -0300 Subject: [PATCH 20/34] Changing the max duration default value to 16 seconds. --- src/pipecat/audio/turn/base_smart_turn.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index 9fc439fb2..2892db7c0 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -23,7 +23,7 @@ class EndOfTurnState(Enum): # Default timing parameters STOP_SECS = 1 PRE_SPEECH_MS = 0 -MAX_DURATION_SECONDS = 8 # Max allowed segment duration +MAX_DURATION_SECONDS = 16 # Max allowed segment duration class SmartTurnParams(BaseModel): From e2fbbb4b4084583483ab75b24fe9bb46d1ae60ac Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 10:43:21 -0300 Subject: [PATCH 21/34] Renaming the smart turn classes. --- examples/foundational/38-smart-turn.py | 4 ++-- examples/foundational/38a-local-smart-turn.py | 4 ++-- src/pipecat/audio/turn/local_smart_turn.py | 2 +- .../audio/turn/{remote_smart_turn.py => smart_turn.py} | 2 +- 4 files changed, 6 insertions(+), 6 deletions(-) rename src/pipecat/audio/turn/{remote_smart_turn.py => smart_turn.py} (98%) diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn.py index 251a13445..fb910b980 100644 --- a/examples/foundational/38-smart-turn.py +++ b/examples/foundational/38-smart-turn.py @@ -9,7 +9,7 @@ import os from dotenv import load_dotenv from loguru import logger -from pipecat.audio.turn.remote_smart_turn import RemoteSmartTurnAnalyzer +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 @@ -37,7 +37,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), vad_audio_passthrough=True, - end_of_turn_analyzer=RemoteSmartTurnAnalyzer(), + end_of_turn_analyzer=SmartTurnAnalyzer(), ), ) diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py index 2be547cd0..8cf3beb48 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 LocalSmartTurnAnalyzer +from pipecat.audio.turn.local_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 @@ -38,7 +38,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), vad_audio_passthrough=True, - end_of_turn_analyzer=LocalSmartTurnAnalyzer(params=SmartTurnParams(stop_secs=5)), + end_of_turn_analyzer=LocalCoreMLSmartTurnAnalyzer(params=SmartTurnParams(stop_secs=5)), ), ) diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index 5143e5f3b..669cabf61 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -25,7 +25,7 @@ except ModuleNotFoundError as e: raise Exception(f"Missing module: {e}") -class LocalSmartTurnAnalyzer(BaseSmartTurn): +class LocalCoreMLSmartTurnAnalyzer(BaseSmartTurn): def __init__(self, **kwargs): super().__init__(**kwargs) # To use this locally, set the environment variable LOCAL_SMART_TURN_MODEL_PATH diff --git a/src/pipecat/audio/turn/remote_smart_turn.py b/src/pipecat/audio/turn/smart_turn.py similarity index 98% rename from src/pipecat/audio/turn/remote_smart_turn.py rename to src/pipecat/audio/turn/smart_turn.py index e24609607..e36478e96 100644 --- a/src/pipecat/audio/turn/remote_smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn.py @@ -16,7 +16,7 @@ from loguru import logger from pipecat.audio.turn.base_smart_turn import BaseSmartTurn -class RemoteSmartTurnAnalyzer(BaseSmartTurn): +class SmartTurnAnalyzer(BaseSmartTurn): def __init__(self, **kwargs): super().__init__(**kwargs) self.remote_smart_turn_url = os.getenv("REMOTE_SMART_TURN_URL") From 8577139d210c563e07a5f2853b61cab21b0dfe21 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 11:39:06 -0300 Subject: [PATCH 22/34] Fixing to keep the last max samples. --- src/pipecat/audio/turn/base_smart_turn.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index 2892db7c0..6f04f3e83 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -133,8 +133,10 @@ class BaseSmartTurn(ABC): logger.debug(f"Segment audio chunks after start index: {len(segment_audio)}") # Limit maximum duration - if len(segment_audio) / self.sample_rate > self._params.max_duration_secs: - segment_audio = segment_audio[: int(self._params.max_duration_secs * self.sample_rate)] + max_samples = int(self._params.max_duration_secs * self.sample_rate) + if len(segment_audio) > max_samples: + # slices the array to keep the last max_samples samples, discarding the earlier part. + segment_audio = segment_audio[-max_samples:] logger.debug(f"Segment audio chunks after limiting duration: {len(segment_audio)}") From b0b38beb19b17b2aab2f99067a950c8254fbe472 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 11:39:48 -0300 Subject: [PATCH 23/34] Returning the max duration back to 8 seconds. --- src/pipecat/audio/turn/base_smart_turn.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index 6f04f3e83..26fdfd83d 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -23,7 +23,7 @@ class EndOfTurnState(Enum): # Default timing parameters STOP_SECS = 1 PRE_SPEECH_MS = 0 -MAX_DURATION_SECONDS = 16 # Max allowed segment duration +MAX_DURATION_SECONDS = 8 # Max allowed segment duration class SmartTurnParams(BaseModel): From e179916c9c2a9587ebf7bedbc02c900af2dad9cf Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 11:49:51 -0300 Subject: [PATCH 24/34] Creating a new param use_only_last_vad_segment --- src/pipecat/audio/turn/base_smart_turn.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index 26fdfd83d..a4c1c530f 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -24,12 +24,14 @@ class EndOfTurnState(Enum): STOP_SECS = 1 PRE_SPEECH_MS = 0 MAX_DURATION_SECONDS = 8 # Max allowed segment duration +USE_ONLY_LAST_VAD_SEGMENT = False class SmartTurnParams(BaseModel): stop_secs: float = STOP_SECS pre_speech_ms: float = PRE_SPEECH_MS max_duration_secs: float = MAX_DURATION_SECONDS + use_only_last_vad_segment: bool = USE_ONLY_LAST_VAD_SEGMENT class BaseSmartTurn(ABC): @@ -97,7 +99,7 @@ class BaseSmartTurn(ABC): def analyze_end_of_turn(self) -> EndOfTurnState: logger.debug("Analyzing End of Turn...") state = self._process_speech_segment(self._audio_buffer) - if state == EndOfTurnState.COMPLETE: + if state == EndOfTurnState.COMPLETE or self._params.use_only_last_vad_segment: self._clear() logger.debug(f"End of Turn result: {state}") return state From 3599761e4e69a7e613f627ed1246bd1a52b5cd4c Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 16:07:03 -0300 Subject: [PATCH 25/34] Changing the default behavior to only use the last vad segment, and increasing the default stop_secs to 3 --- src/pipecat/audio/turn/base_smart_turn.py | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index a4c1c530f..c08fd8fb3 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -21,17 +21,18 @@ class EndOfTurnState(Enum): # Default timing parameters -STOP_SECS = 1 +STOP_SECS = 3 PRE_SPEECH_MS = 0 MAX_DURATION_SECONDS = 8 # Max allowed segment duration -USE_ONLY_LAST_VAD_SEGMENT = False +USE_ONLY_LAST_VAD_SEGMENT = True class SmartTurnParams(BaseModel): stop_secs: float = STOP_SECS pre_speech_ms: float = PRE_SPEECH_MS max_duration_secs: float = MAX_DURATION_SECONDS - use_only_last_vad_segment: bool = USE_ONLY_LAST_VAD_SEGMENT + # not exposing this for now yet until the model can handle it. + # use_only_last_vad_segment: bool = USE_ONLY_LAST_VAD_SEGMENT class BaseSmartTurn(ABC): @@ -99,7 +100,7 @@ class BaseSmartTurn(ABC): def analyze_end_of_turn(self) -> EndOfTurnState: logger.debug("Analyzing End of Turn...") state = self._process_speech_segment(self._audio_buffer) - if state == EndOfTurnState.COMPLETE or self._params.use_only_last_vad_segment: + if state == EndOfTurnState.COMPLETE or USE_ONLY_LAST_VAD_SEGMENT: self._clear() logger.debug(f"End of Turn result: {state}") return state From 53ee3fb64cef1751970dc49f68afde2b88c18421 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 16:14:13 -0300 Subject: [PATCH 26/34] Changing the log levels used in smart_turn --- src/pipecat/audio/turn/smart_turn.py | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/src/pipecat/audio/turn/smart_turn.py b/src/pipecat/audio/turn/smart_turn.py index e36478e96..c6e2f2af1 100644 --- a/src/pipecat/audio/turn/smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn.py @@ -30,16 +30,16 @@ class SmartTurnAnalyzer(BaseSmartTurn): self.session.headers.update({"Connection": "keep-alive"}) def _serialize_array(self, audio_array: np.ndarray) -> bytes: - logger.debug("Serializing NumPy array to bytes...") + logger.trace("Serializing NumPy array to bytes...") buffer = io.BytesIO() np.save(buffer, audio_array) serialized_bytes = buffer.getvalue() - logger.debug(f"Serialized size: {len(serialized_bytes)} bytes") + 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.debug( + logger.trace( f"Sending {len(data_bytes)} bytes as raw body to {self.remote_smart_turn_url}..." ) try: @@ -50,24 +50,24 @@ class SmartTurnAnalyzer(BaseSmartTurn): timeout=60, ) - logger.debug("\n--- Response ---") - logger.debug(f"Status Code: {response.status_code}") + logger.trace("\n--- Response ---") + logger.trace(f"Status Code: {response.status_code}") if response.ok: try: - logger.debug("Response JSON:") - logger.debug(response.json()) + logger.trace("Response JSON:") + logger.trace(response.json()) return response.json() except requests.exceptions.JSONDecodeError: - logger.debug("Response Content (non-JSON):") - logger.debug(response.text) + logger.trace("Response Content (non-JSON):") + logger.trace(response.text) else: - logger.debug("Response Content (Error):") - logger.debug(response.text) + logger.trace("Response Content (Error):") + logger.trace(response.text) response.raise_for_status() except requests.exceptions.RequestException as e: - logger.debug(f"Failed to send raw request to Daily Smart Turn: {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]: From a80f82cdb6a2e476a1771e3aa27d67ba6a3be85a Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 16:28:50 -0300 Subject: [PATCH 27/34] Moving the environment variables to inside the demo. --- examples/foundational/38-smart-turn.py | 4 +++- examples/foundational/38a-local-smart-turn.py | 21 +++++++++++++++++- src/pipecat/audio/turn/local_smart_turn.py | 22 +++---------------- src/pipecat/audio/turn/smart_turn.py | 8 +++---- 4 files changed, 30 insertions(+), 25 deletions(-) diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn.py index fb910b980..f342f93c6 100644 --- a/examples/foundational/38-smart-turn.py +++ b/examples/foundational/38-smart-turn.py @@ -29,6 +29,8 @@ 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( @@ -37,7 +39,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), vad_audio_passthrough=True, - end_of_turn_analyzer=SmartTurnAnalyzer(), + end_of_turn_analyzer=SmartTurnAnalyzer(url=remote_smart_turn_url), ), ) diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py index 8cf3beb48..dffcec573 100644 --- a/examples/foundational/38a-local-smart-turn.py +++ b/examples/foundational/38a-local-smart-turn.py @@ -30,6 +30,23 @@ load_dotenv(override=True) async def run_bot(webrtc_connection: SmallWebRTCConnection): logger.info(f"Starting bot") + # To use this locally, set the environment variable LOCAL_SMART_TURN_MODEL_PATH + # to the path where the smart-turn repo is cloned. + # + # Example setup: + # + # # Git LFS (Large File Storage) + # brew install git-lfs + # # Hugging Face uses LFS to store large model files, including .mlpackage + # git lfs install + # # Clone the repo with the smart_turn_classifier.mlpackage + # git clone https://huggingface.co/pipecat-ai/smart-turn + # + # Then set the env variable: + # export LOCAL_SMART_TURN_MODEL_PATH=./smart-turn + # or add it to your .env file + smart_turn_model_path = os.getenv("LOCAL_SMART_TURN_MODEL_PATH") + transport = SmallWebRTCTransport( webrtc_connection=webrtc_connection, params=TransportParams( @@ -38,7 +55,9 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): vad_enabled=True, vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), vad_audio_passthrough=True, - end_of_turn_analyzer=LocalCoreMLSmartTurnAnalyzer(params=SmartTurnParams(stop_secs=5)), + end_of_turn_analyzer=LocalCoreMLSmartTurnAnalyzer( + smart_turn_model_path=smart_turn_model_path, params=SmartTurnParams(stop_secs=5) + ), ), ) diff --git a/src/pipecat/audio/turn/local_smart_turn.py b/src/pipecat/audio/turn/local_smart_turn.py index 669cabf61..665e4b64d 100644 --- a/src/pipecat/audio/turn/local_smart_turn.py +++ b/src/pipecat/audio/turn/local_smart_turn.py @@ -26,28 +26,12 @@ except ModuleNotFoundError as e: class LocalCoreMLSmartTurnAnalyzer(BaseSmartTurn): - def __init__(self, **kwargs): + def __init__(self, smart_turn_model_path: str, **kwargs): super().__init__(**kwargs) - # To use this locally, set the environment variable LOCAL_SMART_TURN_MODEL_PATH - # to the path where the smart-turn repo is cloned. - # - # Example setup: - # - # # Git LFS (Large File Storage) - # brew install git-lfs - # # Hugging Face uses LFS to store large model files, including .mlpackage - # git lfs install - # # Clone the repo with the smart_turn_classifier.mlpackage - # git clone https://huggingface.co/pipecat-ai/smart-turn - # - # Then set the env variable: - # export LOCAL_SMART_TURN_MODEL_PATH=./smart-turn - # or add it to your .env file - smart_turn_model_path = os.getenv("LOCAL_SMART_TURN_MODEL_PATH") if not smart_turn_model_path: - logger.error("LOCAL_SMART_TURN_MODEL_PATH is not set.") - raise Exception("LOCAL_SMART_TURN_MODEL_PATH environment variable must be provided.") + logger.error("smart_turn_model_path is not set.") + raise Exception("smart_turn_model_path must be provided.") core_ml_model_path = f"{smart_turn_model_path}/coreml/smart_turn_classifier.mlpackage" diff --git a/src/pipecat/audio/turn/smart_turn.py b/src/pipecat/audio/turn/smart_turn.py index c6e2f2af1..5378e4b5b 100644 --- a/src/pipecat/audio/turn/smart_turn.py +++ b/src/pipecat/audio/turn/smart_turn.py @@ -17,13 +17,13 @@ from pipecat.audio.turn.base_smart_turn import BaseSmartTurn class SmartTurnAnalyzer(BaseSmartTurn): - def __init__(self, **kwargs): + def __init__(self, url: str, **kwargs): super().__init__(**kwargs) - self.remote_smart_turn_url = os.getenv("REMOTE_SMART_TURN_URL") + 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 environment variable must be provided.") + 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() From 3ea9cfd2518cd0abfab3375f2afc741ff242c32a Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 16:46:15 -0300 Subject: [PATCH 28/34] Keeping the _speech_triggered as true if the state is incomplete. --- src/pipecat/audio/turn/base_smart_turn.py | 9 +++++---- src/pipecat/transports/base_input.py | 23 +++++++++++++---------- 2 files changed, 18 insertions(+), 14 deletions(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index c08fd8fb3..cc18b2880 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -82,7 +82,7 @@ class BaseSmartTurn(ABC): f"End of Turn complete due to stop_secs. Silence in ms: {self._silence_ms}" ) state = EndOfTurnState.COMPLETE - self._clear() + self._clear(state) else: # Trim buffer to prevent unbounded growth before speech max_buffer_time = ( @@ -101,14 +101,15 @@ class BaseSmartTurn(ABC): logger.debug("Analyzing End of Turn...") state = self._process_speech_segment(self._audio_buffer) if state == EndOfTurnState.COMPLETE or USE_ONLY_LAST_VAD_SEGMENT: - self._clear() + self._clear(state) logger.debug(f"End of Turn result: {state}") return state - def _clear(self): + def _clear(self, turn_state: EndOfTurnState): # Reset internal state for next turn logger.debug("Clearing audio buffer...") - self._speech_triggered = False + # 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._silence_ms = 0 diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 7ae6d3e64..b43b71a49 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -221,6 +221,18 @@ class BaseInputTransport(FrameProcessor): await self.push_frame(UserEndOfTurnFrame()) await self._handle_user_interruption(UserStoppedSpeakingFrame()) + async def _run_turn_analyzer(self, frame: InputAudioRawFrame, vad_state: VADState, previous_vad_state: VADState): + is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING + # If silence exceeds threshold, we are going to receive EndOfTurnState.COMPLETE + end_of_turn_state = self._params.end_of_turn_analyzer.append_audio( + frame.audio, is_speech + ) + if end_of_turn_state == EndOfTurnState.COMPLETE: + await self._handle_end_of_turn_complete(end_of_turn_state) + # Otherwise we are going to trigger to check if the turn is completed based on the VAD + elif vad_state == VADState.QUIET and vad_state != previous_vad_state: + await self._handle_end_of_turn() + async def _audio_task_handler(self): vad_state: VADState = VADState.QUIET while True: @@ -240,16 +252,7 @@ class BaseInputTransport(FrameProcessor): audio_passthrough = self._params.vad_audio_passthrough if self._params.end_of_turn_analyzer: - is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING - # If silence exceeds threshold, we are going to receive EndOfTurnState.COMPLETE - end_of_turn_state = self._params.end_of_turn_analyzer.append_audio( - frame.audio, is_speech - ) - if end_of_turn_state == EndOfTurnState.COMPLETE: - await self._handle_end_of_turn_complete(end_of_turn_state) - # Otherwise we are going to trigger to check if the turn is completed based on the VAD - elif vad_state == VADState.QUIET and vad_state != previous_vad_state: - await self._handle_end_of_turn() + await self._run_turn_analyzer(frame, vad_state, previous_vad_state) # Push audio downstream if passthrough. if audio_passthrough: From a80dc94e9131b23e9125b82978f3e9a62a49e899 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 16:47:17 -0300 Subject: [PATCH 29/34] Fixing ruff format. --- src/pipecat/transports/base_input.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index b43b71a49..5ff7723ec 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -221,12 +221,12 @@ class BaseInputTransport(FrameProcessor): await self.push_frame(UserEndOfTurnFrame()) await self._handle_user_interruption(UserStoppedSpeakingFrame()) - async def _run_turn_analyzer(self, frame: InputAudioRawFrame, vad_state: VADState, previous_vad_state: VADState): + async def _run_turn_analyzer( + self, frame: InputAudioRawFrame, vad_state: VADState, previous_vad_state: VADState + ): is_speech = vad_state == VADState.SPEAKING or vad_state == VADState.STARTING # If silence exceeds threshold, we are going to receive EndOfTurnState.COMPLETE - end_of_turn_state = self._params.end_of_turn_analyzer.append_audio( - frame.audio, is_speech - ) + end_of_turn_state = self._params.end_of_turn_analyzer.append_audio(frame.audio, is_speech) if end_of_turn_state == EndOfTurnState.COMPLETE: await self._handle_end_of_turn_complete(end_of_turn_state) # Otherwise we are going to trigger to check if the turn is completed based on the VAD From 6e06bf97c017ffc16d991a93264afcb5b2b3659f Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 17:21:29 -0300 Subject: [PATCH 30/34] Preventing emitting the UserStartedSpeaking event multiple times. --- src/pipecat/audio/turn/base_smart_turn.py | 4 ++++ src/pipecat/transports/base_input.py | 18 ++++++++++++------ 2 files changed, 16 insertions(+), 6 deletions(-) diff --git a/src/pipecat/audio/turn/base_smart_turn.py b/src/pipecat/audio/turn/base_smart_turn.py index cc18b2880..0716d4e7c 100644 --- a/src/pipecat/audio/turn/base_smart_turn.py +++ b/src/pipecat/audio/turn/base_smart_turn.py @@ -57,6 +57,10 @@ class BaseSmartTurn(ABC): def set_sample_rate(self, sample_rate: int): self._sample_rate = sample_rate + @property + def speech_triggered(self) -> bool: + return self._speech_triggered + def append_audio(self, buffer: bytes, is_speech: bool) -> EndOfTurnState: # Convert raw audio to float32 format and append to the buffer audio_int16 = np.frombuffer(buffer, dtype=np.int16) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 5ff7723ec..767bc158c 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -196,12 +196,18 @@ class BaseInputTransport(FrameProcessor): and new_vad_state != VADState.STOPPING ): frame = None - if new_vad_state == VADState.SPEAKING: - frame = UserStartedSpeakingFrame() - # TODO: need to double check if this is the expected behavior - # Not triggering the UserStoppedSpeakingFrame if the turn analyser is enabled - elif new_vad_state == VADState.QUIET and not self.end_of_turn_analyzer: - frame = UserStoppedSpeakingFrame() + # If the turn analyser is enabled, this will prevent: + # - Creating the UserStoppedSpeakingFrame + # - Creating the UserStartedSpeakingFrame multiple times + can_create_user_frames = ( + self._params.end_of_turn_analyzer is None + or not self._params.end_of_turn_analyzer.speech_triggered + ) + if can_create_user_frames: + if new_vad_state == VADState.SPEAKING: + frame = UserStartedSpeakingFrame() + elif new_vad_state == VADState.QUIET: + frame = UserStoppedSpeakingFrame() if frame: await self._handle_user_interruption(frame) From c71005e249b55bc5d1a9779fa1460a6305932f4c Mon Sep 17 00:00:00 2001 From: Filipi da Silva Fuchter Date: Thu, 17 Apr 2025 17:43:23 -0300 Subject: [PATCH 31/34] Using the default model for OpenAi. Co-authored-by: Mark Backman --- examples/foundational/38-smart-turn.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/foundational/38-smart-turn.py b/examples/foundational/38-smart-turn.py index f342f93c6..03d530b90 100644 --- a/examples/foundational/38-smart-turn.py +++ b/examples/foundational/38-smart-turn.py @@ -50,7 +50,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { From e872ff943ac32ca666d7d1c37664d17c13c421d9 Mon Sep 17 00:00:00 2001 From: Filipi da Silva Fuchter Date: Thu, 17 Apr 2025 17:43:39 -0300 Subject: [PATCH 32/34] Using the default model for OpenAi. Co-authored-by: Mark Backman --- examples/foundational/38a-local-smart-turn.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py index dffcec573..4e687fb82 100644 --- a/examples/foundational/38a-local-smart-turn.py +++ b/examples/foundational/38a-local-smart-turn.py @@ -68,7 +68,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): voice_id="71a7ad14-091c-4e8e-a314-022ece01c121", # British Reading Lady ) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4o") + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) messages = [ { From 61d31d1c402164377077e04c177baafe9082e744 Mon Sep 17 00:00:00 2001 From: Filipi da Silva Fuchter Date: Thu, 17 Apr 2025 17:44:47 -0300 Subject: [PATCH 33/34] Restoring stop_secs to default value. Co-authored-by: Mark Backman --- examples/foundational/38a-local-smart-turn.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/examples/foundational/38a-local-smart-turn.py b/examples/foundational/38a-local-smart-turn.py index 4e687fb82..7baedf10e 100644 --- a/examples/foundational/38a-local-smart-turn.py +++ b/examples/foundational/38a-local-smart-turn.py @@ -56,7 +56,7 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection): vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.2)), vad_audio_passthrough=True, end_of_turn_analyzer=LocalCoreMLSmartTurnAnalyzer( - smart_turn_model_path=smart_turn_model_path, params=SmartTurnParams(stop_secs=5) + smart_turn_model_path=smart_turn_model_path, params=SmartTurnParams() ), ), ) From a390ce13a49d132d1f19502de02226979043bb05 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Thu, 17 Apr 2025 17:53:31 -0300 Subject: [PATCH 34/34] Removing the UserEndOfTurnFrame --- src/pipecat/frames/frames.py | 9 --------- src/pipecat/transports/base_input.py | 2 -- 2 files changed, 11 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 4301c5c1c..72acf1a2a 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -571,15 +571,6 @@ class EmulateUserStoppedSpeakingFrame(SystemFrame): pass -@dataclass -class UserEndOfTurnFrame(SystemFrame): - """Emitted based on the Smart Turn model to indicate that the user has - completed their turn/sentence. - """ - - pass - - @dataclass class BotInterruptionFrame(SystemFrame): """Emitted by when the bot should be interrupted. This will mainly cause the diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 767bc158c..3c7cf868d 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -25,7 +25,6 @@ from pipecat.frames.frames import ( StartInterruptionFrame, StopInterruptionFrame, SystemFrame, - UserEndOfTurnFrame, UserStartedSpeakingFrame, UserStoppedSpeakingFrame, VADParamsUpdateFrame, @@ -224,7 +223,6 @@ class BaseInputTransport(FrameProcessor): async def _handle_end_of_turn_complete(self, state: EndOfTurnState): if state == EndOfTurnState.COMPLETE: - await self.push_frame(UserEndOfTurnFrame()) await self._handle_user_interruption(UserStoppedSpeakingFrame()) async def _run_turn_analyzer(