Integrate product-ws voice demo on port 8000 alongside REST API.

Add src/voice Pipecat pipeline, browser demo at /voice-demo, and config/voice.json.

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

src/main.py

@@ -1,8 +1,8 @@
 from fastapi import FastAPI
-import sys
 from .api.endpoints import router as api_router
 from .core.fastgpt_client import lifespan
 from .core.logging_config import setup_logging
+from .voice.routes import register_voice
 
 # Setup logging first
 setup_logging()
@@ -18,4 +18,5 @@ app = FastAPI(
 def read_root():
     return {"message": "Server is running."}
 
-app.include_router(api_router)
+app.include_router(api_router)
+register_voice(app)

src/voice/__init__.py

@@ -0,0 +1 @@
+"""Voice websocket demo (product-ws / va.ws.v1) powered by Pipecat."""

src/voice/config.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass, field
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+DEFAULT_VOICE_CONFIG = PROJECT_ROOT / "config" / "voice.json"
+
+
+@dataclass(frozen=True)
+class ServerConfig:
+    host: str = "0.0.0.0"
+    port: int = 8000
+    cors_origins: list[str] = field(default_factory=list)
+    serve_webpage: bool = True
+    webpage_mount: str = "/voice-demo"
+
+
+@dataclass(frozen=True)
+class AudioConfig:
+    sample_rate_hz: int = 16000
+    channels: int = 1
+    frame_ms: int = 20
+
+    @property
+    def frame_bytes(self) -> int:
+        return int(self.sample_rate_hz * self.frame_ms / 1000) * self.channels * 2
+
+
+@dataclass(frozen=True)
+class SessionConfig:
+    inactivity_timeout_sec: int = 60
+
+
+@dataclass(frozen=True)
+class VADConfig:
+    confidence: float = 0.7
+    start_secs: float = 0.2
+    stop_secs: float = 0.6
+    min_volume: float = 0.6
+
+
+@dataclass(frozen=True)
+class TurnConfig:
+    vad: VADConfig = field(default_factory=VADConfig)
+    user_speech_timeout_sec: float = 1.0
+    interruption_min_chars: int = 3
+    interruption_use_interim: bool = True
+    interruption_short_replies: list[str] = field(
+        default_factory=lambda: [
+            "是",
+            "是的",
+            "对",
+            "对的",
+            "嗯",
+            "好",
+            "好的",
+            "行",
+            "可以",
+            "没问题",
+            "不是",
+            "不",
+            "不行",
+            "不用",
+            "不要",
+            "没有",
+            "否",
+            "no",
+            "yes",
+            "ok",
+            "okay",
+        ]
+    )
+
+
+@dataclass(frozen=True)
+class AgentConfig:
+    system_prompt: str = "You are a helpful, friendly voice assistant."
+    greeting: str | None = None
+    greeting_mode: str = "generated"
+
+
+@dataclass(frozen=True)
+class LLMConfig:
+    provider: str = "openai"
+    api_key: str = ""
+    base_url: str | None = None
+    model: str = "gpt-4o-mini"
+    temperature: float | None = 0.7
+
+
+@dataclass(frozen=True)
+class STTConfig:
+    provider: str = "openai"
+    app_id: str = ""
+    api_key: str = ""
+    api_secret: str = ""
+    base_url: str | None = None
+    model: str = "gpt-4o-mini-transcribe"
+    language: str | None = "en"
+    domain: str = "iat"
+    accent: str = "mandarin"
+    encoding: str = "raw"
+    frame_size: int = 1280
+    timeout_sec: float = 10.0
+    dynamic_correction: bool = False
+
+
+@dataclass(frozen=True)
+class TTSConfig:
+    provider: str = "openai"
+    app_id: str = ""
+    api_key: str = ""
+    api_secret: str = ""
+    base_url: str | None = None
+    model: str = "gpt-4o-mini-tts"
+    voice: str = "alloy"
+    aue: str = "raw"
+    tte: str = "UTF8"
+    speed: int = 50
+    volume: int = 50
+    pitch: int = 50
+    timeout_sec: float = 30.0
+    source_sample_rate_hz: int | None = None
+
+
+@dataclass(frozen=True)
+class ServicesConfig:
+    llm: LLMConfig = field(default_factory=LLMConfig)
+    stt: STTConfig = field(default_factory=STTConfig)
+    tts: TTSConfig = field(default_factory=TTSConfig)
+
+
+@dataclass(frozen=True)
+class EngineConfig:
+    server: ServerConfig = field(default_factory=ServerConfig)
+    audio: AudioConfig = field(default_factory=AudioConfig)
+    session: SessionConfig = field(default_factory=SessionConfig)
+    turn: TurnConfig = field(default_factory=TurnConfig)
+    agent: AgentConfig = field(default_factory=AgentConfig)
+    services: ServicesConfig = field(default_factory=ServicesConfig)
+
+
+def load_config(path: str | Path | None = None) -> EngineConfig:
+    config_path = Path(path) if path is not None else DEFAULT_VOICE_CONFIG
+    if not config_path.is_absolute():
+        config_path = PROJECT_ROOT / config_path
+    data = json.loads(config_path.read_text(encoding="utf-8"))
+    if not isinstance(data, dict):
+        raise ValueError(f"Config file must contain a JSON object: {config_path}")
+    return config_from_dict(data)
+
+
+def config_from_dict(data: dict) -> EngineConfig:
+    services = _dict(data.get("services"))
+    agent = _dict(data.get("agent"))
+    if agent.get("greeting") == "":
+        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")
+
+    stt = _dict(services.get("stt") or services.get("asr"))
+    if stt.get("language") == "":
+        stt["language"] = None
+
+    turn = _dict(data.get("turn"))
+    vad = _dict(turn.get("vad"))
+
+    return EngineConfig(
+        server=ServerConfig(**_dict(data.get("server"))),
+        audio=AudioConfig(**_dict(data.get("audio"))),
+        session=SessionConfig(**_dict(data.get("session"))),
+        turn=TurnConfig(
+            vad=VADConfig(**vad),
+            user_speech_timeout_sec=float(
+                turn.get("user_speech_timeout_sec", TurnConfig().user_speech_timeout_sec)
+            ),
+            interruption_min_chars=int(
+                turn.get("interruption_min_chars", TurnConfig().interruption_min_chars)
+            ),
+            interruption_use_interim=bool(
+                turn.get("interruption_use_interim", TurnConfig().interruption_use_interim)
+            ),
+            interruption_short_replies=list(
+                turn.get(
+                    "interruption_short_replies",
+                    TurnConfig().interruption_short_replies,
+                )
+            ),
+        ),
+        agent=AgentConfig(**agent),
+        services=ServicesConfig(
+            llm=LLMConfig(**_dict(services.get("llm"))),
+            stt=STTConfig(**stt),
+            tts=TTSConfig(**_dict(services.get("tts"))),
+        ),
+    )
+
+
+def _dict(value: object) -> dict:
+    return dict(value) if isinstance(value, dict) else {}

src/voice/pipeline.py

@@ -0,0 +1,179 @@
+from __future__ import annotations
+
+from loguru import logger
+
+from pipecat.audio.vad.silero import SileroVADAnalyzer
+from pipecat.audio.vad.vad_analyzer import VADParams
+from pipecat.frames.frames import (
+    LLMRunFrame,
+    OutputTransportMessageUrgentFrame,
+    TTSSpeakFrame,
+)
+from pipecat.pipeline.pipeline import Pipeline
+from pipecat.pipeline.runner import PipelineRunner
+from pipecat.pipeline.task import PipelineParams, PipelineTask
+from pipecat.processors.aggregators.llm_context import LLMContext
+from pipecat.processors.aggregators.llm_response_universal import (
+    AssistantTurnStoppedMessage,
+    LLMContextAggregatorPair,
+    LLMUserAggregatorParams,
+    UserTurnStoppedMessage,
+)
+from pipecat.serializers.base_serializer import FrameSerializer
+from pipecat.transports.websocket.fastapi import (
+    FastAPIWebsocketParams,
+    FastAPIWebsocketTransport,
+)
+from pipecat.turns.user_stop.speech_timeout_user_turn_stop_strategy import (
+    SpeechTimeoutUserTurnStopStrategy,
+)
+from pipecat.turns.user_turn_strategies import UserTurnStrategies
+
+from .config import EngineConfig
+from .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 .transcript_stream import ProductTranscriptStreamProcessor
+from .turn_start import InterruptionGateUserTurnStartStrategy
+
+
+async def run_product_voice_pipeline(websocket, config: EngineConfig) -> None:
+    await run_pipeline_with_serializer(
+        websocket,
+        config,
+        serializer=ProductWebsocketSerializer(
+            sample_rate=config.audio.sample_rate_hz,
+            channels=config.audio.channels,
+        ),
+        client_label="Product JSON",
+    )
+
+
+async def run_pipeline_with_serializer(
+    websocket,
+    config: EngineConfig,
+    *,
+    serializer: FrameSerializer,
+    client_label: str,
+) -> None:
+    transport = FastAPIWebsocketTransport(
+        websocket=websocket,
+        params=FastAPIWebsocketParams(
+            audio_in_enabled=True,
+            audio_out_enabled=True,
+            audio_in_sample_rate=config.audio.sample_rate_hz,
+            audio_out_sample_rate=config.audio.sample_rate_hz,
+            audio_in_channels=config.audio.channels,
+            audio_out_channels=config.audio.channels,
+            serializer=serializer,
+            session_timeout=None,
+        ),
+    )
+
+    stt = create_stt_service(config.services.stt, config.audio)
+    llm = create_llm_service(config.services.llm)
+    tts = create_tts_service(config.services.tts, config.audio)
+
+    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})
+
+    context = LLMContext(messages)
+
+    vad_params = VADParams(
+        confidence=config.turn.vad.confidence,
+        start_secs=config.turn.vad.start_secs,
+        stop_secs=config.turn.vad.stop_secs,
+        min_volume=config.turn.vad.min_volume,
+    )
+    user_turn_strategies = UserTurnStrategies(
+        start=[
+            InterruptionGateUserTurnStartStrategy(
+                min_chars_when_bot_speaking=config.turn.interruption_min_chars,
+                allowed_short_replies=config.turn.interruption_short_replies,
+                use_interim=config.turn.interruption_use_interim,
+            ),
+        ],
+        stop=[
+            SpeechTimeoutUserTurnStopStrategy(
+                user_speech_timeout=config.turn.user_speech_timeout_sec,
+            ),
+        ],
+    )
+    user_aggregator, assistant_aggregator = LLMContextAggregatorPair(
+        context,
+        user_params=LLMUserAggregatorParams(
+            vad_analyzer=SileroVADAnalyzer(params=vad_params),
+            user_turn_strategies=user_turn_strategies,
+        ),
+    )
+
+    pipeline = Pipeline(
+        [
+            transport.input(),
+            ProductTextInputProcessor(),
+            stt,
+            ProductTranscriptStreamProcessor(),
+            user_aggregator,
+            llm,
+            ProductTextStreamProcessor(),
+            tts,
+            transport.output(),
+            assistant_aggregator,
+        ]
+    )
+
+    task = PipelineTask(
+        pipeline,
+        params=PipelineParams(
+            audio_in_sample_rate=config.audio.sample_rate_hz,
+            audio_out_sample_rate=config.audio.sample_rate_hz,
+            enable_metrics=True,
+            enable_usage_metrics=True,
+            enable_heartbeats=True,
+        ),
+        idle_timeout_secs=config.session.inactivity_timeout_sec,
+    )
+
+    @transport.event_handler("on_client_connected")
+    async def on_client_connected(_transport, _client):
+        logger.info(f"{client_label} websocket client connected")
+        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()])
+
+    @transport.event_handler("on_client_disconnected")
+    async def on_client_disconnected(_transport, _client):
+        logger.info(f"{client_label} websocket client disconnected")
+        await task.cancel()
+
+    @transport.event_handler("on_session_timeout")
+    async def on_session_timeout(_transport, _client):
+        logger.info(f"{client_label} websocket session timed out")
+        await task.cancel()
+
+    @user_aggregator.event_handler("on_user_turn_stopped")
+    async def on_user_turn_stopped(_aggregator, _strategy, message: UserTurnStoppedMessage):
+        logger.info(f"User: {message.content}")
+        text = (message.content or "").strip()
+        if not text:
+            return
+        await task.queue_frame(
+            OutputTransportMessageUrgentFrame(
+                message={
+                    "type": "input.transcript.final",
+                    "text": text,
+                    "user_id": message.user_id,
+                    "timestamp": message.timestamp,
+                }
+            )
+        )
+
+    @assistant_aggregator.event_handler("on_assistant_turn_stopped")
+    async def on_assistant_turn_stopped(_aggregator, message: AssistantTurnStoppedMessage):
+        logger.info(f"Assistant: {message.content}")
+
+    runner = PipelineRunner(handle_sigint=False)
+    await runner.run(task)

