feat(story-scene-agent): story-to-qualified-stage orchestrator + example
One-line story -> stage brief (parse_stage) -> creative-brain plan -> build via scene_director -> scene-qc inspection/fix loop -> report + screenshots. Reuses the in-repo creative-brain and scene-qc packages; ships a dry-run in-memory executor and tests, an empty examples/ai-stage host, an openCode subagent, and the AI场景导演 integration doc that doubles as the agent-queryable scene-director spec.
This commit is contained in:
西风逍遥游 committed
1 parent
2f2d019140
commit
4e61805c69
17 files changed
+1344
No files matched your search
@@ -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}`。
|
||||
- `<out-dir>/story_scene_report.json`：剧情、舞台要点、生成计划、搭台日志、质检报告与历史。
|
||||
@@ -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<br/>parse_stage 舞台要点"]
|
||||
Stage --> CB["creative-brain<br/>intent→assets→layout→plan"]
|
||||
CB --> SD["scene_director kit<br/>(eve_scene_modify 摆物/光照/摄像机)"]
|
||||
SD --> Shot["eve_screenshot / capture_render_frame<br/>多视角截图"]
|
||||
Shot --> QC["scene-qc-agent<br/>本地小模型筛查→高危VLM复核→融合"]
|
||||
QC -->|不达标| Fix["Fixer 生成修改计划<br/>eve_scene_modify 落地"]
|
||||
Fix --> SD
|
||||
QC -->|达标| Out["报告 + 渲染截图"]
|
||||
VP["vision-prefilter<br/>(可选本地预筛)"] -.-> QC
|
||||
RV["RenderVision<br/>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/<platform>-debug/test/unit_test --testcase='^devtools\.mcp\..*$'
|
||||
```
|
||||
|
||||
## 后续可扩展
|
||||
|
||||
- scene_director 支持外部模型（`.glb`）与动画挂接。
|
||||
- `eve_scene_modify` 增加 `spawn_effect` / 粒子 / 音频编排。
|
||||
- 场景持久化为可回放的编排脚本（story → nut 剧本）。
|
||||
- 与 `eve test` 场景脚本联动的回归断言。
|
||||
@@ -10,6 +10,7 @@
|
||||
- [命令行设计](命令行设计.md)
|
||||
- [测试覆盖](测试覆盖.md)
|
||||
- [AI 与 MCP 支持](AI与MCP支持.md)
|
||||
- [AI 场景导演](AI场景导演.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)。
|
||||
@@ -0,0 +1,7 @@
|
||||
config <- {
|
||||
width = 1024
|
||||
height = 768
|
||||
title = "EVEngine AI Stage"
|
||||
debug = true
|
||||
hotReload = false
|
||||
};
|
||||
@@ -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();
|
||||
}
|
||||
@@ -0,0 +1,5 @@
|
||||
__pycache__/
|
||||
*.pyc
|
||||
test_stage_out/
|
||||
ai_stage_out/
|
||||
stage_duel/
|
||||
@@ -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
|
||||
```
|
||||
|
||||
产物：`<out-dir>/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` 网关。
|
||||
@@ -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"
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,4 @@
|
||||
# story-scene-agent 只需 Python 标准库（std / socket / urllib）。
|
||||
# 可选用 openai 与 OPENAI_API_KEY 让 creative-brain 意图解析与 story 舞台解析走 LLM。
|
||||
# 单元测试用 pytest。
|
||||
pytest
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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())
|
||||
@@ -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)
|
||||
@@ -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}
|
||||
@@ -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())
|
||||
@@ -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
|
||||
@@ -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
|
||||
Reference in new issue
Block a user