diff --git a/xcli_core/cli_core.py b/xcli_core/cli_core.py index 36a33d8..2e15fd3 100644 --- a/xcli_core/cli_core.py +++ b/xcli_core/cli_core.py @@ -12,7 +12,7 @@ from colorama import Fore, Style from .constants import TEXTUAL_AVAILABLE, LOVE_FILE_PATH, DEFAULT_MAX_HISTORY from .config import get_system_config, logger from .cli_base import BaseAICLI -from .safety import get_safety, MODE_UNRESTRICTED, MODE_NORMAL, MODE_MANUAL +from .safety import get_safety, MODE_UNRESTRICTED, MODE_NORMAL, MODE_MANUAL, MODE_PLAN from .cli_clawli import ClawliMixin from .cli_tools import ToolMixin from .cli_code_exec import CodeExecMixin @@ -526,7 +526,16 @@ multi 操作支持一次修改多处： max_loops = 20 while loop_count < max_loops: - response = self._generate_response_with_animation(current_input, liugin_prompts=liugin_prompts) + # ── 计划模式：带引擎错误处理的响应生成 ── + safety = get_safety() + if safety.is_plan_mode(): + response = self._plan_generate(current_input, liugin_prompts) + if response is None: + # 所有引擎都失败，退出循环 + self._display_response("[计划模式] 所有引擎均不可用，任务中断。") + break + else: + response = self._generate_response_with_animation(current_input, liugin_prompts=liugin_prompts) processed_response = self.process_thinking_response(response) processed_response = self._process_code_blocks(processed_response) @@ -865,6 +874,83 @@ multi 操作支持一次修改多处： return f"未找到工具: {tool_name}" + # ── 计划模式：带重试和引擎切换的响应生成 ── + + def _plan_generate(self, current_input: str, liugin_prompts=None) -> Optional[str]: + """ + 计划模式专用响应生成。 + - 自动重试 + - 引擎错误 10 次自动切换 + - 无其他引擎时继续重试但不污染上下文 + """ + safety = get_safety() + max_retries = 5 + + for attempt in range(max_retries): + try: + response = self._generate_response_with_animation(current_input, liugin_prompts=liugin_prompts) + + # 检查是否是引擎错误（空响应或明确错误） + if response and not response.startswith("错误") and not response.startswith("API 调用失败"): + safety.reset_engine_errors() + return response + + # 引擎返回错误 + is_fatal = safety.record_engine_error() + + if is_fatal: + new_engine = safety.try_switch_engine() + if new_engine: + print(f"[计划模式] 引擎连续错误 {safety.plan_engine_error_limit} 次，已切换到 {new_engine}") + continue # 用新引擎重试 + else: + # 无其他引擎，继续重试但不把错误加入上下文 + print(f"[计划模式] 无其他引擎可用，继续重试...") + time.sleep(2) + continue + + # 未达阈值，重试 + continue + + except Exception as e: + is_fatal = safety.record_engine_error() + if is_fatal: + new_engine = safety.try_switch_engine() + if new_engine: + print(f"[计划模式] 引擎异常，已切换到 {new_engine}") + continue + continue + + return "[计划模式] 重试次数已用完。" + + def _plan_review(self, task_description: str, task_result: str): + """ + 计划模式：任务完成后自动审查。 + """ + safety = get_safety() + review_prompt = safety.get_review_prompt(task_result) + + print(f" [计划模式] 自动审查中...") + + try: + review = self._generate_response_with_animation(review_prompt) + if review: + # 检查审查结果是否发现问题 + review_lower = review.lower() + has_issue = any(w in review_lower for w in [ + '问题', '错误', '失败', '遗漏', '修正', '重新', 'issue', 'error', 'fail', + 'bug', 'wrong', 'missing', 'fix', 'retry' + ]) + if has_issue: + print(f" [计划模式] 审查发现问题，自动重试...") + return True # 需要重试 + else: + print(f" [计划模式] 审查通过。") + return False + except Exception: + pass + return False + def _generate_response_with_animation(self, current_input, liugin_prompts=None): """生成AI响应并显示等待动画 — 支持 ESC 跨平台取消""" # TUI 模式：不做动画 @@ -1041,7 +1127,7 @@ multi 操作支持一次修改多处： print(f"{Fore.GREEN}输入 '/tui' 切换到 TUI 模式{Style.RESET_ALL}") print(f"{Fore.GREEN}输入 '/engine list' 查看可用AI引擎{Style.RESET_ALL}") print(f"{Fore.GREEN}输入 '/engine switch <引擎名>' 切换AI引擎{Style.RESET_ALL}") - print(f"{Fore.GREEN}输入 '/safe' 切换安全模式 (普通→人工→无限制){Style.RESET_ALL}") + print(f"{Fore.GREEN}输入 '/safe' 或 '/mode' 切换工作模式 (普通→人工→无限制→计划){Style.RESET_ALL}") print(f"{Fore.GREEN}输入 '/notify' 切换任务完成通知 (开/关){Style.RESET_ALL}") print(f"{Fore.GREEN}输入 '/scheduler' 或 '/remind' 管理定时任务{Style.RESET_ALL}") print(f"{Fore.GREEN}输入 '/memory' 管理记忆系统 (日记/搜索/聊天记录){Style.RESET_ALL}") @@ -1253,7 +1339,7 @@ multi 操作支持一次修改多处： print(f"{Fore.CYAN}已从 TUI 模式返回 CLI 模式{Style.RESET_ALL}") continue - if user_input.startswith('/safe'): + if user_input.startswith('/safe') or user_input.startswith('/mode'): parts = user_input.split() safety = get_safety() if len(parts) > 1: @@ -1264,16 +1350,18 @@ multi 操作支持一次修改多处： safety.set_mode(MODE_NORMAL) elif mode_arg in ('manual', '人工', 'all'): safety.set_mode(MODE_MANUAL) + elif mode_arg in ('plan', '计划', 'auto'): + safety.set_mode(MODE_PLAN) else: - print(f"{Fore.RED}用法: /safe [off|on|manual]{Style.RESET_ALL}") + print(f"{Fore.RED}用法: /mode [off|on|manual|plan]{Style.RESET_ALL}") continue else: - # 无参数：循环切换 safety.cycle_mode() mode_name = safety.get_mode_name() - icons = {"无限制": "", "普通": "", "人工确认": ""} - icon = icons.get(mode_name, "") - print(f"{icon} 安全模式: {mode_name}") + icon = safety.get_mode_icon() + print(f"{icon} 工作模式: {mode_name}") + if mode_name == "计划模式": + print(f" {Fore.CYAN}全自动执行 | 任务审查 | 自动重试 | 引擎故障切换{Style.RESET_ALL}") continue if user_input.startswith('/notify'): diff --git a/xcli_core/safety.py b/xcli_core/safety.py index 8d35a83..04feb7c 100644 --- a/xcli_core/safety.py +++ b/xcli_core/safety.py @@ -1,32 +1,50 @@ """ -安全层 - 统一的指令安全检查 -拦截所有工具调用，通过 AI 识别风险并请求用户确认 +工作模式 - 统一的安全检查 + 工作流控制 +整合安全模式与计划模式，控制 AI 的执行策略 """ import json import threading from typing import Optional, Tuple -# ── 安全模式 ── +# ── 工作模式 ── MODE_UNRESTRICTED = 0 # 无限制：直接执行 MODE_NORMAL = 1 # 普通：AI 识别风险 -MODE_MANUAL = 2 # 人工：所有指令都确认 +MODE_MANUAL = 2 # 人工：所有指令确认 +MODE_PLAN = 3 # 计划：全自动 + 审查 + 重试 + 引擎切换 MODE_NAMES = { MODE_UNRESTRICTED: "无限制", MODE_NORMAL: "普通", MODE_MANUAL: "人工确认", + MODE_PLAN: "计划模式", } -class SafetyLayer: - """统一安全检查层""" +MODE_ICONS = { + MODE_UNRESTRICTED: "", + MODE_NORMAL: "", + MODE_MANUAL: "", + MODE_PLAN: "", +} + + +class WorkMode: + """统一工作模式控制""" def __init__(self): self.mode = MODE_NORMAL self.cli = None - self._analysis_cache = {} # 缓存分析结果 + self._analysis_cache = {} self._lock = threading.Lock() + # ── 计划模式状态 ── + self.plan_engine_errors = 0 # 当前引擎连续错误次数 + self.plan_engine_error_limit = 10 # 触发引擎切换的错误阈值 + self.plan_retry_count = 0 # 当前重试次数 + self.plan_max_retries = 3 # 单步最大重试 + self.plan_review_pending = False # 是否有待审查的任务 + self.plan_last_task = "" # 最近一次任务描述 + def set_cli(self, cli): self.cli = cli @@ -36,26 +54,103 @@ class SafetyLayer: def get_mode_name(self) -> str: return MODE_NAMES.get(self.mode, "未知") + def get_mode_icon(self) -> str: + return MODE_ICONS.get(self.mode, "") + def set_mode(self, mode: int): + old = self.mode self.mode = mode + # 切换模式时重置计划模式状态 + if mode != old: + self.plan_engine_errors = 0 + self.plan_retry_count = 0 + self.plan_review_pending = False def cycle_mode(self) -> str: - """切换到下一个模式，返回新模式名称""" - self.mode = (self.mode + 1) % 3 + """循环切换: 普通 → 人工 → 无限制 → 计划 → 普通""" + self.mode = (self.mode + 1) % 4 return self.get_mode_name() - # ── 核心：拦截检查 ── + def is_plan_mode(self) -> bool: + return self.mode == MODE_PLAN + + # ── 计划模式：引擎错误管理 ── + + def record_engine_error(self) -> bool: + """ + 记录一次引擎错误。 + 返回 True 表示应该切换引擎。 + """ + self.plan_engine_errors += 1 + return self.plan_engine_errors >= self.plan_engine_error_limit + + def reset_engine_errors(self): + """引擎成功响应后重置错误计数""" + self.plan_engine_errors = 0 + + def should_switch_engine(self) -> bool: + return self.plan_engine_errors >= self.plan_engine_error_limit + + def get_available_engines(self) -> list: + """获取当前可用的其他引擎列表""" + if not self.cli: + return [] + engines = [] + for name, engine in getattr(self.cli, 'engines', {}).items(): + if engine != self.cli.current_engine: + engines.append((name, engine)) + return engines + + def try_switch_engine(self) -> Optional[str]: + """ + 尝试切换到另一个可用引擎。 + 返回新引擎名称，或 None 表示无其他引擎。 + """ + available = self.get_available_engines() + if not available: + return None + name, engine = available[0] + old_name = getattr(self.cli.current_engine, 'name', '?') + self.cli.current_engine = engine + self.plan_engine_errors = 0 + return name + + # ── 计划模式：任务审查 ── + + def request_review(self, task_description: str): + """标记需要审查的任务""" + self.plan_review_pending = True + self.plan_last_task = task_description + + def get_review_prompt(self, task_result: str) -> str: + """生成审查提示词""" + return ( + f"你刚才完成了以下任务：\n" + f"任务：{self.plan_last_task}\n" + f"结果：{task_result[:500]}\n\n" + f"请审查执行结果：\n" + f"1. 任务是否完整完成？\n" + f"2. 结果是否正确？\n" + f"3. 是否有遗漏或需要修正的地方？\n\n" + f"如果发现问题，请说明并重新执行。如果没问题，请确认完成。" + ) + + # ── 核心：安全检查 ── def check(self, tool_name: str, tool_args: str) -> Tuple[bool, str]: """ - 检查工具调用是否安全 - 返回: (允许执行, 消息) + 检查工具调用是否安全。 + 计划模式下自动放行（全自动）。 """ + # 计划模式：自动放行，不问用户 + if self.mode == MODE_PLAN: + return True, "" + # 无限制模式：直接放行 if self.mode == MODE_UNRESTRICTED: return True, "" - # 人工确认模式：所有指令都确认 + # 人工确认模式：所有指令确认 if self.mode == MODE_MANUAL: return self._ask_user_confirm( tool_name, tool_args, @@ -70,26 +165,20 @@ class SafetyLayer: # ── AI 风险识别 ── def _ai_analyze(self, tool_name: str, tool_args: str) -> Tuple[bool, str]: - """使用当前引擎独立分析指令风险（不带历史对话）""" - # 快速预检：明显安全的指令直接放行 + """使用当前引擎独立分析指令风险""" if self._is_quick_safe(tool_name, tool_args): return True, "" - # 获取引擎 engine = None if self.cli and hasattr(self.cli, 'current_engine'): engine = self.cli.current_engine if not engine: - # 无引擎时降级为人工确认 return self._ask_user_confirm( tool_name, tool_args, - risk_level="未知", - impact="无法进行 AI 风险分析", - reason="无可用引擎" + risk_level="未知", impact="无法进行 AI 风险分析", reason="无可用引擎" ) - # 检查缓存 cache_key = f"{tool_name}:{tool_args[:100]}" with self._lock: if cache_key in self._analysis_cache: @@ -98,12 +187,9 @@ class SafetyLayer: return True, "" return self._ask_user_confirm( tool_name, tool_args, - risk_level=cached["level"], - impact=cached["impact"], - reason=cached["reason"] + risk_level=cached["level"], impact=cached["impact"], reason=cached["reason"] ) - # 调用 AI 独立分析（不带任何历史对话） analysis_prompt = f"""你是一个安全分析器。分析以下工具调用是否存在安全风险。 工具: {tool_name} @@ -114,85 +200,58 @@ class SafetyLayer: 判断标准: - safe=true: 读取操作、查看信息、无害查询 -- safe=false: 删除文件、执行系统命令、修改配置、网络请求、git 危险操作 -- level 低: 只读操作但可能泄露信息 -- level 中: 修改操作可撤销 -- level 高: 不可逆操作、系统级操作""" +- safe=false: 删除文件、执行系统命令、修改配置、网络请求、git 危险操作""" try: - response = engine.generate_response(analysis_prompt, system_prompt="你是安全分析器。只返回 JSON，不要其他内容。") - + response = engine.generate_response( + analysis_prompt, + system_prompt="你是安全分析器。只返回 JSON，不要其他内容。" + ) if not response: return self._ask_user_confirm( tool_name, tool_args, - risk_level="未知", - impact="AI 分析无响应", - reason="引擎未返回结果" + risk_level="未知", impact="AI 分析无响应", reason="引擎未返回结果" ) - # 解析 JSON result = self._parse_analysis(response) - if result and result.get("safe") is True: - # 缓存安全结果 with self._lock: self._analysis_cache[cache_key] = result return True, "" - # 不安全，请求用户确认 level = result.get("level", "中") if result else "未知" reason = result.get("reason", "AI 识别到潜在风险") if result else "AI 分析失败" impact = result.get("impact", "未知影响") if result else "未知影响" - return self._ask_user_confirm( - tool_name, tool_args, - risk_level=level, - impact=impact, - reason=reason + tool_name, tool_args, risk_level=level, impact=impact, reason=reason ) - except Exception as e: return self._ask_user_confirm( tool_name, tool_args, - risk_level="未知", - impact=f"AI 分析异常: {e}", - reason="分析过程出错" + risk_level="未知", impact=f"AI 分析异常: {e}", reason="分析过程出错" ) def _parse_analysis(self, response: str) -> Optional[dict]: - """解析 AI 返回的分析结果""" import re - - # 尝试直接解析 JSON response = response.strip() - - # 去掉 markdown 代码块 if response.startswith("```"): response = response.split("\n", 1)[-1] if "\n" in response else response[3:] if response.endswith("```"): response = response[:-3] response = response.strip() - try: return json.loads(response) except json.JSONDecodeError: - pass - - # 尝试从文本中提取 JSON - match = re.search(r'\{[^{}]*"safe"[^{}]*\}', response) - if match: - try: - return json.loads(match.group()) - except json.JSONDecodeError: - pass - + match = re.search(r'\{[^{}]*"safe"[^{}]*\}', response) + if match: + try: + return json.loads(match.group()) + except json.JSONDecodeError: + pass return None def _is_quick_safe(self, tool_name: str, tool_args: str) -> bool: - """快速预检：明显安全的操作直接放行""" args = tool_args.strip().lower() - - # 纯读取操作 safe_patterns = { "code_editor": ["read_range", "find ", "regex ", "symbols", "imports", "ast_info", "stats", "structure", "context", "todo", @@ -200,7 +259,7 @@ class SafetyLayer: "git_tools": ["status", "log", "diff", "branch", "remote", "show", "blame", "contributors", "summary"], "file_manager": ["list", "read", "info", "search"], - "cmd_executor": [], # 命令执行不跳过 + "cmd_executor": [], "auto_engineer": ["metrics", "security", "complexity", "duplicates", "suggest", "info"], "task_manager": ["list"], @@ -210,35 +269,23 @@ class SafetyLayer: "network_tools": ["ping", "status", "headers", "ip"], "tool_search": [], } - patterns = safe_patterns.get(tool_name, []) - for p in patterns: - if args.startswith(p): - return True - - return False + return any(args.startswith(p) for p in patterns) # ── 用户交互 ── def _ask_user_confirm(self, tool_name: str, tool_args: str, risk_level: str = "中", impact: str = "", reason: str = "") -> Tuple[bool, str]: - """向用户展示风险信息并请求确认""" from colorama import Fore, Style - # 风险等级颜色 level_colors = { - "低": Fore.YELLOW, - "中": Fore.LIGHTYELLOW_EX, - "高": Fore.RED, - "未知": Fore.LIGHTBLACK_EX, + "低": Fore.YELLOW, "中": Fore.LIGHTYELLOW_EX, + "高": Fore.RED, "未知": Fore.LIGHTBLACK_EX, } level_color = level_colors.get(risk_level, Fore.YELLOW) - # 截断过长的参数 - display_args = tool_args - if len(display_args) > 200: - display_args = display_args[:200] + "..." + display_args = tool_args[:200] + "..." if len(tool_args) > 200 else tool_args print() print(f" {Fore.RED}{'═' * 50}{Style.RESET_ALL}") @@ -255,17 +302,20 @@ class SafetyLayer: try: choice = input(f" {Fore.WHITE}是否执行? (y/n): {Style.RESET_ALL}").strip().lower() - if choice in ('y', 'yes', '是'): - return True, "用户确认执行" - else: - return False, "用户取消执行" + return (choice in ('y', 'yes', '是'), "用户确认执行" if choice in ('y', 'yes', '是') else "用户取消执行") except (KeyboardInterrupt, EOFError): print() return False, "用户中断" -# ── 全局单例 ── -_safety = SafetyLayer() +# ── 兼容旧接口 ── +MODE_UNRESTRICTED = MODE_UNRESTRICTED +MODE_NORMAL = MODE_NORMAL +MODE_MANUAL = MODE_MANUAL -def get_safety() -> SafetyLayer: - return _safety +# ── 全局单例 ── +_work_mode = WorkMode() + +def get_safety() -> WorkMode: + """兼容旧接口：返回 WorkMode 实例""" + return _work_mode