src/voice/protocol.py

@@ -0,0 +1,160 @@
+from __future__ import annotations
+
+import base64
+import json
+from typing import Any
+
+from loguru import logger
+
+from pipecat.frames.frames import (
+    CancelFrame,
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    EndFrame,
+    Frame,
+    InputAudioRawFrame,
+    InputTransportMessageFrame,
+    OutputAudioRawFrame,
+    OutputTransportMessageFrame,
+    OutputTransportMessageUrgentFrame,
+    TextFrame,
+    TranscriptionFrame,
+)
+from pipecat.serializers.base_serializer import FrameSerializer
+
+
+class ProductWebsocketSerializer(FrameSerializer):
+    """Stable app-facing JSON/base64 protocol adapter for Pipecat websocket transport."""
+
+    protocol = "va.ws.v1"
+
+    def __init__(self, *, sample_rate: int, channels: int):
+        super().__init__()
+        self._sample_rate = sample_rate
+        self._channels = channels
+        self._sequence = 0
+
+    async def serialize(self, frame: Frame) -> str | bytes | None:
+        if isinstance(frame, OutputAudioRawFrame):
+            return self._event(
+                "response.audio.delta",
+                audio=base64.b64encode(frame.audio).decode("ascii"),
+                bytes=len(frame.audio),
+                sample_rate=frame.sample_rate,
+                channels=frame.num_channels,
+            )
+
+        if isinstance(frame, BotStartedSpeakingFrame):
+            return self._event("response.audio.started")
+
+        if isinstance(frame, BotStoppedSpeakingFrame):
+            return self._event("response.audio.stopped")
+
+        if isinstance(frame, TranscriptionFrame):
+            return self._event(
+                "input.transcript.final",
+                text=frame.text,
+                user_id=frame.user_id,
+                timestamp=frame.timestamp,
+            )
+
+        if isinstance(frame, TextFrame):
+            return self._event("response.text.delta", text=frame.text)
+
+        if isinstance(frame, (OutputTransportMessageFrame, OutputTransportMessageUrgentFrame)):
+            if self.should_ignore_frame(frame):
+                return None
+            message = frame.message
+            if isinstance(message, dict) and isinstance(message.get("type"), str):
+                event_type = message["type"]
+                payload = {k: v for k, v in message.items() if k != "type"}
+                return self._event(event_type, **payload)
+            return self._event("transport.message", message=message)
+
+        return None
+
+    async def deserialize(self, data: str | bytes) -> Frame | None:
+        if isinstance(data, bytes):
+            return InputAudioRawFrame(
+                audio=data,
+                sample_rate=self._sample_rate,
+                num_channels=self._channels,
+            )
+
+        try:
+            message = json.loads(data)
+        except json.JSONDecodeError as exc:
+            logger.warning(f"Invalid product websocket JSON: {exc}")
+            return None
+
+        if not isinstance(message, dict):
+            logger.warning("Product websocket message must be a JSON object")
+            return None
+
+        message_type = message.get("type")
+        if message_type == "session.start":
+            return InputTransportMessageFrame(
+                message={
+                    "type": "session.started",
+                    "protocol": self.protocol,
+                    "audio": {
+                        "encoding": "pcm_s16le",
+                        "sample_rate": self._sample_rate,
+                        "channels": self._channels,
+                    },
+                }
+            )
+
+        if message_type == "session.stop":
+            return EndFrame()
+
+        if message_type == "response.cancel":
+            return CancelFrame(reason="client_cancelled")
+
+        if message_type == "input.audio":
+            audio = message.get("audio") or message.get("data")
+            if not isinstance(audio, str):
+                logger.warning("input.audio requires base64 'audio' or 'data'")
+                return None
+            try:
+                pcm = base64.b64decode(audio)
+            except ValueError as exc:
+                logger.warning(f"Invalid input.audio base64: {exc}")
+                return None
+            return InputAudioRawFrame(
+                audio=pcm,
+                sample_rate=int(message.get("sample_rate") or self._sample_rate),
+                num_channels=int(message.get("channels") or self._channels),
+            )
+
+        if message_type == "input.text":
+            text = message.get("text")
+            if not isinstance(text, str) or not text.strip():
+                logger.warning("input.text requires non-empty 'text'")
+                return None
+            return InputTransportMessageFrame(
+                message={
+                    "type": "input.text",
+                    "text": text,
+                    "interrupt": bool(message.get("interrupt", True)),
+                }
+            )
+
+        if message_type == "transport.message":
+            payload = message.get("message")
+            return InputTransportMessageFrame(message=payload if isinstance(payload, dict) else message)
+
+        logger.warning(f"Unsupported product websocket message type: {message_type!r}")
+        return None
+
+    def _event(self, event_type: str, **payload: Any) -> str:
+        self._sequence += 1
+        return json.dumps(
+            {
+                "type": event_type,
+                "protocol": self.protocol,
+                "seq": self._sequence,
+                **payload,
+            },
+            ensure_ascii=False,
+        )

