Merge pull request #1409 from pipecat-ai/aleix/segmented-stt-service-vad-events

SegmentedSTTService: use VAD events to detect valid audio

CHANGELOG.md

@@ -149,6 +149,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed a `SegmentedSTTService` issue that was causing audio to be sent
+  prematurely to the STT service. Instead of analyzing the volume in this
+  service we rely on VAD events which use both VAD and volume.
+
 - Fixed a `GeminiMultimodalLiveLLMService` issue that was causing messages to be
   duplicated in the context when pushing `LLMMessagesAppendFrame` frames.
 

pyproject.toml

@@ -31,7 +31,7 @@ dependencies = [
     "pyloudnorm~=0.1.1",
     "resampy~=0.4.3",
     "soxr~=0.5.0",
-    "openai~=1.59.6"
+    "openai~=1.67.0"
 ]
 
 [project.urls]
@@ -39,42 +39,43 @@ Source = "https://github.com/pipecat-ai/pipecat"
 Website = "https://pipecat.ai"
 
 [project.optional-dependencies]
-anthropic = [ "anthropic~=0.47.2" ]
-assemblyai = [ "assemblyai~=0.36.0" ]
-aws = [ "boto3~=1.35.99" ]
-azure = [ "azure-cognitiveservices-speech~=1.42.0"]
+anthropic = [ "anthropic~=0.49.0" ]
+assemblyai = [ "assemblyai~=0.37.0" ]
+aws = [ "boto3~=1.37.16" ]
+azure = [ "azure-cognitiveservices-speech~=1.43.0"]
 canonical = [ "aiofiles~=24.1.0" ]
-cartesia = [ "cartesia~=1.3.1", "websockets~=13.1" ]
+cartesia = [ "cartesia~=1.4.0", "websockets~=13.1" ]
 neuphonic = [ "pyneuphonic~=1.5.13", "websockets~=13.1" ]
 cerebras = []
 deepseek = []
 daily = [ "daily-python~=0.15.0" ]
-deepgram = [ "deepgram-sdk~=3.8.0" ]
+deepgram = [ "deepgram-sdk~=3.10.1" ]
 elevenlabs = [ "websockets~=13.1" ]
-fal = [ "fal-client~=0.5.6" ]
+fal = [ "fal-client~=0.5.9" ]
 fish = [ "ormsgpack~=1.7.0", "websockets~=13.1" ]
 gladia = [ "websockets~=13.1" ]
-google = [ "google-cloud-speech~=2.31.0", "google-cloud-texttospeech~=2.25.0", "google-genai~=1.3.0", "google-generativeai~=0.8.4" ]
+google = [ "google-cloud-speech~=2.31.1", "google-cloud-texttospeech~=2.25.1", "google-genai~=1.7.0", "google-generativeai~=0.8.4" ]
 grok = []
 groq = []
 gstreamer = [ "pygobject~=3.50.0" ]
 fireworks = []
 krisp = [ "pipecat-ai-krisp~=0.3.0" ]
 koala = [ "pvkoala~=2.0.3" ]
-langchain = [ "langchain~=0.3.14", "langchain-community~=0.3.14", "langchain-openai~=0.3.0" ]
-livekit = [ "livekit~=0.19.1", "livekit-api~=0.8.1", "tenacity~=9.0.0" ]
+langchain = [ "langchain~=0.3.20", "langchain-community~=0.3.20", "langchain-openai~=0.3.9" ]
+livekit = [ "livekit~=0.22.0", "livekit-api~=0.8.2", "tenacity~=9.0.0" ]
 lmnt = [ "websockets~=13.1" ]
 local = [ "pyaudio~=0.2.14" ]
 moondream = [ "einops~=0.8.0", "timm~=1.0.13", "transformers~=4.48.0" ]
 nim = []
 noisereduce = [ "noisereduce~=3.0.3" ]
 openai = [ "websockets~=13.1" ]
-openpipe = [ "openpipe~=4.45.0" ]
+openpipe = [ "openpipe~=4.48.0" ]
+openrouter = []
 perplexity = []
 playht = [ "pyht~=0.1.12", "websockets~=13.1" ]
 rime = [ "websockets~=13.1" ]
