wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

user turn stop strategies: don't always wait for transcripts

Browse Source 
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.
This commit is contained in:
Paul Kompfner
2026-05-13 12:37:18 -04:00
parent 5fef239b68
commit 6d21507e95
3 changed files with 88 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4480.fixed.md Normal file
View File

@@ -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.

48
src/pipecat/turns/user_stop/speech_timeout_user_turn_stop_strategy.py
View File

@@ -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:

48
src/pipecat/turns/user_stop/turn_analyzer_user_turn_stop_strategy.py
View File

@@ -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