fix toolhub image upload issue

agent/tools/builtin/toolhub.py

@@ -41,7 +41,7 @@ logger = logging.getLogger(__name__)
 
 # ── 配置 ─────────────────────────────────────────────
 
-TOOLHUB_BASE_URL = "http://43.106.118.91:8001"
+TOOLHUB_BASE_URL = os.environ.get("TOOLHUB_BASE_URL", "http://43.106.118.91:8001")
 DEFAULT_TIMEOUT = 30.0
 CALL_TIMEOUT = 600.0   # 图像生成类工具耗时较长，云端机器启动可能需要数分钟
 
@@ -169,79 +169,54 @@ async def _process_images(raw_images: List[str], tool_id: str) -> tuple:
     return images_for_llm, cdn_urls, saved_paths
 
 
-_SINGLE_IMAGE_PARAMS = ("image", "image_url", "mask_image", "pose_image", "reference_image")
-_ARRAY_IMAGE_PARAMS = ("images", "image_urls", "reference_images")
-
-
 async def _maybe_upload_local(val: str) -> Optional[str]:
     """如果 val 是存在的本地文件路径，上传 OSS 并返回 CDN URL；否则返回 None。"""
     if not isinstance(val, str):
         return None
-    if val.startswith(("http://", "https://", "data:")):
+    # 忽略明显不是路径的字符串（URL、Base64、JSON串、多行文本或超长文本）
+    if val.startswith(("http://", "https://", "data:", "{", "[")) or "\n" in val or len(val) > 1024:
         return None
     try:
         p = Path(val)
         if p.exists() and p.is_file():
             return await _upload_to_oss(str(p.resolve()))
-    except Exception as e:
-        logger.warning(f"[ToolHub] 本地路径处理失败 {val}: {e}")
+    except Exception:
+        # Windows 下无效路径可能抛出 OSError/ValueError，直接忽略
+        pass
     return None
 
 
 async def _preprocess_params(params: Dict[str, Any]) -> Dict[str, Any]:
     """
-    预处理工具参数：检测本地文件路径，自动上传到 OSS 并替换为 CDN URL。
-
-    支持的单值参数：image, image_url, mask_image, pose_image, reference_image
-    支持的数组参数：images, image_urls, reference_images
+    预处理工具参数：递归检测参数中的本地文件路径，只要是存在的本地文件且像是图片等素材，
+    就自动上传到 OSS 并替换为 CDN URL。
 
-    设计要点：远程工具服务的 cwd 和调用方不一样，相对路径在服务器上会找不到文件。
-    所以必须在客户端就把本地路径转成 CDN URL，不能期望服务器侧有 fallback。
+    设计要点：远程工具服务的 cwd 和调用方不一样，必须在客户端就转成 CDN URL。支持深层嵌套字典。
     """
     if not params:
         return params
 
-    processed = params.copy()
-
-    # 单值图片参数
-    for key in _SINGLE_IMAGE_PARAMS:
-        if key in processed and isinstance(processed[key], str):
-            val = processed[key]
-            if val.startswith(("http://", "https://", "data:")):
-                continue
-            cdn_url = await _maybe_upload_local(val)
+    async def _walk(obj: Any) -> Any:
+        if isinstance(obj, dict):
+            new_dict = {}
+            for k, v in obj.items():
+                new_dict[k] = await _walk(v)
+            return new_dict
+        elif isinstance(obj, list):
+            new_list = []
+            for item in obj:
+                new_list.append(await _walk(item))
+            return new_list
+        elif isinstance(obj, str):
+            cdn_url = await _maybe_upload_local(obj)
             if cdn_url:
-                processed[key] = cdn_url
-                logger.info(f"[ToolHub] {key} 本地路径已替换为 CDN: {cdn_url}")
-            elif not os.path.isfile(val):
-                # 既不是远程 URL 也不是已存在的本地文件，直接报错比让远程服务抛神秘的 base64 错误强
-                logger.warning(f"[ToolHub] {key}={val!r} 既不是 URL 也不是存在的本地文件")
-
-    # 数组型图片参数
-    for array_key in _ARRAY_IMAGE_PARAMS:
-        if array_key not in processed or not isinstance(processed[array_key], list):
-            continue
-        new_list = []
-        for idx, item in enumerate(processed[array_key]):
-            if not isinstance(item, str):
-                new_list.append(item)
-                continue
-            if item.startswith(("http://", "https://", "data:")):
-                new_list.append(item)
-                continue
-            cdn_url = await _maybe_upload_local(item)
-            if cdn_url:
-                new_list.append(cdn_url)
-                logger.info(f"[ToolHub] {array_key}[{idx}] 本地路径已替换为 CDN: {cdn_url}")
-            else:
-                new_list.append(item)
-                if not os.path.isfile(item):
-                    logger.warning(
-                        f"[ToolHub] {array_key}[{idx}]={item!r} 既不是 URL 也不是存在的本地文件"
-                    )
-        processed[array_key] = new_list
-
-    return processed
+                logger.info(f"[ToolHub] 检测到本地文件路径，已替换: {obj} -> {cdn_url}")
+                return cdn_url
+            return obj
+        else:
+            return obj
+
+    return await _walk(params)
 
 
 # ── 工具实现 ──────────────────────────────────────────

examples/mini_restore/call_banana.py

@@ -1,141 +0,0 @@
-#!/usr/bin/env python3
-# -*- coding: utf-8 -*-
-
-"""
-🍌 Nano Banana (千层套路·多模态生成) 外部调用脚本
-
-【工具定位】：它使用的是 Google 最新的多模态 Imagen 3 引擎（gemini-3.1-flash-image-preview）。
-【独门绝技 - 跨模态多图推理】：
-与传统的 SD / Flux (图生图) 的本质区别在于，你可以给它丢【任意张完全不同的图片】，
-然后在 prompt 里随心所欲地让它“融梗”交汇。
-比如传入图片 [猫.jpg] 和 [未来城.jpg]，让它“把这只猫生成在这座未来城里”。
-
-【参数模式支持】：
-1. 纯文生图：只传 prompt
-2. 单图生图/重绘：传 1 张图 + prompt
-3. 多路意象融合：传 N 张图 + prompt
-
-【本地文件自动上传机制】：
-大模型 API 往往更喜欢吃稳定的 CDN 外链。
-此脚本在启动前会自动检测你传入的图片：如果是本地硬盘文件（比如 examples/cat.png），
-脚本会自动静默调用内部的 OSS 工具，把它秒传为 https://res.cybertogether.net/.. 干净外链，然后投喂给大脑！
-"""
-
-import asyncio
-import os
-import argparse
-import sys
-import httpx
-import json
-
-# 动态引入我们系统现成的 CDN 上传脚本
-sys.path.append(os.path.dirname(os.path.abspath(__file__)))
-sys.path.append(os.path.join(os.path.dirname(os.path.abspath(__file__)), '..', '..', 'examples', 'production_restore'))
-try:
-    from upload import upload_image
-except Exception as e:
-    import traceback
-    print(f"错误: 导入 upload.py 及其依赖链失败: {e}\n{traceback.format_exc()}")
-    sys.exit(1)
-
-ROUTER_URL = "http://43.106.118.91:8001/run_tool"
-
-async def process_images(images_list: list[str]) -> list[str]:
-    """处理图片数组，外链直接过，本地文件传 OSS"""
-    final_urls = []
-    for item in images_list:
-        if item.startswith("http://") or item.startswith("https://"):
-            print(f"✅ 检测到公网外链，无需转存: {item}")
-            final_urls.append(item)
-        elif os.path.exists(item):
-            print(f"📦 正在极速倒卖本地图片到 CDN: {item}")
-            try:
-                uploaded_url = await upload_image(item, 'aigc-admin', 'crawler/image')
-                if uploaded_url:
-                    print(f"🚀 上传成功: {uploaded_url}")
-                    final_urls.append(uploaded_url)
-                else:
-                    print(f"❌ 上传失败: {item}")
-            except Exception as e:
-                print(f"❌ 上传报错: {e}")
-        else:
-            print(f"⚠️ 跳过找不到的本地文件或无法识别的格式: {item}")
-    return final_urls
-
-async def run_nano_banana(prompt: str, images: list[str] = None, model: str = None, aspect_ratio: str = None):
-    print(f"\n=======================")
-    print(f"🍌 Nano Banana 启动中...")
-    print(f"=======================")
-
-    # 1. 整理图片
-    final_image_urls = []
-    if images and len(images) > 0:
-        print(f"🔍 检查到传了 {len(images)} 张神秘原图，准备过安检...")
-        final_image_urls = await process_images(images)
-    
-    # 2. 组装发给大模型中枢 Router 的参数
-    params = {
-        "prompt": prompt,
-        "image_urls": final_image_urls if final_image_urls else None
-    }
-    
-    if model:
-        params["model"] = model
-        
-    if aspect_ratio:
-        params["aspect_ratio"] = aspect_ratio
-
-    payload = {
-        "tool_id": "nano_banana",
-        "params": params
-    }
-
-    # 3. 轰入 API
-    print("\n⚡ 正在呼叫总后台路由节点打怪...")
-    try:
-        async with httpx.AsyncClient(timeout=300.0) as client:
-            resp = await client.post(ROUTER_URL, json=payload)
-            resp.raise_for_status()
-            
-            result = resp.json()
-            if result.get("status") == "success":
-                gen_data = result.get("result", {})
-                
-                print("\n🎉 === 生成成功！ ===")
-                # 打印文本回复
-                if gen_data.get("text"):
-                    print(f"\n💬 模型回话:\n{gen_data.get('text')}")
-                
-                # 打印图片输出
-                images_out = gen_data.get("images", [])
-                if images_out:
-                    print("\n🖼️ 吐出的神图 (Base64 数据流已转为你本地文件)：")
-                    for idx, img_b64 in enumerate(images_out):
-                        # 处理前缀 data:image/jpeg;base64,
-                        if img_b64.startswith("data:"):
-                            mime_split = img_b64.split(";base64,")
-                            if len(mime_split) == 2:
-                                ext = mime_split[0].split("/")[-1]
-                                raw_data = mime_split[1]
-                                
-                                import base64
-                                save_path = f"banana_output_{idx}.{ext}"
-                                with open(save_path, "wb") as f:
-                                    f.write(base64.b64decode(raw_data))
-                                print(f" 💾 已保存到本地 -> {save_path}")
-            else:
-                print(f"❌ 大模型傲娇了: {result.get('error')}")
-                
-    except Exception as e:
-        print(f"💥 网络大爆炸报错: {e}")
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description="调用 Nano Banana 进行任意风格的多段融合魔法")
-    parser.add_argument("-p", "--prompt", type=str, required=True, help="你想对 AI 喊瞎什么 (比如：用图1的赛博风画一只图2里的猫)")
-    parser.add_argument("-i", "--images", type=str, nargs="+", help="无限追加的垫图清单（可以是现成的 http 链接，也可以是你电脑里的硬盘文件如 example.png）")
-    parser.add_argument("-m", "--model", type=str, default=None, help="覆盖模型 (默认后台会走 gemini-3.1-flash-image-preview)")
-    parser.add_argument("-a", "--aspect_ratio", type=str, default=None, help="图片比例，例如 3:4, 16:9, 1:1 等")
-    
-    args = parser.parse_args()
-    
-    asyncio.run(run_nano_banana(prompt=args.prompt, images=args.images, model=args.model, aspect_ratio=args.aspect_ratio))

