src/commands/swipe_generator.py
Rama main · réplica de lectura del repositorio original.
"""Swipe 候选叙事生成。"""
from __future__ import annotations
import logging
from collections.abc import Callable
from copy import deepcopy
from typing import Any
from src.adventures.graph_v2 import ADVENTURE_GRAPH_FORMAT_V2
from src.commands.economy_effects import (
currency_labels_for_rule,
defer_narrative_effects,
has_economy_proposal,
pending_decision_notice,
unbacked_purchase_notice,
unbacked_payment_notice,
should_warn_unbacked_payment,
)
from src.engine.economy import filter_unconfirmed_purchase_grants
from src.engine import combat_narrative, module_hooks, progression
from src.commands.protocol_repair import repair_malformed_protocol_response
from src.commands.round_actions import format_check_results_constraint
from src.commands.state_update_applier import StateUpdateApplier, discard_unresolved_player_damage
from src.commands.tag_parser import parse_tag_state
from src.commands.story_output import ensure_story_response, bound_story_response
from src.engine.game_instance import GameInstance, restore_players
from src.engine.modules import adventure_runtime_state, progression_state, session_stats
from src.engine.world_state import ensure_world_state
from src.engine.economy import queue_effect_group, reconcile_rollback_snapshot, reverse_round_economy
from src.imagegen.storyboards import normalize_scene_panels, storyboard_panel_metadata, storyboard_source_revision
from src.llm.context_builder import lore_char_budget
from src.llm.parser import normalize_tag_protocol, sanitize_narration
from src.lorebook.retrieval import LoreRetriever
logger = logging.getLogger("trpg")
class SwipeGenerator:
"""为指定历史轮次重新生成一个候选叙事。"""
def __init__(
self,
llm_client: Any,
matcher: Any,
prompt: Any,
state_applier: StateUpdateApplier,
load_world_template: Callable[[str, str], dict | None],
ensure_matcher_for_world: Callable[[str, str], None],
narrative_max_tokens: int,
get_instance: Callable[[tuple], GameInstance | None] | None = None,
save_instance: Callable[[GameInstance], Any] | None = None,
lore_retriever: Any | None = None,
):
self.llm_client = llm_client
self.matcher = matcher
self.lore_retriever = lore_retriever or LoreRetriever(matcher)
self.prompt = prompt
self.state_applier = state_applier
self.load_world_template = load_world_template
self.ensure_matcher_for_world = ensure_matcher_for_world
self.narrative_max_tokens = narrative_max_tokens
self.get_instance = get_instance
self.save_instance = save_instance
# 生图调度回调（GameHandler 注入）：swipe 叙事带新 SCENE_IMAGE 时重新生成该回合图片
self._scene_image_hook = None
def set_scene_image_hook(self, hook) -> None:
self._scene_image_hook = hook
async def generate(self, instance: GameInstance, round_num: int) -> str | None:
"""Run the complete historical rewrite under the shared process barrier."""
if instance._process_lock.locked():
logger.warning(
"process_round/swipe 进行中，跳过并发 generate_swipe: %s",
instance.game_key,
)
return None
async with instance.historical_rewrite() as rewrite_entered:
if not rewrite_entered:
return None
async with instance._process_lock:
session_stats.require_writable(instance)
adventure_runtime_state.require_writable(instance)
module_hooks.preflight(instance, module_hooks.NARRATIVE_VIEW)
expected_run_id = instance.run_id
before = type(instance).from_dict(deepcopy(instance.to_dict()))
before.log = deepcopy(instance.log)
staged = type(instance).from_dict(deepcopy(instance.to_dict()))
staged.log = deepcopy(instance.log)
narration, scene_payload = await self._generate_locked(staged, round_num)
if narration is None:
return None
if self.get_instance:
current = self.get_instance(instance.game_key)
if current is not instance or getattr(current, "run_id", None) != expected_run_id:
return None
instance.replace_persisted_state_from(staged)
try:
if self.save_instance is not None:
await self.save_instance(instance)
except Exception:
instance.replace_persisted_state_from(before)
raise
if scene_payload and self._scene_image_hook:
try:
self._scene_image_hook(
instance,
str(scene_payload.get("prompt") or ""),
int(scene_payload.get("round", round_num) or round_num),
force=True,
panels=scene_payload.get("panels") or [],
compressed_count=int(scene_payload.get("compressed_count") or 0),
)
except Exception:
logger.exception("Swipe 场景图调度失败 (round=%d)", round_num)
return narration
async def _generate_locked(
self, instance: GameInstance, round_num: int,
) -> tuple[str | None, dict[str, Any] | None]:
"""为指定轮生成一个新 swipe（最多 5 个）。"""
target_entry = None
target_idx = -1
for i, entry in enumerate(instance.log):
if entry.get("round") == round_num:
target_entry = entry
target_idx = i
break
if not target_entry:
return None, None
from src.engine.modules import resource_recovery, skill_growth
skill_growth.rewind_preflight(instance, round_num)
resource_recovery.rewind_preflight(instance, round_num)
swipes = target_entry.get("swipes", [])
if len(swipes) >= 5:
logger.warning("Swipe 已达上限 (5), round=%d", round_num)
return None, None
progression.require_writable(instance)
session_stats.require_writable(instance)
if not swipes:
swipes = [target_entry.get("gm_response", "")]
target_entry["swipes"] = swipes
progress_snapshot = target_entry.get("pre_adventure_progress")
snapshot = target_entry.get("pre_state_snapshot", {})
if ((instance.adventure_binding or {}).get("format") == ADVENTURE_GRAPH_FORMAT_V2
and (not isinstance(progress_snapshot, dict) or not isinstance(snapshot, dict) or not snapshot)):
logger.warning("Cannot rewrite v2 adventure without its pre-round snapshots: %s round=%s",
instance.game_key, round_num)
return None, None
if snapshot:
# ADR 0003 branch cut: rounds after the rewritten target belong to
# the discarded branch. Their log entries are dropped together
# with their economy (reverse_round_economy withdraws every
# settlement at or after the target round), and the replay
# continues from the target round itself.
instance.log = [
entry for entry in instance.log
if int(entry.get("round", 0) or 0) <= round_num
]
module_hooks.run(module_hooks.ROUND_REWIND_BEGIN, instance, round_num)
progression.rewind_for_replay(instance, round_num)
reverse_round_economy(instance, round_num)
restore_players(instance, reconcile_rollback_snapshot(instance, snapshot, round_num))
module_hooks.run(
module_hooks.ROUND_REWIND_RESTORE, instance, target_entry, round_num,
module_hooks.REWIND_SOURCE_SWIPE,
)
# 世界真相同属被丢弃的分支：swipe 切回目标轮时，本轮之后写入的
# world ops 一起撤销（ADR 0003 整轮语义）。
world_snapshot = target_entry.get("pre_world_state")
if isinstance(world_snapshot, dict) and world_snapshot:
instance.world_state = ensure_world_state(world_snapshot)
logger.info("Swipe: 已恢复 pre-state snapshot (round=%d)", round_num)
actions_text = "; ".join(
a.get("text", "") for a in target_entry.get("actions", [])
if a.get("user_id") in instance.players
)
actions_text += format_check_results_constraint(
instance, list(target_entry.get("check_results") or [])
)
if instance.world_id:
self.ensure_matcher_for_world(instance.world_id, instance.language)
# 与正常回合同一个 LoreRetriever：swipe 在 staged 克隆上重放同一轮，计时器
# 语义与正常回合一致（匹配结果随后经 replace_persisted_state_from 写回）。
action_actor_uids = sorted({
str(action.get("user_id") or "")
for action in target_entry.get("actions", [])
if str(action.get("user_id") or "") in instance.players
})
lorebook_matches = await self.lore_retriever.retrieve(
instance, actions_text, action_actor_uids=action_actor_uids,
overall_budget=lore_char_budget(
self.llm_client.default if getattr(self, "llm_client", None) else ""
),
)
rule_ctx = self.prompt.load_swipe_rule_context(instance, self.load_world_template)
combat_model_s = rule_ctx.combat_model
world_data = rule_ctx.world_data
currency_labels = currency_labels_for_rule(rule_ctx.rule)
gm_prompt = self.prompt.compose_gm_prompt(
instance, rule_ctx.rule_appendix, world_data=rule_ctx.world_data,
)
pending_combat_event_ids = combat_narrative.pending_event_ids(instance)
pending_combat_events_text = combat_narrative.format_pending_events(instance)
# 构建上下文（仅使用目标轮之前的日志），不临时改写共享 instance.log。
provider_name = self.llm_client.default if self.llm_client else ""
context = await self.prompt.build_user_context(
instance,
gm_prompt,
lorebook_matches,
actions_text,
provider_name=provider_name,
world_data=world_data,
history_override=instance.log[:target_idx],
authoritative_events_text=pending_combat_events_text,
)
response = await self.llm_client.call(
system_prompt=gm_prompt,
user_message=context,
temperature=0.9,
max_tokens=self.narrative_max_tokens,
)
response = await repair_malformed_protocol_response(
self.llm_client,
response,
system_prompt=gm_prompt,
user_message=context,
language=instance.language,
temperature=0.9,
max_tokens=self.narrative_max_tokens,
)
response.content = normalize_tag_protocol(response.content)
response = await ensure_story_response(
instance, response, self.llm_client, system_prompt=gm_prompt,
user_message=context, max_tokens=self.narrative_max_tokens,
round_number=round_num, revision=len(swipes),
)
narration = response.content
if "---" in response.content:
narration = response.content.split("---", 1)[0].strip()
narration = sanitize_narration(narration)
data = parse_tag_state(response.content, combat_model_s)
if getattr(response, "story", None):
bound_story_response(response, data, trigger=500 if instance.language == "zh-CN" else 1200,
limit=400 if instance.language == "zh-CN" else 1100)
narration = response.narration
system_changes: list[str] = []
if should_warn_unbacked_payment(
narration, data, instance.language,
currency_labels=currency_labels,
):
system_changes.append(unbacked_payment_notice(instance.language))
dropped_purchase_items = filter_unconfirmed_purchase_grants(instance, data)
if dropped_purchase_items:
system_changes.append(unbacked_purchase_notice(instance.language))
deferred_effects = defer_narrative_effects(
data, response,
defer_state_update=True,
)
economy_pending = has_economy_proposal(data)
if economy_pending:
system_changes.append(pending_decision_notice(instance.language))
queued_proposals: list[dict[str, Any]] = []
if data.get("state_update"):
discard_unresolved_player_damage(instance, data.get("state_update", {}))
queued_proposals = self.state_applier.apply_state_update(
instance, data.get("state_update", {}),
)
if deferred_effects:
deferred_effects["allowed_player_uids"] = None
queue_effect_group(instance, queued_proposals, deferred_effects)
plot_update = data.get("plot_update")
if isinstance(plot_update, dict) and "adventure_completion" in plot_update:
plot_update = dict(plot_update)
plot_update.pop("adventure_completion", None)
if plot_update and instance.plot_tracker:
try:
instance.plot_tracker.apply_update(
plot_update, progression_state.round_value(instance))
except Exception:
logger.exception("Swipe 剧情更新异常，已跳过 (round=%d)", round_num)
await instance.finish_judgment_with_swipe(
narration, round_num,
state_changes=system_changes if system_changes else None,
story=getattr(response, "story", None),
adventure_completion_candidate=(
data.get("plot_update", {}).get("adventure_completion")
if isinstance(data.get("plot_update"), dict) else None
),
)
combat_narrative.consume_pending_events(
instance, pending_combat_event_ids,
)
scene_payload = None
swipe_prompt = str(data.get("scene_image_prompt") or "").strip()
swipe_panels = data.get("scene_panels")
normalized_panels, compressed_count = normalize_scene_panels(
swipe_panels, merge_same_location=False,
)
# A swipe is a new public-story revision. Keep its storyboard draft
# beside the narration (never derive it later from an old image).
original_panels = deepcopy(target_entry.get("scene_panels") or [])
original_prompt = str(target_entry.get("scene_image_prompt") or "")
target_entry["scene_panels"] = normalized_panels
target_entry["scene_image_prompt"] = swipe_prompt[:300]
target_entry["scene_panel_meta"] = storyboard_panel_metadata(
normalized_panels,
narration=str(target_entry.get("gm_response") or ""),
actions=target_entry.get("actions") or [],
current_scene=str(module_hooks.narrative_view(instance)["scene"] or ""),
source_revision=storyboard_source_revision(target_entry),
)
prompt_history = target_entry.setdefault("swipe_scene_image_prompts", [])
if not isinstance(prompt_history, list):
prompt_history = []
target_entry["swipe_scene_image_prompts"] = prompt_history
while len(prompt_history) < len(target_entry.get("swipes", [])) - 1:
prompt_history.append(original_prompt if not prompt_history else "")
prompt_history.append(swipe_prompt[:300])
swipe_panel_history = target_entry.setdefault("swipe_scene_panels", [])
if not isinstance(swipe_panel_history, list):
swipe_panel_history = []
target_entry["swipe_scene_panels"] = swipe_panel_history
# The first item corresponds to the original swipe; subsequent items
# follow the same order as the public ``swipes`` text list.
while len(swipe_panel_history) < len(target_entry.get("swipes", [])) - 1:
swipe_panel_history.append(deepcopy(original_panels) if not swipe_panel_history else [])
swipe_panel_history.append(deepcopy(normalized_panels))
if not normalized_panels:
target_entry.pop("scene_panel_meta", None)
if swipe_prompt or normalized_panels:
scene_payload = {
"prompt": swipe_prompt,
"round": round_num,
"panels": normalized_panels,
"compressed_count": compressed_count,
}
logger.info("Swipe 生成: round=%d swipe=%d/%d", round_num,
len(swipes) + 1, len(swipes) + 1)
return narration, scene_payload