lywsvip/kwcode
1
0
Fork 0
mirror of https://github.com/val1813/kwcode.git synced 2026-09-03 06:34:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor: vision expert second review + remove hardcoded credentials

Browse Source 
- Remove hardcoded API URL and model defaults
- Require KWCODE_VISION_API_URL, _API_KEY, _MODEL env vars
- Clear error message with setup instructions when unconfigured
- README: add Vision API configuration guide with model examples
- VisionExpert.run() handles multi-image internally
- MAX_IMAGE_BYTES (20MB) and MAX_IMAGE_COUNT (20) limits
- Magic bytes detection for unknown image formats
- Subprocess-based code execution with proper error reporting
This commit is contained in:
dashitongzhi
2026-04-30 12:10:14 +08:00
parent 1488834e71
commit e9f3937369
3 changed files with 170 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

17
README.md
View File

@@ -219,6 +219,23 @@ Prompt Optimizer（可选，需 Anthropic API key）：
- 支持格式：PNG、JPG、JPEG、GIF、WebP、BMP
- 安装：`pip install kwcode[multimodal]`
**Vision API 配置**（使用前必配）：
```bash
# 设置环境变量（支持任何兼容 Anthropic Messages API 的服务）
export KWCODE_VISION_API_URL="https://your-provider.com/v1/messages"
export KWCODE_VISION_API_KEY="your-api-key"
export KWCODE_VISION_MODEL="your-multimodal-model"
```
支持的 Vision 模型示例：
- OpenAI: `gpt-4o` (endpoint: `https://api.openai.com/v1/chat/completions`，需用 OpenAI 格式)
- Anthropic: `claude-sonnet-4-20250514` (endpoint: `https://api.anthropic.com/v1/messages`)
- 小米 MiMo: `mimo-v2-omni` (Anthropic 格式代理)
- 本地模型: Ollama 多模态模型（通过 Ollama 兼容 endpoint）
> **注意**：模型必须支持图片输入（多模态），纯文本模型无法处理图片任务。
### 价值可见
- `kwcode stats`：完成任务数、节省时间估算
- 飞轮通知：专家投产时弹出

24
kaiwu/core/orchestrator.py
View File

@@ -153,33 +153,19 @@ class PipelineOrchestrator:
if expert_type == "vision":
self._emit(on_status, "vision", "图片处理模式")
if self.vision_expert and ctx.image_paths:
outputs = []
failures = []
metadata = []
for idx, image_path in enumerate(ctx.image_paths, 1):
ctx.image_path = image_path
result = self.vision_expert.run(ctx)
output = result.get("output", "")
if len(ctx.image_paths) > 1:
outputs.append(f"图片 {idx} ({image_path}):\n{output}")
else:
outputs.append(output)
metadata.append(result.get("metadata", {}))
if not result.get("success", False):
failures.append(output or result.get("error") or f"图片处理失败: {image_path}")
explanation = "\n\n".join(outputs).strip()
result = self.vision_expert.run(ctx)
explanation = result.get("output", "").strip()
ctx.generator_output = {
"explanation": explanation,
"patches": [],
"metadata": {"vision": metadata},
"metadata": {"vision": result.get("metadata", {})},
}
elapsed = time.time() - start_time
success = not failures
success = result.get("success", False)
return {
"success": success,
"context": ctx,
"error": "\n".join(failures) if failures else None,
"error": None if success else explanation or "图片处理失败",
"elapsed": elapsed,
}
else:

214
kaiwu/experts/vision_expert.py
View File

