fix:websocket连接后不说话的备用关闭方法 (#872)

* update: 增加对httpClient的统一管理

* update: 增加retry机制

* update: 增加retry机制

* fix:websocket连接后不说话的备用关闭方法

---------

Co-authored-by: haotian <haotian@codemao.cn>
Co-authored-by: GoodyHao <865700600@qq.com>

main/xiaozhi-server/config/config_loader.py

@@ -1,18 +1,7 @@
 import os
 import argparse
-import requests
 import yaml
-import time
-
-
-class DeviceNotFoundException(Exception):
-    pass
-
-
-class DeviceBindException(Exception):
-    def __init__(self, bind_code):
-        self.bind_code = bind_code
-        super().__init__(f"设备绑定异常，绑定码: {bind_code}")
+from config.manage_api_client import init_service, get_server_config, get_agent_models
 
 
 # 添加全局配置缓存
@@ -65,97 +54,27 @@ def get_config_file():
     return config_file
 
 
-def _make_api_request(api_url, secret, endpoint, json_data=None):
-    """执行API请求的通用函数
-
-    Args:
-        api_url: API的基础URL
-        secret: API密钥
-        endpoint: API端点
-        json_data: 请求的JSON数据
-
-    Returns:
-        dict: API返回的数据
-
-    Raises:
-        Exception: 当请求失败时抛出异常
-    """
-    if not api_url or not secret:
-        raise Exception("manager-api的url或secret配置错误")
-
-    if "你" in secret:
-        raise Exception("请先配置manager-api的secret")
-
-    max_retries = 10
-    retry_delay = 2  # 秒
-
-    for attempt in range(max_retries):
-        try:
-            response = requests.post(f"{api_url}{endpoint}", json=json_data)
-            if response.status_code == 200:
-                result = response.json()
-                if result.get("code") == 10041:
-                    raise DeviceNotFoundException(result.get("msg"))
-                elif result.get("code") == 10042:
-                    raise DeviceBindException(result.get("msg"))
-                elif result.get("code") != 0:
-                    raise Exception(f"API返回错误: {result.get('msg', '未知错误')}")
-                return result.get("data")
-
-            error_msg = f"manager-api请求失败，状态码: {response.status_code}"
-            try:
-                error_data = response.json()
-                if "msg" in error_data:
-                    error_msg = f"{error_msg}, 错误信息: {error_data['msg']}"
-            except:
-                error_msg = f"{error_msg}, 响应内容: {response.text}"
-
-            if attempt < max_retries - 1:
-                print(f"请求manager-api失败，正在重试 ({attempt + 1}/{max_retries})...")
-                time.sleep(retry_delay)
-            else:
-                raise Exception(error_msg)
-
-        except requests.exceptions.RequestException as e:
-            if attempt < max_retries - 1:
-                print(f"请求manager-api异常，正在重试 ({attempt + 1}/{max_retries})...")
-                time.sleep(retry_delay)
-            else:
-                raise Exception(f"manager-api请求异常: {str(e)}")
-
-
 def get_config_from_api(config):
     """从Java API获取配置"""
-    api_url = config["manager-api"].get("url", "")
-    secret = config["manager-api"].get("secret", "")
+    # 初始化API客户端
+    init_service(config)
+
+    # 获取服务器配置
+    config_data = get_server_config()
+    if config_data is None:
+        raise Exception("Failed to fetch server config from API")
 
-    config_data = _make_api_request(
-        api_url, secret, "/config/server-base", {"secret": secret}
-    )
     config_data["read_config_from_api"] = True
     config_data["manager-api"] = {
-        "url": api_url,
-        "secret": secret,
+        "url": config["manager-api"].get("url", ""),
+        "secret": config["manager-api"].get("secret", ""),
     }
     return config_data
 
 
 def get_private_config_from_api(config, device_id, client_id):
     """从Java API获取私有配置"""
-    api_url = config["manager-api"].get("url", "")
-    secret = config["manager-api"].get("secret", "")
-
-    return _make_api_request(
-        api_url,
-        secret,
-        "/config/agent-models",
-        {
-            "secret": secret,
-            "macAddress": device_id,
-            "clientId": client_id,
-            "selectedModule": config["selected_module"],
-        },
-    )
+    return get_agent_models(device_id, client_id, config["selected_module"])
 
 
 def ensure_directories(config):

main/xiaozhi-server/config/manage_api_client.py

@@ -0,0 +1,154 @@
+import os
+import time
+from typing import Optional, Dict
+
+import httpx
+
+TAG = __name__
+
+
+class DeviceNotFoundException(Exception):
+    pass
+
+
+class DeviceBindException(Exception):
+    def __init__(self, bind_code):
+        self.bind_code = bind_code
+        super().__init__(f"设备绑定异常，绑定码: {bind_code}")
+
+
+class ManageApiClient:
+    _instance = None
+    _client = None
+    _secret = None
+
+    def __new__(cls, config):
+        """单例模式确保全局唯一实例，并支持传入配置参数"""
+        if cls._instance is None:
+            cls._instance = super().__new__(cls)
+            cls._init_client(config)
+        return cls._instance
+
+    @classmethod
+    def _init_client(cls, config):
+        """初始化持久化连接池"""
+        cls.config = config.get("manager-api")
+
+        if not cls.config:
+            raise Exception("manager-api配置错误")
+
+        if not cls.config.get("url") or not cls.config.get("secret"):
+            raise Exception("manager-api的url或secret配置错误")
+
+        if "你" in cls.config.get("secret"):
+            raise Exception("请先配置manager-api的secret")
+
+        cls._secret = cls.config.get("secret")
+        cls.max_retries = cls.config.get("max_retries", 6)  # 最大重试次数
+        cls.retry_delay = cls.config.get("retry_delay", 10)  # 初始重试延迟(秒)
+        # NOTE(goody): 2025/4/16 http相关资源统一管理，后续可以增加线程池或者超时
+        # 后续也可以统一配置apiToken之类的走通用的Auth
+        cls._client = httpx.Client(
+            base_url=cls.config.get("url"),
+            headers={
+                "User-Agent": f"PythonClient/2.0 (PID:{os.getpid()})",
+                "Accept": "application/json",
+            },
+            timeout=cls.config.get("timeout", 30),  # 默认超时时间30秒
+        )
+
+    @classmethod
+    def _request(cls, method: str, endpoint: str, **kwargs) -> Dict:
+        """发送单次HTTP请求并处理响应"""
+        endpoint = endpoint.lstrip("/")
+        response = cls._client.request(method, endpoint, **kwargs)
+        response.raise_for_status()
+
+        result = response.json()
+
+        # 处理API返回的业务错误
+        if result.get("code") == 10041:
+            raise DeviceNotFoundException(result.get("msg"))
+        elif result.get("code") == 10042:
+            raise DeviceBindException(result.get("msg"))
+        elif result.get("code") != 0:
+            raise Exception(f"API返回错误: {result.get('msg', '未知错误')}")
+
+        # 返回成功数据
+        return result.get("data") if result.get("code") == 0 else None
+
+    @classmethod
+    def _should_retry(cls, exception: Exception) -> bool:
+        """判断异常是否应该重试"""
+        # 网络连接相关错误
+        if isinstance(
+            exception, (httpx.ConnectError, httpx.TimeoutException, httpx.NetworkError)
+        ):
+            return True
+
+        # HTTP状态码错误
+        if isinstance(exception, httpx.HTTPStatusError):
+            status_code = exception.response.status_code
+            return status_code in [408, 429, 500, 502, 503, 504]
+
+        return False
+
+    @classmethod
+    def _execute_request(cls, method: str, endpoint: str, **kwargs) -> Dict:
+        """带重试机制的请求执行器"""
+        retry_count = 0
+
+        while retry_count <= cls.max_retries:
+            try:
+                # 执行请求
+                return cls._request(method, endpoint, **kwargs)
+            except Exception as e:
+                # 判断是否应该重试
+                if retry_count < cls.max_retries and cls._should_retry(e):
+                    retry_count += 1
+                    print(
+                        f"{method} {endpoint} 请求失败，将在 {cls.retry_delay:.1f} 秒后进行第 {retry_count} 次重试"
+                    )
+                    time.sleep(cls.retry_delay)
+                    continue
+                else:
+                    # 不重试，直接抛出异常
+                    raise
+
+    @classmethod
+    def safe_close(cls):
+        """安全关闭连接池"""
+        if cls._client:
+            cls._client.close()
+            cls._instance = None
+
+
+def get_server_config() -> Optional[Dict]:
+    """获取服务器基础配置"""
+    return ManageApiClient._instance._execute_request(
+        "POST", "/config/server-base", json={"secret": ManageApiClient._secret}
+    )
+
+
+def get_agent_models(
+    mac_address: str, client_id: str, selected_module: Dict
+) -> Optional[Dict]:
+    """获取代理模型配置"""
+    return ManageApiClient._instance._execute_request(
+        "POST",
+        "/config/agent-models",
+        json={
+            "secret": ManageApiClient._secret,
+            "macAddress": mac_address,
+            "clientId": client_id,
+            "selectedModule": selected_module,
+        },
+    )
+
+
+def init_service(config):
+    ManageApiClient(config)
+
+
+def manage_api_http_safe_close():
+    ManageApiClient.safe_close()

main/xiaozhi-server/core/connection.py

@@ -27,11 +27,8 @@ from core.handle.functionHandler import FunctionHandler
 from plugins_func.register import Action, ActionResponse
 from core.auth import AuthMiddleware, AuthenticationError
 from core.mcp.manager import MCPManager
-from config.config_loader import (
-    get_private_config_from_api,
-    DeviceNotFoundException,
-    DeviceBindException,
-)
+from config.config_loader import get_private_config_from_api
+from config.manage_api_client import DeviceNotFoundException, DeviceBindException
 
 TAG = __name__
 
@@ -112,6 +109,11 @@ class ConnectionHandler:
         self.close_after_chat = False  # 是否在聊天结束后关闭连接
         self.use_function_call_mode = False
 
+        self.timeout_task = None
+        self.timeout_seconds = (
+            int(self.config.get("close_connection_no_voice_time", 120)) + 60
+        )  # 在原来第一道关闭的基础上加60秒，进行二道关闭
+
     async def handle_connection(self, ws):
         try:
             # 获取并验证headers
@@ -150,6 +152,9 @@ class ConnectionHandler:
             self.websocket = ws
             self.session_id = str(uuid.uuid4())
 
+            # 启动超时检查任务
+            self.timeout_task = asyncio.create_task(self._check_timeout())
+
             self.welcome_msg = self.config["xiaozhi"]
             self.welcome_msg["session_id"] = self.session_id
             await self.websocket.send(json.dumps(self.welcome_msg))
@@ -195,6 +200,11 @@ class ConnectionHandler:
 
     async def _route_message(self, message):
         """消息路由"""
+        # 重置超时计时器
+        if self.timeout_task:
+            self.timeout_task.cancel()
+            self.timeout_task = asyncio.create_task(self._check_timeout())
+
         if isinstance(message, str):
             await handleTextMessage(self, message)
         elif isinstance(message, bytes):
@@ -497,7 +507,7 @@ class ConnectionHandler:
         function_id = None
         function_arguments = ""
         content_arguments = ""
- 
+
         for response in llm_responses:
             content, tools_call = response
 
@@ -506,7 +516,7 @@ class ConnectionHandler:
                 tools_call = None
             if content is not None and len(content) > 0:
                 content_arguments += content
-          
+
             if not tool_call_flag and content_arguments.startswith("<tool_call>"):
                 # print("content_arguments", content_arguments)
                 tool_call_flag = True
@@ -679,7 +689,6 @@ class ConnectionHandler:
             self.tts_queue.put(future)
             self.dialogue.put(Message(role="assistant", content=text))
         elif result.action == Action.REQLLM:  # 调用函数后再请求llm生成回复
-
             text = result.result
             if text is not None and len(text) > 0:
                 function_id = function_call_data["id"]
@@ -706,18 +715,14 @@ class ConnectionHandler:
                     Message(role="tool", tool_call_id=function_id, content=text)
                 )
                 self.chat_with_function_calling(text, tool_call=True)
-        elif result.action == Action.NOTFOUND:
+        elif result.action == Action.NOTFOUND or result.action == Action.ERROR:
             text = result.result
             self.recode_first_last_text(text, text_index)
             future = self.executor.submit(self.speak_and_play, text, text_index)
             self.tts_queue.put(future)
             self.dialogue.put(Message(role="assistant", content=text))
         else:
-            text = result.result
-            self.recode_first_last_text(text, text_index)
-            future = self.executor.submit(self.speak_and_play, text, text_index)
-            self.tts_queue.put(future)
-            self.dialogue.put(Message(role="assistant", content=text))
+            pass
 
     def _tts_priority_thread(self):
         while not self.stop_event.is_set():
@@ -831,6 +836,11 @@ class ConnectionHandler:
 
     async def close(self, ws=None):
         """资源清理方法"""
+        # 取消超时任务
+        if self.timeout_task:
+            self.timeout_task.cancel()
+            self.timeout_task = None
+
         # 清理MCP资源
         if hasattr(self, "mcp_manager") and self.mcp_manager:
             await self.mcp_manager.cleanup_all()
@@ -884,3 +894,15 @@ class ConnectionHandler:
             self.close_after_chat = True
         except Exception as e:
             self.logger.bind(tag=TAG).error(f"Chat and close error: {str(e)}")
