Enhance text stream processing by adding synchronization for interrupted assistant turns. Introduce sync_streamed_assistant_context to align LLM context with UI text, and modify ProductTextStreamProcessor to handle interrupted text. Update pipeline to utilize the new functionality for managing assistant turn interruptions.

engine/pipeline.py

@@ -36,7 +36,7 @@ from .config import EngineConfig
 from .product_protocol import ProductWebsocketSerializer
 from .services import create_llm_service, create_stt_service, create_tts_service
 from .text_input import ProductTextInputProcessor
-from .text_stream import ProductTextStreamProcessor
+from .text_stream import ProductTextStreamProcessor, sync_streamed_assistant_context
 from .transcript_stream import ProductTranscriptStreamProcessor
 from .turn_start import InterruptionGateUserTurnStartStrategy
 
@@ -142,6 +142,8 @@ async def run_pipeline_with_serializer(
         ),
     )
 
+    text_stream = ProductTextStreamProcessor()
+
     pipeline = Pipeline(
         [
             transport.input(),
@@ -150,7 +152,7 @@ async def run_pipeline_with_serializer(
             ProductTranscriptStreamProcessor(),
             user_aggregator,
             llm,
-            ProductTextStreamProcessor(),
+            text_stream,
             tts,
             transport.output(),
             assistant_aggregator,
@@ -210,6 +212,14 @@ async def run_pipeline_with_serializer(
     @assistant_aggregator.event_handler("on_assistant_turn_stopped")
     async def on_assistant_turn_stopped(_aggregator, message: AssistantTurnStoppedMessage):
         logger.info(f"Assistant: {message.content}")
+        if message.interrupted:
+            streamed = text_stream.take_interrupted_stream_text()
+            if streamed:
+                sync_streamed_assistant_context(
+                    _aggregator,
+                    streamed_text=streamed,
+                    committed_text=message.content or "",
+                )
 
     runner = PipelineRunner(handle_sigint=False)
     await runner.run(task)

engine/text_stream.py

@@ -1,17 +1,57 @@
 from __future__ import annotations
 
+from typing import Any, Protocol
+
 from pipecat.frames.frames import (
+    CancelFrame,
     Frame,
     InterruptionFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
     LLMTextFrame,
-    OutputTransportMessageFrame,
+    OutputTransportMessageUrgentFrame,
     TTSSpeakFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
 
 
+class _AssistantContextSync(Protocol):
+    @property
+    def context(self) -> Any: ...
+
+
+def sync_streamed_assistant_context(
+    aggregator: _AssistantContextSync,
+    *,
+    streamed_text: str,
+    committed_text: str,
+) -> None:
+    """Align LLM context with UI text after an interrupted assistant turn.
+
+    The assistant aggregator only commits TTS-spoken text on interrupt. Replace
+    or append the streamed LLM text so the next turn sees what the user saw.
+    """
+    streamed = streamed_text.strip()
+    if not streamed or streamed == committed_text.strip():
+        return
+
+    committed = committed_text.strip()
+
+    def _apply(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        updated = list(messages)
+        if committed and updated:
+            last = updated[-1]
+            if isinstance(last, dict) and last.get("role") == "assistant":
+                content = last.get("content")
+                if isinstance(content, str) and content.strip() == committed:
+                    updated[-1] = {"role": "assistant", "content": streamed}
+                    return updated
+        updated.append({"role": "assistant", "content": streamed})
+        return updated
+
+    aggregator.context.transform_messages(_apply)
+
+
 class ProductTextStreamProcessor(FrameProcessor):
     """Mirrors LLM text frames as streaming protocol events.
 
@@ -20,9 +60,8 @@ class ProductTextStreamProcessor(FrameProcessor):
     downstream as ``OutputTransportMessageUrgentFrame``s that the product
     serializer turns into ``response.text.{started,delta,final}`` events.
 
-    Because the events are emitted before the TTS holds onto
+    Urgent frames bypass TTS serialization and transport audio queues so text
-    ``LLMFullResponseEndFrame`` to drain its audio queue, text reaches the
+    reaches the client at least as quickly as synthesized audio.
-    client well ahead of (or at worst, alongside) the synthesized audio.
 
     ``TTSSpeakFrame`` (used by the fixed-greeting code path, which bypasses
     the LLM entirely) is also handled: the processor synthesizes a single
@@ -33,6 +72,12 @@ class ProductTextStreamProcessor(FrameProcessor):
         super().__init__()
         self._aggregation: list[str] = []
         self._turn_active = False
+        self._interrupted_stream_text: str | None = None
+
+    def take_interrupted_stream_text(self) -> str | None:
+        text = self._interrupted_stream_text
+        self._interrupted_stream_text = None
+        return text
 
     async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
         await super().process_frame(frame, direction)
@@ -47,7 +92,7 @@ class ProductTextStreamProcessor(FrameProcessor):
         elif isinstance(frame, LLMFullResponseEndFrame):
             await self.push_frame(frame, direction)
             await self._end_turn(interrupted=False)
-        elif isinstance(frame, InterruptionFrame):
+        elif isinstance(frame, (InterruptionFrame, CancelFrame)):
             await self.push_frame(frame, direction)
             await self._end_turn(interrupted=True)
         elif isinstance(frame, TTSSpeakFrame):
@@ -81,6 +126,8 @@ class ProductTextStreamProcessor(FrameProcessor):
         if not self._turn_active:
             return
         full_text = "".join(self._aggregation)
+        if interrupted and full_text:
+            self._interrupted_stream_text = full_text
         self._turn_active = False
         self._aggregation = []
         await self._emit(
@@ -91,7 +138,7 @@ class ProductTextStreamProcessor(FrameProcessor):
 
     async def _emit(self, event_type: str, **payload: object) -> None:
         await self.push_frame(
-            OutputTransportMessageFrame(
+            OutputTransportMessageUrgentFrame(
                 message={"type": event_type, **payload},
             ),
             FrameDirection.DOWNSTREAM,