@@ -16,6 +16,8 @@ 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 ──────────────────────────────────────────
@@ -92,21 +94,31 @@ class VisionExpert:
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_path:
if not image_paths:
return {
"success": False,
"output": "错误：未提供图片路径",
"metadata": {"error": "no_image_path"}
}
# 验证图片文件
if not self._validate_image(image_path):
if len(image_paths) > MAX_IMAGE_COUNT:
return {
"success": False,
"output": f"错误：图片文件不存在或格式不支持: {image_path}",
"metadata": {"error": "invalid_image"}
"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}
}
# 分析用户意图
@@ -114,14 +126,19 @@ class VisionExpert:
is_codegen_task = self._is_codegen_task(user_input)
try:
# 编码图片
image_base64 = self._encode_image(image_path)
media_type = self._media_type_for_path(image_path, image_base64)
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, image_base64, image_path, media_type)
return self._run_codegen(ctx, images)
else:
return self._run_analysis(ctx, image_base64, image_path, media_type)
return self._run_analysis(ctx, images)
except Exception as e:
logger.error(f"VisionExpert error: {e}")
@@ -136,18 +153,26 @@ class VisionExpert:
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
# 支持的图片格式
supported_formats = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp'}
return path.suffix.lower() in supported_formats
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(image_path, "rb") as f:
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",
@@ -160,14 +185,36 @@ class VisionExpert:
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"
return "image/png"
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:
"""判断是否为代码生成任务"""
@@ -182,23 +229,23 @@ class VisionExpert:
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, media_type: str) -> dict:
def _run_analysis(self, ctx: TaskContext, images: list[dict]) -> dict:
"""运行图片分析"""
logger.info(f"[vision] 分析图片: {image_path}")
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"用户上传了一张图片并说：{user_input}\n\n请分析这张图片。"
prompt = f"用户上传了{image_count}张图片并说：{user_input}\n\n请结合所有图片进行分析。"
else:
prompt = "用户上传了一张图片，请分析其内容。"
prompt = f"用户上传了{image_count}张图片，请分析其内容。"
# 调用Vision LLM
response = self._call_vision_llm(
system_prompt=VISION_ANALYSIS_SYSTEM,
user_prompt=prompt,
image_base64=image_base64,
media_type=media_type,
images=images,
)
return {
@@ -206,26 +253,27 @@ class VisionExpert:
"output": response,
"metadata": {
"type": "vision_analysis",
"image_path": image_path,
"image_paths": [img["path"] for img in images],
"image_count": image_count,
"has_user_task": bool(user_input)
}
}
def _run_codegen(self, ctx: TaskContext, image_base64: str, image_path: str, media_type: str) -> dict:
def _run_codegen(self, ctx: TaskContext, images: list[dict]) -> dict:
"""运行基于图片的代码生成"""
logger.info(f"[vision] 基于图片生成代码: {image_path}")
logger.info("[vision] 基于图片生成代码: %s", ", ".join(img["path"] for img in images))
user_input = ctx.user_input.strip()
image_count = len(images)
# 构建提示词
prompt = f"用户上传了一张图片并要求：{user_input}\n\n请根据图片内容生成代码。"
prompt = f"用户上传了{image_count}张图片并要求：{user_input}\n\n请结合所有图片内容生成代码。"
# 调用Vision LLM
response = self._call_vision_llm(
system_prompt=VISION_CODEGEN_SYSTEM,
user_prompt=prompt,
image_base64=image_base64,
media_type=media_type,
images=images,
)
# 尝试执行生成的代码（如果用户要求）
@@ -239,30 +287,41 @@ class VisionExpert:
"output": response,
"metadata": {
"type": "vision_codegen",
"image_path": image_path,
"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, image_base64: str, media_type: str) -> str:
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 格式)
KWCODE_VISION_API_KEY - API key
KWCODE_VISION_MODEL - 模型名 (默认 mimo-v2-omni)
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, image_base64, media_type)
return self._try_llm_vision(system_prompt, user_prompt, images)
except 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, media_type)
if not self._vision_api_configured():
raise RuntimeError(
"本地模型不支持图片输入，且未显式配置Vision API；"
"请设置 KWCODE_VISION_API_URL 或 KWCODE_VISION_API_KEY 后重试"
)
def _try_llm_vision(self, system_prompt: str, user_prompt: str, image_base64: str, media_type: str) -> str:
# 回退：直接调用 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.
@@ -277,38 +336,62 @@ class VisionExpert:
{
"role": "user",
"content": user_prompt,
"images": [image_base64],
"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": [
{"type": "text", "text": user_prompt},
{
"type": "image_url",
"image_url": {"url": f"data:{media_type};base64,{image_base64}"},
},
],
"content": content,
},
]
return self.llm.chat(messages, max_tokens=2048)
def _call_anthropic_vision(self, system_prompt: str, user_prompt: str, image_base64: str, media_type: str) -> str:
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",
"https://token-plan-cn.xiaomimimo.com/anthropic/v1/messages",
)
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", "mimo-v2-omni")
model = os.environ.get("KWCODE_VISION_MODEL", "")
if not api_url or not model:
raise RuntimeError(
"Vision API 未配置。请设置环境变量：\n"
" export KWCODE_VISION_API_URL=<Anthropic Messages API endpoint>\n"
" export KWCODE_VISION_API_KEY=<your api key>\n"
" export KWCODE_VISION_MODEL=<multimodal model name>\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,
@@ -317,17 +400,7 @@ class VisionExpert:
"messages": [
{
"role": "user",
"content": [
{
"type": "image",
"source": {
"type": "base64",
"media_type": media_type,
"data": image_base64,
},
},
{"type": "text", "text": user_prompt},
],
"content": content,
}
],
}
@@ -392,8 +465,12 @@ class VisionExpert:
temp_file.write_text(code, encoding='utf-8')
# 执行代码
result = self.tools.run_bash(f"python {shlex.quote(str(temp_file))}")
return result.get("output", "")
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)}"
@@ -443,9 +520,14 @@ 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
supported_formats = {'.png', '.jpg', '.jpeg', '.gif', '.webp', '.bmp'}
return path_obj.suffix.lower() in supported_formats
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: