Merge pull request #3537 from xinnan-tech/py-test-tool

Py test tool

main/xiaozhi-server/core/providers/tools/server_plugins/plugin_executor.py

@@ -1,6 +1,8 @@
 """服务端插件工具执行器"""
 
 import asyncio
+import functools
+from concurrent.futures import ThreadPoolExecutor
 from typing import Dict, Any, TYPE_CHECKING
 
 if TYPE_CHECKING:
@@ -8,6 +10,10 @@ if TYPE_CHECKING:
 from ..base import ToolType, ToolDefinition, ToolExecutor
 from plugins_func.register import all_function_registry, module_func_map, Action, ActionResponse
 
+# 插件专用线程池：不用 asyncio 默认池（DNS 解析、ASR、initialize_modules 等都在用），
+# 避免慢插件打满默认池后拖垮所有协程的 DNS 解析和连接建立。
+_PLUGIN_POOL = ThreadPoolExecutor(max_workers=8, thread_name_prefix="plugin")
+
 
 class ServerPluginExecutor(ToolExecutor):
     """服务端插件工具执行器"""
@@ -28,23 +34,21 @@ class ServerPluginExecutor(ToolExecutor):
 
         try:
             # 根据工具类型决定如何调用
-            if hasattr(func_item, "type"):
-                func_type = func_item.type
-                if func_type.code in [4, 5]:  # SYSTEM_CTL, IOT_CTL (需要conn参数)
-                    result = func_item.func(conn, **arguments)
-                elif func_type.code == 2:  # WAIT
-                    result = func_item.func(**arguments)
-                elif func_type.code == 3:  # CHANGE_SYS_PROMPT
-                    result = func_item.func(conn, **arguments)
-                else:
-                    result = func_item.func(**arguments)
-            else:
-                # 默认不传conn参数
-                result = func_item.func(**arguments)
+            func_type = getattr(func_item, "type", None)
+            if func_type is not None and func_type.code in [3, 4, 5]:
+                # CHANGE_SYS_PROMPT, SYSTEM_CTL, IOT_CTL (需要conn参数)
+                call = functools.partial(func_item.func, conn, **arguments)
+            else:  # WAIT 等不传conn（含 type 未指定的插件）
+                call = functools.partial(func_item.func, **arguments)
 
-            # 兼容 async def 工具函数
-            if asyncio.iscoroutine(result):
-                result = await result
+            if asyncio.iscoroutinefunction(func_item.func):
+                result = await call()
+            else:
+                # 同步函数在 loop 线程上直接执行会卡住整个事件循环，放专用线程池。
+                result = await asyncio.get_running_loop().run_in_executor(_PLUGIN_POOL, call)
+                # 兼容同步函数返回协程的写法
+                if asyncio.iscoroutine(result):
+                    result = await result
 
             return result
 

main/xiaozhi-server/core/utils/util.py

@@ -135,7 +135,7 @@ def get_ip_info(ip_addr, logger):
         if is_private_ip(ip_addr):
             ip_addr = ""
         url = f"https://whois.pconline.com.cn/ipJson.jsp?json=true&ip={ip_addr}"
-        resp = requests.get(url).json()
+        resp = requests.get(url, timeout=5).json()
         ip_info = {"city": resp.get("city")}
 
         # 存入缓存

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

