Add voice state tags, SuperTTS configs, and demo WS log groups.

Parse leading <state> tags from LLM replies and emit response.state over the product websocket while stripping tags from TTS/text streams. Add FastGPT+Xfyun voice configs (including state-enabled preset), SuperTTS support, and context sync for interrupted turns. Refresh the voice demo with a state indicator and collapsible audio delta websocket log groups.

Co-authored-by: Cursor <cursoragent@cursor.com>

src/voice/config.py

@@ -26,6 +26,9 @@ def resolve_voice_config_path() -> Path:
 
 DEFAULT_VOICE_CONFIG = resolve_voice_config_path()
 
+SUPPORTED_LLM_PROVIDERS = frozenset({"openai", "fastgpt"})
+_LLM_PROVIDER_ALIASES = {"llm": "openai", "openai": "openai", "fastgpt": "fastgpt"}
+
 
 @dataclass(frozen=True)
 class ServerConfig:
@@ -93,11 +96,20 @@ class TurnConfig:
     )
 
 
+@dataclass(frozen=True)
+class ResponseStateConfig:
+    enabled: bool = False
+    tag: str = "state"
+    event_type: str = "response.state"
+    max_prefix_chars: int = 256
+
+
 @dataclass(frozen=True)
 class AgentConfig:
     system_prompt: str = "You are a helpful, friendly voice assistant."
     greeting: str | None = None
     greeting_mode: str = "generated"
+    response_state: ResponseStateConfig = field(default_factory=ResponseStateConfig)
 
 
 @dataclass(frozen=True)
@@ -106,6 +118,7 @@ class LLMConfig:
     api_key: str = ""
     base_url: str | None = None
     model: str = "gpt-4o-mini"
+    app_id: str | None = None
     temperature: float | None = 0.7
     chat_id: str | None = None
     variables: dict[str, str] = field(default_factory=dict)
@@ -113,6 +126,19 @@ class LLMConfig:
     timeout_sec: float = 60.0
     send_system_prompt: bool = False
 
+    @property
+    def is_fastgpt(self) -> bool:
+        return self.provider == "fastgpt"
+
+    @property
+    def is_openai(self) -> bool:
+        return self.provider == "openai"
+
+    @property
+    def uses_local_context_history(self) -> bool:
+        """Whether the pipeline should seed and maintain local LLM context history."""
+        return not self.is_fastgpt or self.send_system_prompt
+
 
 @dataclass(frozen=True)
 class STTConfig:
@@ -147,6 +173,8 @@ class TTSConfig:
     pitch: int = 50
     timeout_sec: float = 30.0
     source_sample_rate_hz: int | None = None
+    oral_level: str = "mid"
+    text_aggregation_mode: str | None = None
 
 
 @dataclass(frozen=True)
@@ -183,14 +211,24 @@ def config_from_dict(data: dict) -> EngineConfig:
         agent["greeting"] = None
     if agent.get("greeting_mode") not in (None, "generated", "fixed", "off"):
         raise ValueError("agent.greeting_mode must be one of: generated, fixed, off")
+    response_state = ResponseStateConfig(**_dict(agent.pop("response_state")))
+    if response_state.max_prefix_chars < 1:
+        raise ValueError("agent.response_state.max_prefix_chars must be greater than 0")
+    if not response_state.tag:
+        raise ValueError("agent.response_state.tag must not be empty")
+    if not response_state.event_type:
+        raise ValueError("agent.response_state.event_type must not be empty")
 
     stt = _dict(services.get("stt") or services.get("asr"))
     if stt.get("language") == "":
         stt["language"] = None
 
     llm = _dict(services.get("llm"))
+    llm["provider"] = _normalize_llm_provider(llm.get("provider", LLMConfig().provider))
     if llm.get("chat_id") == "":
         llm["chat_id"] = None
+    if llm.get("app_id") == "":
+        llm["app_id"] = None
     if not isinstance(llm.get("variables"), dict):
         llm["variables"] = {}
 
@@ -219,7 +257,7 @@ def config_from_dict(data: dict) -> EngineConfig:
                 )
             ),
         ),
-        agent=AgentConfig(**agent),
+        agent=AgentConfig(**agent, response_state=response_state),
         services=ServicesConfig(
             llm=LLMConfig(**llm),
             stt=STTConfig(**stt),
@@ -230,3 +268,14 @@ def config_from_dict(data: dict) -> EngineConfig:
 
 def _dict(value: object) -> dict:
     return dict(value) if isinstance(value, dict) else {}
+
+
+def _normalize_llm_provider(value: object) -> str:
+    provider = str(value or LLMConfig().provider).strip().lower()
+    normalized = _LLM_PROVIDER_ALIASES.get(provider)
+    if normalized is None:
+        supported = ", ".join(sorted(SUPPORTED_LLM_PROVIDERS | {"llm"}))
+        raise ValueError(
+            f"services.llm.provider must be one of: {supported}; got {value!r}"
+        )
+    return normalized

src/voice/context_sync.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import Any
+
+from pipecat.frames.frames import Frame, InterruptionFrame, LLMMessagesAppendFrame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+from .text_stream import ProductTextStreamProcessor, maybe_sync_assistant_context
+
+
+class AssistantContextSyncProcessor(FrameProcessor):
+    """Sync LLM context to urgent-streamed assistant text before text-input turns.
+
+    ``input.text`` with ``interrupt: true`` queues ``InterruptionFrame`` before
+    ``LLMMessagesAppendFrame``. This processor runs context repair after the
+    interrupt has propagated (including TTS-phase interrupts) and before the new
+    user message is appended.
+    """
+
+    def __init__(
+        self,
+        *,
+        text_stream: ProductTextStreamProcessor,
+        assistant_aggregator: Any,
+    ) -> None:
+        super().__init__()
+        self._text_stream = text_stream
+        self._assistant_aggregator = assistant_aggregator
+        self._sync_on_next_append = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, InterruptionFrame):
+            self._sync_on_next_append = True
+        elif isinstance(frame, LLMMessagesAppendFrame) and self._sync_on_next_append:
+            self._sync_on_next_append = False
+            maybe_sync_assistant_context(self._assistant_aggregator, self._text_stream)
+
+        await self.push_frame(frame, direction)