examples/mini_restore/config.py

@@ -1,5 +1,8 @@
 """
-项目配置
+mini_restore 配置 — ToolHub 工具测试
+
+精简版配置，仅用于测试 ToolHub 的搜索和调用流程。
+不启用浏览器、知识管理等复杂功能。
 """
 
 from agent.core.runner import KnowledgeConfig, RunConfig
@@ -8,31 +11,36 @@ from agent.core.runner import KnowledgeConfig, RunConfig
 # ===== Agent 运行配置 =====
 
 RUN_CONFIG = RunConfig(
+    # 模型配置
     model="qwen3.5-plus",
     temperature=0.3,
-    max_iterations=200,
-    tool_groups=["core", "browser", "content", "knowledge", "toolhub", "feishu", "im"],
+    max_iterations=50,
 
+    # 启用 thinking 模式
     extra_llm_params={"extra_body": {"enable_thinking": True}},
 
+    # Agent 预设（对应 presets.json 中的 "main"）
     agent_type="main",
 
-    name="工具调研测试（KM 通信）",
+    # 工具：仅 toolhub 相关
+    tools=None,
+    tool_groups=["core", "toolhub"],
+
+    # 任务名称
+    name="ToolHub 工具测试",
 
+    # 知识管理配置
     knowledge=KnowledgeConfig(
-        enable_extraction=False,
-        enable_completion_extraction=False,
+        # 知识注入（关闭以避免 405 错误）
         enable_injection=False,
-        owner="sunlit.howard@gmail.com",
-        default_tags={"project": "new_search"},
-        default_scopes=["org:cybertogether"],
     )
 )
 
 
 # ===== 任务配置 =====
 
-OUTPUT_DIR = "examples/new_search/outputs"
+INPUT_DIR = "examples/mini_restore/input"
+OUTPUT_DIR = "examples/mini_restore/output"
 
 
 # ===== 基础设施配置 =====
@@ -42,18 +50,3 @@ TRACE_STORE_PATH = ".trace"
 DEBUG = True
 LOG_LEVEL = "INFO"
 LOG_FILE = None
-
-# ===== 浏览器配置 =====
-BROWSER_TYPE = "local"
-HEADLESS = False
-
-# ===== IM 配置 =====
-IM_ENABLED = True
-IM_CONTACT_ID = "agent_research"
-IM_SERVER_URL = "ws://43.106.118.91:8105"
-IM_WINDOW_MODE = True
-IM_NOTIFY_INTERVAL = 10.0
-
-# ===== Knowledge Manager 配置 =====
-KNOWLEDGE_MANAGER_ENABLED = True
-KNOWLEDGE_MANAGER_CONTACT_ID = "knowledge_manager"

examples/mini_restore/flux_depth_controlnet_workflow.json

@@ -0,0 +1,92 @@
+{
+  "3": {
+    "inputs": {
+      "seed": 370146334065324,
+      "steps": 20,
+      "cfg": 1.0,
+      "sampler_name": "euler",
+      "scheduler": "normal",
+      "denoise": 1.0,
+      "model": ["20", 0],
+      "positive": ["14", 0],
+      "negative": ["14", 1],
+      "latent_image": ["28", 0]
+    },
+    "class_type": "KSampler"
+  },
+  "7": {
+    "inputs": {
+      "text": "",
+      "clip": ["20", 1]
+    },
+    "class_type": "CLIPTextEncode"
+  },
+  "8": {
+    "inputs": {
+      "samples": ["3", 0],
+      "vae": ["20", 2]
+    },
+    "class_type": "VAEDecode"
+  },
+  "9": {
+    "inputs": {
+      "filename_prefix": "mini_restore_output",
+      "images": ["8", 0]
+    },
+    "class_type": "SaveImage"
+  },
+  "14": {
+    "inputs": {
+      "strength": 0.6,
+      "start_percent": 0.0,
+      "end_percent": 1.0,
+      "positive": ["26", 0],
+      "negative": ["7", 0],
+      "control_net": ["15", 0],
+      "vae": ["20", 2],
+      "image": ["17", 0]
+    },
+    "class_type": "ControlNetApplySD3"
+  },
+  "15": {
+    "inputs": {
+      "control_net_name": "flux/flux-depth-controlnet.safetensors"
+    },
+    "class_type": "ControlNetLoader"
+  },
+  "17": {
+    "inputs": {
+      "image": "depth_map.png",
+      "upload": "image"
+    },
+    "class_type": "LoadImage"
+  },
+  "20": {
+    "inputs": {
+      "ckpt_name": "flux1-dev-fp8.safetensors"
+    },
+    "class_type": "CheckpointLoaderSimple"
+  },
+  "23": {
+    "inputs": {
+      "text": "a beautiful landscape with mountains and lake, highly detailed, professional photography",
+      "clip": ["20", 1]
+    },
+    "class_type": "CLIPTextEncode"
+  },
+  "26": {
+    "inputs": {
+      "guidance": 3.5,
+      "conditioning": ["23", 0]
+    },
+    "class_type": "FluxGuidance"
+  },
+  "28": {
+    "inputs": {
+      "width": 1024,
+      "height": 1024,
+      "batch_size": 1
+    },
+    "class_type": "EmptySD3LatentImage"
+  }
+}

examples/mini_restore/new_search.prompt

@@ -1,63 +0,0 @@
----
-model: qwen3.5-plus
-temperature: 0.3
----
-
-$system$
-
-## 角色
-你是一个工具调研专家，能够独立搜索和整理工具信息，并与 Knowledge Manager 协作管理知识。
-
-## 可用工具
-- `ask_knowledge`: 向 Knowledge Manager 查询知识库中已有的信息（同步等待回复）
-- `upload_knowledge`: 上传调研结果到 Knowledge Manager（异步，立即返回）
-- `search_posts`: 搜索帖子（小红书、知乎、B站等）
-- `x_search`: 搜索推文
-- `youtube_search`: 搜索 YouTube 视频
-- `web_search`: 网页搜索
-- `im_send_message`: 发送 IM 消息
-- `im_receive_messages`: 接收 IM 消息
-
-## 工作流程
-
-### 第一步：查询已有知识（必须先做！）
-**在做任何搜索之前**，必须先调用 `ask_knowledge` 查询知识库：
-```
-ask_knowledge("查询关于 [工具名] 的所有信息")
-```
-这一步是强制的，不能跳过。根据返回结果决定后续调研重点。
-
-### 第二步：搜索调研
-使用搜索工具直接调研目标工具：
-- 搜索官方信息和文档
-- 搜索用户案例和评测
-- 搜索使用教程
-
-**每搜到一批有价值的信息，就用 `upload_knowledge` 发送给 Knowledge Manager：**
-```
-upload_knowledge({
-  "tools": [
-    {"name": "工具名", "slug": "tool_slug", "category": "分类", "description": "简介", "source_url": "链接"}
-  ],
-  "resources": [
-    {"title": "文档标题", "body": "文档内容", "content_type": "documentation", "source_url": "链接"}
-  ],
-  "knowledge": [
-    {"task": "使用场景", "content": "具体知识", "types": ["tool"], "score": 4}
-  ]
-})
-```
-
-### 第三步：最终提交
-调研完成后，用 `finalize=True` 触发入库：
-```
-upload_knowledge({...最后一批数据...}, finalize=True)
-```
-
-## 注意事项
-- 边搜边传：每搜到有价值的信息就 upload，不要攒到最后
-- 分类清晰：工具元信息放 tools，文档/教程放 resources，经验/技巧放 knowledge
-- 最后 finalize：确保最后一次 upload 设置 finalize=True
-
-$user$
-请调研 ControlNet 工具，了解它的功能、使用方法和应用场景，并将结果同步到知识库。