src/voice/routes.py

@@ -0,0 +1,92 @@
+from __future__ import annotations
+
+from functools import lru_cache
+from pathlib import Path
+
+from fastapi import APIRouter, FastAPI, WebSocket
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.staticfiles import StaticFiles
+from loguru import logger
+
+from .config import DEFAULT_VOICE_CONFIG, EngineConfig, load_config
+from .pipeline import run_product_voice_pipeline
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+VOICE_DEMO_DIR = PROJECT_ROOT / "static" / "voice-demo"
+
+router = APIRouter(tags=["voice"])
+
+
+@lru_cache(maxsize=1)
+def get_voice_config() -> EngineConfig:
+    return load_config(DEFAULT_VOICE_CONFIG)
+
+
+def _normalize_mount_path(path: str) -> str:
+    normalized = path.strip() or "/voice-demo"
+    if not normalized.startswith("/"):
+        normalized = f"/{normalized}"
+    return normalized.rstrip("/") or "/"
+
+
+@router.get("/voice/health")
+async def voice_health() -> dict[str, object]:
+    config = get_voice_config()
+    mount = (
+        _normalize_mount_path(config.server.webpage_mount)
+        if config.server.serve_webpage
+        else None
+    )
+    return {
+        "status": "healthy",
+        "protocols": {
+            "/ws-product": "va.ws.v1.json_base64",
+        },
+        "features": {
+            "product_text_input": True,
+            "product_text_interrupt": True,
+        },
+        "demo": mount,
+        "llm_provider": config.services.llm.provider,
+        "stt_provider": config.services.stt.provider,
+        "tts_provider": config.services.tts.provider,
+    }
+
+
+@router.websocket("/ws-product")
+async def product_websocket_endpoint(websocket: WebSocket) -> None:
+    await websocket.accept()
+    config = get_voice_config()
+    await run_product_voice_pipeline(websocket, config)
+
+
+def register_voice(app: FastAPI) -> None:
+    """Mount voice websocket routes and optional browser demo static files."""
+    if not DEFAULT_VOICE_CONFIG.exists():
+        logger.warning(f"Voice config not found at {DEFAULT_VOICE_CONFIG}; voice demo disabled")
+        return
+
+    config = get_voice_config()
+    app.include_router(router)
+
+    if config.server.cors_origins:
+        app.add_middleware(
+            CORSMiddleware,
+            allow_origins=config.server.cors_origins,
+            allow_credentials=True,
+            allow_methods=["*"],
+            allow_headers=["*"],
+        )
+
+    if config.server.serve_webpage and VOICE_DEMO_DIR.is_dir():
+        mount = _normalize_mount_path(config.server.webpage_mount)
+        app.mount(
+            mount,
+            StaticFiles(directory=str(VOICE_DEMO_DIR), html=True),
+            name="voice-demo",
+        )
+        logger.info(f"Voice demo mounted at {mount}")
+    else:
+        logger.info("Voice demo static page disabled or missing")
+
+    logger.info("Voice websocket registered at /ws-product")

