Merge pull request #1419 from xinnan-tech/fix-server

update：双麦实时打断

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):

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)

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:
+    if conn.asr is None or not hasattr(conn.asr, "conn") or conn.asr.conn is None:
-        conn.logger.bind(tag=TAG).warning("ASR模块未初始化，继续等待")
+        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)
 
 

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")

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, "嘿，你好呀", [])

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:

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
+        async with self._session_lock:
-        while retry_count < self.max_retries:
+            # 如果正在重连，等待重连完成
-            try:
+            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._current_session_id = str(uuid.uuid4())
-                    self.ws_url,
+            self._session_started = True
-                    additional_headers=headers,
+            self._session_finished = False
-                    max_size=1000000000,
+            self.is_reconnecting = True
-                    ping_interval=None,  # 禁用ping，因为服务器可能不支持
-                    ping_timeout=None,
-                    close_timeout=10,
-                )
 
-                # 发送初始化请求
+            try:
-                request_params = self.construct_request(str(uuid.uuid4()))
+                retry_count = 0
-                try:
+                while retry_count < self.max_retries:
-                    payload_bytes = str.encode(json.dumps(request_params))
+                    try:
-                    payload_bytes = gzip.compress(payload_bytes)
+                        headers = (
-                    full_client_request = self.generate_header()
+                            self.token_auth() if self.auth_method == "token" else None
-                    full_client_request.extend((len(payload_bytes)).to_bytes(4, "big"))
+                        )
-                    full_client_request.extend(payload_bytes)
+                        self.asr_ws = await websockets.connect(
-                    await self.asr_ws.send(full_client_request)
+                            self.ws_url,
-                    logger.bind(tag=TAG).debug(f"发送初始化请求: {request_params}")
+                            additional_headers=headers,
-                except Exception as e:
+                            max_size=1000000000,
-                    logger.bind(tag=TAG).error(f"发送初始化请求失败: {e}")
+                            ping_interval=None,
-                    raise e
+                            ping_timeout=None,
+                            close_timeout=10,
+                        )
 
-                # 等待初始化响应
+                        # 发送初始化请求
-                try:
+                        request_params = self.construct_request(
-                    init_res = await self.asr_ws.recv()
+                            self._current_session_id
-                    result = self.parse_response(init_res)
+                        )
-                    logger.bind(tag=TAG).info(f"ASR服务初始化响应: {result}")
+                        try:
-                except Exception as e:
+                            payload_bytes = str.encode(json.dumps(request_params))
-                    logger.bind(tag=TAG).error(f"ASR服务初始化失败: {e}")
+                            payload_bytes = gzip.compress(payload_bytes)
-                    raise e
+                            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(
+                        try:
-                    target=self._start_monitor_asr_response_thread, daemon=True
+                            init_res = await self.asr_ws.recv()
-                )
+                            self.parse_response(init_res)
-                asr_priority.start()
+                        except Exception as e:
-                return
+                            logger.bind(tag=TAG).error(f"ASR服务初始化失败: {e}")
+                            raise e
 