-riva = [ "nvidia-riva-client~=2.18.0" ]
-sentry = [ "sentry-sdk~=2.20.0" ]
+riva = [ "nvidia-riva-client~=2.19.0" ]
+sentry = [ "sentry-sdk~=2.23.1" ]
 silero = [ "onnxruntime~=1.20.1" ]
 simli = [ "simli-ai~=0.1.10"]
 soundfile = [ "soundfile~=0.13.0" ]
@@ -83,7 +84,6 @@ together = []
 ultravox = [ "transformers~=4.48.0", "vllm~=0.7.3" ]
 websocket = [ "websockets~=13.1", "fastapi~=0.115.6" ]
 whisper = [ "faster-whisper~=1.1.1" ]
-openrouter = []
 
 [tool.setuptools.packages.find]
 # All the following settings are optional:

src/pipecat/processors/aggregators/llm_response.py

@@ -275,7 +275,7 @@ class LLMUserContextAggregator(LLMContextResponseAggregator):
         self,
         context: OpenAILLMContext,
         aggregation_timeout: float = 1.0,
-        bot_interruption_timeout: float = 2.0,
+        bot_interruption_timeout: float = 5.0,
         **kwargs,
     ):
         super().__init__(context=context, role="user", **kwargs)

src/pipecat/services/ai_services.py

@@ -14,7 +14,6 @@ from loguru import logger
 
 from pipecat.adapters.base_llm_adapter import BaseLLMAdapter
 from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter
-from pipecat.audio.utils import calculate_audio_volume, exp_smoothing
 from pipecat.frames.frames import (
     AudioRawFrame,
     BotStartedSpeakingFrame,
@@ -38,6 +37,8 @@ from pipecat.frames.frames import (
     TTSTextFrame,
     TTSUpdateSettingsFrame,
     UserImageRequestFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
     VisionImageRawFrame,
 )
 from pipecat.metrics.metrics import MetricsData
@@ -859,79 +860,64 @@ class STTService(AIService):
 
 
 class SegmentedSTTService(STTService):
-    """SegmentedSTTService is an STTService that will detect speech and will run
-    speech-to-text on speech segments only, instead of a continous stream.
+    """SegmentedSTTService is an STTService that uses VAD events to detect
+    speech and will run speech-to-text on speech segments only, instead of a
+    continous stream. Since it uses VAD it means that VAD needs to be enabled in
+    the pipeline.
+
+    This service always keeps a small audio buffer to take into account that VAD
+    events are delayed from when the user speech really starts.
 
     """
 
-    def __init__(
-        self,
-        *,
-        min_volume: float = 0.6,
-        max_silence_secs: float = 0.3,
-        max_buffer_secs: float = 1.5,
-        sample_rate: Optional[int] = None,
-        **kwargs,
-    ):
+    def __init__(self, *, sample_rate: Optional[int] = None, **kwargs):
         super().__init__(sample_rate=sample_rate, **kwargs)
-        self._min_volume = min_volume
-        self._max_silence_secs = max_silence_secs
-        self._max_buffer_secs = max_buffer_secs
         self._content = None
         self._wave = None
-        self._silence_num_frames = 0
-        # Volume exponential smoothing
-        self._smoothing_factor = 0.2
-        self._prev_volume = 0
-
-    async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
-        # Try to filter out empty background noise
-        volume = self._get_smoothed_volume(frame)
-        if volume >= self._min_volume:
-            # If volume is high enough, write new data to wave file
-            self._wave.writeframes(frame.audio)
-            self._silence_num_frames = 0
-        else:
-            self._silence_num_frames += frame.num_frames
-        self._prev_volume = volume
-
-        # If buffer is not empty and we have enough data or there's been a long
-        # silence, transcribe the audio gathered so far.
-        silence_secs = self._silence_num_frames / self.sample_rate
-        buffer_secs = self._wave.getnframes() / self.sample_rate
-        if self._content.tell() > 0 and (
-            buffer_secs > self._max_buffer_secs or silence_secs > self._max_silence_secs
-        ):
-            self._silence_num_frames = 0
-            self._wave.close()
-            self._content.seek(0)
-            await self.process_generator(self.run_stt(self._content.read()))
-            (self._content, self._wave) = self._new_wave()
+        self._audio_buffer = bytearray()
+        self._audio_buffer_size_1s = 0
+        self._user_speaking = False
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
-        if not self._wave:
-            (self._content, self._wave) = self._new_wave()
+        self._audio_buffer_size_1s = self.sample_rate * 2
 
-    async def stop(self, frame: EndFrame):
-        await super().stop(frame)
-        self._wave.close()
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
 
-    async def cancel(self, frame: CancelFrame):
-        await super().cancel(frame)
-        self._wave.close()
+        if isinstance(frame, UserStartedSpeakingFrame):
+            await self._handle_user_started_speaking(frame)
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            await self._handle_user_stopped_speaking(frame)
+
+    async def _handle_user_started_speaking(self, frame: UserStartedSpeakingFrame):
+        self._user_speaking = True
+
+    async def _handle_user_stopped_speaking(self, frame: UserStoppedSpeakingFrame):
+        self._user_speaking = False
 
-    def _new_wave(self):
         content = io.BytesIO()
-        ww = wave.open(content, "wb")
-        ww.setsampwidth(2)
-        ww.setnchannels(1)
-        ww.setframerate(self.sample_rate)
-        return (content, ww)
+        wav = wave.open(content, "wb")
+        wav.setsampwidth(2)
+        wav.setnchannels(1)
+        wav.setframerate(self.sample_rate)
+        wav.writeframes(self._audio_buffer)
+        wav.close()
+        content.seek(0)
 
-    def _get_smoothed_volume(self, frame: AudioRawFrame) -> float:
-        volume = calculate_audio_volume(frame.audio, frame.sample_rate)
-        return exp_smoothing(volume, self._prev_volume, self._smoothing_factor)
+        await self.process_generator(self.run_stt(content.read()))
+
+        # Start clean.
+        self._audio_buffer.clear()
+
+    async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
+        # If the user is speaking the audio buffer will keep growin.
+        self._audio_buffer += frame.audio
+
+        # If the user is not speaking we keep just a little bit of audio.
+        if not self._user_speaking and len(self._audio_buffer) > self._audio_buffer_size_1s:
+            discarded = len(self._audio_buffer) - self._audio_buffer_size_1s
+            self._audio_buffer = self._audio_buffer[discarded:]
 
 
 class ImageGenService(AIService):

src/pipecat/services/openai.py

@@ -472,22 +472,16 @@ class OpenAITTSService(TTSService):
     """OpenAI Text-to-Speech service that generates audio from text.
 
     This service uses the OpenAI TTS API to generate PCM-encoded audio at 24kHz.
-    When using with DailyTransport, configure the sample rate in DailyParams
-    as shown below:
-
-    DailyParams(
-        audio_out_enabled=True,
-        audio_out_sample_rate=24_000,
-    )
 
     Args:
         api_key: OpenAI API key. Defaults to None.
         voice: Voice ID to use. Defaults to "alloy".
-        model: TTS model to use ("tts-1" or "tts-1-hd"). Defaults to "tts-1".
-        sample_rate: Output audio sample rate in Hz. Defaults to 24000.
+        model: TTS model to use. Defaults to "tts-1".
+        sample_rate: Output audio sample rate in Hz. Defaults to None.
         **kwargs: Additional keyword arguments passed to TTSService.
 
     The service returns PCM-encoded audio at the specified sample rate.
+
     """
 
     OPENAI_SAMPLE_RATE = 24000  # OpenAI TTS always outputs at 24kHz
@@ -497,7 +491,7 @@ class OpenAITTSService(TTSService):
         *,
         api_key: Optional[str] = None,
         voice: str = "alloy",
-        model: Literal["tts-1", "tts-1-hd"] = "tts-1",
+        model: str = "tts-1",
         sample_rate: Optional[int] = None,
         **kwargs,
     ):