""" LLM 工具体系公共模块 统一工具收集、参数校验与执行结果结构,供 AIChat 等插件使用。 """ from __future__ import annotations import json from dataclasses import dataclass from typing import Any, Dict, List, Optional, Tuple from loguru import logger @dataclass class ToolResult: """统一的工具执行结果结构""" success: bool = True message: str = "" need_ai_reply: bool = False already_sent: bool = False send_result_text: bool = False no_reply: bool = False save_to_memory: bool = False @classmethod def from_raw(cls, raw: Any) -> Optional["ToolResult"]: if raw is None: return None if not isinstance(raw, dict): return cls(success=True, message=str(raw)) msg = raw.get("message", "") if not isinstance(msg, str): try: msg = json.dumps(msg, ensure_ascii=False) except Exception: msg = str(msg) return cls( success=bool(raw.get("success", True)), message=msg, need_ai_reply=bool(raw.get("need_ai_reply", False)), already_sent=bool(raw.get("already_sent", False)), send_result_text=bool(raw.get("send_result_text", False)), no_reply=bool(raw.get("no_reply", False)), save_to_memory=bool(raw.get("save_to_memory", False)), ) def collect_tools_with_plugins( tools_config: Dict[str, Any], plugins: Dict[str, Any], ) -> Dict[str, Tuple[str, Dict[str, Any]]]: """ 收集所有插件的 LLM 工具,并保留来源插件名。 Args: tools_config: AIChat 配置中的 [tools] 节 plugins: PluginManager().plugins 映射 Returns: {tool_name: (plugin_name, tool_dict)} """ tools_by_name: Dict[str, Tuple[str, Dict[str, Any]]] = {} mode = tools_config.get("mode", "all") whitelist = set(tools_config.get("whitelist", [])) blacklist = set(tools_config.get("blacklist", [])) for plugin_name, plugin in plugins.items(): if not hasattr(plugin, "get_llm_tools"): continue plugin_tools = plugin.get_llm_tools() or [] for tool in plugin_tools: tool_name = tool.get("function", {}).get("name", "") if not tool_name: continue if mode == "whitelist" and tool_name not in whitelist: continue if mode == "blacklist" and tool_name in blacklist: logger.debug(f"[黑名单] 禁用工具: {tool_name}") continue if tool_name in tools_by_name: logger.warning(f"重复工具名 {tool_name} 来自 {plugin_name},已忽略") continue tools_by_name[tool_name] = (plugin_name, tool) if mode == "whitelist": logger.debug(f"[白名单] 启用工具: {tool_name}") return tools_by_name def collect_tools( tools_config: Dict[str, Any], plugins: Dict[str, Any], ) -> List[Dict[str, Any]]: """仅返回工具定义列表""" return [item[1] for item in collect_tools_with_plugins(tools_config, plugins).values()] def get_tool_schema_map( tools_map: Dict[str, Tuple[str, Dict[str, Any]]], ) -> Dict[str, Dict[str, Any]]: """构建工具名到参数 schema 的映射""" schema_map: Dict[str, Dict[str, Any]] = {} for name, (_plugin_name, tool) in tools_map.items(): fn = tool.get("function", {}) schema_map[name] = fn.get("parameters", {}) or {} return schema_map def validate_tool_arguments( tool_name: str, arguments: Dict[str, Any], schema: Optional[Dict[str, Any]], ) -> Tuple[bool, str, Dict[str, Any]]: """ 轻量校验并补全默认参数。 Returns: (ok, error_message, new_arguments) """ if not schema: return True, "", arguments props = schema.get("properties", {}) or {} required = schema.get("required", []) or [] # 应用默认值 for key, prop in props.items(): if key not in arguments and isinstance(prop, dict) and "default" in prop: arguments[key] = prop["default"] missing = [] for key in required: if key not in arguments or arguments[key] in (None, "", []): missing.append(key) if missing: return False, f"缺少参数: {', '.join(missing)}", arguments # 枚举与基础类型校验 for key, prop in props.items(): if key not in arguments or not isinstance(prop, dict): continue value = arguments[key] if "enum" in prop and value not in prop["enum"]: return False, f"参数 {key} 必须是 {prop['enum']}", arguments expected_type = prop.get("type") if expected_type == "integer": try: arguments[key] = int(value) except Exception: return False, f"参数 {key} 应为整数", arguments elif expected_type == "number": try: arguments[key] = float(value) except Exception: return False, f"参数 {key} 应为数字", arguments elif expected_type == "boolean": if isinstance(value, bool): continue if isinstance(value, str) and value.lower() in ("true", "false", "1", "0"): arguments[key] = value.lower() in ("true", "1") else: return False, f"参数 {key} 应为布尔值", arguments elif expected_type == "string": if not isinstance(value, str): arguments[key] = str(value) return True, "", arguments