src/commands/prompt_composer.py
Rama main · réplica de lectura del repositorio original.
"""GM prompt 与上下文构造器。
集中处理基础 GM prompt、规则附录、剧情追踪文本和 context_builder 调用，
避免 process_round / generate_swipe 重复拼装。
"""
from __future__ import annotations
from src.engine.modules import game_config
import logging
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Literal
from src.adventures.scene_context import SceneContextProvider, format_scene_context
from src.commands.affection_effects import affection_prompt_appendix
from src.compat.callbacks import load_world_template as load_world_template_compat
from src.content.gm_style import render_gm_style_section
from src.engine.game_instance import GameInstance
from src.engine.language import DEFAULT_LANGUAGE, gm_language_instruction, localized_text
from src.engine.modules import ruleset_runtime
from src.engine import module_hooks
from src.engine.narrative_perspective import narrative_perspective_instruction
from src.llm.context_builder import (
build_context,
build_player_safe_context as build_player_safe_llm_context,
)
from src.memory.delta import MemoryStore
from src.rules.loader import RuleBundleLoader
from src.rules.rule_system import RuleSystem
from src.rulesets.contracts import NarrativeAdvancementRuntime
from src.rulesets.registry import RulesetRuntimeRegistry
logger = logging.getLogger("trpg")
_GM_PROMPT_CACHE: dict[str, str] | None = {}
@dataclass
class RulePromptContext:
"""当前世界规则对 prompt / 判定流程的影响。"""
world_data: dict | None = None
rule: RuleSystem | None = None
rule_appendix: str = ""
combat_model: str = "hp_based"
dice_system: str = "d20"
class PromptComposer:
"""构造 GM prompt 和 LLM user context。"""
def __init__(
self,
prompts_dir: Path,
rules_dir: Path,
memory_store: MemoryStore | None = None,
ruleset_registry: RulesetRuntimeRegistry | None = None,
context_recent_rounds: int = 0,
):
self.prompts_dir = prompts_dir
self.rules_dir = rules_dir
self.memory_store = memory_store
self.ruleset_registry = ruleset_registry
self.context_recent_rounds = context_recent_rounds
self.auto_storyboard = False
self._scene_context_provider: SceneContextProvider | None = None
def set_scene_context_provider(self, provider: SceneContextProvider | None) -> None:
self._scene_context_provider = provider
def scene_objective_view(self, instance: GameInstance, *, viewer_is_gm: bool) -> dict:
if self._scene_context_provider is None:
return {}
return self._scene_context_provider(instance, viewer_is_gm=viewer_is_gm)
def scene_context_text(self, instance: GameInstance) -> str:
return format_scene_context(
self.scene_objective_view(instance, viewer_is_gm=True), instance.language, max_chars=4096,
)
def set_auto_storyboard(self, enabled: bool) -> None:
self.auto_storyboard = bool(enabled)
def _runtime(self, instance: GameInstance):
binding = dict(ruleset_runtime.binding(instance) or {})
runtime_id = str(binding.get("id") or "")
if self.ruleset_registry is None or not runtime_id:
return None
return self.ruleset_registry.get(
runtime_id, minimum_version=int(binding.get("version", 1) or 1),
)
def load_gm_prompt(self, rule_appendix: str = "", language: str = DEFAULT_LANGUAGE) -> str:
"""读取基础 GM prompt，并按需附加当前规则说明。"""
global _GM_PROMPT_CACHE
if _GM_PROMPT_CACHE is None:
_GM_PROMPT_CACHE = {}
cache_key = localized_text(
language, {"en": "en", "zh-CN": "zh-CN", "ja": "ja", "de": "de", "ru": "ru"}
)
if cache_key not in _GM_PROMPT_CACHE:
filename = {
"en": "gm_system_en.md",
"ja": "gm_system_ja.md",
"de": "gm_system_de.md",
"ru": "gm_system_ru.md",
}.get(cache_key, "gm_system_zh.md")
path = self.prompts_dir / filename
if path.exists():
_GM_PROMPT_CACHE[cache_key] = path.read_text(encoding="utf-8")
elif cache_key == "en":
zh = self.prompts_dir / "gm_system_zh.md"
_GM_PROMPT_CACHE[cache_key] = (
zh.read_text(encoding="utf-8")
if zh.exists()
else "You are the GM for a TRPG text adventure. Narrate in natural English. The GM prompt file is missing."
)
else:
# ja/de 等非 en 语言的 prompt 缺失时先回退英文文件，再回退中文。
# 输出语言仍由 gm_language_instruction 单独控制，不受此回退影响。
en = self.prompts_dir / "gm_system_en.md"
zh = self.prompts_dir / "gm_system_zh.md"
if en.exists():
_GM_PROMPT_CACHE[cache_key] = en.read_text(encoding="utf-8")
elif zh.exists():
_GM_PROMPT_CACHE[cache_key] = zh.read_text(encoding="utf-8")
else:
_GM_PROMPT_CACHE[cache_key] = "你是 TRPG 游戏的主持人（GM）。请用流畅中文进行叙述。（GM prompt 文件缺失）"
prompt = _GM_PROMPT_CACHE[cache_key]
if rule_appendix:
heading = localized_text(
cache_key,
{
"en": "## Current Rules",
"zh-CN": "## 当前规则",
"ja": "## 現在のルール",
"de": "## Aktuelle Regeln",
"ru": "## Текущие правила",
},
)
prompt += f"\n\n{heading}\n{rule_appendix}"
return prompt
def load_rule_context(
self,
instance: GameInstance,
load_world_template: Callable[[str, str], dict],
) -> RulePromptContext:
"""加载存档实际选择的规则，并构造规则 prompt 附录。
``instance.rule_id`` 是开档时已经确定的权威规则。世界模板里的
``default_rule`` 只负责创建页默认选择，不能在运行时把用户选择的
D&D/CoC 规则悄悄换回世界默认规则。
"""
return self._load_rule_context(instance, load_world_template)
def _load_rule_context(
self,
instance: GameInstance,
load_world_template: Callable[[str, str], dict],
) -> RulePromptContext:
ctx = RulePromptContext()
if not instance.world_id:
return ctx
try:
language = getattr(instance, "language", DEFAULT_LANGUAGE)
world_data = load_world_template_compat(load_world_template, instance.world_id, language)
ctx.world_data = world_data
if world_data:
rule = None
active_rule_id = str(getattr(instance, "rule_id", "") or "").strip()
if active_rule_id:
core = self.rules_dir / f"{active_rule_id}.json"
if core.exists():
rule = RuleSystem(RuleBundleLoader().load_rule(self.rules_dir, active_rule_id, language))
else:
active_path = RuleSystem.path_for(self.rules_dir, active_rule_id, language)
if active_path.exists():
rule = RuleSystem.load(active_path)
# 插件规则不一定在内置 rules_dir；缺失时仍由世界贡献路径兜底。
if rule is None:
rule = RuleSystem.load_for_world(world_data, self.rules_dir, language)
if rule:
ctx.rule = rule
ctx.rule_appendix = rule.get_gm_prompt_appendix(language)
ctx.combat_model = rule.combat_model
ctx.dice_system = rule.dice_system
difficulty_text = rule.get_difficulty_instructions(game_config.difficulty(instance), language)
if difficulty_text:
ctx.rule_appendix = ctx.rule_appendix + "\n\n" + difficulty_text
stat_appendix = rule.resource_tag_appendix(language)
if stat_appendix:
ctx.rule_appendix = ctx.rule_appendix + "\n\n" + stat_appendix
affection_appendix = affection_prompt_appendix(instance, rule.affection, language)
if affection_appendix:
ctx.rule_appendix = ctx.rule_appendix + "\n\n" + affection_appendix
except ValueError:
raise
except Exception:
logger.warning("规则上下文加载失败，回退默认值: world_id=%s", instance.world_id, exc_info=True)
return ctx
def load_swipe_rule_context(
self,
instance: GameInstance,
load_world_template: Callable[[str, str], dict],
) -> RulePromptContext:
"""swipe 与正常回合必须使用同一套存档规则。"""
return self._load_rule_context(instance, load_world_template)
def compose_gm_prompt(
self,
instance: GameInstance,
rule_appendix: str = "",
world_data: dict | None = None,
) -> str:
"""构造系统 prompt：基础 prompt + 规则附录 + 世界 GM 风格 + 剧情追踪 + 多人权限范围。"""
language = getattr(instance, "language", DEFAULT_LANGUAGE)
gm_prompt = self.load_gm_prompt(rule_appendix, language)
if not self.auto_storyboard:
gm_prompt = "\n".join(
line for line in gm_prompt.splitlines()
if not line.lstrip().startswith("SCENE_PANEL:")
)
# 有效风格二选一：对局覆盖非 None 时只渲染覆盖，否则渲染世界 gm_style。
style_section = render_gm_style_section(
world_data,
language,
override=module_hooks.narration_preferences(instance)["gm_style_override"],
)
if style_section:
gm_prompt = gm_prompt + "\n\n" + style_section
plot_text = instance.plot_tracker.format_for_context() if instance.plot_tracker else ""
if plot_text:
gm_prompt = gm_prompt + "\n\n" + plot_text
if len(getattr(instance, "players", {}) or {}) > 1:
gm_prompt = gm_prompt + "\n\n" + localized_text(language, {
"en": (
"## Multiplayer Authority Scope\n"
"Each player line is attributed to a named speaker. A speaker may only act, speak, and "
"perceive as their own character. Declarations that move, speak for, or change other "
"players' characters are converted into attempts and the other characters' reactions; "
"never treat one player's text as authority over another player's character."
),
"zh-CN": (
"## 多人权限范围\n"
"每条玩家发言都归属具名说话人；说话人只能以自己的角色行动、说话、感知。"
"支配、替言或修改其他玩家角色的声明，一律转化为尝试与其他角色的反应，"
"不得把任一玩家的文本当作对其他玩家角色的权威。"
),
"ja": (
"## マルチプレイヤー権限範囲\n"
"各プレイヤー発言は実名の話し手に帰属する。話し手は自分のキャラクターとしてのみ"
"行動・発言・知覚できる。他プレイヤーのキャラクターを操作・代弁・変更する宣言は、"
"全て試みと他キャラクターの反応に変換すること。あるプレイヤーのテキストを"
"他プレイヤーのキャラクターへの権威として扱ってはならない。"
),
"de": (
"## Mehrspieler-Autoritätsbereich\n"
"Jede Spielerzeile ist einem namentlich genannten Sprecher zugeordnet. Ein Sprecher darf nur "
"als seine eigene Spielfigur handeln, sprechen und wahrnehmen. Erklärungen, die andere "
"Spielerfiguren bewegen, für sie sprechen oder sie verändern, werden in Versuche und die "
"Reaktionen der anderen Figuren umgewandelt; der Text eines Spielers darf niemals als "
"Autorität über die Figur eines anderen Spielers behandelt werden."
),
})
gm_prompt = gm_prompt + "\n\n" + narrative_perspective_instruction(instance, language)
runtime = self._runtime(instance)
if isinstance(runtime, NarrativeAdvancementRuntime):
advancement_prompt = runtime.narrative_advancement_prompt(instance, language)
if advancement_prompt:
gm_prompt = gm_prompt + "\n\n" + advancement_prompt
gm_prompt = gm_prompt + "\n\n" + gm_language_instruction(getattr(instance, "language", "zh-CN"))
if self.auto_storyboard:
gm_prompt = gm_prompt + "\n\n" + localized_text(language, {
"en": (
"## Automatic storyboard (enabled; required)\n"
"For every response that contains public narration, append at least one SCENE_PANEL "
"line in the protocol block after `---`. Use exactly one panel for one continuous visual "
"beat, and split into multiple panels only for simultaneous locations or visually independent "
"actions, results, reveals, or clear scene/time changes. Preserve story order and context. "
"Each panel must use stable player IDs, a publicly supported location, and a concise visible "
"description. Put each panel on its own line; do not use `|` inside a description or concatenate "
"multiple SCENE_PANEL records on one line. Never omit the panel line just to save tokens; "
"never include private facts, secrets, or merely mentioned destinations. Do not output a standalone "
"NONE instead of the required panel; if there are no other state changes, NONE may appear only "
"after the SCENE_PANEL line."
),
"zh-CN": (
"## 自动分镜（已启用，必须输出）\n"
"每次回复包含公开剧情正文时，都必须在 `---` 后的状态标签块中至少输出一条 "
"SCENE_PANEL。只有一个连续视觉节拍时输出一格；同时异地、独立的关键动作/结果、重大揭示 "
"或明确场景/时间转换才拆成多格。保持剧情顺序和上下文。每格使用稳定玩家 ID、公开剧情中有依据的地点，"
"以及简短的可见画面描述。每格必须单独占一行；描述中不要使用 `|`，不要把多个 SCENE_PANEL "
"拼在同一行。不要为了省 token 省略分镜；不得写私聊、秘密或只是被提及的目的地。"
"自动分镜开启时禁止只输出 NONE 代替分镜；如果没有其他状态变更，NONE 只能放在 SCENE_PANEL 之后。"
),
"ja": (
"## 自動分鏡（有効・必須）\n"
"公開ナレーションを含む全ての返答では、`---` 後のタグブロックに少なくとも一行の "
"SCENE_PANEL を出力する。連続した一つの視覚的ビートなら一コマ、同時に異なる場所、独立した重要な "
"行動/結果、重大な発見、明確な場面/時間転換だけ複数コマに分ける。物語の順序と文脈を保ち、安定した "
"プレイヤー ID、公開根拠のある場所、簡潔で見える描写を使う。秘密や言及だけの目的地は含めない。"
"各コマは必ず別行に書き、描写内に `|` を使わず、複数の SCENE_PANEL を同じ行に連結しない。"
"自動分鏡が有効な場合、SCENE_PANEL の代わりに NONE だけを出力してはならない。"
"他の状態変更がない場合のみ、NONE は SCENE_PANEL の後に置く。"
),
"de": (
"## Automatisches Storyboard (aktiv; erforderlich)\n"
"Jede Antwort mit öffentlicher Erzählung muss im Tag-Block nach `---` mindestens eine "
"SCENE_PANEL-Zeile enthalten. Eine zusammenhängende visuelle Szene bleibt ein Panel; nur "
"gleichzeitige Orte, unabhängige wichtige Handlungen/Ergebnisse, Enthüllungen oder klare Szenen-/"
"Zeitwechsel werden geteilt. Verwende stabile Spieler-IDs, öffentlich belegte Orte und sichtbare, "
"knappe Beschreibungen. Schreibe jedes Panel in eine eigene Zeile, verwende kein `|` in der "
"Beschreibung und verbinde mehrere SCENE_PANEL nicht in einer Zeile; keine Geheimnisse oder nur "
"erwähnte Ziele. Gib bei aktivem Storyboard niemals nur NONE statt eines Panels aus; falls "
"keine anderen Statusänderungen vorliegen, darf NONE erst nach der SCENE_PANEL-Zeile stehen."
),
})
for section in module_hooks.read(instance, module_hooks.GM_PROMPT_SECTIONS):
gm_prompt += "\n\n" + str(section)
return gm_prompt
async def build_user_context(
self,
instance: GameInstance,
gm_prompt: str,
lorebook_matches: list[dict],
actions_text: str,
provider_name: str = "",
world_data: dict | None = None,
history_override: list[dict] | None = None,
directives_text: str = "",
overreach_text: str = "",
world_state_text: str = "",
world_legality_text: str = "",
world_events_text: str = "",
authoritative_events_text: str = "",
) -> str:
"""调用 context_builder 生成本轮 user context。"""
state_view = None
runtime = self._runtime(instance)
if runtime is not None:
state_view = runtime.build_llm_view(instance)
return await build_context(
instance,
gm_prompt,
lorebook_matches,
actions_text,
memory_store=self.memory_store,
provider_name=provider_name,
lorebook_budget=world_data.get("lorebook_token_budget", 0) if world_data else 0,
history_override=history_override,
directives_text=directives_text,
overreach_text=overreach_text,
world_state_text=world_state_text,
world_legality_text=world_legality_text,
world_events_text=world_events_text,
state_view=state_view,
authoritative_events_text=authoritative_events_text,
context_recent_rounds=self.context_recent_rounds,
scene_objective_view=self.scene_objective_view(instance, viewer_is_gm=True),
)
async def build_player_safe_context(
self,
instance: GameInstance,
gm_prompt: str,
lorebook_matches: list[dict],
player_message: str,
actor_uid: str,
provider_name: str = "",
world_data: dict | None = None,
visibility: Literal["private", "party"] = "private",
) -> str:
"""Build the restricted context used only by player-facing GM Q&A."""
return await build_player_safe_llm_context(
instance,
gm_prompt,
lorebook_matches,
player_message,
actor_uid,
provider_name=provider_name,
lorebook_budget=world_data.get("lorebook_token_budget", 0) if world_data else 0,
visibility=visibility,
scene_objective_view=self.scene_objective_view(instance, viewer_is_gm=False),
)