From 6d21507e95b4b4432f669970a5ec2cc7ee685854 Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 13 May 2026 12:37:18 -0400 Subject: [PATCH] user turn stop strategies: don't always wait for transcripts MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Until now, both TurnAnalyzerUserTurnStopStrategy and SpeechTimeoutUserTurnStopStrategy waited for at least one transcript before ending the user turn. That's the right behavior for cascaded pipelines, where the downstream LLM can't respond until the user's words are recorded in its context — but it's pure latency in pipelines using local turn detection to drive a realtime service like Gemini Live. Add a `require_transcript: bool | None = None` parameter to both strategies. When None (default), it infers from whether an STTMetadataFrame has been seen — a proxy for "does the downstream LLM need the transcript in context?". Explicit True/False overrides the heuristic. When a transcript isn't required, the strategies also skip the STT-waiting timeout in the VAD-stopped handler, so the user turn ends as soon as the analyzer (or speech timer) concludes the turn is complete. --- changelog/4480.fixed.md | 1 + .../speech_timeout_user_turn_stop_strategy.py | 48 ++++++++++++++++--- .../turn_analyzer_user_turn_stop_strategy.py | 48 +++++++++++++++++-- 3 files changed, 88 insertions(+), 9 deletions(-) create mode 100644 changelog/4480.fixed.md diff --git a/changelog/4480.fixed.md b/changelog/4480.fixed.md new file mode 100644 index 000000000..8b686735f --- /dev/null +++ b/changelog/4480.fixed.md @@ -0,0 +1 @@ +- `TurnAnalyzerUserTurnStopStrategy` and `SpeechTimeoutUserTurnStopStrategy` no longer always wait for a transcript before ending the user turn. The wait is still correct for cascaded pipelines, where the downstream LLM can't respond until the user's words are recorded in its context — but it's pure latency in pipelines using local turn detection to drive a realtime service like Gemini Live. The strategies now infer from the presence of an `STTMetadataFrame` whether a transcript is needed, and accept a new `require_transcript: bool | None` parameter to override the heuristic explicitly. 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 254fa60b1..8df8fedae 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 @@ -43,18 +43,43 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): (rearmed on each transcript). stt_timeout has no meaning here since it is defined relative to VAD stop, and STT has already emitted a transcript — so the stt wait is marked done immediately. + + Whether a transcript is required to end the turn is controlled by + ``require_transcript``: + + - ``None`` (default): infer from the pipeline. If an ``STTMetadataFrame`` + has been seen, a transcript is required; otherwise it is not. This is a + heuristic — the underlying question is whether the downstream LLM needs + the user's words recorded in context before responding, which we proxy + by "is there an STT service in the pipeline?". Works for typical + cascaded and speech-to-speech setups. + - ``True`` / ``False``: explicit override when the heuristic doesn't fit. """ - def __init__(self, *, user_speech_timeout: float = 0.6, **kwargs): + def __init__( + self, + *, + user_speech_timeout: float = 0.6, + require_transcript: bool | None = None, + **kwargs, + ): """Initialize the speech timeout-based user turn stop strategy. Args: user_speech_timeout: Time to wait for the user to potentially say more after they pause speaking. Defaults to 0.6 seconds. + require_transcript: Whether to wait for a transcript before ending + the user turn. ``None`` (default) infers from the presence of + an ``STTMetadataFrame``. ``True``/``False`` overrides the + heuristic. **kwargs: Additional keyword arguments. """ super().__init__(**kwargs) self._user_speech_timeout = user_speech_timeout + self._require_transcript = require_transcript + # Set when an STTMetadataFrame is received. Used as the heuristic + # signal when require_transcript is None. + self._has_stt: bool = False 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 @@ -69,6 +94,13 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): self._user_speech_wait_done: bool = False self._stt_wait_done: bool = False + @property + def _transcript_required(self) -> bool: + """Whether the current pipeline requires a transcript to end the turn.""" + if self._require_transcript is not None: + return self._require_transcript + return self._has_stt + async def reset(self): """Reset the strategy to its initial state.""" await super().reset() @@ -106,6 +138,7 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): Always returns CONTINUE so subsequent stop strategies are evaluated. """ if isinstance(frame, STTMetadataFrame): + self._has_stt = True self._stt_timeout = frame.ttfs_p99_latency self._stop_secs_warned = False elif isinstance(frame, VADUserStartedSpeakingFrame): @@ -158,11 +191,12 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): # fallback-mode run of the same timer is superseded here. await self._restart_user_speech_timer() - # stt_timeout is a safety net. Short-circuit it if the transcript is - # already finalized, or if the VAD stop_secs already covered it. + # stt_timeout is a safety net. Short-circuit it if no transcript is + # required (no STT to wait for), the transcript is already finalized, + # or if the VAD stop_secs already covered it. self._stt_wait_done = False effective_stt_wait = max(0.0, self._stt_timeout - self._stop_secs) - if self._transcript_finalized or effective_stt_wait <= 0: + if not self._transcript_required or self._transcript_finalized or effective_stt_wait <= 0: self._stt_wait_done = True else: self._stt_timeout_task = self.task_manager.create_task( @@ -253,9 +287,11 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): Both timers must be done (stt is marked done immediately on the fallback path and when finalization short-circuits the safety net), the user must not be currently speaking, and at least one transcript - must have been received. + must have been received (skipped when a transcript isn't required). """ - if self._vad_user_speaking or not self._text: + if self._vad_user_speaking: + return + if self._transcript_required and not self._text: return if self._user_speech_wait_done and self._stt_wait_done: 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 667e89961..563105809 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 @@ -42,17 +42,42 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): the turn can be triggered immediately once the finalized transcript is received. Otherwise, an STT timeout (adjusted by VAD stop_secs) is used as a fallback. + + Whether a transcript is required to end the turn is controlled by + ``require_transcript``: + + - ``None`` (default): infer from the pipeline. If an ``STTMetadataFrame`` + has been seen, a transcript is required; otherwise it is not. This is a + heuristic — the underlying question is whether the downstream LLM needs + the user's words recorded in context before responding, which we proxy + by "is there an STT service in the pipeline?". Works for typical + cascaded and speech-to-speech setups. + - ``True`` / ``False``: explicit override when the heuristic doesn't fit. """ - def __init__(self, *, turn_analyzer: BaseTurnAnalyzer, **kwargs): + def __init__( + self, + *, + turn_analyzer: BaseTurnAnalyzer, + require_transcript: bool | None = None, + **kwargs, + ): """Initialize the user turn stop strategy. Args: turn_analyzer: The turn detection analyzer instance to detect end of user turn. + require_transcript: Whether to wait for a transcript before ending + the user turn. ``None`` (default) infers from the presence of + an ``STTMetadataFrame``. ``True``/``False`` overrides the + heuristic. **kwargs: Additional keyword arguments. """ super().__init__(**kwargs) self._turn_analyzer = turn_analyzer + self._require_transcript = require_transcript + # Set when an STTMetadataFrame is received. Used as the heuristic + # signal when require_transcript is None. + self._has_stt: bool = False self._stt_timeout: float = 0.0 # STT P99 latency from STTMetadataFrame self._stop_secs: float = 0.0 # VAD stop_secs from VADUserStoppedSpeakingFrame @@ -66,6 +91,13 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): self._timeout_task: asyncio.Task | None = None self._timeout_expired: bool = False + @property + def _transcript_required(self) -> bool: + """Whether the current pipeline requires a transcript to end the turn.""" + if self._require_transcript is not None: + return self._require_transcript + return self._has_stt + async def reset(self): """Reset the strategy to its initial state.""" await super().reset() @@ -109,6 +141,7 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): if isinstance(frame, StartFrame): await self._start(frame) elif isinstance(frame, STTMetadataFrame): + self._has_stt = True self._stt_timeout = frame.ttfs_p99_latency self._stop_secs_warned = False elif isinstance(frame, VADUserStartedSpeakingFrame): @@ -169,6 +202,13 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): # wait for transcriptions. self._turn_complete = state == EndOfTurnState.COMPLETE + if not self._transcript_required: + # No transcript to wait for. Trigger now if the turn is already + # complete; otherwise the analyzer's audio path will trigger once + # it indicates completion. + await self._maybe_trigger_user_turn_stopped() + return + # Start the STT timeout (adjusted by VAD stop_secs since that time already elapsed) timeout = max(0, self._stt_timeout - self._stop_secs) @@ -256,11 +296,13 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): """Trigger user turn stopped if conditions are met. Conditions: - - We have transcription text + - We have transcription text (skipped when a transcript isn't required) - Turn analyzer indicates turn is complete - Either the timeout has elapsed OR we have a finalized transcript """ - if not self._text or not self._turn_complete: + if not self._turn_complete: + return + if self._transcript_required and not self._text: return # For finalized transcripts, trigger immediately