commit 4b3286f5460e8a106c5558e3f0ae4be724d0c153 Author: Mimikko-zeus Date: Wed Jan 7 00:17:46 2026 +0800 Initial commit diff --git a/.env b/.env new file mode 100644 index 0000000..1fbe5b8 --- /dev/null +++ b/.env @@ -0,0 +1,4 @@ +LLM_API_URL=https://api.siliconflow.cn/v1/chat/completions +LLM_API_KEY=sk-fxsxbgatrjjhsnjpkdfgfngukqoqqgitjpxfqfxifcipaqpc +INTENT_MODEL_NAME=Qwen/Qwen2.5-7B-Instruct +GENERATION_MODEL_NAME=Qwen/Qwen2.5-72B-Instruct \ No newline at end of file diff --git a/.env.example b/.env.example new file mode 100644 index 0000000..7b4d2ff --- /dev/null +++ b/.env.example @@ -0,0 +1,4 @@ +LLM_API_URL=https://api.siliconflow.cn/v1/chat/completions +LLM_API_KEY=sk-fxsxbgatrjjhsnjpkdfgfngukqoqqgitjpxfqfxifcipaqpc +INTENT_MODEL_NAME=Qwen/Qwen2.5-7B-Instruct +GENERATION_MODEL_NAME=Qwen/Qwen2.5-72B-Instruct \ No newline at end of file diff --git a/PRD.md b/PRD.md new file mode 100644 index 0000000..389e067 --- /dev/null +++ b/PRD.md @@ -0,0 +1,233 @@ +==================== +【产品目标】 +==================== +- 面向 Windows 小白用户：一句话输入 +- 自动判断任务类型： + - chat：普通对话（如“今天天气怎么样”） + - execution：本地执行任务（文件处理） +- chat： + - 直接调用 LLM 返回文本 +- execution： + - 生成执行计划 + - 生成 Python 代码 + - 安全校验 + - 用户确认 + - 一次性子进程执行 +- 强制工作区副本目录： + workspace/input + workspace/output + workspace/logs +- MVP 明确不做： + - 联网任务（搜索 / 爬取） + - 鼠标 / 键盘自动化 + - 后台常驻 + - 多任务并行 +- 核心安全原则： + - LLM 可以联网“思考” + - Executor（执行器）禁止联网“动手” + +==================== +【项目结构（必须严格按此生成）】 +==================== +LocalAgent/ + main.py + requirements.txt + .env.example + ui/ + chat_view.py + task_guide_view.py + llm/ + client.py + prompts.py + intent/ + classifier.py + labels.py + safety/ + rule_checker.py + llm_reviewer.py + executor/ + sandbox_runner.py + workspace/ + input/ + output/ + logs/ + +==================== +【统一 LLM 调用规则（非常重要）】 +==================== +- 所有模型（包括 qwen2.5:7b-instruct）都通过同一个 API： + https://api.siliconflow.cn/v1/chat/completions +- 不区分“本地 / 云端”客户端 +- 区分只体现在： + - model name + - prompt + - temperature / max_tokens + +.env.example 中必须包含： + +LLM_API_URL=https://api.siliconflow.cn/v1/chat/completions +LLM_API_KEY=your_api_key_here + +# 用于意图识别的小模型 +INTENT_MODEL_NAME=qwen2.5:7b-instruct + +# 用于对话 / 计划 / 代码生成的模型（可先用同一个） +GENERATION_MODEL_NAME=Pro/zai-org/GLM-4.7 + +==================== +【llm/client.py 要求】 +==================== +实现统一的 LLMClient： + +- 使用 requests.post +- URL / API KEY 从 .env 读取 +- 提供方法： + chat( + messages: list[dict], + model: str, + temperature: float, + max_tokens: int + ) -> str + +- payload 结构参考： +{ + "model": model, + "messages": messages, + "stream": false, + "temperature": temperature, + "max_tokens": max_tokens +} + +- headers: + Authorization: Bearer +- 对网络异常 / 非 200 状态码做明确异常抛出 +- 不要在 client 中写任何业务逻辑 + +==================== +【意图识别（核心修改点）】 +==================== +实现 intent/classifier.py： + +- 使用“小参数 LLM”（INTENT_MODEL_NAME，例如 qwen2.5:7b-instruct） +- 目标：二分类 + - chat + - execution +- 要求输出结构化结果： + { + "label": "chat" | "execution", + "confidence": 0.0 ~ 1.0, + "reason": "中文解释，说明为什么这是执行任务/对话任务" + } + +- Prompt 必须极短、强约束、可解析 +- Prompt 模板放在 llm/prompts.py +- 对 LLM 输出： + - 尝试解析 JSON + - 若解析失败 / 字段缺失 → 走兜底逻辑（判为 chat） + +intent/labels.py： +- 定义常量： + CHAT + EXECUTION +- 定义阈值： + EXECUTION_CONFIDENCE_THRESHOLD = 0.6 +- 低于阈值一律判定为 chat（宁可少执行，不可误执行） + +==================== +【Chat Task 流程】 +==================== +- 使用 GENERATION_MODEL_NAME +- messages = 用户原始输入 +- 返回文本直接展示 +- 不触碰本地、不产出文件 + +==================== +【Execution Task 流程】 +==================== +1) 生成执行计划 + - 可用 GENERATION_MODEL_NAME + - 输出中文、可读 + - 明确： + - 会做什么 + - 不会动原文件 + - 输入 / 输出目录 + - 可能失败的情况 + +2) 生成 Python 执行代码 + - MVP 先内置“安全示例代码”： + - 遍历 workspace/input + - 复制文件到 workspace/output + - 不依赖第三方库 + - 不修改原文件 + - 保存为 workspace/task_.py + +3) safety/rule_checker.py（硬规则） + - 静态扫描执行代码： + - 禁止 requests / socket / urllib + - 禁止访问非 workspace 路径 + - 禁止危险操作（os.remove, shutil.rmtree, subprocess 等） + - 若违反，直接 fail + +4) safety/llm_reviewer.py（软规则） + - 使用 GENERATION_MODEL_NAME + - 输入：用户需求 + 执行计划 + 代码 + - 输出：pass / fail + 中文原因 + +5) UI（小白引导式，方案 C） + - 显示： + - 判定原因 reason + - 三步引导： + 1) 把文件复制到 input + 2) 我来处理 + 3) 去 output 取 + - 执行计划 + - 风险提示 + - 【开始执行】按钮 + +6) executor/sandbox_runner.py + - 使用 subprocess 启动一次性 Python 子进程 + - 工作目录限定为 workspace + - 捕获 stdout / stderr + - 写入 workspace/logs/task_.log + - 执行完即退出 + - 执行器层不允许任何联网能力（由 rule_checker 保证） + +==================== +【UI（Tkinter）最小可跑要求】 +==================== +- main.py 启动 Tkinter 窗口 +- 顶部：输入框 + 发送按钮 +- 中部：输出区 +- 当识别为 execution： + - 切换或弹出 task_guide_view +- 执行完成后展示： + - success / partial / failed + - 成功 / 失败数量 + - 日志路径 + +==================== +【requirements.txt（最小集）】 +==================== +- python-dotenv +- requests + +==================== +【最小可跑验收标准】 +==================== +- 未配置 LLM_KEY 时给出明确错误提示 +- 输入“今天天气怎么样” → chat +- 输入“把这个文件夹里的图片复制一份” → execution +- execution 能生成 task_.py 并真正执行 +- output / logs 中有真实文件 + +==================== +【Plan 模式输出要求】 +==================== +1) 先输出整体实现计划（步骤、模块职责） +2) 列出所有文件及其责任说明 +3) 再按文件路径逐个输出代码内容 +4) 确保 main.py 可直接运行 +5) main.py 顶部注释说明： + - 如何配置 .env + - 如何运行 + - 如何测试（往 input 放文件） \ No newline at end of file diff --git a/__pycache__/main.cpython-313.pyc b/__pycache__/main.cpython-313.pyc new file mode 100644 index 0000000..0a159e7 Binary files /dev/null and b/__pycache__/main.cpython-313.pyc differ diff --git a/debug_env.py b/debug_env.py new file mode 100644 index 0000000..29d0eee --- /dev/null +++ b/debug_env.py @@ -0,0 +1,25 @@ +"""调试脚本""" +from pathlib import Path +from dotenv import load_dotenv +import os + +ENV_PATH = Path(__file__).parent / ".env" + +print(f"ENV_PATH: {ENV_PATH}") +print(f"ENV_PATH exists: {ENV_PATH.exists()}") + +# 读取文件内容 +if ENV_PATH.exists(): + print(f"File content:") + print(ENV_PATH.read_text(encoding='utf-8')) + +# 加载环境变量 +result = load_dotenv(ENV_PATH) +print(f"load_dotenv result: {result}") + +# 检查环境变量 +print(f"LLM_API_URL: {os.getenv('LLM_API_URL')}") +print(f"LLM_API_KEY: {os.getenv('LLM_API_KEY')}") +print(f"INTENT_MODEL_NAME: {os.getenv('INTENT_MODEL_NAME')}") +print(f"GENERATION_MODEL_NAME: {os.getenv('GENERATION_MODEL_NAME')}") + diff --git a/executor/__init__.py b/executor/__init__.py new file mode 100644 index 0000000..7e9d55e --- /dev/null +++ b/executor/__init__.py @@ -0,0 +1,2 @@ +# 执行器模块 + diff --git a/executor/__pycache__/__init__.cpython-310.pyc b/executor/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..c3aff9b Binary files /dev/null and b/executor/__pycache__/__init__.cpython-310.pyc differ diff --git a/executor/__pycache__/__init__.cpython-313.pyc b/executor/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..1d313ba Binary files /dev/null and b/executor/__pycache__/__init__.cpython-313.pyc differ diff --git a/executor/__pycache__/sandbox_runner.cpython-310.pyc b/executor/__pycache__/sandbox_runner.cpython-310.pyc new file mode 100644 index 0000000..6ce0596 Binary files /dev/null and b/executor/__pycache__/sandbox_runner.cpython-310.pyc differ diff --git a/executor/__pycache__/sandbox_runner.cpython-313.pyc b/executor/__pycache__/sandbox_runner.cpython-313.pyc new file mode 100644 index 0000000..0a3e37a Binary files /dev/null and b/executor/__pycache__/sandbox_runner.cpython-313.pyc differ diff --git a/executor/sandbox_runner.py b/executor/sandbox_runner.py new file mode 100644 index 0000000..920178c --- /dev/null +++ b/executor/sandbox_runner.py @@ -0,0 +1,240 @@ +""" +沙箱执行器 +在受限环境中执行生成的 Python 代码 +""" + +import os +import sys +import subprocess +import uuid +from datetime import datetime +from pathlib import Path +from typing import Optional +from dataclasses import dataclass + + +@dataclass +class ExecutionResult: + """执行结果""" + success: bool + task_id: str + stdout: str + stderr: str + return_code: int + log_path: str + duration_ms: int + + +class SandboxRunner: + """ + 沙箱执行器 + + 特性： + 1. 使用 subprocess 启动独立 Python 进程 + 2. 工作目录限定为 workspace + 3. 捕获所有输出 + 4. 写入日志文件 + """ + + def __init__(self, workspace_path: Optional[str] = None): + if workspace_path: + self.workspace = Path(workspace_path) + else: + # 默认使用项目根目录下的 workspace + self.workspace = Path(__file__).parent.parent / "workspace" + + self.input_dir = self.workspace / "input" + self.output_dir = self.workspace / "output" + self.logs_dir = self.workspace / "logs" + + # 确保目录存在 + self.input_dir.mkdir(parents=True, exist_ok=True) + self.output_dir.mkdir(parents=True, exist_ok=True) + self.logs_dir.mkdir(parents=True, exist_ok=True) + + def save_task_code(self, code: str, task_id: Optional[str] = None) -> tuple[str, Path]: + """ + 保存任务代码到文件 + + Args: + code: Python 代码 + task_id: 任务 ID（可选，自动生成） + + Returns: + (task_id, code_path) + """ + if not task_id: + task_id = self._generate_task_id() + + code_path = self.workspace / f"task_{task_id}.py" + code_path.write_text(code, encoding='utf-8') + + return task_id, code_path + + def execute(self, code: str, task_id: Optional[str] = None, timeout: int = 60) -> ExecutionResult: + """ + 执行代码 + + Args: + code: Python 代码 + task_id: 任务 ID + timeout: 超时时间（秒） + + Returns: + ExecutionResult: 执行结果 + """ + # 保存代码 + task_id, code_path = self.save_task_code(code, task_id) + + # 准备日志 + log_path = self.logs_dir / f"task_{task_id}.log" + + start_time = datetime.now() + + try: + # 使用 subprocess 执行 + result = subprocess.run( + [sys.executable, str(code_path)], + cwd=str(self.workspace), + capture_output=True, + text=True, + timeout=timeout, + # 不继承父进程的环境变量中的网络代理等 + env=self._get_safe_env() + ) + + end_time = datetime.now() + duration_ms = int((end_time - start_time).total_seconds() * 1000) + + # 写入日志 + self._write_log( + log_path=log_path, + task_id=task_id, + code_path=code_path, + stdout=result.stdout, + stderr=result.stderr, + return_code=result.returncode, + duration_ms=duration_ms + ) + + return ExecutionResult( + success=result.returncode == 0, + task_id=task_id, + stdout=result.stdout, + stderr=result.stderr, + return_code=result.returncode, + log_path=str(log_path), + duration_ms=duration_ms + ) + + except subprocess.TimeoutExpired: + end_time = datetime.now() + duration_ms = int((end_time - start_time).total_seconds() * 1000) + + error_msg = f"执行超时（超过 {timeout} 秒）" + + self._write_log( + log_path=log_path, + task_id=task_id, + code_path=code_path, + stdout="", + stderr=error_msg, + return_code=-1, + duration_ms=duration_ms + ) + + return ExecutionResult( + success=False, + task_id=task_id, + stdout="", + stderr=error_msg, + return_code=-1, + log_path=str(log_path), + duration_ms=duration_ms + ) + + except Exception as e: + end_time = datetime.now() + duration_ms = int((end_time - start_time).total_seconds() * 1000) + + error_msg = f"执行异常: {str(e)}" + + self._write_log( + log_path=log_path, + task_id=task_id, + code_path=code_path, + stdout="", + stderr=error_msg, + return_code=-1, + duration_ms=duration_ms + ) + + return ExecutionResult( + success=False, + task_id=task_id, + stdout="", + stderr=error_msg, + return_code=-1, + log_path=str(log_path), + duration_ms=duration_ms + ) + + def _generate_task_id(self) -> str: + """生成任务 ID""" + timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") + short_uuid = uuid.uuid4().hex[:6] + return f"{timestamp}_{short_uuid}" + + def _get_safe_env(self) -> dict: + """获取安全的环境变量（移除网络代理等）""" + safe_env = os.environ.copy() + + # 移除可能的网络代理设置 + proxy_vars = [ + 'HTTP_PROXY', 'HTTPS_PROXY', 'http_proxy', 'https_proxy', + 'ALL_PROXY', 'all_proxy', 'NO_PROXY', 'no_proxy' + ] + for var in proxy_vars: + safe_env.pop(var, None) + + return safe_env + + def _write_log( + self, + log_path: Path, + task_id: str, + code_path: Path, + stdout: str, + stderr: str, + return_code: int, + duration_ms: int + ): + """写入执行日志""" + log_content = f"""======================================== +任务执行日志 +======================================== +任务 ID: {task_id} +代码文件: {code_path} +执行时间: {datetime.now().strftime("%Y-%m-%d %H:%M:%S")} +耗时: {duration_ms} ms +返回码: {return_code} +状态: {"成功" if return_code == 0 else "失败"} + +======================================== +标准输出 (stdout) +======================================== +{stdout if stdout else "(无输出)"} + +======================================== +标准错误 (stderr) +======================================== +{stderr if stderr else "(无错误)"} +""" + log_path.write_text(log_content, encoding='utf-8') + + +def run_task(code: str, task_id: Optional[str] = None) -> ExecutionResult: + """便捷函数：执行任务""" + runner = SandboxRunner() + return runner.execute(code, task_id) + diff --git a/intent/__init__.py b/intent/__init__.py new file mode 100644 index 0000000..0e1c9d8 --- /dev/null +++ b/intent/__init__.py @@ -0,0 +1,2 @@ +# 意图识别模块 + diff --git a/intent/__pycache__/__init__.cpython-310.pyc b/intent/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..2ebd3b5 Binary files /dev/null and b/intent/__pycache__/__init__.cpython-310.pyc differ diff --git a/intent/__pycache__/__init__.cpython-313.pyc b/intent/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..d93cf15 Binary files /dev/null and b/intent/__pycache__/__init__.cpython-313.pyc differ diff --git a/intent/__pycache__/classifier.cpython-310.pyc b/intent/__pycache__/classifier.cpython-310.pyc new file mode 100644 index 0000000..b16a775 Binary files /dev/null and b/intent/__pycache__/classifier.cpython-310.pyc differ diff --git a/intent/__pycache__/classifier.cpython-313.pyc b/intent/__pycache__/classifier.cpython-313.pyc new file mode 100644 index 0000000..b3ee21d Binary files /dev/null and b/intent/__pycache__/classifier.cpython-313.pyc differ diff --git a/intent/__pycache__/labels.cpython-310.pyc b/intent/__pycache__/labels.cpython-310.pyc new file mode 100644 index 0000000..01dad99 Binary files /dev/null and b/intent/__pycache__/labels.cpython-310.pyc differ diff --git a/intent/__pycache__/labels.cpython-313.pyc b/intent/__pycache__/labels.cpython-313.pyc new file mode 100644 index 0000000..796d3b6 Binary files /dev/null and b/intent/__pycache__/labels.cpython-313.pyc differ diff --git a/intent/classifier.py b/intent/classifier.py new file mode 100644 index 0000000..194ac49 --- /dev/null +++ b/intent/classifier.py @@ -0,0 +1,152 @@ +""" +意图识别器 +使用小参数 LLM 进行意图二分类 +""" + +import os +import json +from pathlib import Path +from typing import Optional +from dataclasses import dataclass +from dotenv import load_dotenv + +from llm.client import get_client, LLMClientError, ENV_PATH +from llm.prompts import INTENT_CLASSIFICATION_SYSTEM, INTENT_CLASSIFICATION_USER +from intent.labels import CHAT, EXECUTION, EXECUTION_CONFIDENCE_THRESHOLD, VALID_LABELS + + +@dataclass +class IntentResult: + """意图识别结果""" + label: str # chat 或 execution + confidence: float # 0.0 ~ 1.0 + reason: str # 中文解释 + raw_response: Optional[str] = None # 原始 LLM 响应（调试用） + + +class IntentClassifier: + """ + 意图分类器 + + 使用小参数 LLM（如 qwen2.5:7b-instruct）进行快速意图识别 + """ + + def __init__(self): + load_dotenv(ENV_PATH) + self.model_name = os.getenv("INTENT_MODEL_NAME") + + def classify(self, user_input: str) -> IntentResult: + """ + 对用户输入进行意图分类 + + Args: + user_input: 用户输入的文本 + + Returns: + IntentResult: 包含 label, confidence, reason 的结果 + """ + try: + client = get_client() + + messages = [ + {"role": "system", "content": INTENT_CLASSIFICATION_SYSTEM}, + {"role": "user", "content": INTENT_CLASSIFICATION_USER.format(user_input=user_input)} + ] + + response = client.chat( + messages=messages, + model=self.model_name, + temperature=0.1, # 低温度，更确定性的输出 + max_tokens=256 + ) + + return self._parse_response(response) + + except LLMClientError as e: + # LLM 调用失败，走兜底逻辑 + return IntentResult( + label=CHAT, + confidence=0.0, + reason=f"意图识别失败（{str(e)}），默认为对话模式" + ) + except Exception as e: + # 其他异常，走兜底逻辑 + return IntentResult( + label=CHAT, + confidence=0.0, + reason=f"意图识别异常（{str(e)}），默认为对话模式" + ) + + def _parse_response(self, response: str) -> IntentResult: + """ + 解析 LLM 响应 + + 尝试解析 JSON，若失败则走兜底逻辑 + """ + try: + # 尝试提取 JSON（LLM 可能会在 JSON 前后加一些文字） + json_str = self._extract_json(response) + data = json.loads(json_str) + + # 验证必要字段 + label = data.get("label", "").lower() + confidence = float(data.get("confidence", 0.0)) + reason = data.get("reason", "无") + + # 验证 label 有效性 + if label not in VALID_LABELS: + return IntentResult( + label=CHAT, + confidence=0.0, + reason=f"无效的意图标签 '{label}'，默认为对话模式", + raw_response=response + ) + + # 应用置信度阈值 + if label == EXECUTION and confidence < EXECUTION_CONFIDENCE_THRESHOLD: + return IntentResult( + label=CHAT, + confidence=confidence, + reason=f"执行任务置信度不足（{confidence:.2f} < {EXECUTION_CONFIDENCE_THRESHOLD}），降级为对话模式。原因: {reason}", + raw_response=response + ) + + return IntentResult( + label=label, + confidence=confidence, + reason=reason, + raw_response=response + ) + + except (json.JSONDecodeError, ValueError, TypeError) as e: + # JSON 解析失败，走兜底逻辑 + return IntentResult( + label=CHAT, + confidence=0.0, + reason=f"响应解析失败，默认为对话模式", + raw_response=response + ) + + def _extract_json(self, text: str) -> str: + """ + 从文本中提取 JSON 字符串 + + LLM 可能会在 JSON 前后添加解释文字，需要提取纯 JSON 部分 + """ + # 尝试找到 JSON 对象的起止位置 + start = text.find('{') + end = text.rfind('}') + + if start != -1 and end != -1 and end > start: + return text[start:end + 1] + + # 如果找不到，返回原文本让 json.loads 报错 + return text + + +# 便捷函数 +def classify_intent(user_input: str) -> IntentResult: + """快速进行意图分类""" + classifier = IntentClassifier() + return classifier.classify(user_input) + diff --git a/intent/labels.py b/intent/labels.py new file mode 100644 index 0000000..128e309 --- /dev/null +++ b/intent/labels.py @@ -0,0 +1,15 @@ +""" +意图标签定义 +""" + +# 意图类型常量 +CHAT = "chat" +EXECUTION = "execution" + +# 执行任务置信度阈值 +# 低于此阈值一律判定为 chat（宁可少执行，不可误执行） +EXECUTION_CONFIDENCE_THRESHOLD = 0.6 + +# 所有有效标签 +VALID_LABELS = {CHAT, EXECUTION} + diff --git a/llm/__init__.py b/llm/__init__.py new file mode 100644 index 0000000..61b410d --- /dev/null +++ b/llm/__init__.py @@ -0,0 +1,2 @@ +# LLM 模块 + diff --git a/llm/__pycache__/__init__.cpython-310.pyc b/llm/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..743fed2 Binary files /dev/null and b/llm/__pycache__/__init__.cpython-310.pyc differ diff --git a/llm/__pycache__/__init__.cpython-313.pyc b/llm/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..8235eb7 Binary files /dev/null and b/llm/__pycache__/__init__.cpython-313.pyc differ diff --git a/llm/__pycache__/client.cpython-310.pyc b/llm/__pycache__/client.cpython-310.pyc new file mode 100644 index 0000000..97fa637 Binary files /dev/null and b/llm/__pycache__/client.cpython-310.pyc differ diff --git a/llm/__pycache__/client.cpython-313.pyc b/llm/__pycache__/client.cpython-313.pyc new file mode 100644 index 0000000..8f3f1a3 Binary files /dev/null and b/llm/__pycache__/client.cpython-313.pyc differ diff --git a/llm/__pycache__/prompts.cpython-310.pyc b/llm/__pycache__/prompts.cpython-310.pyc new file mode 100644 index 0000000..6386c57 Binary files /dev/null and b/llm/__pycache__/prompts.cpython-310.pyc differ diff --git a/llm/__pycache__/prompts.cpython-313.pyc b/llm/__pycache__/prompts.cpython-313.pyc new file mode 100644 index 0000000..adcaa0c Binary files /dev/null and b/llm/__pycache__/prompts.cpython-313.pyc differ diff --git a/llm/client.py b/llm/client.py new file mode 100644 index 0000000..9b9a754 --- /dev/null +++ b/llm/client.py @@ -0,0 +1,124 @@ +""" +LLM 统一调用客户端 +所有模型通过 SiliconFlow API 调用 +""" + +import os +import requests +from pathlib import Path +from typing import Optional +from dotenv import load_dotenv + +# 获取项目根目录 +PROJECT_ROOT = Path(__file__).parent.parent +ENV_PATH = PROJECT_ROOT / ".env" + + +class LLMClientError(Exception): + """LLM 客户端异常""" + pass + + +class LLMClient: + """ + 统一的 LLM 调用客户端 + + 使用方式: + client = LLMClient() + response = client.chat( + messages=[{"role": "user", "content": "你好"}], + model="Qwen/Qwen2.5-7B-Instruct", + temperature=0.7, + max_tokens=1024 + ) + """ + + def __init__(self): + load_dotenv(ENV_PATH) + + self.api_url = os.getenv("LLM_API_URL") + self.api_key = os.getenv("LLM_API_KEY") + + if not self.api_url: + raise LLMClientError("未配置 LLM_API_URL，请检查 .env 文件") + if not self.api_key or self.api_key == "your_api_key_here": + raise LLMClientError("未配置有效的 LLM_API_KEY，请检查 .env 文件") + + def chat( + self, + messages: list[dict], + model: str, + temperature: float = 0.7, + max_tokens: int = 1024 + ) -> str: + """ + 调用 LLM 进行对话 + + Args: + messages: 消息列表，格式为 [{"role": "user/assistant/system", "content": "..."}] + model: 模型名称 + temperature: 温度参数，控制随机性 + max_tokens: 最大生成 token 数 + + Returns: + LLM 生成的文本内容 + + Raises: + LLMClientError: 网络异常或 API 返回错误 + """ + headers = { + "Authorization": f"Bearer {self.api_key}", + "Content-Type": "application/json" + } + + payload = { + "model": model, + "messages": messages, + "stream": False, + "temperature": temperature, + "max_tokens": max_tokens + } + + try: + response = requests.post( + self.api_url, + headers=headers, + json=payload, + timeout=60 + ) + except requests.exceptions.Timeout: + raise LLMClientError("请求超时，请检查网络连接") + except requests.exceptions.ConnectionError: + raise LLMClientError("网络连接失败，请检查网络设置") + except requests.exceptions.RequestException as e: + raise LLMClientError(f"网络请求异常: {str(e)}") + + if response.status_code != 200: + error_msg = f"API 返回错误 (状态码: {response.status_code})" + try: + error_detail = response.json() + if "error" in error_detail: + error_msg += f": {error_detail['error']}" + except: + error_msg += f": {response.text[:200]}" + raise LLMClientError(error_msg) + + try: + result = response.json() + content = result["choices"][0]["message"]["content"] + return content + except (KeyError, IndexError, TypeError) as e: + raise LLMClientError(f"解析 API 响应失败: {str(e)}") + + +# 全局单例（延迟初始化） +_client: Optional[LLMClient] = None + + +def get_client() -> LLMClient: + """获取 LLM 客户端单例""" + global _client + if _client is None: + _client = LLMClient() + return _client + diff --git a/llm/prompts.py b/llm/prompts.py new file mode 100644 index 0000000..c425833 --- /dev/null +++ b/llm/prompts.py @@ -0,0 +1,130 @@ +""" +Prompt 模板集合 +所有与 LLM 交互的 Prompt 统一在此管理 +""" + +# ======================================== +# 意图识别 Prompt +# ======================================== + +INTENT_CLASSIFICATION_SYSTEM = """你是一个意图分类器。判断用户输入是"普通对话"还是"本地执行任务"。 + +规则： +- chat: 闲聊、问答、知识查询（如天气、新闻、解释概念） +- execution: 需要操作本地文件的任务（如复制、移动、重命名、整理文件） + +只输出JSON，格式： +{"label": "chat或execution", "confidence": 0.0到1.0, "reason": "简短中文理由"}""" + +INTENT_CLASSIFICATION_USER = """判断以下输入的意图： +{user_input}""" + + +# ======================================== +# 执行计划生成 Prompt +# ======================================== + +EXECUTION_PLAN_SYSTEM = """你是一个任务规划助手。根据用户需求，生成清晰的执行计划。 + +约束： +1. 所有操作只在 workspace 目录内进行 +2. 输入文件来自 workspace/input +3. 输出文件保存到 workspace/output +4. 绝不修改或删除原始文件 +5. 不进行任何网络操作 + +输出格式（中文）： +## 任务理解 +[简述用户想做什么] + +## 执行步骤 +1. [步骤1] +2. [步骤2] +... + +## 输入输出 +- 输入目录: workspace/input +- 输出目录: workspace/output + +## 风险提示 +[可能失败的情况]""" + +EXECUTION_PLAN_USER = """用户需求：{user_input} + +请生成执行计划。""" + + +# ======================================== +# 代码生成 Prompt +# ======================================== + +CODE_GENERATION_SYSTEM = """你是一个 Python 代码生成器。根据执行计划生成安全的文件处理代码。 + +硬性约束： +1. 只能操作 workspace/input 和 workspace/output 目录 +2. 禁止使用: requests, socket, urllib, subprocess, os.system +3. 禁止删除文件: os.remove, shutil.rmtree, os.unlink +4. 禁止访问 workspace 外的任何路径 +5. 只使用标准库: os, shutil, pathlib, json, csv 等 + +代码模板： +```python +import os +import shutil +from pathlib import Path + +# 工作目录 +WORKSPACE = Path(__file__).parent +INPUT_DIR = WORKSPACE / "input" +OUTPUT_DIR = WORKSPACE / "output" + +def main(): + # 确保输出目录存在 + OUTPUT_DIR.mkdir(exist_ok=True) + + # TODO: 实现具体逻辑 + + print("任务完成") + +if __name__ == "__main__": + main() +``` + +只输出 Python 代码，不要其他解释。""" + +CODE_GENERATION_USER = """执行计划： +{execution_plan} + +用户原始需求：{user_input} + +请生成 Python 代码。""" + + +# ======================================== +# 安全审查 Prompt +# ======================================== + +SAFETY_REVIEW_SYSTEM = """你是一个代码安全审查员。检查代码是否符合安全规范。 + +检查项： +1. 是否只操作 workspace 目录 +2. 是否有网络请求代码 +3. 是否有危险的文件删除操作 +4. 是否有执行外部命令的代码 +5. 代码逻辑是否与用户需求一致 + +输出JSON格式： +{"pass": true或false, "reason": "中文审查结论"}""" + +SAFETY_REVIEW_USER = """用户需求：{user_input} + +执行计划： +{execution_plan} + +待审查代码： +```python +{code} +``` + +请进行安全审查。""" + diff --git a/main.py b/main.py new file mode 100644 index 0000000..3dd14e2 --- /dev/null +++ b/main.py @@ -0,0 +1,518 @@ +""" +LocalAgent - Windows 本地 AI 执行助手 (MVP) + +======================================== +配置说明 +======================================== +1. 复制 .env.example 为 .env +2. 在 .env 中填入你的 SiliconFlow API Key: + LLM_API_KEY=sk-xxxxx + +======================================== +运行方式 +======================================== +方式一：使用 Anaconda + conda create -n localagent python=3.10 + conda activate localagent + pip install -r requirements.txt + python main.py + +方式二：直接运行（需已安装依赖） + python main.py + +======================================== +测试方法 +======================================== +1. 对话测试：输入 "今天天气怎么样" → 应识别为 chat +2. 执行测试： + - 将测试文件放入 workspace/input 目录 + - 输入 "把这些文件复制一份" → 应识别为 execution + - 确认执行后，检查 workspace/output 目录 + +======================================== +""" + +import os +import sys +import tkinter as tk +from tkinter import messagebox +from pathlib import Path +from typing import Optional +from dotenv import load_dotenv +import threading +import queue + +# 确保项目根目录在 Python 路径中 +PROJECT_ROOT = Path(__file__).parent +ENV_PATH = PROJECT_ROOT / ".env" +sys.path.insert(0, str(PROJECT_ROOT)) + +# 在导入其他模块之前先加载环境变量 +load_dotenv(ENV_PATH) + +from llm.client import get_client, LLMClientError +from llm.prompts import ( + EXECUTION_PLAN_SYSTEM, EXECUTION_PLAN_USER, + CODE_GENERATION_SYSTEM, CODE_GENERATION_USER +) +from intent.classifier import classify_intent, IntentResult +from intent.labels import CHAT, EXECUTION +from safety.rule_checker import check_code_safety +from safety.llm_reviewer import review_code_safety +from executor.sandbox_runner import SandboxRunner, ExecutionResult +from ui.chat_view import ChatView +from ui.task_guide_view import TaskGuideView + + +class LocalAgentApp: + """ + LocalAgent 主应用 + + 职责： + 1. 管理 UI 状态切换 + 2. 协调各模块工作流程 + 3. 处理用户交互 + """ + + def __init__(self): + self.workspace = PROJECT_ROOT / "workspace" + self.runner = SandboxRunner(str(self.workspace)) + + # 当前任务状态 + self.current_task: Optional[dict] = None + + # 线程通信队列 + self.result_queue = queue.Queue() + + # 初始化 UI + self._init_ui() + + def _init_ui(self): + """初始化 UI""" + self.root = tk.Tk() + self.root.title("LocalAgent - 本地 AI 助手") + self.root.geometry("800x700") + self.root.configure(bg='#1e1e1e') + + # 设置窗口图标（如果有的话） + try: + self.root.iconbitmap(PROJECT_ROOT / "icon.ico") + except: + pass + + # 主容器 + self.main_container = tk.Frame(self.root, bg='#1e1e1e') + self.main_container.pack(fill=tk.BOTH, expand=True) + + # 聊天视图 + self.chat_view = ChatView(self.main_container, self._on_user_input) + + # 任务引导视图（初始隐藏） + self.task_view: Optional[TaskGuideView] = None + + # 定期检查后台任务结果 + self._check_queue() + + def _check_queue(self): + """检查后台任务队列""" + try: + while True: + callback, args = self.result_queue.get_nowait() + callback(*args) + except queue.Empty: + pass + + # 每 100ms 检查一次 + self.root.after(100, self._check_queue) + + def _run_in_thread(self, func, callback, *args): + """在后台线程运行函数，完成后回调""" + def wrapper(): + try: + result = func(*args) + self.result_queue.put((callback, (result, None))) + except Exception as e: + self.result_queue.put((callback, (None, e))) + + thread = threading.Thread(target=wrapper, daemon=True) + thread.start() + + def _on_user_input(self, user_input: str): + """处理用户输入""" + # 显示用户消息 + self.chat_view.add_message(user_input, 'user') + self.chat_view.set_input_enabled(False) + self.chat_view.add_message("正在分析您的需求...", 'system') + + # 在后台线程进行意图识别 + self._run_in_thread( + classify_intent, + lambda result, error: self._on_intent_result(user_input, result, error), + user_input + ) + + def _on_intent_result(self, user_input: str, intent_result: Optional[IntentResult], error: Optional[Exception]): + """意图识别完成回调""" + if error: + self.chat_view.add_message(f"意图识别失败: {str(error)}", 'error') + self.chat_view.set_input_enabled(True) + return + + if intent_result.label == CHAT: + # 对话模式 + self._handle_chat(user_input, intent_result) + else: + # 执行模式 + self._handle_execution(user_input, intent_result) + + def _handle_chat(self, user_input: str, intent_result: IntentResult): + """处理对话任务""" + self.chat_view.add_message( + f"识别为对话模式 (原因: {intent_result.reason})", + 'system' + ) + self.chat_view.add_message("正在生成回复...", 'system') + + # 在后台线程调用 LLM + def do_chat(): + client = get_client() + model = os.getenv("GENERATION_MODEL_NAME") + return client.chat( + messages=[{"role": "user", "content": user_input}], + model=model, + temperature=0.7, + max_tokens=2048 + ) + + self._run_in_thread( + do_chat, + self._on_chat_result + ) + + def _on_chat_result(self, response: Optional[str], error: Optional[Exception]): + """对话完成回调""" + if error: + self.chat_view.add_message(f"对话失败: {str(error)}", 'error') + else: + self.chat_view.add_message(response, 'assistant') + + self.chat_view.set_input_enabled(True) + + def _handle_execution(self, user_input: str, intent_result: IntentResult): + """处理执行任务""" + self.chat_view.add_message( + f"识别为执行任务 (置信度: {intent_result.confidence:.0%})\n原因: {intent_result.reason}", + 'system' + ) + self.chat_view.add_message("正在生成执行计划...", 'system') + + # 保存用户输入和意图结果 + self.current_task = { + 'user_input': user_input, + 'intent_result': intent_result + } + + # 在后台线程生成执行计划 + self._run_in_thread( + self._generate_execution_plan, + self._on_plan_generated, + user_input + ) + + def _on_plan_generated(self, plan: Optional[str], error: Optional[Exception]): + """执行计划生成完成回调""" + if error: + self.chat_view.add_message(f"生成执行计划失败: {str(error)}", 'error') + self.chat_view.set_input_enabled(True) + self.current_task = None + return + + self.current_task['execution_plan'] = plan + self.chat_view.add_message("正在生成执行代码...", 'system') + + # 在后台线程生成代码 + self._run_in_thread( + self._generate_code, + self._on_code_generated, + self.current_task['user_input'], + plan + ) + + def _on_code_generated(self, code: Optional[str], error: Optional[Exception]): + """代码生成完成回调""" + if error: + self.chat_view.add_message(f"生成代码失败: {str(error)}", 'error') + self.chat_view.set_input_enabled(True) + self.current_task = None + return + + self.current_task['code'] = code + self.chat_view.add_message("正在进行安全检查...", 'system') + + # 硬规则检查（同步，很快） + rule_result = check_code_safety(code) + if not rule_result.passed: + violations = "\n".join(f" • {v}" for v in rule_result.violations) + self.chat_view.add_message( + f"安全检查未通过，任务已取消:\n{violations}", + 'error' + ) + self.chat_view.set_input_enabled(True) + self.current_task = None + return + + # 在后台线程进行 LLM 安全审查 + self._run_in_thread( + review_code_safety, + self._on_safety_reviewed, + self.current_task['user_input'], + self.current_task['execution_plan'], + code + ) + + def _on_safety_reviewed(self, review_result, error: Optional[Exception]): + """安全审查完成回调""" + if error: + self.chat_view.add_message(f"安全审查失败: {str(error)}", 'error') + self.chat_view.set_input_enabled(True) + self.current_task = None + return + + if not review_result.passed: + self.chat_view.add_message( + f"安全审查未通过: {review_result.reason}", + 'error' + ) + self.chat_view.set_input_enabled(True) + self.current_task = None + return + + self.chat_view.add_message("安全检查通过，请确认执行", 'system') + + # 显示任务引导视图 + self._show_task_guide() + + def _generate_execution_plan(self, user_input: str) -> str: + """生成执行计划""" + client = get_client() + model = os.getenv("GENERATION_MODEL_NAME") + + response = client.chat( + messages=[ + {"role": "system", "content": EXECUTION_PLAN_SYSTEM}, + {"role": "user", "content": EXECUTION_PLAN_USER.format(user_input=user_input)} + ], + model=model, + temperature=0.3, + max_tokens=1024 + ) + + return response + + def _generate_code(self, user_input: str, execution_plan: str) -> str: + """生成执行代码""" + client = get_client() + model = os.getenv("GENERATION_MODEL_NAME") + + response = client.chat( + messages=[ + {"role": "system", "content": CODE_GENERATION_SYSTEM}, + {"role": "user", "content": CODE_GENERATION_USER.format( + user_input=user_input, + execution_plan=execution_plan + )} + ], + model=model, + temperature=0.2, + max_tokens=2048 + ) + + # 提取代码块 + code = self._extract_code(response) + return code + + def _extract_code(self, response: str) -> str: + """从 LLM 响应中提取代码""" + import re + + # 尝试提取 ```python ... ``` 代码块 + pattern = r'```python\s*(.*?)\s*```' + matches = re.findall(pattern, response, re.DOTALL) + + if matches: + return matches[0] + + # 尝试提取 ``` ... ``` 代码块 + pattern = r'```\s*(.*?)\s*```' + matches = re.findall(pattern, response, re.DOTALL) + + if matches: + return matches[0] + + # 如果没有代码块，返回原始响应 + return response + + def _show_task_guide(self): + """显示任务引导视图""" + if not self.current_task: + return + + # 隐藏聊天视图 + self.chat_view.get_frame().pack_forget() + + # 创建任务引导视图 + self.task_view = TaskGuideView( + self.main_container, + on_execute=self._on_execute_task, + on_cancel=self._on_cancel_task, + workspace_path=self.workspace + ) + + # 设置内容 + self.task_view.set_intent_result( + self.current_task['intent_result'].reason, + self.current_task['intent_result'].confidence + ) + self.task_view.set_execution_plan(self.current_task['execution_plan']) + + # 显示 + self.task_view.show() + + def _on_execute_task(self): + """执行任务""" + if not self.current_task: + return + + self.task_view.set_buttons_enabled(False) + + # 在后台线程执行 + def do_execute(): + return self.runner.execute(self.current_task['code']) + + self._run_in_thread( + do_execute, + self._on_execution_complete + ) + + def _on_execution_complete(self, result: Optional[ExecutionResult], error: Optional[Exception]): + """执行完成回调""" + if error: + messagebox.showerror("执行错误", f"执行失败: {str(error)}") + else: + self._show_execution_result(result) + # 刷新输出文件列表 + if self.task_view: + self.task_view.refresh_output() + + self._back_to_chat() + + def _show_execution_result(self, result: ExecutionResult): + """显示执行结果""" + if result.success: + status = "执行成功" + else: + status = "执行失败" + + message = f"""{status} + +任务 ID: {result.task_id} +耗时: {result.duration_ms} ms +日志文件: {result.log_path} + +输出: +{result.stdout if result.stdout else '(无输出)'} + +{f'错误信息: {result.stderr}' if result.stderr else ''} +""" + + if result.success: + messagebox.showinfo("执行结果", message) + # 打开 output 目录 + os.startfile(str(self.workspace / "output")) + else: + messagebox.showerror("执行结果", message) + + def _on_cancel_task(self): + """取消任务""" + self.current_task = None + self._back_to_chat() + + def _back_to_chat(self): + """返回聊天视图""" + if self.task_view: + self.task_view.hide() + self.task_view = None + + self.chat_view.get_frame().pack(fill=tk.BOTH, expand=True, padx=10, pady=10) + self.chat_view.set_input_enabled(True) + self.current_task = None + + def run(self): + """运行应用""" + self.root.mainloop() + + +def check_environment(): + """检查运行环境""" + load_dotenv(ENV_PATH) + + api_key = os.getenv("LLM_API_KEY") + + if not api_key or api_key == "your_api_key_here": + print("=" * 50) + print("错误: 未配置 LLM API Key") + print("=" * 50) + print() + print("请按以下步骤配置:") + print("1. 复制 .env.example 为 .env") + print("2. 在 .env 中设置 LLM_API_KEY=你的API密钥") + print() + print("获取 API Key: https://siliconflow.cn") + print("=" * 50) + + # 显示 GUI 错误提示 + root = tk.Tk() + root.withdraw() + messagebox.showerror( + "配置错误", + "未配置 LLM API Key\n\n" + "请按以下步骤配置:\n" + "1. 复制 .env.example 为 .env\n" + "2. 在 .env 中设置 LLM_API_KEY=你的API密钥\n\n" + "获取 API Key: https://siliconflow.cn" + ) + root.destroy() + return False + + return True + + +def main(): + """主入口""" + print("=" * 50) + print("LocalAgent - Windows 本地 AI 执行助手") + print("=" * 50) + + # 检查环境 + if not check_environment(): + sys.exit(1) + + # 创建工作目录 + workspace = PROJECT_ROOT / "workspace" + (workspace / "input").mkdir(parents=True, exist_ok=True) + (workspace / "output").mkdir(parents=True, exist_ok=True) + (workspace / "logs").mkdir(parents=True, exist_ok=True) + + print(f"工作目录: {workspace}") + print(f"输入目录: {workspace / 'input'}") + print(f"输出目录: {workspace / 'output'}") + print(f"日志目录: {workspace / 'logs'}") + print("=" * 50) + + # 启动应用 + app = LocalAgentApp() + app.run() + + +if __name__ == "__main__": + main() diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..08157af --- /dev/null +++ b/requirements.txt @@ -0,0 +1,9 @@ +# LocalAgent MVP 依赖 +# 使用 Anaconda 创建虚拟环境后安装： +# conda create -n localagent python=3.10 +# conda activate localagent +# pip install -r requirements.txt + +python-dotenv>=1.0.0 +requests>=2.31.0 + diff --git a/safety/__init__.py b/safety/__init__.py new file mode 100644 index 0000000..e44309f --- /dev/null +++ b/safety/__init__.py @@ -0,0 +1,2 @@ +# 安全检查模块 + diff --git a/safety/__pycache__/__init__.cpython-310.pyc b/safety/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..64521a4 Binary files /dev/null and b/safety/__pycache__/__init__.cpython-310.pyc differ diff --git a/safety/__pycache__/__init__.cpython-313.pyc b/safety/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..6a00445 Binary files /dev/null and b/safety/__pycache__/__init__.cpython-313.pyc differ diff --git a/safety/__pycache__/llm_reviewer.cpython-310.pyc b/safety/__pycache__/llm_reviewer.cpython-310.pyc new file mode 100644 index 0000000..ff7b24b Binary files /dev/null and b/safety/__pycache__/llm_reviewer.cpython-310.pyc differ diff --git a/safety/__pycache__/llm_reviewer.cpython-313.pyc b/safety/__pycache__/llm_reviewer.cpython-313.pyc new file mode 100644 index 0000000..84999ce Binary files /dev/null and b/safety/__pycache__/llm_reviewer.cpython-313.pyc differ diff --git a/safety/__pycache__/rule_checker.cpython-310.pyc b/safety/__pycache__/rule_checker.cpython-310.pyc new file mode 100644 index 0000000..3b6a094 Binary files /dev/null and b/safety/__pycache__/rule_checker.cpython-310.pyc differ diff --git a/safety/__pycache__/rule_checker.cpython-313.pyc b/safety/__pycache__/rule_checker.cpython-313.pyc new file mode 100644 index 0000000..36c1432 Binary files /dev/null and b/safety/__pycache__/rule_checker.cpython-313.pyc differ diff --git a/safety/llm_reviewer.py b/safety/llm_reviewer.py new file mode 100644 index 0000000..b745154 --- /dev/null +++ b/safety/llm_reviewer.py @@ -0,0 +1,132 @@ +""" +LLM 软规则审查器 +使用 LLM 进行代码安全审查 +""" + +import os +import json +from typing import Optional +from dataclasses import dataclass +from dotenv import load_dotenv + +from llm.client import get_client, LLMClientError, ENV_PATH +from llm.prompts import SAFETY_REVIEW_SYSTEM, SAFETY_REVIEW_USER + + +@dataclass +class LLMReviewResult: + """LLM 审查结果""" + passed: bool + reason: str + raw_response: Optional[str] = None + + +class LLMReviewer: + """ + LLM 安全审查器 + + 使用大模型对代码进行语义级别的安全审查 + """ + + def __init__(self): + load_dotenv(ENV_PATH) + self.model_name = os.getenv("GENERATION_MODEL_NAME") + + def review( + self, + user_input: str, + execution_plan: str, + code: str + ) -> LLMReviewResult: + """ + 审查代码安全性 + + Args: + user_input: 用户原始需求 + execution_plan: 执行计划 + code: 待审查的代码 + + Returns: + LLMReviewResult: 审查结果 + """ + try: + client = get_client() + + messages = [ + {"role": "system", "content": SAFETY_REVIEW_SYSTEM}, + {"role": "user", "content": SAFETY_REVIEW_USER.format( + user_input=user_input, + execution_plan=execution_plan, + code=code + )} + ] + + response = client.chat( + messages=messages, + model=self.model_name, + temperature=0.1, + max_tokens=512 + ) + + return self._parse_response(response) + + except LLMClientError as e: + # LLM 调用失败，保守起见判定为不通过 + return LLMReviewResult( + passed=False, + reason=f"安全审查失败（{str(e)}），出于安全考虑拒绝执行" + ) + except Exception as e: + return LLMReviewResult( + passed=False, + reason=f"安全审查异常（{str(e)}），出于安全考虑拒绝执行" + ) + + def _parse_response(self, response: str) -> LLMReviewResult: + """解析 LLM 响应""" + try: + # 提取 JSON + json_str = self._extract_json(response) + data = json.loads(json_str) + + passed = data.get("pass", False) + reason = data.get("reason", "未提供原因") + + # 确保 passed 是布尔值 + if isinstance(passed, str): + passed = passed.lower() in ('true', 'yes', '1', 'pass') + + return LLMReviewResult( + passed=bool(passed), + reason=reason, + raw_response=response + ) + + except (json.JSONDecodeError, ValueError, TypeError): + # 解析失败，保守判定 + return LLMReviewResult( + passed=False, + reason=f"审查结果解析失败，出于安全考虑拒绝执行", + raw_response=response + ) + + def _extract_json(self, text: str) -> str: + """从文本中提取 JSON""" + start = text.find('{') + end = text.rfind('}') + + if start != -1 and end != -1 and end > start: + return text[start:end + 1] + + return text + + +def review_code_safety( + user_input: str, + execution_plan: str, + code: str +) -> LLMReviewResult: + """便捷函数：审查代码安全性""" + reviewer = LLMReviewer() + return reviewer.review(user_input, execution_plan, code) + diff --git a/safety/rule_checker.py b/safety/rule_checker.py new file mode 100644 index 0000000..be53481 --- /dev/null +++ b/safety/rule_checker.py @@ -0,0 +1,208 @@ +""" +硬规则安全检查器 +静态扫描执行代码，检测危险操作 +""" + +import re +import ast +from typing import List, Tuple +from dataclasses import dataclass + + +@dataclass +class RuleCheckResult: + """规则检查结果""" + passed: bool + violations: List[str] # 违规项列表 + + +class RuleChecker: + """ + 硬规则检查器 + + 静态扫描代码，检测以下危险操作： + 1. 网络请求: requests, socket, urllib, http.client + 2. 危险文件操作: os.remove, shutil.rmtree, os.unlink + 3. 执行外部命令: subprocess, os.system, os.popen + 4. 访问非 workspace 路径 + """ + + # 禁止导入的模块 + FORBIDDEN_IMPORTS = { + 'requests', + 'socket', + 'urllib', + 'urllib.request', + 'urllib.parse', + 'urllib.error', + 'http.client', + 'httplib', + 'ftplib', + 'smtplib', + 'telnetlib', + 'subprocess', + } + + # 禁止调用的函数（模块.函数 或 单独函数名） + FORBIDDEN_CALLS = { + 'os.remove', + 'os.unlink', + 'os.rmdir', + 'os.removedirs', + 'os.system', + 'os.popen', + 'os.spawn', + 'os.spawnl', + 'os.spawnle', + 'os.spawnlp', + 'os.spawnlpe', + 'os.spawnv', + 'os.spawnve', + 'os.spawnvp', + 'os.spawnvpe', + 'os.exec', + 'os.execl', + 'os.execle', + 'os.execlp', + 'os.execlpe', + 'os.execv', + 'os.execve', + 'os.execvp', + 'os.execvpe', + 'shutil.rmtree', + 'shutil.move', # move 可能导致原文件丢失 + 'eval', + 'exec', + 'compile', + '__import__', + } + + # 危险路径模式（正则） + DANGEROUS_PATH_PATTERNS = [ + r'[A-Za-z]:\\', # Windows 绝对路径 + r'\\\\', # UNC 路径 + r'/etc/', + r'/usr/', + r'/bin/', + r'/home/', + r'/root/', + r'\.\./', # 父目录遍历 + r'\.\.', # 父目录 + ] + + def check(self, code: str) -> RuleCheckResult: + """ + 检查代码是否符合安全规则 + + Args: + code: Python 代码字符串 + + Returns: + RuleCheckResult: 检查结果 + """ + violations = [] + + # 1. 检查禁止的导入 + import_violations = self._check_imports(code) + violations.extend(import_violations) + + # 2. 检查禁止的函数调用 + call_violations = self._check_calls(code) + violations.extend(call_violations) + + # 3. 检查危险路径 + path_violations = self._check_paths(code) + violations.extend(path_violations) + + return RuleCheckResult( + passed=len(violations) == 0, + violations=violations + ) + + def _check_imports(self, code: str) -> List[str]: + """检查禁止的导入""" + violations = [] + + try: + tree = ast.parse(code) + + for node in ast.walk(tree): + if isinstance(node, ast.Import): + for alias in node.names: + module_name = alias.name.split('.')[0] + if alias.name in self.FORBIDDEN_IMPORTS or module_name in self.FORBIDDEN_IMPORTS: + violations.append(f"禁止导入模块: {alias.name}") + + elif isinstance(node, ast.ImportFrom): + if node.module: + module_name = node.module.split('.')[0] + if node.module in self.FORBIDDEN_IMPORTS or module_name in self.FORBIDDEN_IMPORTS: + violations.append(f"禁止导入模块: {node.module}") + + except SyntaxError: + # 如果代码有语法错误，使用正则匹配 + for module in self.FORBIDDEN_IMPORTS: + pattern = rf'\bimport\s+{re.escape(module)}\b|\bfrom\s+{re.escape(module)}\b' + if re.search(pattern, code): + violations.append(f"禁止导入模块: {module}") + + return violations + + def _check_calls(self, code: str) -> List[str]: + """检查禁止的函数调用""" + violations = [] + + try: + tree = ast.parse(code) + + for node in ast.walk(tree): + if isinstance(node, ast.Call): + call_name = self._get_call_name(node) + if call_name in self.FORBIDDEN_CALLS: + violations.append(f"禁止调用函数: {call_name}") + + except SyntaxError: + # 如果代码有语法错误，使用正则匹配 + for func in self.FORBIDDEN_CALLS: + pattern = rf'\b{re.escape(func)}\s*\(' + if re.search(pattern, code): + violations.append(f"禁止调用函数: {func}") + + return violations + + def _get_call_name(self, node: ast.Call) -> str: + """获取函数调用的完整名称""" + if isinstance(node.func, ast.Name): + return node.func.id + elif isinstance(node.func, ast.Attribute): + parts = [] + current = node.func + while isinstance(current, ast.Attribute): + parts.append(current.attr) + current = current.value + if isinstance(current, ast.Name): + parts.append(current.id) + return '.'.join(reversed(parts)) + return '' + + def _check_paths(self, code: str) -> List[str]: + """检查危险路径访问""" + violations = [] + + for pattern in self.DANGEROUS_PATH_PATTERNS: + matches = re.findall(pattern, code, re.IGNORECASE) + if matches: + # 排除 workspace 相关的合法路径 + for match in matches: + if 'workspace' not in code[max(0, code.find(match)-50):code.find(match)+50].lower(): + violations.append(f"检测到可疑路径模式: {match}") + break + + return violations + + +def check_code_safety(code: str) -> RuleCheckResult: + """便捷函数：检查代码安全性""" + checker = RuleChecker() + return checker.check(code) + diff --git a/ui/__init__.py b/ui/__init__.py new file mode 100644 index 0000000..03b9341 --- /dev/null +++ b/ui/__init__.py @@ -0,0 +1,2 @@ +# UI 模块 + diff --git a/ui/__pycache__/__init__.cpython-310.pyc b/ui/__pycache__/__init__.cpython-310.pyc new file mode 100644 index 0000000..96adfc5 Binary files /dev/null and b/ui/__pycache__/__init__.cpython-310.pyc differ diff --git a/ui/__pycache__/__init__.cpython-313.pyc b/ui/__pycache__/__init__.cpython-313.pyc new file mode 100644 index 0000000..c3c367b Binary files /dev/null and b/ui/__pycache__/__init__.cpython-313.pyc differ diff --git a/ui/__pycache__/chat_view.cpython-310.pyc b/ui/__pycache__/chat_view.cpython-310.pyc new file mode 100644 index 0000000..ad4db0a Binary files /dev/null and b/ui/__pycache__/chat_view.cpython-310.pyc differ diff --git a/ui/__pycache__/chat_view.cpython-313.pyc b/ui/__pycache__/chat_view.cpython-313.pyc new file mode 100644 index 0000000..dcec184 Binary files /dev/null and b/ui/__pycache__/chat_view.cpython-313.pyc differ diff --git a/ui/__pycache__/task_guide_view.cpython-310.pyc b/ui/__pycache__/task_guide_view.cpython-310.pyc new file mode 100644 index 0000000..e5ae1e3 Binary files /dev/null and b/ui/__pycache__/task_guide_view.cpython-310.pyc differ diff --git a/ui/__pycache__/task_guide_view.cpython-313.pyc b/ui/__pycache__/task_guide_view.cpython-313.pyc new file mode 100644 index 0000000..98063b7 Binary files /dev/null and b/ui/__pycache__/task_guide_view.cpython-313.pyc differ diff --git a/ui/chat_view.py b/ui/chat_view.py new file mode 100644 index 0000000..951de95 --- /dev/null +++ b/ui/chat_view.py @@ -0,0 +1,164 @@ +""" +聊天视图组件 +处理普通对话的 UI 展示 +""" + +import tkinter as tk +from tkinter import scrolledtext +from typing import Callable, Optional + + +class ChatView: + """ + 聊天视图 + + 包含： + - 消息显示区域 + - 输入框 + - 发送按钮 + """ + + def __init__( + self, + parent: tk.Widget, + on_send: Callable[[str], None] + ): + """ + 初始化聊天视图 + + Args: + parent: 父容器 + on_send: 发送消息回调函数 + """ + self.parent = parent + self.on_send = on_send + + self._create_widgets() + + def _create_widgets(self): + """创建 UI 组件""" + # 主框架 + self.frame = tk.Frame(self.parent, bg='#1e1e1e') + self.frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) + + # 标题 + title_label = tk.Label( + self.frame, + text="LocalAgent - 本地 AI 助手", + font=('Microsoft YaHei UI', 16, 'bold'), + fg='#61dafb', + bg='#1e1e1e' + ) + title_label.pack(pady=(0, 10)) + + # 消息显示区域 + self.message_area = scrolledtext.ScrolledText( + self.frame, + wrap=tk.WORD, + font=('Microsoft YaHei UI', 11), + bg='#2d2d2d', + fg='#d4d4d4', + insertbackground='white', + relief=tk.FLAT, + padx=10, + pady=10, + state=tk.DISABLED + ) + self.message_area.pack(fill=tk.BOTH, expand=True, pady=(0, 10)) + + # 配置消息标签样式 + self.message_area.tag_configure('user', foreground='#4fc3f7', font=('Microsoft YaHei UI', 11, 'bold')) + self.message_area.tag_configure('assistant', foreground='#81c784', font=('Microsoft YaHei UI', 11)) + self.message_area.tag_configure('system', foreground='#ffb74d', font=('Microsoft YaHei UI', 10, 'italic')) + self.message_area.tag_configure('error', foreground='#ef5350', font=('Microsoft YaHei UI', 10)) + + # 输入区域框架 + input_frame = tk.Frame(self.frame, bg='#1e1e1e') + input_frame.pack(fill=tk.X) + + # 输入框 + self.input_entry = tk.Entry( + input_frame, + font=('Microsoft YaHei UI', 12), + bg='#3c3c3c', + fg='#ffffff', + insertbackground='white', + relief=tk.FLAT + ) + self.input_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, ipady=8, padx=(0, 10)) + self.input_entry.bind('', self._on_enter_pressed) + + # 发送按钮 + self.send_button = tk.Button( + input_frame, + text="发送", + font=('Microsoft YaHei UI', 11, 'bold'), + bg='#0078d4', + fg='white', + activebackground='#106ebe', + activeforeground='white', + relief=tk.FLAT, + padx=20, + pady=5, + cursor='hand2', + command=self._on_send_clicked + ) + self.send_button.pack(side=tk.RIGHT) + + # 显示欢迎消息 + welcome_msg = ( + "欢迎使用 LocalAgent!\n" + "- 输入问题进行对话\n" + "- 输入文件处理需求（如\"复制文件\"、\"整理图片\"）将触发执行模式" + ) + self.add_message(welcome_msg, 'system') + + def _on_enter_pressed(self, event): + """回车键处理""" + self._on_send_clicked() + + def _on_send_clicked(self): + """发送按钮点击处理""" + text = self.input_entry.get().strip() + if text: + self.input_entry.delete(0, tk.END) + self.on_send(text) + + def add_message(self, message: str, tag: str = 'assistant'): + """ + 添加消息到显示区域 + + Args: + message: 消息内容 + tag: 消息类型 (user/assistant/system/error) + """ + self.message_area.config(state=tk.NORMAL) + + # 添加前缀 + prefix_map = { + 'user': '[你] ', + 'assistant': '[助手] ', + 'system': '[系统] ', + 'error': '[错误] ' + } + prefix = prefix_map.get(tag, '') + + self.message_area.insert(tk.END, "\n" + prefix + message + "\n", tag) + self.message_area.see(tk.END) + self.message_area.config(state=tk.DISABLED) + + def clear_messages(self): + """清空消息区域""" + self.message_area.config(state=tk.NORMAL) + self.message_area.delete(1.0, tk.END) + self.message_area.config(state=tk.DISABLED) + + def set_input_enabled(self, enabled: bool): + """设置输入区域是否可用""" + state = tk.NORMAL if enabled else tk.DISABLED + self.input_entry.config(state=state) + self.send_button.config(state=state) + + def get_frame(self) -> tk.Frame: + """获取主框架""" + return self.frame diff --git a/ui/task_guide_view.py b/ui/task_guide_view.py new file mode 100644 index 0000000..b69544d --- /dev/null +++ b/ui/task_guide_view.py @@ -0,0 +1,524 @@ +""" +任务引导视图组件 +执行任务的引导式 UI - 支持文件拖拽和 Markdown 渲染 +""" + +import tkinter as tk +from tkinter import scrolledtext, messagebox +from tkinter import ttk +from typing import Callable, Optional, List +from pathlib import Path +import shutil +import re + + +class MarkdownText(tk.Text): + """支持简单 Markdown 渲染的 Text 组件""" + + def __init__(self, parent, **kwargs): + super().__init__(parent, **kwargs) + self._setup_tags() + + def _setup_tags(self): + """设置 Markdown 样式标签""" + # 标题样式 + self.tag_configure('h1', font=('Microsoft YaHei UI', 14, 'bold'), foreground='#ffd54f', spacing1=10, spacing3=5) + self.tag_configure('h2', font=('Microsoft YaHei UI', 12, 'bold'), foreground='#81c784', spacing1=8, spacing3=4) + self.tag_configure('h3', font=('Microsoft YaHei UI', 11, 'bold'), foreground='#4fc3f7', spacing1=6, spacing3=3) + + # 列表样式 + self.tag_configure('bullet', foreground='#ce93d8', lmargin1=20, lmargin2=35) + self.tag_configure('numbered', foreground='#ce93d8', lmargin1=20, lmargin2=35) + + # 代码样式 + self.tag_configure('code', font=('Consolas', 10), background='#3c3c3c', foreground='#f8f8f2') + + # 粗体和斜体 + self.tag_configure('bold', font=('Microsoft YaHei UI', 10, 'bold')) + self.tag_configure('italic', font=('Microsoft YaHei UI', 10, 'italic')) + + # 普通文本 + self.tag_configure('normal', font=('Microsoft YaHei UI', 10), foreground='#d4d4d4') + + def set_markdown(self, text: str): + """设置 Markdown 内容并渲染""" + self.config(state=tk.NORMAL) + self.delete(1.0, tk.END) + + lines = text.split('\n') + for line in lines: + self._render_line(line) + + self.config(state=tk.DISABLED) + + def _render_line(self, line: str): + """渲染单行 Markdown""" + stripped = line.strip() + + # 标题 + if stripped.startswith('### '): + self.insert(tk.END, stripped[4:] + '\n', 'h3') + elif stripped.startswith('## '): + self.insert(tk.END, stripped[3:] + '\n', 'h2') + elif stripped.startswith('# '): + self.insert(tk.END, stripped[2:] + '\n', 'h1') + # 无序列表 + elif stripped.startswith('- ') or stripped.startswith('* '): + self.insert(tk.END, ' • ' + stripped[2:] + '\n', 'bullet') + # 有序列表 + elif re.match(r'^\d+\.\s', stripped): + match = re.match(r'^(\d+\.)\s(.*)$', stripped) + if match: + self.insert(tk.END, ' ' + match.group(1) + ' ' + match.group(2) + '\n', 'numbered') + # 普通文本 + else: + # 处理行内格式 + self._render_inline(line + '\n') + + def _render_inline(self, text: str): + """渲染行内 Markdown（粗体、斜体、代码）""" + # 简化处理：直接插入普通文本 + # 完整实现需要更复杂的解析 + self.insert(tk.END, text, 'normal') + + +class DropZone(tk.Frame): + """文件拖拽区域""" + + def __init__( + self, + parent, + title: str, + target_dir: Path, + is_input: bool = True, + **kwargs + ): + super().__init__(parent, **kwargs) + self.target_dir = target_dir + self.is_input = is_input + self.configure(bg='#2d2d2d', relief=tk.GROOVE, bd=2) + + # 确保目录存在 + self.target_dir.mkdir(parents=True, exist_ok=True) + + self._create_widgets(title) + self._setup_drag_drop() + self._refresh_file_list() + + def _create_widgets(self, title: str): + """创建组件""" + # 标题 + title_frame = tk.Frame(self, bg='#2d2d2d') + title_frame.pack(fill=tk.X, padx=5, pady=5) + + tk.Label( + title_frame, + text=title, + font=('Microsoft YaHei UI', 11, 'bold'), + fg='#4fc3f7' if self.is_input else '#81c784', + bg='#2d2d2d' + ).pack(side=tk.LEFT) + + # 打开文件夹按钮 + open_btn = tk.Button( + title_frame, + text="📂", + font=('Microsoft YaHei UI', 10), + bg='#424242', + fg='white', + relief=tk.FLAT, + cursor='hand2', + command=self._open_folder + ) + open_btn.pack(side=tk.RIGHT) + + # 刷新按钮 + refresh_btn = tk.Button( + title_frame, + text="🔄", + font=('Microsoft YaHei UI', 10), + bg='#424242', + fg='white', + relief=tk.FLAT, + cursor='hand2', + command=self._refresh_file_list + ) + refresh_btn.pack(side=tk.RIGHT, padx=(0, 5)) + + # 拖拽提示区域 + self.drop_label = tk.Label( + self, + text="将文件拖拽到此处\n或点击 📂 打开文件夹", + font=('Microsoft YaHei UI', 10), + fg='#888888', + bg='#3c3c3c', + relief=tk.SUNKEN, + padx=20, + pady=15 + ) + self.drop_label.pack(fill=tk.X, padx=5, pady=5) + + # 文件列表 + self.file_listbox = tk.Listbox( + self, + font=('Microsoft YaHei UI', 9), + bg='#2d2d2d', + fg='#d4d4d4', + selectbackground='#0078d4', + relief=tk.FLAT, + height=4 + ) + self.file_listbox.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + # 文件计数 + self.count_label = tk.Label( + self, + text="0 个文件", + font=('Microsoft YaHei UI', 9), + fg='#888888', + bg='#2d2d2d' + ) + self.count_label.pack(pady=(0, 5)) + + def _setup_drag_drop(self): + """设置拖拽功能（Windows 需要 windnd 库，这里用简化方案）""" + # 由于 Tkinter 原生不支持文件拖拽，使用点击打开文件夹的方式 + self.drop_label.bind('', lambda e: self._open_folder()) + + def _open_folder(self): + """打开目标文件夹""" + import os + os.startfile(str(self.target_dir)) + + def _refresh_file_list(self): + """刷新文件列表""" + self.file_listbox.delete(0, tk.END) + + files = list(self.target_dir.glob('*')) + files = [f for f in files if f.is_file()] + + for f in files: + self.file_listbox.insert(tk.END, f.name) + + self.count_label.config(text=f"{len(files)} 个文件") + + def get_files(self) -> List[Path]: + """获取目录中的文件列表""" + files = list(self.target_dir.glob('*')) + return [f for f in files if f.is_file()] + + def clear_files(self): + """清空目录中的文件""" + for f in self.target_dir.glob('*'): + if f.is_file(): + f.unlink() + self._refresh_file_list() + + +class TaskGuideView: + """ + 任务引导视图 + + 小白引导式界面，包含： + - 意图识别结果 + - 文件拖拽区域（输入/输出） + - 执行计划展示（Markdown 渲染） + - 风险提示 + - 执行按钮 + """ + + def __init__( + self, + parent: tk.Widget, + on_execute: Callable[[], None], + on_cancel: Callable[[], None], + workspace_path: Optional[Path] = None + ): + self.parent = parent + self.on_execute = on_execute + self.on_cancel = on_cancel + + if workspace_path: + self.workspace = workspace_path + else: + self.workspace = Path(__file__).parent.parent / "workspace" + + self.input_dir = self.workspace / "input" + self.output_dir = self.workspace / "output" + + self._create_widgets() + + def _create_widgets(self): + """创建 UI 组件""" + # 主框架 + self.frame = tk.Frame(self.parent, bg='#1e1e1e') + + # 标题 + title_label = tk.Label( + self.frame, + text="执行任务确认", + font=('Microsoft YaHei UI', 16, 'bold'), + fg='#ffd54f', + bg='#1e1e1e' + ) + title_label.pack(pady=(10, 15)) + + # 上半部分：文件区域 + file_section = tk.Frame(self.frame, bg='#1e1e1e') + file_section.pack(fill=tk.X, padx=10, pady=5) + + # 输入文件区域 + input_frame = tk.LabelFrame( + file_section, + text=" 📥 输入文件 ", + font=('Microsoft YaHei UI', 11, 'bold'), + fg='#4fc3f7', + bg='#1e1e1e', + relief=tk.GROOVE + ) + input_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(0, 5)) + + self.input_zone = DropZone( + input_frame, + title="待处理文件", + target_dir=self.input_dir, + is_input=True, + bg='#2d2d2d' + ) + self.input_zone.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + # 箭头 + arrow_frame = tk.Frame(file_section, bg='#1e1e1e') + arrow_frame.pack(side=tk.LEFT, padx=10) + tk.Label( + arrow_frame, + text="➡️", + font=('Microsoft YaHei UI', 20), + fg='#ffd54f', + bg='#1e1e1e' + ).pack(pady=30) + + # 输出文件区域 + output_frame = tk.LabelFrame( + file_section, + text=" 📤 输出文件 ", + font=('Microsoft YaHei UI', 11, 'bold'), + fg='#81c784', + bg='#1e1e1e', + relief=tk.GROOVE + ) + output_frame.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(5, 0)) + + self.output_zone = DropZone( + output_frame, + title="处理结果", + target_dir=self.output_dir, + is_input=False, + bg='#2d2d2d' + ) + self.output_zone.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + # 意图识别结果区域 + self._create_intent_section() + + # 执行计划区域（Markdown） + self._create_plan_section() + + # 风险提示区域 + self._create_risk_section() + + # 按钮区域 + self._create_button_section() + + def _create_intent_section(self): + """创建意图识别结果区域""" + section = tk.LabelFrame( + self.frame, + text=" 🎯 意图识别 ", + font=('Microsoft YaHei UI', 11, 'bold'), + fg='#81c784', + bg='#1e1e1e', + relief=tk.GROOVE + ) + section.pack(fill=tk.X, padx=10, pady=5) + + self.intent_label = tk.Label( + section, + text="", + font=('Microsoft YaHei UI', 10), + fg='#d4d4d4', + bg='#1e1e1e', + wraplength=650, + justify=tk.LEFT + ) + self.intent_label.pack(padx=10, pady=8, anchor=tk.W) + + def _create_plan_section(self): + """创建执行计划区域（支持 Markdown）""" + section = tk.LabelFrame( + self.frame, + text=" 📄 执行计划 ", + font=('Microsoft YaHei UI', 11, 'bold'), + fg='#ce93d8', + bg='#1e1e1e', + relief=tk.GROOVE + ) + section.pack(fill=tk.BOTH, expand=True, padx=10, pady=5) + + # 使用 Markdown 渲染的 Text + self.plan_text = MarkdownText( + section, + wrap=tk.WORD, + bg='#2d2d2d', + fg='#d4d4d4', + relief=tk.FLAT, + height=8, + padx=10, + pady=10 + ) + + # 添加滚动条 + scrollbar = ttk.Scrollbar(section, orient=tk.VERTICAL, command=self.plan_text.yview) + self.plan_text.configure(yscrollcommand=scrollbar.set) + + scrollbar.pack(side=tk.RIGHT, fill=tk.Y) + self.plan_text.pack(fill=tk.BOTH, expand=True, padx=5, pady=5) + + def _create_risk_section(self): + """创建风险提示区域""" + section = tk.LabelFrame( + self.frame, + text=" ⚠️ 安全提示 ", + font=('Microsoft YaHei UI', 11, 'bold'), + fg='#ffb74d', + bg='#1e1e1e', + relief=tk.GROOVE + ) + section.pack(fill=tk.X, padx=10, pady=5) + + self.risk_label = tk.Label( + section, + text="• 所有操作仅在 workspace 目录内进行\n• 原始文件不会被修改或删除\n• 执行代码已通过安全检查", + font=('Microsoft YaHei UI', 10), + fg='#d4d4d4', + bg='#1e1e1e', + justify=tk.LEFT + ) + self.risk_label.pack(padx=10, pady=8, anchor=tk.W) + + def _create_button_section(self): + """创建按钮区域""" + button_frame = tk.Frame(self.frame, bg='#1e1e1e') + button_frame.pack(fill=tk.X, padx=10, pady=15) + + # 刷新文件列表按钮 + self.refresh_btn = tk.Button( + button_frame, + text="🔄 刷新文件", + font=('Microsoft YaHei UI', 10), + bg='#424242', + fg='white', + activebackground='#616161', + activeforeground='white', + relief=tk.FLAT, + padx=15, + pady=5, + cursor='hand2', + command=self._refresh_all + ) + self.refresh_btn.pack(side=tk.LEFT, padx=(0, 10)) + + # 取消按钮 + self.cancel_btn = tk.Button( + button_frame, + text="取消", + font=('Microsoft YaHei UI', 11), + bg='#616161', + fg='white', + activebackground='#757575', + activeforeground='white', + relief=tk.FLAT, + padx=20, + pady=5, + cursor='hand2', + command=self.on_cancel + ) + self.cancel_btn.pack(side=tk.RIGHT, padx=(10, 0)) + + # 执行按钮 + self.execute_btn = tk.Button( + button_frame, + text="🚀 开始执行", + font=('Microsoft YaHei UI', 12, 'bold'), + bg='#4caf50', + fg='white', + activebackground='#66bb6a', + activeforeground='white', + relief=tk.FLAT, + padx=30, + pady=8, + cursor='hand2', + command=self._on_execute_clicked + ) + self.execute_btn.pack(side=tk.RIGHT) + + def _refresh_all(self): + """刷新所有文件列表""" + self.input_zone._refresh_file_list() + self.output_zone._refresh_file_list() + + def _on_execute_clicked(self): + """执行按钮点击""" + # 刷新文件列表 + self.input_zone._refresh_file_list() + + # 检查 input 目录是否有文件 + files = self.input_zone.get_files() + + if not files: + result = messagebox.askyesno( + "确认执行", + "输入文件夹为空，确定要继续执行吗？", + icon='warning' + ) + if not result: + return + + self.on_execute() + + def set_intent_result(self, reason: str, confidence: float): + """设置意图识别结果""" + self.intent_label.config( + text=f"识别结果: 执行任务 (置信度: {confidence:.0%})\n原因: {reason}" + ) + + def set_execution_plan(self, plan: str): + """设置执行计划（Markdown 格式）""" + self.plan_text.set_markdown(plan) + + def set_risk_info(self, info: str): + """设置风险提示""" + self.risk_label.config(text=info) + + def set_buttons_enabled(self, enabled: bool): + """设置按钮是否可用""" + state = tk.NORMAL if enabled else tk.DISABLED + self.execute_btn.config(state=state) + self.cancel_btn.config(state=state) + + def refresh_output(self): + """刷新输出文件列表""" + self.output_zone._refresh_file_list() + + def show(self): + """显示视图""" + self.frame.pack(fill=tk.BOTH, expand=True, padx=10, pady=10) + self._refresh_all() + + def hide(self): + """隐藏视图""" + self.frame.pack_forget() + + def get_frame(self) -> tk.Frame: + """获取主框架""" + return self.frame