src/voice/services.py

@@ -0,0 +1,165 @@
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+
+from openai import BadRequestError
+from openai import NOT_GIVEN
+
+from pipecat.frames.frames import ErrorFrame, Frame, TTSAudioRawFrame
+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.transcriptions.language import Language
+
+from .config import AudioConfig, LLMConfig, STTConfig, TTSConfig
+from .xfyun_asr import DEFAULT_XFYUN_ASR_URL, XfyunASRService
+from .xfyun_tts import DEFAULT_XFYUN_TTS_URL, XfyunTTSService
+
+
+def create_stt_service(config: STTConfig, audio: AudioConfig | None = None):
+    if config.provider == "xfyun":
+        sample_rate = audio.sample_rate_hz if audio else 16000
+        return XfyunASRService(
+            app_id=config.app_id,
+            api_key=config.api_key or "",
+            api_secret=config.api_secret,
+            url=config.base_url or DEFAULT_XFYUN_ASR_URL,
+            language=config.language or "zh_cn",
+            domain=config.domain,
+            accent=config.accent,
+            sample_rate=sample_rate,
+            encoding=config.encoding,
+            frame_size=config.frame_size,
+            open_timeout=config.timeout_sec,
+            dynamic_correction=config.dynamic_correction,
+        )
+
+    _require_provider(config.provider, "openai", "stt")
+    return OpenAISTTService(
+        api_key=config.api_key or None,
+        base_url=config.base_url,
+        settings=OpenAISTTService.Settings(
+            model=config.model,
+            language=_language(config.language),
+        ),
+    )
+
+
+def create_llm_service(config: LLMConfig):
+    _require_provider(config.provider, "openai", "llm")
+    return OpenAILLMService(
+        api_key=config.api_key or None,
+        base_url=config.base_url,
+        settings=OpenAILLMService.Settings(
+            model=config.model,
+            temperature=config.temperature if config.temperature is not None else NOT_GIVEN,
+        ),
+    )
+
+
+def create_tts_service(config: TTSConfig, audio: AudioConfig):
+    if config.provider == "xfyun":
+        source_sample_rate = config.source_sample_rate_hz or audio.sample_rate_hz
+        if source_sample_rate not in (8000, 16000):
+            raise ValueError("Xfyun TTS source_sample_rate_hz must be 8000 or 16000")
+        return XfyunTTSService(
+            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_TTS_URL,
+            sample_rate=audio.sample_rate_hz,
+            source_sample_rate=source_sample_rate,
+            encoding=config.aue,
+            text_encoding=config.tte,
+            speed=config.speed,
+            volume=config.volume,
+            pitch=config.pitch,
+            timeout=config.timeout_sec,
+        )
+
+    _require_provider(config.provider, "openai", "tts")
+    service_class = OpenAITTSService if config.voice in VALID_VOICES else OpenAICompatibleTTSService
+    return service_class(
+        api_key=config.api_key or None,
+        base_url=config.base_url,
+        sample_rate=audio.sample_rate_hz,
+        source_sample_rate=config.source_sample_rate_hz,
+        settings=OpenAITTSService.Settings(
+            model=config.model,
+            voice=config.voice,
+        ),
+    )
+
+
+class OpenAICompatibleTTSService(OpenAITTSService):
+    """OpenAI-compatible TTS service that permits provider-specific voice ids."""
+
+    def __init__(self, *, source_sample_rate: int | None = None, **kwargs):
+        super().__init__(**kwargs)
+        self._source_sample_rate = source_sample_rate or OPENAI_SAMPLE_RATE
+
+    async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
+        voice = self._settings.voice
+        if not voice:
+            yield ErrorFrame(error="TTS voice must be specified")
+            return
+
+        try:
+            create_params = {
+                "input": text,
+                "model": self._settings.model,
+                "voice": voice,
+                "response_format": "pcm",
+            }
+
+            if self._settings.instructions:
+                create_params["instructions"] = self._settings.instructions
+
+            if self._settings.speed:
+                create_params["speed"] = self._settings.speed
+
+            async with self._client.audio.speech.with_streaming_response.create(
+                **create_params
+            ) as response:
+                if response.status_code != 200:
+                    error = await response.text()
+                    yield ErrorFrame(
+                        error=f"TTS request failed (status: {response.status_code}, error: {error})"
+                    )
+                    return
+
+                await self.start_tts_usage_metrics(text)
+
+                async def audio_chunks():
+                    async for chunk in response.iter_bytes(self.chunk_size):
+                        if chunk:
+                            yield chunk
+
+                first_frame = True
+                async for frame in self._stream_audio_frames_from_iterator(
+                    audio_chunks(),
+                    in_sample_rate=self._source_sample_rate,
+                    context_id=context_id,
+                ):
+                    if first_frame:
+                        await self.stop_ttfb_metrics()
+                        first_frame = False
+                    yield frame
+        except BadRequestError as exc:
+            yield ErrorFrame(error=f"TTS request failed: {exc}")
+        except Exception as exc:
+            yield ErrorFrame(error=f"TTS request failed: {exc}")
+
+
+def _require_provider(actual: str, expected: str, service_name: str) -> None:
+    if actual != expected:
+        raise ValueError(f"Unsupported {service_name} provider {actual!r}; expected {expected!r}")
+
+
+def _language(value: str | None) -> Language | None:
+    if value is None:
+        return None
+    normalized = value.replace("-", "_").upper()
+    return getattr(Language, normalized, value)