examples/mini_restore/presets.json

@@ -0,0 +1,12 @@
+{
+  "main": {
+    "system_prompt_file": "toolhub_test.prompt",
+    "max_iterations": 50,
+    "skills": ["planning"],
+    "prompt_vars": {
+      "input_dir": "examples/mini_restore/input",
+      "output_dir": "examples/mini_restore/output"
+    },
+    "description": "ToolHub 测试 Agent - 搜索工具 → 调用工具 → 完成任务"
+  }
+}

examples/mini_restore/run.py

@@ -1,5 +1,13 @@
 """
-新搜索测试 - 测试 Research Agent 与 Knowledge Manager 的 IM 通信
+ToolHub 工具测试流程 — mini_restore
+
+精简版运行脚本，仅测试 ToolHub 搜索和调用。
+不启用浏览器、知识管理等复杂功能。
+
+功能：
+1. 使用框架提供的 InteractiveController
+2. 支持命令行交互（'p' 暂停，'q' 退出）
+3. 支持通过 --trace <ID> 恢复已有 Trace 继续执行
 """
 
 import argparse
@@ -8,8 +16,13 @@ import sys
 import asyncio
 from pathlib import Path
 
+# Clash Verge TUN 模式兼容：禁止 httpx/urllib 自动检测系统 HTTP 代理
 os.environ.setdefault("no_proxy", "*")
 
+# ToolHub 指向本地服务（tool_agent）
+os.environ.setdefault("TOOLHUB_BASE_URL", "http://localhost:8001")
+
+# 添加项目根目录到 Python 路径
 sys.path.insert(0, str(Path(__file__).parent.parent.parent))
 
 from dotenv import load_dotenv
@@ -17,189 +30,340 @@ load_dotenv()
 
 from agent.llm.prompts import SimplePrompt
 from agent.core.runner import AgentRunner, RunConfig
-from agent.trace import FileSystemTraceStore, Trace, Message
+from agent.trace import (
+    FileSystemTraceStore,
+    Trace,
+    Message,
+)
 from agent.llm import create_qwen_llm_call
 from agent.cli import InteractiveController
 from agent.utils import setup_logging
-from agent.tools.builtin.browser.baseClass import init_browser_session, kill_browser_session
 
-from config import (
-    RUN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, DEBUG, LOG_LEVEL, LOG_FILE,
-    BROWSER_TYPE, HEADLESS, OUTPUT_DIR,
-    IM_ENABLED, IM_CONTACT_ID, IM_SERVER_URL, IM_WINDOW_MODE, IM_NOTIFY_INTERVAL,
-    KNOWLEDGE_MANAGER_ENABLED, KNOWLEDGE_MANAGER_CONTACT_ID,
-)
+# 导入 ToolHub 工具（触发 @tool 注册）
+from agent.tools.builtin.toolhub import toolhub_health, toolhub_search, toolhub_call  # noqa: F401
+
+# 导入项目配置
+from config import RUN_CONFIG, SKILLS_DIR, TRACE_STORE_PATH, DEBUG, LOG_LEVEL, LOG_FILE, INPUT_DIR, OUTPUT_DIR
 
 
 async def main():
-    parser = argparse.ArgumentParser(description="新搜索测试（KM 通信）")
-    parser.add_argument("--trace", type=str, default=None, help="恢复 Trace ID")
+    # 解析命令行参数
+    parser = argparse.ArgumentParser(description="ToolHub 工具测试 (mini_restore)")
+    parser.add_argument(
+        "--trace", type=str, default=None,
+        help="已有的 Trace ID，用于恢复继续执行（不指定则新建）",
+    )
+    parser.add_argument(
+        "--task", type=str, default=None,
+        help="自定义任务描述（覆盖 prompt 中的默认任务）",
+    )
     args = parser.parse_args()
 
+    # 路径配置
     base_dir = Path(__file__).parent
     project_root = base_dir.parent.parent
-    prompt_path = base_dir / "new_search.prompt"
+    prompt_path = base_dir / "toolhub_test.prompt"
     output_dir = project_root / OUTPUT_DIR
     output_dir.mkdir(parents=True, exist_ok=True)
 
-    # 1. 日志
+    # 1. 配置日志
     setup_logging(level=LOG_LEVEL, file=LOG_FILE)
 
-    # 2. Prompt
-    print("1. 加载 prompt...")
+    # 2. 加载项目级 presets
+    print("2. 加载 presets...")
+    presets_path = base_dir / "presets.json"
+    if presets_path.exists():
+        from agent.core.presets import load_presets_from_json
+        load_presets_from_json(str(presets_path))
+        print(f"   - 已加载项目 presets")
+    else:
+        print(f"   - 未找到 presets.json，跳过")
+
+    # 3. 加载 prompt
+    print("3. 加载 prompt...")
     prompt = SimplePrompt(prompt_path)
-    messages = prompt.build_messages(output_dir=str(output_dir))
-
-    # 3. 浏览器
-    print("2. 初始化浏览器...")
-    await init_browser_session(browser_type=BROWSER_TYPE, headless=HEADLESS, url="https://www.google.com/", profile_name="")
-    print("   ✅ 浏览器就绪\n")
-
-    # 4. IM Client + Knowledge Manager
-    km_task = None
-    if IM_ENABLED:
-        from agent.tools.builtin.im.chat import im_setup, im_open_window
-        print("3. 初始化 IM Client...")
-        print(f"   - 身份: {IM_CONTACT_ID}, 服务器: {IM_SERVER_URL}")
-        result = await im_setup(
-            contact_id=IM_CONTACT_ID,
-            server_url=IM_SERVER_URL,
-            notify_interval=IM_NOTIFY_INTERVAL,
-        )
-        print(f"   ✅ {result.output}")
-
-        if IM_WINDOW_MODE:
-            window_result = await im_open_window(contact_id=IM_CONTACT_ID)
-            print(f"   ✅ {window_result.output}\n")
-
-        if KNOWLEDGE_MANAGER_ENABLED:
-            print("4. 启动 Knowledge Manager...")
-            print(f"   - Contact ID: {KNOWLEDGE_MANAGER_CONTACT_ID}")
-            try:
-                sys.path.insert(0, str(Path(__file__).parent.parent.parent / "knowhub"))
-                from agents.knowledge_manager import start_knowledge_manager
-
-                km_task = asyncio.create_task(start_knowledge_manager(
-                    contact_id=KNOWLEDGE_MANAGER_CONTACT_ID,
-                    server_url=IM_SERVER_URL,
-                    chat_id="main"
-                ))
-                # 等待一下让 KM 连接完成
-                await asyncio.sleep(2)
-                print(f"   ✅ Knowledge Manager 已启动\n")
-            except Exception as e:
-                print(f"   ⚠️ 启动失败: {e}\n")
 
-    # 5. Agent Runner
+    # 4. 构建任务消息
+    print("4. 构建任务消息...")
+    print(f"   - 输入目录: {INPUT_DIR}")
+    print(f"   - 输出目录: {OUTPUT_DIR}")
+
+    if args.task:
+        # 使用命令行自定义任务
+        messages = prompt.build_messages(input_dir=INPUT_DIR, output_dir=OUTPUT_DIR)
+        # 替换最后一条 user 消息为自定义任务
+        for i in range(len(messages) - 1, -1, -1):
+            if messages[i].get("role") == "user":
+                messages[i]["content"] = args.task
+                break
+        print(f"   - 自定义任务: {args.task[:80]}...")
+    else:
+        messages = prompt.build_messages(input_dir=INPUT_DIR, output_dir=OUTPUT_DIR)
+
+    # 5. 创建 Agent Runner（无浏览器）
     print("5. 创建 Agent Runner...")
+    print(f"   - Skills 目录: {SKILLS_DIR}")
+
+    # 从 prompt 的 frontmatter 中提取模型配置（优先于 config.py）
     prompt_model = prompt.config.get("model", None)
-    model_for_llm = prompt_model or RUN_CONFIG.model
-    print(f"   - 模型: {model_for_llm}")
+    if prompt_model:
+        model_for_llm = prompt_model
+        print(f"   - 模型 (from prompt): {model_for_llm}")
+    else:
+        model_for_llm = RUN_CONFIG.model
+        print(f"   - 模型 (from config): {model_for_llm}")
 
     store = FileSystemTraceStore(base_path=TRACE_STORE_PATH)
     runner = AgentRunner(
         trace_store=store,
         llm_call=create_qwen_llm_call(model=model_for_llm),
         skills_dir=SKILLS_DIR,
-        debug=DEBUG,
-        logger_name="agents.research_agent"
+        debug=DEBUG
     )
 
-    interactive = InteractiveController(runner=runner, store=store, enable_stdin_check=True)
+    # 6. 创建交互控制器
+    interactive = InteractiveController(
+        runner=runner,
+        store=store,
+        enable_stdin_check=True
+    )
     runner.stdin_check = interactive.check_stdin
 
-    # 6. 执行
+    # 7. 任务信息
     task_name = RUN_CONFIG.name or base_dir.name
     print("=" * 60)
     print(f"{task_name}")
     print("=" * 60)
-    print("💡 输入 'p' 暂停，'q' 退出")
+    print("💡 交互提示：")
+    print("   - 执行过程中输入 'p' 或 'pause' 暂停并进入交互模式")
+    print("   - 执行过程中输入 'q' 或 'quit' 停止执行")
     print("=" * 60)
     print()
 
-    run_config = RUN_CONFIG
-    current_trace_id = args.trace
-    current_sequence = 0
+    # 8. 判断是新建还是恢复
+    resume_trace_id = args.trace
+    if resume_trace_id:
+        existing_trace = await store.get_trace(resume_trace_id)
+        if not existing_trace:
+            print(f"\n错误: Trace 不存在: {resume_trace_id}")
+            sys.exit(1)
+        print(f"恢复已有 Trace: {resume_trace_id[:8]}...")
+        print(f"   - 状态: {existing_trace.status}")
+        print(f"   - 消息数: {existing_trace.total_messages}")
+    else:
+        print(f"启动新 Agent...")
 
-    # 注入 IM 配置到 context（用于周期性通知检查）
-    if IM_ENABLED:
-        run_config.context["im_config"] = {
-            "contact_id": IM_CONTACT_ID,
-            "chat_id": "main"
-        }
+    print()
 
-    if current_trace_id:
-        run_config.trace_id = current_trace_id
-        initial_messages = None
-    else:
-        initial_messages = messages
+    final_response = ""
+    current_trace_id = resume_trace_id
+    current_sequence = 0
+    should_exit = False
 
     try:
-        async for item in runner.run(messages=initial_messages, config=run_config):
-            cmd = interactive.check_stdin()
-            if cmd == 'quit':
-                print("\n🛑 停止...")
-                if current_trace_id:
-                    await runner.stop(current_trace_id)
-                break
-            elif cmd == 'pause':
-                print("\n⏸️ 暂停...")
-                if current_trace_id:
-                    await runner.stop(current_trace_id)
+        # 配置
+        run_config = RUN_CONFIG
+        if resume_trace_id:
+            initial_messages = None
+            run_config.trace_id = resume_trace_id
+        else:
+            initial_messages = messages
+            run_config.name = f"{task_name}：测试任务"
+
+        while not should_exit:
+            if current_trace_id:
+                run_config.trace_id = current_trace_id
+
+            final_response = ""
+
+            # 如果是恢复 trace，进入交互菜单
+            if current_trace_id and initial_messages is None:
+                check_trace = await store.get_trace(current_trace_id)
+                if check_trace:
+                    if check_trace.status == "completed":
+                        print(f"\n[Trace] ✅ 已完成")
+                        print(f"  - Total messages: {check_trace.total_messages}")
+                        print(f"  - Total cost: ${check_trace.total_cost:.4f}")
+                    elif check_trace.status == "failed":
+                        print(f"\n[Trace] ❌ 已失败: {check_trace.error_message}")
+                    elif check_trace.status == "stopped":
+                        print(f"\n[Trace] ⏸️ 已停止")
+                        print(f"  - Total messages: {check_trace.total_messages}")
+                    else:
+                        print(f"\n[Trace] 📊 状态: {check_trace.status}")
+                        print(f"  - Total messages: {check_trace.total_messages}")
+
+                    current_sequence = check_trace.head_sequence
+
+                    menu_result = await interactive.show_menu(current_trace_id, current_sequence)
+
+                    if menu_result["action"] == "stop":
+                        break
+                    elif menu_result["action"] == "continue":
+                        new_messages = menu_result.get("messages", [])
+                        if new_messages:
+                            initial_messages = new_messages
+                            run_config.after_sequence = menu_result.get("after_sequence")
+                        else:
+                            initial_messages = []
+                            run_config.after_sequence = None
+                        continue
+                    break
+
+            if initial_messages is None:
+                initial_messages = []
+
+            print(f"{'▶️ 开始执行...' if not current_trace_id else '▶️ 继续执行...'}")
+
+            # 执行 Agent
+            paused = False
+            try:
+                async for item in runner.run(messages=initial_messages, config=run_config):
+                    # 检查用户中断
+                    cmd = interactive.check_stdin()
+                    if cmd == 'pause':
+                        print("\n⏸️ 正在暂停执行...")
+                        if current_trace_id:
+                            await runner.stop(current_trace_id)
+                        await asyncio.sleep(0.5)
+
+                        menu_result = await interactive.show_menu(current_trace_id, current_sequence)
+
+                        if menu_result["action"] == "stop":
+                            should_exit = True
+                            paused = True
+                            break
+                        elif menu_result["action"] == "continue":
+                            new_messages = menu_result.get("messages", [])
+                            if new_messages:
+                                initial_messages = new_messages
+                                after_seq = menu_result.get("after_sequence")
+                                if after_seq is not None:
+                                    run_config.after_sequence = after_seq
+                                paused = True
+                                break
+                            else:
+                                initial_messages = []
+                                run_config.after_sequence = None
+                                paused = True
+                                break
+
+                    elif cmd == 'quit':
+                        print("\n🛑 用户请求停止...")
+                        if current_trace_id:
+                            await runner.stop(current_trace_id)
+                        should_exit = True
+                        break
+
+                    # 处理 Trace 对象
+                    if isinstance(item, Trace):
+                        current_trace_id = item.trace_id
+                        if item.status == "running":
+                            print(f"[Trace] 开始: {item.trace_id[:8]}...")
+                        elif item.status == "completed":
+                            print(f"\n[Trace] ✅ 完成")
+                            print(f"  - Total messages: {item.total_messages}")
+                            print(f"  - Total cost: ${item.total_cost:.4f}")
+                        elif item.status == "failed":
+                            print(f"\n[Trace] ❌ 失败: {item.error_message}")
+                        elif item.status == "stopped":
+                            print(f"\n[Trace] ⏸️ 已停止")
+
+                    # 处理 Message 对象
+                    elif isinstance(item, Message):
+                        current_sequence = item.sequence
+
+                        if item.role == "assistant":
+                            content = item.content
+                            if isinstance(content, dict):
+                                text = content.get("text", "")
+                                tool_calls = content.get("tool_calls")
+
+                                if text and not tool_calls:
+                                    final_response = text
+                                    print(f"\n[Response] Agent 回复：")
+                                    print(text)
+                                elif text:
+                                    preview = text[:150] + "..." if len(text) > 150 else text
+                                    print(f"[Assistant] {preview}")
+
+                        elif item.role == "tool":
+                            content = item.content
+                            tool_name = "unknown"
+                            if isinstance(content, dict):
+                                tool_name = content.get("tool_name", "unknown")
+
+                            if item.description and item.description != tool_name:
+                                desc = item.description[:80] if len(item.description) > 80 else item.description
+                                print(f"[Tool Result] ✅ {tool_name}: {desc}...")
+                            else:
+                                print(f"[Tool Result] ✅ {tool_name}")
+
+            except Exception as e:
+                print(f"\n执行出错: {e}")
+                import traceback
+                traceback.print_exc()
+
+            if paused:
+                if should_exit:
+                    break
+                continue
+
+            if should_exit:
                 break
 
-            if isinstance(item, Trace):
-                current_trace_id = item.trace_id
-                if item.status == "running":
-                    print(f"[Trace] 开始: {item.trace_id[:8]}...")
-                elif item.status == "completed":
-                    print(f"\n[Trace] ✅ 完成 (消息: {item.total_messages}, 费用: ${item.total_cost:.4f})")
-                elif item.status == "failed":
-                    print(f"\n[Trace] ❌ 失败: {item.error_message}")
-
-            elif isinstance(item, Message):
-                current_sequence = item.sequence
-                if item.role == "assistant":
-                    content = item.content
-                    if isinstance(content, dict):
-                        text = content.get("text", "")
-                        tool_calls = content.get("tool_calls")
-                        if text and not tool_calls:
-                            print(f"\n[Response] {text}")
-                        elif text:
-                            preview = text[:150] + "..." if len(text) > 150 else text
-                            print(f"[Assistant] {preview}")
-
-                elif item.role == "tool":
-                    content = item.content
-                    tool_name = content.get("tool_name", "unknown") if isinstance(content, dict) else "unknown"
-                    desc = item.description or ""
-                    if desc and desc != tool_name:
-                        desc = desc[:80]
-                        print(f"[Tool] ✅ {tool_name}: {desc}...")
+            # Runner 退出后显示交互菜单
+            if current_trace_id:
+                menu_result = await interactive.show_menu(current_trace_id, current_sequence)
+
+                if menu_result["action"] == "stop":
+                    break
+                elif menu_result["action"] == "continue":
+                    new_messages = menu_result.get("messages", [])
+                    if new_messages:
+                        initial_messages = new_messages
+                        run_config.after_sequence = menu_result.get("after_sequence")
                     else:
