待优化

main/xiaozhi-server/config.yaml

@@ -718,6 +718,7 @@ TTS:
     #各参数意义见开发文档：https://tts.linkerai.top/docs#/default/text_to_speech_tts_get
     type: linkerai
     api_url: https://tts.linkerai.top/tts
+    audio_format: "pcm"
     access_token: "test"
     voice: "OUeAo1mhq6IBExi"
     output_dir: tmp/

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

@@ -331,10 +331,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)
+                if tts_file:
-                self.tts_audio_queue.put(
+                    audio_datas = self._process_audio_file(tts_file)
-                    (SentenceType.MIDDLE, audio_datas, segment_text)
+                    self.tts_audio_queue.put(
-                )
+                        (SentenceType.MIDDLE, audio_datas, segment_text)
-                self.processed_chars += len(full_text)
+                    )
+                    self.processed_chars += len(full_text)
                 return True
         return False

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

@@ -1,3 +1,8 @@
+import asyncio
+import traceback
+import queue
+import requests
+from pathlib import Path
 from core.providers.tts.base import TTSProviderBase
 from core.providers.tts.dto.dto import (
     TTSMessageDTO,
@@ -6,8 +11,7 @@ from core.providers.tts.dto.dto import (
     InterfaceType
 )
 from config.logger import setup_logging
-from core.utils import opus_encoder_utils
+from core.utils import opus_encoder_utils, textUtils
-import requests
 
 TAG = __name__
 logger = setup_logging()
@@ -20,8 +24,6 @@ class TTSProvider(TTSProviderBase):
         self.access_token = config.get("access_token")
         self.voice = config.get("voice")
         self.api_url = config.get("api_url")
-
-        # 根据配置选择音频格式，优先使用opus
         self.audio_format = config.get("audio_format", "opus")
 
         # 创建Opus编码器
@@ -31,15 +33,135 @@ class TTSProvider(TTSProviderBase):
             frame_size_ms=60
         )
 
-    async def text_to_speak(self, text, _):
+        # 添加文本缓冲区
-        """将文本转换为语音（流式）"""
+        self.text_buffer = ""
-        await self.send_text(self.voice, text)
+        # 句子结束标点集合
+        self.sentence_endings = ("。", "？", "！", "；", ":", ".", "?", "!", ";","……")
+        # 逗号类标点（用于第一句话分割）
+        self.comma_endings = ("，", "~", "、", ",", "。", ".", "？", "?", "！", "!", "；", ";", "：",)
+
+        # 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)
+                logger.bind(tag=TAG).debug(
+                    f"TTS任务｜{message.sentence_type.name}｜{message.content_type.name}"
+                )
+
+                if message.sentence_type == SentenceType.FIRST:
+                    # 初始化流式状态
+                    self.tts_audio_first_sentence = True
+                    self.pcm_buffer = bytearray()
+                    self.text_buffer = ""  # 重置文本缓冲区
+
+                elif ContentType.TEXT == message.content_type:
+                    # 将文本添加到缓冲区
+                    self.text_buffer += message.content_detail
+                    # 尝试分割并发送完整句子
+                    self._process_text_buffer()
+
+                elif ContentType.FILE == message.content_type:
+                    # 先处理缓冲区中的剩余文本
+                    self._flush_text_buffer()
+                    # 处理文件类型
+                    if message.content_file and Path(message.content_file).exists():
+                        audio_datas = self._process_audio_file(message.content_file)
+                        self.tts_audio_queue.put(
+                            (SentenceType.MIDDLE, audio_datas, message.content_detail)
+                        )
+
+                if message.sentence_type == SentenceType.LAST:
+                    # 处理缓冲区中的剩余文本
+                    self._flush_text_buffer()
+                    # 发送结束帧
+                    if self.pcm_buffer:
+                        opus_datas = self.wav_to_opus_data_audio_raw(self.pcm_buffer, is_end=True)
+                        self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas, ""))
+                        self.pcm_buffer = bytearray()
+                    self.tts_audio_queue.put((SentenceType.LAST, [], None))
+                    self.text_buffer = ""  # 重置文本缓冲区
+
+            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_text_buffer(self):
+        """处理文本缓冲区，分割并发送完整句子"""
+        while True:
+            # 查找最近的句子结束位置
+            end_pos = -1
+            for punct in self.sentence_endings:
+                pos = self.text_buffer.find(punct)
+                if pos != -1 and (end_pos == -1 or pos < end_pos):
+                    end_pos = pos
+
+            # 如果是第一句话，也允许在逗号处分隔
+            if self.tts_audio_first_sentence and end_pos == -1:
+                for punct in self.comma_endings:
+                    pos = self.text_buffer.find(punct)
+                    if pos != -1 and (end_pos == -1 or pos < end_pos):
+                        end_pos = pos
+
+            # 找到分割点
+            if end_pos != -1:
+                # 提取完整句子
+                sentence = self.text_buffer[:end_pos + 1]
+                sentence = textUtils.get_string_no_punctuation_or_emoji(sentence)
+
+                if not sentence.strip():  # 检查是否为空文本
+                    self.text_buffer = self.text_buffer[end_pos + 1:]
+                    continue
+
+                self.text_buffer = self.text_buffer[end_pos + 1:]
+
+                # 发送句子
+                future = asyncio.run_coroutine_threadsafe(
+                    self.text_to_speak(sentence),
+                    loop=self.conn.loop
+                )
+                future.result()
+
+                # 更新第一句话标志
+                if self.tts_audio_first_sentence:
+                    self.tts_audio_first_sentence = False
+            else:
+                break
+
+    def _flush_text_buffer(self):
+        """处理缓冲区中剩余的文本"""
+        if self.text_buffer:
+            clean_text = textUtils.get_string_no_punctuation_or_emoji(self.text_buffer)
+            if clean_text.strip():  # 检查是否为空文本
+                future = asyncio.run_coroutine_threadsafe(
+                    self.text_to_speak(clean_text),
+                    loop=self.conn.loop
+                )
+                future.result()
+            self.text_buffer = ""
+
+    async def text_to_speak(self, text):
+        # 发送文本
+        await self.send_text(text)
         return
 
-    async def send_text(self, speaker: str, text: str):
+    ###################################################################################
-        """向 TTS 服务发送文本并获取音频流"""
+    # linkerai单流式TTS重写父类的方法--结束
+    ###################################################################################
+
+    async def send_text(self, text: str):
+        """流式处理TTS音频"""
         try:
-            # 构造请求参数
             params = {
                 "tts_text": text,
                 "spk_id": self.voice,
@@ -48,48 +170,38 @@ class TTSProvider(TTSProviderBase):
                 "target_sr": 16000,
                 "audio_format": self.audio_format,
             }
+            headers = {"Authorization": f"Bearer {self.access_token}"}
 
-            # 构造请求头
+            with requests.get(self.api_url, params=params, headers=headers, stream=True) as response:
-            headers = {
+                if response.status_code != 200:
-                "Authorization": f"Bearer {self.access_token}",
+                    logger.error(f"TTS请求失败: {response.status_code}, {response.text}")
-            }
+                    return
 
-            # 发送流式请求
+                logger.debug(f"处理TTS文本: {text}")
-            response = requests.get(
-                self.api_url,
-                params=params,
-                headers=headers,
-                stream=True
-            )
 
-            # 检查响应状态
+                # 流式处理音频数据
-            if response.status_code != 200:
+                for chunk in response.iter_content(chunk_size=1024):
-                logger.bind(tag=TAG).error(f"TTS 请求失败: {response.status_code}, {response.text}")
+                    if chunk:
-                return
+                        # 实时编码并发送音频帧
+                        self.pcm_buffer.extend(chunk)
 
-            # 处理音频流
+                # 处理剩余缓冲区数据
-            audio_frames = []
+                if self.pcm_buffer:
-            for chunk in response.iter_content(chunk_size=1024):
+                    opus_datas = self.wav_to_opus_data_audio_raw(self.pcm_buffer, is_end=True)
-                if chunk:
+                    self.tts_audio_queue.put((SentenceType.MIDDLE, opus_datas, text))
-                    if self.audio_format == "pcm":
+                    self.pcm_buffer = bytearray()
-                        # 将PCM转换为Opus
-                        opus_frames = self.opus_encoder.encode_pcm_to_opus(chunk, False)
-                        audio_frames.extend(opus_frames)
-                    else:
-                        # 直接使用Opus帧
-                        audio_frames.append(chunk)
-
-            # 将音频帧放入队列
-            self.tts_audio_queue.put(
-                (SentenceType.MIDDLE, audio_frames, text)
-            )
 
         except Exception as e:
-            logger.bind(tag=TAG).error(f"TTS 流式处理异常：{str(e)}")
+            logger.error(f"TTS流式处理异常：{str(e)}")
             raise
 
+    # 保持原有方法
+    def wav_to_opus_data_audio_raw(self, raw_data_var, is_end=False):
+        opus_datas = self.opus_encoder.encode_pcm_to_opus(raw_data_var, is_end)
+        return opus_datas
+
     async def close(self):
-        """资源清理方法"""
+        """资源清理"""
         await super().close()
         if hasattr(self, "opus_encoder"):
             self.opus_encoder.close()