Merge pull request #2004 from xinnan-tech/py_IndexStreamTTS

Py index stream tts

main/manager-api/src/main/resources/db/changelog/202508081701.sql

@@ -0,0 +1,32 @@
+-- 添加Index-TTS-vLLM流式TTS供应器
+delete from `ai_model_provider` where id = 'SYSTEM_TTS_IndexStreamTTS';
+INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
+('SYSTEM_TTS_IndexStreamTTS', 'TTS', 'index_stream', 'Index-TTS-vLLM流式语音合成', '[{"key":"api_url","label":"API服务地址","type":"string"},{"key":"voice","label":"默认音色","type":"string"},{"key":"audio_format","label":"音频格式","type":"string"},{"key":"output_dir","label":"输出目录","type":"string"}]', 16, 1, NOW(), 1, NOW());
+
+-- 添加Index-TTS-vLLM流式TTS模型配置
+delete from `ai_model_config` where id = 'TTS_IndexStreamTTS';
+INSERT INTO `ai_model_config` VALUES ('TTS_IndexStreamTTS', 'TTS', 'IndexStreamTTS', 'Index-TTS-vLLM流式语音合成', 0, 1, '{\"type\": \"index_stream\", \"api_url\": \"http://127.0.0.1:11996/tts\", \"voice\": \"jay_klee\", \"audio_format\": \"pcm\", \"output_dir\": \"tmp/\"}', NULL, NULL, 19, NULL, NULL, NULL, NULL);
+
+-- 更新Index-TTS-vLLM流式TTS配置说明
+UPDATE `ai_model_config` SET 
+`doc_link` = 'https://github.com/Ksuriuri/index-tts-vllm',
+`remark` = 'Index-TTS-vLLM流式TTS配置说明：
+1. Index-TTS-vLLM是基于Index-TTS项目的vLLM推理服务，提供流式语音合成功能
+2. 支持多种音色，音质自然，适合各种语音交互场景
+3. 需要先部署Index-TTS-vLLM服务，然后配置API地址
+4. 支持实时流式合成，具有较低的延迟
+5. 支持自定义音色，可在项目assets文件夹下注册新音色
+部署步骤：
+1. 克隆项目：git clone https://github.com/Ksuriuri/index-tts-vllm.git
+2. 安装依赖：pip install -r requirements.txt
+3. 启动服务：python app.py
+4. 服务默认运行在 http://127.0.0.1:11996
+5. 如需其他音色，可到项目assets文件夹下注册
+6. 支持多种音频格式：pcm、wav、mp3等
+如需了解更多配置，请参考：https://github.com/Ksuriuri/index-tts-vllm/blob/master/README.md
+' WHERE `id` = 'TTS_IndexStreamTTS';
+
+-- 添加Index-TTS-vLLM流式TTS音色
+delete from `ai_tts_voice` where tts_model_id = 'TTS_IndexStreamTTS';
+-- 默认音色
+INSERT INTO `ai_tts_voice` VALUES ('TTS_IndexStreamTTS_0001', 'TTS_IndexStreamTTS', 'Jay Klee', 'jay_klee', '中文及中英文混合', NULL, NULL, NULL, NULL, 1, NULL, NULL, NULL, NULL);

main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml

@@ -282,3 +282,10 @@ databaseChangeLog:
         - sqlFile:
             encoding: utf8
             path: classpath:db/changelog/202507081646.sql
+  - changeSet:
+      id: 202508081701
+      author: hrz
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: classpath:db/changelog/202508081701.sql

main/xiaozhi-server/config.yaml

@@ -899,4 +899,12 @@ TTS:
     speed: 1.0  # 语速，1.0 表示正常语速，>1 表示加快，<1 表示减慢
     volume: 1.0  # 音量，1.0 表示正常音量，>1 表示增大，<1 表示减小
     save_path: ./streaming_tts.wav  # 服务器生成的语音文件保存路径
-
+  IndexStreamTTS:
+    # 基于Index-TTS-vLLM项目的TTS接口服务
+    # 参照教程：https://github.com/Ksuriuri/index-tts-vllm/blob/master/README.md
+    type: index_stream
+    api_url: http://127.0.0.1:11996/tts
+    audio_format: "pcm"
+    # 默认音色，如需其他音色可到项目assets文件夹下注册
+    voice: "jay_klee"
+    output_dir: tmp/

main/xiaozhi-server/core/providers/tts/index_stream.py