-                        print(f"[Tool] ✅ {tool_name}")
+                        initial_messages = []
+                        run_config.after_sequence = None
+                    continue
+            break
 
     except KeyboardInterrupt:
         print("\n\n用户中断 (Ctrl+C)")
         if current_trace_id:
             await runner.stop(current_trace_id)
-    finally:
-        if km_task and not km_task.done():
-            print("正在关闭 Knowledge Manager...")
-            km_task.cancel()
-            try:
-                await km_task
-            except asyncio.CancelledError:
-                pass
 
-        try:
-            await kill_browser_session()
-        except Exception:
-            pass
+    # 输出结果
+    if final_response:
+        print()
+        print("=" * 60)
+        print("Agent 响应:")
+        print("=" * 60)
+        print(final_response)
+        print("=" * 60)
+        print()
+
+        output_file = output_dir / "result.txt"
+        with open(output_file, 'w', encoding='utf-8') as f:
+            f.write(final_response)
+
+        print(f"✓ 结果已保存到: {output_file}")
+        print()
 
+    # 可视化提示
     if current_trace_id:
-        print(f"\nTrace ID: {current_trace_id}")
+        print("=" * 60)
+        print("可视化 Step Tree:")
+        print("=" * 60)
+        print("1. 启动 API Server:")
+        print("   python3 api_server.py")
+        print()
+        print("2. 浏览器访问:")
+        print("   http://localhost:8000/api/traces")
+        print()
+        print(f"3. Trace ID: {current_trace_id}")
+        print("=" * 60)
 
 
 if __name__ == "__main__":

examples/mini_restore/test_e2e_proxy.py

@@ -0,0 +1,144 @@
+import os
+import sys
+import json
+import time
+import requests
+import asyncio
+
+# 添加项目根目录到 Python 路径
+sys.path.insert(0, str(os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))))
+from agent.tools.builtin.toolhub import _preprocess_params
+
+ROUTER_URL = "http://127.0.0.1:8001/run_tool"
+
+def call_tool(tool_id, params):
+    print(f"\n[{tool_id}] Calling...")
+    # 就像真正的 agent 流程一样，先预处理参数（将本地文件转化为 CDN 链接）
+    processed_params = asyncio.run(_preprocess_params(params))
+    
+    resp = requests.post(ROUTER_URL, json={"tool_id": tool_id, "params": processed_params})
+    resp.raise_for_status()
+    res = resp.json()
+    if res.get("status") == "error":
+        raise Exception(f"Tool error: {res.get('error')}")
+    print(f"[{tool_id}] Success")
+    return res.get("result", {})
+
+def test_full_workflow():
+    print("=== RunComfy Workflow E2E Test ===")
+    
+    # 0. Check existing machines
+    print("\n0. Checking for existing ComfyUI Instances...")
+    status_res = call_tool("runcomfy_check_status", {})
+    servers = status_res.get("servers", [])
+    
+    server_id = None
+    for s in servers:
+        if s.get("current_status") == "Ready":
+            server_id = s.get("server_id")
+            print(f"Found Ready machine: {server_id}")
+            break
+        elif s.get("current_status") in ("Starting", "Creating") and not server_id:
+            server_id = s.get("server_id")
+            print(f"Found Starting machine: {server_id}. Note: You might need to wait for it to be Ready.")
+
+    if not server_id:
+        raise Exception("\nNo active machine found. Expecting an already running machine. Aborting.")
+    
+    print("\n1. Proceeding with existing machine...")
+
+    try:
+        print("\n2. Loading custom workflow from JSON...")
+        import os
+        workflow_path = os.path.join(os.path.dirname(__file__), "flux_depth_controlnet_workflow.json")
+        with open(workflow_path, "r", encoding="utf-8") as f:
+            workflow_api = json.load(f)
+
+        print("\n3. Injecting correct prompt to fix ControlNet mismatch...")
+        for node_id, node in workflow_api.items():
+            if node.get("class_type") == "CLIPTextEncode":
+                current_text = node["inputs"].get("text", "")
+                if "landscape" in current_text:
+                    new_prompt = (
+                        "A back-view of a person standing in front of a wooden painting easel, "
+                        "looking out at a beautiful landscape with majestic mountains and a serene cyan lake. "
+                        "Highly detailed, masterpiece, realistic photography, cinematic lighting, 8k resolution"
+                    )
+                    node["inputs"]["text"] = new_prompt
+                    print(f"  -> Successfully updated prompt in node {node_id}")
+
+        print("\n4. Executing workflow with ControlNet inputs...")
+        
+        # 收集输入图片文件列表传递给 executor
+        # 我们在这里填入包含本地路径的 url
+        # call_tool 会使用 ToolHub 的 _preprocess_params 自动把这些 url 的本地路径替换为真正的 CDN 链接
+        input_dir = os.path.join(os.path.dirname(__file__), "input")
+        input_files = [
+            {
+                "filename": "depth_map.png",
+                "type": "images",
+                "url": os.path.join(input_dir, "depth_map.png")
+            },
+            {
+                "filename": "background_bokeh_img2.png",
+                "type": "images",
+                "url": os.path.join(input_dir, "background_bokeh_img2.png")
+            },
+            {
+                "filename": "character_ref_back.png",
+                "type": "images",
+                "url": os.path.join(input_dir, "character_ref_back.png")
+            },
+            {
+                "filename": "easel_blank_canvas_img4.png",
+                "type": "images",
+                "url": os.path.join(input_dir, "easel_blank_canvas_img4.png")
+            }
+        ]
+
+        exec_res = call_tool("runcomfy_workflow_executor", {
+            "server_id": server_id,
+            "workflow_api": workflow_api,
+            "input_files": input_files 
+        })
+        
+        print("Execution finished! Prompt ID:", exec_res.get("prompt_id"))
+        
+        images = exec_res.get("images", [])
+        print("Generated images count:", len(images))
+        if images:
+            # 拿到的是 base64 或者是 CDN url
+            img_data = images[0]
+            if isinstance(img_data, dict):
+                print("First image info:", img_data.get("url") or (img_data.get("data", "")[:50] + "..."))
+            else:
+                print("First image info (raw):", img_data[:50] + "...")
+                
+                output_dir = os.path.join(os.path.dirname(__file__), "output")
+                os.makedirs(output_dir, exist_ok=True)
+                output_path = os.path.join(output_dir, "test_output.png")
+                
+                try:
+                    if img_data.startswith("http"):
+                        import requests
+                        img_resp = requests.get(img_data)
+                        img_resp.raise_for_status()
+                        with open(output_path, "wb") as fh:
+                            fh.write(img_resp.content)
+                    else:
+                        import base64
+                        with open(output_path, "wb") as fh:
+                            fh.write(base64.b64decode(img_data))
+                    print(f"🎉 成功！生成的图片已保存至: {output_path}")
+                except Exception as e:
+                    print(f"Failed to save image: {e}")
+        
+    finally:
+        # 4. Cleanup
+        print("\n4. Keep machine alive (Skip cleanup so you can check runcomfy)")
+        # 暂时跳过 stop 方便排查
+        # call_tool("runcomfy_stop_env", {"server_id": server_id})
+
+if __name__ == "__main__":
+    test_full_workflow()
+    os._exit(0)

examples/mini_restore/toolhub_test.prompt

