From 483b643b07a85fe85be72dd990187293c920edad Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Mon, 23 Mar 2026 17:57:51 -0400 Subject: [PATCH] Warn when VAD stop_secs misconfiguration may degrade turn detection Add warnings in SpeechTimeoutUserTurnStopStrategy and TurnAnalyzerUserTurnStopStrategy when stop_secs differs from the recommended default (0.2s) or when stop_secs >= STT p99 latency, which collapses the STT wait timeout to 0s. Document the stop_secs=0.2 assumption in stt_latency.py. --- src/pipecat/services/stt_latency.py | 4 + .../speech_timeout_user_turn_stop_strategy.py | 25 ++++++ .../turn_analyzer_user_turn_stop_strategy.py | 27 +++++++ tests/test_user_turn_stop_strategy.py | 79 +++++++++++++++++++ 4 files changed, 135 insertions(+) diff --git a/src/pipecat/services/stt_latency.py b/src/pipecat/services/stt_latency.py index 351e041a6..4396a787a 100644 --- a/src/pipecat/services/stt_latency.py +++ b/src/pipecat/services/stt_latency.py @@ -13,6 +13,10 @@ transcript is received. These values are used by turn stop strategies to optimize timing. Each STT service publishes its latency via STTMetadataFrame at pipeline start. +All built-in values were measured with VADParams.stop_secs=0.2, the recommended +default. If you change stop_secs, re-run the benchmark with your VAD settings +and pass the measured value to your STT service constructor. + To measure latency for your specific deployment (region, network conditions, self-hosted instances), use the STT benchmark tool: https://github.com/pipecat-ai/stt-benchmark diff --git a/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py index 341a4c1e3..1e127248f 100644 --- a/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py +++ b/src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py @@ -10,6 +10,9 @@ import asyncio import time from typing import Optional +from loguru import logger + +from pipecat.audio.vad.vad_analyzer import VAD_STOP_SECS from pipecat.frames.frames import ( Frame, STTMetadataFrame, @@ -51,6 +54,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): self._user_speech_timeout = user_speech_timeout self._stt_timeout: float = 0.0 # STT P99 latency from STTMetadataFrame self._stop_secs: float = 0.0 # VAD stop_secs from VADUserStoppedSpeakingFrame + self._stop_secs_warned: bool = False self._text = "" self._vad_user_speaking = False @@ -98,6 +102,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): """ if isinstance(frame, STTMetadataFrame): self._stt_timeout = frame.ttfs_p99_latency + self._stop_secs_warned = False elif isinstance(frame, VADUserStartedSpeakingFrame): await self._handle_vad_user_started_speaking(frame) elif isinstance(frame, VADUserStoppedSpeakingFrame): @@ -123,6 +128,26 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): self._stop_secs = frame.stop_secs self._vad_stopped_time = frame.timestamp + if not self._stop_secs_warned: + if self._stop_secs != VAD_STOP_SECS: + self._stop_secs_warned = True + logger.warning( + f"{self}: VAD stop_secs ({self._stop_secs}s) differs from the " + f"recommended default ({VAD_STOP_SECS}s). Built-in p99 latency " + f"values assume stop_secs={VAD_STOP_SECS}. Re-run " + f"https://github.com/pipecat-ai/stt-benchmark with your settings " + f"and pass the TTFS P99 latency result as ttfs_p99_latency to " + f"your STT service." + ) + if self._stt_timeout > 0 and self._stop_secs >= self._stt_timeout: + self._stop_secs_warned = True + logger.warning( + f"{self}: VAD stop_secs ({self._stop_secs}s) >= STT p99 latency " + f"({self._stt_timeout}s). STT wait timeout collapsed to 0s, which " + f"may cause delayed turn detection specified by the " + f"user_turn_stop_timeout parameter in the LLMUserAggregatorParams." + ) + # Start the timeout task timeout = self._calculate_timeout() self._timeout_task = self.task_manager.create_task( diff --git a/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py b/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py index a0df2efbb..895d0976f 100644 --- a/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py +++ b/src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py @@ -9,7 +9,10 @@ import asyncio from typing import Optional +from loguru import logger + from pipecat.audio.turn.base_turn_analyzer import BaseTurnAnalyzer, EndOfTurnState +from pipecat.audio.vad.vad_analyzer import VAD_STOP_SECS from pipecat.frames.frames import ( Frame, InputAudioRawFrame, @@ -54,6 +57,8 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): self._stt_timeout: float = 0.0 # STT P99 latency from STTMetadataFrame self._stop_secs: float = 0.0 # VAD stop_secs from VADUserStoppedSpeakingFrame + self._stop_secs_warned: bool = False + self._text = "" self._turn_complete = False self._vad_user_speaking = False @@ -104,6 +109,7 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): await self._start(frame) elif isinstance(frame, STTMetadataFrame): self._stt_timeout = frame.ttfs_p99_latency + self._stop_secs_warned = False elif isinstance(frame, VADUserStartedSpeakingFrame): await self._handle_vad_user_started_speaking(frame) elif isinstance(frame, VADUserStoppedSpeakingFrame): @@ -163,6 +169,27 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): # Start the STT timeout (adjusted by VAD stop_secs since that time already elapsed) timeout = max(0, self._stt_timeout - self._stop_secs) + + if not self._stop_secs_warned: + if self._stop_secs != VAD_STOP_SECS: + self._stop_secs_warned = True + logger.warning( + f"{self}: VAD stop_secs ({self._stop_secs}s) differs from the " + f"recommended default ({VAD_STOP_SECS}s). Built-in p99 latency " + f"values assume stop_secs={VAD_STOP_SECS}. Re-run " + f"https://github.com/pipecat-ai/stt-benchmark with your settings " + f"and pass the TTFS P99 latency result as ttfs_p99_latency to " + f"your STT service." + ) + if self._stt_timeout > 0 and self._stop_secs >= self._stt_timeout: + self._stop_secs_warned = True + logger.warning( + f"{self}: VAD stop_secs ({self._stop_secs}s) >= STT p99 latency " + f"({self._stt_timeout}s). STT wait timeout collapsed to 0s, which " + f"may cause delayed turn detection specified by the " + f"user_turn_stop_timeout parameter in the LLMUserAggregatorParams." + ) + self._timeout_task = self.task_manager.create_task( self._timeout_handler(timeout), f"{self}::_timeout_handler" ) diff --git a/tests/test_user_turn_stop_strategy.py b/tests/test_user_turn_stop_strategy.py index 85f9f2752..96876fec5 100644 --- a/tests/test_user_turn_stop_strategy.py +++ b/tests/test_user_turn_stop_strategy.py @@ -6,6 +6,7 @@ import asyncio import unittest +from unittest.mock import patch from pipecat.frames.frames import ( InterimTranscriptionFrame, @@ -538,6 +539,84 @@ class TestSpeechTimeoutUserTurnStopStrategy(unittest.IsolatedAsyncioTestCase): self.assertEqual(stop_count, 2) +class TestSpeechTimeoutStopSecsWarnings(unittest.IsolatedAsyncioTestCase): + """Tests for stop_secs misconfiguration warnings.""" + + async def asyncSetUp(self) -> None: + self.task_manager = TaskManager() + self.task_manager.setup(TaskManagerParams(loop=asyncio.get_running_loop())) + + async def _create_strategy(self, stt_timeout=0.35): + strategy = SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=AGGREGATION_TIMEOUT) + await strategy.setup(self.task_manager) + await strategy.process_frame( + STTMetadataFrame(service_name="test", ttfs_p99_latency=stt_timeout) + ) + return strategy + + @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger") + async def test_warns_on_non_default_stop_secs(self, mock_logger): + # Use high stt_timeout so only Warning A fires (stop_secs < stt_timeout) + strategy = await self._create_strategy(stt_timeout=1.0) + + await strategy.process_frame(VADUserStartedSpeakingFrame()) + await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5)) + + mock_logger.warning.assert_called_once() + self.assertIn("differs from the recommended default", mock_logger.warning.call_args[0][0]) + + @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger") + async def test_warns_on_stop_secs_gte_stt_timeout(self, mock_logger): + strategy = await self._create_strategy(stt_timeout=0.35) + + await strategy.process_frame(VADUserStartedSpeakingFrame()) + await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5)) + + # Both warnings fire: non-default stop_secs AND stop_secs >= stt_timeout + self.assertEqual(mock_logger.warning.call_count, 2) + self.assertIn("collapsed to 0s", mock_logger.warning.call_args_list[1][0][0]) + + @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger") + async def test_warns_only_once(self, mock_logger): + # Use high stt_timeout so only Warning A fires + strategy = await self._create_strategy(stt_timeout=1.0) + + # First VAD stop — triggers warning + await strategy.process_frame(VADUserStartedSpeakingFrame()) + await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5)) + self.assertEqual(mock_logger.warning.call_count, 1) + + # Second VAD stop — no duplicate warning + await strategy.process_frame(VADUserStartedSpeakingFrame()) + await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5)) + self.assertEqual(mock_logger.warning.call_count, 1) + + @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger") + async def test_warning_resets_on_new_stt_metadata(self, mock_logger): + # Use high stt_timeout so only Warning A fires + strategy = await self._create_strategy(stt_timeout=1.0) + + await strategy.process_frame(VADUserStartedSpeakingFrame()) + await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5)) + self.assertEqual(mock_logger.warning.call_count, 1) + + # New STTMetadataFrame resets the warned flag + await strategy.process_frame(STTMetadataFrame(service_name="test", ttfs_p99_latency=1.0)) + + await strategy.process_frame(VADUserStartedSpeakingFrame()) + await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.5)) + self.assertEqual(mock_logger.warning.call_count, 2) + + @patch("pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy.logger") + async def test_no_warning_on_default_stop_secs(self, mock_logger): + strategy = await self._create_strategy() + + await strategy.process_frame(VADUserStartedSpeakingFrame()) + await strategy.process_frame(VADUserStoppedSpeakingFrame(stop_secs=0.2)) + + mock_logger.warning.assert_not_called() + + class TestExternalUserTurnStopStrategy(unittest.IsolatedAsyncioTestCase): async def test_external_strategy(self): strategy = ExternalUserTurnStopStrategy()