src/voice/fastgpt_llm.py

@@ -7,11 +7,13 @@ from typing import Any
 import httpx
 from fastgpt_client import AsyncChatClient, FastGPTInteractiveEvent, aiter_stream_events
 from fastgpt_client.exceptions import FastGPTError
+from loguru import logger
 
 from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
     Frame,
+    InterruptionFrame,
     LLMContextFrame,
     LLMFullResponseEndFrame,
     LLMFullResponseStartFrame,
@@ -133,6 +135,24 @@ class FastGPTLLMSettings(LLMSettings):
     detail: bool = False
 
 
+def _default_fastgpt_settings(*, model: str = "fastgpt") -> FastGPTLLMSettings:
+    return FastGPTLLMSettings(
+        model=model,
+        system_instruction=None,
+        temperature=None,
+        max_tokens=None,
+        top_p=None,
+        top_k=None,
+        frequency_penalty=None,
+        presence_penalty=None,
+        seed=None,
+        filter_incomplete_user_turns=False,
+        user_turn_completion_config=None,
+        variables={},
+        detail=False,
+    )
+
+
 class FastGPTLLMService(LLMService):
     """FastGPT LLM service using chatId server-side memory and workflow variables."""
 
@@ -144,18 +164,20 @@ class FastGPTLLMService(LLMService):
         api_key: str,
         base_url: str,
         chat_id: str | None = None,
+        app_id: str | None = None,
         send_system_prompt: bool = False,
         greeting_prompt: str | None = None,
         timeout: float = 60.0,
         settings: FastGPTLLMSettings | None = None,
         **kwargs,
     ) -> None:
-        default_settings = self.Settings(model="fastgpt")
+        default_settings = _default_fastgpt_settings()
         if settings is not None:
             default_settings.apply_update(settings)
         super().__init__(settings=default_settings, **kwargs)
 
         self._chat_id = chat_id or f"voice_{uuid.uuid4().hex[:16]}"
+        self._app_id = (app_id or "").strip()
         self._send_system_prompt = send_system_prompt
         self._greeting_prompt = (greeting_prompt or "你好").strip() or "你好"
         self._client = AsyncChatClient(
@@ -165,6 +187,10 @@ class FastGPTLLMService(LLMService):
         )
         self._active_response = None
 
+    @property
+    def app_id(self) -> str:
+        return self._app_id
+
     @property
     def chat_id(self) -> str:
         return self._chat_id
@@ -183,6 +209,63 @@ class FastGPTLLMService(LLMService):
         await self._close_active_response()
         await super().cancel(frame)
 
+    async def _handle_interruptions(self, _: InterruptionFrame) -> None:
+        await self._close_active_response()
+        await super()._handle_interruptions(_)
+
+    @staticmethod
+    def _welcome_text_from_init_payload(payload: Any) -> str:
+        if not isinstance(payload, dict):
+            return ""
+
+        for container in (payload.get("app"), payload.get("data"), payload):
+            if not isinstance(container, dict):
+                continue
+            nested_app = container.get("app")
+            if isinstance(nested_app, dict):
+                text = FastGPTLLMService._welcome_text_from_app(nested_app)
+                if text:
+                    return text
+            text = FastGPTLLMService._welcome_text_from_app(container)
+            if text:
+                return text
+        return ""
+
+    @staticmethod
+    def _welcome_text_from_app(app_payload: dict[str, Any]) -> str:
+        chat_config = (
+            app_payload.get("chatConfig")
+            if isinstance(app_payload.get("chatConfig"), dict)
+            else {}
+        )
+        return _first_nonempty_text(
+            chat_config.get("welcomeText"),
+            app_payload.get("welcomeText"),
+        )
+
+    async def fetch_welcome_text(self) -> str | None:
+        """Return FastGPT app welcome text from chat init when ``app_id`` is configured."""
+        if not self._app_id:
+            return None
+
+        try:
+            response = await self._client.get_chat_init(
+                appId=self._app_id,
+                chatId=self._chat_id,
+            )
+            response.raise_for_status()
+            text = self._welcome_text_from_init_payload(response.json())
+            if text:
+                logger.info(f"FastGPT welcomeText loaded for appId={self._app_id}")
+            return text or None
+        except FastGPTError as exc:
+            logger.warning(f"FastGPT chat init failed: {exc}")
+        except httpx.HTTPError as exc:
+            logger.warning(f"FastGPT chat init HTTP error: {exc}")
+        except Exception as exc:
+            logger.warning(f"FastGPT chat init error: {exc}")
+        return None
+
     async def _close_active_response(self) -> None:
         response = self._active_response
         self._active_response = None
@@ -216,6 +299,12 @@ class FastGPTLLMService(LLMService):
         messages = self._build_fastgpt_messages(context)
         variables = self._settings.variables or None
 
+        logger.info(
+            "FastGPT chat completion "
+            f"chatId={self._chat_id} appId={self._app_id or '-'} "
+            f"variables={sorted((variables or {}).keys())} messages={messages!r}"
+        )
+
         await self.start_ttfb_metrics()
 
         try:

src/voice/pipeline.py

@@ -32,10 +32,13 @@ from pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy import (
 from pipecat.turns.user_turn_strategies import UserTurnStrategies
 
 from .config import EngineConfig
+from .context_sync import AssistantContextSyncProcessor
+from .fastgpt_llm import FastGPTLLMService
 from .protocol import ProductWebsocketSerializer
 from .services import create_llm_service, create_stt_service, create_tts_service
+from .response_state import StateTagResponseProcessor
 from .text_input import ProductTextInputProcessor
-from .text_stream import ProductTextStreamProcessor, sync_streamed_assistant_context
+from .text_stream import ProductTextStreamProcessor, maybe_sync_assistant_context
 from .transcript_stream import ProductTranscriptStreamProcessor
 from .turn_start import InterruptionGateUserTurnStartStrategy
 
@@ -83,14 +86,15 @@ async def run_pipeline_with_serializer(
         session_variables={"session_id": chat_id, "channel": "voice"},
         greeting_prompt=config.agent.greeting,
     )
-    if llm_config.provider == "fastgpt":
-        logger.info(f"FastGPT chatId={chat_id}")
+    if llm_config.is_fastgpt:
+        logger.info(f"LLM backend=fastgpt chatId={chat_id} appId={llm_config.app_id or '-'}")
+    else:
+        logger.info(f"LLM backend=openai model={llm_config.model}")
 
     tts = create_tts_service(config.services.tts, config.audio)
 
-    use_fastgpt = llm_config.provider == "fastgpt" and not llm_config.send_system_prompt
     messages: list[dict[str, str]] = []
-    if not use_fastgpt:
+    if llm_config.uses_local_context_history:
         messages = [{"role": "system", "content": config.agent.system_prompt}]
         if config.agent.greeting and config.agent.greeting_mode == "generated":
             messages.append({"role": "system", "content": config.agent.greeting})
@@ -126,21 +130,31 @@ async def run_pipeline_with_serializer(
     )
 
     text_stream = ProductTextStreamProcessor()
+    context_sync = AssistantContextSyncProcessor(
+        text_stream=text_stream,
+        assistant_aggregator=assistant_aggregator,
+    )
 
-    pipeline = Pipeline(
+    processors = [
+        transport.input(),
+        ProductTextInputProcessor(),
+        stt,
+        ProductTranscriptStreamProcessor(),
+        context_sync,
+        user_aggregator,
+        llm,
+    ]
+    if config.agent.response_state.enabled:
+        processors.append(StateTagResponseProcessor(config.agent.response_state))
+    processors.extend(
         [
-            transport.input(),
-            ProductTextInputProcessor(),
-            stt,
-            ProductTranscriptStreamProcessor(),
-            user_aggregator,
-            llm,
             text_stream,
             tts,
             transport.output(),
             assistant_aggregator,
         ]
     )
+    pipeline = Pipeline(processors)
 
     task = PipelineTask(
         pipeline,
@@ -160,7 +174,14 @@ async def run_pipeline_with_serializer(
         if config.agent.greeting_mode == "fixed" and config.agent.greeting:
             await task.queue_frames([TTSSpeakFrame(config.agent.greeting)])
         elif config.agent.greeting_mode == "generated":
-            await task.queue_frames([LLMRunFrame()])
+            if isinstance(llm, FastGPTLLMService):
+                welcome = await llm.fetch_welcome_text()
+                if welcome:
+                    await task.queue_frames([TTSSpeakFrame(welcome)])
+                else:
+                    await task.queue_frames([LLMRunFrame()])
+            else:
+                await task.queue_frames([LLMRunFrame()])
 
     @transport.event_handler("on_client_disconnected")
     async def on_client_disconnected(_transport, _client):
@@ -192,14 +213,12 @@ 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 "",
-                )
+        maybe_sync_assistant_context(
+            _aggregator,
+            text_stream,
+            committed_text=message.content or "",
+        )
+        text_stream.take_interrupted_stream_text()
 
     runner = PipelineRunner(handle_sigint=False)
     await runner.run(task)

src/voice/protocol.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import base64
+import binascii
 import json
 from typing import Any
 
@@ -19,10 +20,15 @@ from pipecat.frames.frames import (
     OutputTransportMessageUrgentFrame,
     TextFrame,
     TranscriptionFrame,
+    UserImageRawFrame,
 )
 from pipecat.serializers.base_serializer import FrameSerializer
 
 
+MAX_INPUT_IMAGE_BYTES = 8 * 1024 * 1024
+SUPPORTED_INPUT_IMAGE_MIME_TYPES = {"image/jpeg", "image/png", "image/webp"}
+
+
 class ProductWebsocketSerializer(FrameSerializer):
     """Stable app-facing JSON/base64 protocol adapter for Pipecat websocket transport."""
 
@@ -118,7 +124,7 @@ class ProductWebsocketSerializer(FrameSerializer):
                 return None
             try:
                 pcm = base64.b64decode(audio)
-            except ValueError as exc:
+            except (binascii.Error, ValueError) as exc:
                 logger.warning(f"Invalid input.audio base64: {exc}")
                 return None
             return InputAudioRawFrame(
@@ -127,6 +133,9 @@ class ProductWebsocketSerializer(FrameSerializer):
                 num_channels=int(message.get("channels") or self._channels),
             )
 
+        if message_type == "input.image":
+            return self._deserialize_input_image(message)
+
         if message_type == "input.text":
             text = message.get("text")
             if not isinstance(text, str) or not text.strip():
@@ -147,6 +156,61 @@ class ProductWebsocketSerializer(FrameSerializer):
         logger.warning(f"Unsupported product websocket message type: {message_type!r}")
         return None
 
+    def _deserialize_input_image(self, message: dict[str, Any]) -> Frame | None:
+        encoded = message.get("image") or message.get("data")
+        if not isinstance(encoded, str):
+            logger.warning("input.image requires base64 'image' or 'data'")
+            return None
+
+        mime_type = str(message.get("mime_type") or message.get("media_type") or "image/jpeg")
+        if mime_type not in SUPPORTED_INPUT_IMAGE_MIME_TYPES:
+            logger.warning(
+                "input.image unsupported mime_type "
+                f"{mime_type!r}; expected one of {sorted(SUPPORTED_INPUT_IMAGE_MIME_TYPES)}"
+            )
+            return None
+
+        try:
+            width = int(message.get("width") or 0)
+            height = int(message.get("height") or 0)
+        except (TypeError, ValueError):
+            logger.warning("input.image width and height must be integers")
+            return None
+
+        if width <= 0 or height <= 0:
+            logger.warning("input.image requires positive integer width and height")
+            return None
+
+        if "," in encoded and encoded.lstrip().startswith("data:"):
+            encoded = encoded.split(",", 1)[1]
+
+        try:
+            image = base64.b64decode(encoded, validate=True)
+        except (binascii.Error, ValueError) as exc:
+            logger.warning(f"Invalid input.image base64: {exc}")
+            return None
+
+        if len(image) > MAX_INPUT_IMAGE_BYTES:
+            logger.warning(
+                f"input.image too large: {len(image)} bytes; "
+                f"max is {MAX_INPUT_IMAGE_BYTES} bytes"
+            )
+            return None
+
+        text = message.get("text")
+        if text is not None and not isinstance(text, str):
+            logger.warning("input.image text must be a string when provided")
+            return None
+
+        return UserImageRawFrame(
+            image=image,
+            size=(width, height),
+            format=mime_type,
+            user_id=str(message.get("user_id") or "product-user"),
+            text=text or "Answer using this camera image.",
+            append_to_context=bool(message.get("append_to_context", True)),
+        )
+
     def _event(self, event_type: str, **payload: Any) -> str:
         self._sequence += 1
         return json.dumps(

src/voice/response_state.py

@@ -0,0 +1,136 @@
+from __future__ import annotations
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    Frame,
+    InterruptionFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMTextFrame,
+    OutputTransportMessageUrgentFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+from .config import ResponseStateConfig
+
+
+class StateTagResponseProcessor(FrameProcessor):
+    """Extract a leading state tag from LLM text before text streaming and TTS.
+
+    Expected model output:
+
+        <state>some state</state>spoken response
+
+    The extracted state is emitted as a product protocol event, while only the
+    spoken response text is forwarded downstream. If the model does not produce
+    the tag, the original text is forwarded unchanged.
+    """
+
+    def __init__(self, config: ResponseStateConfig) -> None:
+        super().__init__()
+        self._tag = config.tag
+        self._event_type = config.event_type
+        self._max_prefix_chars = config.max_prefix_chars
+        self._opening_tag = f"<{self._tag}>"
+        self._closing_tag = f"</{self._tag}>"
+        self._start_frame: LLMFullResponseStartFrame | None = None
+        self._buffer = ""
+        self._decided = False
+        self._in_llm_response = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, LLMFullResponseStartFrame):
+            self._start_frame = frame
+            self._buffer = ""
+            self._decided = False
+            self._in_llm_response = True
+            return
+
+        if isinstance(frame, LLMTextFrame) and self._in_llm_response and not self._decided:
+            await self._process_initial_text(frame.text or "", direction)
+            return
+
+        if isinstance(frame, LLMFullResponseEndFrame):
+            if self._in_llm_response:
+                await self._flush_buffer(direction)
+            await self.push_frame(frame, direction)
+            self._reset()
+            return
+
+        if isinstance(frame, (InterruptionFrame, CancelFrame)):
+            if self._in_llm_response:
+                await self._flush_buffer(direction)
+                self._reset()
+            await self.push_frame(frame, direction)
+            return
+
+        await self.push_frame(frame, direction)
+
+    async def _process_initial_text(self, text: str, direction: FrameDirection) -> None:
+        if not text:
+            return
+
+        self._buffer += text
+        decision = self._parse_buffer()
+        if decision is None:
+            return
+
+        self._decided = True
+        state, response_text = decision
+        if state is not None:
+            await self._emit_state(state)
+        await self._push_start(direction)
+        if response_text:
+            await self.push_frame(LLMTextFrame(response_text), direction)
+        self._buffer = ""
+
+    def _parse_buffer(self) -> tuple[str | None, str] | None:
+        stripped = self._buffer.lstrip()
+        if not stripped:
+            return None
+
+        if stripped.startswith(self._opening_tag):
+            state_start = len(self._opening_tag)
+            state_end = stripped.find(self._closing_tag, state_start)
+            if state_end >= 0:
+                response_start = state_end + len(self._closing_tag)
+                return stripped[state_start:state_end].strip(), stripped[response_start:]
+            if len(self._buffer) < self._max_prefix_chars:
+                return None
+            return None, self._buffer
+
+        if self._opening_tag.startswith(stripped) and len(self._buffer) < self._max_prefix_chars:
+            return None
+
+        return None, self._buffer
+
+    async def _flush_buffer(self, direction: FrameDirection) -> None:
+        await self._push_start(direction)
+        if self._buffer:
+            await self.push_frame(LLMTextFrame(self._buffer), direction)
+            self._buffer = ""
+        self._decided = True
+
+    async def _push_start(self, direction: FrameDirection) -> None:
+        if self._start_frame:
+            await self.push_frame(self._start_frame, direction)
+            self._start_frame = None
+
+    async def _emit_state(self, state: str) -> None:
+        await self.push_frame(
+            OutputTransportMessageUrgentFrame(
+                message={
+                    "type": self._event_type,
+                    "state": state,
+                }
+            ),
+            FrameDirection.DOWNSTREAM,
+        )
+
+    def _reset(self) -> None:
+        self._start_frame = None
+        self._buffer = ""
+        self._decided = False
+        self._in_llm_response = False

src/voice/services.py

@@ -10,11 +10,13 @@ from pipecat.services.openai._constants import OPENAI_SAMPLE_RATE
 from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.services.openai.stt import OpenAISTTService
 from pipecat.services.openai.tts import VALID_VOICES, OpenAITTSService
+from pipecat.services.tts_service import TextAggregationMode
 from pipecat.transcriptions.language import Language
 
 from .config import AudioConfig, LLMConfig, STTConfig, TTSConfig
 from .fastgpt_llm import FastGPTLLMService, FastGPTLLMSettings
 from .xfyun_asr import DEFAULT_XFYUN_ASR_URL, XfyunASRService
+from .xfyun_super_tts import DEFAULT_XFYUN_SUPER_TTS_URL, XfyunSuperTTSService
 from .xfyun_tts import DEFAULT_XFYUN_TTS_URL, XfyunTTSService
 
 
@@ -54,12 +56,13 @@ def create_llm_service(
     session_variables: dict | None = None,
     greeting_prompt: str | None = None,
 ):
-    if config.provider == "fastgpt":
+    if config.is_fastgpt:
         variables = {**config.variables, **(session_variables or {})}
         return FastGPTLLMService(
             api_key=config.api_key,
             base_url=config.base_url or "http://localhost:3000",
             chat_id=chat_id or config.chat_id,
+            app_id=config.app_id,
             send_system_prompt=config.send_system_prompt,
             greeting_prompt=greeting_prompt,
             timeout=config.timeout_sec,
@@ -70,7 +73,11 @@ def create_llm_service(
             ),
         )
 
-    _require_provider(config.provider, "openai", "llm")
+    if not config.is_openai:
+        supported = ", ".join(sorted(("openai", "fastgpt", "llm")))
+        raise ValueError(
+            f"Unsupported llm provider {config.provider!r}; expected one of: {supported}"
+        )
     return OpenAILLMService(
         api_key=config.api_key or None,
         base_url=config.base_url,
@@ -102,6 +109,30 @@ def create_tts_service(config: TTSConfig, audio: AudioConfig):
             timeout=config.timeout_sec,
         )
 
+    if config.provider in ("xfyun_super", "xfyun_super_tts"):
+        source_sample_rate = config.source_sample_rate_hz or 24000
+        if source_sample_rate not in (8000, 16000, 24000):
+            raise ValueError(
+                "Xfyun Super TTS source_sample_rate_hz must be 8000, 16000, or 24000"
+            )
+        text_aggregation_mode = config.text_aggregation_mode or TextAggregationMode.TOKEN
+        return XfyunSuperTTSService(
+            app_id=config.app_id,
+            api_key=config.api_key or "",
+            api_secret=config.api_secret,
+            voice=config.voice,
+            url=config.base_url or DEFAULT_XFYUN_SUPER_TTS_URL,
+            sample_rate=audio.sample_rate_hz,
+            source_sample_rate=source_sample_rate,
+            encoding=config.aue,
+            speed=config.speed,
+            volume=config.volume,
+            pitch=config.pitch,
+            oral_level=config.oral_level,
+            text_aggregation_mode=text_aggregation_mode,
+            open_timeout=config.timeout_sec,
+        )
+
     _require_provider(config.provider, "openai", "tts")
     service_class = OpenAITTSService if config.voice in VALID_VOICES else OpenAICompatibleTTSService
     return service_class(

src/voice/text_stream.py

@@ -20,16 +20,31 @@ class _AssistantContextSync(Protocol):
     def context(self) -> Any: ...
 
 
+def _committed_assistant_content(context: Any) -> str:
+    """Return trailing assistant text only when the last context message is assistant."""
+    messages = context.get_messages()
+    if not messages:
+        return ""
+    last = messages[-1]
+    if not isinstance(last, dict) or last.get("role") != "assistant":
+        return ""
+    content = last.get("content")
+    if isinstance(content, str):
+        return content.strip()
+    return ""
+
+
 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.
+    """Align LLM context with urgent-streamed UI text.
 
-    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.
+    The assistant aggregator commits TTS-spoken text; ``ProductTextStreamProcessor``
+    mirrors the LLM stream to the client. Replace or insert the streamed text so
+    the next turn sees what the user read on screen.
     """
     streamed = streamed_text.strip()
     if not streamed or streamed == committed_text.strip():
@@ -39,19 +54,58 @@ def sync_streamed_assistant_context(
 
     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
+        if not updated:
+            updated.append({"role": "assistant", "content": streamed})
+            return updated
+
+        last = updated[-1]
+        if isinstance(last, dict) and last.get("role") == "assistant":
+            content = last.get("content")
+            if isinstance(content, str) and content.strip() != streamed:
+                updated[-1] = {"role": "assistant", "content": streamed}
+            return updated
+
+        if (
+            len(updated) >= 2
+            and isinstance(last, dict)
+            and last.get("role") == "user"
+        ):
+            prev = updated[-2]
+            if isinstance(prev, dict) and prev.get("role") == "user":
+                updated.insert(len(updated) - 1, {"role": "assistant", "content": streamed})
+                return updated
+
+        if isinstance(last, dict) and last.get("role") == "user":
+            updated.append({"role": "assistant", "content": streamed})
+            return updated
+
         updated.append({"role": "assistant", "content": streamed})
         return updated
 
     aggregator.context.transform_messages(_apply)
 
 
+def maybe_sync_assistant_context(
+    aggregator: _AssistantContextSync,
+    text_stream: "ProductTextStreamProcessor",
+    *,
+    committed_text: str | None = None,
+) -> None:
+    committed = (
+        committed_text.strip()
+        if committed_text is not None
+        else _committed_assistant_content(aggregator.context)
+    )
+    streamed = text_stream.last_assistant_stream_text()
+    if not streamed:
+        return
+    sync_streamed_assistant_context(
+        aggregator,
+        streamed_text=streamed,
+        committed_text=committed,
+    )
+
+
 class ProductTextStreamProcessor(FrameProcessor):
     """Mirrors LLM text frames as streaming protocol events.
 
@@ -72,8 +126,12 @@ class ProductTextStreamProcessor(FrameProcessor):
         super().__init__()
         self._aggregation: list[str] = []
         self._turn_active = False
+        self._last_assistant_stream_text = ""
         self._interrupted_stream_text: str | None = None
 
+    def last_assistant_stream_text(self) -> str:
+        return self._last_assistant_stream_text
+
     def take_interrupted_stream_text(self) -> str | None:
         text = self._interrupted_stream_text
         self._interrupted_stream_text = None
@@ -94,7 +152,7 @@ class ProductTextStreamProcessor(FrameProcessor):
             await self._end_turn(interrupted=False)
         elif isinstance(frame, (InterruptionFrame, CancelFrame)):
             await self.push_frame(frame, direction)
-            await self._end_turn(interrupted=True)
+            await self._handle_interrupt()
         elif isinstance(frame, TTSSpeakFrame):
             text = frame.text or ""
             await self.push_frame(frame, direction)
@@ -118,12 +176,24 @@ class ProductTextStreamProcessor(FrameProcessor):
         self._aggregation.append(text)
         await self._emit("response.text.delta", text=text)
 
+    async def _handle_interrupt(self) -> None:
+        if self._turn_active:
+            await self._end_turn(interrupted=True)
+            return
+
+        if self._last_assistant_stream_text:
+            self._interrupted_stream_text = self._last_assistant_stream_text
+
     async def _end_turn(self, *, interrupted: bool) -> None:
         if not self._turn_active:
             return
+
         full_text = "".join(self._aggregation)
+        if full_text:
+            self._last_assistant_stream_text = full_text
         if interrupted and full_text:
             self._interrupted_stream_text = full_text
+
         self._turn_active = False
         self._aggregation = []
         await self._emit(

src/voice/xfyun_super_tts.py

@@ -0,0 +1,391 @@
+from __future__ import annotations
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import json
+import os
+from collections.abc import AsyncGenerator
+from datetime import datetime, timezone
+from email.utils import format_datetime
+from typing import Any
+from urllib.parse import urlencode, urlparse
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    EndFrame,
+    ErrorFrame,
+    Frame,
+    StartFrame,
+    TTSAudioRawFrame,
+    TTSStoppedFrame,
+)
+from pipecat.services.settings import TTSSettings
+from pipecat.services.tts_service import TextAggregationMode, WebsocketTTSService
+from pipecat.utils.tracing.service_decorators import traced_tts
+
+try:
+    from websockets.asyncio.client import connect as websocket_connect
+    from websockets.protocol import State
+except ModuleNotFoundError as exc:
+    logger.error(f"Exception: {exc}")
+    logger.error("In order to use Xfyun Super TTS, install the websockets package.")
+    raise Exception(f"Missing module: {exc}") from exc
+
+from .xfyun_tts import _sanitize_text_for_tts
+
+
+DEFAULT_XFYUN_SUPER_TTS_URL = "wss://cbm01.cn-huabei-1.xf-yun.com/v1/private/mcd9m97e6"
+VALID_SAMPLE_RATES = {8000, 16000, 24000}
+
+
+class XfyunSuperTTSService(WebsocketTTSService):
+    """iFlytek/Xfyun Super Smart TTS using bidirectional WebSocket streaming.
+
+    The service keeps one Xfyun synthesis session open for a Pipecat turn. Each
+    ``run_tts`` call sends a text segment with status 0/1, while ``flush_audio``
+    sends the terminal status 2 frame. Audio arrives on the receive task and is
+    appended to the Pipecat audio context.
+    """
+
+    def __init__(
+        self,
+        *,
+        app_id: str,
+        api_key: str,
+        api_secret: str,
+        voice: str,
+        url: str | None = None,
+        sample_rate: int = 16000,
+        source_sample_rate: int = 24000,
+        encoding: str = "raw",
+        speed: int = 50,
+        volume: int = 50,
+        pitch: int = 50,
+        oral_level: str = "mid",
+        text_aggregation_mode: TextAggregationMode | str | None = TextAggregationMode.TOKEN,
+        open_timeout: float = 30.0,
+        **kwargs,
+    ) -> None:
+        if isinstance(text_aggregation_mode, str):
+            text_aggregation_mode = TextAggregationMode(text_aggregation_mode)
+
+        super().__init__(
+            text_aggregation_mode=text_aggregation_mode,
+            push_text_frames=True,
+            push_stop_frames=False,
+            push_start_frame=True,
+            pause_frame_processing=False,
+            sample_rate=sample_rate,
+            settings=TTSSettings(model=None, voice=voice, language=None),
+            **kwargs,
+        )
+        self._app_id = app_id or os.environ.get("XFYUN_APP_ID", "")
+        self._api_key = api_key or os.environ.get("XFYUN_API_KEY", "")
+        self._api_secret = api_secret or os.environ.get("XFYUN_API_SECRET", "")
+        self._voice = voice
+        self._url = url or DEFAULT_XFYUN_SUPER_TTS_URL
+        self._source_sample_rate = source_sample_rate
+        self._encoding = encoding
+        self._speed = speed
+        self._volume = volume
+        self._pitch = pitch
+        self._oral_level = oral_level
+        self._open_timeout = open_timeout
+
+        self._receive_task: asyncio.Task | None = None
+        self._active_context_id: str | None = None
+        self._started_contexts: set[str] = set()
+        self._seq_by_context: dict[str, int] = {}
+        self._sent_text_bytes_by_context: dict[str, int] = {}
+        self._stream_completed = False
+
+    def can_generate_metrics(self) -> bool:
+        return True
+
+    async def start(self, frame: StartFrame) -> None:
+        await super().start(frame)
+        if not self._app_id or not self._api_key or not self._api_secret:
+            await self.push_error(
+                error_msg="Xfyun Super TTS requires app_id, api_key, and api_secret"
+            )
+            return
+        if self._encoding != "raw":
+            await self.push_error(error_msg="Xfyun Super TTS must use raw PCM audio in Pipecat")
+            return
+        if self._source_sample_rate not in VALID_SAMPLE_RATES:
+            await self.push_error(
+                error_msg=(
+                    "Xfyun Super TTS source_sample_rate must be one of "
+                    f"{sorted(VALID_SAMPLE_RATES)}"
+                )
+            )
+            return
+        await self._connect()
+
+    async def stop(self, frame: EndFrame) -> None:
+        await super().stop(frame)
+        await self._disconnect()
+
+    async def cancel(self, frame: CancelFrame) -> None:
+        await super().cancel(frame)
+        await self._disconnect()
+
+    async def flush_audio(self, context_id: str | None = None) -> None:
+        flush_id = context_id or self.get_active_audio_context_id()
+        if not flush_id or not self._websocket:
+            return
+        if flush_id not in self._started_contexts:
+            return
+
+        logger.trace(f"{self}: flushing Xfyun Super TTS stream {flush_id}")
+        await self._send_request_frame(flush_id, "", status=2)
+
+    async def on_audio_context_interrupted(self, context_id: str) -> None:
+        await self.stop_all_metrics()
+        await self._reset_context(context_id)
+        await self._disconnect()
+        await self._connect()
+        await super().on_audio_context_interrupted(context_id)
+
+    async def _connect(self) -> None:
+        await super()._connect()
+        await self._connect_websocket()
+        if self._websocket and not self._receive_task:
+            self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
+
+    async def _disconnect(self) -> None:
+        await super()._disconnect()
+        if self._receive_task:
+            await self.cancel_task(self._receive_task)
+            self._receive_task = None
+        await self._disconnect_websocket()
+
+    async def _connect_websocket(self) -> None:
+        try:
+            if self._websocket and self._websocket.state is State.OPEN:
+                return
+            logger.debug("Connecting to Xfyun Super TTS")
+            auth_url = _build_auth_url(self._url, self._api_key, self._api_secret)
+            self._websocket = await websocket_connect(
+                auth_url,
+                max_size=None,
+                open_timeout=self._open_timeout,
+            )
+            await self._call_event_handler("on_connected")
+        except Exception as exc:
+            self._websocket = None
+            await self.push_error(
+                error_msg=f"Unable to connect to Xfyun Super TTS: {exc}",
+                exception=exc,
+            )
+            await self._call_event_handler("on_connection_error", f"{exc}")
+
+    async def _disconnect_websocket(self) -> None:
+        try:
+            await self.stop_all_metrics()
+            if self._websocket:
+                logger.debug("Disconnecting from Xfyun Super TTS")
+                await self._websocket.close()
+        except Exception as exc:
+            await self.push_error(
+                error_msg=f"Error closing Xfyun Super TTS websocket: {exc}",
+                exception=exc,
+            )
+        finally:
+            await self.remove_active_audio_context()
+            self._websocket = None
+            self._active_context_id = None
+            self._started_contexts.clear()
+            self._seq_by_context.clear()
+            self._sent_text_bytes_by_context.clear()
+            self._stream_completed = False
+            await self._call_event_handler("on_disconnected")
+
+    def _get_websocket(self):
+        if self._websocket:
+            return self._websocket
+        raise Exception("Websocket not connected")
+
+    async def _receive_messages(self) -> None:
+        async for raw_message in self._get_websocket():
+            try:
+                message = json.loads(raw_message)
+            except json.JSONDecodeError:
+                logger.warning(f"{self}: received non-JSON Xfyun Super TTS message: {raw_message!r}")
+                continue
+
+            header = message.get("header") or {}
+            code = header.get("code", -1)
+            sid = header.get("sid")
+            context_id = self._active_context_id
+
+            if code != 0:
+                error_message = header.get("message", "unknown error")
+                await self.push_error(
+                    error_msg=f"Xfyun Super TTS error code={code}, sid={sid}: {error_message}"
+                )
+                if context_id and self.audio_context_available(context_id):
+                    await self.append_to_audio_context(
+                        context_id, TTSStoppedFrame(context_id=context_id)
+                    )
+                    await self.remove_audio_context(context_id)
+                if context_id:
+                    await self._reset_context(context_id)
+                continue
+
+            audio_obj = (message.get("payload") or {}).get("audio") or {}
+            audio_b64 = audio_obj.get("audio")
+            if audio_b64 and context_id and self.audio_context_available(context_id):
+                await self.stop_ttfb_metrics()
+                audio = base64.b64decode(audio_b64)
+                if self._source_sample_rate != self.sample_rate:
+                    audio = await self._resampler.resample(
+                        audio, self._source_sample_rate, self.sample_rate
+                    )
+                frame = TTSAudioRawFrame(audio, self.sample_rate, 1, context_id=context_id)
+                await self.append_to_audio_context(context_id, frame)
+
+            audio_status = audio_obj.get("status")
+            header_status = header.get("status")
+            if audio_status == 2 or header_status == 2:
+                if context_id and self.audio_context_available(context_id):
+                    await self.append_to_audio_context(
+                        context_id, TTSStoppedFrame(context_id=context_id)
+                    )
+                    await self.remove_audio_context(context_id)
+                if context_id:
+                    await self._reset_context(context_id)
+                self._stream_completed = True
+
+    @traced_tts
+    async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame | None, None]:
+        sanitized = _sanitize_text_for_tts(text)
+        if not sanitized:
+            return
+
+        if not self._is_streaming_tokens:
+            logger.debug(f"{self}: Generating Xfyun Super TTS [{sanitized}]")
+        else:
+            logger.trace(f"{self}: Generating Xfyun Super TTS [{sanitized}]")
+
+        if self._stream_completed and self._websocket:
+            await self._disconnect()
+            await self._connect()
+
+        if not self._websocket or self._websocket.state is State.CLOSED:
+            await self._connect()
+
+        if self._active_context_id and self._active_context_id != context_id:
+            yield ErrorFrame(
+                error=(
+                    "Xfyun Super TTS supports one active synthesis stream per WebSocket; "
+                    f"active={self._active_context_id}, new={context_id}"
+                )
+            )
+            return
+
+        try:
+            status = 0 if context_id not in self._started_contexts else 1
+            await self._send_request_frame(context_id, sanitized, status=status)
+            await self.start_tts_usage_metrics(sanitized)
+        except Exception as exc:
+            yield ErrorFrame(error=f"Xfyun Super TTS request failed: {exc}")
+            yield TTSStoppedFrame(context_id=context_id)
+            await self._disconnect()
+            await self._connect()
+            return
+
+        yield None
+
+    async def _send_request_frame(self, context_id: str, text: str, *, status: int) -> None:
+        if status == 0:
+            self._active_context_id = context_id
+            self._started_contexts.add(context_id)
+
+        seq = self._seq_by_context.get(context_id, 0)
+        text_bytes = text.encode("utf-8")
+        total_bytes = self._sent_text_bytes_by_context.get(context_id, 0) + len(text_bytes)
+        if total_bytes > 65536:
+            raise ValueError("Xfyun Super TTS text must not exceed 64K UTF-8 bytes per stream")
+
+        frame = self._build_request_frame(text, status=status, seq=seq)
+        await self._get_websocket().send(json.dumps(frame, ensure_ascii=False))
+
+        self._seq_by_context[context_id] = seq + 1
+        self._sent_text_bytes_by_context[context_id] = total_bytes
+
+    def _build_request_frame(self, text: str, *, status: int, seq: int) -> dict[str, Any]:
+        return {
+            "header": {
+                "app_id": self._app_id,
+                "status": status,
+            },
+            "parameter": {
+                "oral": {
+                    "oral_level": self._oral_level,
+                },
+                "tts": {
+                    "vcn": self._voice,
+                    "speed": self._speed,
+                    "volume": self._volume,
+                    "pitch": self._pitch,
+                    "bgs": 0,
+                    "reg": 0,
+                    "rdn": 0,
+                    "rhy": 0,
+                    "audio": {
+                        "encoding": self._encoding,
+                        "sample_rate": self._source_sample_rate,
+                        "channels": 1,
+                        "bit_depth": 16,
+                        "frame_size": 0,
+                    },
+                },
+            },
+            "payload": {
+                "text": {
+                    "encoding": "utf8",
+                    "compress": "raw",
+                    "format": "plain",
+                    "status": status,
+                    "seq": seq,
+                    "text": base64.b64encode(text.encode("utf-8")).decode("utf-8"),
+                },
+            },
+        }
+
+    async def _reset_context(self, context_id: str) -> None:
+        self._started_contexts.discard(context_id)
+        self._seq_by_context.pop(context_id, None)
+        self._sent_text_bytes_by_context.pop(context_id, None)
+        if self._active_context_id == context_id:
+            self._active_context_id = None
+
+
+def _build_auth_url(url: str, api_key: str, api_secret: str) -> str:
+    parsed = urlparse(url)
+    if parsed.scheme not in {"ws", "wss"} or not parsed.hostname:
+        raise ValueError(f"invalid Xfyun Super TTS WebSocket URL: {url}")
+
+    host = parsed.hostname
+    path = parsed.path or "/"
+    date = format_datetime(datetime.now(timezone.utc), usegmt=True)
+    request_line = f"GET {path} HTTP/1.1"
+    signature_origin = f"host: {host}\ndate: {date}\n{request_line}"
+    signature_sha = hmac.new(
+        api_secret.encode("utf-8"),
+        signature_origin.encode("utf-8"),
+        digestmod=hashlib.sha256,
+    ).digest()
+    signature = base64.b64encode(signature_sha).decode("utf-8")
+    authorization_origin = (
+        f'api_key="{api_key}", algorithm="hmac-sha256", '
+        f'headers="host date request-line", signature="{signature}"'
+    )
+    authorization = base64.b64encode(authorization_origin.encode("utf-8")).decode("utf-8")
+    query = urlencode({"authorization": authorization, "date": date, "host": host})
+    return f"{url}?{query}"