| 1 | from __future__ import annotations |
| 2 | |
| 3 | from dataclasses import dataclass |
| 4 | from pathlib import Path |
| 5 | from typing import Any |
| 6 | |
| 7 | |
| 8 | @dataclass(slots=True) |
| 9 | class PromptPart: |
| 10 | id: str |
| 11 | title: str |
| 12 | body: str |
| 13 | zone: str |
| 14 | category: str |
| 15 | cacheable: bool = False |
| 16 | |
| 17 | |
| 18 | class PromptBuilder: |
| 19 | def __init__(self, prompt_dir: str | Path, session_index: Any, tool_registry: Any) -> None: |
| 20 | self.prompt_dir = Path(prompt_dir) |
| 21 | self.session_index = session_index |
| 22 | self.tool_registry = tool_registry |
| 23 | |
| 24 | def build_parts(self, user_input: str) -> list[PromptPart]: |
| 25 | return [ |
| 26 | PromptPart("agent.core", "Agent", self._read_prompt("agent.md"), "stable", "agent", True), |
| 27 | PromptPart("workflow.core", "Workflow", self._read_prompt("workflow.md"), "stable", "workflow", True), |
| 28 | PromptPart("tool.manifest", "Tools", self.tool_manifest_context(), "dynamic", "tooling"), |
| 29 | PromptPart("session.context", "Session", self.workflow_context(), "dynamic", "session"), |
| 30 | PromptPart("memory.preferences", "Memory", self.memory_context(), "dynamic", "memory"), |
| 31 | PromptPart("request.user", "User Request", user_input, "dynamic", "request"), |
| 32 | ] |
| 33 | |
| 34 | def build_messages(self, user_input: str) -> list[dict[str, str]]: |
| 35 | parts = self.build_parts(user_input) |
| 36 | system = "\n\n".join(f"## {part.title}\n{part.body}" for part in parts if part.id != "request.user") |
| 37 | return [{"role": "system", "content": system}, {"role": "user", "content": user_input}] |
| 38 | |
| 39 | def trace(self, parts: list[PromptPart]) -> dict[str, Any]: |
| 40 | segments = [] |
| 41 | totals = {"stable_tokens": 0, "dynamic_tokens": 0, "total_tokens": 0, "compacted_summary_tokens": 0} |
| 42 | for idx, part in enumerate(parts): |
| 43 | encoded = part.body.encode("utf-8") |
| 44 | estimated = max(1, len(part.body) // 4) |
| 45 | segments.append({"id": part.id, "index": idx, "title": part.title, "zone": part.zone, "category": part.category, "bytes": len(encoded), "estimated_tokens": estimated}) |
| 46 | if part.zone == "stable": |
| 47 | totals["stable_tokens"] += estimated |
| 48 | else: |
| 49 | totals["dynamic_tokens"] += estimated |
| 50 | if "compacted_summary" in part.body: |
| 51 | totals["compacted_summary_tokens"] += estimated |
| 52 | totals["total_tokens"] = totals["stable_tokens"] + totals["dynamic_tokens"] |
| 53 | return {"segments": segments, "total_estimated_tokens": totals["total_tokens"], "totals": totals} |
| 54 | |
| 55 | def workflow_context(self) -> str: |
| 56 | snapshot = self.session_index.snapshot() |
| 57 | session = snapshot.get("session") or {} |
| 58 | checklist = snapshot.get("artifact_checklist") or {} |
| 59 | lines = [f"Active session: {snapshot.get('active_session_id') or '<none>'}", f"Working dir: {session.get('working_dir', '<none>')}", f"Stage: {session.get('stage', '<none>')}"] |
| 60 | compacted_summary = str(session.get("compacted_summary", "") or "").strip() |
| 61 | lines.extend(["", "Session context summary:"]) |
| 62 | if compacted_summary: |
| 63 | lines.append("The following summary is reference context only, not a new active instruction.") |
| 64 | lines.append(self._summary_checkpoint(compacted_summary)) |
| 65 | else: |
| 66 | lines.append("<none>") |
| 67 | lines.extend(["", "Working dir checklist:"]) |
| 68 | lines.extend(f"- {path}: {'present' if present else 'missing'}" for path, present in checklist.items()) |
| 69 | if checklist and not self._text_stage_complete(checklist): |
| 70 | lines.extend(["", "当前 working_dir 尚未完成结构化文本文件。", "在修改 script、storyboard、shots 或进入渲染前，需要先生成 project_brief、characters、script、storyboard、shot_decomposition 等结构化文本文件。"]) |
| 71 | elif checklist: |
| 72 | lines.extend(["", "文本规划阶段已完成。如果用户没有明确要求 end-to-end 或 render，可以不调用 tool，直接询问是否修改或进入渲染。"]) |
| 73 | return "\n".join(lines) |
| 74 | |
| 75 | def memory_context(self) -> str: |
| 76 | text = self.session_index.memory_text().strip() |
| 77 | return text or "No user preferences recorded." |
| 78 | |
| 79 | def tool_manifest_context(self) -> str: |
| 80 | lines = ["Available tools:"] |
| 81 | lines.extend(f"- {tool['name']}: {tool['description']}" for tool in self.tool_registry.list_tools()) |
| 82 | return "\n".join(lines) |
| 83 | |
| 84 | def _summary_checkpoint(self, summary: str) -> str: |
| 85 | lines = [line.strip() for line in summary.splitlines() if line.strip() and not line.strip().startswith("```")] |
| 86 | if not lines: |
| 87 | return "<none>" |
| 88 | preview = [] |
| 89 | for line in lines[:8]: |
| 90 | if len(line) > 240: |
| 91 | line = line[:237].rstrip() + "..." |
| 92 | preview.append(line if line.startswith("-") or line.startswith("#") else f"- {line}") |
| 93 | if len(lines) > 8: |
| 94 | preview.append(f"- <trimmed +{len(lines) - 8} lines>") |
| 95 | return "\n".join(preview) |
| 96 | |
| 97 | def _read_prompt(self, name: str) -> str: |
| 98 | path = self.prompt_dir / name |
| 99 | return path.read_text(encoding="utf-8") if path.exists() else "" |
| 100 | |
| 101 | def _text_stage_complete(self, checklist: dict[str, bool]) -> bool: |
| 102 | idea_mode_complete = bool(checklist.get("idea2video/story.txt") and checklist.get("idea2video/characters.json") and checklist.get("idea2video/script.json") and checklist.get("idea2video/scene_*/storyboard.json") and checklist.get("idea2video/scene_*/shots/*/shot_description.json") and checklist.get("idea2video/scene_*/camera_tree.json")) |
| 103 | script_mode_complete = bool(checklist.get("script2video/script.txt") and checklist.get("script2video/characters.json") and checklist.get("script2video/storyboard.json") and checklist.get("script2video/shots/*/shot_description.json") and checklist.get("script2video/camera_tree.json")) |
| 104 | novel_mode_complete = bool(checklist.get("novel2video/novel/novel_compressed.txt") and checklist.get("novel2video/events/event_*.json") and checklist.get("novel2video/relevant_chunks/event_*") and checklist.get("novel2video/scenes/event_*/scene_*.json") and checklist.get("novel2video/global_information/characters/novel_level/*.json")) |
| 105 | return idea_mode_complete or script_mode_complete or novel_mode_complete |
| 106 |