From 1fe8cf5289b9c4e2b559c944e05440a651f3cfac Mon Sep 17 00:00:00 2001 From: Paul Kompfner Date: Wed, 20 May 2026 15:09:19 -0400 Subject: [PATCH] Add RealtimeServiceModeConfig to LLMContextAggregatorPair MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Decouple context management from turn frames and transcripts when a realtime LLM service drives the conversation. Three problems with today's behavior: - Some realtime services (Gemini Live, AWS Nova Sonic, Ultravox) emit no UserStarted/StoppedSpeakingFrame at all, so the aggregator — which writes user messages on those frames — doesn't write to context correctly without them. - The workaround (local VAD on the aggregator) generates turn boundaries that don't match the provider's server-side ground truth, and the per-service "do I need it?" rule is hard to keep straight. - When local turn detection is the intended driver, turn-end strategies still wait for transcripts on the latency critical path. Add a realtime_service_mode: RealtimeServiceModeConfig | None = None kwarg on LLMContextAggregatorPair. When set, the pair switches both halves to trailing context writes: user messages are flushed on the first assistant content frame, assistant messages on the next user transcript, both halves on EndFrame. Turn-end strategies stop waiting for transcripts by default. Two fine-grained boolean fields (context_writes_await_turns, turns_await_transcripts) let callers dial back to cascade-style behavior selectively; their invalid combination is rejected in __post_init__. The bifurcation is dispatch-only: seven branch points across the two halves, each at method entry, each delegating to a mode-pure private method. Cross-half coordination uses an asyncio.Lock and a back-reference shared by both halves; the assistant signals user.flush() on LLMFullResponseStartFrame, and the user signals assistant.flush() on the first new transcript after the assistant turn. The mechanism reuses the existing push_aggregation() — no parallel write path. Two new events fire when messages are flushed to context: on_user_message_added and on_assistant_message_added. In cascade mode they coincide with the existing turn-stopped events; in realtime mode (where the turn-stopped event fires before the message is finalized) they're the canonical way to subscribe to "context just updated, here's the text." UserTurnStoppedMessage.content is now typed str | None to reflect that realtime mode fires the event with None. When a RealtimeServiceMetadataFrame arrives and realtime_service_mode is None, the aggregator logs a one-time INFO recommendation pointing users at the option. --- .../aggregators/llm_response_universal.py | 527 +++++++++++++++++- tests/test_context_aggregators_universal.py | 317 +++++++++++ 2 files changed, 817 insertions(+), 27 deletions(-) diff --git a/src/pipecat/processors/aggregators/llm_response_universal.py b/src/pipecat/processors/aggregators/llm_response_universal.py index bc158042e..646c7df66 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 ( LLMThoughtEndFrame, LLMThoughtStartFrame, LLMThoughtTextFrame, + RealtimeServiceMetadataFrame, StartFrame, TextFrame, TranscriptionFrame, @@ -83,7 +84,11 @@ from pipecat.processors.aggregators.llm_context_summarizer import ( from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.turns.user_idle_controller import UserIdleController from pipecat.turns.user_mute import BaseUserMuteStrategy -from pipecat.turns.user_start import BaseUserTurnStartStrategy, UserTurnStartedParams +from pipecat.turns.user_start import ( + BaseUserTurnStartStrategy, + TranscriptionUserTurnStartStrategy, + UserTurnStartedParams, +) from pipecat.turns.user_stop import BaseUserTurnStopStrategy, UserTurnStoppedParams from pipecat.turns.user_turn_completion_mixin import UserTurnCompletionConfig from pipecat.turns.user_turn_controller import UserTurnController @@ -258,6 +263,43 @@ class LLMAssistantAggregatorParams: self.context_summarization_config = None +@dataclass +class RealtimeServiceModeConfig: + """Configure an ``LLMContextAggregatorPair`` for use with a realtime LLM service. + + Both fields default to False (the recommended realtime behavior, dropping + transcript-related waits at both points in the flow). Override individual + fields to dial back to cascade-style behavior selectively. + + Parameters: + context_writes_await_turns: When False (default), context writes are + triggered by the content stream itself (transcripts and assistant + text frames), making writes independent of turn-frame availability + and timing. When True, user messages are written to context on + user-turn-end frames (cascade behavior). + turns_await_transcripts: When False (default), turn-end fires as soon + as VAD signals end of speech, avoiding latency on the critical + path when local turn detection drives a realtime conversation. + When True, turn-end strategies wait for transcripts to arrive + before signalling end-of-turn. + """ + + context_writes_await_turns: bool = False + turns_await_transcripts: bool = False + + def __post_init__(self): + """Validate the field combination.""" + if not self.turns_await_transcripts and self.context_writes_await_turns: + raise ValueError( + "Invalid combination: turns fire early (without transcripts) " + "but context writes wait on those turn frames — context would " + "be written with incomplete user messages. Either set " + "turns_await_transcripts=True (preserve transcript-aware " + "turn-end timing) or context_writes_await_turns=False " + "(decouple writes from turn frames)." + ) + + @dataclass class UserTurnStoppedMessage: """A user turn stopped message containing a user transcript update. @@ -266,13 +308,18 @@ class UserTurnStoppedMessage: the aggregated transcript that is then used in the context. Parameters: - content: The message content/text. + content: The message content/text. ``None`` in realtime mode + (``RealtimeServiceModeConfig(context_writes_await_turns=False)``) + when fired from a user-turn-stop frame, since the user message + hasn't been finalized at that point. Subscribers that need the + finalized text should listen to ``on_user_message_added`` + instead. timestamp: When the user turn started. user_id: Optional identifier for the user. """ - content: str + content: str | None timestamp: str user_id: str | None = None @@ -567,6 +614,9 @@ class LLMUserAggregator(LLMContextAggregator): context: LLMContext, *, params: LLMUserAggregatorParams | None = None, + _realtime_service_mode: RealtimeServiceModeConfig | None = None, + _paired_half: "LLMAssistantAggregator | None" = None, + _pair_lock: asyncio.Lock | None = None, **kwargs, ): """Initialize the user context aggregator. @@ -574,6 +624,14 @@ class LLMUserAggregator(LLMContextAggregator): Args: context: The LLM context for conversation storage. params: Configuration parameters for aggregation behavior. + _realtime_service_mode: Pair-internal. Realtime-mode + configuration propagated from + ``LLMContextAggregatorPair``. Not intended for direct use — + construct the aggregators via the pair. + _paired_half: Pair-internal. Back-reference to the paired + assistant aggregator for cross-half coordination. + _pair_lock: Pair-internal. Shared asyncio lock serializing + cross-half flushes. **kwargs: Additional arguments. """ params = params or LLMUserAggregatorParams() @@ -590,9 +648,23 @@ class LLMUserAggregator(LLMContextAggregator): 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_message_added") self._register_event_handler("on_user_mute_started") self._register_event_handler("on_user_mute_stopped") + # Realtime-mode wiring. Defaults (no config) preserve cascade + # behavior: context writes happen on turn frames, turns wait + # for transcripts. + self._realtime_service_mode = _realtime_service_mode + self._paired_half = _paired_half + self._pair_lock = _pair_lock + if _realtime_service_mode is not None: + self._context_writes_await_turns = _realtime_service_mode.context_writes_await_turns + self._turns_await_transcripts = _realtime_service_mode.turns_await_transcripts + else: + self._context_writes_await_turns = True + self._turns_await_transcripts = True + user_turn_strategies = self._params.user_turn_strategies or UserTurnStrategies() # Deprecated path: translate filter_incomplete_user_turns into @@ -606,8 +678,19 @@ class LLMUserAggregator(LLMContextAggregator): ) self._params.user_turn_strategies = user_turn_strategies + # Realtime mutation: when turns shouldn't wait for transcripts, + # drop the transcription-based start strategy and flip the + # wait_for_transcript flag on stop strategies that expose it. The + # set of strategies that support it intentionally stays narrow — + # the flag was reintroduced specifically for this realtime path. + if not self._turns_await_transcripts: + self._apply_realtime_strategy_mutations(user_turn_strategies) + self._user_is_muted = False self._user_turn_start_timestamp = "" + # Tracks whether the §3.6 recommendation log has already fired + # for this session — see _handle_realtime_service_metadata. + self._realtime_recommendation_logged = False # 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 @@ -717,6 +800,9 @@ class LLMUserAggregator(LLMContextAggregator): await self.push_frame(frame, direction) elif isinstance(frame, LLMSetToolChoiceFrame): self.set_tool_choice(frame.tool_choice) + elif isinstance(frame, RealtimeServiceMetadataFrame): + await self._handle_realtime_service_metadata(frame) + await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) @@ -734,9 +820,16 @@ class LLMUserAggregator(LLMContextAggregator): self._context.add_message({"role": self.role, "content": aggregation}) await self.push_context_frame() + message = UserTurnStoppedMessage( + content=aggregation, timestamp=self._user_turn_start_timestamp + ) + await self._call_event_handler("on_user_message_added", message) + return aggregation async def _start(self, frame: StartFrame): + self._validate_realtime_pairing() + if self._vad_controller: await self._vad_controller.setup(self.task_manager) @@ -748,13 +841,138 @@ 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) + if not self._context_writes_await_turns: + # Realtime: flush trailing user content directly. The + # on_user_turn_stopped event already fired (if turn frames + # were emitted), so don't re-fire it from session end. + await self.push_aggregation() + else: + await self._maybe_emit_user_turn_stopped(on_session_end=True) await self._cleanup() async def _cancel(self, frame: CancelFrame): - await self._maybe_emit_user_turn_stopped(on_session_end=True) + if not self._context_writes_await_turns: + await self.push_aggregation() + else: + await self._maybe_emit_user_turn_stopped(on_session_end=True) await self._cleanup() + def _validate_realtime_pairing(self): + """Validate the realtime-mode wiring set by ``LLMContextAggregatorPair``. + + Realtime mode requires both halves to be paired through the + ``LLMContextAggregatorPair`` so cross-half flushes can find each + other. Direct construction of a half with the private realtime + kwargs is not supported. + """ + if not self._context_writes_await_turns: + if self._paired_half is None: + raise RuntimeError( + f"{self}: realtime_service_mode is configured but this user " + "aggregator has no paired assistant aggregator. Construct " + "the pair via LLMContextAggregatorPair(" + "context, realtime_service_mode=RealtimeServiceModeConfig())." + ) + if self._paired_half is not None: + if ( + self._context_writes_await_turns != self._paired_half._context_writes_await_turns + or self._turns_await_transcripts != self._paired_half._turns_await_transcripts + ): + raise RuntimeError( + f"{self}: realtime-mode config mismatch between user and " + "assistant halves. Use LLMContextAggregatorPair to construct " + "the pair so both halves share the same configuration." + ) + + def _apply_realtime_strategy_mutations(self, user_turn_strategies: UserTurnStrategies) -> None: + """Mutate turn strategies for the realtime ``turns_await_transcripts=False`` path. + + Drops ``TranscriptionUserTurnStartStrategy`` from the start strategies + (transcripts shouldn't start a turn when the realtime service drives + the conversation) and flips ``wait_for_transcript=False`` on stop + strategies that expose the flag, so end-of-turn fires as soon as VAD / + the turn analyzer reports end-of-speech. + """ + custom_strategies = self._params.user_turn_strategies is not None + + start_strategies = user_turn_strategies.start or [] + dropped: list[str] = [] + kept_start: list[BaseUserTurnStartStrategy] = [] + for s in start_strategies: + if isinstance(s, TranscriptionUserTurnStartStrategy): + dropped.append(s.__class__.__name__) + else: + kept_start.append(s) + user_turn_strategies.start = kept_start + + flipped: list[str] = [] + for s in user_turn_strategies.stop or []: + if hasattr(s, "wait_for_transcript"): + try: + s.wait_for_transcript = False + flipped.append(s.__class__.__name__) + except AttributeError: + # Strategy exposes the property but no setter — skip. + pass + + if not dropped and not flipped: + return + + msg = ( + f"{self}: realtime_service_mode(turns_await_transcripts=False) — " + f"dropped {dropped or 'no'} start strategy(ies); set " + f"wait_for_transcript=False on {flipped or 'no'} stop strategy(ies)." + ) + if custom_strategies: + logger.warning(msg) + else: + logger.debug(msg) + + async def _handle_realtime_service_metadata(self, frame: RealtimeServiceMetadataFrame): + """Handle a ``RealtimeServiceMetadataFrame`` broadcast by a realtime LLM service. + + When ``realtime_service_mode`` isn't configured, log a one-time INFO + recommendation pointing the user at the option and warning about the + timing change on ``on_user_turn_stopped``. When it is configured, log + a confirming debug message. Fires at most once per session. + """ + if self._realtime_recommendation_logged: + return + self._realtime_recommendation_logged = True + + if self._realtime_service_mode is None: + logger.info( + f"{self}: detected realtime service `{frame.service_name}` in the " + "pipeline. For correct context-write semantics with realtime " + "services, consider passing " + "realtime_service_mode=RealtimeServiceModeConfig() to " + "LLMContextAggregatorPair. Note: this changes when user messages " + "are written to context — they're written when the assistant " + "response starts rather than when the user-turn-end frame fires. " + "Subscribe to `on_user_message_added` instead of " + "`on_user_turn_stopped` if you need post-write semantics." + ) + else: + logger.debug( + f"{self}: detected realtime service `{frame.service_name}`; " + "realtime_service_mode is configured." + ) + + async def _realtime_handoff_flush(self) -> None: + """Flush pending user aggregation to context. + + Called by the paired assistant half from + ``_realtime_handle_llm_start`` (i.e. on ``LLMFullResponseStartFrame``) + to commit the in-flight user message before the assistant starts + its own turn. No-op when there's no pending content. + """ + if not self._aggregation: + return + # push_aggregation writes the message to context, pushes + # LLMContextFrame, and emits on_user_message_added. + await self.push_aggregation() + self._user_turn_start_timestamp = "" + async def _cleanup(self): if self._vad_controller: await self._vad_controller.cleanup() @@ -826,6 +1044,10 @@ class LLMUserAggregator(LLMContextAggregator): await self.push_context_frame() async def _handle_transcription(self, frame: TranscriptionFrame): + if not self._context_writes_await_turns: + await self._realtime_handle_transcription(frame) + return + text = frame.text # Make sure we really have some text. @@ -839,6 +1061,30 @@ class LLMUserAggregator(LLMContextAggregator): ) ) + async def _realtime_handle_transcription(self, frame: TranscriptionFrame): + """Realtime variant: signal the paired assistant half to flush, then append. + + The first new user transcript after an assistant turn ends is what + commits the assistant's pending message to context. The flush is + idempotent (no-op when nothing pending), so it's safe to call on + every chunk. + """ + if not frame.text.strip(): + return + + if self._paired_half is not None and self._pair_lock is not None: + async with self._pair_lock: + await self._paired_half._realtime_handoff_flush() + + if not self._user_turn_start_timestamp: + self._user_turn_start_timestamp = time_now_iso8601() + + self._aggregation.append( + TextPartForConcatenation( + frame.text, includes_inter_part_spaces=frame.includes_inter_frame_spaces + ) + ) + async def _queued_broadcast_frame(self, frame_cls: type[Frame], **kwargs): """Broadcasts a frame upstream and queues it for internal processing. @@ -903,6 +1149,17 @@ class LLMUserAggregator(LLMContextAggregator): controller: UserTurnController, strategy: BaseUserTurnStopStrategy, ): + if not self._context_writes_await_turns: + # Realtime: turn frames are supplemental — they don't drive + # context writes. Fire the event without pushing aggregation; + # the trailing-write path commits the user message instead. + logger.debug( + f"{self}: User turn inference triggered (strategy: {strategy}) " + "[realtime: event-only, no context push]" + ) + await self._call_event_handler("on_user_turn_inference_triggered", strategy) + return + logger.debug(f"{self}: User turn inference triggered (strategy: {strategy})") # Push aggregation now: this writes the user message segment to @@ -935,6 +1192,17 @@ class LLMUserAggregator(LLMContextAggregator): await self._user_idle_controller.process_frame(UserStoppedSpeakingFrame()) + if not self._context_writes_await_turns: + # Realtime: turn frames are supplemental. The user message + # isn't finalized at turn-stop time — content is None. + # Subscribers wanting the finalized text use + # on_user_message_added instead. + message = UserTurnStoppedMessage( + content=None, timestamp=self._user_turn_start_timestamp + ) + await self._call_event_handler("on_user_turn_stopped", strategy, message) + return + await self._maybe_emit_user_turn_stopped(strategy) async def _on_reset_aggregation( @@ -1030,6 +1298,9 @@ class LLMAssistantAggregator(LLMContextAggregator): context: LLMContext, *, params: LLMAssistantAggregatorParams | None = None, + _realtime_service_mode: RealtimeServiceModeConfig | None = None, + _paired_half: "LLMUserAggregator | None" = None, + _pair_lock: asyncio.Lock | None = None, **kwargs, ): """Initialize the assistant context aggregator. @@ -1037,6 +1308,14 @@ class LLMAssistantAggregator(LLMContextAggregator): Args: context: The OpenAI LLM context for conversation storage. params: Configuration parameters for aggregation behavior. + _realtime_service_mode: Pair-internal. Realtime-mode + configuration propagated from + ``LLMContextAggregatorPair``. Not intended for direct use — + construct the aggregators via the pair. + _paired_half: Pair-internal. Back-reference to the paired + user aggregator for cross-half coordination. + _pair_lock: Pair-internal. Shared asyncio lock serializing + cross-half flushes. **kwargs: Additional arguments. """ params = params or LLMAssistantAggregatorParams() @@ -1048,6 +1327,24 @@ class LLMAssistantAggregator(LLMContextAggregator): ) self._params = params + # Realtime-mode wiring. Defaults (no config) preserve cascade + # behavior: write to context on LLMFullResponseEndFrame. + self._realtime_service_mode = _realtime_service_mode + self._paired_half = _paired_half + self._pair_lock = _pair_lock + if _realtime_service_mode is not None: + self._context_writes_await_turns = _realtime_service_mode.context_writes_await_turns + self._turns_await_transcripts = _realtime_service_mode.turns_await_transcripts + else: + self._context_writes_await_turns = True + self._turns_await_transcripts = True + + # Realtime mode only. Holds the assistant turn's content between + # LLMFullResponseEndFrame (the moment we mark it ready to flush) + # and the next user transcript (the moment we actually write it + # to context). + self._pending_assistant_message_to_flush: dict | None = None + self._function_calls_in_progress: dict[str, FunctionCallInProgressFrame | None] = {} self._function_calls_image_results: dict[str, UserImageRawFrame] = {} self._context_updated_tasks: set[asyncio.Task] = set() @@ -1084,6 +1381,7 @@ class LLMAssistantAggregator(LLMContextAggregator): self._register_event_handler("on_assistant_turn_started") self._register_event_handler("on_assistant_turn_stopped") + self._register_event_handler("on_assistant_message_added") self._register_event_handler("on_assistant_thought") self._register_event_handler("on_summary_applied") @@ -1184,6 +1482,10 @@ class LLMAssistantAggregator(LLMContextAggregator): if self._push_context_on_bot_stopped_speaking and not self._user_speaking: logger.debug(f"{self}: Bot stopped speaking — pushing deferred context frame!") await self.push_context_frame(FrameDirection.UPSTREAM) + elif isinstance(frame, RealtimeServiceMetadataFrame): + # The user half logs the §3.6 recommendation; the assistant + # half just passes the frame through. + await self.push_frame(frame, direction) else: await self.push_frame(frame, direction) @@ -1192,9 +1494,37 @@ class LLMAssistantAggregator(LLMContextAggregator): await self._summarizer.process_frame(frame) async def _start(self, frame: StartFrame): + self._validate_realtime_pairing() if self._summarizer: await self._summarizer.setup(self.task_manager) + def _validate_realtime_pairing(self): + """Validate the realtime-mode wiring set by ``LLMContextAggregatorPair``. + + Realtime mode requires both halves to be paired through the + ``LLMContextAggregatorPair`` so cross-half flushes can find each + other. Direct construction of a half with the private realtime + kwargs is not supported. + """ + if not self._context_writes_await_turns: + if self._paired_half is None: + raise RuntimeError( + f"{self}: realtime_service_mode is configured but this assistant " + "aggregator has no paired user aggregator. Construct the pair " + "via LLMContextAggregatorPair(" + "context, realtime_service_mode=RealtimeServiceModeConfig())." + ) + if self._paired_half is not None: + if ( + self._context_writes_await_turns != self._paired_half._context_writes_await_turns + or self._turns_await_transcripts != self._paired_half._turns_await_transcripts + ): + raise RuntimeError( + f"{self}: realtime-mode config mismatch between user and " + "assistant halves. Use LLMContextAggregatorPair to construct " + "the pair so both halves share the same configuration." + ) + async def push_aggregation(self) -> str: """Push the current assistant aggregation with timestamp.""" if not self._aggregation: @@ -1247,6 +1577,12 @@ class LLMAssistantAggregator(LLMContextAggregator): async def _handle_end_or_cancel(self, frame: Frame): await self._trigger_assistant_turn_stopped(interrupted=isinstance(frame, CancelFrame)) + if not self._context_writes_await_turns: + # Flush any pending assistant content parked by + # _realtime_trigger_assistant_turn_stopped (i.e. the bot + # finished its last reply but no follow-up user transcript + # arrived before the session ended). + await self._realtime_handoff_flush() if self._summarizer: await self._summarizer.cleanup() @@ -1349,26 +1685,7 @@ class LLMAssistantAggregator(LLMContextAggregator): run_llm = True if run_llm and not self._user_speaking: - if self.has_queued_frame(FunctionCallResultFrame): - # Another FunctionCallResultFrame is already queued. Defer the context push - # to bundle all results into a single LLM call instead of triggering one - # inference pass per result. The context will be pushed once the last - # function call in the queue is processed. - logger.debug( - f"{self}: More FunctionCallResultFrames queued — deferring context frame push." - ) - elif self._bot_speaking: - # Defer the context frame push until the bot finishes speaking. If multiple - # function call results arrive while the bot is speaking, they all accumulate - # in the context and a single push is performed once speaking stops, preventing - # the LLM from running multiple times and producing duplicated responses. - # This should be an edge case, since it would require a FunctionCallResultFrame - # being queued between an LLM response start and end frame. - logger.debug(f"{self}: Bot is speaking — deferring context frame push.") - self._push_context_on_bot_stopped_speaking = True - else: - logger.debug(f"{self}: Pushing context frame!") - await self.push_context_frame(FrameDirection.UPSTREAM) + await self._maybe_push_context_after_function_result() # Call the `on_context_updated` callback once the function call result # is added to the context. Also, run this in a separate task to make @@ -1379,6 +1696,42 @@ class LLMAssistantAggregator(LLMContextAggregator): self._context_updated_tasks.add(task) task.add_done_callback(self._context_updated_task_finished) + async def _maybe_push_context_after_function_result(self) -> None: + """Decide whether to push a context frame after a function-call result. + + Dispatched by mode. Cascade re-runs LLM inference by pushing an + ``LLMContextFrame`` upstream (with care to avoid duplicate pushes + while results are queued or the bot is still speaking). Realtime + services consume function results directly via + ``FunctionCallResultFrame``, so the context-driven re-inference + cycle is unnecessary. + """ + if not self._context_writes_await_turns: + # Realtime: the realtime service has the result via + # FunctionCallResultFrame. No context push needed. + return + + if self.has_queued_frame(FunctionCallResultFrame): + # Another FunctionCallResultFrame is already queued. Defer the context push + # to bundle all results into a single LLM call instead of triggering one + # inference pass per result. The context will be pushed once the last + # function call in the queue is processed. + logger.debug( + f"{self}: More FunctionCallResultFrames queued — deferring context frame push." + ) + elif self._bot_speaking: + # Defer the context frame push until the bot finishes speaking. If multiple + # function call results arrive while the bot is speaking, they all accumulate + # in the context and a single push is performed once speaking stops, preventing + # the LLM from running multiple times and producing duplicated responses. + # This should be an edge case, since it would require a FunctionCallResultFrame + # being queued between an LLM response start and end frame. + logger.debug(f"{self}: Bot is speaking — deferring context frame push.") + self._push_context_on_bot_stopped_speaking = True + else: + logger.debug(f"{self}: Pushing context frame!") + await self.push_context_frame(FrameDirection.UPSTREAM) + async def _handle_function_call_intermediate_result( self, frame: FunctionCallResultFrame, in_progress_frame: FunctionCallInProgressFrame ): @@ -1469,6 +1822,20 @@ class LLMAssistantAggregator(LLMContextAggregator): ) async def _handle_llm_start(self, _: LLMFullResponseStartFrame): + if not self._context_writes_await_turns: + await self._realtime_handle_llm_start() + return + await self._trigger_assistant_turn_started() + + async def _realtime_handle_llm_start(self): + """Realtime: flush the paired user half before starting the assistant turn. + + The first content frame of an assistant turn is the trigger to + commit any in-flight user transcript to context. + """ + if self._paired_half is not None and self._pair_lock is not None: + async with self._pair_lock: + await self._paired_half._realtime_handoff_flush() await self._trigger_assistant_turn_started() async def _handle_llm_end(self, _: LLMFullResponseEndFrame): @@ -1606,6 +1973,10 @@ class LLMAssistantAggregator(LLMContextAggregator): await self._call_event_handler("on_assistant_turn_started") async def _trigger_assistant_turn_stopped(self, *, interrupted: bool = False): + if not self._context_writes_await_turns: + await self._realtime_trigger_assistant_turn_stopped(interrupted=interrupted) + return + if not self._assistant_turn_start_timestamp: return @@ -1620,9 +1991,86 @@ class LLMAssistantAggregator(LLMContextAggregator): timestamp=self._assistant_turn_start_timestamp, ) await self._call_event_handler("on_assistant_turn_stopped", message) + if aggregation: + await self._call_event_handler("on_assistant_message_added", message) self._assistant_turn_start_timestamp = "" + async def _realtime_trigger_assistant_turn_stopped(self, *, interrupted: bool): + """Realtime variant: defer the context write or flush on interruption. + + Normal end-of-turn (``interrupted=False``, from + ``LLMFullResponseEndFrame``) parks the message text in a pending + slot — it isn't written to context until the next user transcript + arrives or the session ends. Interruption (``interrupted=True``) + commits immediately, matching today's + ``AssistantTurnStoppedMessage.interrupted`` semantics. + """ + if not self._assistant_turn_start_timestamp: + return + + timestamp = self._assistant_turn_start_timestamp + self._assistant_turn_start_timestamp = "" + + if interrupted: + aggregation = await self.push_aggregation() + if aggregation: + aggregation = self._maybe_strip_turn_completion_markers(aggregation) + message = AssistantTurnStoppedMessage( + content=aggregation, interrupted=True, timestamp=timestamp + ) + await self._call_event_handler("on_assistant_turn_stopped", message) + if aggregation: + await self._call_event_handler("on_assistant_message_added", message) + return + + # Normal end. Park the message for trailing write. + raw_aggregation = self.aggregation_string() + if raw_aggregation: + self._pending_assistant_message_to_flush = { + "raw": raw_aggregation, + "timestamp": timestamp, + } + await self.reset() + stripped = ( + self._maybe_strip_turn_completion_markers(raw_aggregation) if raw_aggregation else "" + ) + message = AssistantTurnStoppedMessage( + content=stripped, interrupted=False, timestamp=timestamp + ) + await self._call_event_handler("on_assistant_turn_stopped", message) + + async def _realtime_handoff_flush(self) -> None: + """Flush pending assistant aggregation to context. + + Called by the paired user half from + ``_realtime_handle_transcription`` when a new transcript arrives, + committing the assistant's deferred message before the user + starts a new turn. No-op when nothing is pending. + """ + if self._pending_assistant_message_to_flush is None: + return + pending = self._pending_assistant_message_to_flush + self._pending_assistant_message_to_flush = None + + raw = pending["raw"] + timestamp = pending["timestamp"] + + # Mirror push_aggregation: write the raw aggregation (with any + # turn-completion markers intact) to context, emit LLMContextFrame + # and the timestamp frame. Markers are stripped only from the + # event-carried text. + self._context.add_message({"role": "assistant", "content": raw}) + await self.push_context_frame() + timestamp_frame = LLMContextAssistantTimestampFrame(timestamp=time_now_iso8601()) + await self.push_frame(timestamp_frame) + + stripped = self._maybe_strip_turn_completion_markers(raw) + message = AssistantTurnStoppedMessage( + content=stripped, interrupted=False, timestamp=timestamp + ) + await self._call_event_handler("on_assistant_message_added", message) + def _maybe_strip_turn_completion_markers(self, text: str) -> str: """Strip turn completion markers from assistant transcript. @@ -1685,6 +2133,7 @@ class LLMContextAggregatorPair: user_params: LLMUserAggregatorParams | None = None, assistant_params: LLMAssistantAggregatorParams | None = None, add_tool_change_messages: bool | None = None, + realtime_service_mode: RealtimeServiceModeConfig | None = None, ): """Initialize the LLM context aggregator pair. @@ -1702,14 +2151,38 @@ class LLMContextAggregatorPair: announcement is added exactly once (the second aggregator's diff is empty by the time it sees the frame). Leave as ``None`` to respect per-params settings. + realtime_service_mode: When provided, configures the pair for + use with a realtime (speech-to-speech) LLM service. + Context writes become trailing — driven by the content + stream itself (transcripts, ``LLMFullResponseStartFrame``) + rather than turn frames — and, by default, turn-end + strategies stop waiting for transcripts. Both halves share + this configuration via a private channel; mismatched + halves are rejected at ``StartFrame``. Defaults to + ``None``, which preserves cascade behavior. """ user_params = user_params or LLMUserAggregatorParams() assistant_params = assistant_params or LLMAssistantAggregatorParams() if add_tool_change_messages is not None: user_params.add_tool_change_messages = add_tool_change_messages assistant_params.add_tool_change_messages = add_tool_change_messages - self._user = LLMUserAggregator(context, params=user_params) - self._assistant = LLMAssistantAggregator(context, params=assistant_params) + + pair_lock = asyncio.Lock() if realtime_service_mode is not None else None + self._user = LLMUserAggregator( + context, + params=user_params, + _realtime_service_mode=realtime_service_mode, + _pair_lock=pair_lock, + ) + self._assistant = LLMAssistantAggregator( + context, + params=assistant_params, + _realtime_service_mode=realtime_service_mode, + _pair_lock=pair_lock, + ) + # Wire the cross-half back-references after both halves exist. + self._user._paired_half = self._assistant + self._assistant._paired_half = self._user def user(self) -> LLMUserAggregator: """Get the user context aggregator. diff --git a/tests/test_context_aggregators_universal.py b/tests/test_context_aggregators_universal.py index a8ecace9d..513de3d88 100644 --- a/tests/test_context_aggregators_universal.py +++ b/tests/test_context_aggregators_universal.py @@ -4,6 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import asyncio import json import unittest @@ -33,6 +34,7 @@ from pipecat.frames.frames import ( LLMThoughtEndFrame, LLMThoughtStartFrame, LLMThoughtTextFrame, + RealtimeServiceMetadataFrame, SpeechControlParamsFrame, StartFrame, TextFrame, @@ -55,6 +57,8 @@ from pipecat.processors.aggregators.llm_response_universal import ( LLMContextAggregatorPair, LLMUserAggregator, LLMUserAggregatorParams, + RealtimeServiceModeConfig, + UserTurnStoppedMessage, ) from pipecat.processors.frame_processor import FrameDirection from pipecat.tests.utils import SleepFrame, run_test @@ -63,6 +67,10 @@ from pipecat.turns.user_mute import ( FunctionCallUserMuteStrategy, MuteUntilFirstBotCompleteUserMuteStrategy, ) +from pipecat.turns.user_start import ( + TranscriptionUserTurnStartStrategy, + VADUserTurnStartStrategy, +) from pipecat.turns.user_stop import SpeechTimeoutUserTurnStopStrategy from pipecat.turns.user_turn_strategies import ( FilterIncompleteUserTurnStrategies, @@ -1651,5 +1659,314 @@ class TestToolChangeMessages(unittest.IsolatedAsyncioTestCase): self.assertFalse(pair.assistant()._add_tool_change_messages) +class TestRealtimeServiceModeConfig(unittest.TestCase): + def test_default_fields_are_realtime(self): + cfg = RealtimeServiceModeConfig() + self.assertFalse(cfg.context_writes_await_turns) + self.assertFalse(cfg.turns_await_transcripts) + + def test_keep_transcripts_keep_writes_on_turn(self): + cfg = RealtimeServiceModeConfig( + turns_await_transcripts=True, context_writes_await_turns=True + ) + self.assertTrue(cfg.context_writes_await_turns) + self.assertTrue(cfg.turns_await_transcripts) + + def test_keep_transcripts_trailing_writes(self): + # Valid third row: turns wait on transcripts but context writes + # are trailing. The plan calls this out as the explicit fine-grained + # case (downstream consumers of user-turn frames want transcripts). + cfg = RealtimeServiceModeConfig(turns_await_transcripts=True) + self.assertFalse(cfg.context_writes_await_turns) + self.assertTrue(cfg.turns_await_transcripts) + + def test_invalid_combination_rejected(self): + # turns fire early but context writes wait → incomplete messages. + with self.assertRaises(ValueError): + RealtimeServiceModeConfig( + turns_await_transcripts=False, context_writes_await_turns=True + ) + + +class TestRealtimeServiceModeAggregator(unittest.IsolatedAsyncioTestCase): + """End-to-end tests for the trailing-write realtime mode.""" + + def _build_pair( + self, + *, + realtime_service_mode: RealtimeServiceModeConfig | None = None, + user_params: LLMUserAggregatorParams | None = None, + ) -> tuple[LLMContext, LLMContextAggregatorPair]: + context = LLMContext() + pair = LLMContextAggregatorPair( + context, + user_params=user_params, + realtime_service_mode=realtime_service_mode, + ) + return context, pair + + async def test_pair_propagates_realtime_mode_to_halves(self): + _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig()) + # The pair wires shared state into both halves. + self.assertIs(pair.user()._paired_half, pair.assistant()) + self.assertIs(pair.assistant()._paired_half, pair.user()) + self.assertIs(pair.user()._pair_lock, pair.assistant()._pair_lock) + self.assertFalse(pair.user()._context_writes_await_turns) + self.assertFalse(pair.user()._turns_await_transcripts) + self.assertFalse(pair.assistant()._context_writes_await_turns) + self.assertFalse(pair.assistant()._turns_await_transcripts) + + async def test_pair_omits_realtime_wiring_when_unset(self): + _, pair = self._build_pair() + # Backreferences are still created (harmless), but no shared lock + # is allocated when the realtime config is absent. + self.assertIsNone(pair.user()._pair_lock) + self.assertIsNone(pair.assistant()._pair_lock) + self.assertTrue(pair.user()._context_writes_await_turns) + self.assertTrue(pair.assistant()._context_writes_await_turns) + + async def test_realtime_strategy_mutations_with_defaults(self): + _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig()) + # The mutated strategies live on the UserTurnController owned by + # the user aggregator. + strategies = pair.user()._user_turn_controller._user_turn_strategies + # TranscriptionUserTurnStartStrategy is dropped. + for s in strategies.start: + self.assertNotIsInstance(s, TranscriptionUserTurnStartStrategy) + # VAD start strategy is preserved. + self.assertTrue(any(isinstance(s, VADUserTurnStartStrategy) for s in strategies.start)) + # Stop strategies that expose wait_for_transcript have it flipped. + for s in strategies.stop: + if hasattr(s, "wait_for_transcript"): + self.assertFalse(s.wait_for_transcript) + + async def test_realtime_strategy_mutations_skipped_when_turns_await_transcripts(self): + _, pair = self._build_pair( + realtime_service_mode=RealtimeServiceModeConfig(turns_await_transcripts=True), + ) + strategies = pair.user()._user_turn_controller._user_turn_strategies + # When turns still wait for transcripts, the transcript start + # strategy stays in the chain. + self.assertTrue( + any(isinstance(s, TranscriptionUserTurnStartStrategy) for s in strategies.start) + ) + + async def test_trailing_write_user_then_assistant_then_user(self): + _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig()) + user, assistant = pair + + user_msg_added: list[UserTurnStoppedMessage] = [] + assistant_msg_added: list[AssistantTurnStoppedMessage] = [] + + @user.event_handler("on_user_message_added") + async def _on_um(_a, msg): + user_msg_added.append(msg) + + @assistant.event_handler("on_assistant_message_added") + async def _on_am(_a, msg): + assistant_msg_added.append(msg) + + context = user.context + + # Sequence: user transcript, assistant response starts (flushes + # user), assistant response ends (parks pending), new user + # transcript (flushes assistant), then EndFrame flushes the new + # user message. + frames_to_send = [ + TranscriptionFrame(text="Hello!", user_id="", timestamp="now"), + SleepFrame(), + LLMFullResponseStartFrame(), + LLMTextFrame("Hi "), + LLMTextFrame("there!"), + LLMFullResponseEndFrame(), + SleepFrame(), + TranscriptionFrame(text="How are you?", user_id="", timestamp="now"), + SleepFrame(), + ] + await run_test( + Pipeline([user, assistant]), + frames_to_send=frames_to_send, + ) + + # Context should contain: user("Hello!"), assistant("Hi there!"), + # user("How are you?"). + messages = context.get_messages() + roles_contents = [(m["role"], m["content"]) for m in messages] + self.assertEqual( + roles_contents, + [ + ("user", "Hello!"), + ("assistant", "Hi there!"), + ("user", "How are you?"), + ], + ) + self.assertEqual([m.content for m in user_msg_added], ["Hello!", "How are you?"]) + self.assertEqual([m.content for m in assistant_msg_added], ["Hi there!"]) + for msg in assistant_msg_added: + self.assertFalse(msg.interrupted) + + async def test_interruption_writes_assistant_immediately(self): + _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig()) + user, assistant = pair + + assistant_messages: list[AssistantTurnStoppedMessage] = [] + + @assistant.event_handler("on_assistant_message_added") + async def _on_am(_a, msg): + assistant_messages.append(msg) + + context = user.context + + frames_to_send = [ + TranscriptionFrame(text="Hi!", user_id="", timestamp="now"), + LLMFullResponseStartFrame(), + LLMTextFrame("Hello "), + SleepFrame(), + InterruptionFrame(), + ] + await run_test( + Pipeline([user, assistant]), + frames_to_send=frames_to_send, + ) + + roles_contents = [(m["role"], m["content"]) for m in context.get_messages()] + # User message written when assistant started; assistant message + # written immediately on interruption with interrupted=True. + self.assertEqual(roles_contents, [("user", "Hi!"), ("assistant", "Hello")]) + self.assertEqual(len(assistant_messages), 1) + self.assertTrue(assistant_messages[0].interrupted) + + async def test_user_turn_stopped_in_realtime_mode_has_none_content(self): + # When VAD turn frames fire in realtime mode, the user-turn-stop + # message carries content=None — the message isn't finalized yet. + _, pair = self._build_pair( + realtime_service_mode=RealtimeServiceModeConfig(), + user_params=LLMUserAggregatorParams( + user_turn_strategies=UserTurnStrategies( + stop=[ + SpeechTimeoutUserTurnStopStrategy( + user_speech_timeout=TRANSCRIPTION_TIMEOUT, + ) + ], + ), + user_turn_stop_timeout=USER_TURN_STOP_TIMEOUT, + ), + ) + user, assistant = pair + + stop_messages: list[UserTurnStoppedMessage] = [] + + @user.event_handler("on_user_turn_stopped") + async def _on_stop(_a, _s, msg): + stop_messages.append(msg) + + frames_to_send = [ + VADUserStartedSpeakingFrame(), + TranscriptionFrame(text="hey", user_id="", timestamp="now"), + VADUserStoppedSpeakingFrame(), + SleepFrame(sleep=TRANSCRIPTION_TIMEOUT + 0.05), + ] + await run_test( + Pipeline([user, assistant]), + frames_to_send=frames_to_send, + ) + self.assertEqual(len(stop_messages), 1) + self.assertIsNone(stop_messages[0].content) + + async def test_realtime_metadata_recommendation_log_when_unconfigured(self): + # Cascade pair receiving a RealtimeServiceMetadataFrame logs the + # one-time recommendation. The user half records the fact via + # _realtime_recommendation_logged. + _, pair = self._build_pair() + user = pair.user() + + frames_to_send = [ + RealtimeServiceMetadataFrame( + service_name="FakeRealtimeLLM", emits_user_turn_frames=False + ), + ] + await run_test(Pipeline([pair.user(), pair.assistant()]), frames_to_send=frames_to_send) + self.assertTrue(user._realtime_recommendation_logged) + + async def test_realtime_metadata_no_log_when_configured(self): + # When realtime mode is opted in, the metadata frame is consumed + # without firing the recommendation log (we still flag the + # one-shot bookkeeping). + _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig()) + user = pair.user() + + frames_to_send = [ + RealtimeServiceMetadataFrame( + service_name="FakeRealtimeLLM", emits_user_turn_frames=False + ), + ] + await run_test(Pipeline([pair.user(), pair.assistant()]), frames_to_send=frames_to_send) + self.assertTrue(user._realtime_recommendation_logged) + + async def test_realtime_mode_requires_paired_half(self): + # Direct construction of a half with realtime mode set but no + # paired_half raises at StartFrame validation. We call the + # validation directly so the error isn't swallowed by the + # pipeline's exception handler. + context = LLMContext() + cfg = RealtimeServiceModeConfig() + user = LLMUserAggregator(context, _realtime_service_mode=cfg) + with self.assertRaises(RuntimeError): + user._validate_realtime_pairing() + assistant = LLMAssistantAggregator(context, _realtime_service_mode=cfg) + with self.assertRaises(RuntimeError): + assistant._validate_realtime_pairing() + + async def test_realtime_mode_rejects_mismatched_halves(self): + # If a user code path constructs halves with mismatched configs, + # StartFrame validation catches it. + context = LLMContext() + lock = asyncio.Lock() + user = LLMUserAggregator( + context, + _realtime_service_mode=RealtimeServiceModeConfig(), + _pair_lock=lock, + ) + assistant = LLMAssistantAggregator( + context, + _realtime_service_mode=RealtimeServiceModeConfig(turns_await_transcripts=True), + _pair_lock=lock, + ) + user._paired_half = assistant + assistant._paired_half = user + with self.assertRaises(RuntimeError): + user._validate_realtime_pairing() + + async def test_function_call_no_context_push_in_realtime_mode(self): + # Realtime services consume function results directly via + # FunctionCallResultFrame, so the aggregator should not push + # LLMContextFrame upstream after a function call result. + _, pair = self._build_pair(realtime_service_mode=RealtimeServiceModeConfig()) + assistant = pair.assistant() + frames_to_send = [ + FunctionCallInProgressFrame( + function_name="get_weather", + tool_call_id="1", + arguments={"location": "Los Angeles"}, + cancel_on_interruption=True, + ), + SleepFrame(), + FunctionCallResultFrame( + function_name="get_weather", + tool_call_id="1", + arguments={"location": "Los Angeles"}, + result={"conditions": "Sunny"}, + ), + SleepFrame(), + ] + _, up_frames = await run_test( + assistant, + frames_to_send=frames_to_send, + ) + # No LLMContextFrame should have been pushed upstream in + # realtime mode (cascade would push one to re-run inference). + self.assertFalse(any(isinstance(f, LLMContextFrame) for f in up_frames)) + + if __name__ == "__main__": unittest.main()