From e08eb48d9a0860ea0c9769720351037e5191d694 Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Thu, 9 Apr 2026 10:46:46 +0800 Subject: [PATCH 1/7] =?UTF-8?q?refactor:=20=E4=BC=98=E5=8C=96=E7=9B=B8?= =?UTF-8?q?=E5=85=B3=E6=89=93=E6=96=AD=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 29 ++++-- .../core/handle/intentHandler.py | 4 +- .../core/handle/receiveAudioHandle.py | 8 +- .../core/handle/sendAudioHandle.py | 7 +- .../core/providers/tts/alibl_stream.py | 95 +++++++++---------- .../core/providers/tts/aliyun_stream.py | 75 +++++++-------- .../xiaozhi-server/core/providers/tts/base.py | 43 ++++++++- .../providers/tts/huoshan_double_stream.py | 45 ++++----- .../core/providers/tts/xunfei_stream.py | 62 ++++++------ 9 files changed, 204 insertions(+), 164 deletions(-) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index d167bf1b..b64afd5d 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -838,20 +838,27 @@ class ConnectionHandler: self.dialogue.update_system_message(self.prompt) def chat(self, query, depth=0): + # 保存当前任务的sentence_id到局部变量，避免被新任务覆盖 + current_sentence_id = None + if query is not None: self.logger.bind(tag=TAG).info(f"大模型收到用户消息: {query}") # 为最顶层时新建会话ID和发送FIRST请求 if depth == 0: - self.sentence_id = str(uuid.uuid4().hex) + current_sentence_id = str(uuid.uuid4().hex) + self.sentence_id = current_sentence_id # 更新共享属性 self.dialogue.put(Message(role="user", content=query)) self.tts.tts_text_queue.put( TTSMessageDTO( - sentence_id=self.sentence_id, + sentence_id=current_sentence_id, sentence_type=SentenceType.FIRST, content_type=ContentType.ACTION, ) ) + else: + # 递归调用时，使用当前的sentence_id + current_sentence_id = self.sentence_id # 设置最大递归深度，避免无限循环，可根据实际需求调整 MAX_DEPTH = 5 @@ -976,7 +983,6 @@ class ConnectionHandler: # 支持多个并行工具调用 - 使用列表存储 tool_calls_list = [] # 格式: [{"id": "", "name": "", "arguments": ""}] content_arguments = "" - self.client_abort = False emotion_flag = True try: for response in llm_responses: @@ -1013,7 +1019,7 @@ class ConnectionHandler: response_message.append(content) self.tts.tts_text_queue.put( TTSMessageDTO( - sentence_id=self.sentence_id, + sentence_id=current_sentence_id, sentence_type=SentenceType.MIDDLE, content_type=ContentType.TEXT, content_detail=content, @@ -1023,7 +1029,7 @@ class ConnectionHandler: self.logger.bind(tag=TAG).error(f"LLM stream processing error: {e}") self.tts.tts_text_queue.put( TTSMessageDTO( - sentence_id=self.sentence_id, + sentence_id=current_sentence_id, sentence_type=SentenceType.MIDDLE, content_type=ContentType.TEXT, content_detail=get_system_error_response(self.config), @@ -1032,7 +1038,7 @@ class ConnectionHandler: if depth == 0: self.tts.tts_text_queue.put( TTSMessageDTO( - sentence_id=self.sentence_id, + sentence_id=current_sentence_id, sentence_type=SentenceType.LAST, content_type=ContentType.ACTION, ) @@ -1085,7 +1091,7 @@ class ConnectionHandler: # 如需要大模型先处理一轮，添加相关处理后的日志情况 if len(response_message) > 0: text_buff = "".join(response_message) - self.tts_MessageText = text_buff + self.tts.store_tts_text(current_sentence_id, text_buff) self.dialogue.put(Message(role="assistant", content=text_buff)) response_message.clear() @@ -1139,7 +1145,7 @@ class ConnectionHandler: # 存储对话内容 if len(response_message) > 0: text_buff = "".join(response_message) - self.tts_MessageText = text_buff + self.tts.store_tts_text(current_sentence_id, text_buff) self.dialogue.put(Message(role="assistant", content=text_buff)) # 更新工具调用统计：如果没有调用工具，增加计数 @@ -1149,7 +1155,7 @@ class ConnectionHandler: if depth == 0: self.tts.tts_text_queue.put( TTSMessageDTO( - sentence_id=self.sentence_id, + sentence_id=current_sentence_id, sentence_type=SentenceType.LAST, content_type=ContentType.ACTION, ) @@ -1282,7 +1288,10 @@ class ConnectionHandler: # 标记任务完成 self.report_queue.task_done() - def clearSpeakStatus(self): + def clearSpeakStatus(self, sentence_id=None): + # 如果sentence_id不匹配，说明是旧轮次的回调，不需要执行 + if sentence_id is not None and sentence_id != self.sentence_id: + return self.client_is_speaking = False self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态") diff --git a/main/xiaozhi-server/core/handle/intentHandler.py b/main/xiaozhi-server/core/handle/intentHandler.py index da0211e9..4d173259 100644 --- a/main/xiaozhi-server/core/handle/intentHandler.py +++ b/main/xiaozhi-server/core/handle/intentHandler.py @@ -219,8 +219,8 @@ async def process_intent_result( def speak_txt(conn: "ConnectionHandler", text): - # 记录文本 - conn.tts_MessageText = text + # 记录文本到 sentence_id 映射 + conn.tts.store_tts_text(conn.sentence_id, text) conn.tts.tts_text_queue.put( TTSMessageDTO( diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 280d72cf..cd53629b 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -26,10 +26,6 @@ async def handleAudioMessage(conn: "ConnectionHandler", audio): if not hasattr(conn, "vad_resume_task") or conn.vad_resume_task.done(): conn.vad_resume_task = asyncio.create_task(resume_vad_detection(conn)) return - # manual 模式下不打断正在播放的内容 - if have_voice: - if conn.client_is_speaking and conn.client_listen_mode != "manual": - await handleAbortMessage(conn) # 设备长时间空闲检测，用于say goodbye await no_voice_close_connect(conn, have_voice) # 接收音频 @@ -94,6 +90,10 @@ async def startToChat(conn: "ConnectionHandler", text): # 意图未被处理，继续常规聊天流程，使用实际文本内容 await send_stt_message(conn, actual_text) + + # 准备开始新会话 + conn.client_abort = False + conn.executor.submit(conn.chat, actual_text) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index d2b24f28..f62cc582 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -45,7 +45,6 @@ async def sendAudioMessage(conn: "ConnectionHandler", sentenceType, audios, text # 发送结束消息（如果是最后一个文本） if sentenceType == SentenceType.LAST: await send_tts_message(conn, "stop", None) - conn.client_is_speaking = False if conn.close_after_chat: await conn.close() @@ -270,6 +269,8 @@ async def send_tts_message(conn: "ConnectionHandler", state, text=None): # TTS播放结束 if state == "stop": + # 保存当前的 sentence_id，用于后续判断是否是当前轮次 + current_sentence_id = conn.sentence_id # 播放提示音 tts_notify = conn.config.get("enable_stop_tts_notify", False) if tts_notify: @@ -283,8 +284,8 @@ async def send_tts_message(conn: "ConnectionHandler", state, text=None): # 停止音频发送循环（仅在流控器已初始化时调用） if hasattr(conn, "audio_rate_controller") and conn.audio_rate_controller: conn.audio_rate_controller.stop_sending() - # 清除服务端讲话状态 - conn.clearSpeakStatus() + # 清除服务端讲话状态，传入 sentence_id 用于判断是否是当前轮次 + conn.clearSpeakStatus(current_sentence_id) # 发送消息到客户端 await conn.websocket.send(json.dumps(message)) diff --git a/main/xiaozhi-server/core/providers/tts/alibl_stream.py b/main/xiaozhi-server/core/providers/tts/alibl_stream.py index 5fb860f2..8bad4ee1 100644 --- a/main/xiaozhi-server/core/providers/tts/alibl_stream.py +++ b/main/xiaozhi-server/core/providers/tts/alibl_stream.py @@ -39,6 +39,7 @@ class TTSProvider(TTSProviderBase): self.ws_url = "wss://dashscope.aliyuncs.com/api-ws/v1/inference/" self.ws = None self._monitor_task = None + self.activate_session = False self.last_active_time = None # 模型和音色配置 @@ -75,9 +76,9 @@ class TTSProvider(TTSProviderBase): current_time = time.time() if self.ws and current_time - self.last_active_time < 60: # 一分钟内才可以复用链接进行连续对话 - logger.bind(tag=TAG).info(f"使用已有链接...") + logger.bind(tag=TAG).debug(f"使用已有链接...") return self.ws - logger.bind(tag=TAG).info("开始建立新连接...") + logger.bind(tag=TAG).debug("开始建立新连接...") self.ws = await websockets.connect( self.ws_url, @@ -87,7 +88,7 @@ class TTSProvider(TTSProviderBase): close_timeout=10, ) - logger.bind(tag=TAG).info("WebSocket连接建立成功") + logger.bind(tag=TAG).debug("WebSocket连接建立成功") self.last_active_time = current_time return self.ws except Exception as e: @@ -101,16 +102,22 @@ class TTSProvider(TTSProviderBase): 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} | 会话ID: {self.conn.sentence_id}" - ) - if message.sentence_type == SentenceType.FIRST: - self.conn.client_abort = False + # 过滤旧消息：检查sentence_id是否匹配 + if message.sentence_id != self.conn.sentence_id: + continue + + logger.bind(tag=TAG).debug( + f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" + ) if self.conn.client_abort: try: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") + asyncio.run_coroutine_threadsafe( + self.finish_session(self.conn.sentence_id), + loop=self.conn.loop, + ) continue except Exception as e: logger.bind(tag=TAG).error(f"取消TTS会话失败: {str(e)}") @@ -121,16 +128,16 @@ class TTSProvider(TTSProviderBase): try: if not getattr(self.conn, "sentence_id", None): self.conn.sentence_id = uuid.uuid4().hex - logger.bind(tag=TAG).info(f"自动生成新的 会话ID: {self.conn.sentence_id}") + logger.bind(tag=TAG).debug(f"自动生成新的 会话ID: {self.conn.sentence_id}") - logger.bind(tag=TAG).info("开始启动TTS会话...") + logger.bind(tag=TAG).debug("开始启动TTS会话...") future = asyncio.run_coroutine_threadsafe( self.start_session(self.conn.sentence_id), loop=self.conn.loop, ) future.result(timeout=self.tts_timeout) self.before_stop_play_files.clear() - logger.bind(tag=TAG).info("TTS会话启动成功") + logger.bind(tag=TAG).debug("TTS会话启动成功") except Exception as e: logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}") continue @@ -146,7 +153,6 @@ class TTSProvider(TTSProviderBase): loop=self.conn.loop, ) future.result(timeout=self.tts_timeout) - logger.bind(tag=TAG).debug("TTS文本发送成功") except Exception as e: logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}") continue @@ -161,12 +167,12 @@ class TTSProvider(TTSProviderBase): if message.sentence_type == SentenceType.LAST: try: - logger.bind(tag=TAG).info("开始结束TTS会话...") + logger.bind(tag=TAG).debug("开始结束TTS会话...") future = asyncio.run_coroutine_threadsafe( self.finish_session(self.conn.sentence_id), loop=self.conn.loop, ) - future.result(timeout=self.tts_timeout) + future.result() except Exception as e: logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}") continue @@ -202,7 +208,6 @@ class TTSProvider(TTSProviderBase): await self.ws.send(json.dumps(continue_task_message)) self.last_active_time = time.time() - logger.bind(tag=TAG).debug(f"已发送文本: {filtered_text}") return except Exception as e: logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}") @@ -216,22 +221,22 @@ class TTSProvider(TTSProviderBase): async def start_session(self, session_id): """启动TTS会话""" - logger.bind(tag=TAG).info(f"开始会话～～{session_id}") + logger.bind(tag=TAG).debug(f"开始会话～～{session_id}") try: - # 检查并清理上一个会话的监听任务 - if ( - self._monitor_task is not None - and isinstance(self._monitor_task, Task) - and not self._monitor_task.done() - ): - logger.bind(tag=TAG).info("检测到未完成的上个会话，关闭监听任务...") + # 上个会话处于激活状态时关闭上个连接新建链接 + if self.activate_session: await self.close() + # 设置会话激活标志 + self.activate_session = True + # 确保连接可用 await self._ensure_connection() # 启动监听任务 - self._monitor_task = asyncio.create_task(self._start_monitor_tts_response()) + if self._monitor_task is None or self._monitor_task.done(): + logger.bind(tag=TAG).debug("启动监听任务...") + self._monitor_task = asyncio.create_task(self._start_monitor_tts_response()) # 发送run-task消息启动会话 run_task_message = { @@ -260,7 +265,7 @@ class TTSProvider(TTSProviderBase): await self.ws.send(json.dumps(run_task_message)) self.last_active_time = time.time() - logger.bind(tag=TAG).info("会话启动请求已发送") + logger.bind(tag=TAG).debug("会话启动请求已发送") except Exception as e: logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}") await self.close() @@ -268,7 +273,7 @@ class TTSProvider(TTSProviderBase): async def finish_session(self, session_id): """结束TTS会话""" - logger.bind(tag=TAG).info(f"关闭会话～～{session_id}") + logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}") try: if self.ws and session_id: # 发送finish-task消息 @@ -285,17 +290,6 @@ class TTSProvider(TTSProviderBase): await self.ws.send(json.dumps(finish_task_message)) self.last_active_time = time.time() - logger.bind(tag=TAG).info("会话结束请求已发送") - # 等待监听任务完成 - if self._monitor_task: - try: - await self._monitor_task - except Exception as e: - logger.bind(tag=TAG).error( - f"等待监听任务完成时发生错误: {str(e)}" - ) - finally: - self._monitor_task = None except Exception as e: logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}") @@ -304,6 +298,8 @@ class TTSProvider(TTSProviderBase): async def close(self): """清理资源""" + await super().close() + self.activate_session = False # 取消监听任务 if self._monitor_task: try: @@ -325,19 +321,13 @@ class TTSProvider(TTSProviderBase): self.last_active_time = None async def _start_monitor_tts_response(self): - """监听TTS响应""" + """监听TTS响应 - 长期运行""" try: - session_finished = False while not self.conn.stop_event.is_set(): try: msg = await self.ws.recv() self.last_active_time = time.time() - # 检查客户端是否中止 - if self.conn.client_abort: - logger.bind(tag=TAG).info("收到打断信息，终止监听TTS响应") - break - if isinstance(msg, str): # JSON控制消息 try: data = json.loads(msg) @@ -348,19 +338,19 @@ class TTSProvider(TTSProviderBase): self.tts_audio_queue.put((SentenceType.FIRST, [], None)) elif event == "result-generated": # 发送缓存的数据 - if self.conn.tts_MessageText: + tts_text = self.get_tts_text(self.conn.sentence_id) + if tts_text: logger.bind(tag=TAG).info( - f"句子语音生成成功： {self.conn.tts_MessageText}" + f"句子语音生成成功： {tts_text}" ) self.tts_audio_queue.put( - (SentenceType.FIRST, [], self.conn.tts_MessageText) + (SentenceType.FIRST, [], tts_text) ) - self.conn.tts_MessageText = None + self.clear_tts_text(self.conn.sentence_id) elif event == "task-finished": logger.bind(tag=TAG).debug("TTS任务完成~") + self.activate_session = False self._process_before_stop_play_files() - session_finished = True - break elif event == "task-failed": error_code = data["header"].get("error_code", "unknown") error_message = data["header"].get("error_message", "未知错误") @@ -383,8 +373,8 @@ class TTSProvider(TTSProviderBase): ) break - # 仅在连接异常且非正常结束时才关闭连接 - if not session_finished and self.ws: + # 连接异常时关闭WebSocket + if self.ws: try: await self.ws.close() except: @@ -392,6 +382,7 @@ class TTSProvider(TTSProviderBase): self.ws = None # 监听任务退出时清理引用 finally: + self.activate_session = False self._monitor_task = None def audio_to_opus_data_stream( diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index f773b710..d9a0ca44 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -135,6 +135,7 @@ class TTSProvider(TTSProviderBase): self.ws_url = f"wss://{self.host}/ws/v1" self.ws = None self._monitor_task = None + self.activate_session = False self.last_active_time = None # 专属tts设置 @@ -188,7 +189,7 @@ class TTSProvider(TTSProviderBase): if self.ws and current_time - self.last_active_time < 10: # 10秒内才可以复用链接进行连续对话 self.task_id = uuid.uuid4().hex - logger.bind(tag=TAG).info(f"使用已有链接..., task_id: {self.task_id}") + logger.bind(tag=TAG).debug(f"使用已有链接..., task_id: {self.task_id}") return self.ws logger.bind(tag=TAG).debug("开始建立新连接...") @@ -214,15 +215,21 @@ class TTSProvider(TTSProviderBase): 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} | 会话ID: {self.conn.sentence_id}" - ) - if message.sentence_type == SentenceType.FIRST: - self.conn.client_abort = False + # 过滤旧消息：检查sentence_id是否匹配 + if message.sentence_id != self.conn.sentence_id: + continue + + logger.bind(tag=TAG).debug( + f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" + ) if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") + asyncio.run_coroutine_threadsafe( + self.finish_session(self.conn.sentence_id), + loop=self.conn.loop, + ) continue if message.sentence_type == SentenceType.FIRST: @@ -252,7 +259,6 @@ class TTSProvider(TTSProviderBase): loop=self.conn.loop, ) future.result(timeout=self.tts_timeout) - logger.bind(tag=TAG).debug("TTS文本发送成功") except Exception as e: logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}") continue @@ -317,22 +323,20 @@ class TTSProvider(TTSProviderBase): async def start_session(self, task_id): logger.bind(tag=TAG).debug("开始会话～～") try: - # 会话开始时检测上个会话的监听状态 - if ( - self._monitor_task is not None - and isinstance(self._monitor_task, Task) - and not self._monitor_task.done() - ): - logger.bind(tag=TAG).info( - "检测到未完成的上个会话，关闭监听任务和连接..." - ) + # 上个会话处于激活状态时关闭上个连接新建链接 + if self.activate_session: await self.close() + # 设置会话激活标志 + self.activate_session = True + # 建立新连接 await self._ensure_connection() # 启动监听任务 - self._monitor_task = asyncio.create_task(self._start_monitor_tts_response()) + if self._monitor_task is None or self._monitor_task.done(): + logger.bind(tag=TAG).debug("启动监听任务...") + self._monitor_task = asyncio.create_task(self._start_monitor_tts_response()) start_request = { "header": { @@ -377,15 +381,7 @@ class TTSProvider(TTSProviderBase): await self.ws.send(json.dumps(stop_request)) logger.bind(tag=TAG).debug("会话结束请求已发送") self.last_active_time = time.time() - if self._monitor_task: - try: - await self._monitor_task - except Exception as e: - logger.bind(tag=TAG).error( - f"等待监听任务完成时发生错误: {str(e)}" - ) - finally: - self._monitor_task = None + except Exception as e: logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}") # 确保清理资源 @@ -394,6 +390,8 @@ class TTSProvider(TTSProviderBase): async def close(self): """资源清理""" + await super().close() + self.activate_session = False if self._monitor_task: try: self._monitor_task.cancel() @@ -413,17 +411,13 @@ class TTSProvider(TTSProviderBase): self.last_active_time = None async def _start_monitor_tts_response(self): - """监听TTS响应""" + """监听TTS响应 - 长期运行""" try: - session_finished = False # 标记会话是否正常结束 while not self.conn.stop_event.is_set(): try: msg = await self.ws.recv() self.last_active_time = time.time() - # 检查客户端是否中止 - if self.conn.client_abort: - logger.bind(tag=TAG).info("收到打断信息，终止监听TTS响应") - break + if isinstance(msg, str): # 文本控制消息 try: data = json.loads(msg) @@ -436,19 +430,19 @@ class TTSProvider(TTSProviderBase): ) elif event_name == "SentenceEnd": # 发送缓存的数据 - if self.conn.tts_MessageText: + tts_text = self.get_tts_text(self.conn.sentence_id) + if tts_text: logger.bind(tag=TAG).info( - f"句子语音生成成功： {self.conn.tts_MessageText}" + f"句子语音生成成功： {tts_text}" ) self.tts_audio_queue.put( - (SentenceType.FIRST, [], self.conn.tts_MessageText) + (SentenceType.FIRST, [], tts_text) ) - self.conn.tts_MessageText = None + self.clear_tts_text(self.conn.sentence_id) elif event_name == "SynthesisCompleted": logger.bind(tag=TAG).debug(f"会话结束～～") + self.activate_session = False self._process_before_stop_play_files() - session_finished = True - break except json.JSONDecodeError: logger.bind(tag=TAG).warning("收到无效的JSON消息") # 二进制消息（音频数据） @@ -462,8 +456,8 @@ class TTSProvider(TTSProviderBase): f"处理TTS响应时出错: {e}\n{traceback.format_exc()}" ) break - # 仅在连接异常时才关闭 - if not session_finished and self.ws: + # 连接异常时关闭WebSocket + if self.ws: try: await self.ws.close() except: @@ -471,6 +465,7 @@ class TTSProvider(TTSProviderBase): self.ws = None # 监听任务退出时清理引用 finally: + self.activate_session = False self._monitor_task = None def audio_to_opus_data_stream( diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index d4440f92..a5c4a7da 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -5,6 +5,7 @@ import queue import asyncio import threading import traceback +import concurrent.futures from core.utils import p3 from datetime import datetime @@ -42,6 +43,8 @@ class TTSProviderBase(ABC): self.tts_audio_first_sentence = True self.before_stop_play_files = [] self.report_on_last = False + # sentence_id 到文本的映射，用于流式TTS获取正确的字幕文本 + self._sentence_text_map = {} self.tts_text_buff = [] self.punctuations = ( @@ -277,14 +280,49 @@ class TTSProviderBase(ABC): ) self.audio_play_priority_thread.start() + def store_tts_text(self, sentence_id, text): + """存储指定 sentence_id 对应的文本，用于流式TTS获取正确的字幕文本 + + Args: + sentence_id: 会话ID + text: 要存储的文本 + """ + if sentence_id and text: + self._sentence_text_map[sentence_id] = text + # 只保留最近 5 个，防止内存泄漏 + if len(self._sentence_text_map) > 5: + oldest = next(iter(self._sentence_text_map)) + del self._sentence_text_map[oldest] + + def get_tts_text(self, sentence_id): + """获取指定 sentence_id 对应的文本 + + Args: + sentence_id: 会话ID + + Returns: + str: 对应的文本，如果不存在返回 None + """ + return self._sentence_text_map.get(sentence_id) + + def clear_tts_text(self, sentence_id): + """清除指定 sentence_id 的文本 + + Args: + sentence_id: 会话ID + """ + if sentence_id in self._sentence_text_map: + del self._sentence_text_map[sentence_id] + # 这里默认是非流式的处理方式 # 流式处理方式请在子类中重写 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.conn.client_abort = False + # 过滤旧消息：检查sentence_id是否匹配 + if message.sentence_id != self.conn.sentence_id: + continue if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") continue @@ -386,6 +424,7 @@ class TTSProviderBase(ABC): async def close(self): """资源清理方法""" + self._sentence_text_map.clear() if hasattr(self, "ws") and self.ws: await self.ws.close() diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index 75a59f94..f95acaba 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -221,7 +221,7 @@ class TTSProvider(TTSProviderBase): try: if self.ws: if self.enable_ws_reuse: - logger.bind(tag=TAG).info(f"使用已有链接...") + logger.bind(tag=TAG).debug(f"使用已有链接...") return self.ws else: try: @@ -272,12 +272,14 @@ class TTSProvider(TTSProviderBase): 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} | 会话ID: {self.conn.sentence_id}" - ) - if message.sentence_type == SentenceType.FIRST: - self.conn.client_abort = False + # 过滤旧消息：检查sentence_id是否匹配 + if message.sentence_id != self.conn.sentence_id: + continue + + logger.bind(tag=TAG).debug( + f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" + ) if self.conn.client_abort: try: @@ -327,7 +329,6 @@ class TTSProvider(TTSProviderBase): loop=self.conn.loop, ) future.result(timeout=self.tts_timeout) - logger.bind(tag=TAG).debug("TTS文本发送成功") except Exception as e: logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}") continue @@ -387,15 +388,8 @@ class TTSProvider(TTSProviderBase): async def start_session(self, session_id): logger.bind(tag=TAG).debug(f"开始会话～～{session_id}") try: - # 等待上一个会话结束，最多等待3次 - for _ in range(3): - if not self.activate_session: - break - logger.bind(tag=TAG).debug(f"等待上一个会话结束...") - await asyncio.sleep(0.1) - else: - # 等待超时，强制清除连接状态 - logger.bind(tag=TAG).debug("等待上一个会话超时，清除连接状态...") + # 上个会话处于激活状态时关闭上个连接新建链接 + if self.activate_session: await self.close() # 设置会话激活标志 @@ -468,6 +462,7 @@ class TTSProvider(TTSProviderBase): async def close(self): """资源清理方法""" + await super().close() self.activate_session = False # 取消监听任务 if self._monitor_task: @@ -525,14 +520,16 @@ class TTSProvider(TTSProviderBase): and res.header.message_type == AUDIO_ONLY_RESPONSE ): # 处理seed-tts-2.0文本字幕 - if self.resource_type and self.conn.tts_MessageText: - logger.bind(tag=TAG).info( - f"句子语音生成成功： {self.conn.tts_MessageText}" - ) - self.tts_audio_queue.put( - (SentenceType.FIRST, [], self.conn.tts_MessageText) - ) - self.conn.tts_MessageText = None + if self.resource_type: + tts_text = self.get_tts_text(self.conn.sentence_id) + if tts_text: + logger.bind(tag=TAG).info( + f"句子语音生成成功： {tts_text}" + ) + self.tts_audio_queue.put( + (SentenceType.FIRST, [], tts_text) + ) + self.clear_tts_text(self.conn.sentence_id) self.wav_to_opus_data_audio_raw_stream(res.payload, callback=self.handle_opus) elif not self.resource_type and res.optional.event == EVENT_TTSSentenceEnd: logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}") diff --git a/main/xiaozhi-server/core/providers/tts/xunfei_stream.py b/main/xiaozhi-server/core/providers/tts/xunfei_stream.py index 7a806014..40d8a11b 100644 --- a/main/xiaozhi-server/core/providers/tts/xunfei_stream.py +++ b/main/xiaozhi-server/core/providers/tts/xunfei_stream.py @@ -117,6 +117,7 @@ class TTSProvider(TTSProviderBase): # WebSocket配置 self.ws = None self._monitor_task = None + self.activate_session = False # 序列号管理 self.text_seq = 0 @@ -128,7 +129,7 @@ class TTSProvider(TTSProviderBase): async def _ensure_connection(self): """确保WebSocket连接可用""" try: - logger.bind(tag=TAG).info("开始建立新连接...") + logger.bind(tag=TAG).debug("开始建立新连接...") # 生成认证URL auth_url = XunfeiWSAuth.create_auth_url( @@ -141,7 +142,7 @@ class TTSProvider(TTSProviderBase): ping_timeout=10, close_timeout=10, ) - logger.bind(tag=TAG).info("WebSocket连接建立成功") + logger.bind(tag=TAG).debug("WebSocket连接建立成功") return self.ws except Exception as e: logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}") @@ -153,16 +154,21 @@ class TTSProvider(TTSProviderBase): while not self.conn.stop_event.is_set(): try: message = self.tts_text_queue.get(timeout=1) + + # 过滤旧消息：检查sentence_id是否匹配 + if message.sentence_id != self.conn.sentence_id: + continue + logger.bind(tag=TAG).debug( - f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}" + f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" ) if message.sentence_type == SentenceType.FIRST: # 重置序列号 self.text_seq = 0 - self.conn.client_abort = False # 增加序列号 self.text_seq += 1 + if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") continue @@ -172,16 +178,16 @@ class TTSProvider(TTSProviderBase): try: if not getattr(self.conn, "sentence_id", None): self.conn.sentence_id = uuid.uuid4().hex - logger.bind(tag=TAG).info(f"自动生成新的 会话ID: {self.conn.sentence_id}") + logger.bind(tag=TAG).debug(f"自动生成新的 会话ID: {self.conn.sentence_id}") - logger.bind(tag=TAG).info("开始启动TTS会话...") + logger.bind(tag=TAG).debug("开始启动TTS会话...") future = asyncio.run_coroutine_threadsafe( self.start_session(self.conn.sentence_id), loop=self.conn.loop, ) future.result(timeout=self.tts_timeout) self.before_stop_play_files.clear() - logger.bind(tag=TAG).info("TTS会话启动成功") + logger.bind(tag=TAG).debug("TTS会话启动成功") except Exception as e: logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}") @@ -199,7 +205,6 @@ class TTSProvider(TTSProviderBase): loop=self.conn.loop, ) future.result(timeout=self.tts_timeout) - logger.bind(tag=TAG).debug("TTS文本发送成功") except Exception as e: logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}") # 不使用continue，确保后续处理不被中断 @@ -216,7 +221,7 @@ class TTSProvider(TTSProviderBase): # 处理会话结束 if message.sentence_type == SentenceType.LAST: try: - logger.bind(tag=TAG).info("开始结束TTS会话...") + logger.bind(tag=TAG).debug("开始结束TTS会话...") asyncio.run_coroutine_threadsafe( self.finish_session(self.conn.sentence_id), loop=self.conn.loop, @@ -257,30 +262,28 @@ class TTSProvider(TTSProviderBase): raise async def start_session(self, session_id): - logger.bind(tag=TAG).info(f"开始会话～～{session_id}") + logger.bind(tag=TAG).debug(f"开始会话～～{session_id}") try: - # 会话开始时检测上个会话的监听状态 - if ( - self._monitor_task is not None - and isinstance(self._monitor_task, Task) - and not self._monitor_task.done() - ): - logger.bind(tag=TAG).info( - "检测到未完成的上个会话，关闭监听任务和连接..." - ) + # 上个会话处于激活状态时关闭上个连接新建链接 + if self.activate_session: await self.close() + # 设置会话激活标志 + self.activate_session = True + # 建立新连接 await self._ensure_connection() # 启动监听任务 - self._monitor_task = asyncio.create_task(self._start_monitor_tts_response()) + if self._monitor_task is None or self._monitor_task.done(): + logger.bind(tag=TAG).debug("启动监听任务...") + self._monitor_task = asyncio.create_task(self._start_monitor_tts_response()) # 发送会话启动请求 start_request = self._build_base_request(status=0) await self.ws.send(json.dumps(start_request)) - logger.bind(tag=TAG).info("会话启动请求已发送") + logger.bind(tag=TAG).debug("会话启动请求已发送") except Exception as e: logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}") # 确保清理资源 @@ -288,13 +291,13 @@ class TTSProvider(TTSProviderBase): raise async def finish_session(self, session_id): - logger.bind(tag=TAG).info(f"关闭会话～～{session_id}") + logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}") try: if self.ws: # 发送会话结束请求 stop_request = self._build_base_request(status=2) await self.ws.send(json.dumps(stop_request)) - logger.bind(tag=TAG).info("会话结束请求已发送") + logger.bind(tag=TAG).debug("会话结束请求已发送") if self._monitor_task: try: @@ -310,6 +313,8 @@ class TTSProvider(TTSProviderBase): async def close(self): """资源清理""" + await super().close() + self.activate_session = False if self._monitor_task: try: self._monitor_task.cancel() @@ -358,17 +363,19 @@ class TTSProvider(TTSProviderBase): ) elif status == 2: logger.bind(tag=TAG).debug("收到结束状态的音频数据，TTS合成完成") + self.activate_session = False self._process_before_stop_play_files() break else: - if self.conn.tts_MessageText: + tts_text = self.get_tts_text(self.conn.sentence_id) + if tts_text: logger.bind(tag=TAG).info( - f"句子语音生成成功： {self.conn.tts_MessageText}" + f"句子语音生成成功： {tts_text}" ) self.tts_audio_queue.put( - (SentenceType.FIRST, [], self.conn.tts_MessageText) + (SentenceType.FIRST, [], tts_text) ) - self.conn.tts_MessageText = None + self.clear_tts_text(self.conn.sentence_id) try: audio_bytes = base64.b64decode(audio_data) self.opus_encoder.encode_pcm_to_opus_stream( @@ -405,6 +412,7 @@ class TTSProvider(TTSProviderBase): self.ws = None # 监听任务退出时清理引用 finally: + self.activate_session = False self._monitor_task = None def to_tts(self, text: str) -> list: From a6904f6ebd945ee69234a11befd1f5d498652732 Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Thu, 9 Apr 2026 17:15:35 +0800 Subject: [PATCH 2/7] =?UTF-8?q?refactor:=20vad=E7=8B=AC=E7=AB=8B=E6=97=B6?= =?UTF-8?q?=E9=97=B4=E6=88=B3=EF=BC=8C=E7=9B=B8=E5=85=B3=E6=AE=8B=E4=BD=99?= =?UTF-8?q?=E7=8A=B6=E6=80=81=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 7 +++---- main/xiaozhi-server/core/handle/receiveAudioHandle.py | 1 + main/xiaozhi-server/core/handle/sendAudioHandle.py | 6 ++++-- main/xiaozhi-server/core/providers/asr/base.py | 6 ++++++ main/xiaozhi-server/core/providers/vad/silero.py | 4 ++-- 5 files changed, 16 insertions(+), 8 deletions(-) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index b64afd5d..c948b815 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -159,6 +159,7 @@ class ConnectionHandler: self.client_voice_window = deque(maxlen=5) self.first_activity_time = 0.0 # 记录首次活动的时间（毫秒） self.last_activity_time = 0.0 # 统一的活动时间戳（毫秒） + self.vad_last_voice_time = 0.0 # 记录用户最后一次说话的时间（毫秒） self.client_voice_stop = False self.last_is_voice = False @@ -1288,10 +1289,7 @@ class ConnectionHandler: # 标记任务完成 self.report_queue.task_done() - def clearSpeakStatus(self, sentence_id=None): - # 如果sentence_id不匹配，说明是旧轮次的回调，不需要执行 - if sentence_id is not None and sentence_id != self.sentence_id: - return + def clearSpeakStatus(self): self.client_is_speaking = False self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态") @@ -1433,6 +1431,7 @@ class ConnectionHandler: self.client_voice_stop = False self.client_voice_window.clear() self.last_is_voice = False + self.vad_last_voice_time = 0.0 # Clear ASR buffers self.asr_audio.clear() diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index cd53629b..7e234a9e 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -77,6 +77,7 @@ async def startToChat(conn: "ConnectionHandler", text): ): await max_out_size(conn) return + # manual 模式下不打断正在播放的内容 if conn.client_is_speaking and conn.client_listen_mode != "manual": await handleAbortMessage(conn) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index f62cc582..789d61f6 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -284,8 +284,10 @@ async def send_tts_message(conn: "ConnectionHandler", state, text=None): # 停止音频发送循环（仅在流控器已初始化时调用） if hasattr(conn, "audio_rate_controller") and conn.audio_rate_controller: conn.audio_rate_controller.stop_sending() - # 清除服务端讲话状态，传入 sentence_id 用于判断是否是当前轮次 - conn.clearSpeakStatus(current_sentence_id) + # 检查是否是当前轮次 + if current_sentence_id != conn.sentence_id: + return + conn.clearSpeakStatus() # 发送消息到客户端 await conn.websocket.send(json.dumps(message)) diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 986dd275..6ede6a5d 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -84,6 +84,12 @@ class ASRProviderBase(ABC): async def handle_voice_stop(self, conn: "ConnectionHandler", asr_audio_task: List[bytes]): """并行处理ASR和声纹识别""" try: + # 如果处于退出流程中，直接关闭连接，不处理新消息 + if conn.close_after_chat or conn.is_exiting: + logger.bind(tag=TAG).info("退出流程中收到新消息，直接关闭连接") + await conn.close() + return + total_start_time = time.monotonic() # 准备音频数据 diff --git a/main/xiaozhi-server/core/providers/vad/silero.py b/main/xiaozhi-server/core/providers/vad/silero.py index 8d3891dc..3682f3b8 100644 --- a/main/xiaozhi-server/core/providers/vad/silero.py +++ b/main/xiaozhi-server/core/providers/vad/silero.py @@ -107,12 +107,12 @@ class VADProvider(VADProviderBase): # 如果之前有声音，但本次没有声音，且与上次有声音的时间差已经超过了静默阈值，则认为已经说完一句话 if conn.client_have_voice and not client_have_voice: - stop_duration = time.time() * 1000 - conn.last_activity_time + stop_duration = time.time() * 1000 - conn.vad_last_voice_time if stop_duration >= self.silence_threshold_ms: conn.client_voice_stop = True if client_have_voice: conn.client_have_voice = True - conn.last_activity_time = time.time() * 1000 + conn.vad_last_voice_time = time.time() * 1000 return client_have_voice except opuslib_next.OpusError as e: From 0e23672be766e105a7811f10f17c00822a07b9f4 Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Thu, 9 Apr 2026 17:58:19 +0800 Subject: [PATCH 3/7] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E7=9B=B4=E6=8E=A5?= =?UTF-8?q?=E5=9B=9E=E5=A4=8D=E6=98=AF=E5=AD=97=E5=B9=95=E6=9C=AA=E8=AE=B0?= =?UTF-8?q?=E5=BD=95?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 1 + 1 file changed, 1 insertion(+) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index c948b815..31c79f5f 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -1212,6 +1212,7 @@ class ConnectionHandler: ]: # 直接回复前端 text = result.response if result.response else result.result self.tts.tts_one_sentence(self, ContentType.TEXT, content_detail=text) + self.tts.store_tts_text(self.sentence_id, text) self.dialogue.put(Message(role="assistant", content=text)) elif result.action == Action.REQLLM: # 收集需要 LLM 处理的工具 From 0f1e0d0d2e28286f9b0a75851afdf607dea9f739 Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Fri, 10 Apr 2026 14:22:36 +0800 Subject: [PATCH 4/7] =?UTF-8?q?=E9=9F=B3=E9=A2=91=E4=BC=9A=E5=9C=A8?= =?UTF-8?q?=E9=98=9F=E5=88=97=E7=A7=AF=E5=8E=8B=E7=AD=89=E5=BE=85=E5=8F=91?= =?UTF-8?q?=E9=80=81=EF=BC=8C=E7=AD=89=E5=BE=85=E5=8D=8F=E7=A8=8B=E7=BB=93?= =?UTF-8?q?=E6=9E=9C=E8=BF=94=E5=9B=9E=E4=B8=8D=E8=AE=BE=E7=BD=AE=E8=B6=85?= =?UTF-8?q?=E6=97=B6=E6=97=B6=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/providers/tts/base.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index a5c4a7da..1a04907a 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -407,7 +407,7 @@ class TTSProviderBase(ABC): sendAudioMessage(self.conn, sentence_type, audio_datas, text), self.conn.loop, ) - future.result(timeout=self.tts_timeout) + future.result() # 记录输出和报告 if self.conn.max_output_size > 0 and text: From 27e5e1e9d7d690798d74f23e16dc5a7077e6f04f Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Mon, 13 Apr 2026 11:33:47 +0800 Subject: [PATCH 5/7] =?UTF-8?q?refactor:=20=E4=BC=98=E5=8C=96=E9=9D=9E?= =?UTF-8?q?=E6=B5=81=E5=BC=8F=E9=9F=B3=E9=A2=91=E8=BF=87=E6=BB=A4=E6=9C=BA?= =?UTF-8?q?=E5=88=B6=20fix:=20=E4=BF=AE=E5=A4=8D=E7=AD=89=E5=BE=85?= =?UTF-8?q?=E4=B8=AD=E8=A2=AB=E6=89=93=E6=96=AD=E9=98=9F=E5=88=97=E8=A2=AB?= =?UTF-8?q?=E5=81=9C=E6=AD=A2=E9=80=A0=E6=88=90=E7=9A=84=E6=AD=BB=E9=94=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/handle/sendAudioHandle.py | 14 ++++++++--- .../xiaozhi-server/core/providers/tts/base.py | 25 +++++++++++-------- 2 files changed, 24 insertions(+), 15 deletions(-) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 789d61f6..0ae801ed 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -17,7 +17,11 @@ AUDIO_FRAME_DURATION = 60 PRE_BUFFER_COUNT = 5 -async def sendAudioMessage(conn: "ConnectionHandler", sentenceType, audios, text): +async def sendAudioMessage(conn: "ConnectionHandler", sentenceType, audios, text, sentence_id=None): + # 跳过旧句子残留音频 + if sentence_id is not None and sentence_id != conn.sentence_id: + return + if conn.tts.tts_audio_first_sentence: conn.logger.bind(tag=TAG).info(f"发送第一段语音: {text}") conn.tts.tts_audio_first_sentence = False @@ -281,12 +285,14 @@ async def send_tts_message(conn: "ConnectionHandler", state, text=None): await sendAudio(conn, audios) # 等待所有音频包发送完成 await _wait_for_audio_completion(conn) - # 停止音频发送循环（仅在流控器已初始化时调用） - if hasattr(conn, "audio_rate_controller") and conn.audio_rate_controller: - conn.audio_rate_controller.stop_sending() + # 检查是否是当前轮次 if current_sentence_id != conn.sentence_id: return + + # 停止音频发送循环（仅在流控器已初始化时调用） + if hasattr(conn, "audio_rate_controller") and conn.audio_rate_controller: + conn.audio_rate_controller.stop_sending() conn.clearSpeakStatus() # 发送消息到客户端 diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 1a04907a..25ade2d6 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -83,7 +83,7 @@ class TTSProviderBase(ABC): def handle_opus(self, opus_data: bytes): logger.bind(tag=TAG).debug(f"推送数据到队列里面帧数～～ {len(opus_data)}") - self.tts_audio_queue.put((SentenceType.MIDDLE, opus_data, None)) + self.tts_audio_queue.put((SentenceType.MIDDLE, opus_data, None, getattr(self, 'current_sentence_id', None))) def handle_audio_file(self, file_audio: bytes, text): self.before_stop_play_files.append((file_audio, text)) @@ -97,7 +97,7 @@ class TTSProviderBase(ABC): try: audio_bytes = asyncio.run(self.text_to_speak(text, None)) if audio_bytes: - self.tts_audio_queue.put((SentenceType.FIRST, None, text)) + self.tts_audio_queue.put((SentenceType.FIRST, None, text, getattr(self, 'current_sentence_id', None))) audio_bytes_to_data_stream( audio_bytes, file_type=self.audio_file_type, @@ -146,7 +146,7 @@ class TTSProviderBase(ABC): logger.bind(tag=TAG).error( f"语音生成失败: {text}，请检查网络或服务是否正常" ) - self.tts_audio_queue.put((SentenceType.FIRST, None, text)) + self.tts_audio_queue.put((SentenceType.FIRST, None, text, getattr(self, 'current_sentence_id', None))) self._process_audio_file_stream(tmp_file, callback=opus_handler) except Exception as e: logger.bind(tag=TAG).error(f"Failed to generate TTS file: {e}") @@ -327,7 +327,7 @@ class TTSProviderBase(ABC): logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") continue if message.sentence_type == SentenceType.FIRST: - # 初始化参数 + self.current_sentence_id = message.sentence_id self.tts_stop_request = False self.processed_chars = 0 self.tts_text_buff = [] @@ -348,7 +348,7 @@ class TTSProviderBase(ABC): if message.sentence_type == SentenceType.LAST: self._process_remaining_text_stream(opus_handler=self.handle_opus) self.tts_audio_queue.put( - (message.sentence_type, [], message.content_detail) + (message.sentence_type, [], message.content_detail, message.sentence_id) ) except queue.Empty: @@ -367,9 +367,12 @@ class TTSProviderBase(ABC): text = None try: try: - sentence_type, audio_datas, text = self.tts_audio_queue.get( - timeout=0.1 - ) + item = self.tts_audio_queue.get(timeout=0.1) + if len(item) == 4: + sentence_type, audio_datas, text, sentence_id = item + else: + sentence_type, audio_datas, text = item + sentence_id = None except queue.Empty: if self.conn.stop_event.is_set(): break @@ -404,7 +407,7 @@ class TTSProviderBase(ABC): # 发送音频 future = asyncio.run_coroutine_threadsafe( - sendAudioMessage(self.conn, sentence_type, audio_datas, text), + sendAudioMessage(self.conn, sentence_type, audio_datas, text, sentence_id), self.conn.loop, ) future.result() @@ -493,9 +496,9 @@ class TTSProviderBase(ABC): def _process_before_stop_play_files(self): for audio_datas, text in self.before_stop_play_files: - self.tts_audio_queue.put((SentenceType.MIDDLE, audio_datas, text)) + self.tts_audio_queue.put((SentenceType.MIDDLE, audio_datas, text, getattr(self, 'current_sentence_id', None))) self.before_stop_play_files.clear() - self.tts_audio_queue.put((SentenceType.LAST, [], None)) + self.tts_audio_queue.put((SentenceType.LAST, [], None, getattr(self, 'current_sentence_id', None))) def _process_remaining_text_stream( self, opus_handler: Callable[[bytes], None] = None From a0497fdc482933dbdf6b8302c2c43c1c2e325e3d Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Tue, 14 Apr 2026 10:18:09 +0800 Subject: [PATCH 6/7] =?UTF-8?q?=E7=BB=99=E5=A4=8D=E7=94=A8=E5=9E=8BTTS?= =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E7=9B=B8=E5=85=B3=E8=BF=87=E6=BB=A4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/alibl_stream.py | 15 ++++++++++++--- .../core/providers/tts/aliyun_stream.py | 9 +++++++++ 2 files changed, 21 insertions(+), 3 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/alibl_stream.py b/main/xiaozhi-server/core/providers/tts/alibl_stream.py index 8bad4ee1..af912ef6 100644 --- a/main/xiaozhi-server/core/providers/tts/alibl_stream.py +++ b/main/xiaozhi-server/core/providers/tts/alibl_stream.py @@ -331,7 +331,16 @@ class TTSProvider(TTSProviderBase): if isinstance(msg, str): # JSON控制消息 try: data = json.loads(msg) - event = data["header"].get("event") + header = data.get("header", {}) + event = header.get("event") + task_id = header.get("task_id") + + # 只处理当前活跃会话的响应 + if task_id and self.conn.sentence_id != task_id: + if event in ["task-finished", "task-failed"]: + logger.bind(tag=TAG).debug(f"收到残余下行结束响应重置会话状态～～") + self.activate_session = False + continue if event == "task-started": logger.bind(tag=TAG).debug("TTS任务启动成功~") @@ -352,8 +361,8 @@ class TTSProvider(TTSProviderBase): self.activate_session = False self._process_before_stop_play_files() elif event == "task-failed": - error_code = data["header"].get("error_code", "unknown") - error_message = data["header"].get("error_message", "未知错误") + error_code = header.get("error_code", "unknown") + error_message = header.get("error_message", "未知错误") logger.bind(tag=TAG).error( f"TTS任务失败: {error_code} - {error_message}" ) diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index d9a0ca44..824e7902 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -423,6 +423,15 @@ class TTSProvider(TTSProviderBase): data = json.loads(msg) header = data.get("header", {}) event_name = header.get("name") + task_id = header.get("task_id") + + # 只处理当前活跃会话的响应 + if task_id and self.task_id != task_id: + if event_name in ["SynthesisCompleted", "TaskFailed"]: + logger.bind(tag=TAG).debug(f"收到残余下行结束响应重置会话状态～～") + self.activate_session = False + continue + if event_name == "SynthesisStarted": logger.bind(tag=TAG).debug("TTS合成已启动") self.tts_audio_queue.put( From ad01b059404afbe413f2ab816709b2c0d8e49096 Mon Sep 17 00:00:00 2001 From: Sakura-RanChen <1908198662@qq.com> Date: Tue, 14 Apr 2026 15:05:55 +0800 Subject: [PATCH 7/7] =?UTF-8?q?=E6=89=93=E6=96=AD=E5=88=A4=E6=96=AD?= =?UTF-8?q?=E4=BC=98=E5=85=88=E7=BA=A7=E8=B0=83=E6=95=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/providers/tts/alibl_stream.py | 16 ++++++++-------- .../core/providers/tts/aliyun_stream.py | 16 ++++++++-------- main/xiaozhi-server/core/providers/tts/base.py | 6 +++--- .../core/providers/tts/huoshan_double_stream.py | 16 ++++++++-------- .../core/providers/tts/xunfei_stream.py | 8 ++++---- 5 files changed, 31 insertions(+), 31 deletions(-) diff --git a/main/xiaozhi-server/core/providers/tts/alibl_stream.py b/main/xiaozhi-server/core/providers/tts/alibl_stream.py index af912ef6..f3032b5e 100644 --- a/main/xiaozhi-server/core/providers/tts/alibl_stream.py +++ b/main/xiaozhi-server/core/providers/tts/alibl_stream.py @@ -103,14 +103,6 @@ class TTSProvider(TTSProviderBase): try: message = self.tts_text_queue.get(timeout=1) - # 过滤旧消息：检查sentence_id是否匹配 - if message.sentence_id != self.conn.sentence_id: - continue - - logger.bind(tag=TAG).debug( - f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" - ) - if self.conn.client_abort: try: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") @@ -123,6 +115,14 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).error(f"取消TTS会话失败: {str(e)}") continue + # 过滤旧消息：检查sentence_id是否匹配 + if message.sentence_id != self.conn.sentence_id: + continue + + logger.bind(tag=TAG).debug( + f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" + ) + if message.sentence_type == SentenceType.FIRST: # 初始化会话 try: diff --git a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py index 824e7902..f9581ee8 100644 --- a/main/xiaozhi-server/core/providers/tts/aliyun_stream.py +++ b/main/xiaozhi-server/core/providers/tts/aliyun_stream.py @@ -216,14 +216,6 @@ class TTSProvider(TTSProviderBase): try: message = self.tts_text_queue.get(timeout=1) - # 过滤旧消息：检查sentence_id是否匹配 - if message.sentence_id != self.conn.sentence_id: - continue - - logger.bind(tag=TAG).debug( - f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" - ) - if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") asyncio.run_coroutine_threadsafe( @@ -232,6 +224,14 @@ class TTSProvider(TTSProviderBase): ) continue + # 过滤旧消息：检查sentence_id是否匹配 + if message.sentence_id != self.conn.sentence_id: + continue + + logger.bind(tag=TAG).debug( + f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" + ) + if message.sentence_type == SentenceType.FIRST: # 初始化参数 try: diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 25ade2d6..4c3dd6be 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -320,12 +320,12 @@ class TTSProviderBase(ABC): while not self.conn.stop_event.is_set(): try: message = self.tts_text_queue.get(timeout=1) - # 过滤旧消息：检查sentence_id是否匹配 - if message.sentence_id != self.conn.sentence_id: - continue if self.conn.client_abort: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") continue + # 过滤旧消息：检查sentence_id是否匹配 + if message.sentence_id != self.conn.sentence_id: + continue if message.sentence_type == SentenceType.FIRST: self.current_sentence_id = message.sentence_id self.tts_stop_request = False diff --git a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py index f95acaba..0d783754 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -273,14 +273,6 @@ class TTSProvider(TTSProviderBase): try: message = self.tts_text_queue.get(timeout=1) - # 过滤旧消息：检查sentence_id是否匹配 - if message.sentence_id != self.conn.sentence_id: - continue - - logger.bind(tag=TAG).debug( - f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" - ) - if self.conn.client_abort: try: logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") @@ -299,6 +291,14 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).error(f"取消TTS会话失败: {str(e)}") continue + # 过滤旧消息：检查sentence_id是否匹配 + if message.sentence_id != self.conn.sentence_id: + continue + + logger.bind(tag=TAG).debug( + f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {message.sentence_id}" + ) + if message.sentence_type == SentenceType.FIRST: # 初始化参数 try: diff --git a/main/xiaozhi-server/core/providers/tts/xunfei_stream.py b/main/xiaozhi-server/core/providers/tts/xunfei_stream.py index 40d8a11b..6cdb8e19 100644 --- a/main/xiaozhi-server/core/providers/tts/xunfei_stream.py +++ b/main/xiaozhi-server/core/providers/tts/xunfei_stream.py @@ -155,6 +155,10 @@ class TTSProvider(TTSProviderBase): try: message = self.tts_text_queue.get(timeout=1) + if self.conn.client_abort: + logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") + continue + # 过滤旧消息：检查sentence_id是否匹配 if message.sentence_id != self.conn.sentence_id: continue @@ -169,10 +173,6 @@ class TTSProvider(TTSProviderBase): # 增加序列号 self.text_seq += 1 - if self.conn.client_abort: - logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程") - continue - if message.sentence_type == SentenceType.FIRST: # 初始化参数 try: