Merge pull request #1446 from xinnan-tech/tts-huoshan-fix

update:优化火山引擎双流TTS连接方式

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

@@ -11,10 +11,8 @@ TAG = __name__
 
 async def handleAudioMessage(conn, audio):
     if conn.vad is None:
-        conn.logger.bind(tag=TAG).warning("VAD模块未初始化，继续等待")
         return
     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)

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

@@ -178,6 +178,9 @@ class TTSProviderBase(ABC):
         while not self.conn.stop_event.is_set():
             try:
                 message = self.tts_text_queue.get(timeout=1)
+                if self.conn.client_abort:
+                    logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
+                    continue
                 if message.sentence_type == SentenceType.FIRST:
                     # 初始化参数
                     self.tts_stop_request = False

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

@@ -3,14 +3,13 @@ import uuid
 import json
 import queue
 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
 from core.providers.tts.base import TTSProviderBase
+from core.handle.abortHandle import handleAbortMessage
 from core.providers.tts.dto.dto import SentenceType, ContentType, InterfaceType
 
 TAG = __name__
@@ -139,7 +138,7 @@ class Response:
 class TTSProvider(TTSProviderBase):
     def __init__(self, config, delete_audio_file):
         super().__init__(config, delete_audio_file)
-        self.ws = None  # 初始化ws属性
+        self.ws = None
         self.interface_type = InterfaceType.DUAL_STREAM
         self.appId = config.get("appid")
         self.access_token = config.get("access_token")
@@ -154,83 +153,43 @@ class TTSProvider(TTSProviderBase):
         self.authorization = config.get("authorization")
         self.header = {"Authorization": f"{self.authorization}{self.access_token}"}
         self.enable_two_way = True
-        self.start_connection_flag = False
         self.tts_text = ""
-        # 合成文字语音后，播放的音频文件列表
         self.before_stop_play_files = []
         self.opus_encoder = opus_encoder_utils.OpusEncoderUtils(
             sample_rate=16000, channels=1, frame_size_ms=60
         )
         check_model_key("TTS", self.access_token)
 
-        # 添加会话状态控制
-        self._session_lock = asyncio.Lock()  # 会话操作的并发锁
-        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):
         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连接可用"""
+        """建立新的WebSocket连接"""
         try:
-            if self.ws is None:
+            logger.bind(tag=TAG).info("开始建立新连接...")
-                logger.bind(tag=TAG).info("WebSocket连接不存在，开始建立新连接...")
+            ws_header = {
-                ws_header = {
+                "X-Api-App-Key": self.appId,
-                    "X-Api-App-Key": self.appId,
+                "X-Api-Access-Key": self.access_token,
-                    "X-Api-Access-Key": self.access_token,
+                "X-Api-Resource-Id": self.resource_id,
-                    "X-Api-Resource-Id": self.resource_id,
+                "X-Api-Connect-Id": uuid.uuid4(),
-                    "X-Api-Connect-Id": uuid.uuid4(),
+            }
-                }
+            self.ws = await websockets.connect(
-                self.ws = await websockets.connect(
+                self.ws_url, additional_headers=ws_header, max_size=1000000000
-                    self.ws_url, additional_headers=ws_header, max_size=1000000000
+            )
-                )
+            logger.bind(tag=TAG).info("WebSocket连接建立成功")
-                self._connection_ready = True
+            return self.ws
-                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)}")
+            logger.bind(tag=TAG).error(f"建立连接失败: {str(e)}")
-            self._connection_ready = False
             self.ws = None
             raise
 
     def tts_text_priority_thread(self):
+        """火山引擎双流式TTS的文本处理线程"""
         logger.bind(tag=TAG).info("TTS文本处理线程启动")
         while not self.conn.stop_event.is_set():
             try:
@@ -239,6 +198,10 @@ class TTSProvider(TTSProviderBase):
                 logger.bind(tag=TAG).info(
                     f"收到TTS任务｜{message.sentence_type.name} ｜ {message.content_type.name} | 会话ID: {self.conn.sentence_id}"
                 )
+                if self.conn.client_abort:
+                    logger.bind(tag=TAG).info("收到打断信息，终止TTS文本处理线程")
+                    continue
+
                 if message.sentence_type == SentenceType.FIRST:
                     # 初始化参数
                     try:
@@ -253,9 +216,8 @@ class TTSProvider(TTSProviderBase):
                         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:
@@ -270,9 +232,8 @@ class TTSProvider(TTSProviderBase):
                             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}"
@@ -292,8 +253,6 @@ class TTSProvider(TTSProviderBase):
                         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:
@@ -302,103 +261,204 @@ 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服务"""
         try:
-            # 确保WebSocket连接可用
+            # 建立新连接
-            if not self._connection_ready or self.ws is None:
+            if self.ws is None:
-                logger.bind(tag=TAG).warning("WebSocket连接不可用，尝试重新连接...")
+                await handleAbortMessage(self.conn)
-                await self._ensure_connection()
+                logger.bind(tag=TAG).error(f"WebSocket连接不存在，终止发送文本")
-
+                return
             # 发送文本
             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 self.ws:
-            if isinstance(e, websockets.ConnectionClosed):
+                try:
-                self._connection_ready = False
+                    await self.ws.close()
+                except:
+                    pass
                 self.ws = None
-                await self._handle_connection_error()
             raise
 
-    ###################################################################################
+    async def start_session(self, session_id):
-    # 火山双流式TTS重写父类的方法--结束
+        logger.bind(tag=TAG).info(f"开始会话～～{session_id}")
-    ###################################################################################
+        try:
-    def _start_monitor_tts_response_thread(self):
+            # 建立新连接
-        # 初始化链接
+            await self._ensure_connection()
-        asyncio.run_coroutine_threadsafe(
+
-            self._start_monitor_tts_response(), loop=self.conn.loop
+            # 启动监听任务
-        )
+            self._monitor_task = asyncio.create_task(self._start_monitor_tts_response())
+
+            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)
+            logger.bind(tag=TAG).info("会话启动请求已发送")
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"启动会话失败: {str(e)}")
+            # 确保清理资源
+            if hasattr(self, "_monitor_task"):
+                try:
+                    self._monitor_task.cancel()
+                    await self._monitor_task
+                except:
+                    pass
+                self._monitor_task = None
+            if self.ws:
+                try:
+                    await self.ws.close()
+                except:
+                    pass
+                self.ws = None
+            raise
+
+    async def finish_session(self, session_id):
+        logger.bind(tag=TAG).info(f"关闭会话～～{session_id}")
+        try:
+            if self.ws:
+                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)
+                logger.bind(tag=TAG).info("会话结束请求已发送")
+
+                # 等待监听任务完成
+                if hasattr(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
+
+                # 关闭连接
+                await self.close()
+        except Exception as e:
+            logger.bind(tag=TAG).error(f"关闭会话失败: {str(e)}")
+            # 确保清理资源
+            if hasattr(self, "_monitor_task"):
+                try:
+                    self._monitor_task.cancel()
+                    await self._monitor_task
+                except:
+                    pass
+                self._monitor_task = None
+            if self.ws:
+                try:
+                    await self.ws.close()
+                except:
+                    pass
+                self.ws = None
+            raise
+
+    async def close(self):
+        """资源清理方法"""
+        if self.ws:
+            try:
+                await self.ws.close()
+            except:
+                pass
+            self.ws = None
 
     async def _start_monitor_tts_response(self):
+        """监听TTS响应"""
         opus_datas_cache = []
-        # 添加标志来区分是否是第一句话
         is_first_sentence = True
-        while not self.conn.stop_event.is_set():
+        try:
-            try:
+            while not self.conn.stop_event.is_set():
-                # 确保 `recv()` 运行在同一个 event loop
+                try:
-                msg = await self.ws.recv()
+                    # 确保 `recv()` 运行在同一个 event loop
-                res = self.parser_response(msg)
+                    msg = await self.ws.recv()
-                self.print_response(res, "send_text res:")
+                    res = self.parser_response(msg)
+                    self.print_response(res, "send_text res:")
 
-                if res.optional.event == EVENT_TTSSentenceStart:
+                    # 检查客户端是否中止
-                    json_data = json.loads(res.payload.decode("utf-8"))
+                    if self.conn.client_abort:
-                    self.tts_text = json_data.get("text", "")
+                        logger.bind(tag=TAG).info("收到打断信息，终止监听TTS响应")
-                    logger.bind(tag=TAG).debug(f"句子语音生成开始: {self.tts_text}")
+                        break
-                    self.tts_audio_queue.put((SentenceType.FIRST, [], self.tts_text))
+
-                    opus_datas_cache = []
+                    if res.optional.event == EVENT_TTSSentenceStart:
-                elif (
+                        json_data = json.loads(res.payload.decode("utf-8"))
-                    res.optional.event == EVENT_TTSResponse
+                        self.tts_text = json_data.get("text", "")
-                    and res.header.message_type == AUDIO_ONLY_RESPONSE
+                        logger.bind(tag=TAG).debug(f"句子语音生成开始: {self.tts_text}")
-                ):
-                    logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
-                    opus_datas = self.wav_to_opus_data_audio_raw(res.payload)
-                    logger.bind(tag=TAG).debug(
-                        f"推送数据到队列里面帧数～～{len(opus_datas)}"
-                    )
-                    if is_first_sentence:
-                        # 第一句话直接发送
                         self.tts_audio_queue.put(
-                            (SentenceType.MIDDLE, opus_datas, self.tts_text)
+                            (SentenceType.FIRST, [], self.tts_text)
                         )
-                    else:
+                        opus_datas_cache = []
-                        # 后续句子缓存
+                    elif (
-                        opus_datas_cache = opus_datas_cache + opus_datas
+                        res.optional.event == EVENT_TTSResponse
-                elif res.optional.event == EVENT_TTSSentenceEnd:
+                        and res.header.message_type == AUDIO_ONLY_RESPONSE
-                    logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}")
+                    ):
-                    if not is_first_sentence:
+                        logger.bind(tag=TAG).debug(f"推送数据到队列里面～～")
-                        # 只有非第一句话才发送缓存的数据
+                        opus_datas = self.wav_to_opus_data_audio_raw(res.payload)
-                        self.tts_audio_queue.put(
+                        logger.bind(tag=TAG).debug(
-                            (SentenceType.MIDDLE, opus_datas_cache, self.tts_text)
+                            f"推送数据到队列里面帧数～～{len(opus_datas)}"
                         )
-                    # 第一句话结束后，将标志设置为False
+                        if is_first_sentence:
-                    is_first_sentence = False
+                            # 第一句话直接发送
-                elif res.optional.event == EVENT_SessionFinished:
-                    logger.bind(tag=TAG).debug(f"会话结束～～")
-                    for tts_file, text in self.before_stop_play_files:
-                        if tts_file and os.path.exists(tts_file):
-                            audio_datas = self._process_audio_file(tts_file)
                             self.tts_audio_queue.put(
-                                (SentenceType.MIDDLE, audio_datas, text)
+                                (SentenceType.MIDDLE, opus_datas, self.tts_text)
                             )
-                    self.before_stop_play_files.clear()
+                        else:
-                    self.tts_audio_queue.put((SentenceType.LAST, [], None))
+                            # 后续句子缓存
-
+                            opus_datas_cache = opus_datas_cache + opus_datas
-                    opus_datas_cache = []
+                    elif res.optional.event == EVENT_TTSSentenceEnd:
-                    is_first_sentence = True
+                        logger.bind(tag=TAG).info(f"句子语音生成成功：{self.tts_text}")
-                    continue
+                        if not is_first_sentence:
-            except websockets.ConnectionClosed:
+                            # 只有非第一句话才发送缓存的数据
-                break  # 连接关闭时退出监听
+                            self.tts_audio_queue.put(
-            except Exception as e:
+                                (SentenceType.MIDDLE, opus_datas_cache, self.tts_text)
-                logger.bind(tag=TAG).error(f"Error in _start_monitor_tts_response: {e}")
+                            )
-                traceback.print_exc()
+                        # 第一句话结束后，将标志设置为False
-                continue
+                        is_first_sentence = False
+                    elif res.optional.event == EVENT_SessionFinished:
+                        logger.bind(tag=TAG).debug(f"会话结束～～")
+                        for tts_file, text in self.before_stop_play_files:
+                            if tts_file and os.path.exists(tts_file):
+                                audio_datas = self._process_audio_file(tts_file)
+                                self.tts_audio_queue.put(
+                                    (SentenceType.MIDDLE, audio_datas, text)
+                                )
+                        self.before_stop_play_files.clear()
+                        self.tts_audio_queue.put((SentenceType.LAST, [], None))
+                        break
+                except websockets.ConnectionClosed:
+                    logger.bind(tag=TAG).warning("WebSocket连接已关闭")
+                    break
+                except Exception as e:
+                    logger.bind(tag=TAG).error(
+                        f"Error in _start_monitor_tts_response: {e}"
+                    )
+                    traceback.print_exc()
+                    break
+        finally:
+            # 确保清理资源
+            if self.ws:
+                try:
+                    await self.ws.close()
+                except:
+                    pass
+                self.ws = None
 
     async def send_event(
         self, header: bytes, optional: bytes | None = None, payload: bytes = None
@@ -413,11 +473,8 @@ class TTSProvider(TTSProviderBase):
                 full_client_request.extend(payload)
             await self.ws.send(full_client_request)
         except websockets.ConnectionClosed:
-            if await self._handle_connection_error():
+            logger.bind(tag=TAG).error(f"ConnectionClosed")
-                # 重连成功后重试发送
+            raise
-                await self.ws.send(full_client_request)
-            else:
-                raise
 
     async def send_text(self, speaker: str, text: str, session_id):
         header = Header(
@@ -540,200 +597,6 @@ class TTSProvider(TTSProviderBase):
             )
         )
 
-    async def finish_connection(self):
-        header = Header(
-            message_type=FULL_CLIENT_REQUEST,
-            message_type_specific_flags=MsgTypeFlagWithEvent,
-            serial_method=JSON,
-        ).as_bytes()
-        optional = Optional(event=EVENT_FinishConnection).as_bytes()
-        payload = str.encode("{}")
-        await self.send_event(header, optional, payload)
-        return
-
-    async def start_session(self, 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)
-
-                    # 如果已有会话未结束，先关闭它
-                    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
-
-                    # 重置会话状态
-                    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).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._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
-
-                    # 确保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
-
-    async def reset(self):
-        # 关闭之前的对话
-        if self.start_connection_flag:
-            await self.finish_connection()
-            self.start_connection_flag = False
-        await self.start_connection()
-        self.start_connection_flag = True
-
-    async def close(self):
-        """资源清理方法"""
-        await self.finish_connection()
-        await self.ws.close()
-
     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