+
+    async def _check_timeout(self):
+        """检查连接超时"""
+        try:
+            while not self.stop_event.is_set():
+                await asyncio.sleep(self.timeout_seconds)
+                if not self.stop_event.is_set():
+                    self.logger.bind(tag=TAG).info("连接超时，准备关闭")
+                    await self.close(self.websocket)
+                    break
+        except Exception as e:
+            self.logger.bind(tag=TAG).error(f"超时检查任务出错: {e}")

main/xiaozhi-server/docker-compose_all.yml

@@ -42,7 +42,6 @@ services:
       - "8002:8002"
     environment:
       - TZ=Asia/Shanghai
-      ##记得改mysql和redis IP 密码
       - SPRING_DATASOURCE_DRUID_URL=jdbc:mysql://xiaozhi-esp32-server-db:3306/xiaozhi_esp32_server?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&nullCatalogMeansCurrent=true&connectTimeout=30000&socketTimeout=30000&autoReconnect=true&failOverReadOnly=false&maxReconnects=10
       - SPRING_DATASOURCE_DRUID_USERNAME=root
       - SPRING_DATASOURCE_DRUID_PASSWORD=123456

main/xiaozhi-server/plugins_func/functions/play_music.py

@@ -9,7 +9,7 @@ import traceback
 from pathlib import Path
 from core.utils import p3
 from core.handle.sendAudioHandle import send_stt_message
-from plugins_func.register import register_function,ToolType, ActionResponse, Action
+from plugins_func.register import register_function, ToolType, ActionResponse, Action
 
 
 TAG = __name__
@@ -18,38 +18,41 @@ logger = setup_logging()
 MUSIC_CACHE = {}
 
 play_music_function_desc = {
-                "type": "function",
-                "function": {
-                    "name": "play_music",
-                    "description": "唱歌、听歌、播放音乐的方法。",
-                    "parameters": {
-                        "type": "object",
-                        "properties": {
-                            "song_name": {
-                                "type": "string",
-                                "description": "歌曲名称，如果用户没有指定具体歌名则为'random', 明确指定的时返回音乐的名字 示例: ```用户:播放两只老虎\n参数：两只老虎``` ```用户:播放音乐 \n参数：random ```"
-                            }
-                        },
-                        "required": ["song_name"]
-                    }
+    "type": "function",
+    "function": {
+        "name": "play_music",
+        "description": "唱歌、听歌、播放音乐的方法。",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "song_name": {
+                    "type": "string",
+                    "description": "歌曲名称，如果用户没有指定具体歌名则为'random', 明确指定的时返回音乐的名字 示例: ```用户:播放两只老虎\n参数：两只老虎``` ```用户:播放音乐 \n参数：random ```",
                 }
-            }
+            },
+            "required": ["song_name"],
+        },
+    },
+}
 
 
-@register_function('play_music', play_music_function_desc, ToolType.SYSTEM_CTL)
+@register_function("play_music", play_music_function_desc, ToolType.SYSTEM_CTL)
 def play_music(conn, song_name: str):
     try:
-        music_intent = f"播放音乐 {song_name}" if song_name != "random" else "随机播放音乐"
+        music_intent = (
+            f"播放音乐 {song_name}" if song_name != "random" else "随机播放音乐"
+        )
 
         # 检查事件循环状态
         if not conn.loop.is_running():
             logger.bind(tag=TAG).error("事件循环未运行，无法提交任务")
-            return ActionResponse(action=Action.RESPONSE, result="系统繁忙", response="请稍后再试")
+            return ActionResponse(
+                action=Action.RESPONSE, result="系统繁忙", response="请稍后再试"
+            )
 
         # 提交异步任务
         future = asyncio.run_coroutine_threadsafe(
-            handle_music_command(conn, music_intent),
-            conn.loop
+            handle_music_command(conn, music_intent), conn.loop
         )
 
         # 非阻塞回调处理
@@ -62,10 +65,14 @@ def play_music(conn, song_name: str):
 
         future.add_done_callback(handle_done)
 
-        return ActionResponse(action=Action.RESPONSE, result="指令已接收", response="正在为您播放音乐")
+        return ActionResponse(
+            action=Action.NONE, result="指令已接收", response="正在为您播放音乐"
+        )
     except Exception as e:
         logger.bind(tag=TAG).error(f"处理音乐意图错误: {e}")
-        return ActionResponse(action=Action.RESPONSE, result=str(e), response="播放音乐时出错了")
+        return ActionResponse(
+            action=Action.RESPONSE, result=str(e), response="播放音乐时出错了"
+        )
 
 
 def _extract_song_name(text):
@@ -105,7 +112,9 @@ def get_music_files(music_dir, music_ext):
             if ext in music_ext:
                 # 添加相对路径
                 music_files.append(str(file.relative_to(music_dir)))
-                music_file_names.append(os.path.splitext(str(file.relative_to(music_dir)))[0])
+                music_file_names.append(
+                    os.path.splitext(str(file.relative_to(music_dir)))[0]
+                )
     return music_files, music_file_names
 
 
@@ -117,15 +126,20 @@ def initialize_music_handler(conn):
             MUSIC_CACHE["music_dir"] = os.path.abspath(
                 MUSIC_CACHE["music_config"].get("music_dir", "./music")  # 默认路径修改
             )
-            MUSIC_CACHE["music_ext"] = MUSIC_CACHE["music_config"].get("music_ext", (".mp3", ".wav", ".p3"))
-            MUSIC_CACHE["refresh_time"] = MUSIC_CACHE["music_config"].get("refresh_time", 60)
+            MUSIC_CACHE["music_ext"] = MUSIC_CACHE["music_config"].get(
+                "music_ext", (".mp3", ".wav", ".p3")
+            )
+            MUSIC_CACHE["refresh_time"] = MUSIC_CACHE["music_config"].get(
+                "refresh_time", 60
+            )
         else:
             MUSIC_CACHE["music_dir"] = os.path.abspath("./music")
             MUSIC_CACHE["music_ext"] = (".mp3", ".wav", ".p3")
             MUSIC_CACHE["refresh_time"] = 60
         # 获取音乐文件列表
-        MUSIC_CACHE["music_files"], MUSIC_CACHE["music_file_names"] = get_music_files(MUSIC_CACHE["music_dir"],
-                                                                                      MUSIC_CACHE["music_ext"])
+        MUSIC_CACHE["music_files"], MUSIC_CACHE["music_file_names"] = get_music_files(
+            MUSIC_CACHE["music_dir"], MUSIC_CACHE["music_ext"]
+        )
         MUSIC_CACHE["scan_time"] = time.time()
     return MUSIC_CACHE
 
@@ -135,15 +149,16 @@ async def handle_music_command(conn, text):
     global MUSIC_CACHE
 
     """处理音乐播放指令"""
-    clean_text = re.sub(r'[^\w\s]', '', text).strip()
+    clean_text = re.sub(r"[^\w\s]", "", text).strip()
     logger.bind(tag=TAG).debug(f"检查是否是音乐命令: {clean_text}")
 
     # 尝试匹配具体歌名
     if os.path.exists(MUSIC_CACHE["music_dir"]):
         if time.time() - MUSIC_CACHE["scan_time"] > MUSIC_CACHE["refresh_time"]:
             # 刷新音乐文件列表
-            MUSIC_CACHE["music_files"], MUSIC_CACHE["music_file_names"] = get_music_files(MUSIC_CACHE["music_dir"],
-                                                                                          MUSIC_CACHE["music_ext"])
+            MUSIC_CACHE["music_files"], MUSIC_CACHE["music_file_names"] = (
+                get_music_files(MUSIC_CACHE["music_dir"], MUSIC_CACHE["music_ext"])
+            )
             MUSIC_CACHE["scan_time"] = time.time()
 
         potential_song = _extract_song_name(clean_text)
@@ -158,6 +173,23 @@ async def handle_music_command(conn, text):
     return True
 
 
+def _get_random_play_prompt(song_name):
+    """生成随机播放引导语"""
+    # 移除文件扩展名
+    clean_name = os.path.splitext(song_name)[0]
+    prompts = [
+        f"正在为您播放，{clean_name}",
+        f"请欣赏歌曲，{clean_name}",
+        f"即将为您播放，{clean_name}",
+        f"为您带来，{clean_name}",
+        f"让我们聆听，{clean_name}",
+        f"接下来请欣赏，{clean_name}",
+        f"为您献上，{clean_name}",
+    ]
+    # 直接使用random.choice，不设置seed
+    return random.choice(prompts)
+
+
 async def play_local_music(conn, specific_file=None):
     global MUSIC_CACHE
     """播放本地音乐文件"""
@@ -180,16 +212,25 @@ async def play_local_music(conn, specific_file=None):
         if not os.path.exists(music_path):
             logger.bind(tag=TAG).error(f"选定的音乐文件不存在: {music_path}")
             return
-        text = f"正在播放{selected_music}"
+        text = _get_random_play_prompt(selected_music)
         await send_stt_message(conn, text)
         conn.tts_first_text_index = 0
         conn.tts_last_text_index = 0
+
+        tts_file = await asyncio.to_thread(conn.tts.to_tts, text)
+        if tts_file is not None and os.path.exists(tts_file):
+            conn.tts_last_text_index = 1
+            opus_packets, _ = conn.tts.audio_to_opus_data(tts_file)
+            conn.audio_play_queue.put((opus_packets, None, 0))
+            os.remove(tts_file)
+
         conn.llm_finish_task = True
+
         if music_path.endswith(".p3"):
-            opus_packets, duration = p3.decode_opus_from_file(music_path)
+            opus_packets, _ = p3.decode_opus_from_file(music_path)
         else:
-            opus_packets, duration = conn.tts.audio_to_opus_data(music_path)
-        conn.audio_play_queue.put((opus_packets, selected_music, 0))
+            opus_packets, _ = conn.tts.audio_to_opus_data(music_path)
+        conn.audio_play_queue.put((opus_packets, None, conn.tts_last_text_index))
 
     except Exception as e:
         logger.bind(tag=TAG).error(f"播放音乐失败: {str(e)}")