@@ -0,0 +1,264 @@
+import os
+import queue
+import asyncio
+import traceback
+import aiohttp
+import requests
+import time
+from config.logger import setup_logging
+from core.utils.tts import MarkdownCleaner
+from core.providers.tts.base import TTSProviderBase
+from core.utils import opus_encoder_utils, textUtils
+from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
+
+TAG = __name__
+logger = setup_logging()
+
+
+class TTSProvider(TTSProviderBase):
+    def __init__(self, config, delete_audio_file):
+        super().__init__(config, delete_audio_file)
+        self.interface_type = InterfaceType.SINGLE_STREAM
+        self.voice = config.get("voice", "xiao_he")
+        if config.get("private_voice"):
+            self.voice = config.get("private_voice")
+        else:
+            self.voice = config.get("voice", "xiao_he")
+        self.api_url = config.get("api_url", "http://8.138.114.124:11996/tts")
+        self.audio_format = "pcm"
+        self.before_stop_play_files = []
+        self.segment_count = 0
+
+        # 创建Opus编码器 需注意接口返回的采样率为24000
+        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
+            sample_rate=24000, channels=1, frame_size_ms=60
+        )
+
+        # 文本缓冲区和PCM缓冲区
+        self.text_buffer = ""
+        self.pcm_buffer = bytearray()
+
+    def tts_text_priority_thread(self):
+        """流式文本处理线程"""
+        while not self.conn.stop_event.is_set():
+            try:
+                message = self.tts_text_queue.get(timeout=1)
+                if message.sentence_type == SentenceType.FIRST:
+                    # 初始化参数
+                    self.tts_stop_request = False
+                    self.processed_chars = 0
+                    self.tts_text_buff = []
+                    self.segment_count = 0
+                    self.before_stop_play_files.clear()
+                elif ContentType.TEXT == message.content_type:
+                    self.tts_text_buff.append(message.content_detail)
+                    segment_text = self._get_segment_text()
+                    if segment_text:
+                        self.to_tts_single_stream(segment_text)
+
+                elif ContentType.FILE == message.content_type:
+                    logger.bind(tag=TAG).info(
+                        f"添加音频文件到待播放列表: {message.content_file}"
+                    )
+                    if message.content_file and os.path.exists(message.content_file):
+                        # 先处理文件音频数据
+                        file_audio = self._process_audio_file(message.content_file)
+                        self.before_stop_play_files.append(
+                            (file_audio, message.content_detail)
+                        )
+
+                if message.sentence_type == SentenceType.LAST:
+                    # 处理剩余的文本
+                    self._process_remaining_text(True)
+
+            except queue.Empty:
+                continue
+            except Exception as e:
+                logger.bind(tag=TAG).error(
+                    f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}"
+                )
+
+    def _process_remaining_text(self, is_last=False):
+        """处理剩余的文本并生成语音
+        Returns:
+            bool: 是否成功处理了文本
+        """
+        full_text = "".join(self.tts_text_buff)
+        remaining_text = full_text[self.processed_chars :]
+        if remaining_text:
+            segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text)
+            if segment_text:
+                self.to_tts_single_stream(segment_text, is_last)
+                self.processed_chars += len(full_text)
+            else:
+                self._process_before_stop_play_files()
+        else:
+            self._process_before_stop_play_files()
+
+    def to_tts_single_stream(self, text, is_last=False):
+        try:
+            max_repeat_time = 5
+            text = MarkdownCleaner.clean_markdown(text)
+            try:
+                asyncio.run(self.text_to_speak(text, is_last))
+            except Exception as e:
+                logger.bind(tag=TAG).warning(
+                    f"语音生成失败{5 - max_repeat_time + 1}次: {text}，错误: {e}"
+                )
+                max_repeat_time -= 1
+
+            if max_repeat_time > 0:
+                logger.bind(tag=TAG).info(
+                    f"语音生成成功: {text}，重试{5 - max_repeat_time}次"
+                )
+            else:
+                logger.bind(tag=TAG).error(
+                    f"语音生成失败: {text}，请检查网络或服务是否正常"
+                )
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}")
+        finally:
+            return None
+
+    async def text_to_speak(self, text, is_last):
+        """流式处理TTS音频，每句只推送一次音频列表"""
+        payload = {"text": text, "character": self.voice}
+
+        frame_bytes = int(
+            self.opus_encoder.sample_rate
+            * self.opus_encoder.channels  # 1
+            * self.opus_encoder.frame_size_ms
+            / 1000
+            * 2
+        )  # 16-bit = 2 bytes
+        try:
+            async with aiohttp.ClientSession() as session:
+                async with session.post(self.api_url, json=payload, timeout=10) as resp:
+
+                    if resp.status != 200:
+                        logger.bind(tag=TAG).error(
+                            f"TTS请求失败: {resp.status}, {await resp.text()}"
+                        )
+                        self.tts_audio_queue.put((SentenceType.LAST, [], None))
+                        return
+
+                    self.pcm_buffer.clear()
+                    opus_datas_cache = []
+
+                    self.tts_audio_queue.put((SentenceType.FIRST, [], text))
+
+                    # 处理音频流数据
+                    async for chunk in resp.content.iter_any():
+                        data = chunk[0] if isinstance(chunk, (list, tuple)) else chunk
+                        if not data:
+                            continue
+
+                        self.pcm_buffer.extend(data)
+
+                        while len(self.pcm_buffer) >= frame_bytes:
+                            frame = bytes(self.pcm_buffer[:frame_bytes])
+                            del self.pcm_buffer[:frame_bytes]
+                            opus = self.opus_encoder.encode_pcm_to_opus(
+                                frame, end_of_stream=False
+                            )
+                            if opus:
+                                if self.segment_count < 10:  # 前10个片段直接发送
+                                    self.tts_audio_queue.put(
+                                        (SentenceType.MIDDLE, opus, None)
+                                    )
+                                    self.segment_count += 1
+                                else:
+                                    opus_datas_cache.extend(opus)
+
+                    # flush 剩余不足一帧的数据
+                    if self.pcm_buffer:
+                        opus = self.opus_encoder.encode_pcm_to_opus(
+                            bytes(self.pcm_buffer), end_of_stream=True
+                        )
+                        if opus:
+                            if self.segment_count < 10:  # 前10个片段直接发送
+                                # 直接发送
+                                self.tts_audio_queue.put(
+                                    (SentenceType.MIDDLE, opus, None)
+                                )
+                                self.segment_count += 1
+                            else:
+                                # 后续片段缓存
+                                opus_datas_cache.extend(opus)
+                        self.pcm_buffer.clear()
+
+                    # 如果不是前10个片段，发送缓存的数据
+                    if self.segment_count >= 10 and opus_datas_cache:
+                        self.tts_audio_queue.put(
+                            (SentenceType.MIDDLE, opus_datas_cache, None)
+                        )
+
+                    # 如果是最后一段，输出音频获取完毕
+                    if is_last:
+                        self._process_before_stop_play_files()
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
+            self.tts_audio_queue.put((SentenceType.LAST, [], None))
+
+    async def close(self):
+        """资源清理"""
+        await super().close()
+        if hasattr(self, "opus_encoder"):
+            self.opus_encoder.close()
+
+    def to_tts(self, text: str) -> list:
+        """非流式TTS处理，用于测试及保存音频文件的场景
+
+        Args:
+            text: 要转换的文本
+
+        Returns:
+            list: 返回opus编码后的音频数据列表
+        """
+        start_time = time.time()
+        text = MarkdownCleaner.clean_markdown(text)
+
+        payload = {"text": text, "character": self.character}
+
+        try:
+            with requests.post(self.api_url, json=payload, timeout=5) as response:
+                if response.status_code != 200:
+                    logger.bind(tag=TAG).error(
+                        f"TTS请求失败: {response.status_code}, {response.text}"
+                    )
+                    return []
+
+                logger.info(f"TTS请求成功: {text}, 耗时: {time.time() - start_time}秒")
+
+                # 使用opus编码器处理PCM数据
+                opus_datas = []
+                pcm_data = response.content
+
+                # 计算每帧的字节数
+                frame_bytes = int(
+                    self.opus_encoder.sample_rate
+                    * self.opus_encoder.channels
+                    * self.opus_encoder.frame_size_ms
+                    / 1000
+                    * 2
+                )
+
+                # 分帧处理PCM数据
+                for i in range(0, len(pcm_data), frame_bytes):
+                    frame = pcm_data[i : i + frame_bytes]
+                    if len(frame) < frame_bytes:
+                        # 最后一帧可能不足，用0填充
+                        frame = frame + b"\x00" * (frame_bytes - len(frame))
+
+                    opus = self.opus_encoder.encode_pcm_to_opus(
+                        frame, end_of_stream=(i + frame_bytes >= len(pcm_data))
+                    )
+                    if opus:
+                        opus_datas.extend(opus)
+
+                return opus_datas
+
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"TTS请求异常: {e}")
+            return []