"""统一工具处理器"""
|
|
import json
|
from typing import Dict, List, Any, Optional
|
from config.logger import setup_logging
|
from plugins_func.loadplugins import auto_import_modules
|
|
from .base import ToolType
|
from plugins_func.register import Action, ActionResponse
|
from .unified_tool_manager import ToolManager
|
from .server_plugins import ServerPluginExecutor
|
from .server_mcp import ServerMCPExecutor
|
from .device_iot import DeviceIoTExecutor
|
from .device_mcp import DeviceMCPExecutor
|
from .mcp_endpoint import MCPEndpointExecutor
|
from core.handle.sendAudioHandle import send_display_message
|
|
|
class UnifiedToolHandler:
|
"""统一工具处理器"""
|
|
def __init__(self, conn):
|
self.conn = conn
|
self.config = conn.config
|
self.logger = setup_logging()
|
|
# 创建工具管理器
|
self.tool_manager = ToolManager(conn)
|
|
# 创建各类执行器
|
self.server_plugin_executor = ServerPluginExecutor(conn)
|
self.server_mcp_executor = ServerMCPExecutor(conn)
|
self.device_iot_executor = DeviceIoTExecutor(conn)
|
self.device_mcp_executor = DeviceMCPExecutor(conn)
|
self.mcp_endpoint_executor = MCPEndpointExecutor(conn)
|
|
# 注册执行器
|
self.tool_manager.register_executor(
|
ToolType.SERVER_PLUGIN, self.server_plugin_executor
|
)
|
self.tool_manager.register_executor(
|
ToolType.SERVER_MCP, self.server_mcp_executor
|
)
|
self.tool_manager.register_executor(
|
ToolType.DEVICE_IOT, self.device_iot_executor
|
)
|
self.tool_manager.register_executor(
|
ToolType.DEVICE_MCP, self.device_mcp_executor
|
)
|
self.tool_manager.register_executor(
|
ToolType.MCP_ENDPOINT, self.mcp_endpoint_executor
|
)
|
|
# 初始化标志
|
self.finish_init = False
|
|
async def _initialize(self):
|
"""异步初始化"""
|
try:
|
# 自动导入插件模块
|
auto_import_modules("plugins_func.functions")
|
|
# 初始化服务端MCP
|
await self.server_mcp_executor.initialize()
|
|
# 初始化MCP接入点
|
await self._initialize_mcp_endpoint()
|
|
# 初始化Home Assistant（如果需要）
|
self._initialize_home_assistant()
|
|
self.finish_init = True
|
self.logger.debug("统一工具处理器初始化完成")
|
|
# 输出当前支持的所有工具列表
|
self.current_support_functions()
|
|
except Exception as e:
|
self.logger.error(f"统一工具处理器初始化失败: {e}")
|
|
async def _initialize_mcp_endpoint(self):
|
"""初始化MCP接入点"""
|
try:
|
from .mcp_endpoint import connect_mcp_endpoint
|
|
# 从配置中获取MCP接入点URL
|
mcp_endpoint_url = self.config.get("mcp_endpoint", "")
|
|
if (
|
mcp_endpoint_url
|
and "你的" not in mcp_endpoint_url
|
and mcp_endpoint_url != "null"
|
):
|
self.logger.info(f"正在初始化MCP接入点: {mcp_endpoint_url}")
|
mcp_endpoint_client = await connect_mcp_endpoint(
|
mcp_endpoint_url, self.conn
|
)
|
|
if mcp_endpoint_client:
|
# 将MCP接入点客户端保存到连接对象中
|
self.conn.mcp_endpoint_client = mcp_endpoint_client
|
self.logger.info("MCP接入点初始化成功")
|
else:
|
self.logger.warning("MCP接入点初始化失败")
|
|
except Exception as e:
|
self.logger.error(f"初始化MCP接入点失败: {e}")
|
|
def _initialize_home_assistant(self):
|
"""初始化Home Assistant提示词"""
|
try:
|
from plugins_func.functions.hass_init import append_devices_to_prompt
|
|
append_devices_to_prompt(self.conn)
|
except ImportError:
|
pass # 忽略导入错误
|
except Exception as e:
|
self.logger.error(f"初始化Home Assistant失败: {e}")
|
|
def get_functions(self) -> List[Dict[str, Any]]:
|
"""获取所有工具的函数描述"""
|
return self.tool_manager.get_function_descriptions()
|
|
def current_support_functions(self) -> List[str]:
|
"""获取当前支持的函数名称列表"""
|
func_names = self.tool_manager.get_supported_tool_names()
|
self.logger.info(f"当前支持的函数列表: {func_names}")
|
return func_names
|
|
def upload_functions_desc(self):
|
"""刷新函数描述列表"""
|
self.tool_manager.refresh_tools()
|
self.logger.info("函数描述列表已刷新")
|
|
def has_tool(self, tool_name: str) -> bool:
|
"""检查是否有指定工具"""
|
return self.tool_manager.has_tool(tool_name)
|
|
async def handle_llm_function_call(
|
self, conn, function_call_data: Dict[str, Any]
|
) -> Optional[ActionResponse]:
|
"""处理LLM函数调用"""
|
try:
|
# 处理多函数调用
|
if "function_calls" in function_call_data:
|
responses = []
|
for call in function_call_data["function_calls"]:
|
result = await self.tool_manager.execute_tool(
|
call["name"], call.get("arguments", {})
|
)
|
responses.append(result)
|
return self._combine_responses(responses)
|
|
# 处理单函数调用
|
function_name = function_call_data["name"]
|
arguments = function_call_data.get("arguments", {})
|
|
# 如果arguments是字符串，尝试解析为JSON
|
if isinstance(arguments, str):
|
try:
|
arguments = json.loads(arguments) if arguments else {}
|
except json.JSONDecodeError:
|
self.logger.error(f"无法解析函数参数: {arguments}")
|
return ActionResponse(
|
action=Action.ERROR,
|
response="无法解析函数参数",
|
)
|
|
self.logger.debug(f"调用函数: {function_name}, 参数: {arguments}")
|
|
# 发送工具调用显示消息到设备
|
try:
|
await send_display_message(self.conn, f"% {function_name}")
|
except Exception as e:
|
self.logger.warning(f"发送工具调用显示消息失败: {e}")
|
|
# 执行工具调用
|
result = await self.tool_manager.execute_tool(function_name, arguments)
|
return result
|
|
except Exception as e:
|
self.logger.error(f"处理function call错误: {e}")
|
return ActionResponse(action=Action.ERROR, response=str(e))
|
|
def _combine_responses(self, responses: List[ActionResponse]) -> ActionResponse:
|
"""合并多个函数调用的响应"""
|
if not responses:
|
return ActionResponse(action=Action.NONE, response="无响应")
|
|
# 如果有任何错误，返回第一个错误
|
for response in responses:
|
if response.action == Action.ERROR:
|
return response
|
|
# 合并所有成功的响应
|
contents = []
|
responses_text = []
|
|
for response in responses:
|
if response.content:
|
contents.append(response.content)
|
if response.response:
|
responses_text.append(response.response)
|
|
# 确定最终的动作类型
|
final_action = Action.RESPONSE
|
for response in responses:
|
if response.action == Action.REQLLM:
|
final_action = Action.REQLLM
|
break
|
|
return ActionResponse(
|
action=final_action,
|
result="; ".join(contents) if contents else None,
|
response="; ".join(responses_text) if responses_text else None,
|
)
|
|
async def register_iot_tools(self, descriptors: List[Dict[str, Any]]):
|
"""注册IoT设备工具"""
|
self.device_iot_executor.register_iot_tools(descriptors)
|
self.tool_manager.refresh_tools()
|
self.logger.info(f"注册了{len(descriptors)}个IoT设备的工具")
|
|
def get_tool_statistics(self) -> Dict[str, int]:
|
"""获取工具统计信息"""
|
return self.tool_manager.get_tool_statistics()
|
|
async def cleanup(self):
|
"""清理资源"""
|
try:
|
await self.server_mcp_executor.cleanup()
|
|
# 清理MCP接入点连接
|
if (
|
hasattr(self.conn, "mcp_endpoint_client")
|
and self.conn.mcp_endpoint_client
|
):
|
await self.conn.mcp_endpoint_client.close()
|
|
self.logger.info("工具处理器清理完成")
|
except Exception as e:
|
self.logger.error(f"工具处理器清理失败: {e}")
|
