diff --git a/changelog/4480.added.md b/changelog/4480.added.md new file mode 100644 index 000000000..e472beaab --- /dev/null +++ b/changelog/4480.added.md @@ -0,0 +1 @@ +- Added the configuration surface to drive a realtime service like Gemini Live from local turn detection without paying user-transcript latency. Cascaded pipelines wait for a transcript before ending the user's turn because the downstream LLM needs the user's words recorded in context — but that wait is pure latency in pipelines using local turn detection to drive a realtime service, which consumes user audio directly. Set `wait_for_transcript_to_end_user_turn=False` on `LLMUserAggregatorParams` to turn this on: the aggregator drops `TranscriptionUserTurnStartStrategy` from the start strategies (so late-arriving realtime transcripts don't trigger new turns), sets `wait_for_transcript=False` on any stop strategies that support it (so the turn ends without waiting for a transcript), fires `on_user_turn_stopped` on the audible end of the turn (with empty content, since the transcript hasn't arrived), and defers the context flush until the transcript arrives. A new `on_user_turn_message_finalized` event fires when the user's message has been written to context — in the default mode it coincides with `on_user_turn_stopped`; in the delayed-transcript mode it fires later. Consumers that want the populated transcript should subscribe to `on_user_turn_message_finalized`. See `examples/realtime/realtime-gemini-live-local-vad.py` for the full pattern. diff --git a/changelog/4480.fixed.md b/changelog/4480.fixed.md deleted file mode 100644 index 8b686735f..000000000 --- a/changelog/4480.fixed.md +++ /dev/null @@ -1 +0,0 @@ -- `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/examples/realtime/realtime-gemini-live-local-vad.py b/examples/realtime/realtime-gemini-live-local-vad.py index c7580f0c6..379cb7364 100644 --- a/examples/realtime/realtime-gemini-live-local-vad.py +++ b/examples/realtime/realtime-gemini-live-local-vad.py @@ -70,10 +70,26 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): }, ], ) + # `wait_for_transcript_to_end_user_turn=False` configures the user + # aggregator for realtime services like Gemini Live that emit user + # transcripts after the audible end of the turn. With this flag the + # aggregator: + # + # - drops `TranscriptionUserTurnStartStrategy` from the default start + # strategies (so late-arriving realtime transcripts don't trigger + # new turns), + # - sets `wait_for_transcript=False` on the default stop strategy + # (so the turn ends without waiting for a transcript), + # - fires `on_user_turn_stopped` on the audible end of the turn with + # empty `message.content` (the transcript hasn't arrived yet), and + # - defers the context flush until the (late) transcript arrives, then + # emits `on_user_turn_message_finalized` with the populated message + # so the user's words land in the LLM context for audit/history. user_aggregator, assistant_aggregator = LLMContextAggregatorPair( context, user_params=LLMUserAggregatorParams( vad_analyzer=SileroVADAnalyzer(), + wait_for_transcript_to_end_user_turn=False, ), ) @@ -107,8 +123,19 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Client disconnected") await task.cancel() + # With `wait_for_transcript_to_end_user_turn=False`, `on_user_turn_stopped` + # fires on the audible end of the turn (before the transcript arrives), so + # its `message.content` is empty. Logged here to make the timing of the + # audible-stop signal visible alongside the later transcript event. @user_aggregator.event_handler("on_user_turn_stopped") async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMessage): + logger.info(f"User turn ended (audible, strategy: {type(strategy).__name__})") + + # `on_user_turn_message_finalized` fires when the user transcript has + # been written to context — later than `on_user_turn_stopped` in this + # mode, since transcripts arrive after the audible turn end. + @user_aggregator.event_handler("on_user_turn_message_finalized") + async def on_user_turn_message_finalized(aggregator, strategy, message: UserTurnStoppedMessage): timestamp = f"[{message.timestamp}] " if message.timestamp else "" line = f"{timestamp}user: {message.content}" logger.info(f"Transcript: {line}") diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index b50eac1f7..13ebf28ec 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -80,6 +80,7 @@ from pipecat.processors.aggregators.llm_context_summarizer import ( SummaryAppliedEvent, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.services.stt_latency import DEFAULT_TTFS_P99 from pipecat.turns.user_idle_controller import UserIdleController from pipecat.turns.user_mute import BaseUserMuteStrategy from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams @@ -127,6 +128,30 @@ class LLMUserAggregatorParams: has been idle (not speaking) for this duration. Set to 0 to disable idle detection. vad_analyzer: Voice Activity Detection analyzer instance. + wait_for_transcript_to_end_user_turn: Defaults to True. Set to + False when using local turn detection to drive a realtime + service like Gemini Live, where the LLM consumes user audio + directly and doesn't need the transcript in context to respond + — so waiting for the transcript before ending the turn is pure + latency. When False, the aggregator: + + - drops ``TranscriptionUserTurnStartStrategy`` from the start + strategies (so late-arriving realtime transcripts don't + trigger new turns), + - sets ``wait_for_transcript=False`` on any stop strategy that + supports it (so the turn ends without waiting for a + transcript), and + - defers the context flush and public ``on_user_turn_stopped`` + event until the (late) transcript arrives or a backstop timer + fires (so the user's words still land in the LLM context). + + Adjustments to user-provided strategies are logged. The + precondition for the deferred-context path: every user turn + produces one transcript in conversational order — turn N's + transcript arrives before any of turn N's assistant content, and + before turn N+1 starts. Violations are logged and + best-effort-recovered by force-flushing the pending turn before + starting the next one. filter_incomplete_user_turns: [DEPRECATED] Use ``user_turn_strategies=FilterIncompleteUserTurnStrategies()`` instead. When enabled, the LLM outputs a turn-completion @@ -157,6 +182,7 @@ class LLMUserAggregatorParams: user_turn_stop_timeout: float = 5.0 user_idle_timeout: float = 0 vad_analyzer: VADAnalyzer | None = None + wait_for_transcript_to_end_user_turn: bool = True filter_incomplete_user_turns: bool = False user_turn_completion_config: UserTurnCompletionConfig | None = None @@ -527,7 +553,18 @@ class LLMUserAggregator(LLMContextAggregator): Event handlers available: - on_user_turn_started: Called when the user turn starts - - on_user_turn_stopped: Called when the user turn ends + - on_user_turn_stopped: Called when the user turn ends. With + ``wait_for_transcript_to_end_user_turn=True`` (default), this fires + with the populated ``UserTurnStoppedMessage.content``. With + ``wait_for_transcript_to_end_user_turn=False`` (realtime + local + turn detection), this fires on the audible end of the turn — before + the transcript has arrived — so ``content`` is empty; subscribe to + ``on_user_turn_message_finalized`` to get the transcript. + - on_user_turn_message_finalized: Called once the user message has + been finalized into the context. Fires in both modes — in the + default mode it coincides with ``on_user_turn_stopped``; in the + delayed-transcript path it fires later, when the transcript + arrives. The ``UserTurnStoppedMessage.content`` is always populated. - on_user_turn_stop_timeout: Called when no user turn stop strategy triggers - on_user_turn_idle: Called when the user has been idle for the configured timeout - on_user_mute_started: Called when the user becomes muted @@ -543,6 +580,10 @@ class LLMUserAggregator(LLMContextAggregator): async def on_user_turn_stopped(aggregator, strategy: BaseUserTurnStopStrategy, message: UserTurnStoppedMessage): ... + @aggregator.event_handler("on_user_turn_message_finalized") + async def on_user_turn_message_finalized(aggregator, strategy: BaseUserTurnStopStrategy, message: UserTurnStoppedMessage): + ... + @aggregator.event_handler("on_user_turn_stop_timeout") async def on_user_turn_stop_timeout(aggregator): ... @@ -586,12 +627,14 @@ class LLMUserAggregator(LLMContextAggregator): self._register_event_handler("on_user_turn_started") self._register_event_handler("on_user_turn_stopped") + self._register_event_handler("on_user_turn_message_finalized") self._register_event_handler("on_user_turn_stop_timeout") self._register_event_handler("on_user_turn_idle") self._register_event_handler("on_user_turn_inference_triggered") self._register_event_handler("on_user_mute_started") self._register_event_handler("on_user_mute_stopped") + user_provided_strategies = self._params.user_turn_strategies is not None user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies() # Deprecated path: translate filter_incomplete_user_turns into @@ -605,6 +648,16 @@ class LLMUserAggregator(LLMContextAggregator): ) self._params.user_turn_strategies = user_turn_strategies + # When the user opts out of waiting for transcripts to end the user + # turn, mutate the strategies to match — drop the transcription start + # strategy, flip `wait_for_transcript=False` on the stop strategies + # that support it. Loud log if the user passed their own strategies + # (we're overwriting parts of their config); quiet log otherwise. + if not self._params.wait_for_transcript_to_end_user_turn: + self._apply_no_transcript_wait_bundle( + user_turn_strategies, user_provided_strategies=user_provided_strategies + ) + self._user_is_muted = False self._user_turn_start_timestamp = "" # Full transcript across the user turn. Each @@ -616,6 +669,15 @@ class LLMUserAggregator(LLMContextAggregator): # inferences fire before finalization. self._full_user_turn_aggregation: str | None = None + # State for the deferred-finalization path that activates when + # `wait_for_transcript_to_end_user_turn=False`. The strategy fires + # turn-stopped on the audible end of the user's turn; we defer the + # context flush and `on_user_turn_stopped` event until the transcript + # arrives (or a backstop timer fires). + self._pending_stop_strategy: BaseUserTurnStopStrategy | None = None + self._pending_inference_trigger: bool = False + self._delayed_transcript_backstop_task: asyncio.Task | None = None + self._user_turn_controller = UserTurnController( user_turn_strategies=user_turn_strategies, user_turn_stop_timeout=self._params.user_turn_stop_timeout, @@ -658,6 +720,75 @@ class LLMUserAggregator(LLMContextAggregator): self._vad_controller.add_event_handler("on_push_frame", self._on_push_frame) self._vad_controller.add_event_handler("on_broadcast_frame", self._on_broadcast_frame) + @property + def _expect_delayed_transcripts(self) -> bool: + """Internal alias for the deferred-finalization mode. + + Active whenever ``wait_for_transcript_to_end_user_turn`` is False, + which is also when the strategy-mutation bundle is applied. The two + always travel together — there's no configuration where one is + active without the other. + """ + return not self._params.wait_for_transcript_to_end_user_turn + + def _apply_no_transcript_wait_bundle( + self, + user_turn_strategies: UserTurnStrategies, + *, + user_provided_strategies: bool, + ): + """Adjust strategies to match ``wait_for_transcript_to_end_user_turn=False``. + + Drops ``TranscriptionUserTurnStartStrategy`` from start strategies + (so late-arriving realtime transcripts don't trigger new turns) and + sets ``wait_for_transcript=False`` on any stop strategy that supports + it (so the turn ends without waiting for a transcript). + + Logs loudly when adjusting user-provided strategies — we're + mutating objects the caller passed in. Logs quietly when only + synthesized defaults are in play. + """ + # Local import to avoid a top-level cycle with `turns.user_start`. + from pipecat.turns.user_start import TranscriptionUserTurnStartStrategy + + adjustments: list[str] = [] + + if user_turn_strategies.start: + filtered = [ + s + for s in user_turn_strategies.start + if not isinstance(s, TranscriptionUserTurnStartStrategy) + ] + dropped = len(user_turn_strategies.start) - len(filtered) + if dropped: + user_turn_strategies.start = filtered + adjustments.append( + f"dropped {dropped} TranscriptionUserTurnStartStrategy from start strategies" + ) + + flipped = 0 + for s in user_turn_strategies.stop or []: + if hasattr(s, "_wait_for_transcript") and s._wait_for_transcript: + s._wait_for_transcript = False + flipped += 1 + if flipped: + adjustments.append( + f"set wait_for_transcript=False on {flipped} stop " + f"strateg{'y' if flipped == 1 else 'ies'}" + ) + + if not adjustments: + return + + message = ( + f"{self}: wait_for_transcript_to_end_user_turn=False adjusted " + f"user turn strategies: {'; '.join(adjustments)}." + ) + if user_provided_strategies: + logger.warning(message) + else: + logger.info(message) + async def cleanup(self): """Clean up processor resources.""" await super().cleanup() @@ -747,13 +878,26 @@ class LLMUserAggregator(LLMContextAggregator): await s.setup(self.task_manager) async def _stop(self, frame: EndFrame): - await self._maybe_emit_user_turn_stopped(on_session_end=True) + await self._finalize_on_session_end() await self._cleanup() async def _cancel(self, frame: CancelFrame): - await self._maybe_emit_user_turn_stopped(on_session_end=True) + await self._finalize_on_session_end() await self._cleanup() + async def _finalize_on_session_end(self): + """Flush any pending turn-stop on session end. + + If a deferred turn finalization is pending (because + ``expect_delayed_transcripts`` is True and the transcript hadn't + arrived yet), replay it first so the user message is captured before + the session shuts down. + """ + if self._pending_stop_strategy is not None or self._pending_inference_trigger: + await self._replay_deferred_turn_finalization(on_session_end=True) + else: + await self._maybe_emit_user_turn_stopped(on_session_end=True) + async def _cleanup(self): if self._vad_controller: await self._vad_controller.cleanup() @@ -884,6 +1028,17 @@ class LLMUserAggregator(LLMContextAggregator): ): logger.debug(f"{self}: User started speaking (strategy: {strategy})") + # Precondition guard for `expect_delayed_transcripts`: if the previous + # turn's finalization is still pending, the precondition (turn N's + # transcript arrives before turn N+1 starts) has been violated. + # Force-flush the previous turn before proceeding. + if self._pending_stop_strategy is not None or self._pending_inference_trigger: + logger.warning( + f"{self}: user turn started while previous turn's transcript was " + f"still pending; flushing previous turn now" + ) + await self._replay_deferred_turn_finalization() + self._user_turn_start_timestamp = time_now_iso8601() self._full_user_turn_aggregation = None @@ -904,6 +1059,12 @@ class LLMUserAggregator(LLMContextAggregator): ): logger.debug(f"{self}: User turn inference triggered (strategy: {strategy})") + if self._expect_delayed_transcripts: + # Defer the push_aggregation and event emission until the + # transcript arrives (or the backstop fires). + self._pending_inference_trigger = True + return + # Push aggregation now: this writes the user message segment to # the context and emits LLMContextFrame, which kicks LLM # inference. Concatenate the segment into @@ -929,19 +1090,106 @@ class LLMUserAggregator(LLMContextAggregator): ): logger.debug(f"{self}: User stopped speaking (strategy: {strategy})") + # Audible-stop side effects fire on the strategy event regardless of + # whether the transcript-ready finalization is deferred. if params.enable_user_speaking_frames: await self.broadcast_frame(UserStoppedSpeakingFrame) await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame()) + if self._expect_delayed_transcripts: + # Fire `on_user_turn_stopped` now to signal the audible end of the + # turn. `content` is empty because the transcript hasn't arrived + # yet — consumers that want the transcript subscribe to + # `on_user_turn_message_finalized` instead, which fires later when + # the transcript lands (or the backstop timer fires). + audible_stop_message = UserTurnStoppedMessage( + content="", timestamp=self._user_turn_start_timestamp + ) + await self._call_event_handler("on_user_turn_stopped", strategy, audible_stop_message) + + self._pending_stop_strategy = strategy + self._delayed_transcript_backstop_task = self.create_task( + self._delayed_transcript_backstop_handler(DEFAULT_TTFS_P99), + f"{self}::delayed_transcript_backstop", + ) + return + await self._maybe_emit_user_turn_stopped(strategy) + async def _delayed_transcript_backstop_handler(self, timeout: float): + """Backstop timer for `expect_delayed_transcripts`. + + Fires after ``timeout`` seconds and replays the deferred turn + finalization with whatever transcript has been captured by then + (possibly nothing). + """ + try: + await asyncio.sleep(timeout) + except asyncio.CancelledError: + return + finally: + self._delayed_transcript_backstop_task = None + + await self._replay_deferred_turn_finalization() + + async def _replay_deferred_turn_finalization(self, *, on_session_end: bool = False): + """Replay the deferred halves of inference-triggered and turn-stopped. + + Called from the backstop timer, the precondition guard in + `_on_user_turn_started`, and the session-end paths. Clears all + pending state before delegating to the normal finalization methods + so they run the unmodified code path. + """ + if self._delayed_transcript_backstop_task: + await self.cancel_task(self._delayed_transcript_backstop_task) + self._delayed_transcript_backstop_task = None + + pending_strategy = self._pending_stop_strategy + had_pending_inference = self._pending_inference_trigger + self._pending_stop_strategy = None + self._pending_inference_trigger = False + + if had_pending_inference: + segment = await self.push_aggregation() + if segment: + if self._full_user_turn_aggregation: + self._full_user_turn_aggregation = ( + f"{self._full_user_turn_aggregation} {segment}".strip() + ) + else: + self._full_user_turn_aggregation = segment + await self._call_event_handler("on_user_turn_inference_triggered", pending_strategy) + + if pending_strategy is not None or on_session_end: + # `on_user_turn_stopped` already fired on the audible stop; + # only fire `on_user_turn_message_finalized` here. + await self._maybe_emit_user_turn_stopped( + pending_strategy, + on_session_end=on_session_end, + emit_stopped_event=False, + ) + async def _on_reset_aggregation( self, controller: UserTurnController, strategy: BaseUserTurnStartStrategy ): logger.debug(f"{self}: Resetting aggregation (strategy: {strategy})") + await self._discard_deferred_turn_finalization() await self.reset() + async def _discard_deferred_turn_finalization(self): + """Drop any pending deferred turn finalization without replaying. + + Called from reset paths (interruption, explicit reset). "Reset" means + "throw it away" — we don't want to flush a partial transcript that + was about to be invalidated anyway. + """ + if self._delayed_transcript_backstop_task: + await self.cancel_task(self._delayed_transcript_backstop_task) + self._delayed_transcript_backstop_task = None + self._pending_stop_strategy = None + self._pending_inference_trigger = False + async def _on_user_turn_stop_timeout(self, controller): await self._call_event_handler("on_user_turn_stop_timeout") @@ -952,19 +1200,28 @@ class LLMUserAggregator(LLMContextAggregator): self, strategy: BaseUserTurnStopStrategy | None = None, on_session_end: bool = False, + *, + emit_stopped_event: bool = True, ): - """Maybe emit user turn stopped event. + """Finalize the user message and emit the corresponding events. Earlier inference triggers in the same turn have already pushed their segments to the context and accumulated them into ``self._full_user_turn_aggregation``. Any aggregation that arrived after the last inference trigger is flushed here so - end-of-turn content is never lost from the public event. + end-of-turn content is never lost from the public events. + + Always emits ``on_user_turn_message_finalized``. Also emits + ``on_user_turn_stopped`` when ``emit_stopped_event=True`` (the + default path); the delayed-transcript path passes False because it + already fired ``on_user_turn_stopped`` on the audible stop. Args: strategy: The strategy that triggered the turn stop. on_session_end: If True, only emit if there's unemitted content (avoids duplicate events when session ends). + emit_stopped_event: Whether to fire ``on_user_turn_stopped`` + alongside ``on_user_turn_message_finalized``. """ segment = await self.push_aggregation() full_aggregation = self._full_user_turn_aggregation @@ -979,7 +1236,9 @@ class LLMUserAggregator(LLMContextAggregator): message = UserTurnStoppedMessage( content=content, timestamp=self._user_turn_start_timestamp ) - await self._call_event_handler("on_user_turn_stopped", strategy, message) + if emit_stopped_event: + await self._call_event_handler("on_user_turn_stopped", strategy, message) + await self._call_event_handler("on_user_turn_message_finalized", strategy, message) self._user_turn_start_timestamp = "" 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 8df8fedae..7ce80e332 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 @@ -44,23 +44,20 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): 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. + Set ``wait_for_transcript=False`` for pipelines where the downstream LLM + doesn't need the user's words recorded in context to respond — typically + when using local turn detection to drive a realtime service like Gemini + Live. In that case the strategy fires the turn stop as soon as the + user_speech_timeout elapses, without waiting for a transcript. Pair with + ``expect_delayed_transcripts=True`` on ``LLMUserAggregatorParams`` so the + aggregator still captures the transcript when it arrives. """ def __init__( self, *, user_speech_timeout: float = 0.6, - require_transcript: bool | None = None, + wait_for_transcript: bool = True, **kwargs, ): """Initialize the speech timeout-based user turn stop strategy. @@ -68,18 +65,15 @@ class SpeechTimeoutUserTurnStopStrategy(BaseUserTurnStopStrategy): 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. + wait_for_transcript: Whether to wait for a transcript before + ending the user turn. Defaults to True. Set to False when + using local turn detection with a realtime service that + doesn't need user transcripts in context to respond. **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._wait_for_transcript = wait_for_transcript 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 @@ -94,13 +88,6 @@ 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() @@ -138,7 +125,6 @@ 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): @@ -191,12 +177,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 no transcript is - # required (no STT to wait for), the transcript is already finalized, - # or if the VAD stop_secs already covered it. + # stt_timeout is a safety net. Short-circuit it if we're not waiting + # for a transcript, if 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 not self._transcript_required or self._transcript_finalized or effective_stt_wait <= 0: + if not self._wait_for_transcript or self._transcript_finalized or effective_stt_wait <= 0: self._stt_wait_done = True else: self._stt_timeout_task = self.task_manager.create_task( @@ -287,11 +273,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 (skipped when a transcript isn't required). + must have been received (skipped when ``wait_for_transcript`` is False). """ if self._vad_user_speaking: return - if self._transcript_required and not self._text: + if self._wait_for_transcript 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 563105809..37c1ff955 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 @@ -43,41 +43,36 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): 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. + Set ``wait_for_transcript=False`` for pipelines where the downstream LLM + doesn't need the user's words recorded in context to respond — typically + when using local turn detection to drive a realtime service like Gemini + Live. In that case the strategy fires the turn stop as soon as the + analyzer concludes the turn is complete, without waiting for a + transcript. Pair with ``expect_delayed_transcripts=True`` on + ``LLMUserAggregatorParams`` so the aggregator still captures the + transcript when it arrives. """ def __init__( self, *, turn_analyzer: BaseTurnAnalyzer, - require_transcript: bool | None = None, + wait_for_transcript: bool = True, **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. + wait_for_transcript: Whether to wait for a transcript before + ending the user turn. Defaults to True. Set to False when + using local turn detection with a realtime service that + doesn't need user transcripts in context to respond. **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._wait_for_transcript = wait_for_transcript self._stt_timeout: float = 0.0 # STT P99 latency from STTMetadataFrame self._stop_secs: float = 0.0 # VAD stop_secs from VADUserStoppedSpeakingFrame @@ -91,13 +86,6 @@ 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() @@ -141,7 +129,6 @@ 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): @@ -202,7 +189,7 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): # wait for transcriptions. self._turn_complete = state == EndOfTurnState.COMPLETE - if not self._transcript_required: + if not self._wait_for_transcript: # 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. @@ -296,13 +283,13 @@ class TurnAnalyzerUserTurnStopStrategy(BaseUserTurnStopStrategy): """Trigger user turn stopped if conditions are met. Conditions: - - We have transcription text (skipped when a transcript isn't required) + - We have transcription text (skipped when ``wait_for_transcript`` is False) - Turn analyzer indicates turn is complete - Either the timeout has elapsed OR we have a finalized transcript """ if not self._turn_complete: return - if self._transcript_required and not self._text: + if self._wait_for_transcript and not self._text: return # For finalized transcripts, trigger immediately