From d3d85c804ca769a21fbe02a449a7fbec5df656db Mon Sep 17 00:00:00 2001 From: Val-sss <154882199@qq.com> Date: Wed, 6 May 2026 11:54:18 +0800 Subject: [PATCH] feat: P2 watchdog + gate accuracy report; remove .bak file - P2-1: Task-level watchdog (300s timeout via threading.Timer, auto-abort stuck tasks) - P2-2: Gate routing accuracy report (/stats command, per-expert_type success rate) - Remove vision_expert.py.bak (keep vision_expert.py as intended feature) Co-Authored-By: Claude Opus 4.6 (1M context) --- kaiwu/cli/main.py | 25 ++ kaiwu/core/orchestrator.py | 18 + kaiwu/experts/vision_expert.py.bak | 546 ----------------------------- kaiwu/stats/value_tracker.py | 34 ++ 4 files changed, 77 insertions(+), 546 deletions(-) delete mode 100644 kaiwu/experts/vision_expert.py.bak diff --git a/kaiwu/cli/main.py b/kaiwu/cli/main.py index 7415003..5ff0892 100644 --- a/kaiwu/cli/main.py +++ b/kaiwu/cli/main.py @@ -441,6 +441,7 @@ REPL_COMMANDS = { "/api": "API配置 (用法: /api show | /api temp | /api default )", "/paste": "从剪贴板粘贴图片", "/image": "添加图片文件 (用法: /image )", + "/stats": "查看任务统计和Gate准确率", "/exit": "退出", } @@ -712,6 +713,30 @@ def _repl(model_path, ollama_url, ollama_model, project_root, verbose, no_search kws = ", ".join(e["trigger_keywords"][:4]) console.print(f" [bold]{e['name']}[/bold] [{lc}] tasks={cnt} sr={sr:.0%} kw=[{kws}]") + elif cmd == "/stats": + from kaiwu.stats.value_tracker import ValueTracker + tracker = ValueTracker() + summary = tracker.get_summary(days=30) + gate_acc = tracker.get_gate_accuracy(days=30) + + console.print("\n [bold]任务统计（近30天）[/bold]") + console.print(f" 总任务: {summary['total_tasks']} | 成功: {summary['succeeded_tasks']} | " + f"成功率: {summary['succeeded_tasks']/max(summary['total_tasks'],1)*100:.0f}%") + console.print(f" 节省时间: ~{summary['time_saved_hours']}h | 累计: {summary['total_all_time']} 个任务") + + if gate_acc: + console.print("\n [bold]Gate路由准确率[/bold]") + console.print(" [dim]类型 总数 成功率 平均耗时 平均重试[/dim]") + for g in gate_acc: + sr = f"{g['success_rate']*100:.0f}%" + console.print( + f" {g['expert_type']:14s} {g['total']:4d} {sr:>5s} " + f"{g['avg_elapsed']:5.1f}s {g['avg_retries']:.1f}次" + ) + else: + console.print("\n [dim]暂无Gate路由数据（需要执行几个任务后才有统计）[/dim]") + console.print() + elif cmd == "/plan": if arg: pending_images = list(getattr(session, '_pending_images', [])) diff --git a/kaiwu/core/orchestrator.py b/kaiwu/core/orchestrator.py index 5169a7c..dede967 100644 --- a/kaiwu/core/orchestrator.py +++ b/kaiwu/core/orchestrator.py @@ -6,6 +6,7 @@ RED-5: Max 3 retries, hardcoded. import logging import time +import threading from typing import Optional from kaiwu.core.context import TaskContext @@ -97,6 +98,17 @@ class PipelineOrchestrator: """ start_time = time.time() + # ── Watchdog: task-level timeout (P2) ── + TASK_TIMEOUT_S = 300 # 5 minutes max per task + _watchdog_triggered = threading.Event() + + def _watchdog_timer(): + _watchdog_triggered.set() + + _watchdog = threading.Timer(TASK_TIMEOUT_S, _watchdog_timer) + _watchdog.daemon = True + _watchdog.start() + # Store project_root for Gate 2 backtest use self._backtest_project_root = project_root @@ -253,6 +265,11 @@ class PipelineOrchestrator: f"任务分类置信度较低({confidence:.0%})，减少重试次数") while ctx.retry_count < max_retries: + # Watchdog check: abort if task exceeded timeout + if _watchdog_triggered.is_set(): + self._emit(on_status, "watchdog", f"任务超时({TASK_TIMEOUT_S}s)，强制终止") + break + success = self._run_sequence(sequence, ctx, on_status) # Notify locator of task result (graph stats + incremental update) @@ -373,6 +390,7 @@ class PipelineOrchestrator: # Clear ephemeral debug info to prevent context pollution ctx.debug_info = "" + _watchdog.cancel() # Clean up watchdog timer elapsed = time.time() - start_time # ── Checkpoint: restore on failure ── diff --git a/kaiwu/experts/vision_expert.py.bak b/kaiwu/experts/vision_expert.py.bak deleted file mode 100644 index d79e1ff..0000000 --- a/kaiwu/experts/vision_expert.py.bak +++ /dev/null @@ -1,546 +0,0 @@ -""" -VisionExpert: 多模态图片处理专家 -支持图片上传、剪贴板粘贴、图片分析和基于图片的代码生成。 - -Pipeline: 图片输入 → 图片分析 → 任务路由 → 代码生成/图片描述 -""" - -import base64 -import logging -import os -import shlex -import tempfile -from pathlib import Path -from typing import Optional - -from kaiwu.core.context import TaskContext - -logger = logging.getLogger(__name__) -MAX_IMAGE_BYTES = 20 * 1024 * 1024 -MAX_IMAGE_COUNT = 20 - -# ── Vision system prompts ────────────────────────────────────────── - -VISION_ANALYSIS_SYSTEM = """\ -你是KWCode的多模态视觉专家。用户上传了一张图片，请根据图片内容提供有用的分析。 - -## 分析类型 -1. **代码截图分析**：如果图片包含代码、错误信息、终端输出 - - 识别代码语言和功能 - - 分析错误信息并提供修复建议 - - 如果用户有相关任务，提供代码修改建议 - -2. **UI/设计图分析**：如果图片是UI设计、网页截图、界面原型 - - 描述布局和设计元素 - - 提供实现建议（HTML/CSS/框架） - - 识别可复用的组件 - -3. **文档/表格分析**：如果图片是文档、表格、图表 - - 提取关键信息 - - 结构化数据 - - 提供数据处理建议 - -4. **通用图片分析**：其他类型图片 - - 详细描述图片内容 - - 识别关键元素 - - 提供相关建议 - -## 输出格式 -- 简洁明了，2-5句话 -- 如果是代码相关，提供具体的修改建议 -- 如果是UI设计，提供实现方向 -- 使用中文回复""" - -VISION_CODEGEN_SYSTEM = """\ -你是KWCode的多模态代码生成专家。用户上传了图片并要求生成代码。 - -## 代码生成规则 -1. **UI截图 → HTML/CSS** - - 使用Tailwind CSS - - 响应式设计 - - 保持视觉一致性 - -2. **错误截图 → 修复代码** - - 分析错误信息 - - 定位问题根源 - - 提供最小化修复 - -3. **设计图 → 组件代码** - - 选择合适的框架（React/Vue/原生） - - 模块化组件设计 - - 可复用性优先 - -## 输出要求 -- 只输出完整可执行的代码 -- 不要解释，不要markdown代码块标记 -- 代码末尾添加使用说明注释""" - - -class VisionExpert: - """多模态图片处理专家：分析图片内容，生成代码或提供分析""" - - def __init__(self, llm, tool_executor=None): - self.llm = llm - self.tools = tool_executor - self._temp_dir = Path(tempfile.mkdtemp(prefix="kwcode_vision_")) - - def run(self, ctx: TaskContext) -> dict: - """ - 处理图片输入任务 - - Args: - ctx: 任务上下文，包含 user_input 和 image_path - - Returns: - dict: 包含 success, output, metadata - """ - image_paths = list(getattr(ctx, 'image_paths', []) or []) - image_path = getattr(ctx, 'image_path', None) - if not image_paths and image_path: - image_paths = [image_path] - - if not image_paths: - return { - "success": False, - "output": "错误：未提供图片路径", - "metadata": {"error": "no_image_path"} - } - if len(image_paths) > MAX_IMAGE_COUNT: - return { - "success": False, - "output": f"错误：一次最多支持 {MAX_IMAGE_COUNT} 张图片", - "metadata": {"error": "too_many_images", "count": len(image_paths)} - } - - # 验证图片文件 - invalid_paths = [path for path in image_paths if not self._validate_image(path)] - if invalid_paths: - return { - "success": False, - "output": f"错误：图片文件不存在、过大或格式不支持: {', '.join(invalid_paths)}", - "metadata": {"error": "invalid_image", "paths": invalid_paths} - } - - # 分析用户意图 - user_input = ctx.user_input.strip() - is_codegen_task = self._is_codegen_task(user_input) - - try: - images = [] - for path in image_paths: - image_base64 = self._encode_image(path) - images.append({ - "path": path, - "base64": image_base64, - "media_type": self._media_type_for_path(path, image_base64), - }) - - if is_codegen_task: - return self._run_codegen(ctx, images) - else: - return self._run_analysis(ctx, images) - - except Exception as e: - logger.error(f"VisionExpert error: {e}") - return { - "success": False, - "output": f"图片处理失败: {str(e)}", - "metadata": {"error": str(e)} - } - - def _validate_image(self, image_path: str) -> bool: - """验证图片文件是否存在且格式支持""" - path = Path(image_path).expanduser() - if not path.exists() or not path.is_file(): - return False - if path.stat().st_size > MAX_IMAGE_BYTES: - return False - - try: - with path.open("rb") as f: - return self._media_type_for_bytes(f.read(16)) is not None - except OSError: - return False - - def _encode_image(self, image_path: str) -> str: - """将图片编码为base64""" - with open(Path(image_path).expanduser(), "rb") as f: - return base64.b64encode(f.read()).decode('utf-8') - - def _media_type_for_path(self, image_path: str, image_base64: str) -> str: - """Return a MIME type compatible with common vision APIs.""" - media_type = self._media_type_for_base64(image_base64) - if media_type: - return media_type - - suffix_map = { - ".png": "image/png", - ".jpg": "image/jpeg", - ".jpeg": "image/jpeg", - ".gif": "image/gif", - ".webp": "image/webp", - ".bmp": "image/bmp", - } - suffix = Path(image_path).suffix.lower() - if suffix in suffix_map: - return suffix_map[suffix] - - return "image/png" - - @staticmethod - def _media_type_for_base64(image_base64: str) -> Optional[str]: - raw_sample = image_base64[:20] - if raw_sample.startswith("iVBORw0KGgo"): - return "image/png" - if raw_sample.startswith("/9j"): - return "image/jpeg" - if raw_sample.startswith("R0lGOD"): - return "image/gif" - if raw_sample.startswith("UklGR"): - return "image/webp" - if raw_sample.startswith("Qk"): - return "image/bmp" - return None - - @staticmethod - def _media_type_for_bytes(raw: bytes) -> Optional[str]: - if raw.startswith(b"\x89PNG\r\n\x1a\n"): - return "image/png" - if raw.startswith(b"\xff\xd8\xff"): - return "image/jpeg" - if raw.startswith((b"GIF87a", b"GIF89a")): - return "image/gif" - if raw.startswith(b"RIFF") and raw[8:12] == b"WEBP": - return "image/webp" - if raw.startswith(b"BM"): - return "image/bmp" - return None - - def _is_codegen_task(self, user_input: str) -> bool: - """判断是否为代码生成任务""" - codegen_keywords = [ - "生成代码", "写代码", "实现", "创建", "生成", - "代码", "函数", "类", "组件", "页面", - "generate", "create", "implement", "code", - "html", "css", "javascript", "python", "react", "vue" - ] - - # 检查是否包含代码相关关键词 - user_input_lower = user_input.lower() - return any(kw in user_input_lower for kw in codegen_keywords) - - def _run_analysis(self, ctx: TaskContext, images: list[dict]) -> dict: - """运行图片分析""" - logger.info("[vision] 分析图片: %s", ", ".join(img["path"] for img in images)) - - # 构建提示词 - user_input = ctx.user_input.strip() - image_count = len(images) - if user_input: - prompt = f"用户上传了{image_count}张图片并说：{user_input}\n\n请结合所有图片进行分析。" - else: - prompt = f"用户上传了{image_count}张图片，请分析其内容。" - - # 调用Vision LLM - response = self._call_vision_llm( - system_prompt=VISION_ANALYSIS_SYSTEM, - user_prompt=prompt, - images=images, - ) - - return { - "success": True, - "output": response, - "metadata": { - "type": "vision_analysis", - "image_paths": [img["path"] for img in images], - "image_count": image_count, - "has_user_task": bool(user_input) - } - } - - def _run_codegen(self, ctx: TaskContext, images: list[dict]) -> dict: - """运行基于图片的代码生成""" - logger.info("[vision] 基于图片生成代码: %s", ", ".join(img["path"] for img in images)) - - user_input = ctx.user_input.strip() - image_count = len(images) - - # 构建提示词 - prompt = f"用户上传了{image_count}张图片并要求：{user_input}\n\n请结合所有图片内容生成代码。" - - # 调用Vision LLM - response = self._call_vision_llm( - system_prompt=VISION_CODEGEN_SYSTEM, - user_prompt=prompt, - images=images, - ) - - # 尝试执行生成的代码（如果用户要求） - if self.tools and self._should_execute_code(user_input): - execution_result = self._execute_generated_code(response) - if execution_result: - response += f"\n\n--- 执行结果 ---\n{execution_result}" - - return { - "success": True, - "output": response, - "metadata": { - "type": "vision_codegen", - "image_paths": [img["path"] for img in images], - "image_count": image_count, - "task": user_input - } - } - - def _call_vision_llm(self, system_prompt: str, user_prompt: str, images: list[dict]) -> str: - """调用支持Vision的LLM (Anthropic Messages API 格式) - - 优先使用 self.llm (如果支持 vision)，否则回退到环境变量配置的 API： - KWCODE_VISION_API_URL - API endpoint (Anthropic Messages API 格式，必填) - KWCODE_VISION_API_KEY - API key (必填) - KWCODE_VISION_MODEL - 模型名 (必填，需支持多模态，如 mimo-v2-omni) - """ - # 尝试通过 self.llm 直接调用（如果后端支持多模态） - if self.llm is not None: - try: - return self._try_llm_vision(system_prompt, user_prompt, images) - except Exception as e: - logger.debug(f"[vision] self.llm 不支持 vision，回退到 API: {e}") - - if not self._vision_api_configured(): - raise RuntimeError( - "本地模型不支持图片输入，且未显式配置Vision API；" - "请设置 KWCODE_VISION_API_URL 或 KWCODE_VISION_API_KEY 后重试" - ) - - # 回退：直接调用 Anthropic Messages API - return self._call_anthropic_vision(system_prompt, user_prompt, images) - - @staticmethod - def _vision_api_configured() -> bool: - return bool(os.environ.get("KWCODE_VISION_API_URL") or os.environ.get("KWCODE_VISION_API_KEY")) - - def _try_llm_vision(self, system_prompt: str, user_prompt: str, images: list[dict]) -> str: - """尝试通过 self.llm 的 chat 接口发送多模态请求""" - # LLMBackend supports two HTTP styles. Native llama.cpp cannot consume - # image payloads here, so force the documented vision API fallback. - mode = getattr(self.llm, "_mode", None) - is_openai_compat = getattr(self.llm, "_is_openai_compat", False) - if mode and mode != "ollama": - raise AttributeError("current LLM backend does not support image chat payloads") - - if mode == "ollama" and not is_openai_compat: - messages = [ - {"role": "system", "content": system_prompt}, - { - "role": "user", - "content": user_prompt, - "images": [img["base64"] for img in images], - }, - ] - return self.llm.chat(messages, max_tokens=2048) - - content = [{"type": "text", "text": user_prompt}] - content.extend( - { - "type": "image_url", - "image_url": {"url": f"data:{img['media_type']};base64,{img['base64']}"}, - } - for img in images - ) - messages = [ - {"role": "system", "content": system_prompt}, - { - "role": "user", - "content": content, - }, - ] - return self.llm.chat(messages, max_tokens=2048) - - def _call_anthropic_vision(self, system_prompt: str, user_prompt: str, images: list[dict]) -> str: - """直接调用 Anthropic Messages API（兼容 xiaomimimo 等代理）""" - import json as _json - import urllib.request - import urllib.error - - api_url = os.environ.get("KWCODE_VISION_API_URL", "") - api_key = os.environ.get("KWCODE_VISION_API_KEY", "") - model = os.environ.get("KWCODE_VISION_MODEL", "") - - if not api_url or not model: - raise RuntimeError( - "Vision API 未配置。请设置环境变量：\n" - " export KWCODE_VISION_API_URL=\n" - " export KWCODE_VISION_API_KEY=\n" - " export KWCODE_VISION_MODEL=\n" - "示例：\n" - " export KWCODE_VISION_API_URL=https://your-provider.com/v1/messages\n" - " export KWCODE_VISION_API_KEY=sk-xxx\n" - " export KWCODE_VISION_MODEL=gpt-4o" - ) - - content = [ - { - "type": "image", - "source": { - "type": "base64", - "media_type": img["media_type"], - "data": img["base64"], - }, - } - for img in images - ] - content.append({"type": "text", "text": user_prompt}) - - payload = { - "model": model, - "max_tokens": 2048, - "system": system_prompt, - "messages": [ - { - "role": "user", - "content": content, - } - ], - } - - headers = { - "Content-Type": "application/json", - "anthropic-version": "2023-06-01", - } - if api_key: - headers["x-api-key"] = api_key - - logger.info(f"[vision] 调用 {api_url} model={model}") - data_bytes = _json.dumps(payload).encode("utf-8") - req = urllib.request.Request(api_url, data=data_bytes, headers=headers, method="POST") - - try: - with urllib.request.urlopen(req, timeout=90) as resp: - result = _json.loads(resp.read().decode("utf-8")) - except urllib.error.HTTPError as e: - body = e.read().decode("utf-8", errors="replace")[:500] - raise RuntimeError(f"Vision API HTTP {e.code}: {body}") from e - except urllib.error.URLError as e: - raise RuntimeError(f"Vision API connection failed: {e.reason}") from e - except TimeoutError as e: - raise RuntimeError("Vision API request timed out") from e - - # 提取文本 - # 检查 API 级别错误 - if result.get("type") == "error" or result.get("isError"): - error_msg = result.get("error", {}).get("message", str(result)) - raise RuntimeError(f"Vision API error: {error_msg}") - - stop_reason = result.get("stop_reason", "") - if stop_reason == "max_tokens": - logger.warning("[vision] 输出被截断 (max_tokens reached)") - - text_parts = [] - for block in result.get("content", []): - if block.get("type") == "text": - text_parts.append(block["text"]) - - usage = result.get("usage", {}) - logger.info( - f"[vision] 完成: {usage.get('input_tokens', '?')} in / " - f"{usage.get('output_tokens', '?')} out tokens" - ) - return "\n".join(text_parts) if text_parts else "[VisionExpert] 模型未返回文本内容" - - def _should_execute_code(self, user_input: str) -> bool: - """判断是否应该执行生成的代码""" - execute_keywords = ["执行", "运行", "测试", "run", "execute", "test"] - return any(kw in user_input.lower() for kw in execute_keywords) - - def _execute_generated_code(self, code: str) -> Optional[str]: - """执行生成的代码""" - if not self.tools: - return None - - try: - # 保存代码到临时文件 - temp_file = self._temp_dir / "generated_code.py" - temp_file.write_text(code, encoding='utf-8') - - # 执行代码 - stdout, stderr, returncode = self.tools.run_bash(f"python {shlex.quote(str(temp_file))}") - output = stdout.strip() - error = stderr.strip() - if returncode != 0: - return f"退出码 {returncode}\n{error or output}".strip() - return output or error - - except Exception as e: - return f"执行失败: {str(e)}" - - def cleanup(self): - """清理临时文件""" - import shutil - if self._temp_dir.exists(): - shutil.rmtree(self._temp_dir) - - -def save_clipboard_image() -> Optional[str]: - """ - 从剪贴板保存图片 - - Returns: - str: 保存的图片路径，如果剪贴板没有图片则返回None - """ - try: - from PIL import Image, ImageGrab - - image = ImageGrab.grabclipboard() - if isinstance(image, Image.Image): - # 生成临时文件路径 - temp_dir = Path(tempfile.mkdtemp(prefix="kwcode_clipboard_")) - temp_path = temp_dir / "clipboard_image.png" - - # 保存图片 - image.save(temp_path, "PNG") - logger.info(f"[vision] 剪贴板图片已保存: {temp_path}") - - return str(temp_path) - else: - logger.debug("[vision] 剪贴板中没有图片") - return None - - except ImportError: - logger.warning("[vision] Pillow未安装，无法处理剪贴板图片") - return None - except Exception as e: - logger.error(f"[vision] 处理剪贴板图片失败: {e}") - return None - - -def validate_image_path(path: str) -> bool: - """验证图片路径是否有效""" - path_obj = Path(path).expanduser() - if not path_obj.exists() or not path_obj.is_file(): - return False - if path_obj.stat().st_size > MAX_IMAGE_BYTES: - return False - - try: - with path_obj.open("rb") as f: - return VisionExpert._media_type_for_bytes(f.read(16)) is not None - except OSError: - return False - - -def get_image_info(image_path: str) -> dict: - """获取图片基本信息""" - try: - from PIL import Image - - with Image.open(image_path) as img: - return { - "format": img.format, - "size": img.size, - "mode": img.mode, - "file_size": os.path.getsize(image_path) - } - except Exception as e: - return {"error": str(e)} diff --git a/kaiwu/stats/value_tracker.py b/kaiwu/stats/value_tracker.py index 6a1a442..22a682c 100644 --- a/kaiwu/stats/value_tracker.py +++ b/kaiwu/stats/value_tracker.py @@ -131,3 +131,37 @@ class ValueTracker: return row["c"] or 0 except Exception: return 0 + + def get_gate_accuracy(self, days: int = 30) -> list[dict]: + """ + Gate routing accuracy report: success rate per expert_type. + P2: Shows which Gate classifications lead to successful outcomes. + """ + since = (datetime.now() - timedelta(days=days)).isoformat() + try: + with self._get_conn() as conn: + rows = conn.execute(""" + SELECT expert_type, + COUNT(*) as total, + SUM(success) as succeeded, + AVG(elapsed_s) as avg_elapsed, + AVG(retry_count) as avg_retries + FROM task_stats + WHERE timestamp > ? AND expert_type != '' + GROUP BY expert_type + ORDER BY total DESC + """, (since,)).fetchall() + + return [ + { + "expert_type": row["expert_type"], + "total": row["total"], + "success_rate": (row["succeeded"] or 0) / row["total"] if row["total"] > 0 else 0, + "avg_elapsed": round(row["avg_elapsed"] or 0, 1), + "avg_retries": round(row["avg_retries"] or 0, 1), + } + for row in rows + ] + except Exception as e: + logger.debug("[value_tracker] get_gate_accuracy failed: %s", e) + return []