add：双向流tts结构改造，火山双向tts语音流式输入输出

main/xiaozhi-server/core/connection.py

@@ -46,6 +46,7 @@ class ConnectionHandler:
         self.session_id = None
         self.prompt = None
         self.welcome_msg = None
+        self.u_id = None
 
         # 客户端状态相关
         self.client_abort = False
@@ -159,9 +160,13 @@ class ConnectionHandler:
             # 异步初始化
             await self.loop.run_in_executor(None, self._initialize_components)
 
-            # tts 消化线程
-            tts_priority = threading.Thread(target=self._tts_priority_thread, daemon=True)
-            tts_priority.start()
+            # 音频播放 消化线程
+            self.stop_event.clear()
+            audio_play_priority = threading.Thread(target=self._audio_play_priority_thread, daemon=True)
+            audio_play_priority.start()
+
+            # 打开音频通道
+            await self.tts.open_audio_channels()
 
             try:
                 async for message in self.websocket:
@@ -325,6 +330,7 @@ class ConnectionHandler:
         function_arguments = ""
         content_arguments = ""
         uuid_str = str(uuid.uuid4()).replace("-", "")
+        self.u_id = uuid_str
         msg_type = None
         for response in llm_responses:
             content, tools_call = response
@@ -554,18 +560,8 @@ class ConnectionHandler:
             self.tts_first_text_index = text_index
         self.tts_last_text_index = text_index
 
-    async def init_and_reset_tts(self):
-        self.stop_event.set()
-        # 释放之前的tts语音监听：重置监听队列
-        await self.tts.reset()
-        # 音频播放 消化线程
-        self.stop_event.clear()
-        audio_play_priority = threading.Thread(target=self._audio_play_priority_thread, daemon=True)
-        audio_play_priority.start()
-
     async def close(self):
         """资源清理方法"""
-
         # 清理其他资源
         self.stop_event.set()
         self.executor.shutdown(wait=False)

main/xiaozhi-server/core/handle/sendAudioHandle.py

@@ -13,7 +13,9 @@ logger = setup_logging()
 
 
 async def sendAudioMessage(conn, ttsMessageDTO: TTSMessageDTO):
-    u_id = None
+    if ttsMessageDTO.u_id != conn.u_id:
+        logger.bind(tag=TAG).info(f"msg id:{ttsMessageDTO.u_id},不是当前对话，当前对话id：{conn.u_id}")
+        return
     # 发送句子开始消息
     if SentenceType.SENTENCE_START == ttsMessageDTO.sentence_type:
         logger.bind(tag=TAG).info(f"发送第一段语音: {ttsMessageDTO.tts_finish_text}")

main/xiaozhi-server/core/handle/textHandle.py

@@ -28,8 +28,6 @@ async def handleTextMessage(conn, message):
             if msg_json["state"] == "start":
                 conn.client_have_voice = True
                 conn.client_voice_stop = False
-                # 打断，开启了行的对话，如果之前有tts存在，销毁掉重新建立tts
-                await conn.init_and_reset_tts()
             elif msg_json["state"] == "stop":
                 conn.client_have_voice = True
                 conn.client_voice_stop = True

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

@@ -50,21 +50,9 @@ class TTSProviderBase(ABC):
         self.executor = ThreadPoolExecutor(max_workers=self.max_workers)
 
     async def open_audio_channels(self):
-        pass
-
-    async def reset(self):
-        try:
-            logger.bind(tag=TAG).info("说明开始了新的对话，重建tts监听")
-            await self.stop_listen_resource()
-            self.tts_text_queue = queue.Queue()
-            self.tts_audio_queue = queue.Queue()
-            # 启动tts_text_queue监听线程
-            self.stop_event.clear()
-            tts_priority = threading.Thread(target=self._tts_text_priority_thread, daemon=True)
-            tts_priority.start()
-        except Exception as e:
-            logger.bind(tag=TAG).error(f"Failed to process TTS text: {e}")
-            traceback.print_exc()
+        # 启动tts_text_queue监听线程
+        tts_priority = threading.Thread(target=self._tts_text_priority_thread, daemon=True)
+        tts_priority.start()
 
     async def stop_listen_resource(self):
         """资源清理方法"""

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

