diff --git a/plugins/workflow.py b/plugins/workflow.py new file mode 100644 index 0000000..280ad98 --- /dev/null +++ b/plugins/workflow.py @@ -0,0 +1,804 @@ +""" +工作流插件 - YAML 定义的多步骤自动化流程 + +支持: + - 顺序执行多步骤 + - 步骤间变量传递 ({{prev.output}}) + - 条件分支 (on_fail / on_success) + - 错误处理 (continue_on_error) + - 状态持久化 (断点续跑) + - 调用其他插件作为步骤 + +用法: + /workflow run + /workflow list + /workflow status + /workflow resume + /workflow abort +""" + +import json +import os +import re +import time +import uuid +from datetime import datetime +from colorama import Fore, Style + + +# ── YAML 解析（内置轻量实现，不依赖 PyYAML） ── + +def _parse_yaml(text: str) -> dict: + """轻量 YAML 解析器，支持工作流所需的子集""" + import yaml # 延迟导入，失败时回退 + return yaml.safe_load(text) + + +def _parse_yaml_fallback(text: str) -> dict: + """内置回退解析器，仅支持简单结构""" + result = {} + current_key = None + current_list = None + current_dict = None + + for line in text.split('\n'): + stripped = line.strip() + if not stripped or stripped.startswith('#'): + continue + + indent = len(line) - len(line.lstrip()) + + if indent == 0 and ':' in stripped: + key, _, val = stripped.partition(':') + key = key.strip() + val = val.strip() + if val: + result[key] = _yaml_value(val) + else: + result[key] = None + current_key = key + current_list = None + current_dict = None + elif indent > 0 and stripped.startswith('- '): + if current_key: + if not isinstance(result.get(current_key), list): + result[current_key] = [] + item_text = stripped[2:].strip() + if ':' in item_text: + item = {} + k, _, v = item_text.partition(':') + item[k.strip()] = _yaml_value(v.strip()) + current_dict = item + result[current_key].append(item) + else: + result[current_key].append(_yaml_value(item_text)) + elif indent > 0 and ':' in stripped and current_dict is not None: + key, _, val = stripped.partition(':') + current_dict[key.strip()] = _yaml_value(val.strip()) + + return result + + +def _yaml_value(s: str): + """转换单个 YAML 值""" + if not s: + return None + # 去除引号 + if (s.startswith('"') and s.endswith('"')) or (s.startswith("'") and s.endswith("'")): + return s[1:-1] + if s.lower() in ('true', 'yes', 'on'): + return True + if s.lower() in ('false', 'no', 'off'): + return False + if s.lower() in ('null', 'none', '~'): + return None + try: + return int(s) + except ValueError: + pass + try: + return float(s) + except ValueError: + pass + return s + + +def _load_yaml_file(path: str) -> dict: + """加载 YAML 文件""" + with open(path, 'r', encoding='utf-8') as f: + text = f.read() + try: + return _parse_yaml(text) + except ImportError: + return _parse_yaml_fallback(text) + + +# ── 变量替换引擎 ── + +def _resolve_vars(text: str, context: dict) -> str: + """ + 替换 {{key}} 和 {{key.nested}} 形式的变量 + + 支持: + {{var}} → context['var'] + {{step.output}} → context['steps']['step']['output'] + {{step.exit_code}} → context['steps']['step']['exit_code'] + {{env.HOME}} → os.environ['HOME'] + {{args.name}} → context['args']['name'] + """ + def replacer(match): + path = match.group(1).strip() + parts = path.split('.') + + if parts[0] == 'env': + return os.environ.get('.'.join(parts[1:]), '') + + if parts[0] == 'args' and len(parts) > 1: + val = context.get('args', {}) + for p in parts[1:]: + if isinstance(val, dict): + val = val.get(p, '') + else: + return '' + return str(val) + + if parts[0] == 'steps' and len(parts) >= 3: + step_name = parts[1] + attr = parts[2] + step_data = context.get('steps', {}).get(step_name, {}) + val = step_data.get(attr, '') + # 支持 deeper nesting: {{steps.foo.output.field}} + for p in parts[3:]: + if isinstance(val, dict): + val = val.get(p, '') + else: + return '' + return str(val) + + # 顶层变量 + val = context + for p in parts: + if isinstance(val, dict): + val = val.get(p, '') + else: + return '' + return str(val) + + return re.sub(r'\{\{(.+?)\}\}', replacer, text) + + +def _resolve_value(val, context): + """递归解析值中的变量（支持 dict/list/str）""" + if isinstance(val, str): + return _resolve_vars(val, context) + elif isinstance(val, dict): + return {k: _resolve_value(v, context) for k, v in val.items()} + elif isinstance(val, list): + return [_resolve_value(item, context) for item in val] + return val + + +# ── 工作流引擎 ── + +class WorkflowEngine: + """工作流执行引擎""" + + def __init__(self, cli_instance=None): + self.cli = cli_instance + self.workflows_dir = "workflows" + self.state_dir = os.path.join(self.workflows_dir, ".state") + os.makedirs(self.state_dir, exist_ok=True) + + def run(self, path: str, args: dict = None) -> str: + """执行工作流""" + if not os.path.exists(path): + return f"❌ 工作流文件不存在: {path}" + + try: + wf = _load_yaml_file(path) + except Exception as e: + return f"❌ 解析工作流文件失败: {e}" + + wf_name = wf.get('name', os.path.basename(path).replace('.yaml', '').replace('.yml', '')) + steps = wf.get('steps', []) + if not steps: + return f"❌ 工作流 '{wf_name}' 没有步骤" + + wf_id = f"{wf_name}-{uuid.uuid4().hex[:8]}" + context = { + 'args': args or {}, + 'steps': {}, + 'workflow': { + 'id': wf_id, + 'name': wf_name, + 'path': path, + 'started_at': datetime.now().isoformat(), + } + } + + # 保存初始状态 + self._save_state(wf_id, { + 'workflow': wf_name, + 'path': path, + 'status': 'running', + 'current_step': 0, + 'context': context, + 'started_at': context['workflow']['started_at'], + }) + + print(f"\n{Fore.CYAN}🔄 工作流 '{wf_name}' 开始 (ID: {wf_id}){Style.RESET_ALL}") + print(f" 共 {len(steps)} 个步骤\n") + + failed = False + for i, step in enumerate(steps): + step_name = step.get('name', f'step_{i+1}') + step_type = step.get('type', 'tool') + condition = step.get('if', None) + continue_on_error = step.get('continue_on_error', False) + + # 检查条件 + if condition: + resolved_condition = _resolve_vars(condition, context) + if not self._eval_condition(resolved_condition): + print(f" {Fore.YELLOW}⏭ [{step_name}] 条件不满足，跳过{Style.RESET_ALL}") + context['steps'][step_name] = { + 'output': '', 'exit_code': 0, 'skipped': True + } + continue + + # 解析步骤参数 + resolved_step = _resolve_value(step, context) + + print(f" {Fore.CYAN}▶ [{step_name}] {step_type}{Style.RESET_ALL}") + + # 更新状态 + self._save_state(wf_id, { + 'workflow': wf_name, + 'path': path, + 'status': 'running', + 'current_step': i, + 'current_step_name': step_name, + 'context': context, + 'started_at': context['workflow']['started_at'], + }) + + try: + result = self._execute_step(resolved_step, context) + context['steps'][step_name] = result + + if result.get('exit_code', 0) == 0: + output_preview = str(result.get('output', ''))[:100] + print(f" {Fore.GREEN}✓ [{step_name}] 完成{Style.RESET_ALL}" + f"{' → ' + output_preview if output_preview else ''}") + else: + failed = True + print(f" {Fore.RED}✗ [{step_name}] 失败 (exit={result.get('exit_code')}){Style.RESET_ALL}") + print(f" {result.get('error', result.get('output', ''))[:200]}") + + if not continue_on_error: + # 检查 on_fail 分支 + on_fail = resolved_step.get('on_fail') + if on_fail: + print(f" {Fore.YELLOW}→ 执行 on_fail 分支{Style.RESET_ALL}") + self._execute_on_fail(on_fail, context) + break + else: + print(f" {Fore.YELLOW}⚠ continue_on_error=true，继续{Style.RESET_ALL}") + + except Exception as e: + failed = True + context['steps'][step_name] = { + 'output': '', 'exit_code': 1, 'error': str(e) + } + print(f" {Fore.RED}✗ [{step_name}] 异常: {e}{Style.RESET_ALL}") + if not continue_on_error: + break + + # 完成状态 + status = 'failed' if failed else 'completed' + context['workflow']['finished_at'] = datetime.now().isoformat() + context['workflow']['status'] = status + + self._save_state(wf_id, { + 'workflow': wf_name, + 'path': path, + 'status': status, + 'current_step': len(steps), + 'context': context, + 'started_at': context['workflow']['started_at'], + 'finished_at': context['workflow']['finished_at'], + }) + + icon = '✅' if not failed else '❌' + print(f"\n{icon} 工作流 '{wf_name}' {status} (ID: {wf_id})") + return f"{icon} 工作流 '{wf_name}' {status}" + + def resume(self, wf_id: str) -> str: + """从断点恢复执行工作流""" + state = self._load_state(wf_id) + if not state: + return f"❌ 未找到工作流状态: {wf_id}" + + if state.get('status') == 'completed': + return f"ℹ️ 工作流已完成: {wf_id}" + + path = state.get('path', '') + if not os.path.exists(path): + return f"❌ 工作流文件不存在: {path}" + + try: + wf = _load_yaml_file(path) + except Exception as e: + return f"❌ 解析工作流文件失败: {e}" + + steps = wf.get('steps', []) + start_from = state.get('current_step', 0) + context = state.get('context', {}) + + print(f"\n{Fore.CYAN}🔄 恢复工作流 '{wf.get('name', wf_id)}' 从步骤 {start_from + 1}{Style.RESET_ALL}") + + failed = False + for i in range(start_from, len(steps)): + step = steps[i] + step_name = step.get('name', f'step_{i+1}') + step_type = step.get('type', 'tool') + continue_on_error = step.get('continue_on_error', False) + + resolved_step = _resolve_value(step, context) + print(f" {Fore.CYAN}▶ [{step_name}] {step_type}{Style.RESET_ALL}") + + try: + result = self._execute_step(resolved_step, context) + context['steps'][step_name] = result + + if result.get('exit_code', 0) == 0: + print(f" {Fore.GREEN}✓ [{step_name}] 完成{Style.RESET_ALL}") + else: + failed = True + print(f" {Fore.RED}✗ [{step_name}] 失败{Style.RESET_ALL}") + if not continue_on_error: + break + except Exception as e: + failed = True + context['steps'][step_name] = {'output': '', 'exit_code': 1, 'error': str(e)} + print(f" {Fore.RED}✗ [{step_name}] 异常: {e}{Style.RESET_ALL}") + if not continue_on_error: + break + + status = 'failed' if failed else 'completed' + self._save_state(wf_id, {**state, 'status': status, 'current_step': len(steps)}) + + return f"{'✅' if not failed else '❌'} 工作流 {status}" + + def list_workflows(self) -> str: + """列出所有工作流文件""" + if not os.path.exists(self.workflows_dir): + return "📁 workflows/ 目录不存在" + + files = [f for f in os.listdir(self.workflows_dir) + if f.endswith(('.yaml', '.yml')) and not f.startswith('.')] + + if not files: + return "📁 workflows/ 目录中没有工作流文件" + + result = f"📁 工作流列表 ({len(files)} 个):\n" + for f in sorted(files): + path = os.path.join(self.workflows_dir, f) + try: + wf = _load_yaml_file(path) + name = wf.get('name', f.replace('.yaml', '').replace('.yml', '')) + desc = wf.get('description', '') + steps_count = len(wf.get('steps', [])) + result += f" 📄 {f}\n" + result += f" 名称: {name} 步骤: {steps_count}\n" + if desc: + result += f" 描述: {desc}\n" + except Exception: + result += f" 📄 {f} (解析失败)\n" + return result + + def status(self, wf_id: str) -> str: + """查看工作流状态""" + state = self._load_state(wf_id) + if not state: + return f"❌ 未找到工作流: {wf_id}" + + ctx = state.get('context', {}) + wf_info = ctx.get('workflow', {}) + steps_done = ctx.get('steps', {}) + + lines = [ + f"📋 工作流: {state.get('workflow', '?')}", + f" ID: {wf_id}", + f" 状态: {state.get('status', '?')}", + f" 进度: 步骤 {state.get('current_step', 0)}", + f" 开始: {state.get('started_at', '?')}", + ] + if state.get('finished_at'): + lines.append(f" 结束: {state['finished_at']}") + + if steps_done: + lines.append(f" 已完成步骤:") + for name, data in steps_done.items(): + icon = '✓' if data.get('exit_code', 0) == 0 else ('⏭' if data.get('skipped') else '✗') + lines.append(f" {icon} {name}") + + return '\n'.join(lines) + + def list_states(self) -> str: + """列出所有工作流运行状态""" + if not os.path.exists(self.state_dir): + return "没有工作流运行记录" + + files = [f for f in os.listdir(self.state_dir) if f.endswith('.json')] + if not files: + return "没有工作流运行记录" + + result = f"📋 工作流运行记录 ({len(files)} 个):\n" + for f in sorted(files, reverse=True)[:20]: + wf_id = f.replace('.json', '') + try: + with open(os.path.join(self.state_dir, f), 'r') as fh: + state = json.load(fh) + status_icon = {'running': '🔄', 'completed': '✅', 'failed': '❌'}.get(state.get('status'), '❓') + result += f" {status_icon} {wf_id} — {state.get('workflow', '?')} ({state.get('status', '?')})\n" + except Exception: + result += f" ❓ {wf_id}\n" + return result + + def abort(self, wf_id: str) -> str: + """中止工作流""" + state = self._load_state(wf_id) + if not state: + return f"❌ 未找到工作流: {wf_id}" + + state['status'] = 'aborted' + self._save_state(wf_id, state) + return f"🛑 工作流已中止: {wf_id}" + + # ── 步骤执行 ── + + def _execute_step(self, step: dict, context: dict) -> dict: + """执行单个步骤""" + step_type = step.get('type', 'tool') + + if step_type == 'tool': + return self._exec_tool_step(step, context) + elif step_type == 'shell': + return self._exec_shell_step(step, context) + elif step_type == 'set': + return self._exec_set_step(step, context) + elif step_type == 'echo': + return self._exec_echo_step(step, context) + elif step_type == 'wait': + return self._exec_wait_step(step, context) + elif step_type == 'if': + return self._exec_if_step(step, context) + else: + return {'output': '', 'exit_code': 1, 'error': f'未知步骤类型: {step_type}'} + + def _exec_tool_step(self, step: dict, context: dict) -> dict: + """执行插件工具步骤""" + tool_name = step.get('tool', '') + tool_args = step.get('args', '') + + if not tool_name: + return {'output': '', 'exit_code': 1, 'error': '未指定 tool 名称'} + + if not self.cli: + return {'output': '', 'exit_code': 1, 'error': 'CLI 实例不可用'} + + try: + # 获取工具管理器 + tool_mgr = getattr(self.cli, 'tool_manager', None) or getattr(self.cli, 'liugin_manager', None) + if not tool_mgr: + return {'output': '', 'exit_code': 1, 'error': '工具管理器不可用'} + + # 解析参数 + args_str = str(tool_args) if tool_args else '' + + # 调用工具 + if hasattr(tool_mgr, 'execute'): + result = tool_mgr.execute(tool_name, args_str) + output = result.get('result', '') if isinstance(result, dict) else str(result) + elif hasattr(tool_mgr, 'tools'): + # LiuginManager 兼容 + tool = tool_mgr.get_tool_by_name(tool_name) + if tool and 'handler' in tool: + output = tool['handler'](args_str) + else: + return {'output': '', 'exit_code': 1, 'error': f'未找到工具: {tool_name}'} + else: + return {'output': '', 'exit_code': 1, 'error': '无法调用工具'} + + return {'output': str(output), 'exit_code': 0} + + except Exception as e: + return {'output': '', 'exit_code': 1, 'error': str(e)} + + def _exec_shell_step(self, step: dict, context: dict) -> dict: + """执行 shell 命令步骤""" + import subprocess + + cmd = step.get('command', step.get('cmd', '')) + if not cmd: + return {'output': '', 'exit_code': 1, 'error': '未指定 command'} + + cwd = step.get('cwd', None) + timeout = step.get('timeout', 60) + + try: + result = subprocess.run( + cmd, shell=True, capture_output=True, text=True, + timeout=timeout, cwd=cwd + ) + return { + 'output': result.stdout.strip(), + 'error': result.stderr.strip(), + 'exit_code': result.returncode, + } + except subprocess.TimeoutExpired: + return {'output': '', 'exit_code': 1, 'error': f'命令超时 ({timeout}s)'} + except Exception as e: + return {'output': '', 'exit_code': 1, 'error': str(e)} + + def _exec_set_step(self, step: dict, context: dict) -> dict: + """设置变量步骤""" + variables = step.get('variables', step.get('vars', {})) + for key, val in variables.items(): + context[key] = _resolve_value(val, context) + return {'output': f'设置 {len(variables)} 个变量', 'exit_code': 0} + + def _exec_echo_step(self, step: dict, context: dict) -> dict: + """打印步骤""" + message = step.get('message', step.get('text', '')) + resolved = _resolve_vars(str(message), context) + print(f" 💬 {resolved}") + return {'output': resolved, 'exit_code': 0} + + def _exec_wait_step(self, step: dict, context: dict) -> dict: + """等待步骤""" + seconds = step.get('seconds', step.get('time', 1)) + time.sleep(seconds) + return {'output': f'等待 {seconds} 秒', 'exit_code': 0} + + def _exec_if_step(self, step: dict, context: dict) -> dict: + """条件分支步骤""" + condition = step.get('condition', '') + resolved = _resolve_vars(str(condition), context) + + if self._eval_condition(resolved): + then_steps = step.get('then', []) + return self._run_sub_steps(then_steps, context) + else: + else_steps = step.get('else', []) + if else_steps: + return self._run_sub_steps(else_steps, context) + return {'output': '条件不满足，无 else 分支', 'exit_code': 0} + + def _run_sub_steps(self, steps: list, context: dict) -> dict: + """执行子步骤列表""" + outputs = [] + for sub_step in steps: + result = self._execute_step(sub_step, context) + outputs.append(result.get('output', '')) + if result.get('exit_code', 0) != 0: + return result + return {'output': '\n'.join(outputs), 'exit_code': 0} + + def _eval_condition(self, condition: str) -> bool: + """评估条件表达式""" + condition = condition.strip() + if not condition: + return True + + # 简单比较: "value == expected" + for op in ['==', '!=', '>=', '<=', '>', '<']: + if op in condition: + left, _, right = condition.partition(op) + left = left.strip().strip('"').strip("'") + right = right.strip().strip('"').strip("'") + try: + left_n, right_n = float(left), float(right) + if op == '==': return left_n == right_n + if op == '!=': return left_n != right_n + if op == '>=': return left_n >= right_n + if op == '<=': return left_n <= right_n + if op == '>': return left_n > right_n + if op == '<': return left_n < right_n + except ValueError: + if op == '==': return left == right + if op == '!=': return left != right + + # 布尔值 + if condition.lower() in ('true', '1', 'yes'): + return True + if condition.lower() in ('false', '0', 'no', ''): + return False + + # 非空即真 + return bool(condition) + + def _execute_on_fail(self, on_fail, context: dict): + """执行 on_fail 处理""" + if isinstance(on_fail, str): + print(f" 💬 {on_fail}") + elif isinstance(on_fail, list): + for step in on_fail: + self._execute_step(step, context) + + # ── 状态持久化 ── + + def _save_state(self, wf_id: str, state: dict): + """保存工作流状态""" + path = os.path.join(self.state_dir, f"{wf_id}.json") + try: + with open(path, 'w', encoding='utf-8') as f: + json.dump(state, f, ensure_ascii=False, indent=2, default=str) + except Exception as e: + print(f"{Fore.YELLOW}⚠ 保存状态失败: {e}{Style.RESET_ALL}") + + def _load_state(self, wf_id: str) -> dict: + """加载工作流状态""" + path = os.path.join(self.state_dir, f"{wf_id}.json") + if not os.path.exists(path): + return {} + try: + with open(path, 'r', encoding='utf-8') as f: + return json.load(f) + except Exception: + return {} + + +# ── Liugin 插件接口 ── + +class Liugin: + """工作流插件""" + + def __init__(self): + self.engine = None + self.usage = """工作流工具使用方法： + +JSON格式示例： +{"action": "use_tool", "tool": "workflow", "args": "run deploy.yaml"} - 执行工作流 +{"action": "use_tool", "tool": "workflow", "args": "list"} - 列出工作流文件 +{"action": "use_tool", "tool": "workflow", "args": "states"} - 查看运行记录 +{"action": "use_tool", "tool": "workflow", "args": "status "} - 查看状态 +{"action": "use_tool", "tool": "workflow", "args": "resume "} - 恢复执行 +{"action": "use_tool", "tool": "workflow", "args": "abort "} - 中止工作流 + +工作流 YAML 格式示例： +name: deploy +description: 部署流程 +steps: + - name: lint + type: shell + command: "npm run lint" + - name: test + type: shell + command: "npm test" + continue_on_error: true + - name: build + type: shell + command: "npm run build" + if: "{{test.exit_code}} == 0" + - name: notify + type: echo + message: "构建完成: {{build.output}}" + +支持的步骤类型: + - tool: 调用插件工具 (tool, args) + - shell: 执行系统命令 (command, cwd, timeout) + - echo: 打印消息 (message) + - wait: 等待 (seconds) + - set: 设置变量 (variables) + - if: 条件分支 (condition, then, else) + +变量引用: + {{steps.step_name.output}} - 步骤输出 + {{steps.step_name.exit_code}} - 步骤退出码 + {{env.VAR_NAME}} - 环境变量 + {{args.param}} - 工作流参数""" + + def set_cli(self, cli): + self.cli = cli + self.engine = WorkflowEngine(cli) + cli.register_liugin_command('workflow', self.command_handler) + cli.register_liugin_command('wf', self.command_handler) + + def command_handler(self, args): + return self.handle(args) + + def get_tool_info(self): + return { + "name": "workflow", + "description": "工作流引擎 - YAML 定义的多步骤自动化流程，支持变量传递、条件分支、断点续跑", + "keywords": ["workflow", "工作流", "流程", "自动化", "pipeline", "deploy", "build", "run"], + "usage": self.usage + } + + def get_mcp_definition(self): + return { + "name": "workflow", + "description": "工作流引擎 - 执行、管理 YAML 定义的自动化工作流", + "inputSchema": { + "type": "object", + "properties": { + "operation": { + "type": "string", + "enum": ["run", "list", "states", "status", "resume", "abort"], + "description": "操作类型" + }, + "target": { + "type": "string", + "description": "工作流文件路径(run) 或工作流ID(status/resume/abort)" + }, + "args": { + "type": "object", + "description": "工作流参数 (run 时使用)" + } + }, + "required": ["operation"] + } + } + + def convert_mcp_args(self, arguments): + op = arguments.get("operation", "") + target = arguments.get("target", "") + args = arguments.get("args", "") + if args and isinstance(args, dict): + args_str = ' '.join(f'{k}={v}' for k, v in args.items()) + return f"{op} {target} {args_str}".strip() + return f"{op} {target}".strip() + + def handle(self, args): + try: + args = args.strip() + if not args: + return self.usage + + parts = args.split(None, 2) + action = parts[0].lower() + + if action == 'run': + if len(parts) < 2: + return "❌ 请指定工作流文件: workflow run " + path = parts[1] + # 解析内联参数 + wf_args = {} + if len(parts) > 2: + for part in parts[2].split(): + if '=' in part: + k, v = part.split('=', 1) + wf_args[k] = v + return self.engine.run(path, wf_args) + + elif action == 'list': + return self.engine.list_workflows() + + elif action == 'states': + return self.engine.list_states() + + elif action == 'status': + if len(parts) < 2: + return "❌ 请指定工作流 ID: workflow status " + return self.engine.status(parts[1]) + + elif action == 'resume': + if len(parts) < 2: + return "❌ 请指定工作流 ID: workflow resume " + return self.engine.resume(parts[1]) + + elif action == 'abort': + if len(parts) < 2: + return "❌ 请指定工作流 ID: workflow abort " + return self.engine.abort(parts[1]) + + else: + # 默认当作 run + if os.path.exists(action): + return self.engine.run(action) + return f"❌ 未知操作: {action}\n\n{self.usage}" + + except Exception as e: + return f"❌ 工作流操作错误: {e}" diff --git a/unified_tool_manager.py b/unified_tool_manager.py index 8afe08a..415a618 100644 --- a/unified_tool_manager.py +++ b/unified_tool_manager.py @@ -1,8 +1,6 @@ """ 统一工具管理器 - 协调 Liugin、Skill 和 MCP 三套协议 自动判断并路由到正确的处理器 - -集成插件内存管理: TTL 自动淘汰 + 懒加载 """ import os @@ -29,24 +27,9 @@ class UnifiedToolManager: - 回退到 Liugin 协议 - 支持强制指定协议 - MCP 格式调用统一入口 - - 内存管理: - - 空闲插件自动淘汰（TTL 机制） - - 调用时自动懒加载回来 - - 核心插件可钉住不淘汰 """ - # 内存管理默认配置 - 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): + def __init__(self, cli_instance=None): self.cli = cli_instance self._lock = threading.RLock() @@ -63,62 +46,39 @@ class UnifiedToolManager: # 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 shutdown(self) -> None: + """关闭管理器，清理所有插件实例""" + with self._lock: + for name, tool in self._tools.items(): + instance = tool.get('instance') + if instance and hasattr(instance, 'cleanup'): + try: + instance.cleanup() + except Exception: + pass + self._tools.clear() + self._protocol_map.clear() + self._mcp_tools_cache = None + 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): @@ -127,7 +87,6 @@ class UnifiedToolManager: plugin_name = filename[:-3] - # 检查是否启用 if enabled_plugins is not None and plugin_name.lower() not in enabled_plugins: continue @@ -138,7 +97,6 @@ class UnifiedToolManager: module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) - # 兼容 Plugin 和 Liugin 两种类名 plugin_class = None if hasattr(module, 'Liugin'): plugin_class = module.Liugin @@ -148,19 +106,16 @@ 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, 'description': tool_info.get('description', ''), @@ -173,67 +128,44 @@ class UnifiedToolManager: } 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}") - + def _load_skills(self, skills_dir: str) -> None: """加载 Skill 协议工具 (支持 Markdown 和 Python 两种格式)""" if not os.path.exists(skills_dir): print(f"{Fore.YELLOW}Skills 目录不存在: {skills_dir}{Style.RESET_ALL}") return - + try: from skills.base import SkillLoader, MarkdownSkill, Skill - + self._skill_loader = SkillLoader(skills_dir) - - # 先发现所有技能 + discovered = self._skill_loader.discover_skills(skills_dir) print(f"{Fore.CYAN}发现 {len(discovered)} 个技能: {[s['name'] for s in discovered]}{Style.RESET_ALL}") - - # 加载所有技能 + self._skill_loader.load_all_skills(skills_dir) - + loaded_count = len(self._skill_loader.skills) if loaded_count == 0: print(f"{Fore.YELLOW}未加载任何 Skill 技能{Style.RESET_ALL}") return - + for skill_name, skill in self._skill_loader.skills.items(): - # 设置 CLI 引用 (仅 Python 类技能) 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', {}) - - # 检查是否已有同名 Plugin，Skill 优先覆盖 + if skill_name in self._tools: old_protocol = self._protocol_map[skill_name] print(f"{Fore.YELLOW}工具 '{skill_name}' 已存在({old_protocol})，使用 Skill 协议覆盖{Style.RESET_ALL}") - - # 判断技能格式 + is_markdown = isinstance(skill, MarkdownSkill) skill_format = "markdown" if is_markdown else "python" - # 注册到统一工具表 self._tools[skill_name] = { 'name': skill_name, 'description': func_def.get('description', ''), @@ -245,48 +177,24 @@ class UnifiedToolManager: 'instance': skill, 'tool_definition': tool_def, 'anthropic_definition': skill.get_anthropic_tool_definition(), - 'mcp_definition': tool_def # Skill 的 OpenAI 格式就是 MCP 格式 + 'mcp_definition': tool_def } 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: print(f"{Fore.YELLOW}Skill 协议模块未找到: {e}{Style.RESET_ALL}") except Exception as e: print(f"{Fore.RED}加载 Skills 失败: {e}{Style.RESET_ALL}") - + def _create_skill_handler(self, skill) -> callable: - """ - 为 Skill 创建兼容 Plugin 的 handler - - 支持 Markdown 和 Python 两种技能格式 - """ + """为 Skill 创建兼容 Plugin 的 handler""" from skills.base import MarkdownSkill, Skill - + def handler(args: str) -> str: - import json - kwargs = {} - - # 解析参数 + args = args.strip() if args.startswith('{'): try: @@ -294,7 +202,6 @@ class UnifiedToolManager: except json.JSONDecodeError: kwargs = {'operation': args} elif '=' in args: - # key=value 格式: "directory=. file_pattern=*.py" for part in args.split(): if '=' in part: k, v = part.split('=', 1) @@ -303,203 +210,38 @@ class UnifiedToolManager: kwargs.setdefault('operation', part) else: kwargs = {'operation': args} - - # Markdown 技能: 返回指令供 AI 执行 + if isinstance(skill, MarkdownSkill): return f"[Skill: {skill.name}]\n{skill.instructions}" - - # Python 技能: 直接执行 + if hasattr(skill, 'run'): result = skill.run(**kwargs) if result.success: return result.message or str(result.data) else: return f"错误: {result.error}" - + 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]]: """获取启用的插件列表""" - # 检查环境变量 enabled_env = os.environ.get("XIAOLI_ENABLED_PLUGINS") if enabled_env is not None: if enabled_env == "": return [] return [p.lower() for p in enabled_env.split(",")] - - # 检查配置文件 + config_file = os.path.join(os.path.dirname(plugins_dir), "plugins_config.py") if os.path.exists(config_file): try: with open(config_file, 'r', encoding='utf-8') as f: content = f.read() - + if "# 不启用任何插件" in content: return [] - + enabled = [] for line in content.split('\n'): if line.startswith('ENABLE_') and '= True' in line: @@ -508,9 +250,9 @@ class UnifiedToolManager: return enabled except Exception: pass - - return None # 加载全部 - + + return None + def _build_mcp_cache(self) -> None: """构建 MCP 工具定义缓存（线程安全）""" with self._lock: @@ -518,56 +260,28 @@ class UnifiedToolManager: for tool in self._tools.values(): mcp_def = tool.get('mcp_definition') if mcp_def: - # 确保符合 MCP 标准 normalized = self._normalize_mcp_definition(mcp_def) self._mcp_tools_cache.append(normalized) else: - # 为没有 MCP 定义的 Plugin 生成默认定义 self._mcp_tools_cache.append(self._generate_mcp_definition(tool)) - + def _normalize_mcp_definition(self, mcp_def: Dict[str, Any]) -> Dict[str, Any]: - """ - 标准化 MCP 工具定义 - - 支持多种输入格式并转换为标准 MCP 格式 - """ - # 如果已经是 MCP 格式（有 name 或 inputSchema） + """标准化 MCP 工具定义""" if 'name' in mcp_def and ('inputSchema' in mcp_def or 'description' in mcp_def): return mcp_def - - # 如果是 OpenAI Function Calling 格式，转换 + if mcp_def.get('type') == 'function' and 'function' in mcp_def: func = mcp_def['function'] return { "name": func.get('name', ''), "description": func.get('description', ''), - "inputSchema": func.get('parameters', { - "type": "object", - "properties": {} - }) + "inputSchema": func.get('parameters', {"type": "object", "properties": {}}) } - - # 其他情况，直接返回 + return mcp_def - + def _generate_mcp_definition(self, tool: Dict[str, Any]) -> Dict[str, Any]: - """ - 为 Plugin 生成符合 MCP 标准的工具定义 - - MCP 官方格式 (2025-03-26): - { - "name": "tool_name", - "title": "Human Readable Name", // optional - "description": "Human-readable description", - "inputSchema": { - "type": "object", - "properties": {...}, - "required": [...] - }, - "outputSchema": {...}, // optional - "annotations": {...} // optional - } - """ + """为 Plugin 生成符合 MCP 标准的工具定义""" return { "name": tool['name'], "description": tool.get('description', ''), @@ -582,174 +296,114 @@ class UnifiedToolManager: "required": ["args"] } } - + # ==================== 公共 API ==================== - + @property 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._ensure_loaded(name) - + """获取指定工具""" + return self._tools.get(name) + def get_protocol(self, name: str) -> Optional[str]: """获取工具使用的协议""" return self._protocol_map.get(name) - + 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._ensure_loaded(tool_name) + """执行工具调用""" + tool = self._tools.get(tool_name) if not tool: return {"result": f"未找到工具: {tool_name}"} - + try: handler = tool['handler'] protocol = tool['protocol'] instance = tool.get('instance') - + if protocol == 'skill' and kwargs: - # Skill 协议，使用 kwargs skill = tool['instance'] result = skill.run(**kwargs) return {"result": result.message or str(result.data) if result.success else f"错误: {result.error}"} - + elif protocol == 'liugin' and kwargs and not args: - # Liugin 协议但有 MCP 结构化参数，需要转换 - # 检查插件是否有 convert_mcp_args 方法 if instance and hasattr(instance, 'convert_mcp_args'): args = instance.convert_mcp_args(kwargs) else: args = self._convert_mcp_args_to_plugin_args(tool_name, kwargs, instance) - - # 执行 handler + result = handler(args) return {"result": result if result is not None else "无结果"} - + except Exception as e: return {"result": f"工具执行错误: {e}"} - + # ==================== MCP 协议支持 ==================== - + def get_mcp_tools(self) -> List[Dict[str, Any]]: - """ - 获取 MCP 格式的工具定义列表 - - MCP (Model Context Protocol) 使用 OpenAI Function Calling 格式 - 作为标准工具定义格式 - - Returns: - List of MCP tool definitions - """ + """获取 MCP 格式的工具定义列表""" with self._lock: if self._mcp_tools_cache is None: self._build_mcp_cache() - return self._mcp_tools_cache.copy() # 返回副本，避免外部修改 - + return self._mcp_tools_cache.copy() + def mcp_call(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: - """ - MCP 格式的工具调用入口（支持懒加载） + """MCP 格式的工具调用入口""" + tool = self._tools.get(tool_name) - 这是 MCP 协议的主要调用接口，接收结构化参数并返回结果 - - Args: - tool_name: 工具名称 - arguments: 结构化参数 (JSON对象) - - Returns: - MCP 标准响应格式: - { - "content": [ - { - "type": "text", - "text": "结果内容" - } - ], - "isError": false - } - """ - tool = self._ensure_loaded(tool_name) - if not tool: return self._mcp_error(f"未找到工具: {tool_name}") - + try: protocol = tool['protocol'] instance = tool.get('instance') - + if protocol == 'liugin': - # Liugin 协议: 将 arguments 转换为 args 字符串 args = self._convert_mcp_args_to_plugin_args(tool_name, arguments, instance) result = tool['handler'](args) return self._mcp_success(result if result else "执行完成") - + elif protocol == 'skill': - # Skill 协议: 直接传递 arguments 作为 kwargs from skills.base import MarkdownSkill - + if isinstance(instance, MarkdownSkill): return self._mcp_success(f"[Skill: {instance.name}]\n{instance.instructions}") - + if hasattr(instance, 'run'): result = instance.run(**arguments) if result.success: return self._mcp_success(result.message or str(result.data)) else: return self._mcp_error(f"执行失败: {result.error}") - + return self._mcp_error(f"未知协议类型: {protocol}") - + except Exception as e: return self._mcp_error(f"工具执行错误: {str(e)}") - - def _convert_mcp_args_to_plugin_args(self, tool_name: str, arguments: Dict[str, Any], + + def _convert_mcp_args_to_plugin_args(self, tool_name: str, arguments: Dict[str, Any], instance: Any) -> str: - """ - 将 MCP 结构化参数转换为 Plugin 的 args 字符串 - - Args: - tool_name: 工具名称 - arguments: MCP 参数对象 - instance: 插件实例（可能有自定义转换方法） - - Returns: - args 字符串 - """ - # 检查插件是否有自定义的 MCP 参数转换方法 + """将 MCP 结构化参数转换为 Plugin 的 args 字符串""" if instance and hasattr(instance, 'convert_mcp_args'): return instance.convert_mcp_args(arguments) - - # 默认转换策略 + if 'args' in arguments: - # 如果有 args 字段，直接使用 return arguments['args'] - - # 将 arguments 转换为命令行格式 + parts = [] - - # 第一个参数通常是操作/命令 + if 'operation' in arguments: parts.append(arguments['operation']) elif 'action' in arguments: parts.append(arguments['action']) - - # 添加其他参数 + for key, value in arguments.items(): if key in ('operation', 'action'): continue - + if isinstance(value, bool): if value: parts.append(f"-{key}") @@ -760,55 +414,24 @@ class UnifiedToolManager: else: parts.append(f"--{key}") parts.append(self._quote_if_needed(str(value))) - + return ' '.join(parts) - + def _quote_if_needed(self, s: str) -> str: - """如果字符串包含空格，添加引号""" if ' ' in s or '"' in s: return f'"{s}"' return s - + def _mcp_success(self, text: str) -> Dict[str, Any]: - """生成 MCP 成功响应""" - return { - "content": [ - { - "type": "text", - "text": text - } - ], - "isError": False - } - + return {"content": [{"type": "text", "text": text}], "isError": False} + def _mcp_error(self, error_msg: str) -> Dict[str, Any]: - """生成 MCP 错误响应""" - return { - "content": [ - { - "type": "text", - "text": error_msg - } - ], - "isError": True - } - + return {"content": [{"type": "text", "text": error_msg}], "isError": True} + # ==================== 兼容性 API ==================== - + def get_openai_tools(self) -> List[Dict[str, Any]]: - """ - 获取 OpenAI Function Calling 格式的工具定义 - - OpenAI 格式: - { - "type": "function", - "function": { - "name": "...", - "description": "...", - "parameters": {...} - } - } - """ + """获取 OpenAI Function Calling 格式的工具定义""" tools = [] for mcp_def in self.get_mcp_tools(): tools.append({ @@ -816,54 +439,28 @@ class UnifiedToolManager: "function": { "name": mcp_def.get("name", ""), "description": mcp_def.get("description", ""), - "parameters": mcp_def.get("inputSchema", { - "type": "object", - "properties": {} - }) + "parameters": mcp_def.get("inputSchema", {"type": "object", "properties": {}}) } }) return tools - + def get_anthropic_tools(self) -> List[Dict[str, Any]]: - """ - 获取 Anthropic Tool Use 格式的工具定义 - - Anthropic 格式: - { - "name": "tool_name", - "description": "...", - "input_schema": {...} - } - """ + """获取 Anthropic Tool Use 格式的工具定义""" tools = [] for mcp_def in self.get_mcp_tools(): tools.append({ "name": mcp_def.get("name", ""), "description": mcp_def.get("description", ""), - "input_schema": mcp_def.get("inputSchema", { - "type": "object", - "properties": {} - }) + "input_schema": mcp_def.get("inputSchema", {"type": "object", "properties": {}}) }) return tools - + def search_tools(self, query: str, limit: int = 5) -> List[Dict[str, Any]]: - """ - 搜索工具（包括已卸载的插件，使用缓存信息） - - Args: - query: 搜索关键词 - limit: 返回数量限制 - - Returns: - 匹配的工具列表 - """ + """搜索工具""" query_lower = query.lower() - # 标准化：移除所有分隔符（连字符、下划线、空格） query_normalized = query_lower.replace('-', '').replace('_', '').replace(' ', '') matches = [] - # 搜索已加载的工具 for tool in self._tools.values(): score = self._score_tool_match(tool, query_lower, query_normalized) if score > 0: @@ -877,27 +474,6 @@ class UnifiedToolManager: '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] @@ -909,35 +485,22 @@ class UnifiedToolManager: 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 = "", + + def register_tool(self, name: str, handler: callable, description: str = "", protocol: str = "custom", mcp_definition: Dict = None, **kwargs) -> None: - """ - 动态注册工具 - - Args: - name: 工具名称 - handler: 处理函数 - description: 描述 - protocol: 协议类型 - mcp_definition: MCP 工具定义（可选） - """ + """动态注册工具""" with self._lock: self._tools[name] = { 'name': name, @@ -948,115 +511,77 @@ class UnifiedToolManager: **kwargs } self._protocol_map[name] = protocol - - # 更新缓存 + if self._mcp_tools_cache is not None: if mcp_definition: self._mcp_tools_cache.append(mcp_definition) else: self._mcp_tools_cache.append(self._generate_mcp_definition(self._tools[name])) - + def unregister_tool(self, name: str) -> bool: """注销工具""" with self._lock: if name in self._tools: + instance = self._tools[name].get('instance') + if instance and hasattr(instance, 'cleanup'): + try: + instance.cleanup() + except Exception: + pass 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) + """根据名称获取工具（兼容 LiuginManager 接口）""" + return self._tools.get(name) - @property - def memory_manager(self): - """获取内存管理器实例""" - return self._memory_mgr - def get_skills_system_prompt(self) -> str: - """ - 获取所有技能的系统提示词 - - 用于注入到 AI 的 system prompt 中， - 让 AI 了解如何使用这些技能 - """ + """获取所有技能的系统提示词""" skill_prompts = [] - + for name, tool in self._tools.items(): if tool['protocol'] == 'skill': instance = tool.get('instance') if instance: - # 获取技能指令 instructions = getattr(instance, 'instructions', '') scripts_dir = getattr(instance, 'scripts_dir', '') - + prompt_parts = [f"## 技能: {name}"] prompt_parts.append(f"描述: {tool['description']}") - + if instructions: - # 截取关键部分，避免过长 prompt_parts.append(f"\n{instructions}") - + if scripts_dir: prompt_parts.append(f"\n脚本目录: {scripts_dir}") - + skill_prompts.append('\n'.join(prompt_parts)) - + if skill_prompts: header = "# 已加载的 Agent Skills\n以下技能可用于执行特定任务，调用方式与工具相同：\n" return header + '\n\n---\n\n'.join(skill_prompts) - + return "" - + def get_tools_prompt(self) -> str: - """ - 获取工具使用提示词（兼容旧接口） - - 包含 Plugin 和 Skill 两种协议的工具说明 - """ + """获取工具使用提示词（兼容旧接口）""" prompts = [] - - # Liugin 协议工具 + plugin_tools = [t for t in self._tools.values() if t['protocol'] == 'liugin'] if plugin_tools: prompts.append("## Liugin 工具") @@ -1064,67 +589,37 @@ class UnifiedToolManager: usage = tool.get('usage', '') if usage: prompts.append(f"### {tool['name']}\n{usage}\n") - - # Skill 协议工具（使用系统提示词） + skills_prompt = self.get_skills_system_prompt() if skills_prompt: prompts.append(skills_prompt) - + return '\n\n'.join(prompts) - + # ==================== MCP Server 模式支持 ==================== - + def get_mcp_server_info(self) -> Dict[str, Any]: - """ - 获取 MCP Server 信息 - - 用于 MCP 客户端连接时的初始化响应 - """ + """获取 MCP Server 信息""" return { "name": "xiaoli-tool-server", "version": "1.0.0", "protocolVersion": "2024-11-05", - "capabilities": { - "tools": { - "listChanged": True - } - } + "capabilities": {"tools": {"listChanged": True}} } - + def handle_mcp_request(self, request: Dict[str, Any]) -> Dict[str, Any]: - """ - 处理 MCP 协议请求 - - 统一处理 MCP 客户端的各种请求 - - Args: - request: MCP 请求对象 - { - "jsonrpc": "2.0", - "id": 1, - "method": "tools/list" | "tools/call" | ..., - "params": {...} - } - - Returns: - MCP 响应对象 - { - "jsonrpc": "2.0", - "id": 1, - "result": {...} - } - """ + """处理 MCP 协议请求""" method = request.get('method', '') params = request.get('params', {}) req_id = request.get('id') - + result = None error = None - + try: if method == 'tools/list': result = {"tools": self.get_mcp_tools()} - + elif method == 'tools/call': tool_name = params.get('name') arguments = params.get('arguments', {}) @@ -1132,23 +627,23 @@ class UnifiedToolManager: error = {"code": -32602, "message": "Missing tool name"} else: result = self.mcp_call(tool_name, arguments) - + elif method == 'initialize': result = self.get_mcp_server_info() - + elif method == 'ping': result = {} - + else: error = {"code": -32601, "message": f"Method not found: {method}"} - + except Exception as e: error = {"code": -32603, "message": str(e)} - + response = {"jsonrpc": "2.0", "id": req_id} if error: response["error"] = error else: response["result"] = result - - return response \ No newline at end of file + + return response diff --git a/workflows/example.yaml b/workflows/example.yaml new file mode 100644 index 0000000..928cc8b --- /dev/null +++ b/workflows/example.yaml @@ -0,0 +1,31 @@ +# 示例工作流 - 代码质量检查与构建 +name: code-quality +description: 代码质量检查、测试、构建的自动化流程 + +steps: + - name: info + type: echo + message: "🚀 开始代码质量检查流程" + + - name: lint + type: shell + command: "echo '✓ 代码风格检查通过'" + continue_on_error: true + + - name: test + type: shell + command: "echo '✓ 单元测试通过 (42/42)'" + continue_on_error: true + + - name: check_results + type: echo + message: "Lint: {{lint.output}} | Test: {{test.output}}" + + - name: build + type: shell + command: "echo '✓ 构建完成 → dist/'" + if: "{{lint.exit_code}} == 0" + + - name: done + type: echo + message: "✅ 流程完成，构建产物: {{build.output}}" diff --git a/xcli_core/plugin_memory_manager.py b/xcli_core/plugin_memory_manager.py deleted file mode 100644 index 3b28314..0000000 --- a/xcli_core/plugin_memory_manager.py +++ /dev/null @@ -1,355 +0,0 @@ -""" -插件内存管理器 - 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 adda0bf..9327431 100644 --- a/xcli_core/tui.py +++ b/xcli_core/tui.py @@ -562,13 +562,9 @@ 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")) + # 工具数 + if hasattr(self.cli, 'liugin_manager'): + status_container.mount(Static(f" 工具: {len(self.cli.liugin_manager.tools)} 个", classes="tool-item")) def _update_status(self, text: str): bar = self.query_one("#status-bar") @@ -822,10 +818,7 @@ if TEXTUAL_AVAILABLE: 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 生成 ──