-            except websockets.exceptions.WebSocketException as e:
+                        # 启动接收ASR结果的异步任务
-                retry_count += 1
+                        with self.thread_lock:
-                if retry_count < self.max_retries:
+                            if (
-                    logger.bind(tag=TAG).warning(
+                                self.asr_thread is None
-                        f"WebSocket连接失败，正在进行第{retry_count}次重试: {e}"
+                                or not self.asr_thread.is_alive()
-                    )
+                            ):
-                    await asyncio.sleep(self.retry_delay)
+                                logger.bind(tag=TAG).info("创建新的ASR监听线程...")
-                else:
+                                self.asr_thread = threading.Thread(
-                    logger.bind(tag=TAG).error(
+                                    target=self._start_monitor_asr_response_thread,
-                        f"WebSocket连接失败，已达到最大重试次数: {e}"
+                                    daemon=True,
-                    )
+                                )
-                    raise
+                                self.asr_thread.start()
-            except Exception as e:
+                                # 等待一小段时间确保线程启动
-                logger.bind(tag=TAG).error(f"WebSocket连接发生未知错误: {e}")
+                                await asyncio.sleep(0.1)
-                raise
+                                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(
+        try:
-            self._forward_asr_results(), loop=self.conn.loop
+            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)
+                        async with self.reconnect_lock:
-                        continue
+                            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 (
+                    await asyncio.sleep(self.reconnect_cooldown)
-                        retry_count < self.max_retries
+                    continue
-                        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
-                                )  # 继续等待，以便后续重试
                 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()

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():

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)
+        try:
-        ws_header = {
+            await super().open_audio_channels(conn)
-            "X-Api-App-Key": self.appId,
+            await self._ensure_connection()
-            "X-Api-Access-Key": self.access_token,
+            tts_priority = threading.Thread(
-            "X-Api-Resource-Id": self.resource_id,
+                target=self._start_monitor_tts_response_thread, daemon=True
-            "X-Api-Connect-Id": uuid.uuid4(),
+            )
-        }
+            tts_priority.start()
-        self.ws = await websockets.connect(
+        except Exception as e:
-            self.ws_url, additional_headers=ws_header, max_size=1000000000
+            logger.bind(tag=TAG).error(f"Failed to open audio channels: {str(e)}")
-        )
+            self.ws = None
-        tts_priority = threading.Thread(
+            raise
-            target=self._start_monitor_tts_response_thread, daemon=True
+
-        )
+    async def _ensure_connection(self):
-        tts_priority.start()
+        """确保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(
+                logger.bind(tag=TAG).info(
-                    f"TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name}"
+                    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(
+                    try:
-                        self.start_session(self.conn.sentence_id), loop=self.conn.loop
+                        logger.bind(tag=TAG).info("开始启动TTS会话...")
-                    )
-                    future.result()
-                    self.tts_audio_first_sentence = True
-                    self.before_stop_play_files.clear()
-                elif ContentType.TEXT == message.content_type:
-                    if message.content_detail:
                         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(
+                    try:
-                        self.finish_session(self.conn.sentence_id), loop=self.conn.loop
+                        logger.bind(tag=TAG).info("开始结束TTS会话...")
-                    )
+                        future = asyncio.run_coroutine_threadsafe(
-                    future.result()
+                            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, _):
-        # 发送文本
+        """发送文本到TTS服务"""
-        await self.send_text(self.speaker, text, self.conn.sentence_id)
+        try:
-        return
+            # 确保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)
+        try:
-        if optional is not None:
+            full_client_request = bytearray(header)
-            full_client_request.extend(optional)
+            if optional is not None:
-        if payload is not None:
+                full_client_request.extend(optional)
-            payload_size = len(payload).to_bytes(4, "big", signed=True)
+            if payload is not None:
-            full_client_request.extend(payload_size)
+                payload_size = len(payload).to_bytes(4, "big", signed=True)
-            full_client_request.extend(payload)
+                full_client_request.extend(payload_size)
-        await self.ws.send(full_client_request)
+                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}")
+        logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
-        async with self._session_lock:
+        try:
-            # 如果已有会话未结束，先关闭它
+            async with self._session_lock:
-            if self._session_started and not self._session_finished:
+                try:
-                logger.bind(tag=TAG).warning(
+                    # 确保连接可用
-                    f"发现未关闭的会话 {self._current_session_id}，正在关闭..."
+                    logger.bind(tag=TAG).info("检查WebSocket连接状态...")
-                )
+                    await asyncio.wait_for(self._ensure_connection(), timeout=5)
-                await self.finish_session(self._current_session_id)
 
-            # 重置会话状态
+                    # 如果已有会话未结束，先关闭它
-            self._current_session_id = session_id
+                    if self._session_started and not self._session_finished:
-            self._session_started = True
+                        logger.bind(tag=TAG).warning(
-            self._session_finished = False
+                            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,
+                    self._current_session_id = session_id
-                message_type_specific_flags=MsgTypeFlagWithEvent,
+                    self._session_started = True
-                serial_method=JSON,
+                    self._session_finished = False
-            ).as_bytes()
+                    logger.bind(tag=TAG).info(
-            optional = Optional(
+                        f"会话状态已更新 - 开始: {self._session_started}, 结束: {self._session_finished}"
-                event=EVENT_StartSession, sessionId=session_id
+                    )
-            ).as_bytes()
+
-            payload = self.get_payload_bytes(
+                    header = Header(
-                event=EVENT_StartSession, speaker=self.speaker
+                        message_type=FULL_CLIENT_REQUEST,
-            )
+                        message_type_specific_flags=MsgTypeFlagWithEvent,
-            await self.send_event(header, optional, payload)
+                        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}")
+        logger.bind(tag=TAG).info(f"关闭会话～～{session_id}")
-        async with self._session_lock:
+        try:
-            # 检查会话状态
+            async with self._session_lock:
-            if not self._session_started:
+                try:
-                logger.bind(tag=TAG).warning(f"尝试关闭未开始的会话 {session_id}")
+                    # 检查会话状态
-                return
+                    if not self._session_started:
+                        logger.bind(tag=TAG).warning(
+                            f"尝试关闭未开始的会话 {session_id}"
+                        )
+                        return
 
-            if self._session_finished:
+                    if self._session_finished:
-                logger.bind(tag=TAG).warning(f"会话 {session_id} 已经关闭")
+                        logger.bind(tag=TAG).warning(f"会话 {session_id} 已经关闭")
-                return
+                        return
 
-            if self._current_session_id != session_id:
+                    if self._current_session_id != session_id:
-                logger.bind(tag=TAG).warning(
+                        logger.bind(tag=TAG).warning(
-                    f"尝试关闭错误的会话 {session_id}，当前会话为 {self._current_session_id}"
+                            f"尝试关闭错误的会话 {session_id}，当前会话为 {self._current_session_id}"
-                )
+                        )
-                return
+                        # 即使会话ID不匹配，也尝试关闭当前会话
+                        if self._current_session_id:
+                            session_id = self._current_session_id
 
-            header = Header(
+                    # 确保WebSocket连接可用
-                message_type=FULL_CLIENT_REQUEST,
+                    if self.ws is None:
-                message_type_specific_flags=MsgTypeFlagWithEvent,
+                        logger.bind(tag=TAG).warning(
-                serial_method=JSON,
+                            "WebSocket连接不存在，尝试重新连接..."
-            ).as_bytes()
+                        )
-            optional = Optional(
+                        await asyncio.wait_for(self._ensure_connection(), timeout=5)
-                event=EVENT_FinishSession, sessionId=session_id
-            ).as_bytes()
-            payload = str.encode("{}")
-            await self.send_event(header, optional, payload)
 
-            # 更新会话状态
+                    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