Merge pull request #3060 from xinnan-tech/py_test_end

Py test end

main/manager-api/src/main/resources/db/changelog/202604011035.sql

@@ -0,0 +1 @@
+INSERT INTO `sys_params` (id, param_code, param_value, value_type, param_type, remark) VALUES (312, 'tool_call_timeout', '30', 'number', 1, '工具调用超时时间(秒)');

main/manager-api/src/main/resources/db/changelog/db.changelog-master.yaml

@@ -592,4 +592,10 @@ databaseChangeLog:
         - sqlFile:
             encoding: utf8
             path: classpath:db/changelog/202604010930.sql
-
+  - changeSet:
+      id: 202604011035
+      author: RanChen
+      changes:
+        - sqlFile:
+            encoding: utf8
+            path: classpath:db/changelog/202604011035.sql

main/xiaozhi-server/config.yaml

@@ -61,6 +61,8 @@ delete_audio: true
 close_connection_no_voice_time: 120
 # TTS请求超时时间(秒)
 tts_timeout: 10
+# 工具调用超时时间(秒)
+tool_call_timeout: 30
 # 开启唤醒词加速
 enable_wakeup_words_response_cache: true
 # 开场是否回复唤醒词

main/xiaozhi-server/core/connection.py

@@ -32,7 +32,7 @@ from core.providers.asr.dto.dto import InterfaceType
 from core.handle.textHandle import handleTextMessage
 from core.providers.tools.unified_tool_handler import UnifiedToolHandler
 from plugins_func.loadplugins import auto_import_modules
-from plugins_func.register import Action
+from plugins_func.register import Action, ActionResponse
 from core.auth import AuthenticationError
 from config.config_loader import get_private_config_from_api
 from core.providers.tts.dto.dto import ContentType, TTSMessageDTO, SentenceType
@@ -148,6 +148,8 @@ class ConnectionHandler:
         self.memory = _memory
         self.intent = _intent
 
+        self.is_exiting = False  # 标记是否正在执行退出流程
+
         # 为每个连接单独管理声纹识别
         self.voiceprint_provider = None
 
@@ -327,6 +329,10 @@ class ConnectionHandler:
 
     async def _route_message(self, message):
         """消息路由"""
+        # 退出状态丢弃所有消息
+        if self.is_exiting:
+           return
+
         # 检查是否已经获取到真实的绑定状态
         if not self.bind_completed_event.is_set():
             # 还没有获取到真实状态，等待直到获取到真实状态或超时
@@ -1102,14 +1108,29 @@ class ConnectionHandler:
                     )
                     futures_with_data.append((future, tool_call_data, tool_input))
 
+                # 工具调用超时时间，可配置，默认30秒
+                tool_call_timeout = int(self.config.get("tool_call_timeout", 30))
                 # 等待协程结束（实际等待时长为最慢的那个）
                 tool_results = []
-                for future, tool_call_data, tool_input in futures_with_data:
-                    result = future.result()
-                    tool_results.append((result, tool_call_data))
 
-                    # 使用公共方法上报工具调用结果
+                for future, tool_call_data, tool_input in futures_with_data:
-                    enqueue_tool_report(self, tool_call_data['name'], tool_input, str(result.result) if result.result else None, report_tool_call=False)
+                    try:
+                        result = future.result(timeout=tool_call_timeout)
+                        tool_results.append((result, tool_call_data))
+                        # 使用公共方法上报工具调用结果
+                        enqueue_tool_report(self, tool_call_data['name'], tool_input, str(result.result) if result.result else None, report_tool_call=False)
+
+                    except Exception as e:
+                        self.logger.bind(tag=TAG).error(
+                            f"工具调用超时或异常: {tool_call_data['name']}, 错误: {e}"
+                        )
+                        # 超时时返回错误响应，避免整个流程卡死
+                        tool_results.append((
+                            ActionResponse(action=Action.ERROR, result="哎呀，网络遇到点问题，请稍后再试下！"),
+                            tool_call_data
+                        ))
+                        # 上报工具调用错误
+                        enqueue_tool_report(self, tool_call_data['name'], tool_input, str(e), report_tool_call=False)
 
                 # 统一处理工具调用结果
                 if tool_results:

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

@@ -7,6 +7,10 @@ TAG = __name__
 
 
 async def handleAbortMessage(conn: "ConnectionHandler"):
+    if conn.close_after_chat or conn.is_exiting:
+        conn.logger.bind(tag=TAG).info("退出流程中被打断，直接关闭连接")
+        return
+        
     conn.logger.bind(tag=TAG).info("Abort message received")
     # 设置成打断状态，会自动打断llm、tts任务
     conn.client_abort = True

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

@@ -33,6 +33,10 @@ async def handle_user_intent(conn: "ConnectionHandler", text):
     if await check_direct_exit(conn, filtered_text):
         return True
 
+    # 明确再见不被打断
+    if conn.is_exiting:
+        return True
+
     # 检查是否是唤醒词
     if await checkWakeupWords(conn, filtered_text):
         return True
@@ -58,6 +62,7 @@ async def check_direct_exit(conn: "ConnectionHandler", text):
         if text == cmd:
             conn.logger.bind(tag=TAG).info(f"识别到明确的退出命令: {text}")
             await send_stt_message(conn, text)
+            conn.is_exiting = True
             await conn.close()
             return True
     return False
@@ -156,7 +161,9 @@ async def process_intent_result(
             # 使用executor执行函数调用和结果处理
             def process_function_call():
                 conn.dialogue.put(Message(role="user", content=original_text))
-
+                
+                # 工具调用超时时间
+                tool_call_timeout = int(conn.config.get("tool_call_timeout", 30))
                 # 使用统一工具处理器处理所有工具调用
                 try:
                     result = asyncio.run_coroutine_threadsafe(
@@ -164,11 +171,11 @@ async def process_intent_result(
                             conn, function_call_data
                         ),
                         conn.loop,
-                    ).result()
+                    ).result(timeout=tool_call_timeout)
                 except Exception as e:
                     conn.logger.bind(tag=TAG).error(f"工具调用失败: {e}")
                     result = ActionResponse(
-                        action=Action.ERROR, result=str(e), response=str(e)
+                        action=Action.ERROR, result="工具调用超时，请一会再试下哈", response="工具调用超时，请一会再试下哈"
                     )
 
                 # 上报工具调用结果

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

@@ -15,6 +15,8 @@ TAG = __name__
 
 
 async def handleAudioMessage(conn: "ConnectionHandler", audio):
+    if conn.is_exiting:
+        return
     # 当前片段是否有人说话
     have_voice = conn.vad.is_vad(conn, audio)
     # 如果设备刚刚被唤醒，短暂忽略VAD检测

main/xiaozhi-server/core/providers/llm/openai/openai.py

@@ -17,8 +17,22 @@ class LLMProvider(LLMProviderBase):
             self.base_url = config.get("base_url")
         else:
             self.base_url = config.get("url")
-        timeout = config.get("timeout", 300)
+        
-        self.timeout = int(timeout) if timeout else 300
+        timeout_config = config.get("timeout")
+        if isinstance(timeout_config, dict):
+            # 细粒度超时配置
+            custom_timeout = httpx.Timeout(
+                pool=timeout_config.get("pool", 2.0),
+                connect=timeout_config.get("connect", 3.0),
+                write=timeout_config.get("write", 5.0),
+                read=timeout_config.get("read", 60.0)
+            )
+        elif isinstance(timeout_config, (int, float)) and timeout_config > 0:
+            # 兼容旧的单一超时配置（整数或浮点数）
+            custom_timeout = httpx.Timeout(timeout_config)
+        else:
+            # 未配置或配置无效，使用默认值
+            custom_timeout = httpx.Timeout(300)
 
         param_defaults = {
             "max_tokens": int,
@@ -45,7 +59,7 @@ class LLMProvider(LLMProviderBase):
         model_key_msg = check_model_key("LLM", self.api_key)
         if model_key_msg:
             logger.bind(tag=TAG).error(model_key_msg)
-        self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url, timeout=httpx.Timeout(self.timeout))
+        self.client = openai.OpenAI(api_key=self.api_key, base_url=self.base_url, timeout=custom_timeout)
 
     @staticmethod
     def normalize_dialogue(dialogue):

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

@@ -128,7 +128,7 @@ class TTSProvider(TTSProviderBase):
                             self.start_session(self.conn.sentence_id),
                             loop=self.conn.loop,
                         )
-                        future.result()
+                        future.result(timeout=self.tts_timeout)
                         self.before_stop_play_files.clear()
                         logger.bind(tag=TAG).info("TTS会话启动成功")
                     except Exception as e:
@@ -145,7 +145,7 @@ class TTSProvider(TTSProviderBase):
                                 self.text_to_speak(message.content_detail, None),
                                 loop=self.conn.loop,
                             )
-                            future.result()
+                            future.result(timeout=self.tts_timeout)
                             logger.bind(tag=TAG).debug("TTS文本发送成功")
                         except Exception as e:
                             logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
