Add configuration files for FastGPT and Xfyun voice services, enhancing LLM service capabilities. Update LLMConfig to include chat_id, variables, detail, and timeout settings. Refactor create_llm_service to support FastGPT integration and adjust pipeline to handle chat_id and greeting prompts. Implement context synchronization for interrupted assistant turns in text streaming.

src/voice/text_stream.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+from typing import Any, Protocol
+
 from pipecat.frames.frames import (
+    CancelFrame,
     Frame,
     InterruptionFrame,
     LLMFullResponseEndFrame,
@@ -12,34 +15,95 @@ from pipecat.frames.frames import (
 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."""
+    """Mirrors LLM text frames as streaming protocol events.
+
+    Placed between the LLM service and the TTS service, this processor
+    observes the LLM's text frames as they're emitted and forwards them
+    downstream as ``OutputTransportMessageUrgentFrame``s that the product
+    serializer turns into ``response.text.{started,delta,final}`` events.
+
+    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
+    started/delta/final sequence for its fixed text.
+    """
 
     def __init__(self) -> None:
         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)
 
         if isinstance(frame, LLMFullResponseStartFrame):
+            await self.push_frame(frame, direction)
             await self._start_turn()
         elif isinstance(frame, LLMTextFrame):
+            await self.push_frame(frame, direction)
             if frame.text:
                 await self._delta(frame.text)
         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):
             text = frame.text or ""
+            await self.push_frame(frame, direction)
             await self._start_turn()
             if text:
                 await self._delta(text)
             await self._end_turn(interrupted=False)
-
-        await self.push_frame(frame, direction)
+        else:
+            await self.push_frame(frame, direction)
 
     async def _start_turn(self) -> None:
         if self._turn_active:
@@ -58,6 +122,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(