feat(asset-catalog): 简化分析工具+新增汇总和AI推理脚本

- analyze_meshes: 移除失败的 LineTrace 和六面分析，只保留基础属性 (包围盒、锚点、复杂度、Socket、材质槽位、自动Tag) - build_database: 新增离线汇总脚本，将 meshes/*.json 扁平化为统一的 asset_database.json - ai_inference: 新增 GLM API 批量推理脚本，按类别分批推断语义信息 (朝向、表面高度、功能描述、交互属性、降级方案) Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-10 00:50:56 +08:00
parent a951009dae
commit 7a2436a612
3 changed files with 518 additions and 222 deletions
@@ -0,0 +1,366 @@
+"""GLM API 批量推理脚本 — 资产语义增强。
+
+读取 asset_database.json，按类别分批调用 GLM API（OpenAI 兼容接口），
+推断每个资产的朝向、功能、交互属性等语义信息。
+
+此脚本不依赖 UE5，可离线运行。
+
+环境变量:
+  GLM_API_KEY   — API Key（必需）
+  GLM_API_BASE  — API Base URL（默认 https://open.bigmodel.cn/api/paas/v4）
+  GLM_MODEL     — 模型名（默认 glm-4-flash）
+
+用法:
+  python ai_inference.py [--input FILE] [--output FILE] [--batch-size N] [--dry-run]
+"""
+
+import sys
+import os
+import json
+import argparse
+import time
+from pathlib import Path
+
+_SCRIPT_DIR = Path(__file__).resolve().parent
+if str(_SCRIPT_DIR) not in sys.path:
+    sys.path.insert(0, str(_SCRIPT_DIR))
+
+from _common import read_json, write_json, iso_now
+
+TAG = "[AIInference]"
+
+# ---- 配置 ----
+DEFAULT_API_BASE = "https://open.bigmodel.cn/api/paas/v4"
+DEFAULT_MODEL = "glm-4-flash"
+DEFAULT_BATCH_SIZE = 20
+
+# ---- Prompt 模板 ----
+SYSTEM_PROMPT = """你是一个 UE5 游戏开发助手，专注于为 3D 模型资产提取语义信息。
+
+你会收到一批 StaticMesh 资产的几何数据（名称、尺寸、锚点、材质等），
+需要推断每个资产的以下语义字段：
+
+1. facing_direction: 主朝向方向
+   - 格式: "+X", "-X", "+Y", "-Y", null
+   - 判断依据: 根据资产类型和包围盒尺寸推断"正面"方向
+   - 对称物体（桌子、球、圆柱等）设为 null
+   - 电视/显示器: 屏幕面方向; 门: 开门方向; 椅子: 坐面方向
+
+2. surface_height_cm: 可用水平表面高度 (cm)
+   - 如桌面高度、台面高度、床面高度
+   - 对于无水平表面的物体（墙体、栏杆）设为 null
+   - 使用锚点类型判断: CENTER_BOTTOM 时直接取 max.z
+
+3. function_description: 一句话功能描述 (中文, ≤50字)
+
+4. placement_notes: 放置注意事项 (中文, ≤80字)
+   - 锚点位置提醒、旋转建议、缩放注意等
+
+5. interaction_hints: 交互提示标签数组
+   - 可选值: can_sit, can_place_items, can_open, can_interact_screen,
+     climbable, electronics, lighting, container, decoration,
+     pass_through, driveable, weapon, vegetation
+   - 可多选
+
+6. plausible_substitutes: BSP 基础体降级方案 (中文, 或 null)
+   - 当原始资产不可用时，用什么 Cube/Cylinder + 缩放 + 材质替代
+
+你必须返回严格的 JSON 数组，每个元素对应输入中的一个资产，顺序一致。
+不要输出任何 JSON 之外的内容。"""
+
+BATCH_PROMPT_TEMPLATE = """以下是 {category} 类别的 {count} 个 StaticMesh 资产数据：
+
+{assets_json}
+
+请为每个资产推断语义信息，返回 JSON 数组。每个元素包含:
+facing_direction, surface_height_cm, function_description,
+placement_notes, interaction_hints, plausible_substitutes"""
+
+
+def _get_api_config():
+    """从环境变量读取 API 配置。"""
+    api_key = os.environ.get("GLM_API_KEY", "")
+    api_base = os.environ.get("GLM_API_BASE", DEFAULT_API_BASE)
+    model = os.environ.get("GLM_MODEL", DEFAULT_MODEL)
+
+    if not api_key:
+        print(f"{TAG} 错误: 未设置 GLM_API_KEY 环境变量")
+        print(f"{TAG}   设置方法: set GLM_API_KEY=your_key_here")
+        return None
+
+    return {
+        "api_key": api_key,
+        "api_base": api_base,
+        "model": model,
+    }
+
+
+def _classify_batch_category(tags: dict) -> str:
+    """根据 tags 推断批次类别名。"""
+    style = tags.get("style", [])
+    if not style:
+        return "未分类"
+    # 取第一个非空 style tag 作为类别
+    return style[0] if style[0] != "unknown" else "未分类"
+
+
+def _prepare_batches(assets: dict, batch_size: int) -> list:
+    """将资产按类别分批。
+
+    Returns:
+        list of (category_name, batch_index, [asset_keys])
+    """
+    # 按类别分组
+    by_category = {}
+    for name, data in assets.items():
+        cat = _classify_batch_category(data.get("tags", {}))
+        by_category.setdefault(cat, []).append(name)
+
+    # 每个类别内按 batch_size 切分
+    batches = []
+    for cat, keys in sorted(by_category.items()):
+        for i in range(0, len(keys), batch_size):
+            batch_keys = keys[i:i + batch_size]
+            batches.append((cat, i // batch_size + 1, batch_keys))
+
+    return batches
+
+
+def _format_asset_for_prompt(name: str, data: dict) -> dict:
+    """提取 prompt 中需要的精简字段。"""
+    bb = data.get("bounding_box", {})
+    pivot = data.get("pivot", {})
+    tags = data.get("tags", {})
+
+    return {
+        "name": name,
+        "size_cm": bb.get("size"),
+        "pivot_type": pivot.get("type"),
+        "pivot_offset_xy": pivot.get("offset_xy"),
+        "material_count": len(data.get("material_slots", [])),
+        "socket_count": len(data.get("sockets", [])),
+        "tags_style": tags.get("style", []),
+    }
+
+
+def _call_glm_api(config: dict, category: str, assets_json: str) -> list:
+    """调用 GLM API 进行推理。
+
+    Returns:
+        推理结果列表，失败时返回空列表
+    """
+    try:
+        from openai import OpenAI
+    except ImportError:
+        print(f"{TAG} 错误: 需要安装 openai 包")
+        print(f"{TAG}   pip install openai")
+        return []
+
+    client = OpenAI(
+        api_key=config["api_key"],
+        base_url=config["api_base"],
+    )
+
+    user_prompt = BATCH_PROMPT_TEMPLATE.format(
+        category=category,
+        count=assets_json.count('"name"'),
+        assets_json=assets_json,
+    )
+
+    try:
+        response = client.chat.completions.create(
+            model=config["model"],
+            messages=[
+                {"role": "system", "content": SYSTEM_PROMPT},
+                {"role": "user", "content": user_prompt},
+            ],
+            temperature=0.1,
+            response_format={"type": "json_object"},
+        )
+    except Exception as e:
+        print(f"{TAG} API 调用失败: {e}")
+        return []
+
+    content = response.choices[0].message.content.strip()
+    if not content:
+        print(f"{TAG} API 返回空内容")
+        return []
+
+    # 解析 JSON
+    try:
+        result = json.loads(content)
+        # 如果返回的是 {"results": [...]} 形式，提取数组
+        if isinstance(result, dict):
+            for key in ("results", "data", "assets", "items"):
+                if key in result and isinstance(result[key], list):
+                    return result[key]
+            # 如果找不到数组，可能是单条结果被包在 dict 里
+            print(f"{TAG} 警告: 返回 JSON 不是数组，尝试整体作为单条: {list(result.keys())}")
+            return [result]
+        if isinstance(result, list):
+            return result
+    except json.JSONDecodeError as e:
+        print(f"{TAG} JSON 解析失败: {e}")
+        print(f"{TAG} 原始响应前 200 字: {content[:200]}")
+        return []
+
+    return []
+
+
+def run_inference(
+    input_file: Path,
+    output_file: Path,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+    dry_run: bool = False,
+) -> dict:
+    """执行批量 AI 推理。
+
+    Args:
+        input_file: asset_database.json 路径
+        output_file: 输出路径（通常与 input 相同，就地更新）
+        batch_size: 每批资产数量
+        dry_run: 只打印批次信息，不调用 API
+
+    Returns:
+        统计信息 dict
+    """
+    # 读取数据库
+    if not input_file.exists():
+        print(f"{TAG} 错误: {input_file} 不存在，请先运行 build_database")
+        return {"status": "error", "reason": "no input file"}
+
+    db = read_json(input_file)
+    assets = db.get("assets", {})
+    if not assets:
+        print(f"{TAG} 错误: 数据库中无资产")
+        return {"status": "error", "reason": "empty assets"}
+
+    # 过滤: 跳过已推理的
+    pending = {}
+    for name, data in assets.items():
+        if data.get("ai_inference") is not None:
+            continue
+        pending[name] = data
+
+    print(f"{TAG} 数据库共 {len(assets)} 个资产，待推理 {len(pending)} 个")
+
+    if not pending:
+        print(f"{TAG} 全部已推理，无需处理")
+        return {"status": "ok", "total": len(assets), "inferred": 0, "skipped": len(assets)}
+
+    # 分批
+    batches = _prepare_batches(pending, batch_size)
+    print(f"{TAG} 分为 {len(batches)} 批")
+
+    if dry_run:
+        for cat, idx, keys in batches:
+            print(f"{TAG}   批次 [{cat}] #{idx}: {len(keys)} 个资产")
+            for k in keys[:3]:
+                print(f"{TAG}     - {k}")
+            if len(keys) > 3:
+                print(f"{TAG}     ... 还有 {len(keys) - 3} 个")
+        return {"status": "ok", "dry_run": True, "batches": len(batches)}
+
+    # 获取 API 配置
+    config = _get_api_config()
+    if not config:
+        return {"status": "error", "reason": "no api key"}
+
+    print(f"{TAG} API: {config['api_base']} 模型: {config['model']}")
+
+    # 逐批推理
+    total_inferred = 0
+    total_failed = 0
+
+    for batch_idx, (cat, batch_num, keys) in enumerate(batches):
+        print(f"{TAG} 批次 [{batch_idx + 1}/{len(batches)}] [{cat}] #{batch_num}: {len(keys)} 个资产...")
+
+        # 构造 prompt 数据
+        prompt_assets = []
+        for k in keys:
+            prompt_assets.append(_format_asset_for_prompt(k, pending[k]))
+
+        assets_json = json.dumps(prompt_assets, ensure_ascii=False, indent=2)
+
+        # 调用 API
+        results = _call_glm_api(config, cat, assets_json)
+
+        if not results:
+            print(f"{TAG}   ⚠️ 推理失败，跳过此批")
+            total_failed += len(keys)
+            continue
+
+        # 合并结果
+        matched = 0
+        for i, k in enumerate(keys):
+            if i < len(results):
+                assets[k]["ai_inference"] = results[i]
+                matched += 1
+                total_inferred += 1
+            else:
+                print(f"{TAG}   ⚠️ {k}: API 返回结果不足")
+
+        print(f"{TAG}   ✅ 匹配 {matched}/{len(keys)}")
+
+        # 每批都保存一次（断点续跑）
+        db["assets"] = assets
+        # 更新 metadata
+        if "ai_inference_metadata" not in db["metadata"]:
+            db["metadata"]["ai_inference_metadata"] = {
+                "model": config["model"],
+                "api_base": config["api_base"],
+                "started_at": iso_now(),
+            }
+        db["metadata"]["ai_inference_metadata"]["last_batch"] = batch_idx + 1
+        db["metadata"]["ai_inference_metadata"]["total_batches"] = len(batches)
+        db["metadata"]["ai_inference_metadata"]["inferred"] = total_inferred
+        db["metadata"]["ai_inference_metadata"]["failed"] = total_failed
+        write_json(db, output_file)
+
+        # 限速: 避免触发 API 限流
+        if batch_idx < len(batches) - 1:
+            time.sleep(1)
+
+    db["metadata"]["ai_inference_metadata"]["completed_at"] = iso_now()
+    write_json(db, output_file)
+
+    print(f"{TAG} 推理完成: {total_inferred} 成功, {total_failed} 失败")
+    return {
+        "status": "ok",
+        "total": len(assets),
+        "inferred": total_inferred,
+        "failed": total_failed,
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser(description="GLM API 批量推理 — 资产语义增强")
+    parser.add_argument(
+        "--input", type=str,
+        default=str(_SCRIPT_DIR.parent.parent / "output" / "asset-catalog" / "asset_database.json"),
+        help="输入 asset_database.json 路径",
+    )
+    parser.add_argument(
+        "--output", type=str, default=None,
+        help="输出路径 (默认与 input 相同，就地更新)",
+    )
+    parser.add_argument(
+        "--batch-size", type=int, default=DEFAULT_BATCH_SIZE,
+        help=f"每批资产数量 (默认 {DEFAULT_BATCH_SIZE})",
+    )
+    parser.add_argument(
+        "--dry-run", action="store_true",
+        help="只打印批次信息，不调用 API",
+    )
+    args = parser.parse_args()
+
+    input_file = Path(args.input)
+    output_file = Path(args.output) if args.output else input_file
+
+    result = run_inference(input_file, output_file, args.batch_size, args.dry_run)
+    if result.get("status") != "ok" and not result.get("dry_run"):
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    main()