Merge pull request #1484 from xinnan-tech/py_linkerai_single

Py linkerai single
2026-07-22 07:03:53 +08:00 · 2025-06-05 16:13:21 +08:00
parent ca1beb956a 01eb416a03
commit 5a01bfcdbf
6 changed files with 283 additions and 16 deletions
@@ -0,0 +1,20 @@
+-- 增加LinkeraiTTS供应器和模型配置
+delete from `ai_model_provider` where id = 'SYSTEM_TTS_LinkeraiTTS';
+INSERT INTO `ai_model_provider` (`id`, `model_type`, `provider_code`, `name`, `fields`, `sort`, `creator`, `create_date`, `updater`, `update_date`) VALUES
+('SYSTEM_TTS_LinkeraiTTS', 'TTS', 'linkerai', 'Linkerai语音合成', '[{"key":"api_url","label":"API地址","type":"string"},{"key":"audio_format","label":"音频格式","type":"string"},{"key":"access_token","label":"访问令牌","type":"string"},{"key":"voice","label":"默认音色","type":"string"}]', 14, 1, NOW(), 1, NOW());
+
+delete from `ai_model_config` where id = 'TTS_LinkeraiTTS';
+INSERT INTO `ai_model_config` VALUES ('TTS_LinkeraiTTS', 'TTS', 'LinkeraiTTS', 'Linkerai语音合成', 0, 1, '{\"type\": \"linkerai\", \"api_url\": \"https://tts.linkerai.cn/tts\", \"audio_format\": \"pcm\", \"access_token\": \"U4YdYXVfpwWnk2t5Gp822zWPCuORyeJL\", \"voice\": \"OUeAo1mhq6IBExi\"}', NULL, NULL, 17, NULL, NULL, NULL, NULL);
+
+-- LinkeraiTTS模型配置说明文档
+UPDATE `ai_model_config` SET 
+`doc_link` = 'https://tts.linkerai.cn/docs',
+`remark` = 'Linkerai语音合成服务配置说明：
+1. 访问 https://linkerai.cn 注册并获取访问令牌
+2. 默认的access_token供测试使用，请勿用于商业用途
+3. 支持声音克隆功能，可自行上传音频，填入voice参数
+4. 如果voice参数为空，将使用默认声音' WHERE `id` = 'TTS_LinkeraiTTS';
+
+
+delete from `ai_tts_voice` where tts_model_id = 'TTS_LinkeraiTTS';
+INSERT INTO `ai_tts_voice` VALUES ('TTS_LinkeraiTTS_0001', 'TTS_LinkeraiTTS', '芷若', 'OUeAo1mhq6IBExi', '中文', NULL, NULL, 1, NULL, NULL, NULL, NULL);
@@ -190,4 +190,11 @@ databaseChangeLog:
      changes:
        - sqlFile:
            encoding: utf8
-            path: classpath:db/changelog/202506032232.sql
+            path: classpath:db/changelog/202506032232.sql
+  - changeSet:
+      id: 202506051538
+      author: hrz
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: classpath:db/changelog/202506051538.sql
@@ -324,7 +324,7 @@ VAD:
    type: silero
    threshold: 0.5
    model_dir: models/snakers4_silero-vad
-    min_silence_duration_ms: 700  # 如果说话停顿比较长，可以把这个值设置大一些
+    min_silence_duration_ms: 200  # 如果说话停顿比较长，可以把这个值设置大一些

 LLM:
  # 所有openai类型均可以修改超参，以AliLLM为例
@@ -755,4 +755,15 @@ TTS:
    headers: # 自定义请求头
      # Authorization: Bearer xxxx
    format: mp3 # 接口返回的音频格式
+    output_dir: tmp/
+  LinkeraiTTS:
+    type: linkerai
+    api_url: https://tts.linkerai.cn/tts
+    audio_format: "pcm"
+    # 默认的access_token供大家测试时免费使用的，此access_token请勿用于商业用途
+    # 如果效果不错，可自行申请token，申请地址：https://linkerai.cn
+    # 各参数意义见开发文档：https://tts.linkerai.cn/docs
+    # 支持声音克隆，可自行上传音频，填入voice参数，voice参数为空时，使用默认声音
+    access_token: "U4YdYXVfpwWnk2t5Gp822zWPCuORyeJL"
+    voice: "OUeAo1mhq6IBExi"
    output_dir: tmp/
@@ -37,6 +37,7 @@ class TTSProviderBase(ABC):
        self.tts_text_queue = queue.Queue()
        self.tts_audio_queue = queue.Queue()
        self.tts_audio_first_sentence = True
