fix(tools): 服务端插件的同步函数放线程池，不阻塞事件循环 (#3528)

ServerPluginExecutor.execute 在事件循环上被 await，而 hass_state、天气等插件用 requests 同步阻塞，工具执行期间整个循环停转、正在发的音频节拍也停。同步函数改用 run_in_executor 放到线程池；async def 插件照常 await；传参规则不变。附测试。

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

@@ -1,6 +1,7 @@
 """服务端插件工具执行器"""
 
 import asyncio
+import functools
 from typing import Dict, Any, TYPE_CHECKING
 
 if TYPE_CHECKING:
@@ -30,21 +31,23 @@ class ServerPluginExecutor(ToolExecutor):
             # 根据工具类型决定如何调用
             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)
+                if 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
+                    call = functools.partial(func_item.func, **arguments)
             else:
                 # 默认不传conn参数
-                result = func_item.func(**arguments)
+                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:
+                # 同步插件函数（requests 等阻塞调用）放到线程池：这个方法是在事件循环上被 await 的，
+                # 直接调用会把整个循环卡住，正在给设备发的音频节拍也会停。
+                result = await asyncio.get_running_loop().run_in_executor(None, call)
+                # 兼容同步函数返回协程的写法
+                if asyncio.iscoroutine(result):
+                    result = await result
 
             return result
 

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

@@ -0,0 +1,62 @@
+"""服务端插件执行器：同步插件函数在线程池里跑，不卡事件循环；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
+
+_real_sleep = asyncio.tasks.sleep  # conftest 在 mock 模式把 asyncio.sleep 换成空函数
+
+
+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(monkeypatch):
+    monkeypatch.setattr(asyncio, "sleep", _real_sleep)
+    conn = SimpleNamespace(config={})
+
+    def slow_sync(conn_arg, entity_id):
+        time.sleep(0.3)  # 模拟 requests 调 HA
+        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 asyncio.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) >= 8                      # 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)