Merge branch 'test-mcp' into py_test_mcp

README.md

@@ -141,11 +141,10 @@
 本项目提供两种部署方式，请根据您的具体需求选择：
 
 #### 🚀 部署方式选择
-
-| 部署方式 | 特点 | 适用场景 | Docker部署文档 | 源码部署文档 | 视频教程 | 
+| 部署方式 | 特点 | 适用场景 | 部署文档 | 配置要求 | 视频教程 | 
 |---------|------|---------|---------|---------|---------|
-| **最简化安装** | 智能对话、IOT功能，数据存储在配置文件 | 低配置环境，无需数据库 | [Docker只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E5%8F%AA%E8%BF%90%E8%A1%8Cserver) | [本地源码只运行Server](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E5%8F%AA%E8%BF%90%E8%A1%8Cserver)| - | 
-| **全模块安装** | 智能对话、IOT、OTA、智控台，数据存储在数据库 | 完整功能体验 |[Docker运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | [本地源码运行全模块](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | [本地源码视频教程](https://www.bilibili.com/video/BV1wBJhz4Ewe) | 
+| **最简化安装** | 智能对话、IOT功能，数据存储在配置文件 | 低配置环境，无需数据库 | [Docker版](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E5%8F%AA%E8%BF%90%E8%A1%8Cserver) / [源码部署](./docs/Deployment.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E5%8F%AA%E8%BF%90%E8%A1%8Cserver)| 如果使用`FunASR`要2核4G，如果全API，要2核2G | - | 
+| **全模块安装** | 智能对话、IOT、OTA、智控台，数据存储在数据库 | 完整功能体验 |[Docker版](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%B8%80docker%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) / [源码部署](./docs/Deployment_all.md#%E6%96%B9%E5%BC%8F%E4%BA%8C%E6%9C%AC%E5%9C%B0%E6%BA%90%E7%A0%81%E8%BF%90%E8%A1%8C%E5%85%A8%E6%A8%A1%E5%9D%97) | 如果使用`FunASR`要4核8G，如果全API，要2核4G| [本地源码视频教程](https://www.bilibili.com/video/BV1wBJhz4Ewe) | 
 
 
 > 💡 提示：以下是按最新代码部署后的测试平台，有需要可烧录测试，并发为6个，每天会清空数据
@@ -158,9 +157,23 @@ OTA接口地址: https://2662r3426b.vicp.fun/xiaozhi/ota/
 Websocket接口地址: wss://2662r3426b.vicp.fun/xiaozhi/v1/
 ```
 
+#### 🚩 配置说明和推荐
+> [!Note]
+> 本项目默认的配置是`入门全免费`设置，如果想效果更优，推荐使用`全流式配置`。
+> 
+> 本项目自`0.5.2`版本，已支持整个生命周期全流式，相比`0.5`版本以前，响应速度提升约`2.5秒`
+
+
+| 模块名称 | 入门全免费设置 | 全流式配置 |
+|---------|---------|------|
+| ASR(语音识别) | FunASR(本地) | ✅DoubaoASR(火山流式语音识别) |
+| LLM(大模型) | ChatGLMLLM(智谱glm-4-flash) | ✅DoubaoLLM(火山doubao-1-5-pro-32k-250115) |
+| TTS(语音合成) | EdgeTTS(微软语音) | ✅HuoshanDoubleStreamTTS(火山双流式语音合成) |
+| Intent(意图识别) | function_call(函数调用) | ✅function_call(函数调用) |
+| Memory(记忆功能) | mem_local_short(本地短期记忆） | ✅mem_local_short（本地短期记忆） |
+
 ---
 ## 功能清单 ✨
-
 ### 已实现 ✅
 
 | 功能模块 | 描述 |

main/manager-api/src/main/java/xiaozhi/common/constant/Constant.java

@@ -227,7 +227,7 @@ public interface Constant {
     /**
      * 版本号
      */
-    public static final String VERSION = "0.5.1";
+    public static final String VERSION = "0.5.2";
 
     /**
      * 无效固件URL

main/xiaozhi-server/config/logger.py

@@ -4,7 +4,7 @@ from loguru import logger
 from config.config_loader import load_config
 from config.settings import check_config_file
 
-SERVER_VERSION = "0.5.1"
+SERVER_VERSION = "0.5.2"
 _logger_initialized = False
 
 
@@ -48,7 +48,11 @@ def setup_logging():
 
     # 第一次初始化时配置日志
     if not _logger_initialized:
-        logger.configure(extra={"selected_module": log_config.get("selected_module", "00000000000000")})  # 新增配置
+        logger.configure(
+            extra={
+                "selected_module": log_config.get("selected_module", "00000000000000")
+            }
+        )  # 新增配置
         log_format = log_config.get(
             "log_format",
             "<green>{time:YYMMDD HH:mm:ss}</green>[{version}_{extra[selected_module]}][<light-blue>{extra[tag]}</light-blue>]-<level>{level}</level>-<light-green>{message}</light-green>",
@@ -62,7 +66,9 @@ def setup_logging():
         log_format = log_format.replace("{version}", SERVER_VERSION)
         log_format = log_format.replace("{selected_module}", selected_module_str)
         log_format_file = log_format_file.replace("{version}", SERVER_VERSION)
-        log_format_file = log_format_file.replace("{selected_module}", selected_module_str)
+        log_format_file = log_format_file.replace(
+            "{selected_module}", selected_module_str
+        )
 
         log_level = log_config.get("log_level", "INFO")
         log_dir = log_config.get("log_dir", "tmp")
@@ -116,15 +122,25 @@ def update_module_string(selected_module_str):
         log_format = log_format.replace("{version}", SERVER_VERSION)
         log_format = log_format.replace("{selected_module}", selected_module_str)
         log_format_file = log_format_file.replace("{version}", SERVER_VERSION)
-        log_format_file = log_format_file.replace("{selected_module}", selected_module_str)
+        log_format_file = log_format_file.replace(
+            "{selected_module}", selected_module_str
+        )
 
         logger.remove()
-        logger.add(sys.stdout, format=log_format, level=log_config.get("log_level", "INFO"), filter=formatter)
         logger.add(
-            os.path.join(log_config.get("log_dir", "tmp"), log_config.get("log_file", "server.log")),
+            sys.stdout,
+            format=log_format,
+            level=log_config.get("log_level", "INFO"),
+            filter=formatter,
+        )
+        logger.add(
+            os.path.join(
+                log_config.get("log_dir", "tmp"),
+                log_config.get("log_file", "server.log"),
+            ),
             format=log_format_file,
             level=log_config.get("log_level", "INFO"),
-            filter=formatter
+            filter=formatter,
         )
 
     except Exception as e:

main/xiaozhi-server/core/connection.py

@@ -84,6 +84,7 @@ class ConnectionHandler:
 
         # 客户端状态相关
         self.client_abort = False
+        self.client_is_speaking = False
         self.client_listen_mode = "auto"
 
         # 线程任务相关
@@ -624,7 +625,7 @@ class ConnectionHandler:
         function_arguments = ""
         content_arguments = ""
         text_index = 0
-
+        self.client_abort = False
         for response in llm_responses:
             if self.client_abort:
                 break
@@ -704,17 +705,23 @@ class ConnectionHandler:
                     "arguments": function_arguments,
                 }
 
-                # 处理MCP工具调用
+                # 处理Server端MCP工具调用
                 if self.mcp_manager.is_mcp_tool(function_name):
                     result = self._handle_mcp_tool_call(function_call_data)
                 elif hasattr(self, "mcp_client") and self.mcp_client.has_tool(function_name):
-                    # 如果是MCP工具调用 
+                    # 如果是小智端MCP工具调用 
                     self.logger.bind(tag=TAG).debug(
-                        f"调用MCP工具: {function_name}, 参数: {function_arguments}"
+                        f"调用小智端MCP工具: {function_name}, 参数: {function_arguments}"
                     )
-                    result = asyncio.run_coroutine_threadsafe(call_mcp_tool(self, self.mcp_client, function_name, function_arguments), self.loop).result()
-                    self.logger.bind(tag=TAG).debug(f"MCP工具调用结果: {result}")
-                    result = ActionResponse(action=Action.REQLLM, result=result, response="")
+                    try:
+                        result = asyncio.run_coroutine_threadsafe(call_mcp_tool(self, self.mcp_client, function_name, function_arguments), self.loop).result()
+                        self.logger.bind(tag=TAG).debug(f"MCP工具调用结果: {result}")
+                        result = ActionResponse(action=Action.REQLLM, result=result, response="")
+                    except Exception as e:
+                        self.logger.bind(tag=TAG).error(f"MCP工具调用失败: {e}")
+                        result = ActionResponse(
+                            action=Action.REQLLM, result="MCP工具调用失败", response=""
+                        )
                 else:
                     # 处理系统函数
                     result = self.func_handler.handle_llm_function_call(
@@ -867,6 +874,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/functionHandler.py

@@ -61,7 +61,7 @@ class FunctionHandler:
         self.function_registry.register_function("plugin_loader")
         self.function_registry.register_function("get_time")
         self.function_registry.register_function("get_lunar")
-        self.function_registry.register_function("handle_speaker_volume_or_screen_brightness")
+        # self.function_registry.register_function("handle_speaker_volume_or_screen_brightness")
 
     def register_config_functions(self):
         """注册配置中的函数,可以不同客户端使用不同的配置"""

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

@@ -104,7 +104,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/mcpHandle.py

@@ -7,33 +7,36 @@ TAG = __name__
 class MCPClient:
     """MCPClient，用于管理MCP状态和工具"""
     def __init__(self):
-        self.tools = []
+        self.tools = {}  # Dictionary for O(1) lookup
         self.ready = False
         self.call_results = {}  # To store Futures for tool call responses
         self.next_id = 1
         self.lock = asyncio.Lock()
+        self._cached_available_tools = None # Cache for get_available_tools
 
-    async def has_tool(self, name: str) -> bool:
-        async with self.lock:
-            for tool in self.tools:
-                if tool["name"] == name:
-                    return True
-            return False
+    def has_tool(self, name: str) -> bool:
+        return name in self.tools
 
     def get_available_tools(self) -> list:
-        # async with self.lock:
+        # Check if the cache is valid
+        if self._cached_available_tools is not None:
+            return self._cached_available_tools
+
+        # If cache is not valid, regenerate the list
         result = []
-        for tool in self.tools:
+        for tool_name, tool_data in self.tools.items():
             function_def = {
-                "name": tool["name"],
-                "description": tool["description"],
+                "name": tool_data["name"],
+                "description": tool_data["description"],
                 "parameters": {
-                    "type": tool["inputSchema"].get("type", "object"),
-                    "properties": tool["inputSchema"].get("properties", {}),
-                    "required": tool["inputSchema"].get("required", [])
+                    "type": tool_data["inputSchema"].get("type", "object"),
+                    "properties": tool_data["inputSchema"].get("properties", {}),
+                    "required": tool_data["inputSchema"].get("required", [])
                 }
             }
             result.append({"type": "function", "function": function_def})
+        
+        self._cached_available_tools = result # Store the generated list in cache
         return result
 
     async def is_ready(self) -> bool:
@@ -46,8 +49,9 @@ class MCPClient:
 
     async def add_tool(self, tool_data: dict):
         async with self.lock:
-            self.tools.append(tool_data)
-
+            self.tools[tool_data["name"]] = tool_data
+            self._cached_available_tools = None # Invalidate the cache when a tool is added
+            
     async def get_next_id(self) -> int:
         async with self.lock:
             current_id = self.next_id
@@ -232,7 +236,7 @@ async def call_mcp_tool(conn, mcp_client: MCPClient, tool_name: str, args: str =
     if not await mcp_client.is_ready():
         raise RuntimeError("MCP客户端尚未准备就绪")
 
-    if not await mcp_client.has_tool(tool_name):
+    if not mcp_client.has_tool(tool_name):
         raise ValueError(f"工具 {tool_name} 不存在")
 
     tool_call_id = await mcp_client.get_next_id()
@@ -272,4 +276,4 @@ async def call_mcp_tool(conn, mcp_client: MCPClient, tool_name: str, args: str =
         raise TimeoutError("工具调用请求超时")
     except Exception as e:
         await mcp_client.cleanup_call_result(tool_call_id)
-        raise e
+        raise e

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)
 
 

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

@@ -59,6 +59,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,7 @@ 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
-        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()

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

main/xiaozhi-server/test/test_page.html

@@ -1484,6 +1484,23 @@
                                 if (message.text && message.text !== '😊') {
                                     addMessage(message.text);
                                 }
+                            }else if (message.type === 'mcp') {
+                                const payload = message.payload || {};
+                                log(`服务器下发: ${JSON.stringify(message)}`, 'info');
+                                if (payload) {
+                                    // 模拟小智客户端行为
+                                    if(payload.method === 'tools/list'){
+                                        const replay_message = JSON.stringify({"session_id":"","type":"mcp","payload":{"jsonrpc":"2.0","id":2,"result":{"tools":[{"name":"self.get_device_status","description":"Provides the real-time information of the device, including the current status of the audio speaker, screen, battery, network, etc.\nUse this tool for: \n1. Answering questions about current condition (e.g. what is the current volume of the audio speaker?)\n2. As the first step to control the device (e.g. turn up / down the volume of the audio speaker, etc.)","inputSchema":{"type":"object","properties":{}}},{"name":"self.audio_speaker.set_volume","description":"Set the volume of the audio speaker. If the current volume is unknown, you must call `self.get_device_status` tool first and then call this tool.","inputSchema":{"type":"object","properties":{"volume":{"type":"integer","minimum":0,"maximum":100}},"required":["volume"]}},{"name":"self.screen.set_brightness","description":"Set the brightness of the screen.","inputSchema":{"type":"object","properties":{"brightness":{"type":"integer","minimum":0,"maximum":100}},"required":["brightness"]}},{"name":"self.screen.set_theme","description":"Set the theme of the screen. The theme can be 'light' or 'dark'.","inputSchema":{"type":"object","properties":{"theme":{"type":"string"}},"required":["theme"]}}]}}})
+                                        websocket.send(replay_message);
+                                        log(`回复MCP消息: ${replay_message}`, 'info');
+                                    } else if(payload.method === 'tools/call'){
+                                        // 模拟回复
+                                        const replay_message = JSON.stringify({"session_id":"9f261599","type":"mcp","payload":{"jsonrpc":"2.0","id": payload.id,"result":{"content":[{"type":"text","text":"true"}],"isError":false}}})
+                                        websocket.send(replay_message);
+                                        log(`回复MCP消息: ${replay_message}`, 'info');
+                                    }
+                                }
+                                
                             } else {
                                 // 未知消息类型
                                 log(`未知消息类型: ${message.type}`, 'info');
@@ -1523,7 +1540,10 @@
                     device_id: config.deviceId,
                     device_name: config.deviceName,
                     device_mac: config.deviceMac,
-                    token: config.token
+                    token: config.token,
+                    features: {
+                        mcp: true
+                    }
                 };
 
                 log('发送hello握手消息', 'info');