From fdd16ad3c75e4352a108a07f32d18d338fab7e3b Mon Sep 17 00:00:00 2001 From: hrz <1710360675@qq.com> Date: Mon, 31 Mar 2025 21:40:48 +0800 Subject: [PATCH] =?UTF-8?q?update:=201=E3=80=81await=20self.close(ws)?= =?UTF-8?q?=E5=B7=B2=E7=BB=8F=E5=9C=A8finally=E6=9C=89=E5=85=B3=E9=97=AD?= =?UTF-8?q?=E9=80=BB=E8=BE=91=E4=BA=86=EF=BC=8C=E6=97=A0=E9=9C=80=E5=9C=A8?= =?UTF-8?q?except=E9=87=8C=E5=85=B3=E9=97=AD=E3=80=82=202=E3=80=81func=5Fh?= =?UTF-8?q?andler=E5=92=8Cmcp=5Fmanager=E7=9A=84=E5=8A=A0=E8=BD=BD?= =?UTF-8?q?=E5=BF=85=E9=A1=BB=E7=94=A8=E5=BC=82=E6=AD=A5=E5=AE=9E=E7=8E=B0?= =?UTF-8?q?=EF=BC=8C=E5=87=8F=E5=B0=91=E5=94=A4=E9=86=92=E8=BF=9E=E6=8E=A5?= =?UTF-8?q?=E5=A5=BD=E4=BD=BF=203=E3=80=81=E5=85=B3=E4=BA=8Eiot=E7=AD=89?= =?UTF-8?q?=E5=BE=85func=5Fhandler=E9=97=AE=E9=A2=98=EF=BC=8C=E8=A6=81?= =?UTF-8?q?=E5=9C=A8iot=E6=B6=88=E6=81=AF=E9=82=A3=E9=87=8C=E5=81=9A?= =?UTF-8?q?=E7=AD=89=E5=BE=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- main/xiaozhi-server/core/connection.py | 289 ++++++++++++++++--------- 1 file changed, 192 insertions(+), 97 deletions(-) diff --git a/main/xiaozhi-server/core/connection.py b/main/xiaozhi-server/core/connection.py index 790e2879..eac944a3 100644 --- a/main/xiaozhi-server/core/connection.py +++ b/main/xiaozhi-server/core/connection.py @@ -13,7 +13,11 @@ from plugins_func.loadplugins import auto_import_modules from config.logger import setup_logging from core.utils.dialogue import Message, Dialogue from core.handle.textHandle import handleTextMessage -from core.utils.util import get_string_no_punctuation_or_emoji, extract_json_from_string, get_ip_info +from core.utils.util import ( + get_string_no_punctuation_or_emoji, + extract_json_from_string, + get_ip_info, +) from concurrent.futures import ThreadPoolExecutor, TimeoutError from core.handle.sendAudioHandle import sendAudioMessage from core.handle.receiveAudioHandle import handleAudioMessage @@ -26,7 +30,7 @@ from core.mcp.manager import MCPManager TAG = __name__ -auto_import_modules('plugins_func.functions') +auto_import_modules("plugins_func.functions") class TTSException(RuntimeError): @@ -34,7 +38,9 @@ class TTSException(RuntimeError): class ConnectionHandler: - def __init__(self, config: Dict[str, Any], _vad, _asr, _llm, _tts, _memory, _intent): + def __init__( + self, config: Dict[str, Any], _vad, _asr, _llm, _tts, _memory, _intent + ): self.config = config self.logger = setup_logging() self.auth = AuthMiddleware(config) @@ -99,23 +105,8 @@ class ConnectionHandler: self.is_device_verified = False # 添加设备验证状态标志 self.close_after_chat = False # 是否在聊天结束后关闭连接 self.use_function_call_mode = False - if self.config["selected_module"]["Intent"] == 'function_call': + if self.config["selected_module"]["Intent"] == "function_call": self.use_function_call_mode = True - - try: - # 加载插件 - if self.use_function_call_mode: - self.func_handler = FunctionHandler(self) - self.logger.bind(tag=TAG).info("初始化FunctionHandler成功") - else: - self.func_handler = None - self.logger.bind(tag=TAG).info("未启用function_call模式，跳过FunctionHandler初始化") - except Exception as e: - self.logger.bind(tag=TAG).error(f"初始化FunctionHandler失败: {str(e)}") - self.func_handler = None - self.use_function_call_mode = False - - self.mcp_manager = MCPManager(self) async def handle_connection(self, ws): try: @@ -123,7 +114,9 @@ class ConnectionHandler: self.headers = dict(ws.request.headers) # 获取客户端ip地址 self.client_ip = ws.remote_address[0] - self.logger.bind(tag=TAG).info(f"{self.client_ip} conn - Headers: {self.headers}") + self.logger.bind(tag=TAG).info( + f"{self.client_ip} conn - Headers: {self.headers}" + ) # 进行认证 await self.auth.authenticate(self.headers) @@ -138,10 +131,14 @@ class ConnectionHandler: await self.websocket.send(json.dumps(self.welcome_msg)) # Load private configuration if device_id is provided bUsePrivateConfig = self.config.get("use_private_config", False) - self.logger.bind(tag=TAG).info(f"bUsePrivateConfig: {bUsePrivateConfig}, device_id: {device_id}") + self.logger.bind(tag=TAG).info( + f"bUsePrivateConfig: {bUsePrivateConfig}, device_id: {device_id}" + ) if bUsePrivateConfig and device_id: try: - self.private_config = PrivateConfig(device_id, self.config, self.auth_code_gen) + self.private_config = PrivateConfig( + device_id, self.config, self.auth_code_gen + ) await self.private_config.load_or_create() # 判断是否已经绑定 owner = self.private_config.get_owner() @@ -154,23 +151,33 @@ class ConnectionHandler: if all([llm, tts]): self.llm = llm self.tts = tts - self.logger.bind(tag=TAG).info(f"Loaded private config and instances for device {device_id}") + self.logger.bind(tag=TAG).info( + f"Loaded private config and instances for device {device_id}" + ) else: - self.logger.bind(tag=TAG).error(f"Failed to create instances for device {device_id}") + self.logger.bind(tag=TAG).error( + f"Failed to create instances for device {device_id}" + ) self.private_config = None except Exception as e: - self.logger.bind(tag=TAG).error(f"Error initializing private config: {e}") + self.logger.bind(tag=TAG).error( + f"Error initializing private config: {e}" + ) self.private_config = None raise # 异步初始化 self.executor.submit(self._initialize_components) # tts 消化线程 - self.tts_priority_thread = threading.Thread(target=self._tts_priority_thread, daemon=True) + self.tts_priority_thread = threading.Thread( + target=self._tts_priority_thread, daemon=True + ) self.tts_priority_thread.start() # 音频播放 消化线程 - self.audio_play_priority_thread = threading.Thread(target=self._audio_play_priority_thread, daemon=True) + self.audio_play_priority_thread = threading.Thread( + target=self._audio_play_priority_thread, daemon=True + ) self.audio_play_priority_thread.start() try: @@ -181,12 +188,10 @@ class ConnectionHandler: except AuthenticationError as e: self.logger.bind(tag=TAG).error(f"Authentication failed: {str(e)}") - await self.close(ws) return except Exception as e: stack_trace = traceback.format_exc() self.logger.bind(tag=TAG).error(f"Connection error: {str(e)}-{stack_trace}") - await self.close(ws) return finally: self._save_and_close(ws) @@ -214,34 +219,46 @@ class ConnectionHandler: self.prompt = self.private_config.private_config.get("prompt", self.prompt) self.dialogue.put(Message(role="system", content=self.prompt)) + """加载插件""" + self.func_handler = FunctionHandler(self) + self.mcp_manager = MCPManager(self) """加载记忆""" device_id = self.headers.get("device-id", None) self.memory.init_memory(device_id, self.llm) - + """为意图识别设置LLM，优先使用专用LLM""" # 检查是否配置了专用的意图识别LLM intent_llm_name = self.config["Intent"]["intent_llm"]["llm"] - + # 记录开始初始化意图识别LLM的时间 intent_llm_init_start = time.time() - - if not self.use_function_call_mode and intent_llm_name and intent_llm_name in self.config["LLM"]: + + if ( + not self.use_function_call_mode + and intent_llm_name + and intent_llm_name in self.config["LLM"] + ): # 如果配置了专用LLM，则创建独立的LLM实例 from core.utils import llm as llm_utils + intent_llm_config = self.config["LLM"][intent_llm_name] intent_llm_type = intent_llm_config.get("type", intent_llm_name) intent_llm = llm_utils.create_instance(intent_llm_type, intent_llm_config) - self.logger.bind(tag=TAG).info(f"为意图识别创建了专用LLM: {intent_llm_name}, 类型: {intent_llm_type}") - + self.logger.bind(tag=TAG).info( + f"为意图识别创建了专用LLM: {intent_llm_name}, 类型: {intent_llm_type}" + ) + self.intent.set_llm(intent_llm) else: # 否则使用主LLM self.intent.set_llm(self.llm) self.logger.bind(tag=TAG).info("意图识别使用主LLM") - + # 记录意图识别LLM初始化耗时 intent_llm_init_time = time.time() - intent_llm_init_start - self.logger.bind(tag=TAG).info(f"意图识别LLM初始化完成，耗时: {intent_llm_init_time:.4f}秒") + self.logger.bind(tag=TAG).info( + f"意图识别LLM初始化完成，耗时: {intent_llm_init_time:.4f}秒" + ) """加载位置信息""" self.client_ip_info = get_ip_info(self.client_ip) @@ -251,7 +268,9 @@ class ConnectionHandler: self.dialogue.update_system_message(self.prompt) """加载MCP工具""" - asyncio.run_coroutine_threadsafe(self.mcp_manager.initialize_servers(), self.loop) + asyncio.run_coroutine_threadsafe( + self.mcp_manager.initialize_servers(), self.loop + ) def change_system_prompt(self, prompt): self.prompt = prompt @@ -283,7 +302,9 @@ class ConnectionHandler: def chat(self, query): if self.isNeedAuth(): self.llm_finish_task = True - future = asyncio.run_coroutine_threadsafe(self._check_and_broadcast_auth_code(), self.loop) + future = asyncio.run_coroutine_threadsafe( + self._check_and_broadcast_auth_code(), self.loop + ) future.result() return True @@ -294,13 +315,14 @@ class ConnectionHandler: try: start_time = time.time() # 使用带记忆的对话 - future = asyncio.run_coroutine_threadsafe(self.memory.query_memory(query), self.loop) + future = asyncio.run_coroutine_threadsafe( + self.memory.query_memory(query), self.loop + ) memory_str = future.result() self.logger.bind(tag=TAG).debug(f"记忆内容: {memory_str}") llm_responses = self.llm.response( - self.session_id, - self.dialogue.get_llm_dialogue_with_memory(memory_str) + self.session_id, self.dialogue.get_llm_dialogue_with_memory(memory_str) ) except Exception as e: self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}") @@ -330,7 +352,7 @@ class ConnectionHandler: # 找到分割点则处理 if last_punct_pos != -1: - segment_text_raw = current_text[:last_punct_pos + 1] + segment_text_raw = current_text[: last_punct_pos + 1] segment_text = get_string_no_punctuation_or_emoji(segment_text_raw) if segment_text: # 强制设置空字符，测试TTS出错返回语音的健壮性 @@ -338,7 +360,9 @@ class ConnectionHandler: # segment_text = " " text_index += 1 self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit(self.speak_and_play, segment_text, text_index) + future = self.executor.submit( + self.speak_and_play, segment_text, text_index + ) self.tts_queue.put(future) processed_chars += len(segment_text_raw) # 更新已处理字符位置 @@ -350,12 +374,16 @@ class ConnectionHandler: if segment_text: text_index += 1 self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit(self.speak_and_play, segment_text, text_index) + future = self.executor.submit( + self.speak_and_play, segment_text, text_index + ) self.tts_queue.put(future) self.llm_finish_task = True self.dialogue.put(Message(role="assistant", content="".join(response_message))) - self.logger.bind(tag=TAG).debug(json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False)) + self.logger.bind(tag=TAG).debug( + json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False) + ) return True def chat_with_function_calling(self, query, tool_call=False): @@ -363,7 +391,9 @@ class ConnectionHandler: """Chat with function calling for intent detection using streaming""" if self.isNeedAuth(): self.llm_finish_task = True - future = asyncio.run_coroutine_threadsafe(self._check_and_broadcast_auth_code(), self.loop) + future = asyncio.run_coroutine_threadsafe( + self._check_and_broadcast_auth_code(), self.loop + ) future.result() return True @@ -372,7 +402,7 @@ class ConnectionHandler: # Define intent functions functions = None - if hasattr(self, 'func_handler'): + if hasattr(self, "func_handler"): functions = self.func_handler.get_functions() response_message = [] processed_chars = 0 # 跟踪已处理的字符位置 @@ -381,7 +411,9 @@ class ConnectionHandler: start_time = time.time() # 使用带记忆的对话 - future = asyncio.run_coroutine_threadsafe(self.memory.query_memory(query), self.loop) + future = asyncio.run_coroutine_threadsafe( + self.memory.query_memory(query), self.loop + ) memory_str = future.result() # self.logger.bind(tag=TAG).info(f"对话记录: {self.dialogue.get_llm_dialogue_with_memory(memory_str)}") @@ -390,7 +422,7 @@ class ConnectionHandler: llm_responses = self.llm.response_with_functions( self.session_id, self.dialogue.get_llm_dialogue_with_memory(memory_str), - functions=functions + functions=functions, ) except Exception as e: self.logger.bind(tag=TAG).error(f"LLM 处理出错 {query}: {e}") @@ -411,7 +443,9 @@ class ConnectionHandler: content = response["content"] tools_call = None if content is not None and len(content) > 0: - if len(response_message) <= 0 and (content == "```" or "" in content): + if len(response_message) <= 0 and ( + content == "```" or "" in content + ): tool_call_flag = True if tools_call is not None: @@ -450,14 +484,19 @@ class ConnectionHandler: # 找到分割点则处理 if last_punct_pos != -1: - segment_text_raw = current_text[:last_punct_pos + 1] - segment_text = get_string_no_punctuation_or_emoji(segment_text_raw) + segment_text_raw = current_text[: last_punct_pos + 1] + segment_text = get_string_no_punctuation_or_emoji( + segment_text_raw + ) if segment_text: text_index += 1 self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit(self.speak_and_play, segment_text, text_index) + future = self.executor.submit( + self.speak_and_play, segment_text, text_index + ) self.tts_queue.put(future) - processed_chars += len(segment_text_raw) # 更新已处理字符位置 + # 更新已处理字符位置 + processed_chars += len(segment_text_raw) # 处理function call if tool_call_flag: @@ -468,7 +507,9 @@ class ConnectionHandler: try: content_arguments_json = json.loads(a) function_name = content_arguments_json["name"] - function_arguments = json.dumps(content_arguments_json["arguments"], ensure_ascii=False) + function_arguments = json.dumps( + content_arguments_json["arguments"], ensure_ascii=False + ) function_id = str(uuid.uuid4().hex) except Exception as e: bHasError = True @@ -477,16 +518,19 @@ class ConnectionHandler: bHasError = True response_message.append(content_arguments) if bHasError: - self.logger.bind(tag=TAG).error(f"function call error: {content_arguments}") + self.logger.bind(tag=TAG).error( + f"function call error: {content_arguments}" + ) else: function_arguments = json.loads(function_arguments) if not bHasError: self.logger.bind(tag=TAG).info( - f"function_name={function_name}, function_id={function_id}, function_arguments={function_arguments}") + f"function_name={function_name}, function_id={function_id}, function_arguments={function_arguments}" + ) function_call_data = { "name": function_name, "id": function_id, - "arguments": function_arguments + "arguments": function_arguments, } # 处理MCP工具调用 @@ -494,8 +538,10 @@ class ConnectionHandler: result = self._handle_mcp_tool_call(function_call_data) else: # 处理系统函数 - result = self.func_handler.handle_llm_function_call(self, function_call_data) - self._handle_function_result(result, function_call_data, text_index+1) + result = self.func_handler.handle_llm_function_call( + self, function_call_data + ) + self._handle_function_result(result, function_call_data, text_index + 1) # 处理最后剩余的文本 full_text = "".join(response_message) @@ -505,15 +551,21 @@ class ConnectionHandler: if segment_text: text_index += 1 self.recode_first_last_text(segment_text, text_index) - future = self.executor.submit(self.speak_and_play, segment_text, text_index) + future = self.executor.submit( + self.speak_and_play, segment_text, text_index + ) self.tts_queue.put(future) # 存储对话内容 if len(response_message) > 0: - self.dialogue.put(Message(role="assistant", content="".join(response_message))) + self.dialogue.put( + Message(role="assistant", content="".join(response_message)) + ) self.llm_finish_task = True - self.logger.bind(tag=TAG).debug(json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False)) + self.logger.bind(tag=TAG).debug( + json.dumps(self.dialogue.get_llm_dialogue(), indent=4, ensure_ascii=False) + ) return True @@ -526,13 +578,16 @@ class ConnectionHandler: try: args_dict = json.loads(function_arguments) except json.JSONDecodeError: - self.logger.bind(tag=TAG).error(f"无法解析 function_arguments: {function_arguments}") - return ActionResponse(action=Action.REQLLM, result="参数解析失败", response="") - - tool_result = asyncio.run_coroutine_threadsafe(self.mcp_manager.execute_tool( - function_name, - args_dict - ), self.loop).result() + self.logger.bind(tag=TAG).error( + f"无法解析 function_arguments: {function_arguments}" + ) + return ActionResponse( + action=Action.REQLLM, result="参数解析失败", response="" + ) + + tool_result = asyncio.run_coroutine_threadsafe( + self.mcp_manager.execute_tool(function_name, args_dict), self.loop + ).result() # meta=None content=[TextContent(type='text', text='北京当前天气:\n温度: 21°C\n天气: 晴\n湿度: 6%\n风向: 西北 风\n风力等级: 5级', annotations=None)] isError=False content_text = "" if tool_result is not None and tool_result.content is not None: @@ -542,16 +597,19 @@ class ConnectionHandler: content_text = content.text elif content_type == "image": pass - + if len(content_text) > 0: - return ActionResponse(action=Action.REQLLM, result=content_text, response="") - + return ActionResponse( + action=Action.REQLLM, result=content_text, response="" + ) + except Exception as e: self.logger.bind(tag=TAG).error(f"MCP工具调用错误: {e}") - return ActionResponse(action=Action.REQLLM, result="工具调用出错", response="") + return ActionResponse( + action=Action.REQLLM, result="工具调用出错", response="" + ) return ActionResponse(action=Action.REQLLM, result="工具调用出错", response="") - def _handle_function_result(self, result, function_call_data, text_index): if result.action == Action.RESPONSE: # 直接回复前端 @@ -567,14 +625,26 @@ class ConnectionHandler: function_id = function_call_data["id"] function_name = function_call_data["name"] function_arguments = function_call_data["arguments"] - self.dialogue.put(Message(role='assistant', - tool_calls=[{"id": function_id, - "function": {"arguments": function_arguments, - "name": function_name}, - "type": 'function', - "index": 0}])) + self.dialogue.put( + Message( + role="assistant", + tool_calls=[ + { + "id": function_id, + "function": { + "arguments": function_arguments, + "name": function_name, + }, + "type": "function", + "index": 0, + } + ], + ) + ) - self.dialogue.put(Message(role="tool", tool_call_id=function_id, content=text)) + self.dialogue.put( + Message(role="tool", tool_call_id=function_id, content=text) + ) self.chat_with_function_calling(text, tool_call=True) elif result.action == Action.NOTFOUND: text = result.result @@ -608,15 +678,23 @@ class ConnectionHandler: tts_timeout = self.config.get("tts_timeout", 10) tts_file, text, text_index = future.result(timeout=tts_timeout) if text is None or len(text) <= 0: - self.logger.bind(tag=TAG).error(f"TTS出错：{text_index}: tts text is empty") + self.logger.bind(tag=TAG).error( + f"TTS出错：{text_index}: tts text is empty" + ) elif tts_file is None: - self.logger.bind(tag=TAG).error(f"TTS出错： file is empty: {text_index}: {text}") + self.logger.bind(tag=TAG).error( + f"TTS出错： file is empty: {text_index}: {text}" + ) else: - self.logger.bind(tag=TAG).debug(f"TTS生成：文件路径: {tts_file}") + self.logger.bind(tag=TAG).debug( + f"TTS生成：文件路径: {tts_file}" + ) if os.path.exists(tts_file): opus_datas, duration = self.tts.audio_to_opus_data(tts_file) else: - self.logger.bind(tag=TAG).error(f"TTS出错：文件不存在{tts_file}") + self.logger.bind(tag=TAG).error( + f"TTS出错：文件不存在{tts_file}" + ) except TimeoutError: self.logger.bind(tag=TAG).error("TTS超时") except Exception as e: @@ -624,16 +702,30 @@ class ConnectionHandler: if not self.client_abort: # 如果没有中途打断就发送语音 self.audio_play_queue.put((opus_datas, text, text_index)) - if self.tts.delete_audio_file and tts_file is not None and os.path.exists(tts_file): + if ( + self.tts.delete_audio_file + and tts_file is not None + and os.path.exists(tts_file) + ): os.remove(tts_file) except Exception as e: self.logger.bind(tag=TAG).error(f"TTS任务处理错误: {e}") self.clearSpeakStatus() asyncio.run_coroutine_threadsafe( - self.websocket.send(json.dumps({"type": "tts", "state": "stop", "session_id": self.session_id})), - self.loop + self.websocket.send( + json.dumps( + { + "type": "tts", + "state": "stop", + "session_id": self.session_id, + } + ) + ), + self.loop, + ) + self.logger.bind(tag=TAG).error( + f"tts_priority priority_thread: {text} {e}" ) - self.logger.bind(tag=TAG).error(f"tts_priority priority_thread: {text} {e}") def _audio_play_priority_thread(self): while not self.stop_event.is_set(): @@ -645,11 +737,14 @@ class ConnectionHandler: if self.stop_event.is_set(): break continue - future = asyncio.run_coroutine_threadsafe(sendAudioMessage(self, opus_datas, text, text_index), - self.loop) + future = asyncio.run_coroutine_threadsafe( + sendAudioMessage(self, opus_datas, text, text_index), self.loop + ) future.result() except Exception as e: - self.logger.bind(tag=TAG).error(f"audio_play_priority priority_thread: {text} {e}") + self.logger.bind(tag=TAG).error( + f"audio_play_priority priority_thread: {text} {e}" + ) def speak_and_play(self, text, text_index=0): if text is None or len(text) <= 0: @@ -682,15 +777,15 @@ class ConnectionHandler: # 触发停止事件并清理资源 if self.stop_event: self.stop_event.set() - + # 立即关闭线程池 if self.executor: self.executor.shutdown(wait=False, cancel_futures=True) self.executor = None - + # 清空任务队列 self._clear_queues() - + if ws: await ws.close() elif self.websocket: