shuiyuPyLab/xiaoli-cli
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

feat: 工作模式系统 - 整合安全模式 + 计划模式

Browse Source 
工作模式 (/mode 或 /safe 切换):
- 普通: AI识别风险，有风险才确认 (原安全模式)
- 人工确认: 所有指令都需确认 (原安全模式)
- 无限制: 直接执行 (原安全模式)
- 计划模式: 全自动 + 审查 + 重试 + 引擎切换 (新增)

计划模式行为:
- 全自动执行，不询问用户
- 引擎错误10次自动切换到其他引擎
- 无其他引擎时继续重试，但不把引擎错误加入上下文
- 任务完成后自动审查结果
- 审查发现问题自动重试

实现:
- safety.py: SafetyLayer → WorkMode，新增 MODE_PLAN
- cli_core.py: 新增 _plan_generate / _plan_review
- /safe 和 /mode 命令均可切换
This commit is contained in:
xiaoli-cli bot committed 2026-06-23 10:11:06 +00:00
1 parent 3f7df28423
commit 810a057d25
2 files changed
+238 -100
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

xcli_core/cli_core.py
+97 -9
View File
@@ -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'):
xcli_core/safety.py
+141 -91
View File
@@ -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