diff --git a/.gitignore b/.gitignore index 60862a4..6ff0960 100644 --- a/.gitignore +++ b/.gitignore @@ -7,3 +7,4 @@ build/ .pytest_cache/ test_project/ .eggs/ +.DS_Store diff --git a/README.md b/README.md index 9ac3c1f..e9d8034 100644 --- a/README.md +++ b/README.md @@ -211,6 +211,14 @@ Prompt Optimizer（可选，需 Anthropic API key）： ### Office 文档 - Excel / PPT / Word 生成 +### 多模态图片处理 +- `/paste` 命令：从剪贴板粘贴图片 +- `/image ` 命令：添加图片文件 +- 图片分析：代码截图、UI设计图、文档表格 +- 基于图片的代码生成：UI截图→HTML/CSS、错误截图→修复代码 +- 支持格式：PNG、JPG、JPEG、GIF、WebP、BMP +- 安装：`pip install kwcode[multimodal]` + ### 价值可见 - `kwcode stats`：完成任务数、节省时间估算 - 飞轮通知：专家投产时弹出 @@ -355,6 +363,8 @@ kwcode --plan "重构数据库连接层" /memory 查看项目记忆 /init 初始化项目规则文件 /cd <路径> 切换项目目录 +/paste 从剪贴板粘贴图片 +/image <路径> 添加图片文件 /help 显示帮助 ``` @@ -418,7 +428,8 @@ kaiwu/ │ ├── verifier.py # [元专家] 语法检查 + pytest │ ├── debug_subagent.py # [元专家] 运行时调试（sys.settrace） │ ├── reviewer.py # [元专家] 需求对齐审查 -│ └── search_augmentor.py # 搜索增强 + BM25 + CE 重排 +│ ├── search_augmentor.py # 搜索增强 + BM25 + CE 重排 +│ └── vision_expert.py # [元专家] 多模态图片处理 ├── search/ │ ├── reranker.py # Cross-Encoder 可选重排 │ ├── duckduckgo.py # SearXNG + DDG 并行搜索 diff --git a/docs/kwcode_architecture.png b/docs/kwcode_architecture.png new file mode 100644 index 0000000..7283f5d Binary files /dev/null and b/docs/kwcode_architecture.png differ diff --git a/kaiwu/cli/main.py b/kaiwu/cli/main.py index 962f83a..3e77d4f 100644 --- a/kaiwu/cli/main.py +++ b/kaiwu/cli/main.py @@ -161,6 +161,10 @@ def _build_pipeline(model_path, ollama_url, ollama_model, project_root, verbose) from kaiwu.experts.chat_expert import ChatExpert chat_expert = ChatExpert(llm=llm, search_augmentor=search) + # Vision Expert (多模态图片处理) + from kaiwu.experts.vision_expert import VisionExpert + vision_expert = VisionExpert(llm=llm, tool_executor=tools) + # Debug Subagent (问题1修复：实例化并注入) from kaiwu.experts.debug_subagent import DebugSubagent debug_subagent = DebugSubagent(llm, tools) @@ -183,6 +187,7 @@ def _build_pipeline(model_path, ollama_url, ollama_model, project_root, verbose) ab_tester=ab_tester, chat_expert=chat_expert, debug_subagent=debug_subagent, + vision_expert=vision_expert, ) # Wire circular reference: ABTester needs orchestrator for backtest ab_tester.orchestrator = orchestrator @@ -192,17 +197,26 @@ def _build_pipeline(model_path, ollama_url, ollama_model, project_root, verbose) # ── Single task execution ───────────────────────────────────── -def _run_task(task, gate, orchestrator, memory, project_root, verbose, plan=False, no_search=False): +def _run_task(task, gate, orchestrator, memory, project_root, verbose, plan=False, no_search=False, image_paths=None): """Execute a single task through the pipeline. Returns success bool.""" from kaiwu.core.orchestrator import EXPERT_SEQUENCES from rich.progress import Progress, SpinnerColumn, TextColumn + # 处理图片上下文 + if image_paths: + logger.info(f"[main] 任务包含 {len(image_paths)} 张图片") + # 将图片路径添加到任务描述中 + image_context = "\n".join([f"[图片: {img}]" for img in image_paths]) + task_with_images = f"{task}\n\n{image_context}" + else: + task_with_images = task + # Gate (with spinner) with Progress(SpinnerColumn(), TextColumn("{task.description}"), transient=True, console=console) as progress: spin = progress.add_task("分析任务...", total=None) try: - gate_result = gate.classify(task, memory_context=memory.load(project_root)) + gate_result = gate.classify(task_with_images, memory_context=memory.load(project_root)) except Exception as e: progress.stop() console.print(f"\n [red]模型调用失败：{e}[/red]") @@ -298,12 +312,13 @@ def _run_task(task, gate, orchestrator, memory, project_root, verbose, plan=Fals logger.debug("[main] 预搜索失败: %s", e) result = orchestrator.run( - user_input=task, + user_input=task_with_images, gate_result=gate_result, project_root=project_root, on_status=_status_fn, no_search=no_search, pre_search_results=pre_search, + image_paths=image_paths, ) # 保存最后结果供conversation_history使用（问题7） orchestrator._last_result = result @@ -393,6 +408,8 @@ REPL_COMMANDS = { "/plan": "计划模式 (用法: /plan <任务> 或 /plan 后输入任务)", "/multi": "多任务模式 (用法: /multi 后按提示输入多个任务)", "/api": "API配置 (用法: /api show | /api temp | /api default )", + "/paste": "从剪贴板粘贴图片", + "/image": "添加图片文件 (用法: /image )", "/exit": "退出", } @@ -642,6 +659,35 @@ def _repl(model_path, ollama_url, ollama_model, project_root, verbose): elif cmd == "/multi": _handle_multi_command(arg, gate, orchestrator, project_root, console) + elif cmd == "/paste": + from kaiwu.experts.vision_expert import save_clipboard_image + image_path = save_clipboard_image() + if image_path: + console.print(f" [green]图片已从剪贴板保存: {image_path}[/green]") + console.print(" [dim]现在可以输入任务描述，图片将作为上下文[/dim]") + # 存储图片路径供后续任务使用 + if not hasattr(session, '_pending_images'): + session._pending_images = [] + session._pending_images.append(image_path) + else: + console.print(" [yellow]剪贴板中没有图片[/yellow]") + + elif cmd == "/image": + if not arg: + console.print(" [yellow]用法: /image <图片路径>[/yellow]") + else: + from kaiwu.experts.vision_expert import validate_image_path + image_path = arg.strip() + if validate_image_path(image_path): + console.print(f" [green]图片已添加: {image_path}[/green]") + console.print(" [dim]现在可以输入任务描述，图片将作为上下文[/dim]") + # 存储图片路径供后续任务使用 + if not hasattr(session, '_pending_images'): + session._pending_images = [] + session._pending_images.append(image_path) + else: + console.print(f" [red]图片文件不存在或格式不支持: {image_path}[/red]") + elif cmd == "/api": api_parts = user_input.split() result = _handle_api_command(api_parts, ollama_url, ollama_model) @@ -677,6 +723,13 @@ def _repl(model_path, ollama_url, ollama_model, project_root, verbose): f"耗时{pruner._last_compress_ms:.1f}ms）[/dim]" ) + # 处理待处理的图片 + pending_images = getattr(session, '_pending_images', []) + if pending_images: + console.print(f" [cyan]图片上下文: {len(pending_images)} 张图片[/cyan]") + for img_path in pending_images: + console.print(f" - {img_path}") + t0 = time.perf_counter() # P2: Small model forces plan mode (问题6修复：用户可通过 no_search 间接控制) effective_plan = plan_next or (model_strategy.force_plan_mode and not no_search) @@ -689,7 +742,13 @@ def _repl(model_path, ollama_url, ollama_model, project_root, verbose): verbose=verbose, plan=effective_plan, no_search=False, + image_paths=pending_images if pending_images else None, ) + + # 清除已使用的图片 + if pending_images: + session._pending_images = [] + elapsed = time.perf_counter() - t0 plan_next = False # Reset plan flag diff --git a/kaiwu/core/gate.py b/kaiwu/core/gate.py index 5c11609..976fa72 100644 --- a/kaiwu/core/gate.py +++ b/kaiwu/core/gate.py @@ -27,6 +27,7 @@ expert_type选项： - doc：写注释、文档、README（仅限代码相关文档，用户明确提到已有文件+docstring/注释） - office：仅限生成Excel(.xlsx)/Word(.docx)/PPT(.pptx)办公文档，不包括代码文件 - chat：问候、闲聊、非编码问题、询问天气、询问知识 +- vision：图片分析、图片代码生成、UI截图分析、设计图实现（用户上传了图片或提到图片相关内容） difficulty选项：easy | hard hard条件（满足任意一条）： @@ -48,6 +49,7 @@ subtask_hint：仅difficulty=hard时填写，逗号分隔的子任务简述。 - "修复src/xxx.py" → locator_repair - "重构src/xxx.py" → refactor - 不确定时优先选codegen或locator_repair，不要选office +- 图片相关任务（上传图片、分析截图、根据设计图生成代码） → vision 示例： - "你好" → {{"expert_type":"chat","task_summary":"问候","difficulty":"easy","needs_search":false,"subtask_hint":""}} @@ -57,6 +59,8 @@ subtask_hint：仅difficulty=hard时填写，逗号分隔的子任务简述。 - "生成一个Excel报表" → {{"expert_type":"office","task_summary":"Excel报表","difficulty":"easy","needs_search":false,"subtask_hint":""}} - "根据内容生成项目介绍PPT" → {{"expert_type":"office","task_summary":"项目PPT","difficulty":"easy","needs_search":false,"subtask_hint":""}} - "做个幻灯片汇报" → {{"expert_type":"office","task_summary":"汇报PPT","difficulty":"easy","needs_search":false,"subtask_hint":""}} +- "分析这张截图" → {{"expert_type":"vision","task_summary":"图片分析","difficulty":"easy","needs_search":false,"subtask_hint":""}} +- "根据UI设计图生成HTML代码" → {{"expert_type":"vision","task_summary":"UI代码生成","difficulty":"easy","needs_search":false,"subtask_hint":""}} 格式：{{"expert_type":"...","task_summary":"10字内","difficulty":"...","needs_search":false,"subtask_hint":""}} @@ -66,14 +70,14 @@ subtask_hint：仅difficulty=hard时填写，逗号分隔的子任务简述。 GATE_GRAMMAR = r''' root ::= "{" ws expert-type "," ws task-summary "," ws difficulty "}" ws expert-type ::= "\"expert_type\"" ws ":" ws "\"" expert-val "\"" -expert-val ::= "locator_repair" | "codegen" | "refactor" | "doc" | "office" | "chat" +expert-val ::= "locator_repair" | "codegen" | "refactor" | "doc" | "office" | "chat" | "vision" task-summary ::= "\"task_summary\"" ws ":" ws string difficulty ::= "\"difficulty\"" ws ":" ws ("\"easy\"" | "\"hard\"") string ::= "\"" [^"]* "\"" ws ::= [ \t\n]* ''' -VALID_EXPERT_TYPES = {"locator_repair", "codegen", "refactor", "doc", "office", "chat"} +VALID_EXPERT_TYPES = {"locator_repair", "codegen", "refactor", "doc", "office", "chat", "vision"} VALID_DIFFICULTIES = {"easy", "hard"} diff --git a/kaiwu/core/orchestrator.py b/kaiwu/core/orchestrator.py index 6672286..8ac4bbe 100644 --- a/kaiwu/core/orchestrator.py +++ b/kaiwu/core/orchestrator.py @@ -36,6 +36,7 @@ EXPERT_SEQUENCES = { "doc": ["locator", "generator"], "office": ["office"], "chat": ["chat"], + "vision": ["vision"], } @@ -59,6 +60,7 @@ class PipelineOrchestrator: ab_tester: ABTester | None = None, chat_expert: ChatExpert | None = None, debug_subagent=None, + vision_expert=None, ): self.locator = locator self.generator = generator @@ -66,6 +68,7 @@ class PipelineOrchestrator: self.search_augmentor = search_augmentor self.office_handler = office_handler self.chat_expert = chat_expert + self.vision_expert = vision_expert self.tools = tool_executor self.memory = memory self.registry = registry @@ -85,6 +88,7 @@ class PipelineOrchestrator: no_search: bool = False, skip_checkpoint: bool = False, pre_search_results: str = "", + image_paths: list = None, ) -> dict: """ Execute the expert pipeline. @@ -104,6 +108,11 @@ class PipelineOrchestrator: expert_system_prompt=gate_result.get("system_prompt", ""), ) + # 处理图片路径 + if image_paths: + ctx.image_paths = image_paths + logger.info(f"[orchestrator] 任务包含 {len(image_paths)} 张图片") + expert_type = gate_result.get("expert_type", "locator_repair") # ── Pre-search results injection (P1-B) ── @@ -140,6 +149,30 @@ class PipelineOrchestrator: "elapsed": elapsed, } + # vision类型：图片处理任务 + if expert_type == "vision": + self._emit(on_status, "vision", "图片处理模式") + if self.vision_expert and ctx.image_paths: + # 使用第一个图片进行处理 + ctx.image_path = ctx.image_paths[0] + result = self.vision_expert.run(ctx) + elapsed = time.time() - start_time + return { + "success": result.get("success", False), + "context": ctx, + "error": result.get("error"), + "elapsed": elapsed, + } + else: + ctx.generator_output = {"explanation": "图片处理功能需要配置Vision专家", "patches": []} + elapsed = time.time() - start_time + return { + "success": False, + "context": ctx, + "error": "Vision专家未配置或未提供图片", + "elapsed": elapsed, + } + # Gate 3: AB test — check if a candidate expert should be used for this task ab_candidate_name = None ab_used_new = False diff --git a/kaiwu/experts/vision_expert.py b/kaiwu/experts/vision_expert.py new file mode 100644 index 0000000..ea52462 --- /dev/null +++ b/kaiwu/experts/vision_expert.py @@ -0,0 +1,425 @@ +""" +VisionExpert: 多模态图片处理专家 +支持图片上传、剪贴板粘贴、图片分析和基于图片的代码生成。 + +Pipeline: 图片输入 → 图片分析 → 任务路由 → 代码生成/图片描述 +""" + +import base64 +import logging +import os +import tempfile +from pathlib import Path +from typing import Optional, Union + +from kaiwu.core.context import TaskContext + +logger = logging.getLogger(__name__) + +# ── 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_path = getattr(ctx, 'image_path', None) + + if not image_path: + return { + "success": False, + "output": "错误：未提供图片路径", + "metadata": {"error": "no_image_path"} + } + + # 验证图片文件 + if not self._validate_image(image_path): + return { + "success": False, + "output": f"错误：图片文件不存在或格式不支持: {image_path}", + "metadata": {"error": "invalid_image"} + } + + # 分析用户意图 + user_input = ctx.user_input.strip() + is_codegen_task = self._is_codegen_task(user_input) + + try: + # 编码图片 + image_base64 = self._encode_image(image_path) + + if is_codegen_task: + return self._run_codegen(ctx, image_base64, image_path) + else: + return self._run_analysis(ctx, image_base64, image_path) + + 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) + if not path.exists(): + return False + + # 支持的图片格式 + supported_formats = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp'} + return path.suffix.lower() in supported_formats + + def _encode_image(self, image_path: str) -> str: + """将图片编码为base64""" + with open(image_path, "rb") as f: + return base64.b64encode(f.read()).decode('utf-8') + + 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, image_base64: str, image_path: str) -> dict: + """运行图片分析""" + logger.info(f"[vision] 分析图片: {image_path}") + + # 构建提示词 + user_input = ctx.user_input.strip() + if user_input: + prompt = f"用户上传了一张图片并说：{user_input}\n\n请分析这张图片。" + else: + prompt = "用户上传了一张图片，请分析其内容。" + + # 调用Vision LLM + response = self._call_vision_llm( + system_prompt=VISION_ANALYSIS_SYSTEM, + user_prompt=prompt, + image_base64=image_base64 + ) + + return { + "success": True, + "output": response, + "metadata": { + "type": "vision_analysis", + "image_path": image_path, + "has_user_task": bool(user_input) + } + } + + def _run_codegen(self, ctx: TaskContext, image_base64: str, image_path: str) -> dict: + """运行基于图片的代码生成""" + logger.info(f"[vision] 基于图片生成代码: {image_path}") + + user_input = ctx.user_input.strip() + + # 构建提示词 + prompt = f"用户上传了一张图片并要求：{user_input}\n\n请根据图片内容生成代码。" + + # 调用Vision LLM + response = self._call_vision_llm( + system_prompt=VISION_CODEGEN_SYSTEM, + user_prompt=prompt, + image_base64=image_base64 + ) + + # 尝试执行生成的代码（如果用户要求） + 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_path": image_path, + "task": user_input + } + } + + def _call_vision_llm(self, system_prompt: str, user_prompt: str, image_base64: str) -> str: + """调用支持Vision的LLM (Anthropic Messages API 格式) + + 优先使用 self.llm (如果支持 vision)，否则回退到环境变量配置的 API： + KWCODE_VISION_API_URL - API endpoint (默认 Anthropic 格式) + 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, image_base64) + except (AttributeError, TypeError, Exception) as e: + logger.debug(f"[vision] self.llm 不支持 vision，回退到 API: {e}") + + # 回退：直接调用 Anthropic Messages API + return self._call_anthropic_vision(system_prompt, user_prompt, image_base64) + + def _try_llm_vision(self, system_prompt: str, user_prompt: str, image_base64: str) -> str: + """尝试通过 self.llm 的 chat 接口发送多模态请求""" + messages = [ + {"role": "system", "content": system_prompt}, + { + "role": "user", + "content": [ + {"type": "text", "text": user_prompt}, + { + "type": "image_url", + "image_url": {"url": f"data:image/png;base64,{image_base64}"}, + }, + ], + }, + ] + return self.llm.chat(messages, max_tokens=2048) + + def _call_anthropic_vision(self, system_prompt: str, user_prompt: str, image_base64: str) -> str: + """直接调用 Anthropic Messages API（兼容 xiaomimimo 等代理）""" + import json as _json + import urllib.request + import urllib.error + + api_url = os.environ.get( + "KWCODE_VISION_API_URL", + "https://token-plan-cn.xiaomimimo.com/anthropic/v1/messages", + ) + api_key = os.environ.get("KWCODE_VISION_API_KEY", "") + model = os.environ.get("KWCODE_VISION_MODEL", "mimo-v2-omni") + + # 检测图片格式（从 base64 开头字节判断） + media_type = "image/png" + raw_sample = image_base64[:20] + if raw_sample.startswith("/9j"): + media_type = "image/jpeg" + elif raw_sample.startswith("R0lGOD"): + media_type = "image/gif" + elif raw_sample.startswith("UklGR"): + media_type = "image/webp" + + payload = { + "model": model, + "max_tokens": 2048, + "system": system_prompt, + "messages": [ + { + "role": "user", + "content": [ + { + "type": "image", + "source": { + "type": "base64", + "media_type": media_type, + "data": image_base64, + }, + }, + {"type": "text", "text": user_prompt}, + ], + } + ], + } + + 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 + + # 提取文本 + # 检查 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') + + # 执行代码 + result = self.tools.run_bash(f"python {temp_file}") + return result.get("output", "") + + 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) + if not path_obj.exists(): + return False + + supported_formats = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp'} + return path_obj.suffix.lower() in supported_formats + + +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/pyproject.toml b/pyproject.toml index 1ffa3a8..fb9d764 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -30,6 +30,7 @@ dependencies = [ [project.optional-dependencies] rerank = ["sentence-transformers>=2.7.0"] +multimodal = ["Pillow>=10.0", "pyperclip>=1.8"] [project.scripts] kwcode = "kaiwu.cli.main:app"