diff --git a/.opencode/agent/story-scene-brain.md b/.opencode/agent/story-scene-brain.md new file mode 100644 index 00000000..9c840646 --- /dev/null +++ b/.opencode/agent/story-scene-brain.md @@ -0,0 +1,45 @@ +--- +description: 剧情舞台导演：把一句话剧情解析成场景需求，经 creative-brain 生成计划、用 scene_director kit 搭台、scene-qc 质检迭代，产出合格舞台与渲染截图。 +mode: subagent +temperature: 0.2 +--- + +你是 **EVEngine 剧情舞台导演**（story-scene brain）。玩家/导演给一句话剧情，你负责把它变成 +一个 **质量过关、可直接截图交付的 3D 舞台**：自动摆道具、调光照、架摄像机、多视角质检、迭代修复。 + +## 能力定位 + +你是一站式编排中枢，复用它人已落地的基础 AI 工具，不重复造轮子： + +| 环节 | 工具 | +|------|------| +| 故事 → 舞台要点 | `tools/story-scene-agent/story_scene_agent/story.py`（parse_stage） | +| 生成计划 | `tools/creative-brain`（intent → assets → layout → plan） | +| 搭台 | 引擎 MCP `eve_scene_modify` / `eve_camera_generate` + `scene_director.nut` kit | +| 质检/修复 | `tools/scene-qc-agent`（ReAct，三源融合，硬上限） | + +## 调用方式 + +```bash +# 干跑（无引擎/模型，验证闭环） +cd tools/story-scene-agent +python -m story_scene_agent.cli "月光下骑士与巨龙在古堡前对峙" --dry-run --trace + +# 真实引擎（先起：eve run --debug --mcp-port=7529 examples/ai-stage） +python -m story_scene_agent.cli \ + "月光下骑士与巨龙在古堡前对峙" --port 7529 --out-dir stage_duel --trace +``` + +## 执行要点 + +1. **先干跑再真机**：先 `--dry-run` 确认故事能解析、计划能生成、质检闭环能收敛； + 确认引擎 MCP / 模型可用后再去掉 `--dry-run`。 +2. 剧情里强调的氛围 / 时段（月光、黄昏、森林、战场…）会覆盖 creative-brain 的默认光照， + 由 `parse_stage` 抽取（可加 `OPENAI_API_KEY` 启用 LLM 精修，否则关键词启发式）。 +3. `--max-rounds` 控制质检迭代上限（默认 3），避免死循环；不达标会自动调整道具 / 光照 / 摄像机重检。 +4. 报告 + 截图落在 `--out-dir`，其中 `story_scene_report.json` 含 stage/plan/build/qc 全链路。 + +## 输出 + +- 终端打印 `{passed, score, rounds_used, props, dry_run, report}`。 +- `/story_scene_report.json`：剧情、舞台要点、生成计划、搭台日志、质检报告与历史。 diff --git a/docs/dev/AI场景导演.md b/docs/dev/AI场景导演.md new file mode 100644 index 00000000..06f32ae7 --- /dev/null +++ b/docs/dev/AI场景导演.md @@ -0,0 +1,131 @@ +# AI 场景导演（Scene Director） + +> 状态：整合打通（story-scene-agent 编排 + scene_director kit + 引擎 MCP 新工具）。 +> 编排入口：`tools/story-scene-agent/`；搭台 kit：`src/scripts/scene_director.nut`； +> 引擎侧调度：`src/engine/devtools/McpServer.cpp`。 + +把散落在仓库各处的 AI 能力（creative-brain 生成、scene-qc 质检修复、vision-prefilter +预筛选、RenderVision 渲染描述、asset-pipeline 素材供给、引擎 MCP 驱动）整合成一条 +**可被 Agent 端到端驱动**的流水线：给一句话剧情，就能在引擎里自动摆道具、调光照、架摄像机、 +多视角截图、质检迭代修复，最终交付一个质量过关的舞台。 + +## 目标与效果 + +- **Agent 可驱动**：无需手改脚本，Agent 通过 MCP 直接搭建 / 调整 / 查看 / 改进场景。 +- **自动摆物 / 调摄像机 / 看渲染结果 / 迭代改进**：`eve_scene_modify` 摆物与调光， + `eve_camera_generate` 生成机位，`eve_screenshot` / `capture_render_frame` / `eve_render_describe` + 看结果，scene-qc 自动修复迭代。 +- **剧情 → 合格舞台**：`story-scene-agent` 一句话剧情 → 舞台要点 → 计划 → 搭台 → 质检 → 交付。 + +## 端到端流程 + +```mermaid +flowchart LR + Story["一句话剧情"] --> Stage["story-scene-agent
parse_stage 舞台要点"] + Stage --> CB["creative-brain
intent→assets→layout→plan"] + CB --> SD["scene_director kit
(eve_scene_modify 摆物/光照/摄像机)"] + SD --> Shot["eve_screenshot / capture_render_frame
多视角截图"] + Shot --> QC["scene-qc-agent
本地小模型筛查→高危VLM复核→融合"] + QC -->|不达标| Fix["Fixer 生成修改计划
eve_scene_modify 落地"] + Fix --> SD + QC -->|达标| Out["报告 + 渲染截图"] + VP["vision-prefilter
(可选本地预筛)"] -.-> QC + RV["RenderVision
eve_render_describe"] -.-> QC +``` + +## 组件与打通点 + +| 组件 | 角色 | 打通方式 | +|------|------|----------| +| `scene_director.nut` | 搭台 kit：spawn/move/scale/rotate/remove/material/lighting/camera/cameras/info/modify/reset | 嵌入为 `eve.sceneDirectorScript`，MCP 工具自动安装 | +| `eve_scene_modify` | Agent 摆物/调整的统一入口（action+target+params） | 薄调度，转发 `scene_director.modify` | +| `eve_scene_info` | 权威场景真值（props/pos/scale/yaw/tint） | 转发 `scene_director.info` | +| `eve_camera_generate` | 围绕已摆场景生成标准质检机位 | 转发 `scene_director.cameras` | +| `eve_scene_reset` | 清场重置 | 转发 `scene_director.reset` | +| `tools/creative-brain` | 意图解析 / 资产匹配 / 布局 / 计划 | `brain.mcp` 现在把计划步骤翻译成真实 `scene_director.modify` 动作 | +| `tools/scene-qc-agent` | ReAct 巡检 / 三源融合 / 修复迭代 | `config.example.json` 已对齐到真实引擎工具名；截图支持读盘→base64 | +| `tools/vision-prefilter` | 本地轻量视觉预筛选（可选） | 可作 scene-qc `local_screen` 的 OpenAI 兼容后端 | +| `RenderVision` | 引擎内置渲染描述 | `eve_render_describe` 供质检 / 调试 | + +## 引擎 MCP 新工具（Agent 驱动核心） + +| 工具 | 说明 | 参数 | +|------|------|------| +| `eve_scene_director_install` | 安装搭台 kit（幂等，工具首次使用自动调用） | — | +| `eve_scene_director_status` | kit 状态（installed / propCount / hasCamera） | — | +| `eve_scene_reset` | 清空道具、摄像机、恢复默认光照 | — | +| `eve_scene_modify` | 场景动作 | `action`, `target`, `params` | +| `eve_camera_generate` | 生成标准机位 | `count` | +| `eve_scene_info` | 场景真值 | — | + +### `eve_scene_modify` 的 action + +`add_object|spawn|place`、`move_object|move`、`scale`、`rotate|rotation`、 +`remove_object|remove`、`visibility`、`material`、`lighting|set_lighting`、 +`camera`、`cameras`、`info`、`list`、`reset`。 + +参数示例： + +```jsonc +// 摆一棵树 +{ "action": "add_object", "target": "tree_0", + "params": { "id": "tree_0", "kind": "tree", "pos": [4, 0, 3], "scale": [1.2,1.2,1.2], "seed": 5 } } +// 调月光 +{ "action": "lighting", "target": "scene", + "params": { "timeOfDay": "night", "atmosphere": "moonlit", "intensity": 0.8 } } +// 架摄像机 +{ "action": "camera", "target": "scene", "params": { "eye": [0,8,16], "target": [0,1,0], "fov": 55 } } +``` + +### 支持的道具 `kind` + +- 图元：`box|cube`、`sphere|ball`、`cylinder|pillar|column|trunk` +- 程序化网格：`tree`、`rock`、`bush`、`skyscraper`、`hexplanet`、`marchingcubes`、 + `fence`、`stonewall`、`bridge`、`greatwall`、`hedge`、`chevaldefrise` +- 地面：`ground|floor`（自动压扁） + +## 编排入口：story-scene-agent + +```bash +# 起引擎空舞台 + MCP +eve run --debug --mcp-port=7529 examples/ai-stage + +# 干跑（无引擎/模型，验证闭环） +cd tools/story-scene-agent +python -m story_scene_agent.cli "月光下骑士与巨龙在古堡前对峙" --dry-run + +# 真实搭台 + 质检 +python -m story_scene_agent.cli \ + "月光下骑士与巨龙在古堡前对峙" --port 7529 --out-dir stage_duel +``` + +产物：`stage_duel/story_scene_report.json`（stage / plan / build / qc / final）+ 截图。 + +## Agent 工作流建议（参考 TEngine 的 AI 开发工作流） + +1. **任务分级**：简单加一个道具 → 直接调 `eve_scene_modify`；整幕搭台 → 用 + `story-scene-agent` 全链路；跨模块（素材缺失、特效、动画）→ 并行触发 + `asset-pipeline-brain` / `scene-qc-brain`。 +2. **按需查规范**：Agent 摆物前先 `eve_scene_director_status` + 读本文件「支持的 kind / + action」，避免猜 API；会话内缓存，不重复查。 +3. **先干跑再真机**：`--dry-run` 验证闭环，再连真引擎。 +4. **冲突标注**：kit / 文档与实际引擎行为不符时，以代码为准并记录差异。 + +## 测试 + +```bash +# Python 侧（creative-brain / story-scene-agent 的改动） +python -m pytest tools/creative-brain/tests tools/story-scene-agent/tests -q +# scene-qc +(cd tools/scene-qc-agent && python -m unittest discover -s tests -p "test_*.py") + +# 引擎侧（MCP 工具 / kit） +./build/-debug/test/unit_test --testcase='^devtools\.mcp\..*$' +``` + +## 后续可扩展 + +- scene_director 支持外部模型（`.glb`）与动画挂接。 +- `eve_scene_modify` 增加 `spawn_effect` / 粒子 / 音频编排。 +- 场景持久化为可回放的编排脚本（story → nut 剧本）。 +- 与 `eve test` 场景脚本联动的回归断言。 diff --git a/docs/dev/README.md b/docs/dev/README.md index b5faf1e9..70b97628 100644 --- a/docs/dev/README.md +++ b/docs/dev/README.md @@ -10,6 +10,7 @@ - [命令行设计](命令行设计.md) - [测试覆盖](测试覆盖.md) - [AI 与 MCP 支持](AI与MCP支持.md) +- [AI 场景导演](AI场景导演.md)（剧情 → 自动搭台 → 质检迭代 → 交付） ## 功能设计 diff --git a/examples/ai-stage/README.md b/examples/ai-stage/README.md new file mode 100644 index 00000000..24a29743 --- /dev/null +++ b/examples/ai-stage/README.md @@ -0,0 +1,25 @@ +# AI Stage + +一个"空舞台"示例：启动时把 `scene_director` 搭台 kit 安装进虚拟机，并摆好默认地面 / +光照 / 摄像机。AI Agent（`tools/story-scene-agent`、`creative-brain`、 +`scene-qc-agent`）通过引擎 MCP 直接往这个舞台上摆放道具、调整摄像机、截图质检。 + +## 运行 + +```bash +make build/win32-debug +# 或先起引擎 MCP： +build/win32-debug/src/engine/eve.exe run --debug --mcp-port=7529 examples/ai-stage +``` + +## Agent 接入 + +```bash +# 一句话剧情 → 搭台 → 质检迭代 → 报告 + 截图 +cd tools/story-scene-agent +python -m story_scene_agent.cli "月光下骑士与巨龙在古堡前对峙" --port 7529 --out-dir stage_duel +# 干跑（无需引擎/模型）验证闭环 +python -m story_scene_agent.cli "月光下骑士与巨龙在古堡前对峙" --dry-run +``` + +详见 [`docs/dev/AI场景导演.md`](../../docs/dev/AI场景导演.md)。 \ No newline at end of file diff --git a/examples/ai-stage/config.nut b/examples/ai-stage/config.nut new file mode 100644 index 00000000..7a9773f8 --- /dev/null +++ b/examples/ai-stage/config.nut @@ -0,0 +1,7 @@ +config <- { + width = 1024 + height = 768 + title = "EVEngine AI Stage" + debug = true + hotReload = false +}; \ No newline at end of file diff --git a/examples/ai-stage/main.nut b/examples/ai-stage/main.nut new file mode 100644 index 00000000..5500216a --- /dev/null +++ b/examples/ai-stage/main.nut @@ -0,0 +1,52 @@ +// AI Stage — an empty 3D stage that hosts the `scene_director` authoring kit. +// +// Purpose: a neutral, always-ready scene the agents build into over MCP. +// - story-scene-agent / creative-brain place props via eve_scene_modify +// - scene-qc / RenderVision frame cameras and screenshot via MCP +// Run with DevTools + MCP so agents can connect: +// eve run --debug --mcp-port=7529 examples/ai-stage + +// Install the embedded scene-director kit (idempotent). `sceneDirectorScript` +// is baked into the engine from src/scripts/scene_director.nut. +if ("sceneDirectorScript" in eve && eve.sceneDirectorScript != "") { + try { + compilestring(eve.sceneDirectorScript)(); + } catch (e) { + print("scene_director install failed: " + e + "\n"); + } +} else if (file_exists("scene_director.nut")) { + try { + dofile("scene_director.nut"); + } catch (e) { + print("scene_director file load failed: " + e + "\n"); + } +} + +if (!("sd" in getroottable())) sd <- null; +if (scene_director != null) sd = scene_director; + +// Minimal default stage so the window is never empty before an agent builds. +if (sd != null) { + sd.reset(); + sd.spawn({ + id = "terrain", kind = "ground", + pos = [0.0, 0.0, 0.0], + scale = [40.0, 1.0, 40.0], + tint = [0.40, 0.45, 0.42], + roughness = 0.95, + }); + sd.set_lighting({ timeOfDay = "day", atmosphere = "clear", intensity = 1.0 }); + sd.camera({ eye = [0.0, 10.0, 20.0], target = [0.0, 1.0, 0.0], fov = 55.0 }); +} else { + gfx.setBackgroundColor(0.10, 0.13, 0.19, 1.0); +} + +function eve_update(dt) { + // Idle host: agents drive the scene over MCP (eve_scene_modify / camera / + // screenshot). Nothing to tick here. +} + +function eve_render() { + gfx.clear(); + gfx.render3D(); +} \ No newline at end of file diff --git a/tools/story-scene-agent/.gitignore b/tools/story-scene-agent/.gitignore new file mode 100644 index 00000000..e2ca331a --- /dev/null +++ b/tools/story-scene-agent/.gitignore @@ -0,0 +1,5 @@ +__pycache__/ +*.pyc +test_stage_out/ +ai_stage_out/ +stage_duel/ diff --git a/tools/story-scene-agent/README.md b/tools/story-scene-agent/README.md new file mode 100644 index 00000000..a62d7b24 --- /dev/null +++ b/tools/story-scene-agent/README.md @@ -0,0 +1,68 @@ +# story-scene-agent · 剧情 → 合格舞台 + +把「玩家 / 导演的一句话剧情」变成 **可交付、质量过关的舞台**： +自动摆道具、调光照、架摄像机、多视角截图、质检迭代修复，最终输出报告 + 渲染截图。 + +``` +故事 ──► stage brief ──► creative-brain 生成计划 ──► scene_director 搭台 + └──► scene-qc 多视角质检 + 自动修复（ReAct，硬上限）──► 报告 + 截图 +``` + +## 打通了哪些 AI 能力 + +| 组件 | 角色 | 位置 | +|------|------|------| +| `scene_director.nut` | 引擎脚本侧搭台 kit（摆道具 / 光照 / 摄像机 / 信息） | `src/scripts/` | +| 引擎 MCP 新工具 | `eve_scene_modify` / `eve_scene_info` / `eve_camera_generate` / `eve_scene_reset` | `src/engine/devtools/McpServer.cpp` | +| creative-brain | 意图解析 → 资产匹配 → 布局 → 生成计划 | `tools/creative-brain/` | +| scene-qc-agent | 机位 → 截图 → 本地小模型筛查 → 高危 VLM 复核 → 融合 → 修复 | `tools/scene-qc-agent/` | +| vision-prefilter | 本地轻量视觉预筛选（可选，可作 `local_screen` 后端） | `tools/vision-prefilter/` | +| RenderVision | 引擎内置渲染描述（`eve_render_describe`） | `src/engine/devtools/RenderVision.*` | + +## 快速开始 + +```bash +# 1) 起引擎空舞台 + MCP +eve run --debug --mcp-port=7529 examples/ai-stage + +# 2) 干跑（无引擎 / 无模型，验证闭环） +cd tools/story-scene-agent +python -m story_scene_agent.cli "月光下骑士与巨龙在古堡前对峙" --dry-run + +# 3) 连真引擎搭台 + 质检（在 tools/story-scene-agent 目录下） +python -m story_scene_agent.cli \ + "月光下骑士与巨龙在古堡前对峙" --port 7529 --out-dir stage_duel +``` + +产物：`/story_scene_report.json`（含 stage / plan / build / qc / final）+ 截图。 + +## 参数 + +| 参数 | 说明 | +|------|------| +| `story` | 一句话剧情（必填） | +| `--config` | 配置文件（默认内置） | +| `--scene-id` | 场景标识 | +| `--port` | 引擎 MCP 端口 | +| `--max-rounds` | 质检迭代硬上限 | +| `--dry-run` | 不连引擎/模型，仿真闭环 | +| `--trace` | 打印 ReAct 轨迹 | +| `--out-dir` | 输出目录 | + +> 以模块方式运行（`python -m story_scene_agent.cli`），以便解析相对导入与复用 +> `creative-brain` / `scene-qc-agent` 同仓库代码。若直接 `python cli.py` 需自行处理 +> 包路径。 + +## 测试 + +```bash +python -m pytest tools/story-scene-agent/tests -q +``` + +## 依赖 + +- 运行：Python 3.10+，**标准库即可**（复用 `creative-brain/brain` 与 + `scene-qc-agent/scene_qc_agent` 同仓库代码，无需 pip 安装第三方）。 +- 质检如需远端 VLM：在 `../scene-qc-agent/config.example.json` 配 + `models.vlm_review.base_url/api_key/model`；本地小模型可配 Ollama 或 + `tools/vision-prefilter` 网关。 \ No newline at end of file diff --git a/tools/story-scene-agent/config.example.json b/tools/story-scene-agent/config.example.json new file mode 100644 index 00000000..b79cd68f --- /dev/null +++ b/tools/story-scene-agent/config.example.json @@ -0,0 +1,24 @@ +{ + "engine": { + "host": "127.0.0.1", + "port": 7529 + }, + "creative": { + "catalog": "../creative-brain/catalogs/assets.example.json", + "seed": 2026, + "tile": 2.0 + }, + "stage": { + "default_camera": { "eye": [0.0, 12.0, 22.0], "target": [0.0, 1.0, 0.0], "fov": 55.0 } + }, + "qc": { + "config": "../scene-qc-agent/config.example.json", + "max_rounds": 3, + "targets": ["scene"], + "save_snapshots": true + }, + "output": { + "out_dir": "ai_stage_out", + "report": "story_scene_report.json" + } +} \ No newline at end of file diff --git a/tools/story-scene-agent/requirements.txt b/tools/story-scene-agent/requirements.txt new file mode 100644 index 00000000..e2996f50 --- /dev/null +++ b/tools/story-scene-agent/requirements.txt @@ -0,0 +1,4 @@ +# story-scene-agent 只需 Python 标准库（std / socket / urllib）。 +# 可选用 openai 与 OPENAI_API_KEY 让 creative-brain 意图解析与 story 舞台解析走 LLM。 +# 单元测试用 pytest。 +pytest \ No newline at end of file diff --git a/tools/story-scene-agent/story_scene_agent/__init__.py b/tools/story-scene-agent/story_scene_agent/__init__.py new file mode 100644 index 00000000..721d2436 --- /dev/null +++ b/tools/story-scene-agent/story_scene_agent/__init__.py @@ -0,0 +1,24 @@ +"""story-scene-agent — EVEngine 剧情场景编排大脑。 + +把「玩家一句话剧情」变成「可交付的合格舞台」： +story → stage brief → creative-brain 生成计划 → scene_director 搭台 → +scene-qc 多视角质检 + 自动修复 → 输出报告与渲染截图。 + +复用仓库内既有 AI 工具： + - tools/creative-brain/brain 意图解析 / 资产匹配 / 布局 / 计划 + - tools/scene-qc-agent 巡检 / 融合 / 修复 / ReAct 迭代 + - src/scripts/scene_director.nut 引擎脚本侧搭台 kit + - 引擎 MCP（eve_scene_modify / eve_scene_info / eve_camera_generate / eve_screenshot） +""" + +from .config import DefaultConfig, load_config +from .story import StageBrief, parse_stage +from .orchestrator import StorySceneBrain + +__all__ = [ + "DefaultConfig", + "load_config", + "StageBrief", + "parse_stage", + "StorySceneBrain", +] \ No newline at end of file diff --git a/tools/story-scene-agent/story_scene_agent/cli.py b/tools/story-scene-agent/story_scene_agent/cli.py new file mode 100644 index 00000000..9efe5f3f --- /dev/null +++ b/tools/story-scene-agent/story_scene_agent/cli.py @@ -0,0 +1,10 @@ +"""story-scene-agent CLI entrypoint.""" + +from __future__ import annotations + +import sys + +from .orchestrator import main + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/tools/story-scene-agent/story_scene_agent/config.py b/tools/story-scene-agent/story_scene_agent/config.py new file mode 100644 index 00000000..4be706ad --- /dev/null +++ b/tools/story-scene-agent/story_scene_agent/config.py @@ -0,0 +1,68 @@ +"""Config loading for story-scene-agent (deep-merge over defaults).""" + +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any, Dict + +REPO_ROOT = Path(__file__).resolve().parents[3] +SCENE_DIRECTOR_NUT = REPO_ROOT / "src" / "scripts" / "scene_director.nut" +CREATIVE_BRAIN_DIR = REPO_ROOT / "tools" / "creative-brain" +SCENE_QC_DIR = REPO_ROOT / "tools" / "scene-qc-agent" + + +def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: + out = dict(base) + for k, v in override.items(): + if isinstance(v, dict) and isinstance(out.get(k), dict): + out[k] = _deep_merge(out[k], v) + else: + out[k] = v + return out + + +DefaultConfig: Dict[str, Any] = { + "engine": {"host": "127.0.0.1", "port": 7529, "connect_timeout_s": 5.0}, + "creative": { + "catalog": str(CREATIVE_BRAIN_DIR / "catalogs" / "assets.example.json"), + "seed": 2026, + "tile": 2.0, + }, + "stage": { + "default_camera": {"eye": [0.0, 12.0, 22.0], "target": [0.0, 1.0, 0.0], "fov": 55.0}, + "focal_distance": 2.0, + }, + "qc": { + "config": str(SCENE_QC_DIR / "config.example.json"), + "max_rounds": 3, + "targets": ["scene"], + "save_snapshots": True, + }, + "output": {"out_dir": "ai_stage_out", "report": "story_scene_report.json"}, +} + + +def load_config(path: str = "") -> Dict[str, Any]: + """Return deep-merged config (defaults <- file).""" + cfg = json.loads(json.dumps(DefaultConfig)) + if path: + with open(path, "r", encoding="utf-8") as f: + cfg = _deep_merge(cfg, json.load(f)) + return cfg + + +def load_scene_director_source() -> str: + try: + return SCENE_DIRECTOR_NUT.read_text(encoding="utf-8") + except OSError: + return "" + + +def load_qc_config(path: str = "") -> Dict[str, Any]: + """Load the scene-qc agent config (defaults to its example file).""" + p = Path(path) if path else Path(DefaultConfig["qc"]["config"]) + if not p.is_absolute(): + p = (REPO_ROOT / p).resolve() + with open(p, "r", encoding="utf-8") as f: + return json.load(f) \ No newline at end of file diff --git a/tools/story-scene-agent/story_scene_agent/executor.py b/tools/story-scene-agent/story_scene_agent/executor.py new file mode 100644 index 00000000..08d35771 --- /dev/null +++ b/tools/story-scene-agent/story_scene_agent/executor.py @@ -0,0 +1,251 @@ +"""Scene build executor: plan steps + stage brief -> live scene. + +Two implementations: + - McpSceneExecutor: drives the real engine through MCP + the scene_director kit. + - MemorySceneExecutor: in-memory scene (dry-run / tests / planning preview). + +Both expose the same surface so the orchestrator stays engine-agnostic. +""" + +from __future__ import annotations + +import sys +from typing import Any, Dict, List, Optional + +from .config import load_scene_director_source + + +class SceneExecutor: + """Interface: build / mutate / inspect a staged scene.""" + + def connect(self) -> None: ... + + def close(self) -> None: ... + + def reset(self) -> None: ... + + def install_kit(self) -> bool: ... + + def spawn(self, params: Dict[str, Any]) -> Dict[str, Any]: ... + + def set_lighting(self, params: Dict[str, Any]) -> Dict[str, Any]: ... + + def set_camera(self, params: Dict[str, Any]) -> Dict[str, Any]: ... + + def modify(self, action: str, target: str, params: Dict[str, Any]) -> Dict[str, Any]: ... + + def info(self) -> Dict[str, Any]: ... + + def cameras(self, count: int = 6) -> List[Dict[str, Any]]: ... + + def screenshot(self, path: str) -> Dict[str, Any]: ... + + def run_script(self, source: str) -> str: ... + + +class McpSceneExecutor(SceneExecutor): + """Real engine via the creative-brain TCP MCP client.""" + + def __init__(self, host: str = "127.0.0.1", port: int = 7529): + # Import lazily so the package works without the creative-brain checkout. + from brain.mcp import McpClient # type: ignore + + self.client = McpClient(host, port) + self._kit_installed = False + + def connect(self) -> None: + self.client.connect() + + def close(self) -> None: + try: + self.client.close() + except Exception: + pass + + def install_kit(self) -> bool: + source = load_scene_director_source() + if not source: + return False + try: + self.client.run_script(source) + self._kit_installed = True + return True + except Exception: + return False + + def _ensure_kit(self) -> None: + if not self._kit_installed: + self.install_kit() + + def _modify(self, action: str, target: str, params: Dict[str, Any]) -> Dict[str, Any]: + self._ensure_kit() + raw = self.client.call_tool("eve_scene_modify", + {"action": action, "target": target, "params": params}) + return _parse_dict(raw) + + def reset(self) -> None: + self._ensure_kit() + try: + self.client.call_tool("eve_scene_reset") + except Exception: + pass + + def spawn(self, params: Dict[str, Any]) -> Dict[str, Any]: + return self._modify("add_object", str(params.get("id", "prop")), params) + + def set_lighting(self, params: Dict[str, Any]) -> Dict[str, Any]: + return self._modify("lighting", "scene", params) + + def set_camera(self, params: Dict[str, Any]) -> Dict[str, Any]: + return self._modify("camera", "scene", params) + + def modify(self, action: str, target: str, params: Dict[str, Any]) -> Dict[str, Any]: + return self._modify(action, target, params) + + def info(self) -> Dict[str, Any]: + self._ensure_kit() + return _parse_dict(self.client.call_tool("eve_scene_info")) + + def cameras(self, count: int = 6) -> List[Dict[str, Any]]: + self._ensure_kit() + raw = self.client.call_tool("eve_camera_generate", {"count": count}) + parsed = _parse_dict(raw) + if isinstance(parsed, list): + return parsed + return [] + + def screenshot(self, path: str) -> Dict[str, Any]: + raw = self.client.call_tool("eve_screenshot", {"path": path}) + return _parse_dict(raw) + + def run_script(self, source: str) -> str: + return self.client.run_script(source) + + +class MemorySceneExecutor(SceneExecutor): + """Deterministic in-memory scene: no engine / no MCP (tests & dry-run).""" + + def __init__(self): + self.props: Dict[str, Dict[str, Any]] = {} + self.lighting: Dict[str, Any] = {} + self.camera: Dict[str, Any] = {} + self._counter = 0 + + def connect(self) -> None: + pass + + def close(self) -> None: + pass + + def reset(self) -> None: + self.props.clear() + self.lighting = {"timeOfDay": "day", "intensity": 1.0} + self.camera = {} + + def install_kit(self) -> bool: + return bool(load_scene_director_source()) + + def spawn(self, params: Dict[str, Any]) -> Dict[str, Any]: + pid = str(params.get("id", f"prop_{self._counter}")) + self._counter += 1 + self.props[pid] = { + "id": pid, + "kind": params.get("kind", "box"), + "pos": list(params.get("pos", [0.0, 0.5, 0.0])), + "scale": list(params.get("scale", [1.0, 1.0, 1.0])), + "yaw_deg": params.get("yaw_deg", 0.0), + "tint": list(params.get("tint", [1.0, 1.0, 1.0])), + } + return {"ok": True, "id": pid} + + def set_lighting(self, params: Dict[str, Any]) -> Dict[str, Any]: + self.lighting.update(params) + return {"ok": True} + + def set_camera(self, params: Dict[str, Any]) -> Dict[str, Any]: + self.camera.update(params) + return {"ok": True} + + def modify(self, action: str, target: str, params: Dict[str, Any]) -> Dict[str, Any]: + p = dict(params or {}) + if action == "add_object": + return self.spawn(p) + if action == "move_object": + if target in self.props: + pos = p.get("pos") or [p.get("x", 0), p.get("y", 0), p.get("z", 0)] + self.props[target]["pos"] = list(pos) + return {"ok": True, "id": target} + if action == "remove_object": + if target in self.props: + del self.props[target] + return {"ok": True, "id": target} + if action == "scale": + if target in self.props and "scale" in p: + self.props[target]["scale"] = list(p["scale"]) + return {"ok": True, "id": target} + if action == "rotate": + if target in self.props and "yaw_deg" in p: + self.props[target]["yaw_deg"] = p["yaw_deg"] + return {"ok": True, "id": target} + if action == "lighting": + return self.set_lighting(p) + if action == "camera": + return self.set_camera(p) + if action == "cameras": + return {"cameras": self.cameras(int(p.get("count", 6)))} + if action == "info": + return self.info() + if action == "list": + return {"ids": list(self.props.keys())} + return {"ok": False, "error": f"unknown action: {action}"} + + def info(self) -> Dict[str, Any]: + return { + "count": len(self.props), + "props": [ + {"id": v["id"], "kind": v["kind"], "pos": v["pos"], + "scale": v["scale"], "yaw_deg": v["yaw_deg"], "tint": v["tint"]} + for v in self.props.values() + ], + } + + def cameras(self, count: int = 6) -> List[Dict[str, Any]]: + cams = [] + for i in range(max(1, count)): + ang = (i * 2.0 * 3.14159265) / max(1, count) + cams.append({ + "id": f"cam_{i}", + "eye": [12.0 * _cos(ang), 8.0, 12.0 * _sin(ang)], + "target": [0.0, 1.0, 0.0], + "fov": 55.0, + }) + return cams + + def screenshot(self, path: str) -> Dict[str, Any]: + return {"path": path, "width": 800, "height": 600} + + def run_script(self, source: str) -> str: + return "dry-run script ok" + + +def _cos(a: float) -> float: + import math + + return math.cos(a) + + +def _sin(a: float) -> float: + import math + + return math.sin(a) + + +def _parse_dict(raw: str) -> Any: + import json + + if not raw: + return {} + try: + return json.loads(raw) + except json.JSONDecodeError: + return {"raw": raw} diff --git a/tools/story-scene-agent/story_scene_agent/orchestrator.py b/tools/story-scene-agent/story_scene_agent/orchestrator.py new file mode 100644 index 00000000..e1287f39 --- /dev/null +++ b/tools/story-scene-agent/story_scene_agent/orchestrator.py @@ -0,0 +1,302 @@ +"""Orchestrator: story -> stage -> plan -> build -> QC loop -> finalize. + +Reuses the existing AI tools as building blocks: + - creative-brain (brain.plan.build_plan) for intent / assets / layout / plan + - scene-qc-agent (SceneQCBrain + QcPipeline + Fixer) for quality iteration + - the scene_director kit for placing / lighting / camera +""" + +from __future__ import annotations + +import json +import os +import sys +from pathlib import Path +from typing import Any, Dict, List, Optional + +from .config import ( + CREATIVE_BRAIN_DIR, + SCENE_QC_DIR, + DefaultConfig, + load_config, + load_qc_config, +) +from .executor import McpSceneExecutor, MemorySceneExecutor, SceneExecutor +from .story import StageBrief, parse_stage + +# Make sibling tool packages importable (creative-brain / scene-qc-agent). +for _pkg in (str(CREATIVE_BRAIN_DIR), str(SCENE_QC_DIR)): + if _pkg not in sys.path: + sys.path.insert(0, _pkg) + + +class StorySceneBrain: + """End-to-end story -> qualified scene director.""" + + def __init__(self, cfg: Dict[str, Any], dry_run: bool = False, trace: bool = False): + import json as _json + + base = _json.loads(_json.dumps(DefaultConfig)) + self.cfg = _deep_merge(base, cfg) + self.dry_run = dry_run + self.trace = trace + + # ------------------------------------------------------------------ build + def build_plan(self, story: str, stage: StageBrief) -> Any: + """Creative-brain plan + stage-director overrides.""" + from brain.plan import build_plan # type: ignore + + catalog = self.cfg["creative"].get("catalog") or DefaultConfig["creative"]["catalog"] + seed = int(self.cfg["creative"].get("seed", 2026)) + plan = build_plan(story, catalog_path=catalog, seed=seed) + # Stage-director intent wins over the generic lighting guess. + plan.config.lighting.timeOfDay = stage.timeOfDay + plan.config.lighting.atmosphere = stage.atmosphere + plan.config.name = stage.name or plan.config.name + return plan + + def build_scene(self, executor: SceneExecutor, plan: Any, stage: StageBrief) -> List[dict]: + """Execute the plan + stage brief against the live/in-memory scene.""" + log: List[dict] = [] + tile = float(self.cfg["creative"].get("tile", 2.0)) + + executor.reset() + log.append({"step": "reset", "ok": True}) + + # Terrain ground (from plan.terrain, sized by config width/height). + terr = plan.config.terrain + w = int(terr.width or 32) + h = int(terr.height or 32) + r = executor.spawn({ + "id": "terrain", + "kind": "ground", + "pos": [0.0, 0.0, 0.0], + "scale": [max(8.0, w * tile), 1.0, max(8.0, h * tile)], + "tint": [0.42, 0.46, 0.40], + "roughness": 0.95, + }) + log.append({"step": "terrain", "ok": bool(r.get("ok"))}) + + # Lighting (stage overrides from story atmosphere / time-of-day). + lit = executor.set_lighting({ + "timeOfDay": plan.config.lighting.timeOfDay, + "atmosphere": plan.config.lighting.atmosphere, + "intensity": float(plan.config.lighting.intensity), + }) + log.append({"step": "lighting", "ok": bool(lit.get("ok"))}) + + # Place props from the layout steps (grid cells -> world coords). + placed = 0 + for step in plan.steps: + if step.action != "place": + continue + x = int(step.params.get("x", 0)) + y = int(step.params.get("y", 0)) + pid = f"{step.target}_{x}_{y}" + res = executor.spawn({ + "id": pid, + "kind": str(step.target), + "pos": _cell_to_world(x, y, w, h, tile), + "seed": int(step.params.get("seed", 1)), + }) + if res.get("ok"): + placed += 1 + log.append({"step": "place", "id": pid, "ok": bool(res.get("ok"))}) + + # Focal centerpiece from the story (deliberate, at stage origin). + centerpiece = stage.centerpiece + if centerpiece: + res = executor.spawn({ + "id": "centerpiece", + "kind": centerpiece, + "pos": [0.0, 0.0, 0.0], + "scale": [1.6, 1.6, 1.6], + "yaw_deg": 0.0, + "seed": 7, + }) + log.append({"step": "centerpiece", "kind": centerpiece, "ok": bool(res.get("ok"))}) + + # Default framing camera. + cam = self.cfg["stage"].get("default_camera", DefaultConfig["stage"]["default_camera"]) + camres = executor.set_camera(dict(cam)) + log.append({"step": "camera", "ok": bool(camres.get("ok"))}) + + info = executor.info() + log.append({"step": "info", "count": int(info.get("count", 0))}) + return log + + # ------------------------------------------------------------------- QC + def _qc_runtime(self, client, screenshot_dir: str = ""): + from scene_qc_agent.config import load_config as qc_load_config # type: ignore + from scene_qc_agent.fixer import Fixer # type: ignore + from scene_qc_agent.models import OpenAIClient # type: ignore + from scene_qc_agent.pipeline import QcPipeline # type: ignore + from scene_qc_agent.react import SceneQCBrain # type: ignore + from scene_qc_agent.tools import ToolRegistry # type: ignore + + qc_path = self.cfg["qc"].get("config") + overrides = { + "dry_run": self.dry_run, + "iteration": {"max_rounds": int(self.cfg["qc"].get("max_rounds", 3))}, + "output": {"trace": bool(self.trace or self.cfg.get("trace", False)), + "save_snapshots": bool(self.cfg["qc"].get("save_snapshots", False))}, + } + if screenshot_dir: + overrides["output"]["screenshot_dir"] = os.path.abspath(screenshot_dir) + qc_cfg = qc_load_config(qc_path, overrides) + + dry = self.dry_run + registry = ToolRegistry(client if not dry else None, qc_cfg, dry_run=dry) + + def _model(tier: str): + spec = qc_cfg["models"].get(tier) or {} + if dry or not spec.get("enabled") or not spec.get("model"): + return None + try: + return OpenAIClient(spec) + except Exception: + return None + + local = _model("local_screen") + vlm = _model("vlm_review") + planner = _model("fix_planner") + brain = _model("brain") if qc_cfg["models"].get("brain") else None + + pipeline = QcPipeline(registry, local, vlm, qc_cfg) + fixer = Fixer(registry, planner, qc_cfg) + react = SceneQCBrain(pipeline, fixer, brain, qc_cfg, trace=bool(self.trace)) + return pipeline, fixer, react + + def run_qc(self, client, scene_id: str, targets: List[str], screenshot_dir: str = ""): + _, _, react = self._qc_runtime(client, screenshot_dir=screenshot_dir) + report, history, rounds = react.run(scene_id, targets) + return report, history, rounds + + # ------------------------------------------------------------- finalize + def finalize(self, executor: SceneExecutor, stage: StageBrief, + qc_report, out_dir: str) -> Dict[str, Any]: + os.makedirs(out_dir, exist_ok=True) + shots: List[dict] = [] + if not self.dry_run: + try: + cams = executor.cameras(2) + for i, cam in enumerate(cams[:2]): + path = os.path.join(out_dir, f"shot_{i}.png") + res = executor.screenshot(path) + shots.append({"camera": cam.get("id", f"cam_{i}"), "path": res.get("path", path)}) + except Exception as e: # screenshot is best-effort + shots.append({"error": str(e)}) + return {"out_dir": out_dir, "shots": shots, + "info": executor.info()} + + # ---------------------------------------------------------------- run + def run(self, story: str, scene_id: str = "", out_dir: str = "") -> Dict[str, Any]: + stage = parse_stage(story, name=scene_id or "story-scene") + plan = self.build_plan(story, stage) + + executor: SceneExecutor + if self.dry_run: + executor = MemorySceneExecutor() + else: + executor = McpSceneExecutor(self.cfg["engine"]["host"], int(self.cfg["engine"]["port"])) + + try: + if not self.dry_run: + executor.connect() + build_log = self.build_scene(executor, plan, stage) + + out_dir = out_dir or self.cfg["output"].get("out_dir", "ai_stage_out") + targets = list(self.cfg["qc"].get("targets", ["scene"])) + qc_client = executor.client if not self.dry_run else None + qc_report, qc_history, rounds = self.run_qc( + qc_client, stage.name or scene_id, targets, screenshot_dir=out_dir) + final = self.finalize(executor, stage, qc_report, out_dir) + + result = { + "story": story, + "stage": stage.to_dict(), + "plan": plan.to_dict(), + "build": build_log, + "qc": { + "passed": bool(qc_report.passed), + "score": float(qc_report.score), + "rounds_used": int(rounds), + "summary": qc_report.summary, + "report": qc_report.to_dict(), + "history": qc_history, + }, + "final": final, + "dry_run": self.dry_run, + } + self._save_report(result, out_dir) + return result + finally: + executor.close() + + def _save_report(self, result: Dict[str, Any], out_dir: str) -> str: + os.makedirs(out_dir, exist_ok=True) + path = os.path.join(out_dir, self.cfg["output"].get("report", "story_scene_report.json")) + with open(path, "w", encoding="utf-8") as f: + json.dump(result, f, ensure_ascii=False, indent=2) + return path + + +def _cell_to_world(x: int, y: int, width: int, height: int, tile: float) -> List[float]: + return [round((x - width / 2.0) * tile, 3), 0.0, round((y - height / 2.0) * tile, 3)] + + +def _deep_merge(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]: + out = dict(base) + for k, v in override.items(): + if isinstance(v, dict) and isinstance(out.get(k), dict): + out[k] = _deep_merge(out[k], v) + else: + out[k] = v + return out + + +def main(argv: Optional[List[str]] = None) -> int: + import argparse + + ap = argparse.ArgumentParser( + prog="story-scene-agent", + description="剧情 -> 合格舞台：故事解析 -> 生成计划 -> 搭台 -> 质检迭代 -> 交付") + ap.add_argument("story", help="玩家/导演的一句话剧情，如“月光下骑士与巨龙在古堡前对峙”") + ap.add_argument("--config", default="", help="配置文件（默认内置）") + ap.add_argument("--scene-id", default="story-scene", help="场景标识") + ap.add_argument("--port", type=int, default=None, help="引擎 MCP 端口（覆盖配置）") + ap.add_argument("--max-rounds", type=int, default=None, help="质检迭代硬上限（覆盖配置）") + ap.add_argument("--dry-run", action="store_true", help="不连引擎/模型，使用仿真闭环") + ap.add_argument("--trace", action="store_true", help="打印 ReAct 轨迹") + ap.add_argument("--out-dir", default="", help="输出目录（报告 + 截图）") + args = ap.parse_args(argv) + + cfg = load_config(args.config) + if args.port: + cfg["engine"]["port"] = args.port + if args.max_rounds: + cfg["qc"]["max_rounds"] = args.max_rounds + if args.trace: + cfg["trace"] = True + + brain = StorySceneBrain(cfg, dry_run=args.dry_run, trace=args.trace) + result = brain.run(args.story, scene_id=args.scene_id, out_dir=args.out_dir) + + summary = { + "passed": result["qc"]["passed"], + "score": result["qc"]["score"], + "rounds_used": result["qc"]["rounds_used"], + "props": result["final"]["info"].get("count", 0), + "dry_run": result["dry_run"], + "report": os.path.join(args.out_dir or cfg["output"]["out_dir"], + cfg["output"]["report"]), + } + print(json.dumps(summary, ensure_ascii=False)) + if result["qc"]["history"]: + print(json.dumps([{ "round": h["round"], "phase": h["phase"], "passed": h["passed"] } + for h in result["qc"]["history"]], ensure_ascii=False)) + return 0 if result["qc"]["passed"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/tools/story-scene-agent/story_scene_agent/story.py b/tools/story-scene-agent/story_scene_agent/story.py new file mode 100644 index 00000000..23050a45 --- /dev/null +++ b/tools/story-scene-agent/story_scene_agent/story.py @@ -0,0 +1,232 @@ +"""Story → stage brief: what the scene must contain and feel like. + +The Creative Brain already parses style / terrain / lighting / vegetation from +free text. This module adds the *stage-directing* layer the story ask really +needs: focal elements (cast / centerpiece), camera intent, atmosphere keywords, +and a hint count of story beats — so the orchestrator can place a deliberate +centerpiece and frame it, instead of scattering props uniformly. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field, asdict +from typing import Any, Dict, List, Optional + +# Thematic keyword -> procedural prop kind the kit can spawn. +THEME_KINDS: Dict[str, str] = { + "castle": "skyscraper", + "fortress": "skyscraper", + "keep": "skyscraper", + "tower": "skyscraper", + "palace": "skyscraper", + "城堡": "skyscraper", + "王宫": "skyscraper", + "要塞": "skyscraper", + "高塔": "skyscraper", + "塔": "skyscraper", + "dragon": "rock", + "monster": "rock", + "beast": "rock", + "巨龙": "rock", + "龙": "rock", + "怪物": "rock", + "魔兽": "rock", + "knight": "pillar", + "hero": "pillar", + "guard": "pillar", + "statue": "pillar", + "骑士": "pillar", + "勇士": "pillar", + "雕像": "pillar", + "立柱": "pillar", + "mountain": "rock", + "hill": "rock", + "boulder": "rock", + "stone": "rock", + "岩石": "rock", + "巨石": "rock", + "山石": "rock", + "tree": "tree", + "forest": "tree", + "wood": "tree", + "森林": "tree", + "树": "tree", + "bush": "bush", + "灌木": "bush", + "wall": "stonewall", + "城墙": "stonewall", + "石墙": "stonewall", + "bridge": "bridge", + "桥": "bridge", + "fence": "fence", + "篱笆": "fence", + "栅栏": "fence", + "hedge": "hedge", + "树篱": "hedge", + "chevaldefrise": "chevaldefrise", + "treasure": "box", + "chest": "box", + "coin": "sphere", + "jewel": "sphere", + "crystal": "sphere", + "temple": "box", + "house": "box", + "hut": "box", + "宝箱": "box", + "宝藏": "box", + "宝石": "sphere", + "水晶": "sphere", + "神庙": "box", + "殿堂": "box", +} + +# Atmosphere / time-of-day keywords (checked before generic style). +TIME_OF_DAY: List[tuple] = [ + ("night", ["night", "moonlit", "moonlight", "midnight", "dark", "noir", "月光", "月下", "夜晚", "深夜", "夜色", "漆黑"]), + ("dusk", ["dusk", "sunset", "twilight", "evening", "黄昏", "傍晚", "落日", "暮色"]), + ("dawn", ["dawn", "sunrise", "morning", "拂晓", "黎明", "清晨"]), + ("day", ["day", "noon", "midday", "sunny", "bright", "afternoon", "白天", "正午", "晴朗"]), +] + +# Generic mood keywords -> atmosphere label (passed through to the plan). +MOOD_LABELS: List[tuple] = [ + ("moonlit", ["moonlit", "moonlight", "月光", "月下"]), + ("dusk", ["dusk", "黄昏", "傍晚"]), + ("rainy", ["rain", "rainy", "storm", "雨"]), + ("foggy", ["fog", "mist", "foggy", "雾"]), + ("sunny", ["sunny", "clear", "晴"]), + ("eerie", ["eerie", "haunted", "ghost", "幽", "诡异"]), + ("epic", ["epic", "battle", "war", "conflict", "对决", "战场"]), + ("peaceful", ["peaceful", "quiet", "calm", "宁静"]), + ("cyberpunk", ["cyberpunk", "neon", "synth", "赛博"]), + ("fantasy", ["fantasy", "magic", "enchant", "魔法", "奇幻"]), +] + + +def _contains(text: str, keywords: List[str]) -> bool: + low = text.lower() + return any(k in low for k in keywords) + + +def _extract_kinds(text: str) -> List[str]: + """Collect prop kinds mentioned in the story, most specific first.""" + seen: List[str] = [] + low = text.lower() + for keyword, kind in THEME_KINDS.items(): + if keyword in low and kind not in seen: + seen.append(kind) + return seen + + +def _time_of_day(text: str) -> str: + for tod, keys in TIME_OF_DAY: + if _contains(text, keys): + return tod + return "day" + + +def _atmosphere(text: str) -> str: + for label, keys in MOOD_LABELS: + if _contains(text, keys): + return label + return "generic" + + +def _sentences(text: str) -> List[str]: + parts = re.split(r"[。！？!?；;.\n]+", text) + return [p.strip() for p in parts if p.strip()] + + +@dataclass +class StageBrief: + """Stage-directing summary extracted from a story request.""" + + name: str = "story-scene" + atmosphere: str = "generic" + timeOfDay: str = "day" + beats: int = 1 + focal: List[Dict[str, Any]] = field(default_factory=list) # [{kind, label}] + cast: List[str] = field(default_factory=list) # focal entity labels + centerpiece: Optional[str] = None # main prop kind + notes: List[str] = field(default_factory=list) + + def to_dict(self) -> dict: + return asdict(self) + + +def parse_stage(story: str, name: str = "story-scene", llm: Any = None) -> StageBrief: + """Parse a story request into a StageBrief (heuristic; LLM optional).""" + story = (story or "").strip() + brief = StageBrief(name=name) + + if not story: + return brief + + brief.atmosphere = _atmosphere(story) + brief.timeOfDay = _time_of_day(story) + brief.beats = max(1, len(_sentences(story))) + + kinds = _extract_kinds(story) + focal: List[Dict[str, Any]] = [] + for kind in kinds: + focal.append({"kind": kind, "label": f"{kind}-focal"}) + if focal: + brief.focal = focal + brief.centerpiece = focal[0]["kind"] + brief.cast = [f["label"] for f in focal] + + # Keep a lightweight narrative note for the build phase. + brief.notes.append(story[:200]) + + # Optional LLM refinement (OpenAI-compatible) — best effort, never fatal. + if llm is not None: + try: + raw = llm.complete( + [{ + "role": "user", + "content": ( + "你是舞台导演。把这段剧情抽成舞台要点：" + "atmosphere(英文), timeOfDay(dawn|day|dusk|night), " + "centerpiece(道具种类：skyscraper|rock|pillar|tree|bush|stonewall|bridge" + "|box|sphere|cylinder), cast(角色/焦点实体名数组)。" + f"只输出JSON对象。剧情：{story}" + ), + }] + ) + parsed = _safe_json(raw) + if isinstance(parsed, dict): + if isinstance(parsed.get("atmosphere"), str): + brief.atmosphere = parsed["atmosphere"] + if parsed.get("timeOfDay") in ("dawn", "day", "dusk", "night"): + brief.timeOfDay = parsed["timeOfDay"] + if isinstance(parsed.get("centerpiece"), str): + brief.centerpiece = parsed["centerpiece"] + if isinstance(parsed.get("cast"), list): + brief.cast = [str(c) for c in parsed["cast"] if str(c)] + except Exception: + pass + return brief + + +def _safe_json(text: str) -> Optional[Any]: + import json + + if not text: + return None + t = text.strip() + if t.startswith("```"): + t = t.split("```", 2)[1] if "```" in t[3:] else t + t = t.strip().lstrip("json").strip() + try: + return json.loads(t) + except json.JSONDecodeError: + import re + + m = re.search(r"\{.*\}", t, re.DOTALL) + if m: + try: + return json.loads(m.group(0)) + except json.JSONDecodeError: + return None + return None \ No newline at end of file diff --git a/tools/story-scene-agent/tests/test_story_scene_agent.py b/tools/story-scene-agent/tests/test_story_scene_agent.py new file mode 100644 index 00000000..eb39a796 --- /dev/null +++ b/tools/story-scene-agent/tests/test_story_scene_agent.py @@ -0,0 +1,95 @@ +"""story-scene-agent unit tests (dry-run: no engine, no network, no models).""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +PKG_DIR = Path(__file__).resolve().parents[1] +if str(PKG_DIR) not in sys.path: + sys.path.insert(0, str(PKG_DIR)) + +REPO = Path(__file__).resolve().parents[3] +CREATIVE = REPO / "tools" / "creative-brain" +if str(CREATIVE) not in sys.path: + sys.path.insert(0, str(CREATIVE)) + +from story_scene_agent.executor import MemorySceneExecutor # noqa: E402 +from story_scene_agent.orchestrator import StorySceneBrain # noqa: E402 +from story_scene_agent.story import parse_stage # noqa: E402 + + +def test_parse_stage_night_forest(): + brief = parse_stage("月光下的幽暗森林，树影摇曳，冒险者在木桥边遇到巨龙的宝藏") + assert brief.timeOfDay == "night" + assert brief.beats >= 1 + assert "tree" in {f["kind"] for f in brief.focal} + assert brief.centerpiece is not None + + +def test_parse_stage_empty(): + brief = parse_stage("") + assert brief.beats == 1 + assert brief.focal == [] + assert brief.centerpiece is None + + +def test_plan_build_no_llm(): + brain = StorySceneBrain({}, dry_run=True) + stage = parse_stage("snowy tundra with scattered rocks at dusk") + plan = brain.build_plan("snowy tundra with scattered rocks at dusk", stage) + actions = [s.action for s in plan.steps] + assert "terrain" in actions + assert "set_lighting" in actions + assert "place" in actions + assert plan.config.lighting.timeOfDay == stage.timeOfDay + + +def test_memory_executor_roundtrip(): + ex = MemorySceneExecutor() + ex.reset() + assert ex.spawn({"id": "hero", "kind": "pillar", "pos": [1.0, 0.0, 2.0]})["ok"] + assert ex.spawn({"id": "dragon", "kind": "rock"})["ok"] + info = ex.info() + assert info["count"] == 2 + assert ex.modify("move_object", "hero", {"pos": [5.0, 1.0, 5.0]})["ok"] + assert ex.props["hero"]["pos"] == [5.0, 1.0, 5.0] + assert ex.modify("remove_object", "dragon", {})["ok"] + assert ex.info()["count"] == 1 + assert len(ex.cameras(3)) == 3 + assert ex.cameras(3)[0]["id"] == "cam_0" + + +def test_full_dryrun_loop(tmp_path): + out_dir = str(tmp_path) + cfg = { + "creative": {"catalog": str(CREATIVE / "catalogs" / "assets.example.json"), "seed": 7}, + "stage": {"default_camera": {"eye": [0, 12, 22], "target": [0, 1, 0], "fov": 55}}, + "qc": {"max_rounds": 3, "targets": ["scene"]}, + "output": {"out_dir": out_dir}, + } + brain = StorySceneBrain(cfg, dry_run=True, trace=False) + result = brain.run("moonlit duel between a knight and a dragon before an old castle", + scene_id="duel", out_dir=out_dir) + assert result["stage"]["beats"] >= 1 + assert result["plan"]["steps"] + assert result["build"], "build log must not be empty" + assert result["final"]["info"]["count"] > 0 + assert "passed" in result["qc"] + assert isinstance(result["qc"]["rounds_used"], int) + report = Path(out_dir) / brain.cfg["output"]["report"] + assert report.exists() + data = json.loads(report.read_text(encoding="utf-8")) + assert data["story"] + assert data["stage"]["atmosphere"] + + +def test_creative_brain_snippet_targets_kit(): + from brain.mcp import snippet_for_step # type: ignore + + src = snippet_for_step("place", "tree", {"x": 4, "y": 5, "seed": 9}) + assert "scene_director.modify" in src + assert "add_object" in src + src2 = snippet_for_step("set_lighting", "scene", {"timeOfDay": "night"}) + assert "lighting" in src2 \ No newline at end of file