From b1bfaf5a5c5f70a83350e5e6758bf0070db32de8 Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Fri, 30 May 2025 15:47:32 +0800 Subject: [PATCH] =?UTF-8?q?update=EF=BC=9A=E5=8F=8C=E9=BA=A6=E5=AE=9E?= =?UTF-8?q?=E6=97=B6=E6=89=93=E6=96=AD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 4 +- .../xiaozhi-server/core/handle/helloHandle.py | 8 +- .../core/handle/receiveAudioHandle.py | 12 +- .../core/handle/sendAudioHandle.py | 3 + main/xiaozhi-server/core/handle/textHandle.py | 1 + .../xiaozhi-server/core/providers/asr/base.py | 1 - .../core/providers/asr/doubao.py | 287 +++++++++---- .../xiaozhi-server/core/providers/tts/base.py | 1 + .../providers/tts/huoshan_double_stream.py | 388 ++++++++++++++---- 9 files changed, 519 insertions(+), 186 deletions(-) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index b8f06f70..8dbecb34 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -83,6 +83,7 @@ class ConnectionHandler: # 客户端状态相关 self.client_abort = False + self.client_is_speaking = False self.client_listen_mode = "auto" # 线程任务相关 @@ -615,7 +616,7 @@ class ConnectionHandler: function_arguments = "" content_arguments = "" text_index = 0 - + self.client_abort = False for response in llm_responses: if self.client_abort: break @@ -850,6 +851,7 @@ class ConnectionHandler: self.report_queue.task_done() def clearSpeakStatus(self): + self.client_is_speaking = False self.logger.bind(tag=TAG).debug(f"清除服务端讲话状态") async def close(self, ws=None): diff --git a/main/xiaozhi-server/core/handle/helloHandle.py b/main/xiaozhi-server/core/handle/helloHandle.py index 6fd40401..45ef941a 100644 --- a/main/xiaozhi-server/core/handle/helloHandle.py +++ b/main/xiaozhi-server/core/handle/helloHandle.py @@ -92,7 +92,13 @@ async def wakeupWordsResponse(conn): """唤醒词响应""" wakeup_word = random.choice(WAKEUP_CONFIG["words"]) - result = conn.llm.response_no_stream(conn.config["prompt"], wakeup_word) + question = ( + "此刻用户正在和你说```" + + wakeup_word + + "```。\n请你根据以上用户的内容，进行简短回复，文字内容控制在15个字以内。\n" + + "请勿对这条内容本身进行任何解释和回应，仅返回对用户的内容的回复。" + ) + result = conn.llm.response_no_stream(conn.config["prompt"], question) if result is None or result == "": return tts_file = await asyncio.to_thread(conn.tts.to_tts, result) diff --git a/main/xiaozhi-server/core/handle/receiveAudioHandle.py b/main/xiaozhi-server/core/handle/receiveAudioHandle.py index 6776dd0f..b829afb5 100644 --- a/main/xiaozhi-server/core/handle/receiveAudioHandle.py +++ b/main/xiaozhi-server/core/handle/receiveAudioHandle.py @@ -1,6 +1,7 @@ from core.handle.sendAudioHandle import send_stt_message from core.handle.intentHandler import handle_user_intent from core.utils.output_counter import check_device_output_limit +from core.handle.abortHandle import handleAbortMessage import time from core.handle.sendAudioHandle import SentenceType from core.utils.util import audio_to_data @@ -12,11 +13,14 @@ async def handleAudioMessage(conn, audio): if conn.vad is None: conn.logger.bind(tag=TAG).warning("VAD模块未初始化，继续等待") return - if conn.asr is None: - conn.logger.bind(tag=TAG).warning("ASR模块未初始化，继续等待") + if conn.asr is None or not hasattr(conn.asr, "conn") or conn.asr.conn is None: + conn.logger.bind(tag=TAG).warning("ASR模块未初始化或通道未就绪，继续等待") return # 当前片段是否有人说话 have_voice = conn.vad.is_vad(conn, audio) + if have_voice: + if conn.client_is_speaking: + await handleAbortMessage(conn) # 设备长时间空闲检测，用于say goodbye await no_voice_close_connect(conn, have_voice) # 接收音频 @@ -35,6 +39,8 @@ async def startToChat(conn, text): ): await max_out_size(conn) return + if conn.client_is_speaking: + await handleAbortMessage(conn) # 首先进行意图分析 intent_handled = await handle_user_intent(conn, text) @@ -72,7 +78,7 @@ async def no_voice_close_connect(conn, have_voice): return prompt = end_prompt.get("prompt") if not prompt: - prompt = "请你以“时间过得真快”未来头，用富有感情、依依不舍的话来结束这场对话吧。！" + prompt = "请你以```时间过得真快```未来头，用富有感情、依依不舍的话来结束这场对话吧。！" await startToChat(conn, prompt) diff --git a/main/xiaozhi-server/core/handle/sendAudioHandle.py b/main/xiaozhi-server/core/handle/sendAudioHandle.py index 6625838c..85a71c8e 100644 --- a/main/xiaozhi-server/core/handle/sendAudioHandle.py +++ b/main/xiaozhi-server/core/handle/sendAudioHandle.py @@ -62,6 +62,7 @@ async def sendAudioMessage(conn, sentenceType, audios, text): # 发送结束消息（如果是最后一个文本） if conn.llm_finish_task and sentenceType == SentenceType.LAST: await send_tts_message(conn, "stop", None) + conn.client_is_speaking = False if conn.close_after_chat: await conn.close() @@ -88,6 +89,7 @@ async def sendAudio(conn, audios, pre_buffer=True): # 播放剩余音频帧 for opus_packet in remaining_audios: if conn.client_abort: + conn.client_abort = False return # 每分钟重置一次计时器 @@ -141,4 +143,5 @@ async def send_stt_message(conn, text): await conn.websocket.send( json.dumps({"type": "stt", "text": stt_text, "session_id": conn.session_id}) ) + conn.client_is_speaking = True await send_tts_message(conn, "start") diff --git a/main/xiaozhi-server/core/handle/textHandle.py b/main/xiaozhi-server/core/handle/textHandle.py index 602f72aa..043a8b16 100644 --- a/main/xiaozhi-server/core/handle/textHandle.py +++ b/main/xiaozhi-server/core/handle/textHandle.py @@ -58,6 +58,7 @@ async def handleTextMessage(conn, message): # 如果是唤醒词，且关闭了唤醒词回复，就不用回答 await send_stt_message(conn, original_text) await send_tts_message(conn, "stop", None) + conn.client_is_speaking = False elif is_wakeup_words: # 上报纯文字数据（复用ASR上报功能，但不提供音频数据） enqueue_asr_report(conn, "嘿，你好呀", []) diff --git a/main/xiaozhi-server/core/providers/asr/base.py b/main/xiaozhi-server/core/providers/asr/base.py index 1a3a384d..dbd1c339 100644 --- a/main/xiaozhi-server/core/providers/asr/base.py +++ b/main/xiaozhi-server/core/providers/asr/base.py @@ -47,7 +47,6 @@ class ASRProviderBase(ABC): if self.conn.client_voice_stop: asr_audio_task = copy.deepcopy(self.conn.asr_audio) self.conn.asr_audio.clear() - self.conn.client_abort = False # 音频太短了，无法识别 self.conn.reset_vad_states() if len(asr_audio_task) > 15: diff --git a/main/xiaozhi-server/core/providers/asr/doubao.py b/main/xiaozhi-server/core/providers/asr/doubao.py index 9ef1d411..da4098ad 100644 --- a/main/xiaozhi-server/core/providers/asr/doubao.py +++ b/main/xiaozhi-server/core/providers/asr/doubao.py @@ -33,6 +33,21 @@ class ASRProvider(ASRProviderBase): self.max_retries = 3 self.retry_delay = 2 # 重试延迟秒数 self.recv_lock = asyncio.Lock() # 添加接收锁 + self.reconnect_lock = asyncio.Lock() # 添加重连锁 + self.last_reconnect_time = 0 # 上次重连时间 + self.reconnect_cooldown = 1 # 增加重连冷却时间到10秒 + self.reconnect_count = 0 # 当前重连次数 + self.max_reconnect_count = 3 # 减少最大重连次数到3次 + self.asr_thread = None # ASR监听线程 + self.thread_lock = threading.Lock() # 线程管理锁 + self.is_reconnecting = False # 添加重连状态标志 + + # 添加会话管理相关属性 + self._session_lock = asyncio.Lock() # 会话操作的并发锁 + self._current_session_id = None # 当前会话ID + self._session_started = False # 会话是否已开始 + self._session_finished = False # 会话是否已结束 + self._session_close_event = asyncio.Event() # 添加会话关闭事件 self.appid = str(config.get("appid")) self.cluster = config.get("cluster") @@ -60,7 +75,6 @@ class ASRProvider(ASRProviderBase): self.asr_ws = None self.forward_task = None self.conn = None - self.asr_thread = None ################################################################################### # 豆包流式ASR重写父类的方法--开始 @@ -68,10 +82,18 @@ class ASRProvider(ASRProviderBase): async def open_audio_channels(self, conn): await super().open_audio_channels(conn) - retry_count = 0 - while retry_count < self.max_retries: - try: - # 确保关闭旧的连接 + async with self._session_lock: + # 如果正在重连，等待重连完成 + if self.is_reconnecting: + logger.bind(tag=TAG).info("等待当前重连完成...") + await self._session_close_event.wait() + self._session_close_event.clear() + + # 如果已有会话未结束，先关闭它 + if self._session_started and not self._session_finished: + logger.bind(tag=TAG).warning( + f"发现未关闭的会话 {self._current_session_id}，正在关闭..." + ) if self.asr_ws is not None: try: await self.asr_ws.close() @@ -79,62 +101,94 @@ class ASRProvider(ASRProviderBase): logger.bind(tag=TAG).warning(f"关闭旧连接时发生错误: {e}") finally: self.asr_ws = None + self._session_finished = True + self._session_close_event.set() - headers = self.token_auth() if self.auth_method == "token" else None - self.asr_ws = await websockets.connect( - self.ws_url, - additional_headers=headers, - max_size=1000000000, - ping_interval=None, # 禁用ping，因为服务器可能不支持 - ping_timeout=None, - close_timeout=10, - ) + # 重置会话状态 + self._current_session_id = str(uuid.uuid4()) + self._session_started = True + self._session_finished = False + self.is_reconnecting = True - # 发送初始化请求 - request_params = self.construct_request(str(uuid.uuid4())) - try: - payload_bytes = str.encode(json.dumps(request_params)) - payload_bytes = gzip.compress(payload_bytes) - full_client_request = self.generate_header() - full_client_request.extend((len(payload_bytes)).to_bytes(4, "big")) - full_client_request.extend(payload_bytes) - await self.asr_ws.send(full_client_request) - logger.bind(tag=TAG).debug(f"发送初始化请求: {request_params}") - except Exception as e: - logger.bind(tag=TAG).error(f"发送初始化请求失败: {e}") - raise e + try: + retry_count = 0 + while retry_count < self.max_retries: + try: + headers = ( + self.token_auth() if self.auth_method == "token" else None + ) + self.asr_ws = await websockets.connect( + self.ws_url, + additional_headers=headers, + max_size=1000000000, + ping_interval=None, + ping_timeout=None, + close_timeout=10, + ) - # 等待初始化响应 - try: - init_res = await self.asr_ws.recv() - result = self.parse_response(init_res) - logger.bind(tag=TAG).info(f"ASR服务初始化响应: {result}") - except Exception as e: - logger.bind(tag=TAG).error(f"ASR服务初始化失败: {e}") - raise e + # 发送初始化请求 + request_params = self.construct_request( + self._current_session_id + ) + try: + payload_bytes = str.encode(json.dumps(request_params)) + payload_bytes = gzip.compress(payload_bytes) + full_client_request = self.generate_header() + full_client_request.extend( + (len(payload_bytes)).to_bytes(4, "big") + ) + full_client_request.extend(payload_bytes) + await self.asr_ws.send(full_client_request) + except Exception as e: + logger.bind(tag=TAG).error(f"发送初始化请求失败: {e}") + raise e - # 启动接收ASR结果的异步任务 - asr_priority = threading.Thread( - target=self._start_monitor_asr_response_thread, daemon=True - ) - asr_priority.start() - return + # 等待初始化响应 + try: + init_res = await self.asr_ws.recv() + self.parse_response(init_res) + except Exception as e: + logger.bind(tag=TAG).error(f"ASR服务初始化失败: {e}") + raise e - except websockets.exceptions.WebSocketException as e: - retry_count += 1 - if retry_count < self.max_retries: - logger.bind(tag=TAG).warning( - f"WebSocket连接失败，正在进行第{retry_count}次重试: {e}" - ) - await asyncio.sleep(self.retry_delay) - else: - logger.bind(tag=TAG).error( - f"WebSocket连接失败，已达到最大重试次数: {e}" - ) - raise - except Exception as e: - logger.bind(tag=TAG).error(f"WebSocket连接发生未知错误: {e}") - raise + # 启动接收ASR结果的异步任务 + with self.thread_lock: + if ( + self.asr_thread is None + or not self.asr_thread.is_alive() + ): + logger.bind(tag=TAG).info("创建新的ASR监听线程...") + self.asr_thread = threading.Thread( + target=self._start_monitor_asr_response_thread, + daemon=True, + ) + self.asr_thread.start() + # 等待一小段时间确保线程启动 + await asyncio.sleep(0.1) + if not self.asr_thread.is_alive(): + logger.bind(tag=TAG).error("ASR监听线程启动失败") + raise Exception("ASR监听线程启动失败") + logger.bind(tag=TAG).info("ASR监听线程已启动") + return + + except websockets.exceptions.WebSocketException as e: + retry_count += 1 + if retry_count < self.max_retries: + logger.bind(tag=TAG).warning( + f"WebSocket连接失败，正在进行第{retry_count}次重试: {e}" + ) + await asyncio.sleep(self.retry_delay) + else: + logger.bind(tag=TAG).warning( + f"WebSocket连接失败，已达到最大重试次数: {e}" + ) + raise + except Exception as e: + logger.bind(tag=TAG).error(f"WebSocket连接发生未知错误: {e}") + raise + finally: + self.is_reconnecting = False + self._session_close_event.set() async def receive_audio(self, audio, _): if not isinstance(audio, bytes): @@ -150,7 +204,7 @@ class ASRProvider(ASRProviderBase): if self.asr_ws: await self.asr_ws.send(audio_request) except Exception as e: - logger.bind(tag=TAG).error(f"发送音频数据时发生错误: {e}") + logger.bind(tag=TAG).debug(f"发送音频数据时发生错误: {e}") ################################################################################### # 豆包流式ASR重写父类的方法--结束 @@ -246,23 +300,61 @@ class ASRProvider(ASRProviderBase): def _start_monitor_asr_response_thread(self): # 初始化链接 - asyncio.run_coroutine_threadsafe( - self._forward_asr_results(), loop=self.conn.loop - ) + try: + with self.thread_lock: + if self.conn is None or self.conn.loop is None: + logger.bind(tag=TAG).error( + "无法启动ASR监听线程：conn或loop未初始化" + ) + return + + try: + logger.bind(tag=TAG).info("开始启动ASR监听...") + asyncio.run_coroutine_threadsafe( + self._forward_asr_results(), loop=self.conn.loop + ) + logger.bind(tag=TAG).info("ASR监听已启动") + except Exception as e: + logger.bind(tag=TAG).error(f"启动ASR监听线程失败: {e}") + except Exception as e: + logger.bind(tag=TAG).error(f"ASR监听线程发生未预期的错误: {e}") async def _forward_asr_results(self): try: while not self.conn.stop_event.is_set(): try: if self.asr_ws is None: - logger.bind(tag=TAG).info("尝试重新连接ASR服务...") - await self.open_audio_channels(self.conn) - continue + # 检查是否需要重连 + async with self.reconnect_lock: + current_time = asyncio.get_event_loop().time() + if ( + current_time - self.last_reconnect_time + < self.reconnect_cooldown + ): + await asyncio.sleep(1) + continue + + if self.reconnect_count >= self.max_reconnect_count: + logger.bind(tag=TAG).error( + "达到最大重连次数限制，停止重连" + ) + await asyncio.sleep(self.reconnect_cooldown) + self.reconnect_count = 0 + continue + + self.last_reconnect_time = current_time + self.reconnect_count += 1 + logger.bind(tag=TAG).info( + f"尝试重新连接ASR服务... (第{self.reconnect_count}次)" + ) + await self.open_audio_channels(self.conn) + continue # 使用锁来确保同一时间只有一个协程在接收数据 async with self.recv_lock: response = await self.asr_ws.recv() result = self.parse_response(response) + # 检查是否需要重连 if result.get("need_reconnect", False): logger.bind(tag=TAG).info( @@ -290,6 +382,7 @@ class ASRProvider(ASRProviderBase): self.text = utterance["text"] await self.handle_voice_stop(None) break + except websockets.ConnectionClosed: logger.bind(tag=TAG).debug("ASR服务连接已关闭，准备重连...") # 确保关闭旧连接 @@ -301,34 +394,16 @@ class ASRProvider(ASRProviderBase): finally: self.asr_ws = None - retry_count = 0 - while ( - retry_count < self.max_retries - and not self.conn.stop_event.is_set() - ): - try: - logger.bind(tag=TAG).info( - f"正在进行第{retry_count + 1}次重连尝试..." - ) - await self.open_audio_channels(self.conn) - break - except Exception as e: - retry_count += 1 - if retry_count < self.max_retries: - logger.bind(tag=TAG).warning( - f"重连失败，等待{self.retry_delay}秒后重试: {e}" - ) - await asyncio.sleep(self.retry_delay) - else: - logger.bind(tag=TAG).error( - f"重连失败，已达到最大重试次数: {e}" - ) - await asyncio.sleep( - self.retry_delay - ) # 继续等待，以便后续重试 + # 等待冷却时间 + await asyncio.sleep(self.reconnect_cooldown) + continue + except Exception as e: if not self.conn.stop_event.is_set(): - await asyncio.sleep(2) # 增加重试延迟 + logger.bind(tag=TAG).error(f"ASR监听发生错误: {e}") + await asyncio.sleep(self.retry_delay) + continue + except Exception as e: logger.bind(tag=TAG).error(f"ASR监听线程发生错误: {e}") # 确保在发生严重错误时也能继续尝试重连 @@ -422,8 +497,38 @@ class ASRProvider(ASRProviderBase): f"ASR错误: {error_message} (错误码: {error_code})" ) - # 如果是识别相关错误（>=1020），标记需要重连 - if error_code >= 1020: + # 如果是识别相关错误，标记需要重连 + if error_code >= 1020 or error_code == 1001: result["need_reconnect"] = True return result + + async def close_session(self): + """关闭当前会话""" + async with self._session_lock: + if not self._session_started: + logger.bind(tag=TAG).warning("尝试关闭未开始的会话") + return + + if self._session_finished: + logger.bind(tag=TAG).warning( + f"会话 {self._current_session_id} 已经关闭" + ) + return + + try: + if self.asr_ws is not None: + await self.asr_ws.close() + except Exception as e: + logger.bind(tag=TAG).warning(f"关闭WebSocket连接时发生错误: {e}") + finally: + self.asr_ws = None + self._session_finished = True + self._session_started = False + self._current_session_id = None + # 重置重连计数 + self.reconnect_count = 0 + + async def close(self): + """资源清理方法""" + await self.close_session() diff --git a/main/xiaozhi-server/core/providers/tts/base.py b/main/xiaozhi-server/core/providers/tts/base.py index 0b17eef5..36b6ecbf 100644 --- a/main/xiaozhi-server/core/providers/tts/base.py +++ b/main/xiaozhi-server/core/providers/tts/base.py @@ -216,6 +216,7 @@ class TTSProviderBase(ABC): logger.bind(tag=TAG).error( f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" ) + continue def _audio_play_priority_thread(self): while not self.conn.stop_event.is_set(): 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 25548958..de50c18c 100644 --- a/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py +++ b/main/xiaozhi-server/core/providers/tts/huoshan_double_stream.py @@ -6,6 +6,7 @@ import asyncio import threading import traceback import websockets +import time from config.logger import setup_logging from core.utils import opus_encoder_utils from core.utils.util import check_model_key @@ -138,6 +139,7 @@ class Response: class TTSProvider(TTSProviderBase): def __init__(self, config, delete_audio_file): super().__init__(config, delete_audio_file) + self.ws = None # 初始化ws属性 self.interface_type = InterfaceType.DUAL_STREAM self.appId = config.get("appid") self.access_token = config.get("access_token") @@ -166,59 +168,133 @@ class TTSProvider(TTSProviderBase): self._current_session_id = None # 当前会话ID self._session_started = False # 会话是否已开始 self._session_finished = False # 会话是否已结束 + self._connection_ready = False # 连接是否就绪 + self._reconnect_attempts = 0 # 重连尝试次数 + self._max_reconnect_attempts = 3 # 最大重连次数 ################################################################################### # 火山双流式TTS重写父类的方法--开始 ################################################################################### async def open_audio_channels(self, conn): - await super().open_audio_channels(conn) - 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 = 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() + try: + await super().open_audio_channels(conn) + await self._ensure_connection() + tts_priority = threading.Thread( + target=self._start_monitor_tts_response_thread, daemon=True + ) + tts_priority.start() + except Exception as e: + logger.bind(tag=TAG).error(f"Failed to open audio channels: {str(e)}") + self.ws = None + raise + + async def _ensure_connection(self): + """确保WebSocket连接可用""" + try: + if self.ws is None: + logger.bind(tag=TAG).info("WebSocket连接不存在，开始建立新连接...") + 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 = await websockets.connect( + self.ws_url, additional_headers=ws_header, max_size=1000000000 + ) + self._connection_ready = True + self._reconnect_attempts = 0 + logger.bind(tag=TAG).info("WebSocket连接建立成功") + else: + # 尝试发送ping来检查连接是否还活着 + try: + logger.bind(tag=TAG).debug("检查WebSocket连接状态...") + pong_waiter = await self.ws.ping() + await asyncio.wait_for(pong_waiter, timeout=1.0) + logger.bind(tag=TAG).debug("WebSocket连接状态正常") + except (asyncio.TimeoutError, websockets.ConnectionClosed): + # 如果ping失败，重新建立连接 + logger.bind(tag=TAG).warning("WebSocket连接已断开，准备重新连接...") + try: + await self.ws.close() + except: + pass + self.ws = None + self._connection_ready = False + # 重新建立连接 + await self._ensure_connection() + except Exception as e: + logger.bind(tag=TAG).error(f"确保连接失败: {str(e)}") + self._connection_ready = False + self.ws = None + raise def tts_text_priority_thread(self): + logger.bind(tag=TAG).info("TTS文本处理线程启动") while not self.conn.stop_event.is_set(): try: + logger.bind(tag=TAG).debug("等待TTS文本队列消息...") message = self.tts_text_queue.get(timeout=1) - logger.bind(tag=TAG).debug( - f"TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name}" + logger.bind(tag=TAG).info( + f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}" ) if message.sentence_type == SentenceType.FIRST: # 初始化参数 - future = asyncio.run_coroutine_threadsafe( - self.start_session(self.conn.sentence_id), loop=self.conn.loop - ) - future.result() - self.tts_audio_first_sentence = True - self.before_stop_play_files.clear() - elif ContentType.TEXT == message.content_type: - if message.content_detail: + try: + logger.bind(tag=TAG).info("开始启动TTS会话...") future = asyncio.run_coroutine_threadsafe( - self.text_to_speak(message.content_detail, None), + self.start_session(self.conn.sentence_id), loop=self.conn.loop, ) future.result() + self.tts_audio_first_sentence = True + self.before_stop_play_files.clear() + logger.bind(tag=TAG).info("TTS会话启动成功") + except Exception as e: + logger.bind(tag=TAG).error(f"启动TTS会话失败: {str(e)}") + # 直接跳过当前消息，不重新入队 + time.sleep(1) + continue + elif ContentType.TEXT == message.content_type: + if message.content_detail: + try: + logger.bind(tag=TAG).info( + f"开始发送TTS文本: {message.content_detail}" + ) + future = asyncio.run_coroutine_threadsafe( + self.text_to_speak(message.content_detail, None), + loop=self.conn.loop, + ) + future.result() + logger.bind(tag=TAG).info("TTS文本发送成功") + except Exception as e: + logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}") + # 直接跳过当前消息，不重新入队 + time.sleep(1) + continue 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: - future = asyncio.run_coroutine_threadsafe( - self.finish_session(self.conn.sentence_id), loop=self.conn.loop - ) - future.result() + try: + logger.bind(tag=TAG).info("开始结束TTS会话...") + future = asyncio.run_coroutine_threadsafe( + self.finish_session(self.conn.sentence_id), + loop=self.conn.loop, + ) + future.result() + logger.bind(tag=TAG).info("TTS会话结束成功") + except Exception as e: + logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}") + # 直接跳过当前消息，不重新入队 + time.sleep(1) + continue except queue.Empty: continue @@ -226,11 +302,30 @@ class TTSProvider(TTSProviderBase): logger.bind(tag=TAG).error( f"处理TTS文本失败: {str(e)}, 类型: {type(e).__name__}, 堆栈: {traceback.format_exc()}" ) + # 如果是WebSocket连接关闭错误，等待一段时间后继续 + if "non-exist session" in str(e): + time.sleep(1) + continue async def text_to_speak(self, text, _): - # 发送文本 - await self.send_text(self.speaker, text, self.conn.sentence_id) - return + """发送文本到TTS服务""" + try: + # 确保WebSocket连接可用 + if not self._connection_ready or self.ws is None: + logger.bind(tag=TAG).warning("WebSocket连接不可用，尝试重新连接...") + await self._ensure_connection() + + # 发送文本 + await self.send_text(self.speaker, text, self.conn.sentence_id) + return + except Exception as e: + logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}") + # 如果是连接问题，尝试重新连接 + if isinstance(e, websockets.ConnectionClosed): + self._connection_ready = False + self.ws = None + await self._handle_connection_error() + raise ################################################################################### # 火山双流式TTS重写父类的方法--结束 @@ -308,14 +403,21 @@ class TTSProvider(TTSProviderBase): async def send_event( self, header: bytes, optional: bytes | None = None, payload: bytes = None ): - full_client_request = bytearray(header) - if optional is not None: - full_client_request.extend(optional) - if payload is not None: - payload_size = len(payload).to_bytes(4, "big", signed=True) - full_client_request.extend(payload_size) - full_client_request.extend(payload) - await self.ws.send(full_client_request) + try: + full_client_request = bytearray(header) + if optional is not None: + full_client_request.extend(optional) + if payload is not None: + payload_size = len(payload).to_bytes(4, "big", signed=True) + full_client_request.extend(payload_size) + full_client_request.extend(payload) + await self.ws.send(full_client_request) + except websockets.ConnectionClosed: + if await self._handle_connection_error(): + # 重连成功后重试发送 + await self.ws.send(full_client_request) + else: + raise async def send_text(self, speaker: str, text: str, session_id): header = Header( @@ -450,63 +552,154 @@ class TTSProvider(TTSProviderBase): return async def start_session(self, session_id): - logger.bind(tag=TAG).debug(f"开始会话～～{session_id}") - async with self._session_lock: - # 如果已有会话未结束，先关闭它 - if self._session_started and not self._session_finished: - logger.bind(tag=TAG).warning( - f"发现未关闭的会话 {self._current_session_id}，正在关闭..." - ) - await self.finish_session(self._current_session_id) + logger.bind(tag=TAG).info(f"开始会话～～{session_id}") + try: + async with self._session_lock: + try: + # 确保连接可用 + logger.bind(tag=TAG).info("检查WebSocket连接状态...") + await asyncio.wait_for(self._ensure_connection(), timeout=5) - # 重置会话状态 - self._current_session_id = session_id - self._session_started = True - self._session_finished = False + # 如果已有会话未结束，先关闭它 + if self._session_started and not self._session_finished: + logger.bind(tag=TAG).warning( + f"发现未关闭的会话 {self._current_session_id}，正在关闭..." + ) + try: + await asyncio.wait_for( + self.finish_session(self._current_session_id), timeout=5 + ) + except Exception as e: + logger.bind(tag=TAG).error(f"关闭旧会话失败: {str(e)}") + # 强制重置会话状态 + self._session_started = False + self._session_finished = True + self._current_session_id = None - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional( - event=EVENT_StartSession, sessionId=session_id - ).as_bytes() - payload = self.get_payload_bytes( - event=EVENT_StartSession, speaker=self.speaker - ) - await self.send_event(header, optional, payload) + # 重置会话状态 + self._current_session_id = session_id + self._session_started = True + self._session_finished = False + logger.bind(tag=TAG).info( + f"会话状态已更新 - 开始: {self._session_started}, 结束: {self._session_finished}" + ) + + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional( + event=EVENT_StartSession, sessionId=session_id + ).as_bytes() + payload = self.get_payload_bytes( + event=EVENT_StartSession, speaker=self.speaker + ) + await asyncio.wait_for( + self.send_event(header, optional, payload), timeout=5 + ) + logger.bind(tag=TAG).info("会话启动请求已发送") + except Exception as e: + logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}") + self._session_started = False + self._session_finished = True + self._current_session_id = None + raise + except asyncio.TimeoutError: + logger.bind(tag=TAG).error(f"启动会话超时: {session_id}") + # 超时后强制重置会话状态 + self._session_started = False + self._session_finished = True + self._current_session_id = None + # 尝试关闭WebSocket连接 + if self.ws: + try: + await self.ws.close() + except: + pass + self.ws = None + except Exception as e: + logger.bind(tag=TAG).error(f"启动会话时发生未知错误: {str(e)}") + # 发生未知错误时也重置会话状态 + self._session_started = False + self._session_finished = True + self._current_session_id = None async def finish_session(self, session_id): - logger.bind(tag=TAG).debug(f"关闭会话～～{session_id}") - async with self._session_lock: - # 检查会话状态 - if not self._session_started: - logger.bind(tag=TAG).warning(f"尝试关闭未开始的会话 {session_id}") - return + logger.bind(tag=TAG).info(f"关闭会话～～{session_id}") + try: + async with self._session_lock: + try: + # 检查会话状态 + if not self._session_started: + logger.bind(tag=TAG).warning( + f"尝试关闭未开始的会话 {session_id}" + ) + return - if self._session_finished: - logger.bind(tag=TAG).warning(f"会话 {session_id} 已经关闭") - return + if self._session_finished: + logger.bind(tag=TAG).warning(f"会话 {session_id} 已经关闭") + return - if self._current_session_id != session_id: - logger.bind(tag=TAG).warning( - f"尝试关闭错误的会话 {session_id}，当前会话为 {self._current_session_id}" - ) - return + if self._current_session_id != session_id: + logger.bind(tag=TAG).warning( + f"尝试关闭错误的会话 {session_id}，当前会话为 {self._current_session_id}" + ) + # 即使会话ID不匹配，也尝试关闭当前会话 + if self._current_session_id: + session_id = self._current_session_id - header = Header( - message_type=FULL_CLIENT_REQUEST, - message_type_specific_flags=MsgTypeFlagWithEvent, - serial_method=JSON, - ).as_bytes() - optional = Optional( - event=EVENT_FinishSession, sessionId=session_id - ).as_bytes() - payload = str.encode("{}") - await self.send_event(header, optional, payload) + # 确保WebSocket连接可用 + if self.ws is None: + logger.bind(tag=TAG).warning( + "WebSocket连接不存在，尝试重新连接..." + ) + await asyncio.wait_for(self._ensure_connection(), timeout=5) - # 更新会话状态 + header = Header( + message_type=FULL_CLIENT_REQUEST, + message_type_specific_flags=MsgTypeFlagWithEvent, + serial_method=JSON, + ).as_bytes() + optional = Optional( + event=EVENT_FinishSession, sessionId=session_id + ).as_bytes() + payload = str.encode("{}") + await asyncio.wait_for( + self.send_event(header, optional, payload), timeout=5 + ) + logger.bind(tag=TAG).info("会话结束请求已发送") + + # 更新会话状态 + self._session_finished = True + self._session_started = False + self._current_session_id = None + logger.bind(tag=TAG).info( + "会话状态已更新 - 开始: False, 结束: True" + ) + except Exception as e: + logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}") + # 即使发生错误，也要重置会话状态 + self._session_finished = True + self._session_started = False + self._current_session_id = None + raise + except asyncio.TimeoutError: + logger.bind(tag=TAG).error(f"关闭会话超时: {session_id}") + # 超时后强制重置会话状态 + self._session_finished = True + self._session_started = False + self._current_session_id = None + # 尝试关闭WebSocket连接 + if self.ws: + try: + await self.ws.close() + except: + pass + self.ws = None + except Exception as e: + logger.bind(tag=TAG).error(f"关闭会话时发生未知错误: {str(e)}") + # 发生未知错误时也重置会话状态 self._session_finished = True self._session_started = False self._current_session_id = None @@ -527,3 +720,20 @@ class TTSProvider(TTSProviderBase): 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 _handle_connection_error(self): + """处理连接错误""" + if self._reconnect_attempts < self._max_reconnect_attempts: + self._reconnect_attempts += 1 + logger.bind(tag=TAG).warning( + f"尝试重新连接 (第{self._reconnect_attempts}次)" + ) + try: + await self._ensure_connection() + return True + except Exception as e: + logger.bind(tag=TAG).error(f"重新连接失败: {str(e)}") + return False + else: + logger.bind(tag=TAG).error("达到最大重连次数，放弃重连") + return False