@@ -180,8 +180,8 @@ class TTSProvider(TTSProviderBase):
                         if len(chunk_total) % 2 == 0 and chunk_total[-2:] == b'\x00\x00':
                             audio = self._get_audio_from_tts(chunk_total)
                             audio_raw = audio_raw + audio.raw_data
-                            # 长度凑够2贞开始发送，60ms*4=240ms
-                            if len(audio_raw) >= 7680:
+                            # 长度凑够2贞开始发送，60ms*2=120ms
+                            if len(audio_raw) >= 3840:
                                 opus_datas = self.wav_to_opus_data_audio_raw(audio_raw)
                                 if index == 0:
                                     yield TTSMessageDTO(u_id=u_id, msg_type=MsgType.TTS_TEXT_RESPONSE,

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

@@ -152,7 +152,7 @@ class TTSProvider(TTSProviderBase):
         self.tts_text = ""
 
     async def open_audio_channels(self):
-        self.loop_tts = asyncio.new_event_loop()
+        await super().open_audio_channels()
         ws_header = {
             "X-Api-App-Key": self.appId,
             "X-Api-Access-Key": self.access_token,
@@ -316,6 +316,7 @@ class TTSProvider(TTSProviderBase):
 
     async def close(self):
         """资源清理方法"""
+        await self.finish_connection()
         await self.ws.close()
 
     async def text_to_speak(self, u_id, text, is_last_text=False, is_first_text=False):

main/xiaozhi-server/core/websocket_server.py

@@ -12,7 +12,7 @@ class WebSocketServer:
     def __init__(self, config: dict):
         self.config = config
         self.logger = setup_logging()
-        self._vad, self._asr, self._llm, self._memory, self.intent = self._create_processing_instances()
+        self._vad, self._asr, self._llm, self._tts, self._memory, self.intent = self._create_processing_instances()
         self.active_connections = set()  # 添加全局连接记录
 
     def _create_processing_instances(self):
@@ -41,6 +41,14 @@ class WebSocketServer:
                 self.config["LLM"][self.config["selected_module"]["LLM"]]['type'],
                 self.config["LLM"][self.config["selected_module"]["LLM"]],
             ),
+            tts.create_instance(
+                self.config["selected_module"]["TTS"]
+                if not 'type' in self.config["TTS"][self.config["selected_module"]["TTS"]]
+                else
+                self.config["TTS"][self.config["selected_module"]["TTS"]]["type"],
+                self.config["TTS"][self.config["selected_module"]["TTS"]],
+                self.config["delete_audio"]
+            ),
             memory.create_instance(memory_cls_name, memory_cfg),
             intent.create_instance(
                 self.config["selected_module"]["Intent"]
@@ -71,16 +79,7 @@ class WebSocketServer:
         """处理新连接，每次创建独立的ConnectionHandler"""
         # 创建ConnectionHandler时传入当前server实例
         # tts 变成链接的时候创建，避免并非问题
-        f_tts = tts.create_instance(
-            self.config["selected_module"]["TTS"]
-            if not 'type' in self.config["TTS"][self.config["selected_module"]["TTS"]]
-            else
-            self.config["TTS"][self.config["selected_module"]["TTS"]]["type"],
-            self.config["TTS"][self.config["selected_module"]["TTS"]],
-            self.config["delete_audio"]
-        )
-        await f_tts.open_audio_channels()
-        handler = ConnectionHandler(self.config, self._vad, self._asr, self._llm, f_tts, self._memory, self.intent)
+        handler = ConnectionHandler(self.config, self._vad, self._asr, self._llm, self._tts, self._memory, self.intent)
         self.active_connections.add(handler)
         try:
             await handler.handle_connection(websocket)