diff --git a/engine/pipeline.py b/engine/pipeline.py index d32f2f7..84abca3 100644 --- a/engine/pipeline.py +++ b/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) diff --git a/engine/text_stream.py b/engine/text_stream.py index ea6dcbd..8628c46 100644 --- a/engine/text_stream.py +++ b/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 - ``LLMFullResponseEndFrame`` to drain its audio queue, text reaches the - client well ahead of (or at worst, alongside) the synthesized audio. + Urgent frames bypass TTS serialization and transport audio queues so text + reaches the client at least as quickly as 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,