src/voice/text_input.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+from loguru import logger
+
+from pipecat.frames.frames import Frame, InputTransportMessageFrame, LLMMessagesAppendFrame
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+class ProductTextInputProcessor(FrameProcessor):
+    """Converts product text-input transport messages into LLM turns."""
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if not isinstance(frame, InputTransportMessageFrame):
+            await self.push_frame(frame, direction)
+            return
+
+        message = frame.message
+        if not isinstance(message, dict) or message.get("type") != "input.text":
+            await self.push_frame(frame, direction)
+            return
+
+        text = str(message.get("text") or "").strip()
+        if not text:
+            return
+
+        if message.get("interrupt", True):
+            logger.info("Text input interrupting current response")
+            await self.broadcast_interruption()
+
+        await self.push_frame(
+            LLMMessagesAppendFrame(
+                messages=[{"role": "user", "content": text}],
+                run_llm=True,
+            ),
+            FrameDirection.DOWNSTREAM,
+        )

src/voice/text_stream.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+from pipecat.frames.frames import (
+    Frame,
+    InterruptionFrame,
+    LLMFullResponseEndFrame,
+    LLMFullResponseStartFrame,
+    LLMTextFrame,
+    OutputTransportMessageUrgentFrame,
+    TTSSpeakFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+class ProductTextStreamProcessor(FrameProcessor):
+    """Mirrors LLM text frames as streaming protocol events."""
+
+    def __init__(self) -> None:
+        super().__init__()
+        self._aggregation: list[str] = []
+        self._turn_active = False
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, LLMFullResponseStartFrame):
+            await self._start_turn()
+        elif isinstance(frame, LLMTextFrame):
+            if frame.text:
+                await self._delta(frame.text)
+        elif isinstance(frame, LLMFullResponseEndFrame):
+            await self._end_turn(interrupted=False)
+        elif isinstance(frame, InterruptionFrame):
+            await self._end_turn(interrupted=True)
+        elif isinstance(frame, TTSSpeakFrame):
+            text = frame.text or ""
+            await self._start_turn()
+            if text:
+                await self._delta(text)
+            await self._end_turn(interrupted=False)
+
+        await self.push_frame(frame, direction)
+
+    async def _start_turn(self) -> None:
+        if self._turn_active:
+            return
+        self._turn_active = True
+        self._aggregation = []
+        await self._emit("response.text.started")
+
+    async def _delta(self, text: str) -> None:
+        if not self._turn_active:
+            await self._start_turn()
+        self._aggregation.append(text)
+        await self._emit("response.text.delta", text=text)
+
+    async def _end_turn(self, *, interrupted: bool) -> None:
+        if not self._turn_active:
+            return
+        full_text = "".join(self._aggregation)
+        self._turn_active = False
+        self._aggregation = []
+        await self._emit(
+            "response.text.final",
+            text=full_text,
+            interrupted=interrupted,
+        )
+
+    async def _emit(self, event_type: str, **payload: object) -> None:
+        await self.push_frame(
+            OutputTransportMessageUrgentFrame(
+                message={"type": event_type, **payload},
+            ),
+            FrameDirection.DOWNSTREAM,
+        )

src/voice/transcript_stream.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from pipecat.frames.frames import (
+    Frame,
+    InterimTranscriptionFrame,
+    OutputTransportMessageUrgentFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+
+
+class ProductTranscriptStreamProcessor(FrameProcessor):
+    """Mirrors interim STT frames to the product websocket protocol."""
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, InterimTranscriptionFrame):
+            await self.push_frame(
+                OutputTransportMessageUrgentFrame(
+                    message={
+                        "type": "input.transcript.interim",
+                        "text": frame.text,
+                        "user_id": frame.user_id,
+                        "timestamp": frame.timestamp,
+                    }
+                ),
+                FrameDirection.DOWNSTREAM,
+            )
+
+        await self.push_frame(frame, direction)

src/voice/turn_start.py

@@ -0,0 +1,86 @@
+from __future__ import annotations
+
+import re
+
+from loguru import logger
+from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    Frame,
+    InterimTranscriptionFrame,
+    TranscriptionFrame,
+)
+from pipecat.turns.types import ProcessFrameResult
+from pipecat.turns.user_start.base_user_turn_start_strategy import BaseUserTurnStartStrategy
+
+
+_COUNTABLE_TEXT_RE = re.compile(r"[\w\u4e00-\u9fff]", re.UNICODE)
+
+
+class InterruptionGateUserTurnStartStrategy(BaseUserTurnStartStrategy):
+    """Starts user turns only after likely intentional speech."""
+
+    def __init__(
+        self,
+        *,
+        min_chars_when_bot_speaking: int,
+        allowed_short_replies: list[str],
+        use_interim: bool = True,
+        **kwargs,
+    ):
+        super().__init__(**kwargs)
+        self._min_chars_when_bot_speaking = min_chars_when_bot_speaking
+        self._allowed_short_replies = {
+            self._normalize_text(reply) for reply in allowed_short_replies if reply.strip()
+        }
+        self._use_interim = use_interim
+        self._bot_speaking = False
+
+    async def reset(self):
+        await super().reset()
+
+    async def process_frame(self, frame: Frame) -> ProcessFrameResult:
+        if isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_speaking = True
+            return ProcessFrameResult.CONTINUE
+        if isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_speaking = False
+            return ProcessFrameResult.CONTINUE
+        if isinstance(frame, InterimTranscriptionFrame) and self._use_interim:
+            return await self._handle_transcription(frame.text, interim=True)
+        if isinstance(frame, TranscriptionFrame):
+            return await self._handle_transcription(frame.text, interim=False)
+
+        return ProcessFrameResult.CONTINUE
+
+    async def _handle_transcription(self, text: str, *, interim: bool) -> ProcessFrameResult:
+        normalized = self._normalize_text(text)
+        if not normalized:
+            return ProcessFrameResult.CONTINUE
+
+        if not self._bot_speaking:
+            await self.trigger_user_turn_started()
+            return ProcessFrameResult.STOP
+
+        should_interrupt = self._should_interrupt(normalized)
+        logger.debug(
+            f"{self} interruption_gate text={text!r} normalized={normalized!r} "
+            f"should_interrupt={should_interrupt} interim={interim}"
+        )
+
+        if should_interrupt:
+            await self.trigger_user_turn_started()
+            return ProcessFrameResult.STOP
+
+        await self.trigger_reset_aggregation()
+        return ProcessFrameResult.CONTINUE
+
+    def _should_interrupt(self, normalized: str) -> bool:
+        return (
+            normalized in self._allowed_short_replies
+            or len(normalized) >= self._min_chars_when_bot_speaking
+        )
+
+    @staticmethod
+    def _normalize_text(text: str) -> str:
+        return "".join(_COUNTABLE_TEXT_RE.findall(text.lower()))

src/voice/xfyun_asr.py

