From e3e90d38aa995c054c54fd23bd30d1c4df81fddc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 5 May 2026 16:31:56 -0700 Subject: [PATCH] Preserve full user transcript across multiple inferences in one turn MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a stop-strategy chain splits inference-triggered from finalization (e.g. `LLMTurnCompletionUserTurnStopStrategy` gating a deferred detector), more than one inference can fire inside a single user turn — each adds the new transcription segment to the context. Previously each inference overwrote `_pending_user_turn_aggregation`, so the eventual `on_user_turn_stopped` event surfaced only the segment from the last inference, dropping anything the user said before it. Concatenate each segment into `_full_user_turn_aggregation` instead of overwriting, and combine that running buffer with any post-final- inference segment when emitting the public event. --- .../aggregators/llm_response_universal.py | 58 +++++++++------- tests/test_context_aggregators_universal.py | 69 +++++++++++++++++++ 2 files changed, 103 insertions(+), 24 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index 019490929..f136b848d 100644 --- a/src/pipecat/processors/aggregators/llm_response_universal.py +++ b/src/pipecat/processors/aggregators/llm_response_universal.py @@ -588,10 +588,14 @@ class LLMUserAggregator(LLMContextAggregator): self._user_is_muted = False self._user_turn_start_timestamp = "" - # Aggregation captured at inference-trigger time and surfaced again - # in `on_user_turn_stopped`. Set to None when no inference-triggered - # event has fired since the last finalization. - self._pending_user_turn_aggregation: str | None = None + # Full transcript across the user turn. Each + # `_on_user_turn_inference_triggered` push captures only the + # new segment since the previous push (push_aggregation resets + # `_aggregation` after writing to context); we accumulate those + # segments here so the eventual `on_user_turn_stopped` event + # surfaces the full turn transcript even when several + # inferences fire before finalization. + self._full_user_turn_aggregation: str | None = None self._user_turn_controller = UserTurnController( user_turn_strategies=user_turn_strategies, @@ -862,7 +866,7 @@ class LLMUserAggregator(LLMContextAggregator): logger.debug(f"{self}: User started speaking (strategy: {strategy})") self._user_turn_start_timestamp = time_now_iso8601() - self._pending_user_turn_aggregation = None + self._full_user_turn_aggregation = None if params.enable_user_speaking_frames: await self.broadcast_frame(UserStartedSpeakingFrame) @@ -881,12 +885,20 @@ class LLMUserAggregator(LLMContextAggregator): ): logger.debug(f"{self}: User turn inference triggered (strategy: {strategy})") - # Push aggregation now: this writes the user message to the context - # and pushes LLMContextFrame, which is what kicks LLM inference. - # `on_user_turn_stopped` later fires when the turn is semantically - # final and surfaces the aggregated message via the public event. - aggregation = await self.push_aggregation() - self._pending_user_turn_aggregation = aggregation + # Push aggregation now: this writes the user message segment to + # the context and emits LLMContextFrame, which kicks LLM + # inference. Concatenate the segment into + # `_full_user_turn_aggregation` so multiple inferences in the + # same turn don't lose earlier segments from the eventual + # `on_user_turn_stopped` event. + 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", strategy) @@ -924,27 +936,25 @@ class LLMUserAggregator(LLMContextAggregator): ): """Maybe emit user turn stopped event. - The aggregation has typically already been pushed at - inference-trigger time and is cached in - ``self._pending_user_turn_aggregation``. Any aggregation that has - accumulated since the last inference-trigger (e.g. transcriptions - that arrived between inference trigger and finalization) is flushed - here so end-of-turn content is never lost. + 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. 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). """ - aggregation = await self.push_aggregation() - previous_aggregation = self._pending_user_turn_aggregation - self._pending_user_turn_aggregation = None + segment = await self.push_aggregation() + full_aggregation = self._full_user_turn_aggregation + self._full_user_turn_aggregation = None - content = None - if aggregation and previous_aggregation: - content = f"{previous_aggregation} {aggregation}".strip() + if segment and full_aggregation: + content = f"{full_aggregation} {segment}".strip() else: - content = previous_aggregation or aggregation + content = full_aggregation or segment if not on_session_end or content: message = UserTurnStoppedMessage( diff --git a/tests/test_context_aggregators_universal.py b/tests/test_context_aggregators_universal.py index 01ca70c80..d16a34c43 100644 --- a/tests/test_context_aggregators_universal.py +++ b/tests/test_context_aggregators_universal.py @@ -697,6 +697,75 @@ class TestLLMUserAggregator(unittest.IsolatedAsyncioTestCase): self.assertEqual(events, ["inference_triggered", "stopped"]) + async def test_multiple_inferences_in_one_turn_preserve_aggregation(self): + """Two inference triggers before finalization should preserve the full user transcript. + + When the LLM marks the first inference incomplete (○ / ◐) and the + user keeps speaking, the deferred upstream strategy fires a + second inference. Both the public ``on_user_turn_stopped`` event + and the conversation context should reflect the full user + utterance, not just the segment from the last inference. + """ + from pipecat.frames.frames import UserTurnCompletedFrame + from pipecat.turns.user_stop import LLMTurnCompletionUserTurnStopStrategy, deferred + + gating = LLMTurnCompletionUserTurnStopStrategy() + upstream = deferred( + SpeechTimeoutUserTurnStopStrategy(user_speech_timeout=TRANSCRIPTION_TIMEOUT) + ) + context = LLMContext() + user_aggregator = LLMUserAggregator( + context, + params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies(stop=[upstream, gating]), + ), + ) + + inference_count = 0 + stop_message = None + + @user_aggregator.event_handler("on_user_turn_inference_triggered") + async def on_inference_triggered(aggregator, strategy): + nonlocal inference_count + inference_count += 1 + + @user_aggregator.event_handler("on_user_turn_stopped") + async def on_stopped(aggregator, strategy, message): + nonlocal stop_message + stop_message = message + + pipeline = Pipeline([user_aggregator]) + + frames_to_send = [ + VADUserStartedSpeakingFrame(), + TranscriptionFrame(text="I'm thinking", user_id="", timestamp="now"), + SleepFrame(), + VADUserStoppedSpeakingFrame(), + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.1), + # First inference fired here. Imagine the LLM returned ○; + # the turn is not yet finalized, so the user keeps talking. + VADUserStartedSpeakingFrame(), + TranscriptionFrame(text="about pizza", user_id="", timestamp="now"), + SleepFrame(), + VADUserStoppedSpeakingFrame(), + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.1), + # Second inference fired here. Now the LLM returns ✓ and the + # turn finalizes via UserTurnCompletedFrame. + UserTurnCompletedFrame(), + SleepFrame(), + ] + await run_test(pipeline, frames_to_send=frames_to_send) + + self.assertEqual(inference_count, 2) + self.assertIsNotNone(stop_message) + # The public event should report the full transcript, even + # though each inference push only writes its own segment to + # the context. + self.assertEqual(stop_message.content, "I'm thinking about pizza") + + user_messages = [m for m in context.get_messages() if m.get("role") == "user"] + self.assertEqual([m["content"] for m in user_messages], ["I'm thinking", "about pizza"]) + class TestLLMAssistantAggregator(unittest.IsolatedAsyncioTestCase): async def test_empty(self):