From fb41c79b1425b487562fc684c700764e342df02d Mon Sep 17 00:00:00 2001 From: OpenClaw Date: Fri, 8 May 2026 00:40:01 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=8F=92=E4=BB=B6=E5=86=85=E5=AD=98?= =?UTF-8?q?=E7=AE=A1=E7=90=86=20-=20TTL=E8=87=AA=E5=8A=A8=E6=B7=98?= =?UTF-8?q?=E6=B1=B0=20+=20=E6=87=92=E5=8A=A0=E8=BD=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 plugin_memory_manager.py: - 后台线程定期扫描空闲插件，超过TTL自动从内存卸载 - AI调用已卸载插件时自动重新加载（懒加载） - 核心插件(tool_search)可钉住永不淘汰 - 可配置: TTL(300s), 最大空闲数(10), 扫描间隔(60s), 钉住列表 改造 unified_tool_manager.py: - 集成 PluginMemoryManager - execute/mcp_call/search_tools/get_tool 统一走 _ensure_loaded - search_tools 可搜索已卸载插件（使用缓存信息） - 新增 get_all_tools_info/get_tool_by_name/memory_manager 属性 改造 tui.py: - /status 命令显示内存管理详细状态 - 侧栏显示已加载/已卸载工具数 --- unified_tool_manager.py | 451 ++++++++++++++++++++++++----- xcli_core/plugin_manager.py | 4 + xcli_core/plugin_memory_manager.py | 355 +++++++++++++++++++++++ xcli_core/tui.py | 25 +- 4 files changed, 761 insertions(+), 74 deletions(-) create mode 100644 xcli_core/plugin_memory_manager.py diff --git a/unified_tool_manager.py b/unified_tool_manager.py index 478af12..8afe08a 100644 --- a/unified_tool_manager.py +++ b/unified_tool_manager.py @@ -1,96 +1,143 @@ """ 统一工具管理器 - 协调 Liugin、Skill 和 MCP 三套协议 自动判断并路由到正确的处理器 + +集成插件内存管理: TTL 自动淘汰 + 懒加载 """ import os import json +import logging import threading from typing import Dict, List, Any, Optional, Union, Callable from colorama import Fore, Style +logger = logging.getLogger(__name__) + class UnifiedToolManager: """ 统一工具管理器 - + 协调三种协议: 1. Liugin 协议 (plugins/) - 小狸自有协议 2. Skill 协议 (skills/) - Agent Skills 开源协议 3. MCP 协议 - Model Context Protocol (Anthropic/OpenAI 标准) - + 路由策略: - 优先使用 Skill 协议（如果存在同名技能） - 回退到 Liugin 协议 - 支持强制指定协议 - MCP 格式调用统一入口 + + 内存管理: + - 空闲插件自动淘汰（TTL 机制） + - 调用时自动懒加载回来 + - 核心插件可钉住不淘汰 """ - - def __init__(self, cli_instance=None): + + # 内存管理默认配置 + DEFAULT_PLUGIN_TTL = 300 # 5分钟 + DEFAULT_MAX_IDLE_PLUGINS = 10 # 最多保留10个空闲插件 + DEFAULT_CHECK_INTERVAL = 60 # 60秒检查一次 + DEFAULT_PINNED_PLUGINS = ['tool_search'] # 默认钉住的插件 + + def __init__(self, cli_instance=None, + plugin_ttl: int = None, + max_idle_plugins: int = None, + check_interval: int = None, + pinned_plugins: list = None): self.cli = cli_instance self._lock = threading.RLock() - + # 两个独立的加载器 self._liugin_manager = None self._skill_loader = None - + # 统一的工具注册表 self._tools: Dict[str, Dict[str, Any]] = {} - + # 协议映射: tool_name -> protocol_type self._protocol_map: Dict[str, str] = {} - + # MCP 工具定义缓存 self._mcp_tools_cache: Optional[List[Dict]] = None + + # ── 原始文件路径缓存（用于懒加载） ── + self._plugin_paths: Dict[str, str] = {} # name -> file_path + self._skill_dirs: Dict[str, str] = {} # name -> skill_dir + + # ── 插件内存管理器 ── + from xcli_core.plugin_memory_manager import PluginMemoryManager + self._memory_mgr = PluginMemoryManager( + plugin_ttl=plugin_ttl or self.DEFAULT_PLUGIN_TTL, + max_idle_plugins=max_idle_plugins or self.DEFAULT_MAX_IDLE_PLUGINS, + check_interval=check_interval or self.DEFAULT_CHECK_INTERVAL, + pinned_plugins=pinned_plugins or self.DEFAULT_PINNED_PLUGINS, + ) + self._memory_mgr.set_callbacks( + load_fn=self._lazy_load_plugin, + unload_fn=self._unload_plugin, + ) def initialize(self, plugins_dir: str = "plugins", skills_dir: str = "skills") -> None: """ 初始化并加载所有工具 - + Args: plugins_dir: Liugin 目录 skills_dir: Skill 目录 """ # 加载 Liugins + self._plugins_dir = plugins_dir + self._skills_dir = skills_dir self._load_plugins(plugins_dir) - + # 加载 Skills self._load_skills(skills_dir) - + # 构建缓存 self._build_mcp_cache() - + + # 启动内存管理器 + self._memory_mgr.start() + print(f"{Fore.GREEN}[统一工具管理器] 加载完成: {len(self._tools)} 个工具{Style.RESET_ALL}") print(f" - Liugin 协议: {sum(1 for p in self._protocol_map.values() if p == 'liugin')}") print(f" - Skill 协议: {sum(1 for p in self._protocol_map.values() if p == 'skill')}") + + # 打印内存管理状态 + status = self._memory_mgr.get_status() + print(f" - 内存管理: TTL={status['ttl']}s, 最大空闲={status['max_idle']}, " + f"钉住={status['pinned']}") def _load_plugins(self, plugins_dir: str) -> None: - """加载 Liugin 协议工具""" + """加载 Liugin 协议工具（并注册到内存管理器）""" import importlib.util - + if not os.path.exists(plugins_dir): return - + # 获取启用的插件列表 enabled_plugins = self._get_enabled_plugins(plugins_dir) - + for filename in os.listdir(plugins_dir): if not filename.endswith('.py') or filename == '__init__.py': continue - + plugin_name = filename[:-3] - + # 检查是否启用 if enabled_plugins is not None and plugin_name.lower() not in enabled_plugins: continue - + plugin_path = os.path.join(plugins_dir, filename) - + try: spec = importlib.util.spec_from_file_location(plugin_name, plugin_path) module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) - + # 兼容 Plugin 和 Liugin 两种类名 plugin_class = None if hasattr(module, 'Liugin'): @@ -100,19 +147,19 @@ class UnifiedToolManager: if plugin_class: plugin_instance = plugin_class() - + # 设置 CLI 引用 if hasattr(plugin_instance, 'set_cli') and self.cli: plugin_instance.set_cli(self.cli) - + tool_info = plugin_instance.get_tool_info() tool_name = tool_info.get('name', plugin_name) - + # 获取 MCP 工具定义（如果插件提供） mcp_definition = None if hasattr(plugin_instance, 'get_mcp_definition'): mcp_definition = plugin_instance.get_mcp_definition() - + # 注册到统一工具表 self._tools[tool_name] = { 'name': tool_name, @@ -125,7 +172,23 @@ class UnifiedToolManager: 'mcp_definition': mcp_definition } self._protocol_map[tool_name] = 'liugin' - + + # 记录路径（用于懒加载） + self._plugin_paths[tool_name] = plugin_path + + # 注册到内存管理器 + self._memory_mgr.register( + name=tool_name, + path=plugin_path, + protocol='liugin', + info_cache={ + 'name': tool_name, + 'description': tool_info.get('description', ''), + 'keywords': tool_info.get('keywords', []), + 'usage': tool_info.get('usage', ''), + } + ) + except Exception as e: print(f"{Fore.RED}加载 Liugin '{plugin_name}' 失败: {e}{Style.RESET_ALL}") @@ -169,7 +232,7 @@ class UnifiedToolManager: # 判断技能格式 is_markdown = isinstance(skill, MarkdownSkill) skill_format = "markdown" if is_markdown else "python" - + # 注册到统一工具表 self._tools[skill_name] = { 'name': skill_name, @@ -185,6 +248,24 @@ class UnifiedToolManager: 'mcp_definition': tool_def # Skill 的 OpenAI 格式就是 MCP 格式 } self._protocol_map[skill_name] = 'skill' + + # 记录路径（用于懒加载） + skill_path = os.path.join(skills_dir, skill_name) + self._skill_dirs[skill_name] = skill_path + + # 注册到内存管理器 + self._memory_mgr.register( + name=skill_name, + path=skill_path, + protocol='skill', + info_cache={ + 'name': skill_name, + 'description': func_def.get('description', ''), + 'keywords': getattr(skill, 'tags', []), + 'format': skill_format, + } + ) + print(f"{Fore.GREEN}已加载 Skill ({skill_format}): {skill_name}{Style.RESET_ALL}") except ImportError as e: @@ -238,7 +319,168 @@ class UnifiedToolManager: return f"未知的技能类型: {type(skill)}" return handler - + + # ==================== 懒加载/卸载 ==================== + + def _lazy_load_plugin(self, name: str) -> Optional[Dict[str, Any]]: + """ + 懒加载单个插件（由内存管理器回调） + + Args: + name: 插件名 + + Returns: + tool_dict 如果成功，None 如果失败 + """ + import importlib.util + + # ── 尝试 Liugin 插件 ── + plugin_path = self._plugin_paths.get(name) + if plugin_path and os.path.exists(plugin_path): + return self._load_single_plugin(name, plugin_path) + + # ── 尝试 Skill ── + skill_dir = self._skill_dirs.get(name) + if skill_dir and os.path.exists(skill_dir): + return self._load_single_skill(name, skill_dir) + + return None + + def _load_single_plugin(self, name: str, plugin_path: str) -> Optional[Dict[str, Any]]: + """加载单个 Liugin 插件""" + import importlib.util + + try: + plugin_name = os.path.splitext(os.path.basename(plugin_path))[0] + spec = importlib.util.spec_from_file_location(plugin_name, plugin_path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + + plugin_class = getattr(module, 'Liugin', None) or getattr(module, 'Plugin', None) + if not plugin_class: + return None + + plugin_instance = plugin_class() + if hasattr(plugin_instance, 'set_cli') and self.cli: + plugin_instance.set_cli(self.cli) + + tool_info = plugin_instance.get_tool_info() + tool_name = tool_info.get('name', name) + + mcp_definition = None + if hasattr(plugin_instance, 'get_mcp_definition'): + mcp_definition = plugin_instance.get_mcp_definition() + + tool_dict = { + 'name': tool_name, + 'description': tool_info.get('description', ''), + 'keywords': tool_info.get('keywords', []), + 'usage': tool_info.get('usage', ''), + 'handler': plugin_instance.handle, + 'protocol': 'liugin', + 'instance': plugin_instance, + 'mcp_definition': mcp_definition + } + + with self._lock: + self._tools[tool_name] = tool_dict + self._protocol_map[tool_name] = 'liugin' + + return tool_dict + + except Exception as e: + logger.error(f"懒加载插件 '{name}' 失败: {e}") + return None + + def _load_single_skill(self, name: str, skill_dir: str) -> Optional[Dict[str, Any]]: + """加载单个 Skill""" + try: + from skills.base import SkillLoader, MarkdownSkill, Skill + + loader = SkillLoader(os.path.dirname(skill_dir)) + skill = loader.load_skill(skill_dir) + + if not skill: + return None + + if hasattr(skill, 'set_cli') and self.cli: + skill.set_cli(self.cli) + + tool_def = skill.get_tool_definition() + func_def = tool_def.get('function', {}) + + is_markdown = isinstance(skill, MarkdownSkill) + skill_format = "markdown" if is_markdown else "python" + + tool_dict = { + 'name': name, + 'description': func_def.get('description', ''), + 'keywords': getattr(skill, 'tags', []), + 'usage': getattr(skill, 'instructions', '')[:500] if is_markdown else getattr(skill, 'usage', ''), + 'handler': self._create_skill_handler(skill), + 'protocol': 'skill', + 'format': skill_format, + 'instance': skill, + 'tool_definition': tool_def, + 'anthropic_definition': skill.get_anthropic_tool_definition(), + 'mcp_definition': tool_def + } + + with self._lock: + self._tools[name] = tool_dict + self._protocol_map[name] = 'skill' + + return tool_dict + + except Exception as e: + logger.error(f"懒加载 Skill '{name}' 失败: {e}") + return None + + def _unload_plugin(self, name: str) -> None: + """ + 卸载插件实例（由内存管理器回调） + + 从工具表中移除，释放实例引用，但保留路径缓存以便重新加载 + """ + with self._lock: + if name in self._tools: + tool = self._tools[name] + # 清理实例引用 + instance = tool.get('instance') + if instance and hasattr(instance, 'cleanup'): + try: + instance.cleanup() + except Exception: + pass + del self._tools[name] + # 不删除 _protocol_map 和路径缓存，懒加载时需要 + + # 重建 MCP 缓存 + self._build_mcp_cache() + + def _ensure_loaded(self, name: str) -> Optional[Dict[str, Any]]: + """ + 确保插件已加载（调用入口统一使用） + + 如果插件已加载，更新使用时间戳并返回 tool_dict + 如果插件已卸载，自动懒加载后返回 tool_dict + 如果插件不存在，返回 None + """ + # 先检查是否在工具表中 + with self._lock: + tool = self._tools.get(name) + if tool: + self._memory_mgr.mark_used(name) + return tool + + # 不在工具表中，可能是已卸载，尝试懒加载 + result = self._memory_mgr.ensure_loaded(name) + if result: + self._memory_mgr.mark_used(name) + return result + + return None + def _get_enabled_plugins(self, plugins_dir: str) -> Optional[List[str]]: """获取启用的插件列表""" # 检查环境变量 @@ -347,10 +589,10 @@ class UnifiedToolManager: def tools(self) -> List[Dict[str, Any]]: """获取所有工具列表（兼容 LiuginManager 接口）""" return list(self._tools.values()) - + def get_tool(self, name: str) -> Optional[Dict[str, Any]]: - """获取指定工具""" - return self._tools.get(name) + """获取指定工具（支持懒加载）""" + return self._ensure_loaded(name) def get_protocol(self, name: str) -> Optional[str]: """获取工具使用的协议""" @@ -358,18 +600,18 @@ class UnifiedToolManager: def execute(self, tool_name: str, args: str = "", **kwargs) -> Dict[str, Any]: """ - 执行工具调用 - + 执行工具调用（支持懒加载） + Args: tool_name: 工具名称 args: 参数字符串（Plugin 格式） **kwargs: 关键字参数（Skill 格式或 MCP arguments） - + Returns: {"result": "..."} 格式的结果 """ - tool = self._tools.get(tool_name) - + tool = self._ensure_loaded(tool_name) + if not tool: return {"result": f"未找到工具: {tool_name}"} @@ -418,14 +660,14 @@ class UnifiedToolManager: def mcp_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: """ - MCP 格式的工具调用入口 - + MCP 格式的工具调用入口（支持懒加载） + 这是 MCP 协议的主要调用接口，接收结构化参数并返回结果 - + Args: tool_name: 工具名称 arguments: 结构化参数 (JSON对象) - + Returns: MCP 标准响应格式: { @@ -438,7 +680,7 @@ class UnifiedToolManager: "isError": false } """ - tool = self._tools.get(tool_name) + tool = self._ensure_loaded(tool_name) if not tool: return self._mcp_error(f"未找到工具: {tool_name}") @@ -607,12 +849,12 @@ class UnifiedToolManager: def search_tools(self, query: str, limit: int = 5) -> List[Dict[str, Any]]: """ - 搜索工具 - + 搜索工具（包括已卸载的插件，使用缓存信息） + Args: query: 搜索关键词 limit: 返回数量限制 - + Returns: 匹配的工具列表 """ @@ -620,44 +862,69 @@ class UnifiedToolManager: # 标准化：移除所有分隔符（连字符、下划线、空格） query_normalized = query_lower.replace('-', '').replace('_', '').replace(' ', '') matches = [] - + + # 搜索已加载的工具 for tool in self._tools.values(): - score = 0 - name = tool['name'].lower() - name_normalized = name.replace('-', '').replace('_', '') - desc = tool['description'].lower() - keywords = [kw.lower() for kw in tool.get('keywords', [])] - - # 名称完全匹配（忽略分隔符差异） - if query_lower == name or query_normalized == name_normalized: - score += 100 - # 名称包含（标准化后匹配） - elif query_normalized in name_normalized or query_lower in name: - score += 50 - # 描述包含 - if query_lower in desc: - score += 30 - # 关键词匹配 - for kw in keywords: - kw_normalized = kw.replace('-', '').replace('_', '') - if query_lower in kw or query_normalized in kw_normalized: - score += 20 - break - + score = self._score_tool_match(tool, query_lower, query_normalized) if score > 0: matches.append({ 'name': tool['name'], 'description': tool['description'], 'usage': tool.get('usage', ''), - 'keywords': keywords, + 'keywords': [kw.lower() for kw in tool.get('keywords', [])], 'protocol': tool['protocol'], 'instance': tool.get('instance'), 'score': score }) - + + # 搜索已卸载但注册的插件（使用缓存信息） + for name in self._memory_mgr.get_unloaded_names(): + # 跳过已加载的（上面已搜索） + if name in self._tools: + continue + info = self._memory_mgr.get_info_cache(name) + if info: + score = self._score_tool_match(info, query_lower, query_normalized) + if score > 0: + matches.append({ + 'name': info.get('name', name), + 'description': info.get('description', ''), + 'usage': info.get('usage', ''), + 'keywords': info.get('keywords', []), + 'protocol': info.get('protocol', 'liugin'), + 'instance': None, + 'score': score, + '_unloaded': True, # 标记为已卸载 + }) + # 按分数排序 matches.sort(key=lambda x: x['score'], reverse=True) return matches[:limit] + + def _score_tool_match(self, tool: Dict, query_lower: str, query_normalized: str) -> int: + """计算工具匹配分数""" + score = 0 + name = tool.get('name', '').lower() + name_normalized = name.replace('-', '').replace('_', '') + desc = tool.get('description', '').lower() + keywords = [kw.lower() for kw in tool.get('keywords', [])] + + # 名称完全匹配（忽略分隔符差异） + if query_lower == name or query_normalized == name_normalized: + score += 100 + # 名称包含（标准化后匹配） + elif query_normalized in name_normalized or query_lower in name: + score += 50 + # 描述包含 + if query_lower in desc: + score += 30 + # 关键词匹配 + for kw in keywords: + kw_normalized = kw.replace('-', '').replace('_', '') + if query_lower in kw or query_normalized in kw_normalized: + score += 20 + break + return score def register_tool(self, name: str, handler: callable, description: str = "", protocol: str = "custom", mcp_definition: Dict = None, **kwargs) -> None: @@ -695,10 +962,56 @@ class UnifiedToolManager: if name in self._tools: del self._tools[name] del self._protocol_map[name] + # 同时从内存管理器注销 + if name in self._plugin_paths: + del self._plugin_paths[name] + if name in self._skill_dirs: + del self._skill_dirs[name] # 重建缓存 self._build_mcp_cache() return True return False + + def get_all_tools_info(self) -> List[Dict[str, Any]]: + """ + 获取所有工具的简要信息（包括已卸载的） + + 用于系统提示词，让 AI 知道有哪些工具可用 + 已卸载的工具只有基本信息，调用时会自动加载 + """ + result = [] + # 已加载的 + for tool in self._tools.values(): + result.append({ + 'name': tool['name'], + 'description': tool.get('description', ''), + 'keywords': tool.get('keywords', []), + 'protocol': tool.get('protocol', ''), + 'loaded': True, + }) + # 已卸载的（从内存管理器缓存读取） + for name in self._memory_mgr.get_unloaded_names(): + if name in self._tools: + continue + info = self._memory_mgr.get_info_cache(name) + if info: + result.append({ + 'name': info.get('name', name), + 'description': info.get('description', ''), + 'keywords': info.get('keywords', []), + 'protocol': info.get('protocol', 'liugin'), + 'loaded': False, + }) + return result + + def get_tool_by_name(self, name: str) -> Optional[Dict[str, Any]]: + """根据名称获取工具（兼容 LiuginManager 接口，支持懒加载）""" + return self._ensure_loaded(name) + + @property + def memory_manager(self): + """获取内存管理器实例""" + return self._memory_mgr def get_skills_system_prompt(self) -> str: """ diff --git a/xcli_core/plugin_manager.py b/xcli_core/plugin_manager.py index cda3b3d..09378f4 100644 --- a/xcli_core/plugin_manager.py +++ b/xcli_core/plugin_manager.py @@ -130,3 +130,7 @@ class LiuginManager: if tool.get('name', '').lower() == tool_name.lower(): return tool return None + + def get_tool_info_only(self, tool_name): + """根据名称获取工具信息（不含 handler，用于搜索已卸载插件）""" + return self.get_tool_by_name(tool_name) diff --git a/xcli_core/plugin_memory_manager.py b/xcli_core/plugin_memory_manager.py new file mode 100644 index 0000000..3b28314 --- /dev/null +++ b/xcli_core/plugin_memory_manager.py @@ -0,0 +1,355 @@ +""" +插件内存管理器 - TTL 自动淘汰 + 懒加载 + +机制: + 1. 每个插件记录 last_used 时间戳 + 2. 后台线程定期扫描，超过 TTL 未使用的插件从内存卸载 + 3. AI 调用已卸载的插件时，自动重新加载 + 4. 核心插件（tool_search 等）可钉住，永不淘汰 + +配置: + - plugin_ttl: 插件空闲多久后淘汰（秒），默认 300（5分钟） + - max_idle_plugins: 最多保留多少个空闲插件，超出后淘汰最久未用的，默认 10 + - check_interval: 后台扫描间隔（秒），默认 60 + - pinned_plugins: 钉住的插件名列表，永不淘汰 +""" + +import os +import time +import threading +import importlib.util +from typing import Dict, Optional, List, Any, Callable +from colorama import Fore, Style + +from .config import logger + + +class PluginMemoryManager: + """插件内存管理器 - 自动淘汰空闲插件，按需懒加载""" + + # 默认钉住的插件（核心工具，不淘汰） + DEFAULT_PINNED = {'tool_search'} + + def __init__(self, + plugin_ttl: int = 300, + max_idle_plugins: int = 10, + check_interval: int = 60, + pinned_plugins: Optional[List[str]] = None): + self.plugin_ttl = plugin_ttl + self.max_idle_plugins = max_idle_plugins + self.check_interval = check_interval + self.pinned = set(pinned_plugins or self.DEFAULT_PINNED) + + # ── 插件元数据 ── + # name -> {path, protocol, last_used, info_cache, ...} + self._registry: Dict[str, Dict[str, Any]] = {} + self._lock = threading.RLock() + + # ── 已加载的实例引用（由 UnifiedToolManager 提供注册/注销回调） ── + self._load_callback: Optional[Callable] = None # (name) -> tool_dict or None + self._unload_callback: Optional[Callable] = None # (name) -> None + + # ── 后台清理线程 ── + self._cleanup_thread: Optional[threading.Thread] = None + self._running = False + + # ── 统计 ── + self.stats = { + 'evictions': 0, + 'lazy_loads': 0, + 'pins_hit': 0, + } + + # ==================== 生命周期 ==================== + + def start(self) -> None: + """启动后台清理线程""" + if self._running: + return + self._running = True + self._cleanup_thread = threading.Thread( + target=self._cleanup_loop, daemon=True, name="plugin-gc" + ) + self._cleanup_thread.start() + print(f"{Fore.CYAN}[插件内存管理] 已启动 (TTL={self.plugin_ttl}s, " + f"最大空闲={self.max_idle_plugins}, 检查间隔={self.check_interval}s){Style.RESET_ALL}") + + def stop(self) -> None: + """停止后台清理线程""" + self._running = False + if self._cleanup_thread: + self._cleanup_thread.join(timeout=5) + self._cleanup_thread = None + + # ==================== 回调注册 ==================== + + def set_callbacks(self, + load_fn: Callable[[str], Optional[Dict]], + unload_fn: Callable[[str], None]) -> None: + """ + 注册加载/卸载回调（由 UnifiedToolManager 提供） + + Args: + load_fn: 加载插件函数，接收插件名，返回 tool_dict 或 None + unload_fn: 卸载插件函数，接收插件名 + """ + self._load_callback = load_fn + self._unload_callback = unload_fn + + # ==================== 插件注册 ==================== + + def register(self, name: str, path: str, protocol: str = 'liugin', + info_cache: Optional[Dict] = None) -> None: + """ + 注册一个插件的元数据（加载时调用） + + Args: + name: 插件名 + path: 插件文件路径 + protocol: 协议类型 (liugin/skill) + info_cache: 缓存的工具信息（name, description, keywords 等） + """ + with self._lock: + self._registry[name] = { + 'path': path, + 'protocol': protocol, + 'last_used': time.time(), + 'loaded': True, + 'info_cache': info_cache or {}, + } + + def mark_used(self, name: str) -> None: + """标记插件被使用（每次工具调用时）""" + with self._lock: + if name in self._registry: + self._registry[name]['last_used'] = time.time() + + def mark_loaded(self, name: str) -> None: + """标记插件已加载""" + with self._lock: + if name in self._registry: + self._registry[name]['loaded'] = True + self._registry[name]['last_used'] = time.time() + + def mark_unloaded(self, name: str) -> None: + """标记插件已卸载""" + with self._lock: + if name in self._registry: + self._registry[name]['loaded'] = False + + # ==================== 懒加载 ==================== + + def ensure_loaded(self, name: str) -> Optional[Dict]: + """ + 确保插件已加载，如果已卸载则自动重新加载 + + Args: + name: 插件名 + + Returns: + tool_dict 如果加载成功，None 如果失败 + """ + with self._lock: + entry = self._registry.get(name) + if not entry: + return None # 未注册的插件，不管 + + if entry['loaded']: + # 已加载，更新时间戳 + entry['last_used'] = time.time() + return None # 返回 None 表示不需要重新加载 + + # 需要重新加载（锁外执行，避免死锁） + print(f"{Fore.CYAN}[插件内存管理] 懒加载插件: {name}{Style.RESET_ALL}") + result = self._do_load(name) + if result: + self.stats['lazy_loads'] += 1 + print(f"{Fore.GREEN}[插件内存管理] 懒加载成功: {name}{Style.RESET_ALL}") + else: + print(f"{Fore.RED}[插件内存管理] 懒加载失败: {name}{Style.RESET_ALL}") + return result + + def _do_load(self, name: str) -> Optional[Dict]: + """执行实际的插件加载""" + if not self._load_callback: + return None + + with self._lock: + entry = self._registry.get(name) + if not entry: + return None + + try: + result = self._load_callback(name) + if result: + with self._lock: + if name in self._registry: + self._registry[name]['loaded'] = True + self._registry[name]['last_used'] = time.time() + return result + except Exception as e: + logger.error(f"懒加载插件 '{name}' 失败: {e}") + return None + + # ==================== 淘汰 ==================== + + def evict(self, name: str) -> bool: + """ + 淘汰指定插件 + + Args: + name: 插件名 + + Returns: + 是否成功淘汰 + """ + if name in self.pinned: + self.stats['pins_hit'] += 1 + return False + + with self._lock: + entry = self._registry.get(name) + if not entry or not entry['loaded']: + return False + + # 执行卸载（锁外） + if self._unload_callback: + try: + self._unload_callback(name) + except Exception as e: + logger.error(f"卸载插件 '{name}' 失败: {e}") + return False + + with self._lock: + if name in self._registry: + self._registry[name]['loaded'] = False + + self.stats['evictions'] += 1 + print(f"{Fore.YELLOW}[插件内存管理] 已淘汰: {name} " + f"(空闲 {self._idle_time(name):.0f}s){Style.RESET_ALL}") + return True + + def _idle_time(self, name: str) -> float: + """获取插件空闲时间""" + with self._lock: + entry = self._registry.get(name) + if not entry: + return 0 + return time.time() - entry['last_used'] + + # ==================== 后台清理 ==================== + + def _cleanup_loop(self) -> None: + """后台清理循环""" + while self._running: + try: + time.sleep(self.check_interval) + if not self._running: + break + self._cleanup_once() + except Exception as e: + logger.error(f"插件清理线程异常: {e}") + + def _cleanup_once(self) -> None: + """执行一次清理""" + with self._lock: + # 收集所有已加载且未钉住的插件 + candidates = [] + for name, entry in self._registry.items(): + if entry['loaded'] and name not in self.pinned: + idle = time.time() - entry['last_used'] + if idle >= self.plugin_ttl: + candidates.append((name, idle)) + + if not candidates: + return + + # 按空闲时间排序，最久未用的优先淘汰 + candidates.sort(key=lambda x: x[1], reverse=True) + + # 也检查总数限制 + with self._lock: + loaded_count = sum(1 for e in self._registry.values() + if e['loaded'] and e.get('name') not in self.pinned) + + evicted = 0 + for name, idle in candidates: + # 如果已超过最大空闲数限制 或 超过 TTL + if loaded_count - evicted > self.max_idle_plugins or idle >= self.plugin_ttl: + if self.evict(name): + evicted += 1 + + if evicted: + print(f"{Fore.CYAN}[插件内存管理] 本轮淘汰 {evicted} 个插件, " + f"累计淘汰 {self.stats['evictions']} 次, " + f"懒加载 {self.stats['lazy_loads']} 次{Style.RESET_ALL}") + + # ==================== 查询接口 ==================== + + def is_loaded(self, name: str) -> bool: + """检查插件是否已加载""" + with self._lock: + entry = self._registry.get(name) + return entry['loaded'] if entry else False + + def is_registered(self, name: str) -> bool: + """检查插件是否已注册""" + with self._lock: + return name in self._registry + + def get_info_cache(self, name: str) -> Optional[Dict]: + """获取插件的缓存信息（不需要加载实例）""" + with self._lock: + entry = self._registry.get(name) + return entry.get('info_cache') if entry else None + + def get_all_registered(self) -> List[str]: + """获取所有已注册的插件名""" + with self._lock: + return list(self._registry.keys()) + + def get_loaded_names(self) -> List[str]: + """获取当前已加载的插件名""" + with self._lock: + return [n for n, e in self._registry.items() if e['loaded']] + + def get_unloaded_names(self) -> List[str]: + """获取当前已卸载的插件名""" + with self._lock: + return [n for n, e in self._registry.items() if not e['loaded']] + + def get_status(self) -> Dict[str, Any]: + """获取内存管理状态""" + with self._lock: + loaded = sum(1 for e in self._registry.values() if e['loaded']) + unloaded = sum(1 for e in self._registry.values() if not e['loaded']) + return { + 'total': len(self._registry), + 'loaded': loaded, + 'unloaded': unloaded, + 'pinned': list(self.pinned), + 'ttl': self.plugin_ttl, + 'max_idle': self.max_idle_plugins, + 'check_interval': self.check_interval, + 'stats': dict(self.stats), + } + + def format_status(self) -> str: + """格式化状态信息（用于 /status 命令）""" + s = self.get_status() + lines = [ + f" 插件内存管理:", + f" 总注册: {s['total']} 已加载: {s['loaded']} 已卸载: {s['unloaded']}", + f" TTL: {s['ttl']}s 最大空闲: {s['max_idle']} 扫描间隔: {s['check_interval']}s", + f" 钉住: {', '.join(s['pinned']) or '无'}", + f" 统计: 淘汰 {s['stats']['evictions']} 次, " + f"懒加载 {s['stats']['lazy_loads']} 次, " + f"钉住保护 {s['stats']['pins_hit']} 次", + ] + + # 显示已卸载的插件 + unloaded = self.get_unloaded_names() + if unloaded: + lines.append(f" 已卸载: {', '.join(unloaded)}") + + return '\n'.join(lines) diff --git a/xcli_core/tui.py b/xcli_core/tui.py index 0238374..b658947 100644 --- a/xcli_core/tui.py +++ b/xcli_core/tui.py @@ -562,6 +562,14 @@ if TEXTUAL_AVAILABLE: status_container.mount(Static(f" 工具: {len(self.bridge.tools())} 个", classes="tool-item")) status_container.mount(Static(f" 引擎: {len(self.bridge.engines())} 个", classes="tool-item")) + # 内存管理状态 + if hasattr(self.cli, 'liugin_manager') and hasattr(self.cli.liugin_manager, 'memory_manager'): + mgr = self.cli.liugin_manager.memory_manager + s = mgr.get_status() + status_container.mount(Static(f" 已加载: {s['loaded']}", classes="tool-item")) + if s['unloaded'] > 0: + status_container.mount(Static(f" 已卸载: {s['unloaded']}", classes="tool-item")) + def _update_status(self, text: str): bar = self.query_one("#status-bar") engine = self.bridge.current_engine() @@ -807,11 +815,18 @@ if TEXTUAL_AVAILABLE: self._system('\n'.join(lines)) def _show_status(self): - self._system(f""" 系统状态: - 引擎: {self.bridge.current_engine()} - 引擎数: {len(self.bridge.engines())} - 工具数: {len(self.bridge.tools())} - 对话数: {len(self.bridge.history())}""") + status_lines = [ + f" 系统状态:", + f" 引擎: {self.bridge.current_engine()}", + f" 引擎数: {len(self.bridge.engines())}", + f" 工具数: {len(self.bridge.tools())}", + f" 对话数: {len(self.bridge.history())}", + ] + # 内存管理状态 + if hasattr(self.cli, 'liugin_manager') and hasattr(self.cli.liugin_manager, 'memory_manager'): + mgr = self.cli.liugin_manager.memory_manager + status_lines.append(mgr.format_status()) + self._system('\n'.join(status_lines)) # ── AI 生成 ──