@@ -0,0 +1,353 @@
+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,
+    Frame,
+    InterimTranscriptionFrame,
+    TranscriptionFrame,
+    UserStoppedSpeakingFrame,
+    VADUserStartedSpeakingFrame,
+)
+from pipecat.processors.frame_processor import FrameDirection
+from pipecat.services.settings import STTSettings
+from pipecat.services.stt_service import STTService
+from pipecat.transcriptions.language import Language
+from pipecat.utils.time import time_now_iso8601
+from websockets.asyncio.client import connect as websocket_connect
+from websockets.protocol import State
+
+
+DEFAULT_XFYUN_ASR_URL = "wss://iat-api.xfyun.cn/v2/iat"
+
+
+class XfyunASRService(STTService):
+    """iFlytek/Xfyun streaming voice dictation service for Pipecat."""
+
+    def __init__(
+        self,
+        *,
+        app_id: str,
+        api_key: str,
+        api_secret: str,
+        url: str | None = None,
+        language: str = "zh_cn",
+        domain: str = "iat",
+        accent: str = "mandarin",
+        sample_rate: int = 16000,
+        encoding: str = "raw",
+        frame_size: int = 1280,
+        open_timeout: float = 10.0,
+        dynamic_correction: bool = False,
+        **kwargs,
+    ) -> None:
+        super().__init__(
+            sample_rate=sample_rate,
+            settings=STTSettings(model=None, language=language),
+            **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._url = url or DEFAULT_XFYUN_ASR_URL
+        self._language = language
+        self._domain = domain
+        self._accent = accent
+        self._encoding = encoding
+        self._frame_size = frame_size
+        self._open_timeout = open_timeout
+        self._dynamic_correction = dynamic_correction
+
+        self._websocket = None
+        self._receive_task = None
+        self._audio_buffer = bytearray()
+        self._sent_first_frame = False
+        self._sent_final_frame = False
+        self._finalizing_turn = False
+        self._partials: list[str] = []
+        self._last_text = ""
+
+    async def cleanup(self) -> None:
+        await self._close_utterance()
+        await super().cleanup()
+
+    async def stop(self, frame: EndFrame) -> None:
+        await self._close_utterance()
+        await super().stop(frame)
+
+    async def cancel(self, frame: CancelFrame) -> None:
+        await self._close_utterance()
+        await super().cancel(frame)
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection) -> None:
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, UserStoppedSpeakingFrame):
+            # Aggregator-level turn end (broadcast once per logical user turn).
+            # This is the only boundary that finalizes/closes the xfyun
+            # websocket, so brief VAD pauses do not restart the ASR session.
+            await self._finish_utterance()
+        elif isinstance(frame, VADUserStartedSpeakingFrame):
+            await self._start_utterance()
+
+    async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame | None, None]:
+        if not audio:
+            yield None
+            return
+
+        if not self._websocket or self._websocket.state is not State.OPEN:
+            await self._start_utterance()
+
+        self._audio_buffer.extend(audio)
+        await self._flush_audio_buffer(final=False)
+        yield None
+
+    async def _start_utterance(self) -> None:
+        if self._websocket and self._websocket.state is State.OPEN:
+            return
+
+        if not self._app_id or not self._api_key or not self._api_secret:
+            await self.push_error("Xfyun ASR requires app_id, api_key, and api_secret")
+            return
+
+        if self.sample_rate not in (8000, 16000):
+            await self.push_error("Xfyun ASR sample rate must be 8000 or 16000")
+            return
+
+        self._audio_buffer.clear()
+        self._partials = []
+        self._last_text = ""
+        self._sent_first_frame = False
+        self._sent_final_frame = False
+
+        auth_url = _build_auth_url(self._url, self._api_key, self._api_secret)
+        try:
+            self._websocket = await websocket_connect(
+                auth_url,
+                max_size=None,
+                open_timeout=self._open_timeout,
+            )
+        except Exception as exc:
+            await self.push_error(f"Xfyun ASR connection failed: {exc}", exception=exc)
+            self._websocket = None
+            return
+
+        self._receive_task = self.create_task(
+            self._receive_messages(),
+            name="xfyun_asr_receive",
+        )
+
+    async def _finish_utterance(self) -> None:
+        if not self._websocket or self._websocket.state is not State.OPEN:
+            return
+
+        await self._flush_audio_buffer(final=True)
+        if not self._sent_first_frame:
+            await self._close_utterance()
+            return
+
+        if not self._sent_final_frame:
+            self._finalizing_turn = True
+            await self._send_payload({"data": {"status": 2}})
+            self.request_finalize()
+            self._sent_final_frame = True
+
+    async def _close_utterance(self) -> None:
+        current_task = asyncio.current_task()
+        if self._receive_task and self._receive_task is not current_task:
+            await self.cancel_task(self._receive_task)
+            self._receive_task = None
+
+        websocket = self._websocket
+        self._websocket = None
+        if websocket and websocket.state is State.OPEN:
+            try:
+                await websocket.close()
+            except Exception:
+                pass
+
+        self._audio_buffer.clear()
+        self._sent_first_frame = False
+        self._sent_final_frame = False
+        self._finalizing_turn = False
+
+    async def _flush_audio_buffer(self, *, final: bool) -> None:
+        while len(self._audio_buffer) >= self._frame_size:
+            chunk = bytes(self._audio_buffer[: self._frame_size])
+            del self._audio_buffer[: self._frame_size]
+            await self._send_audio_chunk(chunk, status=1)
+
+        if final and self._audio_buffer:
+            chunk = bytes(self._audio_buffer)
+            self._audio_buffer.clear()
+            await self._send_audio_chunk(chunk, status=1)
+
+    async def _send_audio_chunk(self, audio: bytes, *, status: int) -> None:
+        if not audio:
+            return
+
+        if not self._sent_first_frame:
+            business = {
+                "language": self._language,
+                "domain": self._domain,
+                "accent": self._accent,
+            }
+            if self._dynamic_correction:
+                business["dwa"] = "wpgs"
+
+            payload = {
+                "common": {"app_id": self._app_id},
+                "business": business,
+                "data": {
+                    "status": 0,
+                    "format": f"audio/L16;rate={self.sample_rate}",
+                    "encoding": self._encoding,
+                    "audio": base64.b64encode(audio).decode("utf-8"),
+                },
+            }
+            self._sent_first_frame = True
+        else:
+            payload = {
+                "data": {
+                    "status": status,
+                    "format": f"audio/L16;rate={self.sample_rate}",
+                    "encoding": self._encoding,
+                    "audio": base64.b64encode(audio).decode("utf-8"),
+                }
+            }
+
+        await self._send_payload(payload)
+
+    async def _send_payload(self, payload: dict[str, Any]) -> None:
+        if not self._websocket or self._websocket.state is not State.OPEN:
+            return
+        await self._websocket.send(json.dumps(payload, ensure_ascii=False))
+
+    async def _receive_messages(self) -> None:
+        websocket = self._websocket
+        if not websocket:
+            return
+
+        try:
+            async for message in websocket:
+                await self._process_response(json.loads(message))
+        except Exception as exc:
+            if self._websocket is websocket:
+                await self.push_error(f"Xfyun ASR receive failed: {exc}", exception=exc)
+        finally:
+            if self._websocket is websocket:
+                self._websocket = None
+            self._receive_task = None
+
+    async def _process_response(self, payload: dict[str, Any]) -> None:
+        code = payload.get("code", -1)
+        if code != 0:
+            message = payload.get("message", "unknown error")
+            sid = payload.get("sid")
+            await self.push_error(f"Xfyun ASR error code={code}, sid={sid}, message={message}")
+            return
+
+        data = payload.get("data")
+        if not isinstance(data, dict):
+            return
+
+        is_final_response = data.get("status") == 2
+        recognition = data.get("result")
+        if isinstance(recognition, dict):
+            text = self._apply_recognition_result(recognition)
+            if text and text != self._last_text:
+                self._last_text = text
+                if not self._finalizing_turn and not is_final_response:
+                    await self.push_frame(
+                        InterimTranscriptionFrame(
+                            text,
+                            self._user_id,
+                            time_now_iso8601(),
+                            _language_or_none(self._language),
+                            result=payload,
+                        )
+                    )
+
+        if is_final_response:
+            final_text = self._last_text
+            if final_text:
+                self.confirm_finalize()
+                await self.push_frame(
+                    TranscriptionFrame(
+                        final_text,
+                        self._user_id,
+                        time_now_iso8601(),
+                        _language_or_none(self._language),
+                        result=payload,
+                    )
+                )
+            await self._close_utterance()
+
+    def _apply_recognition_result(self, recognition: dict[str, Any]) -> str:
+        partial = _extract_text_from_result(recognition)
+        if not partial:
+            return self._last_text
+
+        if self._dynamic_correction and recognition.get("pgs") == "rpl" and recognition.get("rg"):
+            start, end = recognition["rg"]
+            if 1 <= start <= len(self._partials):
+                self._partials[start - 1 : end] = [partial]
+            else:
+                logger.debug(f"Ignoring out-of-range Xfyun replacement rg={recognition['rg']}")
+        else:
+            self._partials.append(partial)
+
+        return "".join(self._partials)
+
+
+def _extract_text_from_result(result: dict[str, Any]) -> str:
+    words: list[str] = []
+    for item in result.get("ws", []):
+        for candidate in item.get("cw", []):
+            word = candidate.get("w")
+            if word:
+                words.append(word)
+    return "".join(words)
+
+
+def _build_auth_url(url: str, api_key: str, api_secret: str) -> str:
+    parsed = urlparse(url)
+    host = parsed.netloc
+    path = parsed.path or "/v2/iat"
+    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}"
+
+
+def _language_or_none(value: str) -> Language | None:
+    try:
+        return Language(value)
+    except ValueError:
+        return None

src/voice/xfyun_tts.py