+        self.before_stop_play_files = []

        self.tts_text_buff = []
        self.punctuations = (
@@ -324,6 +325,14 @@ class TTSProviderBase(ABC):
            os.remove(tts_file)
        return audio_datas

+    def _process_before_stop_play_files(self):
+        for tts_file, text in self.before_stop_play_files:
+            if tts_file and os.path.exists(tts_file):
+                audio_datas = self._process_audio_file(tts_file)
+                self.tts_audio_queue.put((SentenceType.MIDDLE, audio_datas, text))
+        self.before_stop_play_files.clear()
+        self.tts_audio_queue.put((SentenceType.LAST, [], None))
+
    def _process_remaining_text(self):
        """处理剩余的文本并生成语音

@@ -336,10 +345,11 @@ class TTSProviderBase(ABC):
            segment_text = textUtils.get_string_no_punctuation_or_emoji(remaining_text)
            if segment_text:
                tts_file = self.to_tts(segment_text)
-                audio_datas = self._process_audio_file(tts_file)
-                self.tts_audio_queue.put(
-                    (SentenceType.MIDDLE, audio_datas, segment_text)
-                )
-                self.processed_chars += len(full_text)
+                if tts_file:
+                    audio_datas = self._process_audio_file(tts_file)
+                    self.tts_audio_queue.put(
+                        (SentenceType.MIDDLE, audio_datas, segment_text)
+                    )
+                    self.processed_chars += len(full_text)
                return True
        return False
@@ -154,7 +154,6 @@ class TTSProvider(TTSProviderBase):
        self.header = {"Authorization": f"{self.authorization}{self.access_token}"}
        self.enable_two_way = True
        self.tts_text = ""
-        self.before_stop_play_files = []
        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
            sample_rate=16000, channels=1, frame_size_ms=60
        )
@@ -432,14 +431,7 @@ class TTSProvider(TTSProviderBase):
                        is_first_sentence = False
                    elif res.optional.event == EVENT_SessionFinished:
                        logger.bind(tag=TAG).debug(f"会话结束～～")
-                        for tts_file, text in self.before_stop_play_files:
-                            if tts_file and os.path.exists(tts_file):
-                                audio_datas = self._process_audio_file(tts_file)
-                                self.tts_audio_queue.put(
-                                    (SentenceType.MIDDLE, audio_datas, text)
-                                )
-                        self.before_stop_play_files.clear()
-                        self.tts_audio_queue.put((SentenceType.LAST, [], None))
+                        self._process_before_stop_play_files()
                        break
                except websockets.ConnectionClosed:
                    logger.bind(tag=TAG).warning("WebSocket连接已关闭")
@@ -0,0 +1,227 @@
+import queue
+import asyncio
+import traceback
+import aiohttp
+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.access_token = config.get("access_token")
+        self.voice = config.get("voice")
+        self.api_url = config.get("api_url")
+        self.audio_format = "pcm"
+        self.before_stop_play_files = []
+        self.segment_count = 0  # 添加片段计数器
+
+        # 创建Opus编码器
+        self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
+            sample_rate=16000, channels=1, frame_size_ms=60
+        )
+
+        # 添加文本缓冲区
+        self.text_buffer = ""
+
+        # PCM缓冲区
+        self.pcm_buffer = bytearray()
+
+    ###################################################################################
+    # linkerai单流式TTS重写父类的方法--开始
+    ###################################################################################
+
+    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.tts_audio_first_sentence = True
+                    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(segment_text)
+
+                elif ContentType.FILE == message.content_type:
+                    logger.bind(tag=TAG).info(
+                        f"添加音频文件到待播放列表: {message.content_file}"
+                    )
+                    self.before_stop_play_files.append(
+                        (message.content_file, 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(segment_text, is_last)
+                self.processed_chars += len(full_text)
+            else:
+                self._process_before_stop_play_files()
+
+    def to_tts(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
+
+    ###################################################################################
+    # linkerai单流式TTS重写父类的方法--结束
+    ###################################################################################
+
+    async def text_to_speak(self, text, is_last):
+        """流式处理TTS音频，每句只推送一次音频列表"""
+        await self._tts_request(text, is_last)
+
+    async def close(self):
+        """资源清理"""
+        await super().close()
+        if hasattr(self, "opus_encoder"):
+            self.opus_encoder.close()
+
+    async def _tts_request(self, text: str, is_last: bool) -> None:
+        params = {
+            "tts_text": text,
+            "spk_id": self.voice,
+            "frame_durition": 60,
+            "stream": "true",
+            "target_sr": 16000,
+            "audio_format": "pcm",
+            "instruct_text": "请生成一段自然流畅的语音",
+        }
+        headers = {
+            "Authorization": f"Bearer {self.access_token}",
+            "Content-Type": "application/json",
+        }
+
+        # 一帧 PCM 所需字节数：60 ms &times; 16 kHz &times; 1 ch &times; 2 B = 1 920
+        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.get(
+                    self.api_url, params=params, headers=headers, timeout=10
+                ) as resp:
+
+                    if resp.status != 200:
+                        logger.error(f"TTS请求失败: {resp.status}, {await resp.text()}")
+                        self.tts_audio_queue.put((SentenceType.LAST, [], None))
+                        return
+
+                    self.pcm_buffer.clear()
+                    opus_datas_cache = []
+
+                    # 兼容 iter_chunked / iter_chunks / iter_any
+                    async for chunk in resp.content.iter_any():
+                        data = chunk[0] if isinstance(chunk, (list, tuple)) else chunk
+                        if not data:
+                            continue
+
+                        # 拼到 buffer
+                        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, text)
+                                    )
+                                    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, text)
+                                )
+                                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, text)
+                        )
+
+                    # 如果是最后一段，输出音频获取完毕
+                    if is_last:
+                        self._process_before_stop_play_files()
+
+        except Exception as e:
+            logger.error(f"TTS请求异常: {e}")
+            self.tts_audio_queue.put((SentenceType.LAST, [], None))