diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index f79e3509b..f72cfcf4c 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -55,6 +55,7 @@ from pipecat.frames.frames import ( LLMThoughtStartFrame, LLMThoughtTextFrame, StartFrame, + STTMetadataFrame, TextFrame, TranscriptionFrame, TranslationFrame, @@ -701,10 +702,14 @@ class LLMUserAggregator(LLMContextAggregator): # (`_aggregator_gathers_transcripts == True`): # `on_user_turn_stopped` has fired with empty content, and the # aggregator is waiting on `_transcript_gather_task` before - # finalizing the user message into context. + # finalizing the user message into context. The gather window + # duration is taken from the last `STTMetadataFrame` seen + # (`STTMetadataFrame.ttfs_p99_latency`), falling back to + # `DEFAULT_TTFS_P99` if no STT service has reported one. self._gathering_for_strategy: BaseUserTurnStopStrategy | None = None self._inference_during_gather: bool = False self._transcript_gather_task: asyncio.Task | None = None + self._stt_ttfs_p99_latency: float | None = None self._user_turn_controller = UserTurnController( user_turn_strategies=user_turn_strategies, @@ -861,6 +866,13 @@ class LLMUserAggregator(LLMContextAggregator): # Interim transcriptions and translations are consumed here # and not pushed downstream, same as final TranscriptionFrame. pass + elif isinstance(frame, STTMetadataFrame): + # Record the STT service's reported P99 TTFS so the + # transcript-gather timer can size itself to the real + # latency. Frame is also pushed downstream so other + # processors keep seeing it. + self._stt_ttfs_p99_latency = frame.ttfs_p99_latency + await self.push_frame(frame, direction) elif isinstance(frame, LLMRunFrame): await self._handle_llm_run(frame) elif isinstance(frame, LLMMessagesAppendFrame): @@ -1151,8 +1163,13 @@ class LLMUserAggregator(LLMContextAggregator): await self._call_event_handler("on_user_turn_stopped", strategy, end_of_turn_message) self._gathering_for_strategy = strategy + gather_timeout = ( + self._stt_ttfs_p99_latency + if self._stt_ttfs_p99_latency is not None + else DEFAULT_TTFS_P99 + ) self._transcript_gather_task = self.create_task( - self._transcript_gather_handler(DEFAULT_TTFS_P99), + self._transcript_gather_handler(gather_timeout), f"{self}::transcript_gather", ) return diff --git a/tests/test_context_aggregators_universal.py b/tests/test_context_aggregators_universal.py index 85120a47f..81c29b5f4 100644 --- a/tests/test_context_aggregators_universal.py +++ b/tests/test_context_aggregators_universal.py @@ -830,6 +830,64 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase): user_messages = [m for m in context.get_messages() if m.get("role") == "user"] self.assertEqual([m["content"] for m in user_messages], ["Hello!"]) + async def test_no_wait_for_transcript_uses_stt_metadata_for_gather_timer(self): + """The transcript-gather timer prefers the STT-reported P99 TTFS + over ``DEFAULT_TTFS_P99``. With a long ``DEFAULT_TTFS_P99`` and + a short STT-reported value, the gather completes by the shorter + time — if the timer fell back to ``DEFAULT_TTFS_P99``, this test + would hang. + """ + from unittest.mock import patch + + from pipecat.frames.frames import STTMetadataFrame + from pipecat.processors.aggregators import llm_response_universal + + with patch.object(llm_response_universal, "DEFAULT_TTFS_P99", 60.0): + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies( + stop=[ + SpeechTimeoutUserTurnStopStrategy( + user_speech_timeout=TRANSCRIPTION_TIMEOUT + ) + ], + ), + wait_for_transcript_to_end_user_turn=False, + ), + ) + + events: list[tuple[str, str | None]] = [] + + @user_aggregator.event_handler("on_user_turn_stopped") + async def on_stopped(aggregator, strategy, message): + events.append(("stopped", message.content)) + + @user_aggregator.event_handler("on_user_turn_message_finalized") + async def on_finalized(aggregator, strategy, message): + events.append(("finalized", message.content)) + + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + # STT service advertises its P99 TTFS latency. + STTMetadataFrame(service_name="TestSTT", ttfs_p99_latency=TRANSCRIPTION_TIMEOUT), + VADUserStartedSpeakingFrame(), + SleepFrame(), + VADUserStoppedSpeakingFrame(), + # Let the user_speech_timeout fire so the strategy + # fires turn-stopped. + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05), + TranscriptionFrame(text="Hello!", user_id="", timestamp="now"), + # Wait for the transcript-gather timer to fire (sized + # to the STT-reported TTFS, not DEFAULT_TTFS_P99). + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + self.assertEqual(events, [("stopped", None), ("finalized", "Hello!")]) + async def test_no_wait_for_transcript_no_transcripts_arrive(self): """When no transcripts arrive, the transcript-gather timer still runs — ``on_user_turn_message_finalized`` fires with empty