@@ -0,0 +1,257 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import os
+import re
+import unicodedata
+from collections.abc import AsyncGenerator, AsyncIterator
+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 ErrorFrame, Frame
+from pipecat.services.settings import TTSSettings
+from pipecat.services.tts_service import TTSService
+from websockets.asyncio.client import connect
+
+
+DEFAULT_XFYUN_TTS_URL = "wss://tts-api.xfyun.cn/v2/tts"
+
+# Strip characters Xfyun's online TTS cannot synthesize. The engine silently
+# rejects (or returns empty audio for) text containing emoji and other
+# non-BMP symbols, which surfaces as "request finished without audio data".
+_EMOJI_AND_SYMBOL_RE = re.compile(
+    "["
+    "\U0001F300-\U0001FAFF"  # misc pictographs, emoji, symbols, transport, etc.
+    "\U00002600-\U000027BF"  # misc symbols and dingbats
+    "\U0001F1E6-\U0001F1FF"  # regional indicators (flags)
+    "\uFE00-\uFE0F"           # variation selectors
+    "\u200D"                  # zero-width joiner
+    "]",
+    flags=re.UNICODE,
+)
+
+
+class XfyunTTSService(TTSService):
+    """iFlytek/Xfyun online TTS service for Pipecat.
+
+    Xfyun's API is not OpenAI-compatible. It uses a signed WebSocket URL,
+    receives one JSON request per synthesis, and streams text WebSocket
+    messages containing base64-encoded audio chunks. This service requests
+    raw PCM so the chunks can become Pipecat audio frames without MP3 decode.
+    """
+
+    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 = 16000,
+        encoding: str = "raw",
+        text_encoding: str = "UTF8",
+        speed: int = 50,
+        volume: int = 50,
+        pitch: int = 50,
+        timeout: float = 30.0,
+        **kwargs,
+    ) -> None:
+        super().__init__(
+            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_TTS_URL
+        self._source_sample_rate = source_sample_rate
+        self._encoding = encoding
+        self._text_encoding = text_encoding
+        self._speed = speed
+        self._volume = volume
+        self._pitch = pitch
+        self._timeout = timeout
+        self._last_failure_detail: str | None = None
+
+    async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]:
+        if not text:
+            return
+
+        if not self._app_id or not self._api_key or not self._api_secret:
+            yield ErrorFrame(error="Xfyun TTS requires app_id, api_key, and api_secret")
+            return
+
+        sanitized = _sanitize_text_for_tts(text)
+        if not sanitized:
+            logger.debug(
+                f"{self}: skipping Xfyun TTS, text became empty after sanitization "
+                f"(original={text!r})"
+            )
+            return
+
+        if sanitized != text:
+            logger.debug(
+                f"{self}: sanitized Xfyun TTS text "
+                f"(original={text!r}, sanitized={sanitized!r})"
+            )
+
+        if len(sanitized.encode("utf-8")) >= 8000:
+            yield ErrorFrame(error="Xfyun TTS text must be less than 8000 UTF-8 bytes")
+            return
+
+        if self._encoding != "raw":
+            yield ErrorFrame(error="Xfyun TTS is configured for PCM output; set aue/encoding to raw")
+            return
+
+        try:
+            await self.start_tts_usage_metrics(sanitized)
+
+            first_frame = True
+            async for frame in self._stream_audio_frames_from_iterator(
+                self._iter_audio_chunks(sanitized),
+                in_sample_rate=self._source_sample_rate,
+                context_id=context_id,
+            ):
+                if first_frame:
+                    await self.stop_ttfb_metrics()
+                    first_frame = False
+                yield frame
+
+            if first_frame:
+                detail = self._last_failure_detail or "no audio frames received"
+                yield ErrorFrame(
+                    error=(
+                        f"Xfyun TTS request finished without audio data ({detail}); "
+                        f"text={sanitized!r}"
+                    )
+                )
+        except Exception as exc:
+            yield ErrorFrame(error=f"Xfyun TTS request failed: {exc}")
+
+    async def _iter_audio_chunks(self, text: str) -> AsyncIterator[bytes]:
+        request = self._build_request_frame(text)
+        auth_url = _build_auth_url(self._url, self._api_key, self._api_secret)
+
+        self._last_failure_detail = None
+        frames_received = 0
+        audio_bytes_received = 0
+        last_status: int | None = None
+        last_sid: str | None = None
+        saw_status_2 = False
+
+        async with connect(auth_url, max_size=None, open_timeout=self._timeout) as websocket:
+            await websocket.send(json.dumps(request, ensure_ascii=False))
+
+            async for raw_message in websocket:
+                frames_received += 1
+                payload = json.loads(raw_message)
+                code = payload.get("code", -1)
+                sid = payload.get("sid")
+                if sid:
+                    last_sid = sid
+                if code != 0:
+                    err_msg = payload.get("message", "unknown error")
+                    raise RuntimeError(f"code={code}, sid={sid}, message={err_msg}")
+
+                data = payload.get("data")
+                if not isinstance(data, dict):
+                    continue
+
+                last_status = data.get("status", last_status)
+
+                audio_b64 = data.get("audio")
+                if audio_b64:
+                    audio_bytes = base64.b64decode(audio_b64)
+                    audio_bytes_received += len(audio_bytes)
+                    yield audio_bytes
+
+                if data.get("status") == 2:
+                    saw_status_2 = True
+                    break
+
+        if audio_bytes_received == 0:
+            self._last_failure_detail = (
+                f"frames={frames_received}, audio_bytes=0, "
+                f"last_status={last_status}, saw_status_2={saw_status_2}, sid={last_sid}"
+            )
+            logger.warning(
+                f"{self}: Xfyun TTS produced no audio ({self._last_failure_detail})"
+            )
+
+    def _build_request_frame(self, text: str) -> dict[str, Any]:
+        business: dict[str, Any] = {
+            "aue": self._encoding,
+            "auf": f"audio/L16;rate={self._source_sample_rate}",
+            "vcn": self._voice,
+            "speed": self._speed,
+            "volume": self._volume,
+            "pitch": self._pitch,
+            "tte": self._text_encoding,
+        }
+
+        return {
+            "common": {"app_id": self._app_id},
+            "business": business,
+            "data": {
+                "status": 2,
+                "text": base64.b64encode(text.encode("utf-8")).decode("utf-8"),
+            },
+        }
+
+
+def _sanitize_text_for_tts(text: str) -> str:
+    """Strip characters Xfyun's online TTS cannot synthesize.
+
+    The Xfyun ``/v2/tts`` engine silently drops or rejects emoji, pictographs,
+    dingbats, regional-indicator flags, variation selectors, and zero-width
+    joiners.  When such characters appear in the input the synthesis can
+    finish without any audio data ("Xfyun TTS request finished without audio
+    data").  We also drop control characters (other than common whitespace)
+    and "Symbol, Other" codepoints, then collapse runs of whitespace.
+    """
+    if not text:
+        return text
+
+    cleaned = _EMOJI_AND_SYMBOL_RE.sub("", text)
+    filtered: list[str] = []
+    for ch in cleaned:
+        category = unicodedata.category(ch)
+        if category == "So":
+            continue
+        if category.startswith("C") and ch not in ("\n", "\r", "\t"):
+            continue
+        filtered.append(ch)
+    return re.sub(r"\s+", " ", "".join(filtered)).strip()
+
+
+def _build_auth_url(url: str, api_key: str, api_secret: str) -> str:
+    parsed = urlparse(url)
+    host = parsed.netloc
+    path = parsed.path or "/v2/tts"
+    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}"