diff --git a/main/xiaozhi-server/config/config_loader.py b/main/xiaozhi-server/config/config_loader.py index d71da2d3..49a66558 100644 --- a/main/xiaozhi-server/config/config_loader.py +++ b/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): diff --git a/main/xiaozhi-server/config/manage_api_client.py b/main/xiaozhi-server/config/manage_api_client.py new file mode 100644 index 00000000..1f583857 --- /dev/null +++ b/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() diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 22cb981e..b3a78ee2 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/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(""): # 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}") diff --git a/main/xiaozhi-server/docker-compose_all.yml b/main/xiaozhi-server/docker-compose_all.yml index 81b227b1..eac7bef5 100644 --- a/main/xiaozhi-server/docker-compose_all.yml +++ b/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 diff --git a/main/xiaozhi-server/plugins_func/functions/play_music.py b/main/xiaozhi-server/plugins_func/functions/play_music.py index 20148b5f..0b283c8e 100644 --- a/main/xiaozhi-server/plugins_func/functions/play_music.py +++ b/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)}")