@@ -268,7 +268,8 @@ async def get_weather(conn: "ConnectionHandler", location: str = None, lang: str
                 location = cached_ip_info.get("city")
             else:
                 # 缓存未命中，调用API获取
-                ip_info = get_ip_info(client_ip, logger)
+                # get_ip_info 是同步 requests 调用，放线程池，避免卡住主事件循环
+                ip_info = await asyncio.to_thread(get_ip_info, client_ip, logger)
                 if ip_info:
                     cache_manager.set(CacheType.IP_INFO, client_ip, ip_info)
                     location = ip_info.get("city")

main/xiaozhi-server/plugins_func/register.py

@@ -83,7 +83,11 @@ module_func_map = {}
 
 
 def register_function(name, desc, type=None):
-    """注册函数到函数注册字典的装饰器"""
+    """注册函数到函数注册字典的装饰器
+
+    同步函数在工作线程中执行，函数体内不可调用 asyncio 事件循环 API；
+    需要异步（httpx、await）请定义为 async def。
+    """
 
     def decorator(func):
         all_function_registry[name] = FunctionItem(name, desc, func, type)

main/xiaozhi-server/tests/core/providers/tools/test_plugin_executor_offload.py

@@ -0,0 +1,134 @@
+"""服务端插件执行器：同步插件函数在线程池里跑，不卡事件循环；async 插件照常 await；异常变成 ERROR 响应。"""
+from __future__ import annotations
+
+import asyncio
+import time
+from types import SimpleNamespace
+
+from core.providers.tools.server_plugins.plugin_executor import ServerPluginExecutor
+from plugins_func.register import Action, ActionResponse, ToolType, all_function_registry
+
+
+def _register(name, func, tool_type):
+    item = SimpleNamespace(name=name, func=func, type=tool_type, description={})
+    all_function_registry[name] = item
+    return item
+
+
+def test_sync_plugin_runs_off_loop_and_async_plugin_is_awaited():
+    conn = SimpleNamespace(config={})
+
+    async def paced_sleep(delay):
+        # 不能用 asyncio.sleep 计时：conftest 在 mock 模式会把它换成不让出循环的桩，
+        # 这里放到线程池里睡，保证 ticker 每拍都真正让出事件循环。
+        await asyncio.get_running_loop().run_in_executor(None, time.sleep, delay)
+
+    def slow_sync(conn_arg, entity_id):
+        time.sleep(0.3)  # 模拟阻塞 IO
+        return ActionResponse(Action.RESPONSE, None, f"{entity_id} 已打开")
+
+    async def fast_async(question):
+        await asyncio.sleep(0)
+        return ActionResponse(Action.REQLLM, f"答：{question}", None)
+
+    def boom(conn_arg):
+        raise RuntimeError("连不上")
+
+    _register("t_slow_sync", slow_sync, ToolType.SYSTEM_CTL)
+    _register("t_fast_async", fast_async, ToolType.WAIT)
+    _register("t_boom", boom, ToolType.SYSTEM_CTL)
+    try:
+        async def go():
+            ex = ServerPluginExecutor(conn)
+            ticks = []
+
+            async def ticker():
+                while True:
+                    ticks.append(time.monotonic())
+                    await paced_sleep(0.02)
+
+            t = asyncio.create_task(ticker())
+            r = await ex.execute(conn, "t_slow_sync", {"entity_id": "light.living"})
+            t.cancel()
+            assert r.action == Action.RESPONSE and r.response == "light.living 已打开"
+            assert len(ticks) >= 5                      # 300 ms 里事件循环一直在转（没被 time.sleep 卡住）
+            r2 = await ex.execute(conn, "t_fast_async", {"question": "几点"})
+            assert r2.action == Action.REQLLM and r2.result == "答：几点"
+            r3 = await ex.execute(conn, "t_boom", {})
+            assert r3.action == Action.ERROR and "连不上" in r3.response
+            r4 = await ex.execute(conn, "t_missing", {})
+            assert r4.action == Action.NOTFOUND
+        asyncio.run(go())
+    finally:
+        for n in ("t_slow_sync", "t_fast_async", "t_boom"):
+            all_function_registry.pop(n, None)
+
+
+def test_conn_passing_rules_and_compat_branches():
+    """code 3/4/5 传 conn；WAIT/type=None 不传；同步函数返回协程会被兜底 await。"""
+    conn = SimpleNamespace(config={})
+    seen = {}
+
+    def prompt_fn(conn_arg):
+        seen["prompt"] = conn_arg
+        return ActionResponse(Action.RESPONSE, None, "切角色")
+
+    def wait_fn():
+        seen["wait"] = True
+        return ActionResponse(Action.RESPONSE, None, "等")
+
+    def no_type_fn():
+        seen["no_type"] = True
+        return ActionResponse(Action.RESPONSE, None, "无类型")
+
+    def sync_returns_coro(conn_arg):
+        seen["coro"] = conn_arg
+
+        async def inner():
+            return ActionResponse(Action.RESPONSE, None, "协程兜底")
+
+        return inner()
+
+    _register("t_prompt", prompt_fn, ToolType.CHANGE_SYS_PROMPT)
+    _register("t_wait", wait_fn, ToolType.WAIT)
+    _register("t_no_type", no_type_fn, None)
+    _register("t_sync_coro", sync_returns_coro, ToolType.SYSTEM_CTL)
+    try:
+        async def go():
+            ex = ServerPluginExecutor(conn)
+            r1 = await ex.execute(conn, "t_prompt", {})
+            assert seen["prompt"] is conn and r1.response == "切角色"
+            r2 = await ex.execute(conn, "t_wait", {})
+            assert "wait" in seen and r2.response == "等"
+            r3 = await ex.execute(conn, "t_no_type", {})
+            assert "no_type" in seen and r3.response == "无类型"
+            r4 = await ex.execute(conn, "t_sync_coro", {})
+            assert seen["coro"] is conn and r4.response == "协程兜底"
+        asyncio.run(go())
+    finally:
+        for n in ("t_prompt", "t_wait", "t_no_type", "t_sync_coro"):
+            all_function_registry.pop(n, None)
+
+
+def test_sync_plugin_uses_dedicated_pool():
+    """同步插件必须跑在专用池上，不能用 asyncio 默认池（DNS 解析等共用默认池）。"""
+    import threading
+
+    conn = SimpleNamespace(config={})
+    seen = {}
+
+    def sync_fn():
+        seen["thread"] = threading.current_thread().name
+        return ActionResponse(Action.RESPONSE, None, "ok")
+
+    _register("t_pool", sync_fn, ToolType.WAIT)
+    try:
+        async def go():
+            ex = ServerPluginExecutor(conn)
+            r = await ex.execute(conn, "t_pool", {})
+            assert r.response == "ok"
+
+        asyncio.run(go())
+        assert seen["thread"].startswith("plugin"), seen["thread"]  # 专用池线程名前缀
+    finally:
+        all_function_registry.pop("t_pool", None)