@@ -0,0 +1,67 @@
+---
+model: qwen3.5-plus
+temperature: 0.3
+---
+
+$system$
+
+## 角色
+你是一个 ToolHub 工具测试 Agent。你的职责是通过 ToolHub 远程工具库发现并调度合适的工具（特别是 ComfyUI 相关工具）来完成复杂的图文生成任务。
+
+## 可用工具 (ToolHub)
+
+| 工具 | 用途 |
+|------|------|
+| `toolhub_search(keyword=...)` | 搜索/发现可用工具。返回 tool_id、参数、以及**生命周期分组（Group）和调用顺序**。 |
+| `toolhub_call(tool_id=..., params={...})` | 调用指定工具。图片参数传本地路径。 |
+
+## 工作流程
+
+### 第一步：环境检查
+1. 调用 `toolhub_search(keyword="runcomfy")` 或 `toolhub_search(keyword="builder")` 寻找 ComfyUI 相关工具。
+2. 仔细阅读搜索结果，识别是否有 `runcomfy_launch` -> `runcomfy_executor` -> `runcomfy_stop` 这样的序列，或 `runcomfy_workflow_builder` 这样的辅助工具。
+
+### 第二步：素材确认
+使用 `list_dir` 和 `read_file` 确认 `%input_dir%` 中的素材：
+- `character_ref_back.png`: 人物背面参考
+- `background_bokeh_img2.png`: 背景参考
+- `depth_map.png`: 深度图（适用于 ControlNet）
+- `easel_blank_canvas_img4.png`: 道具/构图参考
+
+### 第三步：设计并执行交互
+针对 `runcomfy` 任务：
+1. **构建 Workflow**: 
+   - 如果有 `runcomfy_workflow_builder` 工具，优先调用它来生成包含 ControlNet 的 workflow JSON。
+   - 必须在 workflow 中集成 ControlNet (Depth)，使用 `depth_map.png` 路径。
+   - 融入人物和背景素材的描述。
+3. **执行生命周期**:
+   - `launch_comfy_env`: 启动远程 ComfyUI 实例。记得使用 `check_comfy_env_status` 直到机器状态为 Ready 才能拿到最终的 `server_id`。
+   - `runcomfy_workflow_executor`: 传入构建好的 workflow JSON，执行图片生成任务。
+     **重要规范 - 输入图片参数格式：** 如果流程包含本地参考图/深度图输入，严禁传入随意格式，必须在 `input_files` 参数中构造成对象数组格式，例如：
+     ```json
+     "input_files": [
+         { "filename": "depth_map.png", "type": "images", "url": "%input_dir%/depth_map.png" },
+         { "filename": "character.png", "type": "images", "url": "%input_dir%/character_ref_back.png" }
+     ]
+     ```
+     （只要 `url` 中提供的是本地路径，底层代理就会自动帮我们上传并替换成远端 CDN 链接给 Executor）
+   - `runcomfy_stop_env`: 任务完成后（或失败后）务必释放实例资源。
+
+### 第四步：报告结果
+输出结构化的结果报告，包含每个步骤的状态、使用的工具、生成的图片路径。
+
+## 路径约定
+- 输入素材：`%input_dir%`
+- 输出结果：`%output_dir%`
+
+## 注意事项
+1. **严禁硬编码**: 必须根据 `toolhub_search` 的实时返回结果来决定工具参数。
+2. **异常释放**: 即便 executor 报错，也应尝试调用 `stop` (如分组定义了 stop 步骤)。
+3. **ControlNet 必选**: 任务明确要求带 ControlNet，必须在 workflow 中体现。
+
+$user$
+请执行以下任务：
+1. 检查 `%input_dir%` 下的素材文件。
+2. 使用 `runcomfy` 相关工具（可能涉及 builder 构造 workflow，以及 launch/executor/stop 生命周期），通过 ComfyUI 执行一次带 ControlNet (Depth) 的图像生成。
+3. 将 `%input_dir%/depth_map.png` 用作 ControlNet 深度引导。
+4. 将生成结果保存到 `%output_dir%` 目录，并给出详细的执行报告。

examples/mini_restore/upload.py

@@ -1,70 +0,0 @@
-import asyncio
-import os
-import sys
-import httpx
-from cyber_sdk.ali_oss import upload_localfile
-
-async def upload_image(local_file_path: str, bucket_name: str = 'aigc-admin', bucket_path: str = 'template') -> str:
-    """Uploads a local image to OSS and returns the CDN URL."""
-    print(f"Uploading {local_file_path} to {bucket_name}/{bucket_path}...")
-    # Use forward slashes so cyber_sdk's .split('/') can correctly extract filename on Windows
-    safe_path = os.path.abspath(local_file_path).replace("\\", "/")
-    result = await upload_localfile(
-        file_path=safe_path, 
-        bucket_path=bucket_path, 
-        bucket_name=bucket_name
-    )
-    print("Upload SDK Response:", result)
-    
-    oss_object_key = result.get('oss_object_key')
-    if oss_object_key:
-        cdn_url = f"https://res.cybertogether.net/{oss_object_key}"
-        return cdn_url
-    return None
-
-async def download_image(url: str, save_path: str):
-    """Downloads an image from an HTTP link and saves it locally."""
-    print(f"Downloading from {url} to {save_path}...")
-    async with httpx.AsyncClient(timeout=30.0) as client:
-        resp = await client.get(url)
-        resp.raise_for_status()
-        with open(save_path, 'wb') as f:
-            f.write(resp.content)
-    print(f"Download completed: {save_path}")
-
-async def main():
-    if len(sys.argv) < 2:
-        print("Usage:")
-        print("  Upload: python upload.py <file_path>")
-        print("  Download: python upload.py download <url> <save_path>")
-        
-        # Self-test block if no param is given
-        print("\n--- Running Self Test ---")
-        test_file = 'img_1_gen.png'
-        if not os.path.exists(test_file):
-            print(f"Creating a dummy 1x1 PNG at {test_file} for testing.")
-            with open(test_file, 'wb') as f:
-                f.write(b'\x89PNG\r\n\x1a\n\x00\x00\x00\rIHDR\x00\x00\x00\x01\x00\x00\x00\x01\x08\x06\x00\x00\x00\x1f\x15\xc4\x89\x00\x00\x00\nIDATx\x9cc\x00\x01\x00\x00\x05\x00\x01\r\n-\xb4\x00\x00\x00\x00IEND\xaeB`\x82')
-        
-        url = await upload_image(test_file, 'aigc-admin', 'crawler/image')
-        print(f"\nExtracted URL: {url}")
-        
-        if url:
-            # Download it back
-            download_path = "downloaded_dummy.png"
-            await download_image(url, download_path)
-            
-    elif sys.argv[1] == 'download':
-        if len(sys.argv) >= 4:
-            await download_image(sys.argv[2], sys.argv[3])
-        else:
-            print("Error: Missing parameters for download.")
-            print("Usage: python upload.py download <url> <save_path>")
-    else:
-        # Upload context
-        file_path = sys.argv[1]
-        url = await upload_image(file_path, 'aigc-admin', 'crawler/image')
-        print(f"\nFinal CDN URL: {url}")
-
-if __name__ == '__main__':
-    asyncio.run(main())

examples/mini_restore/workflow_loop.py

@@ -1,422 +0,0 @@
-import sys
-import os
-import asyncio
-import json
-import base64
-import re
-
-# 将项目根目录加入，方便导入内部包
-sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../../')))
-
-from agent.tools.builtin.toolhub import toolhub_call
-from agent.llm.gemini import create_gemini_llm_call
-
-from dotenv import load_dotenv
-load_dotenv()
-
-try:
-    gemini_llm_call = create_gemini_llm_call()
-except ValueError as e:
-    print(f"初始化 Gemini 失败: {e}，请检查 .env。")
-    sys.exit(1)
-
-from agent.tools.builtin.content import content_search
-
-# -----------------
-# Utility Functions
-# -----------------
-def encode_image(image_path: str) -> str:
-    with open(image_path, "rb") as image_file:
-        return base64.b64encode(image_file.read()).decode('utf-8')
-
-def get_base64_url(image_path: str) -> str:
-    b64_data = encode_image(image_path)
-    ext = image_path.split('.')[-1].lower()
-    if ext == 'jpg': ext = 'jpeg'
-    return f"data:image/{ext};base64,{b64_data}"
-
-# -----------------
-# Tools definitions
-# -----------------
-async def call_banana_tool(prompt: str, aspect_ratio: str = None, reference_image: str = None, is_final: bool = True) -> str:
-    """包装 call_banana.py 生成图片，返回一张图的路径"""
-    print(f"\n[Tool] ✨ 正在调用 call_banana 生成图片 (is_final={is_final}), Prompt: {prompt[:50]}...")
-    script_path = os.path.join(os.path.dirname(__file__), "call_banana.py")
-    
-    env = os.environ.copy()
-    env["PYTHONIOENCODING"] = "utf-8"
-    
-    cmd_args = [sys.executable, script_path, "-p", prompt]
-    if aspect_ratio:
-        cmd_args.extend(["-a", aspect_ratio])
-    if reference_image:
-        cmd_args.extend(["-i", reference_image])
-        
-    process = await asyncio.create_subprocess_exec(
-        *cmd_args,
-        stdout=asyncio.subprocess.PIPE,
-        stderr=asyncio.subprocess.PIPE,
-        env=env
-    )
-    stdout, stderr = await process.communicate()
-    output = stdout.decode('utf-8', errors='replace')
-    err_output = stderr.decode('utf-8', errors='replace')
-    if err_output:
-        output += "\n" + err_output
-    
-    match = re.search(r"已保存到本地 -> (.+)", output)
-    if match:
-        path = match.group(1).strip()
-        print(f"[Tool] ✅ call_banana 返回图片路径: {path}")
-        return path
-    else:
-        print(f"[Tool] ❌ call_banana 执行失败:\n{output}")
-        return f"Tool Execution Failed. output:\n{output}"
-
-async def search_tool(keyword: str) -> str:
-    print(f"\n[Tool] 🔍 启动小红书调研, 关键词: {keyword}")
-    try:
-        result = await content_search(platform="xhs", keyword=keyword, max_count=3)
-        return result.output
-    except Exception as e:
-        return f"查询失败: {e}"
-
-def get_agent_tools():
-    return [
-        {
-            "type": "function",
-            "function": {
-                "name": "search_tool",
-                "description": "如果需要了解某个风格如何写 Prompt（例如“写实风格提示词”），调用此工具进行小红书全网搜索，返回总结经验以更新你的参数。",
-                "parameters": {
-                    "type": "object",
-                    "properties": {
-                        "keyword": {
-                            "type": "string",
-                            "description": "搜索关键词"
-                        }
-                    },
-                    "required": ["keyword"]
-                }
-            }
-        },
-        {
-            "type": "function",
-            "function": {
-                "name": "call_banana_tool",
-                "description": "使用此工具通过给定的详细提示词生成图片。工具将返回生成图片的本地保存路径。",
-                "parameters": {
-                    "type": "object",
-                    "properties": {
-                        "prompt": {
-                            "type": "string",
-                            "description": "英语或中文详细的生图提示词"
-                        },
-                        "aspect_ratio": {
-                            "type": "string",
-                            "description": "（可选）你期望生成的图片宽高比，例如 3:4, 16:9, 1:1，请根据目标参考图的比例传入该参数"
-                        },
-                        "reference_image": {
-                            "type": "string",
-                            "description": "（动作控制底图）如果你在这一步设 is_final=true，请将你在上一阶段生成的【辅助骨架素材(is_final=false)】产生的本地路径填入此处。绝对禁止传入原始目标照片！"
-                        },
-                        "is_final": {
-                            "type": "boolean",
-                            "description": "指示本次生成是否是本轮次的最终产物。如果你需要先生成一张『白底火柴人/3D骨架』作为辅助垫图素材，请设为 false；拿到素材后，你必须继续将它的本地路径填给 `reference_image` 并使用最终 Prompt 和 is_final=true 完成最后合成。"
-                        }
-                    },
-                    "required": ["prompt"]
-                }
-            }
-        }
-    ]
-
-
-
-# -----------------
-# Main Workflow Loop
-# -----------------
-
-def get_base64_url(image_path: str) -> str:
-    with open(image_path, "rb") as image_file:
-        b64_data = base64.b64encode(image_file.read()).decode('utf-8')
-    ext = image_path.split('.')[-1].lower()
-    if ext == 'jpg': ext = 'jpeg'
-    return f"data:image/{ext};base64,{b64_data}"
-
-async def main():
-    import argparse
-    import os
-    import json
-    
-    default_target = os.path.join(os.path.dirname(os.path.abspath(__file__)), "input", "img_1.png")
-    parser = argparse.ArgumentParser(description="多智能体画图自动优化 Workflow")
-    parser.add_argument("-t", "--target", default=default_target, help="你想逼近的目标参考图本地路径")
-    parser.add_argument("-p", "--pose", default=None, help="你提供的姿势参考图（如果有的话，给 Agent 用来走捷径垫底）")
-    parser.add_argument("-m", "--max_loops", type=int, default=15, help="优化的最大迭代论调")
-    parser.add_argument("-r", "--resume", action="store_true", help="是否从上次的 history.json 继续运行")
-    args = parser.parse_args()
-    
-    target_image = args.target
-    pose_image = args.pose
-
-    print("\n" + "="*50)
-    print("🤖 启动双 Agent 生图闭环工作流 (纯 Vision-Language 架构)")
-    print("="*50)
-    
-    if not os.path.exists(target_image):
-        print(f"⚠️ 找不到目标图片: {target_image}")
-        print("提示: 系统依然会运行寻找文件，但 Agent 2 将无法给出评估。可随便放一个图片来模拟。")
-    
-    sys_content = f"你是一个高度自治的闭环生图优化 AI 架构师。你的目标是：生成一张与【目标参考图】在主角姿势、整体结构上无限接近的图片。\n你拥有极强的视觉反思能力和 Prompt 编写能力。\n\n【核心工作流与防坑指南】：\n- 你会看到你的【目标参考图】和你的【往期历史尝试与生成结果】。\n- 请你先利用你的**多模态火眼金睛**，无情地对自己上一轮生成的图片进行找茬。绝不允许说客套话！重点对比人物骨架、姿势和构图的偏离程度。\n- 紧接着，请在反思的基础上，直接重构或调整你的 Prompt，并在一次回复中调用 `call_banana_tool` 下发生图指令！\n- 【防作弊铁律】：你**绝对禁止**直接将【目标参考图】的路径传进 `reference_image` 来作弊！如果你想用图生图垫出完美动作，必须使用【中间素材战法】亲手画一张骨架出来垫。\n- 【中间素材战法】：如果原图姿态过于刁钻复杂，**要求你必须**分两步走：\n   第一步：设置 `is_final=false` 并写一段专门用于抽出单一维度的动作骨架/白模 Prompt（如: \"a generic white 3d mannequin jumping in mid-air, clean white background, high contrast skeleton\"），专门用于抽出干净的辅助骨架。\n   第二步：拿到这只纯净骨架的本地路径后，在同回合的下一次调用中，把这只骨架当做 `reference_image` 垫进去，配合你华丽的最终描述（如: \"a neon cyberpunk assassin jumping\"），设置 `is_final=true` 完成高阶对齐兼防污染！ \n\n"
-    
-    if pose_image and os.path.exists(pose_image):
-        sys_content += f"【🔥终极开挂特权】：\n天啊！用户居然为你额外提供了一张极致完美的【姿势参考图】！既然有了这张现成的动作骨架底图，你**立刻抛弃**两步走去抽骨架的方法。你应当直接使用特权，将这张姿势参考图的绝对物理路径 `{os.path.abspath(pose_image)}` 作为 `reference_image` 无脑传给引擎，配合你的终极词汇，并在第一回合内设置 `is_final=true` 完成终极绝杀生成！\n\n"
-
-    sys_content += "流程要求：\n1. 仔细分析差异，在你的纯文本回复段落写出【犀利的反思和执行步骤】。\n2. 反思结束后，使用工具发号施令。\n3. 当调用 `is_final=true` 时，视为你的本轮彻底结束。"
-
-    system_msg = {
-        "role": "system",
-        "content": sys_content
-    }
-
-    max_loops = args.max_loops
-    current_generation_loop_count = 0
-    last_gen_info = None
-    prompt_history = [] # 记录完整的历史 Prompt 轨迹，防止反复抽卡
-    
-    history_file = os.path.join(os.path.dirname(os.path.abspath(__file__)), "history.json")
-    if args.resume and os.path.exists(history_file):
-        try:
-            with open(history_file, "r", encoding="utf-8") as f:
-                prompt_history = json.load(f)
-            if prompt_history:
-                current_generation_loop_count = len(prompt_history)
-                last_gen_info = prompt_history[-1]
-                print(f"✅ [状态恢复] 已成功从 history.json 加载 {current_generation_loop_count} 轮历史，即将开始第 {current_generation_loop_count + 1} 轮...")
-        except Exception as e:
-            print(f"⚠️ [状态恢复失败] 读取历史记录报错: {e}，将重新开始第一轮。")
-            prompt_history = []
-
-    while current_generation_loop_count < max_loops:
-        print(f"\n" + "="*40)
-        print(f"🔄 优化循环: 第 {current_generation_loop_count + 1}/{max_loops} 轮")
-        print("="*40)
-        
-        # 每轮重置上下文，只保留 system message 和含有"上次结果"的 initial user message
-        messages = [system_msg]
-        
-        if last_gen_info is None:
-            try:
-                target_b64_url = get_base64_url(target_image)
-                content_list = [
-                    {"type": "text", "text": "【首轮启动】\n这是你需要逼近的【目标参考图】。现在请你仔细观察它，提炼出一份初步生图 Prompt。\n因为是第一轮，请直接凭借直觉观察，并使用 call_banana_tool 生成原型。"},
-                    {"type": "image_url", "image_url": {"url": target_b64_url}}
-                ]
-                
-                if pose_image and os.path.exists(pose_image):
-                    content_list.append({"type": "text", "text": "并且，下面是用户良心为你提供的【开挂级·姿势参考图】！你可以直接在接下来的提示词工具调用中将此图拿去垫图！"})
-                    content_list.append({"type": "image_url", "image_url": {"url": get_base64_url(pose_image)}})
-                    
-                messages.append({
-                    "role": "user",
-                    "content": content_list
-                })
-            except Exception as e:
-                messages.append({
-                    "role": "user",
-                    "content": f"目标图片读取失败({e})，请盲猜一个初始 Prompt 用 call_banana_tool 生成。"
-                })
-        else:
-            try:
-                target_b64_url = get_base64_url(target_image)
-                user_content = [
-                    {"type": "text", "text": "【持续干预闭环】\n这是不可动摇的【目标参考图】，它是一切评判的唯一基准："},
-                    {"type": "image_url", "image_url": {"url": target_b64_url}}
-                ]
-                
-                if pose_image and os.path.exists(pose_image):
-                    user_content.append({"type": "text", "text": "【外挂辅助】\n这是不可动摇的【姿势参考图】，请毫不犹豫地拿它去填进 reference_image 控制动作："})
-                    user_content.append({"type": "image_url", "image_url": {"url": get_base64_url(pose_image)}})
-                    
-                user_content.append({"type": "text", "text": "\n==== 【你的历史试错轨迹】 ====\n为了防止你在这场试错过程中来回打转（所谓的废卡反复抽卡），我为你列出了你*从古至今*所有的失败作品和对应的提示词！请认真观察下面每一张你过去的废片：\n"})
-                
-                for i, record in enumerate(prompt_history):
-                    user_content.append({"type": "text", "text": f"-- 第 {i+1} 轮 --\n[上次使用的 Prompt]:\n{record['prompt']}\n[此轮的废片结果]:"})
-                    
-                    try: 
-                        img_path = record.get("image_paths", [record.get("image_path")])[0]
-                        # 节约上下文 Token 和视觉注意力：只渲染第一张（由于打底盲测）和最近一次的历史原图，中间的全部折叠仅保留反思文本
-                        if i == 0 or i == len(prompt_history) - 1:
-                            user_content.append({"type": "image_url", "image_url": {"url": get_base64_url(img_path)}})
-                        else:
-                            user_content.append({"type": "text", "text": "*(由于历史过于久远，中间轮次图片已省去展示，请聚焦于下面你对它的纯文本反思)*"})
-                    except:
-                        pass
-                        
-                    if record.get("feedback"):
-                        user_content.append({"type": "text", "text": f"[你在本轮结束后的反思]:\n{record['feedback']}\n"})
-                
-                user_content.append({"type": "text", "text": "====================\n\n现在，结合上述轨迹与那张【目标参考图】，请在回复中写出最新的【极度苛刻自我反思】，然后立马调用工具生成这轮新的 Prompt！"})
-                
-                messages.append({"role": "user", "content": user_content})
-                
-            except Exception as e:
-                messages.append({"role": "user", "content": f"上下文读取失败 ({e})。请重试用 call_banana_tool 生成。"})
-
-        # Agent 1 内部工具调研微循环 (Agent 1 minor logic loop)
-        agent1_finished_generation = False
-        consecutive_empty = 0
-        
-        while not agent1_finished_generation:
-            print(f"---\n💬 正在请求 Agent 1 (Prompt 师)...")
-            # 这里 Agent 1 也换成 qwen-vl-max，这样它才能看到传给它的上一轮图片
-            response = await gemini_llm_call(
-                messages=messages,
-                model="gemini-3.1-pro-preview",
-                tools=get_agent_tools()
-            )
-            
-            content = response.get("content", "")
-            tool_calls = response.get("tool_calls")
-            
-            if content:
-                print(f"\n[Agent 1 思考]:\n{content}")
-                
-            if not tool_calls and not content:
-                consecutive_empty += 1
-                if consecutive_empty >= 3:
-                    print("Agent 连续多次无有意义输出，强制跳出本轮。")
-                    break
-            else:
-                consecutive_empty = 0
-
-            # 保持上下文
-            assistant_reply = {"role": "assistant"}
-            if content: assistant_reply["content"] = content
-            if tool_calls: assistant_reply["tool_calls"] = tool_calls
-            if "raw_gemini_parts" in response: assistant_reply["raw_gemini_parts"] = response["raw_gemini_parts"]
-            messages.append(assistant_reply)
-
-            if tool_calls:
-                for tc in tool_calls:
-                    func_name = tc["function"]["name"]
-                    args_dict = json.loads(tc["function"]["arguments"])
-                    tc_id = tc["id"]
-                    
-                    if func_name == "search_tool":
-                        res = await search_tool(**args_dict)
-                        messages.append({
-                            "role": "tool",
-                            "tool_call_id": tc_id,
-                            "content": str(res)
-                        })
-                    
-                    elif func_name == "call_banana_tool":
-                        is_final = args_dict.get("is_final", True)
-                        print(f"\n⚙️ 节点发起了生图请求 (是否为终极图: {is_final})！")
-                        gen_path = await call_banana_tool(**args_dict)
-                        
-                        if os.path.exists(gen_path):
-                            ext = gen_path.split('.')[-1]
-                            import shutil
-                            if is_final:
-                                new_gen_path = f"gen_loop_{current_generation_loop_count + 1}.{ext}"
-                            else:
-                                import uuid
-                                new_gen_path = f"gen_loop_{current_generation_loop_count + 1}_material_{str(uuid.uuid4())[:8]}.{ext}"
-                            shutil.move(gen_path, new_gen_path)
-                            gen_path = new_gen_path
-                            print(f"[文件管理] 生图结果已重命名并保存为: {new_gen_path}")
-                        
-                        prompt_used = args_dict.get("prompt", "")
-                        
-                        messages.append({
-                            "role": "tool",
-                            "tool_call_id": tc_id,
-                            "content": f"已成功生成，图片路径: {os.path.abspath(gen_path)}"
-                        })
-                        
-                        if is_final:
-                            agent1_finished_generation = True
-                            current_generation_loop_count += 1
-                            
-                            last_gen_info = {
-                                "prompt": prompt_used,
-                                "image_path": gen_path,
-                                "feedback": content if content else "无反思内容"
-                            }
-                            
-                            prompt_history.append(last_gen_info)
-                            try:
-                                with open(history_file, "w", encoding="utf-8") as f:
-                                    json.dump(prompt_history, f, ensure_ascii=False, indent=2)
-                            except Exception as e:
-                                print(f"[警告] 历史记录保存失败: {e}")
-                            break # 跳出 tool_calls for loop 并进入下一大轮
-                        else:
-                            print(f"[战术回馈] 这是辅助素材，已将路径返回给 Agent1 继续思考。")
-            else:
-                # 没调工具
-                print("\n[控制中心] Agent 1 没有继续使用任何工具。结束其周期。")
-                agent1_finished_generation = True
-                break
-                
-    print("\n🎉 工作流闭环成功完成或达到了最大迭代次数。")
-    
-    # 最后由评估专家出具一份最完善的多维度最终报告
-    if len(prompt_history) > 0 and os.path.exists(target_image):
-        print("\n" + "="*50)
-        print("🏆 正在生成【专家最终多维度反馈报告】...")
-        print("="*50)
-        
-        first_gen_record = prompt_history[0]
-        last_gen_record = prompt_history[-1]
-        
-        # 兼容旧版本的单图记录和新版本的多图记录
-        first_gen = first_gen_record.get("image_paths", [first_gen_record.get("image_path")])[0]
-        last_gen = last_gen_record.get("image_paths", [last_gen_record.get("image_path")])[0]
-        
-        if first_gen and last_gen and os.path.exists(first_gen) and os.path.exists(last_gen):
-            try:
-                target_b64 = encode_image(target_image)
-                first_b64 = encode_image(first_gen)
-                last_b64 = encode_image(last_gen)
-                target_ext = target_image.split('.')[-1].lower()
-                first_ext = first_gen.split('.')[-1].lower()
-                last_ext = last_gen.split('.')[-1].lower()
-                
-                # 构建供最终分析的文字轨迹
-                full_history_text = "【历次 Prompt 与专家反馈的演进轨迹】\n"
-                for i, record in enumerate(prompt_history):
-                    full_history_text += f"-- 第 {i+1} 轮 --\n[Prompt]: {record['prompt']}\n[反馈]: {record['feedback']}\n\n"
-
-                final_messages = [
-                    {
-                        "role": "system",
-                        "content": "你是首席AI打样架构师。目前的生图迭代优化工作流已拉下帷幕。你不需要拘泥于打分，而是要通过回顾整个演进历程，总结出‘最好用的 Prompt 模板’和‘最精准的评估反馈维度模板’。"
-                    },
-                    {
-                        "role": "user",
-                        "content": [
-                            {"type": "text", "text": "【目标参考图（原图）】:"},
-                            {"type": "image_url", "image_url": {"url": f"data:image/{target_ext if target_ext != 'jpg' else 'jpeg'};base64,{target_b64}"}},
-                            {"type": "text", "text": "这是最初第1轮盲试的生成图:"},
-                            {"type": "image_url", "image_url": {"url": f"data:image/{first_ext if first_ext != 'jpg' else 'jpeg'};base64,{first_b64}"}},
-                            {"type": "text", "text": f"这是经过迭代后的【最终生成图】:"},
-                            {"type": "image_url", "image_url": {"url": f"data:image/{last_ext if last_ext != 'jpg' else 'jpeg'};base64,{last_b64}"}},
-                            {"type": "text", "text": f"下面是 {len(prompt_history)} 轮迭代中，Prompt 和专家反馈的完整变迁记录：\n\n{full_history_text}\n\n请结合首尾图片的巨大差异以及中间的踩坑过程，深度复盘：\n1. 在构建生图 Prompt 时，哪些描述方式、句型或结构最能有效命中模型？请提炼出一个【最终版高转化率 Prompt 语法模板】。\n2. 在进行视觉反馈时，哪些维度的批评和建议对 Prompt 师是最具指导意义的？请提炼出一个【最终版高维度视觉评估反馈模板】。\n这两个模版需要具备极强的通用性和实战复用价值！"}
-                        ]
-                    }
-                ]
-                
-                response = await gemini_llm_call(
-                    messages=final_messages,
-                    model="gemini-3.1-pro-preview"
-                )
-                print(f"\n[Agent 2] 📋 【最终多维度评估报告】:\n{response['content']}\n")
-            except Exception as e:
-                print(f"最终报告生成失败: {e}")
-
-if __name__ == "__main__":
-    asyncio.run(main())