update:优化流式tts

main/xiaozhi-server/core/connection.py

@@ -315,7 +315,9 @@ class ConnectionHandler:
             self.asr = self._asr
         if self.tts is None:
             self.tts = self._tts
-        self.tts.startSession(self)
+        # 使用事件循环运行异步方法
+        asyncio.run_coroutine_threadsafe(self.tts.open_audio_channels(self), self.loop)
+
         """加载记忆"""
         self._initialize_memory()
         """加载意图识别"""

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

@@ -4,6 +4,7 @@ import queue
 import os
 import json
 import threading
+from enum import Enum
 from core.utils import p3
 from core.handle.sendAudioHandle import sendAudioMessage
 from core.handle.reportHandle import enqueue_tts_report
@@ -16,6 +17,14 @@ TAG = __name__
 logger = setup_logging()
 
 
+class TTSImplementationType(Enum):
+    """TTS实现类型枚举"""
+
+    NON_STREAMING = "non_streaming"  # 非流式实现
+    SINGLE_STREAMING = "single_streaming"  # 单流式实现
+    DOUBLE_STREAMING = "double_streaming"  # 双流式实现
+
+
 class TTSProviderBase(ABC):
     def __init__(self, config, delete_audio_file):
         self.conn = None
@@ -27,12 +36,20 @@ class TTSProviderBase(ABC):
         self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
             sample_rate=16000, channels=1, frame_size_ms=60
         )
+        # 添加实现类型属性，默认为非流式
+        self.interface_type = TTSImplementationType.NON_STREAMING
 
     @abstractmethod
     def generate_filename(self):
         pass
 
     def to_tts(self, text):
+        """如果是流式实现，一般没有文件生成，我们返回枚举值"""
+        if self.interface_type != TTSImplementationType.SINGLE_STREAMING:
+            asyncio.run(self.text_to_speak(text, None))
+            return self.interface_type.value
+
+        """以下是非流式实现，会返回文件"""
         tmp_file = self.generate_filename()
         try:
             max_repeat_time = 5
@@ -75,7 +92,7 @@ class TTSProviderBase(ABC):
         """音频文件转换为Opus编码"""
         return audio_to_data(audio_file_path, is_opus=True)
 
-    def startSession(self, conn):
+    async def open_audio_channels(self, conn):
         self.conn = conn
         self.tts_timeout = conn.config.get("tts_timeout", 10)
         # tts 消化线程
@@ -110,10 +127,18 @@ class TTSProviderBase(ABC):
                 try:
                     logger.bind(tag=TAG).debug("正在处理TTS任务...")
                     tts_file, text, _ = future.result(timeout=self.tts_timeout)
+
+                    # 如果tts_file返回流式标识，则不继续处理
                     if tts_file is None:
                         logger.bind(tag=TAG).error(
                             f"TTS出错： file is empty: {text_index}: {text}"
                         )
+                    elif (
+                        tts_file == TTSImplementationType.SINGLE_STREAMING.value
+                        or tts_file == TTSImplementationType.DOUBLE_STREAMING.value
+                    ):
+                        logger.bind(tag=TAG).debug(f"TTS生成：流式标识: {tts_file}")
+                        continue
                     else:
                         logger.bind(tag=TAG).debug(f"TTS生成：文件路径: {tts_file}")
                         if os.path.exists(tts_file):
@@ -134,7 +159,21 @@ class TTSProviderBase(ABC):
                 except TimeoutError:
                     logger.bind(tag=TAG).error("TTS超时")
                 except Exception as e:
-                    logger.bind(tag=TAG).error(f"TTS出错: {e}")
+                    import traceback
+
+                    error_info = {
+                        "error_type": type(e).__name__,
+                        "error_message": str(e),
+                        "stack_trace": traceback.format_exc(),
+                        "text_index": text_index,
+                        "text": text,
+                        "tts_file": tts_file,
+                        "audio_format": getattr(self.conn, "audio_format", None),
+                        "interface_type": self.interface_type.value,
+                    }
+                    logger.bind(tag=TAG).error(
+                        f"TTS处理出错: {json.dumps(error_info, ensure_ascii=False)}"
+                    )
                 if not self.conn.client_abort:
                     # 如果没有中途打断就发送语音
                     self.audio_play_queue.put((audio_datas, text, text_index))
@@ -154,7 +193,7 @@ class TTSProviderBase(ABC):
                             {
                                 "type": "tts",
                                 "state": "stop",
-                                "session_id": self.session_id,
+                                "session_id": self.conn.session_id,
                             }
                         )
                     ),
@@ -185,3 +224,14 @@ class TTSProviderBase(ABC):
     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 start_session(self, session_id):
+        pass
+
+    async def finish_session(self, session_id):
+        pass
+
+    async def close(self):
+        """资源清理方法"""
+        if hasattr(self, "ws") and self.ws:
+            await self.ws.close()

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

@@ -1,5 +1,4 @@
 import asyncio
-import io
 import os
 import threading
 import traceback
@@ -10,7 +9,7 @@ from datetime import datetime
 import websockets
 
 from config.logger import setup_logging
-from core.providers.tts.base import TTSProviderBase
+from core.providers.tts.base import TTSProviderBase, TTSImplementationType
 
 TAG = __name__
 logger = setup_logging()
@@ -151,34 +150,24 @@ class TTSProvider(TTSProviderBase):
         self.enable_two_way = True
         self.start_connection_flag = False
         self.tts_text = ""
+        self.interface_type = TTSImplementationType.DOUBLE_STREAMING
 
-    def startSession(self, conn):
+    async def open_audio_channels(self, conn):
-        self.conn = conn
+        await super().open_audio_channels(conn)
-        self.tts_timeout = conn.config.get("tts_timeout", 10)
-        # tts 消化线程
-        self.tts_priority_thread = threading.Thread(
-            target=self._tts_priority_thread, daemon=True
-        )
-        self.tts_priority_thread.start()
-
-        # 音频播放 消化线程
-        self.audio_play_priority_thread = threading.Thread(
-            target=self._audio_play_priority_thread, daemon=True
-        )
-        self.audio_play_priority_thread.start()
         ws_header = {
             "X-Api-App-Key": self.appId,
             "X-Api-Access-Key": self.access_token,
             "X-Api-Resource-Id": self.resource_id,
             "X-Api-Connect-Id": uuid.uuid4(),
         }
-        self.ws = websockets.connect(
+        self.ws = await websockets.connect(
             self.ws_url, additional_headers=ws_header, max_size=1000000000
         )
         tts_priority = threading.Thread(
             target=self._start_monitor_tts_response_thread(), daemon=True
         )
         tts_priority.start()
+        await self.start_session(conn.session_id)
 
     def generate_filename(self, extension=".wav"):
         return os.path.join(
@@ -381,7 +370,7 @@ class TTSProvider(TTSProviderBase):
 
     async def _start_monitor_tts_response(self):
         chunk_total = b""
-        while not self.stop_event.is_set():
+        while not self.conn.stop_event.is_set():
             try:
                 msg = await self.ws.recv()  # 确保 `recv()` 运行在同一个 event loop
                 res = self.parser_response(msg)