@@ -166,7 +166,7 @@ class TTSProvider(TTSProviderBase):
                             self.finish_session(self.conn.sentence_id),
                             loop=self.conn.loop,
                         )
-                        future.result()
+                        future.result(timeout=self.tts_timeout)
                     except Exception as e:
                         logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}")
                         continue

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

@@ -233,7 +233,7 @@ class TTSProvider(TTSProviderBase):
                             self.start_session(self.task_id),
                             loop=self.conn.loop,
                         )
-                        future.result()
+                        future.result(timeout=self.tts_timeout)
                         self.before_stop_play_files.clear()
                         logger.bind(tag=TAG).debug("TTS会话启动成功")
 
@@ -251,7 +251,7 @@ class TTSProvider(TTSProviderBase):
                                 self.text_to_speak(message.content_detail, None),
                                 loop=self.conn.loop,
                             )
-                            future.result()
+                            future.result(timeout=self.tts_timeout)
                             logger.bind(tag=TAG).debug("TTS文本发送成功")
                         except Exception as e:
                             logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
@@ -271,7 +271,7 @@ class TTSProvider(TTSProviderBase):
                             self.finish_session(self.task_id),
                             loop=self.conn.loop,
                         )
-                        future.result()
+                        future.result(timeout=self.tts_timeout)
                     except Exception as e:
                         logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}")
                         continue

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

@@ -36,6 +36,7 @@ class TTSProviderBase(ABC):
         self.delete_audio_file = delete_audio_file
         self.audio_file_type = "wav"
         self.output_file = config.get("output_dir", "tmp/")
+        self.tts_timeout = int(config.get("tts_timeout", 15))
         self.tts_text_queue = queue.Queue()
         self.tts_audio_queue = queue.Queue()
         self.tts_audio_first_sentence = True
@@ -368,7 +369,7 @@ class TTSProviderBase(ABC):
                     sendAudioMessage(self.conn, sentence_type, audio_datas, text),
                     self.conn.loop,
                 )
-                future.result()
+                future.result(timeout=self.tts_timeout)
 
                 # 记录输出和报告
                 if self.conn.max_output_size > 0 and text:

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

@@ -309,7 +309,7 @@ class TTSProvider(TTSProviderBase):
                             self.start_session(self.conn.sentence_id),
                             loop=self.conn.loop,
                         )
-                        future.result()
+                        future.result(timeout=self.tts_timeout)
                         self.before_stop_play_files.clear()
                         logger.bind(tag=TAG).debug("TTS会话启动成功")
                     except Exception as e:
@@ -326,7 +326,7 @@ class TTSProvider(TTSProviderBase):
                                 self.text_to_speak(message.content_detail, None),
                                 loop=self.conn.loop,
                             )
-                            future.result()
+                            future.result(timeout=self.tts_timeout)
                             logger.bind(tag=TAG).debug("TTS文本发送成功")
                         except Exception as e:
                             logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")
@@ -346,7 +346,7 @@ class TTSProvider(TTSProviderBase):
                             self.finish_session(self.conn.sentence_id),
                             loop=self.conn.loop,
                         )
-                        future.result()
+                        future.result(timeout=self.tts_timeout)
                     except Exception as e:
                         logger.bind(tag=TAG).error(f"结束TTS会话失败: {str(e)}")
                         continue

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

@@ -179,7 +179,7 @@ class TTSProvider(TTSProviderBase):
                             self.start_session(self.conn.sentence_id),
                             loop=self.conn.loop,
                         )
-                        future.result()
+                        future.result(timeout=self.tts_timeout)
                         self.before_stop_play_files.clear()
                         logger.bind(tag=TAG).info("TTS会话启动成功")
 
@@ -198,7 +198,7 @@ class TTSProvider(TTSProviderBase):
                                 self.text_to_speak(message.content_detail, None),
                                 loop=self.conn.loop,
                             )
-                            future.result()
+                            future.result(timeout=self.tts_timeout)
                             logger.bind(tag=TAG).debug("TTS文本发送成功")
                         except Exception as e:
                             logger.bind(tag=TAG).error(f"发送TTS文本失败: {str(e)}")

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

@@ -31,6 +31,7 @@ handle_exit_intent_function_desc = {
     "handle_exit_intent", handle_exit_intent_function_desc, ToolType.SYSTEM_CTL
 )
 def handle_exit_intent(conn: "ConnectionHandler", say_goodbye: str | None = None):
+    conn.is_exiting = True
     # 处理退出意图
     try:
         if say_goodbye is None: