""" story_engine.py - StoryWeaver 叙事引擎 职责： 1. 封装故事生成的核心逻辑（两阶段生成策略） 2. 生成连贯、有文学色彩的剧情段落 3. 生成 3 个后续选项供玩家选择 4. 输出结构化的状态变更数据 两阶段生成策略 (Chain of Thought)： - 第一阶段：让 Qwen 生成 JSON 格式的剧情大纲（事件、地点变化、NPC反应、状态变更） - 第二阶段：基于大纲生成具体的描写文本 + 3 个选项 目的： - 第一阶段便于程序解析状态变化，保证数据准确 - 第二阶段保证文本质量和文学色彩 """ import logging import random import re from typing import Any, Optional from demo_rules import ( DEEP_FOREST_BARRIER_SEEN_FLAG, FOREST_CAUSE_OBJECTIVE, FOREST_GOBLIN_DEFEATED_FLAG, FOREST_TROLL_BOSS_OBJECTIVE, FOREST_TROLL_DEFEATED_FLAG, FOREST_TROLL_HOARD_CLAIMED_FLAG, FOREST_TROLL_HOARD_PENDING_FLAG, FOREST_TROLL_INTRO_SEEN_FLAG, FOREST_TROLL_TRACKS_FOUND_FLAG, FOREST_TROLL_TRAVEL_OBJECTIVE, FERRY_ROUTE_UNLOCKED_FLAG, GUARDIAN_INTRO_SEEN_FLAG, LOCATION_MAP_REQUIREMENTS, MAIN_QUEST_ID, MAIN_QUEST_TROLL_ID, MINE_RUMOR_HEARD_FLAG, REPORT_TO_CHIEF_OBJECTIVE, SIDE_QUEST_FERRY_ID, SIDE_QUEST_GUARDIAN_ID, SIDE_QUEST_TRAVELER_ID, TRAVELER_ENCOUNTERED_FLAG, TRAVELER_RUMOR_HEARD_FLAG, DEFAULT_OPTION_COUNT, MAX_OPTION_COUNT, build_arrival_event, build_adjacent_actions, build_contextual_actions, get_battle_encounter, build_goal_directed_actions, build_map_actions, build_scene_actions, build_shop_menu_actions, build_village_chief_follow_up_actions, merge_demo_options, resolve_trade, ) from combat_engine import get_monster_profile, resolve_combat from utils import call_qwen, call_qwen_stream, safe_json_call, extract_json_from_text, DEFAULT_MODEL from state_manager import GameState logger = logging.getLogger("StoryWeaver") # ============================================================ # 变更日志合并工具 # ============================================================ # 用于从 "饱食度自然衰减: 100 → 97" 或 "饱食度: 97 → 95" 中提取属性名和数值 _CHANGE_LOG_PATTERN = re.compile( r"^(.+?)(?:自然衰减)?:\s*(\S+)\s*→\s*(\S+)$" ) def _merge_change_logs(tick_log: list[str], action_log: list[str]) -> list[str]: """ 合并时间流逝日志与行动变更日志，避免同一属性出现两行。 例如： tick_log: ["饱食度自然衰减: 100 → 97"] action_log: ["饱食度: 97 → 95"] 合并为： ["饱食度: 100 → 95"] 原理：按属性名匹配，如果 tick_log 中某属性的终值 == action_log 中同属性的起始值， 则合并为一条：属性名: tick起始值 → action终值。 """ # 从 tick_log 中提取 {属性名: (起始值, 终值, 原始索引)} tick_attrs: dict[str, tuple[str, str, int]] = {} for i, line in enumerate(tick_log): m = _CHANGE_LOG_PATTERN.match(line.strip()) if m: attr_name = m.group(1).strip() tick_attrs[attr_name] = (m.group(2), m.group(3), i) # 标记哪些 tick_log 条目被合并掉了 tick_merged: set[int] = set() merged_results: list[str] = [] for line in action_log: m = _CHANGE_LOG_PATTERN.match(line.strip()) if m: attr_name = m.group(1).strip() action_start = m.group(2) action_end = m.group(3) if attr_name in tick_attrs: tick_start, tick_end, tick_idx = tick_attrs[attr_name] if tick_end == action_start: # 可以合并：使用 tick 的起始值 + action 的终值 if tick_start != action_end: merged_results.append(f"{attr_name}: {tick_start} → {action_end}") # 如果起始值==终值（变了又变回来），直接省略此条 tick_merged.add(tick_idx) continue merged_results.append(line) # 输出：未被合并的 tick_log 条目 + 合并后的 action_log 条目 remaining_tick = [ line for i, line in enumerate(tick_log) if i not in tick_merged ] return remaining_tick + merged_results def _normalize_markers(text: str) -> str: """ 标准化 LLM 输出中的分隔标记，处理常见变体格式。 LLM 有时会输出与预期略有不同的标记格式，例如： - 多余的空格: "--- STORY_TEXT ---" - 不同的连字符数量: "----STORY_TEXT----" - 大小写变化: "---story_text---" - 下划线变空格: "---STORY TEXT---" 此函数将这些变体统一为标准格式。 """ text = re.sub(r'-{2,}\s*STORY[_ ]?TEXT\s*-{2,}', '---STORY_TEXT---', text, flags=re.IGNORECASE) text = re.sub(r'-{2,}\s*OPTIONS[_ ]?JSON\s*-{2,}', '---OPTIONS_JSON---', text, flags=re.IGNORECASE) text = re.sub(r'-{2,}\s*STATE[_ ]?JSON\s*-{2,}', '---STATE_JSON---', text, flags=re.IGNORECASE) text = re.sub(r'-{2,}\s*THINKING\s*-{2,}', '---THINKING---', text, flags=re.IGNORECASE) return text def _build_telemetry( engine_mode: str, *, used_fallback: bool = False, fallback_reason: str | None = None, consistency_issues_count: int = 0, validation_issues_count: int = 0, outline_regenerated: bool = False, ) -> dict: """构建供日志与评估脚本使用的轻量运行元信息。""" return { "engine_mode": engine_mode, "used_fallback": used_fallback, "fallback_reason": fallback_reason, "consistency_issues_count": consistency_issues_count, "validation_issues_count": validation_issues_count, "outline_regenerated": outline_regenerated, } # ============================================================ # Prompt 模板设计 # ============================================================ # ------------------------------------------------------------ # 第一阶段 Prompt：生成剧情大纲（结构化 JSON） # # 设计思路： # - System Prompt 注入当前完整状态（来自 state_manager.to_prompt()） # - 要求 LLM 严格输出 JSON，包含：事件描述、状态变更、NPC 反应 # - 低温度 (0.3) 确保 JSON 结构稳定 # - 明确指定每个字段的含义和格式 # ------------------------------------------------------------ OUTLINE_SYSTEM_PROMPT_TEMPLATE = """你是一个专业的 RPG 叙事引擎的规划模块。你的任务是根据玩家的行动和当前世界状态，生成剧情大纲。 {world_state} 【你的任务】 根据玩家的行动意图，生成一个 JSON 格式的剧情大纲。你必须： 1. 考虑当前所有状态（HP、位置、NPC态度、背包物品等） 2. 确保剧情发展合理，不违反一致性约束 3. 生成合理的状态变更数据 4. 考虑剧情的趣味性和戏剧张力 请严格按以下 JSON 格式输出（不要输出任何其他文字）： {{ "event_summary": "简短描述发生了什么事（一句话）", "event_type": "事件类型（COMBAT/DIALOGUE/MOVE/ITEM/QUEST/TRADE/REST/DISCOVERY）", "involved_npcs": ["涉及的NPC名称列表"], "location": "玩家在本回合行动后的最终地点名称（若未移动，则为当前地点）", "state_changes": {{ "hp_change": 0, "mp_change": 0, "gold_change": 0, "exp_change": 0, "morale_change": 0, "sanity_change": 0, "hunger_change": 0, "karma_change": 0, "new_location": null, "items_gained": [], "items_lost": [], "skills_gained": [], "status_effects_added": [], "status_effects_removed": [], "npc_changes": {{}}, "quest_updates": {{}}, "weather_change": null, "time_change": null, "global_flags_set": {{}}, "world_event": null, "equip": {{}}, "title_change": null }}, "npc_reactions": {{ "NPC名称": "NPC的反应描述" }}, "scene_atmosphere": "场景氛围描述（天气、光线、声音等）", "consequence_tags": ["后果标签列表，如 angered_dragon"], "is_reversible": true, "danger_hint": "如果有潜在危险，在这里提示" }} 注意： - state_changes 中只填写确实发生变化的字段，未变化的字段可以省略或设为 null/0/空 - npc_changes 格式: {{"NPC名称": {{"attitude": "新态度", "relationship_change": 数值, "memory_add": "新记忆", "hp_change": 数值}}}} - quest_updates 格式: {{"任务ID": {{"objectives_completed": ["完成的目标"], "status": "新状态"}}}} - 确保所有数值变更合理（例如战斗伤害应考虑攻防差值） - 【关键】所有数值变更必须精确，禁止使用“恢复一些”“大幅降低”等模糊描述，必须给出精确数字（如 hunger_change: 14）。 - 【关键】time_change 字段只允许以下值之一："清晨""上午""正午""下午""黄昏""夜晚""深夜"，不要填写其他格式（如"30分钟""两小时后"等都是非法值）。如果本回合没有发生时间跳跃（例如休息、等待、长途旅行等），请设为 null（系统会自动推进一个时段）。只有当剧情需要跳跃多个时段时才设置此字段。 - 【关键】游戏内的货币单位统一为"金币"，对应 gold_change 字段。严禁使用"银币""铜币""银两""钱币"等其他货币名称。任何财物/钱财类收获必须通过 gold_change 字段表达，严禁将钱币放入 items_gained。举例：击败怪物掉落 3 金币 → gold_change: 3，绝对不要在 items_gained 中放入"铜币""银币""金币"等。 - 【关键 —— 装备规则】当玩家装备某个物品时，必须在 equip 字段中指定对应槽位和物品名称（如 "weapon": "小刀"）。系统会自动将装备的物品从背包移到装备栏，并将替换下来的旧装备放回背包。因此装备操作时不要在 items_lost/items_gained 中重复处理该物品。合法槽位：weapon / armor / accessory / helmet / boots。卸下装备时将对应槽位设为 null。 - 【关键】status_effects_added 中每个效果必须是一个对象，且必须包含以下字段： - "name": 中文效果名称（如"中毒""祝福""饱腹"），必须是中文，不要使用英文字段名 - "description": 获得原因的简短说明（如"食用面包后精力充沛""被毒蛇咬伤"） - "effect_type": "buff" 或 "debuff" 或 "neutral" - "stat_modifiers": 具体属性影响的字典（如 {{"hp": 5}} 表示每回合恢复5HP） - "duration": 持续回合数（正整数，-1表示永久） 示例: {{"name": "饱腹", "description": "吃了面包后恢复了体力", "effect_type": "buff", "stat_modifiers": {{"hp": 2}}, "duration": 3}} - 【关键】state_changes 是程序更新游戏状态的唯一数据源，你输出的 JSON 将被直接用于更新游戏状态。请先在心中做算术验证，确保变更后的数值不超出合法范围（HP ∈ [0, max_hp]，饱食度/士气/理智 ∈ [0, 100]）。 - 【关键】请在生成状态变更 JSON 时，确保数值的精确性和一致性。不要在 JSON 中使用模糊或近似的数值。 - 【关键】如果某个字段没有发生变化，请完全省略该字段或将其设为 null / 0 / 空。绝对不要将实际值写成字符串 "None"。特别注意：new_location 仅在玩家移动到新地点时填写（必须是具体的地点名称），weather_change 仅在天气真正改变时填写（必须是具体天气名称），title_change 仅在称号真正变更时填写（必须是具体称号名称），world_event 仅在发生世界级大事件时填写。如果没有变化，这些字段统一设为 null 或省略。 - 【关键】quest_updates 中的 status 字段只允许以下值："active"（进行中）、"completed"（已完成）、"failed"（已失败）。不要使用英文大写或其他变体如 "IN_PROGRESS" "ACTIVE" 等。 - 【关键】status_effects_added 不是每回合都必须添加的。只有在剧情确实发生了影响角色身心状态的重要事件时才添加状态效果（例如中毒、受到祝福、吃了特殊食物等）。普通的对话、行走、观察等日常行为不应产生状态效果。大约每 3-5 个回合才可能自然产生一次状态效果。同时，状态效果应正向（buff）和负向（debuff）均衡分布，不要总是给予负面效果。 - 【极其重要 —— 物品消耗规则】items_lost 字段仅用于记录真正消失的物品。判断标准如下： - 消耗品（药水、食物、卷轴、弹药等一次性物品）：使用后会消失，应放入 items_lost。 - 非消耗品（哨子、武器、护甲、工具、乐器、钥匙、火把、绳索等可重复使用的物品）：使用后仍然保留在背包中，绝对不要放入 items_lost。 - 举例：吹响哨子 → 哨子仍在背包中（不放入 items_lost）；吃了面包 → 面包消失（放入 items_lost）；使用火把照明 → 火把仍在（不放入 items_lost）。 - 在背包列表中，标注了[消耗品]的物品才是消耗品，标注[可重复使用]的物品绝对不能因使用而消失。 - 如果不确定某个物品是否为消耗品，就不要将它放入 items_lost。 """ # ------------------------------------------------------------ # 第二阶段 Prompt：基于大纲生成文本 + 选项 # # 设计思路： # - 将第一阶段的大纲作为硬性约束注入 # - 要求生成有文学色彩的描写段落 # - 额外生成 3 个选项，每个选项有标签和描述 # - 选项应覆盖不同策略（如激进/保守/探索） # - 中等温度 (0.8) 增加文学创意 # ------------------------------------------------------------ NARRATIVE_SYSTEM_PROMPT_TEMPLATE = """【最高优先级指令 ── 输出格式】 你的输出必须严格遵守以下格式，这是最高优先级的要求，违反将导致系统崩溃： 1. 先输出 ---STORY_TEXT--- 标记（独占一行） 2. 然后直接写故事内容，不加任何前缀（禁止“好的”“以下是”等） 3. 故事写完后输出 ---OPTIONS_JSON--- 标记（独占一行） 4. 最后输出 JSON 格式的 3 个选项数组 绝对不要省略这两个标记。 你是一个经验丰富的奇幻小说家，正在为一个互动 RPG 游戏编写剧情文本。 {world_state} 【剧情大纲（必须严格遵守，不可偏离）】 {outline} 【实际状态变更记录（数值必须以此为准，不可偏离）】 {actual_changes} 当你在叙事中提及任何属性数值变化（如 HP、饱食度、金币等）时，必须与上方「实际状态变更记录」完全一致，不要自行推算或编造数值。 【你的任务】 1. 基于以上大纲，写一段剧情描写（200-400字，使用中文）。 - 使用第二人称（“你”）叙述 - 【最重要】文本开头必须直接回应玩家的行动或选择，让玩家明确感受到自己的操作产生了效果。例如玩家选择“和铁匠交谈”，开头就应写玩家走向铁匠并开始对话，而不是先写一大段环境描写。 - 文风要求：简洁、自然、有画面感。像写小说而不是写诗歌。少用比喻和修辞，多用具体的动作和对话。不要堆砌华丽辞藻，不要每句都加形容词。 - 描写要具体：与其说“空气中弥漫着岁月的沉淀”，不如说“柜台上摆着一壶凉透的茶”。用具体的事物代替抽象的感受。 - 禁止使用以下老套表达：“阳光洒下”“微风拂过”“空气中弥漫着”“XX与 XX 交织在一起”“如同XX般”。如果你压根想不到新鲜的表达，就用最普通的叙述即可。 - 【反重复】严禁在不同回合反复使用相同或极度相似的动作描写和身体反应描写（如“喉结上下一滚”“握紧了拳头”“深吸一口气”等）。每段描写中的动作细节必须是全新的，绝不能与之前出现过的动作描写雷同。如果你不确定是否用过某个表达，就换一种完全不同的描写方式。 - 如果大纲中涉及 NPC，要呈现 NPC 的性格特点 - NPC 姓名规则：当某个 NPC 在本次冒险中第一次出场时，必须先通过外貌、职业等特征描写他（如“铁匠铺里一个肩宽体壮的矮人”），然后通过自然的方式引出名字（如自我介绍、别人称呼、招牌上写着等）。绝对不要在没有铺垫的情况下直接使用名字。 - 如果是战斗场景，描写要紧张刺激 - 游戏内的货币统一称为"金币"，不要使用"银币""铜币""银两"等其他名称 2. 在文本之后，生成恰好 3 个后续选项供玩家选择。选项应： - 覆盖不同策略方向（如：激进/谨慎/探索/社交 等） - 每个选项都可能导向不同的剧情分支 - 简洁明了，让玩家一眼就能理解 - 【关键约束】选项中提及的所有人物、物品和地点，必须已经在本回合的剧情文本中出现过，或在之前的冒险历史中铺垫过。绝对禁止在选项中凭空引入未经铺垫的新 NPC、新地点或新物品。 - 【关键约束】如果本回合玩家获得了新物品（武器、食物、药水、装备等），至少一个选项应涉及使用、装备或查看该物品。例如：获得武器→选项之一为"装备XX"；获得食物→选项之一为"食用XX恢复体力"。 - 【关键约束】选项中如果涉及使用某个物品，该物品必须当前在玩家背包中。不要生成使用玩家不拥有的物品的选项。请仔细检查上方的背包列表，确保选项中提及的物品确实在背包中。 请严格按以下格式输出（先是剧情文本，然后是分隔符，最后是JSON选项）： ---STORY_TEXT--- （在这里写剧情描写文本） ---OPTIONS_JSON--- [ {{"id": 1, "text": "选项描述", "action_type": "动作类型（如ATTACK/TALK/MOVE/EXPLORE/USE_ITEM等）"}}, {{"id": 2, "text": "选项描述", "action_type": "动作类型"}}, {{"id": 3, "text": "选项描述", "action_type": "动作类型"}} ] """ # ------------------------------------------------------------ # 死亡结局 Prompt # ------------------------------------------------------------ DEATH_NARRATIVE_PROMPT = """你是一个才华横溢的奇幻小说家。玩家的角色已经死亡，请为他写一段庄严、有诗意的死亡结局描写。 {death_context} {world_state} 请写一段 150-250 字的死亡结局描写（使用第二人称"你"叙述），包含： 1. 角色倒下的场景描写 2. 回顾此次冒险的点滴 3. 一句意味深长的结语 然后生成 2 个选项： ---STORY_TEXT--- （死亡结局文本） ---OPTIONS_JSON--- [ {{"id": 1, "text": "重新开始冒险", "action_type": "RESTART"}}, {{"id": 2, "text": "接受命运，结束游戏", "action_type": "QUIT"}} ] """ # ------------------------------------------------------------ # 开场叙事 Prompt # ------------------------------------------------------------ OPENING_NARRATIVE_PROMPT = """【最高优先级指令 ── 输出格式】 你的输出必须严格遵守以下格式，这是最高优先级的要求，违反将导致系统崩溃： 1. 先输出 ---STORY_TEXT--- 标记（独占一行） 2. 然后直接写故事内容（200-400字），不加任何前缀（禁止“好的”“以下是”等） 3. 故事写完后输出 ---OPTIONS_JSON--- 标记（独占一行） 4. 最后输出 JSON 格式的 3 个选项数组 绝对不要省略这两个标记。 你是一个经验丰富的奇幻小说家，正在为一个互动 RPG 游戏编写开场白。 {world_state} 【你的任务】 写一段游戏开场叙事（200-400字，中文，第二人称“你”叙述）： 1. 介绍主角（{player_name}）抵达这个村庄的背景 2. 用具体的场景和细节描写村庄（而不是笼统的氛围形容词） 3. 暗示冒险即将开始（可以提到村民们的忧虑、远处森林的阴影等） 4. 以一种引人入胜的方式结束，激发玩家的探索欲望 文风要求：简洁自然，像写小说而不是写诗歌。少用比喻、少用形容词，多用具体的动作和场景细节。禁止使用“阳光洒下”“微风拂过”“空气中弥漫着”等老套表达。 然后生成 3 个初始选项： ---STORY_TEXT--- （开场叙事文本） ---OPTIONS_JSON--- [ {{"id": 1, "text": "前往村庄广场，与村长交谈", "action_type": "TALK"}}, {{"id": 2, "text": "先去旅店休息一下，顺便打听消息", "action_type": "MOVE"}}, {{"id": 3, "text": "在村子里四处走走，观察环境", "action_type": "EXPLORE"}} ] """ # ------------------------------------------------------------ # 合并式 Prompt：一次调用完成大纲 + 叙事 + 选项 # # 设计思路： # - 将两阶段生成合并为一次 API 调用，减少一半延迟 # - THINKING 区域供模型内部规划（不展示给玩家），确保一致性 # - STORY_TEXT 区域放在中间，支持流式输出（用户最先看到文字） # - STATE_JSON 和 OPTIONS_JSON 放在末尾，供程序解析 # ------------------------------------------------------------ MERGED_SYSTEM_PROMPT_TEMPLATE = """【最高优先级指令 ── 输出格式】 你的输出必须严格遵守以下分隔标记格式。这是最高优先级的要求，违反将导致系统崩溃： - ---THINKING--- （规划区域） - ---STORY_TEXT--- （故事文本，200-400字，不加任何前缀） - ---STATE_JSON--- （状态变更 JSON） - ---OPTIONS_JSON--- （3 个选项的 JSON 数组） 每个区域必须以对应的 --- 标记开头（独占一行）。绝对不要省略任何标记。故事文本区域只放纯叙事文字，禁止混入 JSON。 你是一个专业的 RPG 叙事引擎，兼具剧情规划与文学描写能力。 {world_state} 【你的任务】 根据玩家的行动意图，在一次输出中完成以下所有工作： 1. 在 THINKING 标签内进行简短的剧情规划（不展示给玩家） 2. 写一段 200-400 字的剧情描写 3. 输出结构化的状态变更 JSON 4. 生成 3 个后续选项 【状态变更规则】 - 只填写确实发生变化的字段，未变化的设为 null/0/空或省略 - 所有数值变更必须精确，禁止模糊描述如"恢复一些" - time_change 仅允许："清晨""上午""正午""下午""黄昏""夜晚""深夜"，无跳跃则 null - 货币统一为"金币"（gold_change），严禁使用"铜币""银币""银两"等。任何钱财收获只通过 gold_change 表达，严禁将钱币放入 items_gained - 装备规则：装备物品时用 equip 字段指定槽位和物品名（如 "weapon": "小刀"），系统自动处理背包⇌装备栏转移。装备时不要在 items_lost/items_gained 重复处理该物品。合法槽位：weapon/armor/accessory/helmet/boots。卸下装备设槽位为 null - status_effects_added 中每个效果须含：name/description/effect_type/stat_modifiers/duration - 数值不超合法范围（HP∈[0,max_hp]，饱食度/士气/理智∈[0,100]） - new_location/weather_change/title_change/world_event 仅真正变化时填写，否则 null - quest_updates.status 只允许 "active"/"completed"/"failed" - 消耗品（药水、食物等一次性物品）使用后放入 items_lost；非消耗品（武器、工具等）使用后不放入 items_lost - 仅重要事件才添加状态效果，普通日常行为不产生 - npc_changes 格式: {{"NPC名称": {{"attitude": "新态度", "relationship_change": 数值, "memory_add": "新记忆", "hp_change": 数值}}}} - quest_updates 格式: {{"任务ID": {{"objectives_completed": ["完成的目标"], "status": "新状态"}}}} 【叙事写作规则】 - 使用第二人称"你"叙述 - 文本开头必须直接回应玩家的行动/选择 - 文风简洁自然有画面感，像写小说而非诗歌 - 多用具体动作和对话，少用比喻修辞 - 禁用："阳光洒下""微风拂过""空气中弥漫着""如同XX般" - 严禁重复相似的动作描写（如"握紧拳头""深吸一口气"等不可每回合重复） - NPC 首次出场需先描写外貌/特征，再自然引出名字 - 战斗场景要紧张刺激 - 货币统一称"金币" 【剧情多样性规则 ── 极其重要】 - 禁止连续两个回合出现相同类型的敌人（如上回合打了哥布林，本回合不能再打哥布林） - 连续战斗不得超过 2 个回合，之后必须穿插非战斗事件（对话/探索/发现/交易） - 鼓励引入新NPC、新线索、支线事件，让世界有"活"的感觉 - 如果世界状态中标注了【本回合随机事件】，必须将其融入叙事 - 尽量引导玩家前往不同的区域探索，而非反复停留在同一地点 【选项规则】 - 恰好 3 个选项，覆盖不同策略方向（激进/谨慎/探索/社交等） - 选项中的人物/物品/地点必须已在当前或之前剧情中出现过 - 获得新物品时至少一个选项涉及使用该物品 - 使用物品的选项中该物品必须在背包中 【输出格式（严格遵守，按以下顺序输出）】 ---THINKING--- （简短剧情规划：事件类型、涉及NPC、关键状态变更概要，3-5句话即可） ---STORY_TEXT--- （200-400 字剧情描写） ---STATE_JSON--- {{ "event_summary": "一句话描述", "event_type": "COMBAT/DIALOGUE/MOVE/ITEM/QUEST/TRADE/REST/DISCOVERY", "involved_npcs": [], "location": "玩家在本回合行动后的最终地点名称（若未移动，则为当前地点）", "state_changes": {{ "hp_change": 0, "mp_change": 0, "gold_change": 0, "exp_change": 0, "morale_change": 0, "sanity_change": 0, "hunger_change": 0, "karma_change": 0, "new_location": null, "items_gained": [], "items_lost": [], "skills_gained": [], "status_effects_added": [], "status_effects_removed": [], "npc_changes": {{}}, "quest_updates": {{}}, "weather_change": null, "time_change": null, "global_flags_set": {{}}, "world_event": null, "equip": {{}}, "title_change": null }}, "consequence_tags": [], "is_reversible": true }} ---OPTIONS_JSON--- [ {{"id": 1, "text": "选项描述", "action_type": "动作类型（ATTACK/TALK/MOVE/EXPLORE/USE_ITEM等）"}}, {{"id": 2, "text": "选项描述", "action_type": "动作类型"}}, {{"id": 3, "text": "选项描述", "action_type": "动作类型"}} ] """ class StoryEngine: """ 叙事引擎 —— 负责故事内容的生成 核心工作流程： 1. 接收玩家意图（来自 nlu_engine 解析） 2. 第一阶段：调用 Qwen 生成剧情大纲 (JSON) 3. 一致性检查（通过 state_manager） 4. 第二阶段：调用 Qwen 基于大纲生成文学文本 + 选项 5. 返回完整结果（文本 + 选项 + 状态变更） """ def __init__( self, game_state: GameState, model: str = DEFAULT_MODEL, *, enable_rule_text_polish: bool = False, ): self.game_state = game_state self.model = model self.enable_rule_text_polish = enable_rule_text_polish def _looks_like_overnight_request(self, player_intent: dict[str, Any]) -> bool: intent_name = str(player_intent.get("intent", "")).upper() if intent_name == "OVERNIGHT_REST": return True combined_text = f"{player_intent.get('details', '')} {player_intent.get('raw_input', '')}" return intent_name == "REST" and any( keyword in combined_text for keyword in ("过夜", "睡一晚", "住一晚", "睡到天亮") ) def _build_rule_text_polish_prompt(self, base_text: str, *, context_tag: str) -> str: return ( "你是 RPG 游戏的文本润色助手。你的任务是在不改变事实的前提下，" "把给定的基准文本扩写得更自然、更有画面感。\n\n" "【硬性约束】\n" "1. 不要创造新内容，不要新增人物、地点、物品、任务、敌人、线索或背景设定。\n" "2. 不要改变当前文本中的结果，不要改变战斗胜负、任务状态、奖励、数值变化、物品得失或剧情结论。\n" "3. 只需要做文本的扩写和润色，不要把它改写成新的剧情。\n" "4. 只能使用基准文本已经明确给出的信息，最多补充动作、感官、语气和环境细节。\n" "5. 输出只包含润色后的最终文本，不要解释，不要加标题。\n\n" f"【场景标签】{context_tag}\n" f"【基准文本】\n{base_text}" ) def _polish_rule_text(self, base_text: str, *, context_tag: str) -> str: normalized_base_text = str(base_text or "").strip() if not normalized_base_text or not self.enable_rule_text_polish: return normalized_base_text prompt = self._build_rule_text_polish_prompt( normalized_base_text, context_tag=context_tag, ) messages = [ {"role": "system", "content": prompt}, {"role": "user", "content": "请只输出润色后的最终文本。"}, ] try: polished_text = str( call_qwen( messages, model=self.model, temperature=0.3, max_tokens=600, ) ).strip() except Exception as exc: logger.warning("规则文本润色失败，回退到基准文本: %s", exc) return normalized_base_text if not polished_text: return normalized_base_text if "---STORY_TEXT---" in polished_text: polished_text = polished_text.split("---STORY_TEXT---", 1)[1].strip() if "---OPTIONS_JSON---" in polished_text: polished_text = polished_text.split("---OPTIONS_JSON---", 1)[0].strip() if len(polished_text) < max(10, len(normalized_base_text) // 2): return normalized_base_text return polished_text def _build_environment_impact_note(self, tick_log: list[str]) -> str: if not tick_log: return "" notes: list[str] = [] if any("天气变为:" in line for line in tick_log): weather_note = { "小雨": "细雨很快打湿了衣角和地面，你每一步都得更留心脚下。", "浓雾": "雾气一压下来，周围的轮廓立刻变得模糊，连远处的动静都难辨方向。", "暴风雨": "风雨压得人几乎睁不开眼，呼吸和判断都跟着沉了下来。", "晴朗": "天色放亮以后，胸口那股压抑感也跟着松了些。", "多云": "云层压住了天光，四周看着比先前更沉静了一点。", "大雪": "雪意裹着寒气往身上钻，连指尖都变得迟钝了些。", }.get(self.game_state.world.weather, "") if weather_note: notes.append(weather_note) if any("光照变化:" in line for line in tick_log): if self.game_state.world.light_level in {"昏暗", "幽暗", "漆黑"}: notes.append("周围的光线明显暗了下去，你不得不把注意力更多放在近处。") elif self.game_state.world.light_level in {"明亮", "柔和"}: notes.append("视野重新舒展开来，紧绷的神经也稍稍缓了一口气。") return "".join(dict.fromkeys(notes)) @staticmethod def _combat_advantage_label(power_gap: int) -> str: if power_gap >= 8: return "胜算较高" if power_gap >= 2: return "略占上风" if power_gap >= -2: return "势均力敌" if power_gap >= -8: return "胜算偏低" return "胜算很低" def _build_combat_preview_note(self, options: list[dict[str, Any]]) -> str: attack_targets: list[str] = [] for option in options: if str(option.get("action_type", "")).upper() != "ATTACK": continue target = option.get("target") if not isinstance(target, str) or not target.strip(): continue if target not in attack_targets: attack_targets.append(target) if not attack_targets: return "" self.game_state.refresh_combat_stats() player = self.game_state.player player_power = int(player.attack_power) + int(player.level) * 2 player_defense = int(player.defense_power) lines = ["【战斗评估】"] for monster_name in attack_targets[:2]: monster = get_monster_profile(monster_name, game_state=self.game_state) monster_power = int(monster["defense"]) + int(monster["difficulty"]) * 3 power_gap = player_power - monster_power advantage = self._combat_advantage_label(power_gap) loss_low = max(1, int(monster["attack"]) - player_defense - 3) loss_high = max(1, int(monster["attack"]) - player_defense) lose_warning = "" if power_gap < 0: lose_warning = "；若判定失败，实际掉血会更高" lines.append( f"- {monster_name}｜HP {monster['hp']} 攻击 {monster['attack']} 防御 {monster['defense']} 难度 {monster['difficulty']}｜" f"你的战力 {player_power} vs 对方战力 {monster_power}（{advantage}），" f"基础承伤预估 {loss_low}~{loss_high}{lose_warning}。" ) return "\n" + "\n".join(lines) def _build_rule_final_result( self, *, story_text: str, options: list[dict], tick_log: list[str], change_log: list[str], state_changes: dict[str, Any] | None = None, engine_mode: str = "deterministic_rule", strict_options: bool = False, minimum_options: int | None = None, ) -> dict[str, Any]: validated_options = self._validate_options( options, enrich=not strict_options, minimum=( DEFAULT_OPTION_COUNT if minimum_options is None and not strict_options else int(minimum_options or 0) ), maximum=MAX_OPTION_COUNT, ) environment_note = self._build_environment_impact_note(tick_log) combined_story_text = story_text if environment_note and environment_note not in combined_story_text: combined_story_text = f"{combined_story_text}{environment_note}" combat_preview_note = "" if "rule_attack" not in engine_mode: combat_preview_note = self._build_combat_preview_note(validated_options) final_story_text = self._polish_rule_text( combined_story_text, context_tag=engine_mode, ) if combat_preview_note and combat_preview_note not in final_story_text: final_story_text = f"{final_story_text}\n\n{combat_preview_note}" return { "story_text": final_story_text, "options": validated_options, "location": str(getattr(self.game_state, "current_location", None) or self.game_state.player.location or "未知之地"), "state_changes": state_changes or {}, "change_log": _merge_change_logs(tick_log, change_log), "outline": None, "consistency_issues": [], "telemetry": _build_telemetry(engine_mode=engine_mode, used_fallback=False), } def _current_map_name(self) -> str | None: inventory = list(self.game_state.player.inventory) location_name = str(self.game_state.player.location) preferred_map = LOCATION_MAP_REQUIREMENTS.get(location_name) if preferred_map and preferred_map in inventory: return preferred_map for item_name in inventory: if "地图" in str(item_name): return str(item_name) return None def _main_quest(self): return self.game_state.world.quests.get(MAIN_QUEST_ID) def _build_quest_reward_changes(self, quest_id: str) -> dict[str, Any]: quest = self.game_state.world.quests.get(str(quest_id)) if quest is None: return {} reward_changes: dict[str, Any] = {} if int(quest.rewards.gold or 0): reward_changes["gold_change"] = int(quest.rewards.gold) if int(quest.rewards.experience or 0): reward_changes["exp_change"] = int(quest.rewards.experience) if int(quest.rewards.karma_change or 0): reward_changes["karma_change"] = int(quest.rewards.karma_change) owned_items = set(self.game_state.player.inventory) | { str(item_name) for item_name in self.game_state.player.equipment.values() if item_name } reward_items = [ str(item_name) for item_name in quest.rewards.items if str(item_name) not in owned_items ] if reward_items: reward_changes["items_gained"] = reward_items return reward_changes def _build_main_quest_reward_changes(self) -> dict[str, Any]: return self._build_quest_reward_changes(MAIN_QUEST_ID) def _discover_locations(self, *location_names: str) -> None: for location_name in location_names: if location_name not in self.game_state.world.locations: continue location = self.game_state.world.locations[location_name] location.is_discovered = True if location_name not in self.game_state.world.discovered_locations: self.game_state.world.discovered_locations.append(location_name) def _relocate_npc(self, npc_name: str, new_location: str) -> None: npc = self.game_state.world.npcs.get(npc_name) if npc is None or new_location not in self.game_state.world.locations: return old_location = npc.location if old_location == new_location: return if old_location in self.game_state.world.locations: old_scene = self.game_state.world.locations[old_location] if npc_name in old_scene.npcs_present: old_scene.npcs_present.remove(npc_name) npc.location = new_location new_scene = self.game_state.world.locations[new_location] if npc_name not in new_scene.npcs_present: new_scene.npcs_present.append(npc_name) def _build_location_flavor_text(self, location_name: str) -> str: location = self.game_state.world.locations.get(str(location_name)) if location is None: return f"你来到{location_name}，下意识放慢脚步，先确认四周有没有新的动静。" description = str(location.description or "").strip() ambient = str(location.ambient_description or "").strip() lead = f"你来到{location_name}。" if description: lead = f"{lead}{description}" if ambient: return f"{lead}\n{ambient}" return lead def _build_trade_failure_text( self, merchant_name: str, item_name: str, reason: str, price: int, ) -> str: merchant_lines = { ("铁匠格林", "insufficient_gold"): ( f"铁匠格林把{item_name}重新搁回铁毡旁，粗着嗓子说道：" f"“这把家伙值 {price} 金币。钱袋没装满之前，别急着碰它。”" ), ("杂货商人阿尔", "insufficient_gold"): ( f"杂货商人阿尔推了推眼镜，把{item_name}往柜台里收了收：" f"“先把账算明白再说，{price} 金币，一枚都不能少。”" ), ("旅店老板娘莉娜", "insufficient_gold"): ( f"莉娜朝你摊开手，无奈地笑了笑：" f"“我也想帮你，可这份{item_name}还是得收 {price} 金币。”" ), } if (merchant_name, reason) in merchant_lines: return merchant_lines[(merchant_name, reason)] reason_text = { "merchant_not_here": f"{merchant_name}现在不在你面前，这笔买卖暂时谈不成。", "item_not_sold_here": f"{merchant_name}摇了摇头，表示这里并不卖{item_name}。", "invalid_merchant": "你面前没有能完成这笔交易的人。", "unknown_item": f"{merchant_name}听完后皱了皱眉，像是没听说过{item_name}。", } return reason_text.get(reason, f"{merchant_name}摇了摇头，示意这笔交易暂时做不成。") def _build_battle_story_text( self, *, enemy_name: str, location_name: str, outcome: str, low_stamina: bool, reward_items: list[str], ) -> str: exertion = "你几乎是咬着牙硬撑，挥出的每一下都格外吃力。" if low_stamina else "" reward_text = "" if reward_items: reward_text = f"你顺手从{enemy_name}留下的东西里翻出了{reward_items[0]}。" if enemy_name == "森林巨魔": if outcome == "forced_retreat": return ( f"{exertion} 森林巨魔抡起巨臂砸碎你脚边的树根，震得你胸口发闷。" "你被迫借着林间地形连连后撤，才勉强从它的追击范围里脱身。" ).strip() if outcome == "pyrrhic_win": return ( f"{exertion} 你几乎是踩着最后一口气把兵刃送进森林巨魔的要害，" "它发出一声撕裂林海的哀嚎后轰然倒地。巨魔身后的巢穴深处隐约闪着大量金币的反光，像是堆着它掠来的财物。" ).strip() if outcome == "normal_win": return ( f"{exertion} 你稳住节奏避开森林巨魔最沉重的扑砸，趁它露出破绽时连斩数剑，" "终于将这头庞然大物逼得跪倒在地。随着巨魔轰然倒下，后方巢穴里大量金币的反光也显露出来。" ).strip() return ( f"{exertion} 你抢在森林巨魔完全站稳前便连续压制，剑锋数次精准斩进它旧伤最深处。" "巨魔连最后的反扑都没撑起来便倒在你脚下，巢穴深处散落的大量金币也随之暴露在火光里。" ).strip() if outcome == "forced_retreat": return ( f"{exertion} {enemy_name}在{location_name}的阴影里逼得太紧，你只能狼狈后撤，" "勉强把局面从彻底失控的边缘拉回来。" ).strip() if outcome == "pyrrhic_win": return ( f"{exertion} 你硬是在近身缠斗里压住了{enemy_name}，" f"直到对方终于翻倒在地。{reward_text}" ).strip() if outcome == "normal_win": return ( f"{exertion} {enemy_name}扑上来的瞬间，你稳住节奏迎了上去，" f"几下交手后便把它压倒在地。{reward_text}" ).strip() return ( f"{exertion} 你抢在{enemy_name}反应过来前便彻底占了上风，" f"对方几乎没有组织起像样的反击。{reward_text}" ).strip() def _build_combat_narration_prompt( self, *, monster_name: str, location_name: str, outcome: str, player_hp_loss: int, reward_items: list[str], ) -> str: reward_text = "无" if reward_items: reward_text = "、".join(str(item) for item in reward_items) return ( "你是 RPG 游戏战斗叙事模块。\n" f"当前事件：玩家攻击了{monster_name}。系统判定结果：{outcome}。" f"玩家损失生命值：{player_hp_loss}。请根据这个结果生动地描写战斗过程，严禁更改胜负结果。\n\n" "【硬性约束】\n" "1. 严禁改写系统判定结果，禁止把失败写成胜利或把胜利写成失败。\n" "2. 严禁修改玩家损失生命值，文本里出现的扣血信息必须是给定数值。\n" "3. 不要新增不存在的角色、地点或奖励。\n" "4. 只输出最终叙事正文，不要加标题，不要解释。\n\n" f"【战斗地点】{location_name}\n" f"【怪物】{monster_name}\n" f"【系统结果】{outcome}\n" f"【玩家生命损失】{player_hp_loss}\n" f"【本次战斗掉落】{reward_text}\n" ) def _build_combat_fallback_story( self, *, monster_name: str, location_name: str, outcome: str, player_hp_loss: int, reward_items: list[str], ) -> str: reward_text = "" if reward_items: reward_text = f" 你从{monster_name}身上取得了{reward_items[0]}。" if outcome == "win": return ( f"你在{location_name}与{monster_name}展开短促而激烈的交锋，最终稳住节奏将其击败。" f"这场战斗让你损失了{player_hp_loss}点生命值。{reward_text}" ).strip() return ( f"你在{location_name}与{monster_name}硬拼一轮后落入下风，被迫后撤。" f"这一战让你损失了{player_hp_loss}点生命值。" ) def _generate_combat_narration( self, *, monster_name: str, location_name: str, outcome: str, player_hp_loss: int, reward_items: list[str], ) -> str: fallback_story = self._build_combat_fallback_story( monster_name=monster_name, location_name=location_name, outcome=outcome, player_hp_loss=player_hp_loss, reward_items=reward_items, ) prompt = self._build_combat_narration_prompt( monster_name=monster_name, location_name=location_name, outcome=outcome, player_hp_loss=player_hp_loss, reward_items=reward_items, ) messages = [ {"role": "system", "content": prompt}, {"role": "user", "content": "请直接输出战斗叙事正文。"}, ] try: generated = str( call_qwen( messages, model=self.model, temperature=0.4, max_tokens=500, ) ).strip() except Exception as exc: logger.warning("战斗叙事生成失败，使用模板文本: %s", exc) return fallback_story if not generated: return fallback_story return generated def _quest_updates_matching(self, matcher) -> dict[str, dict[str, list[str]]]: updates: dict[str, dict[str, list[str]]] = {} for quest_id, quest in self.game_state.world.quests.items(): if quest.status != "active": continue completed_objectives = [ objective for objective, completed in quest.objectives.items() if not completed and matcher(objective) ] if completed_objectives: updates[quest_id] = {"objectives_completed": completed_objectives} return updates def _activate_quest_objective( self, state_changes: dict[str, Any], quest_id: str, objective_name: str, ) -> bool: quest = self.game_state.world.quests.get(quest_id) if quest is None or quest.status == "completed": return False quest_updates = state_changes.setdefault("quest_updates", {}) quest_update = dict(quest_updates.get(quest_id, {})) changed = False if quest.status != "active" and quest_update.get("status") != "active": quest_update["status"] = "active" changed = True if objective_name in quest.objectives and not quest.objectives.get(objective_name, False): completed_objectives = list(quest_update.get("objectives_completed", [])) if objective_name not in completed_objectives: completed_objectives.append(objective_name) quest_update["objectives_completed"] = completed_objectives changed = True if changed: quest_updates[quest_id] = quest_update elif not quest_updates: state_changes.pop("quest_updates", None) return changed @staticmethod def _merge_reward_changes(state_changes: dict[str, Any], reward_changes: dict[str, Any]) -> None: if reward_changes.get("gold_change"): state_changes["gold_change"] = state_changes.get("gold_change", 0) + int( reward_changes["gold_change"] ) if reward_changes.get("exp_change"): state_changes["exp_change"] = state_changes.get("exp_change", 0) + int( reward_changes["exp_change"] ) if reward_changes.get("karma_change"): state_changes["karma_change"] = state_changes.get("karma_change", 0) + int( reward_changes["karma_change"] ) for reward_item in reward_changes.get("items_gained", []): state_changes.setdefault("items_gained", []).append(reward_item) def _build_route_steps(self, destination: str) -> list[str]: destination = str(destination or "") if not destination or destination == self.game_state.player.location: return [] start = str(self.game_state.player.location) visited = {start} queue: list[tuple[str, list[str]]] = [(start, [])] while queue: current, path = queue.pop(0) location = self.game_state.world.locations.get(current) if location is None: continue for neighbor in location.connected_to: if neighbor in visited: continue visited.add(neighbor) new_path = path + [neighbor] if neighbor == destination: return new_path queue.append((neighbor, new_path)) return [] def _maybe_resolve_guided_deep_forest_travel( self, player_intent: dict, destination: str, ) -> dict[str, Any] | None: if destination != "森林深处": return None main_quest = self.game_state.world.quests.get(MAIN_QUEST_TROLL_ID) if ( main_quest is None or main_quest.status != "active" or "森林之钥" not in self.game_state.player.inventory or self.game_state.player.location == "森林深处" ): return None route_steps = self._build_route_steps(destination) if not route_steps: return None tick_log: list[str] = [] change_log: list[str] = [] final_state_changes: dict[str, Any] = {} for step in route_steps: tick_log.extend(self.game_state.tick_time(player_intent)) step_changes: dict[str, Any] = {"new_location": step} if step == "森林深处": quest_updates = self._quest_updates_matching(lambda objective: step in objective) if quest_updates: step_changes["quest_updates"] = quest_updates step_change_log = self.game_state.apply_changes(step_changes) change_log.extend(step_change_log) final_state_changes = step_changes if step == "黑暗森林入口" and random.random() < 0.25: self.game_state.last_interacted_npc = None options = [ { "id": 1, "text": "击退拦路野狼", "action_type": "ATTACK", "target": "野狼", "priority": 120, }, { "id": 2, "text": "退回村口小路整顿", "action_type": "MOVE", "target": "村口小路", "priority": 112, }, ] if "黑暗森林地图" in self.game_state.player.inventory: options.append( { "id": 3, "text": "查看地图", "action_type": "VIEW_MAP", "target": None, "priority": 108, } ) story_text = ( "你沿着村口小路一路赶赴森林深处，穿过黑暗森林入口时，右侧灌木忽然被猛地撞开。\n" "一头被血腥味刺激得双眼发红的野狼伏低身子拦在路中，獠牙间不断溢出低吼，显然把你当成了新的猎物。" ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=final_state_changes, engine_mode="rule_move_guided_deep_forest_ambush", strict_options=True, ) self.game_state.last_interacted_npc = None forest_troll_intro = ( main_quest.status == "active" and not self.game_state.world.global_flags.get(FOREST_TROLL_DEFEATED_FLAG) and not self.game_state.world.global_flags.get(FOREST_TROLL_INTRO_SEEN_FLAG) ) if forest_troll_intro: intro_changes = { "global_flags_set": { FOREST_TROLL_INTRO_SEEN_FLAG: True, "arrival::deep_forest": True, } } intro_change_log = self.game_state.apply_changes(intro_changes) change_log.extend(intro_change_log) final_state_changes.setdefault("global_flags_set", {}).update(intro_changes["global_flags_set"]) story_text = ( "你穿过黑暗森林外层一路深入，越往前走，树根间残留的重压痕迹就越发清晰。\n" "等你拨开垂落的藤蔓，前方那株断裂古树后缓缓站起一道庞大黑影，浑身裹着泥苔与旧伤。\n" "森林巨魔终于现身了，它低吼着拍碎身旁的树根，显然不打算给你任何后退的余地。" ) return self._build_rule_final_result( story_text=story_text, options=build_scene_actions(self.game_state, destination), tick_log=tick_log, change_log=change_log, state_changes=final_state_changes, engine_mode="rule_move_guided_deep_forest_intro", strict_options=True, ) story_text = ( "你沿着已经摸熟的林间路线重新赶到森林深处，潮湿空气里依旧弥漫着腐朽与血腥混杂的气味。\n" "远处枝干不时传来沉闷震响，说明那头庞然巨物还没有离开这片区域。" ) return self._build_rule_final_result( story_text=story_text, options=build_scene_actions(self.game_state, destination), tick_log=tick_log, change_log=change_log, state_changes=final_state_changes, engine_mode="rule_move_guided_deep_forest", strict_options=True, ) def _objective_requires_npc(self, npc_name: str) -> bool: for quest in self.game_state.world.quests.values(): if quest.status != "active": continue for objective, completed in quest.objectives.items(): if not completed and npc_name in objective: return True return False def _extract_option_npc(self, text: str) -> str | None: mentioned = [ npc_name for npc_name in self.game_state.world.npcs.keys() if npc_name in text ] if len(mentioned) == 1: return mentioned[0] return None def _resolve_option_location(self, option: dict) -> str | None: target = option.get("target") if isinstance(target, dict): target = target.get("location") or target.get("target") if isinstance(target, str) and target in self.game_state.world.locations: return target text = str(option.get("text", "")) mentioned = [ location_name for location_name in self.game_state.world.locations.keys() if location_name in text ] if len(mentioned) == 1: return mentioned[0] return None @staticmethod def _looks_like_travel_text(text: str) -> bool: travel_markers = ( "前往", "去往", "赶往", "出发", "移动", "沿着", "深入", "进入", "走向", "走进", ) return any(marker in text for marker in travel_markers) def _normalize_option(self, option: dict) -> dict[str, Any] | None: if not isinstance(option, dict): return None text = str(option.get("text", "")).strip() if not text: return None normalized: dict[str, Any] = dict(option) action_type = str(normalized.get("action_type", "EXPLORE") or "EXPLORE").upper() normalized["action_type"] = action_type normalized["text"] = text current_location = self.game_state.world.locations.get(self.game_state.player.location) npc_name = self._extract_option_npc(text) move_target = self._resolve_option_location(normalized) travel_like = self._looks_like_travel_text(text) if npc_name: npc = self.game_state.world.npcs.get(npc_name) if npc and npc.location == self.game_state.player.location: normalized["action_type"] = "TALK" normalized["target"] = npc_name normalized["text"] = f"与{npc_name}对话" if normalized["action_type"] == "TALK": talk_target = normalized.get("target") if not isinstance(talk_target, str) or talk_target not in self.game_state.world.npcs: talk_target = npc_name if not talk_target: return None npc = self.game_state.world.npcs.get(str(talk_target)) if npc is None or npc.location != self.game_state.player.location: return None if ( str(talk_target) == self.game_state.last_interacted_npc and not self._objective_requires_npc(str(talk_target)) ): return None normalized["target"] = str(talk_target) normalized["text"] = f"与{talk_target}对话" return normalized if normalized["action_type"] == "MOVE" or travel_like: if move_target is None: return None target_location = self.game_state.world.locations.get(move_target) if current_location is None or target_location is None: return None if move_target not in current_location.connected_to: return None if ( not target_location.is_accessible and target_location.required_item and target_location.required_item not in self.game_state.player.inventory ): return None preserve_move_text = bool(normalized.get("preserve_text")) normalized["action_type"] = "MOVE" normalized["target"] = move_target normalized["text"] = text if preserve_move_text else f"前往{move_target}" return normalized if normalized["action_type"] == "EXPLORE" and travel_like and move_target is None: return None return normalized def _rule_trade_response(self, player_intent: dict) -> dict[str, Any] | None: target = player_intent.get("target") if not isinstance(target, dict): return None merchant_name = str(target.get("merchant") or "") item_name = str(target.get("item") or target.get("item_name") or "") confirm = bool(target.get("confirm")) if not merchant_name or not item_name: return None tick_log = self.game_state.tick_time(player_intent) before_gold = self.game_state.player.gold trade_result = resolve_trade( self.game_state, merchant_name=merchant_name, item_name=item_name, confirm=confirm, ) self.game_state.last_interacted_npc = merchant_name if trade_result.get("reason") == "awaiting_confirmation": price = int(trade_result.get("price", 0)) story_text = f"{merchant_name}拿出{item_name}，报价 {price} 金币，等待你的最终确认。" options = [ { "id": 1, "text": f"确认花费{price}金币购买{item_name}", "action_type": "TRADE", "target": {"merchant": merchant_name, "item": item_name, "confirm": True}, "priority": 120, }, { "id": 2, "text": "先回到商品列表", "action_type": "SHOP_MENU", "target": merchant_name, "priority": 90, }, ] return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=[], engine_mode="rule_trade_quote", strict_options=True, ) if trade_result.get("applied"): price = int(trade_result.get("price", 0)) change_log = [ f"金币: {before_gold} → {self.game_state.player.gold}", f"获得物品: {item_name}", ] story_text = f"你向{merchant_name}支付了 {price} 金币，稳稳地把{item_name}收进背包。" post_purchase_options = [ { "text": "继续看看还有什么可买", "action_type": "SHOP_MENU", "target": merchant_name, "priority": 130, }, { "text": "暂时离开柜台", "action_type": "SCENE_OPTIONS", "target": self.game_state.player.location, "priority": 88, }, ] options = merge_demo_options( post_purchase_options, trade_result.get("follow_up_actions", []), build_contextual_actions(self.game_state, recent_gain=item_name), build_goal_directed_actions(self.game_state), limit=MAX_OPTION_COUNT, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes={"gold_change": -price, "items_gained": [item_name]}, engine_mode="rule_trade_apply", strict_options=True, ) reason = str(trade_result.get("reason", "trade_failed")) price = int(trade_result.get("price", 0)) story_text = self._build_trade_failure_text( merchant_name, item_name, reason, price, ) return self._build_rule_final_result( story_text=story_text, options=[ { "text": "回到商品列表", "action_type": "SHOP_MENU", "target": merchant_name, "priority": 100, }, { "text": "暂不购买，先离开柜台", "action_type": "SCENE_OPTIONS", "target": self.game_state.player.location, "priority": 90, }, ], tick_log=tick_log, change_log=[], engine_mode="rule_trade_reject", strict_options=True, ) def _rule_equip_response(self, player_intent: dict) -> dict[str, Any] | None: item_name = player_intent.get("target") if not isinstance(item_name, str) or not item_name: return None item_info = self.game_state.world.item_registry.get(item_name) if item_info is None or item_name not in self.game_state.player.inventory: return None slot_map = { "weapon": "weapon", "armor": "armor", "accessory": "accessory", } slot_name = slot_map.get(item_info.item_type) if slot_name is None: return None tick_log = self.game_state.tick_time(player_intent) change_log = self.game_state.apply_changes({"equip": {slot_name: item_name}}) story_text = f"你迅速调整姿态，将{item_name}装备在身上，接下来的行动明显更有把握。" options = merge_demo_options( [], build_goal_directed_actions(self.game_state), build_contextual_actions(self.game_state), limit=3, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes={"equip": {slot_name: item_name}}, engine_mode="rule_equip", ) def _rule_shop_menu_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() != "SHOP_MENU": return None merchant_name = player_intent.get("target") if not isinstance(merchant_name, str) or not merchant_name: return None npc = self.game_state.world.npcs.get(merchant_name) if npc is None or npc.location != self.game_state.player.location: return None tick_log = self.game_state.tick_time(player_intent) self.game_state.last_interacted_npc = merchant_name story_text = f"{merchant_name}把货架重新朝你推近了一些，示意你慢慢挑选需要的物资。" return self._build_rule_final_result( story_text=story_text, options=build_shop_menu_actions(self.game_state, merchant_name), tick_log=tick_log, change_log=[], engine_mode="rule_shop_menu", strict_options=True, ) def _rule_scene_options_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() != "SCENE_OPTIONS": return None target = player_intent.get("target") if isinstance(target, dict): location_name = str(target.get("location") or self.game_state.player.location) else: location_name = str(target or self.game_state.player.location) if location_name != self.game_state.player.location: return None tick_log = self.game_state.tick_time(player_intent) self.game_state.last_interacted_npc = None story_text = "你暂时离开了柜台，把注意力重新放回眼前的场景与接下来的计划。" return self._build_rule_final_result( story_text=story_text, options=build_scene_actions(self.game_state, location_name), tick_log=tick_log, change_log=[], engine_mode="rule_scene_options", strict_options=True, ) def _extract_requested_shop_item(self, merchant_name: str, player_intent: dict[str, Any]) -> str | None: merchant = self.game_state.world.npcs.get(merchant_name) if merchant is None or not merchant.can_trade: return None text_blob = f"{player_intent.get('raw_input', '')} {player_intent.get('details', '')}".strip() if not text_blob: return None for item_name in merchant.shop_inventory: if str(item_name) in text_blob: return str(item_name) return None def _find_shop_merchant_at(self, location_name: str) -> str | None: for npc in self.game_state.world.npcs.values(): if npc.location == location_name and npc.can_trade: return npc.name return None def _maybe_chain_trade_after_move( self, *, player_intent: dict[str, Any], destination: str, tick_log: list[str], change_log: list[str], state_changes: dict[str, Any], ) -> dict[str, Any] | None: merchant_name = self._find_shop_merchant_at(destination) if not merchant_name: return None requested_item = self._extract_requested_shop_item(merchant_name, player_intent) if not requested_item: return None quote = resolve_trade( self.game_state, merchant_name=merchant_name, item_name=requested_item, confirm=False, ) if str(quote.get("reason")) != "awaiting_confirmation": return None price = int(quote.get("price", 0)) self.game_state.last_interacted_npc = merchant_name story_text = ( f"你赶到{destination}后，直接向{merchant_name}说明要买{requested_item}。" f"{merchant_name}把货物递到柜台前，报价 {price} 金币，等你最后确认。" ) options = [ { "id": 1, "text": f"确认花费{price}金币购买{requested_item}", "action_type": "TRADE", "target": {"merchant": merchant_name, "item": requested_item, "confirm": True}, "priority": 125, }, { "id": 2, "text": "先看看其他商品", "action_type": "SHOP_MENU", "target": merchant_name, "priority": 108, }, { "id": 3, "text": "暂时离开柜台", "action_type": "SCENE_OPTIONS", "target": destination, "priority": 92, }, ] return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_move_chain_trade_quote", strict_options=True, ) def _rule_move_response(self, player_intent: dict) -> dict[str, Any] | None: target = player_intent.get("target") if isinstance(target, dict): destination = str(target.get("location") or "") else: destination = str(target or "") if not destination: return None guided_travel_result = self._maybe_resolve_guided_deep_forest_travel( player_intent, destination, ) if guided_travel_result is not None: return guided_travel_result if destination == "森林深处" and self.game_state.player.location == "黑暗森林入口": if not self.game_state.world.global_flags.get(FOREST_GOBLIN_DEFEATED_FLAG): tick_log = self.game_state.tick_time(player_intent) story_text = ( "入口那只哥布林还在断木间虎视眈眈地盯着你。现在硬往深处闯，" "等于把后背和退路一起交给它。你得先把眼前这道麻烦解决。" ) options = merge_demo_options( build_scene_actions(self.game_state, "黑暗森林入口"), build_goal_directed_actions(self.game_state), limit=MAX_OPTION_COUNT, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=[], engine_mode="rule_move_deep_forest_enemy_block", strict_options=True, ) if not self.game_state.world.global_flags.get(FOREST_TROLL_TRACKS_FOUND_FLAG): tick_log = self.game_state.tick_time(player_intent) story_text = ( "你刚迈向更深的林荫，就又停住了脚步。倒地的哥布林脚边拖出一串凌乱泥痕，" "比起贸然深入，你更该先弄清它究竟是从什么东西面前逃出来的。" ) options = merge_demo_options( build_goal_directed_actions(self.game_state), build_scene_actions(self.game_state, "黑暗森林入口"), limit=MAX_OPTION_COUNT, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=[], engine_mode="rule_move_deep_forest_hold", strict_options=True, ) if "森林之钥" not in self.game_state.player.inventory: tick_log = self.game_state.tick_time(player_intent) state_changes = { "global_flags_set": { DEEP_FOREST_BARRIER_SEEN_FLAG: True, } } story_text = ( "你沿着巨大脚印试着往林中更深处逼近，没走多远，前方纠缠的古树根间便亮起一圈黯绿微光。" "那道封锁像是在拦着所有贸然闯入的人，你只能先退回入口，意识到若想继续追进去，" "必须先拿到能够开启道路的森林之钥。" ) change_log = self.game_state.apply_changes(state_changes) options = merge_demo_options( build_scene_actions(self.game_state, "黑暗森林入口"), build_goal_directed_actions(self.game_state), limit=MAX_OPTION_COUNT, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_move_deep_forest_blocked", strict_options=True, ) is_valid, rejection_reason = self.game_state.pre_validate_action( { "intent": "MOVE", "target": destination, "details": player_intent.get("details", ""), "raw_input": player_intent.get("raw_input", ""), } ) if not is_valid: return self._build_rule_final_result( story_text=f"你无法这样移动：{rejection_reason}", options=[], tick_log=[], change_log=[], engine_mode="rule_move_reject", ) tick_log = self.game_state.tick_time(player_intent) quest_updates = self._quest_updates_matching(lambda objective: destination in objective) state_changes: dict[str, Any] = {"new_location": destination} if quest_updates: state_changes["quest_updates"] = quest_updates traveler_intro = ( destination == "村口小路" and self.game_state.world.global_flags.get(TRAVELER_RUMOR_HEARD_FLAG) and not self.game_state.world.global_flags.get(TRAVELER_ENCOUNTERED_FLAG) ) guardian_intro = ( destination == "精灵遗迹" and not self.game_state.world.global_flags.get(GUARDIAN_INTRO_SEEN_FLAG) and self.game_state.world.npcs.get("遗迹守护者") is not None and self.game_state.world.npcs["遗迹守护者"].location == "精灵遗迹" ) forest_troll_intro = ( destination == "森林深处" and self.game_state.world.quests.get(MAIN_QUEST_TROLL_ID) is not None and self.game_state.world.quests[MAIN_QUEST_TROLL_ID].status == "active" and not self.game_state.world.global_flags.get(FOREST_TROLL_DEFEATED_FLAG) and not self.game_state.world.global_flags.get(FOREST_TROLL_INTRO_SEEN_FLAG) ) if traveler_intro: self._relocate_npc("神秘旅人", "村口小路") state_changes.setdefault("global_flags_set", {})[TRAVELER_ENCOUNTERED_FLAG] = True if guardian_intro: state_changes.setdefault("global_flags_set", {})[GUARDIAN_INTRO_SEEN_FLAG] = True state_changes.setdefault("global_flags_set", {})["arrival::elf_ruins"] = True if forest_troll_intro: state_changes.setdefault("global_flags_set", {})[FOREST_TROLL_INTRO_SEEN_FLAG] = True state_changes.setdefault("global_flags_set", {})["arrival::deep_forest"] = True change_log = self.game_state.apply_changes(state_changes) self.game_state.last_interacted_npc = None if traveler_intro: story_text = ( "你刚踏上村口小路，就看见路边枯树下立着一道裹得严严实实的身影。" "神秘旅人像是早就在等人，听见你的脚步后缓缓抬头，兜帽下的目光先审视了你一遍。" ) return self._build_rule_final_result( story_text=story_text, options=build_scene_actions(self.game_state, destination), tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_move_traveler_intro", strict_options=True, ) if guardian_intro: story_text = ( "石柱间的符文在你靠近时次第亮起，一道穿着褪色绿袍的身影随之从阴影里走出。" "遗迹守护者拦在前方，目光冷静而锐利，像是在判断你是否有资格继续踏入这片古老圣地。" ) return self._build_rule_final_result( story_text=story_text, options=build_scene_actions(self.game_state, destination), tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_move_guardian_intro", strict_options=True, ) if forest_troll_intro: story_text = ( "你拨开垂落的藤蔓继续深入，脚下的腐叶忽然被一阵沉重震颤掀得四散。" "前方一株断裂古树后缓缓站起一道巨影，浑身裹着泥苔与旧伤，獠牙间喷出的热气像闷雷般滚过林间。" "森林巨魔终于现身了，它低吼着拍碎身旁的树根，显然不打算给你任何后退的余地。" ) return self._build_rule_final_result( story_text=story_text, options=build_scene_actions(self.game_state, destination), tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_move_forest_troll_intro", strict_options=True, ) chained_trade = self._maybe_chain_trade_after_move( player_intent=player_intent, destination=destination, tick_log=tick_log, change_log=change_log, state_changes=state_changes, ) if chained_trade is not None: return chained_trade arrival_event = build_arrival_event(self.game_state, destination) if arrival_event and not self.game_state.world.global_flags.get(arrival_event["event_key"]): self.game_state.world.global_flags[arrival_event["event_key"]] = True return self._build_rule_final_result( story_text=arrival_event["story_text"], options=arrival_event["options"], tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_move_arrival", strict_options=True, ) scene_actions = build_scene_actions(self.game_state, destination) if scene_actions and scene_actions != build_adjacent_actions(self.game_state): story_text = self._build_location_flavor_text(destination) return self._build_rule_final_result( story_text=story_text, options=scene_actions, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_move_scene", strict_options=True, ) story_text = self._build_location_flavor_text(destination) return self._build_rule_final_result( story_text=story_text, options=[], tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_move", ) def _rule_view_map_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() not in {"VIEW_MAP", "MAP"}: return None if not any("地图" in item for item in self.game_state.player.inventory): return self._build_rule_final_result( story_text="你暂时还没有可查看的地图。", options=[], tick_log=[], change_log=[], engine_mode="rule_view_map_reject", ) tick_log = self.game_state.tick_time(player_intent) current_location = self.game_state.world.locations.get(self.game_state.player.location) visible_routes = build_map_actions(self.game_state) visible_text = "、".join(option.get("text", "") for option in visible_routes) if visible_routes else "眼下没有新的可确认路线" map_name = self._current_map_name() or "地图" story_text = ( f"你摊开{map_name}，沿着纸上的线条重新确认周围道路。\n" f"这张图目前能明确指向的去处有：{visible_text}。" ) return self._build_rule_final_result( story_text=story_text, options=visible_routes, tick_log=tick_log, change_log=[], engine_mode="rule_view_map", strict_options=True, ) def _rule_use_item_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() != "USE_ITEM": return None item_name = player_intent.get("target") if not isinstance(item_name, str) or item_name not in self.game_state.player.inventory: return None tick_log = self.game_state.tick_time(player_intent) if item_name == "火把": state_changes = { "status_effects_added": [ { "name": "火把照明", "effect_type": "buff", "description": "火光驱散了黑暗，短时间内减轻夜间探索惩罚。", "duration": 3, } ] } story_text = "你点燃火把，橙黄色的火光立刻把周围的黑暗压退了一圈。" else: consumable_effects = { "小型治疗药水": {"hp_change": 30, "items_lost": [item_name]}, "面包": {"hunger_change": 10, "items_lost": [item_name]}, "烤肉": {"hunger_change": 25, "items_lost": [item_name]}, "麦酒": {"morale_change": 10, "sanity_change": -5, "items_lost": [item_name]}, "草药包": {"hp_change": 20, "items_lost": [item_name]}, "解毒药水": {"items_lost": [item_name], "status_effects_removed": ["中毒"]}, } state_changes = consumable_effects.get(item_name) if state_changes is None: return None story_text = f"你谨慎地使用了{item_name}，身体状态随之发生了变化。" change_log = self.game_state.apply_changes(state_changes) return self._build_rule_final_result( story_text=story_text, options=[], tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_use_item", ) def _rule_talk_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() != "TALK": return None npc_name = player_intent.get("target") if not isinstance(npc_name, str) or npc_name not in self.game_state.world.npcs: return None npc = self.game_state.world.npcs[npc_name] if npc.location != self.game_state.player.location: return None tick_log = self.game_state.tick_time(player_intent) self.game_state.last_interacted_npc = npc_name if npc_name == "村长老伯": quest_updates = self._quest_updates_matching( lambda objective: ( objective == "与村长对话了解情况" or ( objective == REPORT_TO_CHIEF_OBJECTIVE and self.game_state.world.global_flags.get(FOREST_TROLL_TRACKS_FOUND_FLAG) ) ) ) else: quest_updates = self._quest_updates_matching( lambda objective: npc_name in objective or npc_name.replace("老伯", "") in objective ) state_changes: dict[str, Any] = {} first_briefing = False if quest_updates: state_changes["quest_updates"] = quest_updates first_briefing = True if npc.can_trade: _merchant_greet: dict[str, str] = { "铁匠格林": ( "铁匠格林头也不抬，把手里的铁块翻了个面，用下巴朝货架示意。" "「这些都是现货，看上哪件说。」他的话少得像省了钱，但货架上每把武器都擦得油亮。" ), "旅店老板娘莉娜": ( "莉娜正用布抹着柜台，见你走近便扬起笑来。" "「要点什么？厨房刚出了烤肉，闻起来不错。」她边说边把食物清单推过来，顺口加了一句，「外面有危险就回来，这里随时有热汤。」" ), "杂货商人阿尔": ( "杂货商人阿尔推了推眼镜，把算盘搭在柜台上。" "「没有讲价的，价格我标得很公道。」他嘴角带着职业性的微笑，目光已快速扫过你的背包，「想要什么，这里什么都有。」" ), } story_text = _merchant_greet.get( npc_name, f"你来到{npc_name}面前，对方抬手示意你看向货架，把能立即出售的物资一一报给你。", ) return self._build_rule_final_result( story_text=story_text, options=build_shop_menu_actions(self.game_state, npc_name), tick_log=tick_log, change_log=[], state_changes={}, engine_mode="rule_talk_shop", strict_options=True, ) if npc_name == "村长老伯": main_quest = self._main_quest() report_due = bool( main_quest and main_quest.status == "active" and main_quest.objectives.get(FOREST_CAUSE_OBJECTIVE) and not main_quest.objectives.get(REPORT_TO_CHIEF_OBJECTIVE) and self.game_state.world.global_flags.get(FOREST_TROLL_TRACKS_FOUND_FLAG) ) if report_due: quest_update = state_changes.setdefault("quest_updates", {}).setdefault( MAIN_QUEST_ID, {}, ) completed_objectives = list(quest_update.get("objectives_completed", [])) if REPORT_TO_CHIEF_OBJECTIVE not in completed_objectives: completed_objectives.append(REPORT_TO_CHIEF_OBJECTIVE) quest_update["objectives_completed"] = completed_objectives quest_update["status"] = "completed" self._merge_reward_changes(state_changes, self._build_main_quest_reward_changes()) state_changes.setdefault("quest_updates", {}).setdefault( MAIN_QUEST_TROLL_ID, {}, )["status"] = "active" state_changes.setdefault("npc_changes", {}).setdefault( npc_name, {}, )["relationship_change"] = 8 story_text = ( "村长老伯听完你的汇报，脸色一点点沉了下去，连握着拐杖的手都紧了几分。" "“巨大脚印、一路把哥布林撵到入口……”他低声重复了一遍，随即抬头看向你，" "“那多半是森林深处那头森林巨魔醒了。要是它真开始躁动，外层这些怪物只会越逃越乱。”\n" "他从怀里摸出一把泛着黯绿微光的古钥匙，郑重地放到你掌心里。" "“这是祖上传下来的森林之钥，能打开深处旧封锁。拿上奖励，再去铁匠铺把趁手的武器和防具备齐，" "然后立刻赶往森林深处，把那头巨魔彻底解决。”" ) change_log = self.game_state.apply_changes(state_changes) return self._build_rule_final_result( story_text=story_text, options=build_village_chief_follow_up_actions(self.game_state), tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_talk_main_quest_report", strict_options=True, ) if first_briefing and "村庄地图" not in self.game_state.player.inventory: state_changes.setdefault("items_gained", []).append("村庄地图") if first_briefing: state_changes.setdefault("npc_changes", {}).setdefault( npc_name, {}, )["relationship_change"] = 5 story_text = ( "村长老伯把你招到石井旁，先警惕地朝四周看了一眼，这才压低声音开口。" "“这两天黑暗森林很不对劲，猎人还没进林子深处，就有怪物往外窜，已经伤了两个村民。”\n" "他说着把一张反复折过的村庄地图塞到你手里，手指在村口小路和森林入口之间点了点。" "“先别逞强。去杂货铺备个火把，手里没趁手家伙的话，再去铁匠那儿挑一件。准备好了，就沿这条路去黑暗森林入口看看。”" ) elif self.game_state.world.global_flags.get(FOREST_TROLL_DEFEATED_FLAG): story_text = ( "村长老伯一见你走近，原本一直绷着的肩背终于松了下来。" "他郑重地朝你点了点头，“这次全村都该感谢你。森林巨魔倒下以后，外头那些躁动的怪物也安分多了。”" "他把拐杖稳稳拄在地上，又补了一句，“你替村子挡下了这场祸事，往后若还想上路，这里的人都会记得你的功劳。”" ) elif main_quest and main_quest.status == "completed": story_text = ( "村长老伯见你回来，神色总算松了一些，但声音依旧沉稳。" "“钥匙既然交到你手里，接下来就别把这件事当成普通清剿了。”" "他用拐杖轻轻点了点地面，“森林巨魔不是靠一腔热血能解决的，准备够了再进去。”" ) else: story_text = ( "村长老伯抬起头看了你一眼，语气比初见时更低沉些。" "“黑暗森林那边的事有进展就立刻回来告诉我。”他顿了顿，又补上一句，" "“要是还没查清楚，就别急着往深处送命，先把入口附近的痕迹看仔细。”" ) change_log = self.game_state.apply_changes(state_changes) if state_changes else [] return self._build_rule_final_result( story_text=story_text, options=build_village_chief_follow_up_actions(self.game_state), tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_talk", strict_options=True, ) if npc_name == "渡口老渔夫": has_map = "山麓地图" in self.game_state.player.inventory first_sidequest_talk = self._activate_quest_objective( state_changes, SIDE_QUEST_FERRY_ID, "与渡口老渔夫交谈", ) # 只要玩家没有山麓地图就给（无论是否首次交谈） if not has_map: state_changes.setdefault("items_gained", []).append("山麓地图") self._discover_locations("废弃矿洞入口", "山麓盗贼营", "精灵遗迹") if first_briefing or first_sidequest_talk: state_changes.setdefault("npc_changes", {}).setdefault( npc_name, {}, )["relationship_change"] = 5 if first_briefing or first_sidequest_talk or not has_map: story_text = ( "渡口老渔夫放下手里的渔网，眯着眼打量了你一番。\n" "「对岸那条矿道，夜里老是有蓝光往外漏。我在这河上钓了四十年鱼，从没见过那种颜色。」\n" "他从衣袋里掏出一张折得有些破损的图，推到你面前：\n" "「这是走惯了山路的人留下的，拿着吧，省得你乱闯。矿道不好走，小心骨头。」\n" "你在地图上标记了废弃矿洞入口、山麓盗贼营和精灵遗迹的位置。" ) else: story_text = ( "渡口老渔夫抬头看了你一眼，又把目光移回河面。\n" "「那边情况怎么样了？矿道里还有没有动静？」他顿了顿，低声说，「最近夜里声音更频了。」" ) change_log = self.game_state.apply_changes(state_changes) if state_changes else [] options = build_scene_actions(self.game_state, "河边渡口") return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_talk", strict_options=True, ) if npc_name == "遗迹守护者": first_sidequest_talk = self._activate_quest_objective( state_changes, SIDE_QUEST_GUARDIAN_ID, "与遗迹守护者交谈", ) if first_briefing or first_sidequest_talk: state_changes.setdefault("npc_changes", {}).setdefault( npc_name, {}, )["relationship_change"] = 3 state_changes.setdefault("global_flags_set", {})[GUARDIAN_INTRO_SEEN_FLAG] = True story_text = ( "穿绿袍的半精灵沉默了片刻，然后缓缓开口，声音有些嘶哑。\n" "「你来此地，是因为好奇，还是感受到了召唤？」她盯着你，「这片遗迹已被遗忘了三百年，" "而你是近年来第一个没有被石柱符文驱逐的人。」\n" "她转身走向内层石柱，「若你愿意接受试炼，遗迹将告诉你它守护的秘密。」" ) else: story_text = ( "遗迹守护者从石柱旁抬起头，用疲惫的眼神看向你。" "「试炼尚未完成。这里的秘密，需要你用行动来换取。」" ) change_log = self.game_state.apply_changes(state_changes) if state_changes else [] options = build_scene_actions(self.game_state, "精灵遗迹") return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_talk", strict_options=True, ) if npc_name == "神秘旅人": first_sidequest_talk = self._activate_quest_objective( state_changes, SIDE_QUEST_TRAVELER_ID, "与神秘旅人交谈", ) if first_briefing or first_sidequest_talk: state_changes.setdefault("npc_changes", {}).setdefault( npc_name, {}, )["relationship_change"] = 2 state_changes.setdefault("global_flags_set", {})[TRAVELER_ENCOUNTERED_FLAG] = True story_text = ( "兜帽下的眼睛在昏暗的旅店里闪着锐光。\n" "「你也察觉到了异变？」他压低声音，「森林深处有东西醒来了，那不是普通的怪物活动。」\n" "他把手指放在桌上，上面有奇异的魔法纹路，「我需要一件古老的遗物，只有在古老废墟中才能找到。" "如果你愿意帮我，报酬不会让你失望。」" ) else: story_text = ( "神秘旅人朝你微微点头，兜帽遮住了他大半张脸。" "「还没有消息吗？那件东西一定在某个废墟里，你往精灵遗迹方向找找。」" ) change_log = self.game_state.apply_changes(state_changes) if state_changes else [] options = build_adjacent_actions(self.game_state) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_talk", ) # 通用 NPC 对话：简短交谈，给出方向 story_text = f"你与{npc_name}简短地交谈了几句，对方提供了一些关于当前局势的看法，让你对下一步有了更具体的方向。" change_log = self.game_state.apply_changes(state_changes) if state_changes else [] options = merge_demo_options( build_goal_directed_actions(self.game_state), build_adjacent_actions(self.game_state), limit=3, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_talk", ) def _rule_attack_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() not in {"ATTACK", "COMBAT"}: return None enemy_name = str(player_intent.get("target") or "") location_name = str(self.game_state.player.location) encounter = get_battle_encounter(location_name, enemy_name) if encounter is None: return None defeated_flag = str(encounter.get("defeated_flag") or "") if defeated_flag and self.game_state.world.global_flags.get(defeated_flag): return self._build_rule_final_result( story_text=f"{enemy_name}已经倒在这片区域里，你暂时没有再对它挥刀的必要。", options=build_scene_actions(self.game_state, location_name), tick_log=[], change_log=[], engine_mode="rule_attack_already_resolved", strict_options=True, ) tick_log = self.game_state.tick_time(player_intent) self.game_state.refresh_combat_stats() battle_result = resolve_combat( self.game_state.player, enemy_name, game_state=self.game_state, ) outcome = str(battle_result.get("outcome", "lose")).lower() hp_loss = max(1, int(battle_result.get("player_hp_loss", 1))) if self.game_state.player.stamina <= 10: hp_loss += 2 state_changes: dict[str, Any] = {"hp_change": -hp_loss} if outcome == "lose": state_changes["morale_change"] = -8 state_changes["mp_change"] = -6 else: state_changes["mp_change"] = -4 reward_items = [ item_name for item_name in encounter.get("reward_items", []) if item_name not in self.game_state.player.inventory ] encounter_quest_id = str(encounter.get("quest_id") or MAIN_QUEST_ID) if outcome == "win": if reward_items: state_changes["items_gained"] = reward_items if defeated_flag: state_changes.setdefault("global_flags_set", {})[defeated_flag] = True quest_objectives = list(encounter.get("quest_objectives", [])) if quest_objectives: quest_update = state_changes.setdefault("quest_updates", {}).setdefault( encounter_quest_id, {}, ) completed_objectives = list(quest_update.get("objectives_completed", [])) for objective_name in quest_objectives: if objective_name not in completed_objectives: completed_objectives.append(objective_name) quest_update["objectives_completed"] = completed_objectives if enemy_name == "森林巨魔": state_changes.setdefault("quest_updates", {}).setdefault( MAIN_QUEST_TROLL_ID, {}, )["status"] = "completed" self._merge_reward_changes( state_changes, self._build_quest_reward_changes(MAIN_QUEST_TROLL_ID), ) state_changes.setdefault("global_flags_set", {})[FOREST_TROLL_HOARD_PENDING_FLAG] = True state_changes.setdefault("global_flags_set", {})[FOREST_TROLL_HOARD_CLAIMED_FLAG] = False story_text = self._generate_combat_narration( monster_name=enemy_name, location_name=location_name, outcome=outcome, player_hp_loss=hp_loss, reward_items=reward_items, ) change_log = self.game_state.apply_changes(state_changes) if self.game_state.is_game_over(): return self._generate_death_narrative() recent_gain = reward_items[0] if reward_items else None options = merge_demo_options( build_contextual_actions(self.game_state, recent_gain=recent_gain), build_scene_actions(self.game_state, location_name), build_goal_directed_actions(self.game_state), limit=MAX_OPTION_COUNT, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_attack", strict_options=True, ) def _rule_explore_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() != "EXPLORE": return None location_name = str(self.game_state.player.location) raw_input = str(player_intent.get("raw_input", "")) details = str(player_intent.get("details", "")) target = player_intent.get("target") if ( location_name == "黑暗森林入口" and self.game_state.world.global_flags.get(FOREST_GOBLIN_DEFEATED_FLAG) and not self.game_state.world.global_flags.get(FOREST_TROLL_TRACKS_FOUND_FLAG) and ( str(target) == "黑暗森林入口" or "痕迹" in raw_input or "脚印" in raw_input or "痕迹" in details or "脚印" in details or "调查" in raw_input ) ): tick_log = self.game_state.tick_time(player_intent) state_changes = { "global_flags_set": { FOREST_TROLL_TRACKS_FOUND_FLAG: True, }, "quest_updates": { MAIN_QUEST_ID: { "objectives_completed": [FOREST_CAUSE_OBJECTIVE], } }, } story_text = ( "你蹲下身检查哥布林留下的泥痕，很快就发现它脚边还叠着另一串截然不同的印记。" "那道巨大脚印深得像是有人把整块石头硬压进了湿土里，步距也远超寻常野兽。" "顺着痕迹往林中望去，你意识到这只哥布林并不是守在入口伏击，而是被更深处某个庞然大物一路逼逃到了这里。" ) change_log = self.game_state.apply_changes(state_changes) options = merge_demo_options( build_scene_actions(self.game_state, "黑暗森林入口"), build_goal_directed_actions(self.game_state), limit=MAX_OPTION_COUNT, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_explore_forest_tracks", strict_options=True, ) return None def _rule_rumor_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() != "RUMOR": return None target = player_intent.get("target") if not isinstance(target, dict): return None source = str(target.get("source") or "") topic = str(target.get("topic") or "") if not source or not topic: return None tick_log = self.game_state.tick_time(player_intent) state_changes: dict[str, Any] = {} if topic == "rumor_menu" and source == "布告栏": story_text = ( "你停在村庄广场边的布告栏前，几张被风吹得卷边的纸贴得歪歪斜斜。" "上面除了寻物和招工告示，还夹着几条最近在村里传得最凶的异闻。" ) options = [ { "text": "看看关于神秘旅人的留言", "action_type": "RUMOR", "target": {"source": "布告栏", "topic": "traveler"}, }, { "text": "看看关于矿洞鬼火的传闻", "action_type": "RUMOR", "target": {"source": "布告栏", "topic": "mine"}, }, { "text": "离开布告栏，回到广场中央", "action_type": "SCENE_OPTIONS", "target": "村庄广场", }, ] return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=[], engine_mode="rule_rumor_board_menu", strict_options=True, ) if topic == "rumor_menu" and source == "旅店老板娘莉娜": story_text = ( "莉娜把手里的木杯擦干，身体微微前倾，声音压得很低。" "“你要问最近的怪事，我这儿倒是听来两条最像真的。”她朝楼梯和门外各瞥了一眼，示意你挑一条继续问。" ) options = [ { "text": "打听旅店里的神秘旅人", "action_type": "RUMOR", "target": {"source": "旅店老板娘莉娜", "topic": "traveler"}, }, { "text": "打听矿洞晚上闹鬼的传闻", "action_type": "RUMOR", "target": {"source": "旅店老板娘莉娜", "topic": "mine"}, }, { "text": "先聊到这里，看看别的安排", "action_type": "SCENE_OPTIONS", "target": "村庄旅店", }, ] return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=[], engine_mode="rule_rumor_inn_menu", strict_options=True, ) if topic == "traveler": traveler = self.game_state.world.npcs["神秘旅人"] traveler.schedule[self.game_state.world.time_of_day] = "村口小路" self._relocate_npc("神秘旅人", "村口小路") self._discover_locations("村口小路") state_changes["global_flags_set"] = {TRAVELER_RUMOR_HEARD_FLAG: True} if source == "旅店老板娘莉娜": story_text = ( "莉娜回头看了眼楼上，确认那道裹得严严实实的身影不在附近，这才继续往下说。" "“那人白天很少久待，刚才天还亮着时我瞧见他往村口小路去了，像是在等什么人。”" "她把声音放得更轻，“你要是真想问他，就趁现在去，晚了他未必还在。”" ) else: story_text = ( "纸条上用潦草字迹写着：‘有个把自己裹得严严实实的外乡人，傍晚常在村口小路徘徊。’" "旁边还有人补了一句：‘像是在等敢接近他的人。’" ) change_log = self.game_state.apply_changes(state_changes) options = merge_demo_options( build_scene_actions(self.game_state), limit=MAX_OPTION_COUNT, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_rumor_traveler", strict_options=True, ) if topic == "mine": self._discover_locations("河边渡口") state_changes["global_flags_set"] = { MINE_RUMOR_HEARD_FLAG: True, FERRY_ROUTE_UNLOCKED_FLAG: True, } if source == "旅店老板娘莉娜": story_text = ( "莉娜一听你提起矿洞鬼火，脸上的笑意顿时收了几分。" "“这事我不敢乱编。”她把杯子轻轻放回柜台，“前几天渡口那位老渔夫亲眼见过对岸矿洞冒蓝光。”" "她朝门外一扬下巴，“你要查，就先去河边渡口找他。他比村里谁都熟那片山麓。”" ) else: story_text = ( "布告栏角落那张纸写着：‘河对岸旧矿洞夜里闹鬼，渡口老渔夫见过蓝火。’" "下面还被人重重画了一道线，像是在提醒后来者：想问详情，就去河边渡口。" ) change_log = self.game_state.apply_changes(state_changes) options = merge_demo_options( build_scene_actions(self.game_state), limit=MAX_OPTION_COUNT, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_rumor_mine", strict_options=True, ) return None def _rule_claim_reward_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() != "CLAIM_REWARD": return None target = player_intent.get("target") if not isinstance(target, dict) or str(target.get("source") or "") != "forest_troll_hoard": return None if self.game_state.player.location != "森林深处": return None if not self.game_state.world.global_flags.get(FOREST_TROLL_HOARD_PENDING_FLAG): return None tick_log = self.game_state.tick_time(player_intent) state_changes: dict[str, Any] = { "gold_change": 180, "global_flags_set": { FOREST_TROLL_HOARD_PENDING_FLAG: False, FOREST_TROLL_HOARD_CLAIMED_FLAG: True, }, } story_text = ( "你拨开巨魔尸体后方垂下的藤蔓，在树洞般的巢穴里看见了散落成堆的金币、断裂的饰链和被掠来的补给。" "你把最值钱的那一袋金币收入囊中，总算让这场恶战有了实打实的回报。" ) change_log = self.game_state.apply_changes(state_changes) return self._build_rule_final_result( story_text=story_text, options=build_scene_actions(self.game_state, "森林深处"), tick_log=tick_log, change_log=change_log, state_changes=state_changes, engine_mode="rule_claim_reward", strict_options=True, ) def _rule_rest_response(self, player_intent: dict) -> dict[str, Any] | None: if str(player_intent.get("intent", "")).upper() != "REST": return None overnight_result = self._rule_overnight_rest_response(player_intent) if overnight_result is not None: return overnight_result # pre_validate_action already ensures rest_available; double-check here loc = self.game_state.world.locations.get(self.game_state.player.location) if loc is None or not loc.rest_available: return self._build_rule_final_result( story_text="这里不适合休息，你需要找一个安全的地方，比如旅店或营地。", options=[], tick_log=[], change_log=[], engine_mode="rule_rest_reject", ) tick_log = self.game_state.tick_time(player_intent) rest_effects = self.game_state.get_rest_rule_effects() # 休息时也恢复体力 if self.game_state.player.stamina < self.game_state.player.max_stamina: stamina_gain = 30 if loc.shop_available else 20 rest_effects["stamina_change"] = stamina_gain change_log = self.game_state.apply_changes(rest_effects) if rest_effects else [] if loc.shop_available: story_text = ( "你在旅店里找了张床铺，头一碰枕头便沉沉睡去。" "再醒来时，窗外天色已经变了，身上的疲惫和淤青都轻了不少。" ) else: story_text = ( "你靠着营地旁的树根坐下，简单整理了一下伤口和装备。" "短暂的休息让体力恢复了一些，四周倒也安静，没有新的动静。" ) options = merge_demo_options( build_goal_directed_actions(self.game_state), build_scene_actions(self.game_state), limit=3, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=rest_effects, engine_mode="rule_rest", ) def _rule_overnight_rest_response(self, player_intent: dict) -> dict[str, Any] | None: intent_name = str(player_intent.get("intent", "")).upper() if intent_name not in {"REST", "OVERNIGHT_REST"}: return None if not self._looks_like_overnight_request(player_intent): return None if not self.game_state.can_overnight_rest(): return self._build_rule_final_result( story_text="现在还不能在这里过夜。晚上七点后，且仅限旅店或溪边营地。", options=[], tick_log=[], change_log=[], engine_mode="rule_overnight_rest_reject", ) tick_log, overnight_effects = self.game_state.prepare_overnight_rest() change_log = self.game_state.apply_changes(overnight_effects) if overnight_effects else [] if self.game_state.player.location == "村庄旅店": story_text = ( "你在旅店里安顿下来，热汤和床铺把一路积下的疲惫都慢慢压了下去。" "等你再睁开眼时，窗外已经是次日清晨，身上的伤势和紧绷感都恢复了大半，只剩下肚子空了不少。" ) else: story_text = ( "你在溪边营地升起余火，把装备收拾妥当后靠着树根守到睡意彻底压来。" "这一夜没有新的惊动，等天色重新发亮时，你的体力与精神都恢复了，只是野外熬了一晚，肚子也明显更空了。" ) options = merge_demo_options( build_goal_directed_actions(self.game_state), build_scene_actions(self.game_state), build_contextual_actions(self.game_state), limit=3, ) return self._build_rule_final_result( story_text=story_text, options=options, tick_log=tick_log, change_log=change_log, state_changes=overnight_effects, engine_mode="rule_overnight_rest", ) def _maybe_resolve_rule_action(self, player_intent: dict) -> dict[str, Any] | None: intent_name = str(player_intent.get("intent", "")).upper() rule_handlers = { "ATTACK": self._rule_attack_response, "COMBAT": self._rule_attack_response, "TRADE": self._rule_trade_response, "EQUIP": self._rule_equip_response, "SHOP_MENU": self._rule_shop_menu_response, "SCENE_OPTIONS": self._rule_scene_options_response, "MOVE": self._rule_move_response, "VIEW_MAP": self._rule_view_map_response, "MAP": self._rule_view_map_response, "USE_ITEM": self._rule_use_item_response, "TALK": self._rule_talk_response, "RUMOR": self._rule_rumor_response, "CLAIM_REWARD": self._rule_claim_reward_response, "OVERNIGHT_REST": self._rule_overnight_rest_response, "REST": self._rule_rest_response, "EXPLORE": self._rule_explore_response, } handler = rule_handlers.get(intent_name) if handler is None: return None return handler(player_intent) def generate_opening(self) -> dict: """ 生成游戏开场叙事。 Returns: { "story_text": "开场叙事文本", "options": [{"id": 1, "text": "...", "action_type": "..."}], "state_changes": {} } """ logger.info("生成游戏开场叙事...") prompt = OPENING_NARRATIVE_PROMPT.format( world_state=self.game_state.to_prompt(), player_name=self.game_state.player.name, ) messages = [ {"role": "system", "content": prompt}, {"role": "user", "content": "请开始讲述故事的开场。"}, ] raw_text = call_qwen(messages, model=self.model, temperature=0.9, max_tokens=2000) story_text, options = self._parse_story_response(raw_text) options = self._validate_options(options) # 开场没有状态变更 return { "story_text": story_text, "options": options, "state_changes": {}, "change_log": [], "telemetry": _build_telemetry(engine_mode="opening", used_fallback=False), } def generate_story(self, player_intent: dict) -> dict: """ 核心方法：根据玩家意图生成完整的故事响应。 两阶段生成流程： 1. 生成剧情大纲（结构化 JSON） 2. 一致性检查 3. 基于大纲生成文学文本 + 选项 4. 应用状态变更 Args: player_intent: NLU 引擎解析出的意图 { "intent": "ATTACK", "target": "哥布林", "details": "用剑攻击", "raw_input": "我想攻击那个哥布林" } Returns: { "story_text": "剧情文本", "options": [选项列表], "state_changes": {状态变更}, "change_log": ["变更日志"], "outline": {大纲}, "consistency_issues": ["一致性问题"], } """ logger.info(f"生成故事响应，玩家意图: {player_intent}") rule_result = self._maybe_resolve_rule_action(player_intent) if rule_result is not None: return rule_result outline_regenerated = False # ============================================ # 推进时间（行动前，时间自然流逝） # 设计思路：将 tick_time 放在大纲生成之前，确保： # 1. LLM 看到的是 tick 后的真实状态 # 2. apply_changes 的 change_log 与状态栏显示一致 # 3. 不会出现“文本说100、状态栏显示97”的同步Bug # ============================================ tick_log = self.game_state.tick_time(player_intent) # ============================================ # 第一阶段：生成剧情大纲 # ============================================ outline = self._generate_outline(player_intent) if outline is None: # 大纲生成失败 —— 降级处理 logger.error("大纲生成失败，使用降级叙事") return self._fallback_response( player_intent, tick_log, fallback_reason="outline_generation_failed", engine_mode="two_stage", ) extracted_location = str(outline.get("location") or "").strip() or "未知之地" # ============================================ # 处理时间冲突：如果大纲指定了 time_change（时间跳跃）， # 则移除 tick_time 自动推进的"时间流逝"日志， # 避免出现"时间流逝: 上午→正午"和"时段变为: 下午"两条冲突记录。 # ============================================ state_changes = outline.get("state_changes", {}) if state_changes.get("time_change"): tick_log = [line for line in tick_log if not line.startswith("时间流逝:")] # ============================================ # 一致性检查 # ============================================ consistency_issues = self.game_state.check_consistency(state_changes) if consistency_issues: logger.warning(f"发现一致性问题: {consistency_issues}") # 尝试修复：重新生成大纲，附带一致性约束 outline = self._regenerate_outline_with_fixes(player_intent, consistency_issues) if outline is None: return self._fallback_response( player_intent, tick_log, fallback_reason="outline_regeneration_failed", engine_mode="two_stage", ) outline_regenerated = True extracted_location = str(outline.get("location") or "").strip() or "未知之地" state_changes = outline.get("state_changes", {}) # 再次检查 consistency_issues = self.game_state.check_consistency(state_changes) # 移除与非法物品相关的状态变更（安全网） if consistency_issues: state_changes = self._strip_invalid_item_effects(state_changes) # ============================================ # 清理状态变更：阻止非消耗品被错误移除 # ============================================ event_type = outline.get("event_type", "") state_changes, sanitize_warnings = self._sanitize_state_changes(state_changes, event_type) state_changes, item_rule_notes = self._fill_missing_consumable_effects(state_changes, player_intent) state_changes, rest_rule_notes = self._fill_missing_rest_effects(state_changes, player_intent) if sanitize_warnings: logger.info(f"状态变更清理: {sanitize_warnings}") # ============================================ # 应用状态变更 # ============================================ change_log = self.game_state.apply_changes(state_changes) # 校验状态合法性 is_valid, validation_issues = self.game_state.validate() # 记录事件 self.game_state.log_event( event_type=outline.get("event_type", "UNKNOWN"), description=outline.get("event_summary", ""), player_action=player_intent.get("raw_input", ""), involved_npcs=outline.get("involved_npcs", []), state_changes=state_changes, consequence_tags=outline.get("consequence_tags", []), is_reversible=outline.get("is_reversible", True), ) # ============================================ # 检查是否游戏结束 # ============================================ if self.game_state.is_game_over(): return self._generate_death_narrative() # ============================================ # 第二阶段：生成文学文本 + 选项 # （将实际 change_log 注入叙事 Prompt，确保文本数值与状态一致） # ============================================ story_text, options = self._generate_narrative(outline, change_log) # 验证生成的选项：确保不引用玩家没有的物品 options = self._validate_options(options) # 合并 tick_log 和 change_log 中的重复属性条目 merged_log = _merge_change_logs( tick_log, change_log + validation_issues + item_rule_notes + rest_rule_notes, ) return { "story_text": story_text, "options": options, "location": extracted_location, "state_changes": state_changes, "change_log": merged_log, "outline": outline, "consistency_issues": consistency_issues, "telemetry": _build_telemetry( engine_mode="two_stage", used_fallback=False, consistency_issues_count=len(consistency_issues), validation_issues_count=len(validation_issues), outline_regenerated=outline_regenerated, ), } def _generate_outline(self, player_intent: dict) -> Optional[dict]: """ 第一阶段：生成剧情大纲。 使用低温度 (0.3) 确保 JSON 结构稳定， 将完整世界状态注入 System Prompt。 """ system_prompt = OUTLINE_SYSTEM_PROMPT_TEMPLATE.format( world_state=self.game_state.to_prompt(), ) # 构造用户消息：包含意图的完整描述 user_message = ( f"玩家行动: {player_intent.get('raw_input', '未知行动')}\n" f"解析意图: {player_intent.get('intent', 'UNKNOWN')}\n" f"目标: {player_intent.get('target', '无')}\n" f"细节: {player_intent.get('details', '无')}" ) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message}, ] result = safe_json_call(messages, model=self.model, temperature=0.3, max_tokens=1500) if result and isinstance(result, dict): logger.info(f"大纲生成成功: {result.get('event_summary', 'N/A')}") return result logger.error("大纲生成失败，未能获取有效 JSON") return None def _regenerate_outline_with_fixes( self, player_intent: dict, issues: list[str] ) -> Optional[dict]: """ 当一致性检查发现问题时，重新生成大纲。 将发现的矛盾作为额外约束注入 Prompt。 """ logger.info("重新生成大纲（附带一致性修正）...") system_prompt = OUTLINE_SYSTEM_PROMPT_TEMPLATE.format( world_state=self.game_state.to_prompt(), ) issues_text = "\n".join(f"- {issue}" for issue in issues) user_message = ( f"玩家行动: {player_intent.get('raw_input', '未知行动')}\n" f"解析意图: {player_intent.get('intent', 'UNKNOWN')}\n" f"目标: {player_intent.get('target', '无')}\n" f"细节: {player_intent.get('details', '无')}\n\n" f"【⚠️ 注意：上一次生成存在以下矛盾，请修正】\n{issues_text}\n" f"请重新生成符合逻辑的剧情大纲。" ) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message}, ] result = safe_json_call(messages, model=self.model, temperature=0.2, max_tokens=1500) if result and isinstance(result, dict): logger.info(f"修正后大纲生成成功: {result.get('event_summary', 'N/A')}") return result return None def _generate_narrative(self, outline: dict, change_log: list[str] | None = None) -> tuple[str, list[dict]]: """ 第二阶段：基于大纲生成文学文本 + 3 个选项。 使用中等温度 (0.8) 增加文学创意。 Args: outline: 第一阶段生成的剧情大纲 change_log: apply_changes 返回的实际状态变更日志，用于约束叙事中的数值描写 """ import json actual_changes_text = "\n".join(change_log) if change_log else "无状态变化" system_prompt = NARRATIVE_SYSTEM_PROMPT_TEMPLATE.format( world_state=self.game_state.to_prompt(), outline=json.dumps(outline, ensure_ascii=False, indent=2), actual_changes=actual_changes_text, ) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": "请基于以上大纲，生成剧情描写和选项。"}, ] raw_text = call_qwen( messages, model=self.model, temperature=0.8, max_tokens=2000 ) story_text, options = self._parse_story_response(raw_text) return story_text, options def _generate_death_narrative(self) -> dict: """生成死亡结局叙事""" logger.info("生成死亡结局...") death_context = self.game_state.get_death_narrative_context() prompt = DEATH_NARRATIVE_PROMPT.format( death_context=death_context, world_state=self.game_state.to_prompt(), ) messages = [ {"role": "system", "content": prompt}, {"role": "user", "content": "请为这位冒险者写一段死亡结局。"}, ] raw_text = call_qwen(messages, model=self.model, temperature=0.9, max_tokens=1500) story_text, options = self._parse_story_response(raw_text) return { "story_text": story_text, "options": options, "location": "未知之地", "state_changes": {}, "change_log": ["游戏结束"], "outline": None, "consistency_issues": [], "telemetry": _build_telemetry(engine_mode="death_narrative", used_fallback=False), } @staticmethod def _clean_story_text(story_text: str) -> str: """ 清理故事文本中残留的 JSON 选项数组和格式标记。 LLM 有时会将选项 JSON 嵌入到故事文本中（尤其在标记之间）， 此方法移除这些 JSON 片段，仅保留纯叙事文本。 """ # 移除看起来像选项 JSON 数组的内容: [{"id": ...}, ...] cleaned = re.sub( r'\[\s*\{\s*"id"\s*:.*?\]', '', story_text, flags=re.DOTALL, ) # 移除残留的标记 for marker in ["---STORY_TEXT---", "---OPTIONS_JSON---", "---STATE_JSON---", "---THINKING---"]: cleaned = cleaned.replace(marker, "") # 移除可能因清理产生的多余空行（保留最多一个空行） cleaned = re.sub(r'\n{3,}', '\n\n', cleaned) return cleaned.strip() def _parse_story_response(self, raw_text: str) -> tuple[str, list[dict]]: """ 解析 LLM 返回的故事响应，分离文本和选项。 采用 **暴力提取策略**： 1. 优先尝试标准标记格式 2. 如都失败，默认将整段文本视为故事文本 3. 从末尾向前查找 JSON 块 ([...] 或 {...})，若找到则切除并解析为选项 4. 找不到任何 JSON 则整段当故事 + 硬性默认选项 """ # ★ 调试日志 logger.debug(f"原始 API 返回内容: {raw_text}") story_text = "" options = [] if not raw_text or not raw_text.strip(): logger.warning("API 返回内容为空") return "你环顾四周，思考着接下来该做什么...", self._generate_default_options() # 标准化标记格式 normalized = _normalize_markers(raw_text) # ========== 策略 1：标准标记格式 ========== if "---STORY_TEXT---" in normalized and "---OPTIONS_JSON---" in normalized: story_start = normalized.index("---STORY_TEXT---") + len("---STORY_TEXT---") options_start = normalized.index("---OPTIONS_JSON---") story_part = normalized[story_start:options_start].strip() options_part = normalized[options_start + len("---OPTIONS_JSON---"):].strip() story_text = story_part parsed_options = extract_json_from_text(options_part) if isinstance(parsed_options, list) and len(parsed_options) > 0: options = parsed_options logger.info("标准格式解析成功") # ========== 策略 1b：只有 STORY_TEXT 没有 OPTIONS_JSON ========== elif "---STORY_TEXT---" in normalized: logger.warning("找到 STORY_TEXT 标记但缺少 OPTIONS_JSON，暴力提取") story_start = normalized.index("---STORY_TEXT---") + len("---STORY_TEXT---") remaining = normalized[story_start:].strip() # 从 remaining 末尾查找 JSON story_text, options = self._brute_force_extract(remaining) # ========== 策略 2：完全没有标记 → 暴力提取 ========== if not story_text.strip(): logger.warning("响应格式不标准，使用暴力提取策略") story_text, options = self._brute_force_extract(normalized) # 清理 story_text 中可能残留的 JSON 和标记 story_text = self._clean_story_text(story_text) # 移除常见的 AI 前缀 story_text = re.sub(r'^(好的[，,]?\s*(这是|以下是|下面是).*?[：:]\s*)', '', story_text, flags=re.MULTILINE) # 如果 story_text 仍为空，使用兜底文本 if not story_text.strip(): logger.warning("解析后 story_text 为空，使用兜底文本") story_text = "你环顾四周，思考着接下来该做什么..." # 确保至少有默认选项 if not options: logger.info("未提取到有效选项，使用默认选项") options = self._generate_default_options() options = self._ensure_option_count(options) return story_text, options def _brute_force_extract(self, text: str) -> tuple[str, list[dict]]: """ 暴力提取策略：默认整段文本都是故事，从末尾向前查找 JSON 块并切除。 Args: text: 待解析的文本 Returns: (story_text, options_list) """ options = [] story_text = text # 第一步：尝试从末尾向前找 JSON 数组 [...] json_array_matches = list(re.finditer(r'\[\s*\{.*?\}\s*\]', text, re.DOTALL)) if json_array_matches: last_match = json_array_matches[-1] try: parsed = extract_json_from_text(last_match.group()) if isinstance(parsed, list) and len(parsed) > 0: # 检查是否看起来像选项（至少一个元素有 text 或 id 字段） if any(isinstance(item, dict) and ("text" in item or "id" in item) for item in parsed): options = parsed story_text = text[:last_match.start()].strip() logger.info(f"暴力提取：从末尾找到 JSON 数组，提取了 {len(options)} 个选项") except Exception: pass # 第二步：如果没找到数组，尝试找独立的 JSON 对象 {...} if not options: json_obj_matches = list(re.finditer(r'\{[^{}]*"(?:text|id|action_type)"[^{}]*\}', text, re.DOTALL)) if json_obj_matches and len(json_obj_matches) >= 2: first_match_start = json_obj_matches[0].start() last_match_end = json_obj_matches[-1].end() json_block = text[first_match_start:last_match_end] wrapped = "[" + json_block + "]" try: parsed = extract_json_from_text(wrapped) if isinstance(parsed, list) and len(parsed) > 0: options = parsed story_text = text[:first_match_start].strip() logger.info(f"暴力提取：通过独立 JSON 对象拼合，提取了 {len(options)} 个选项") except Exception: pass # 第三步：完全没有 JSON 影子 → 整段当故事 if not options: logger.info("暴力提取：未发现任何 JSON，整段文本作为故事，使用默认选项") story_text = text return story_text, options def _generate_default_options(self, limit: int = DEFAULT_OPTION_COUNT) -> list[dict]: """生成默认选项（当 LLM 未能提供有效选项时的降级方案）""" available = self.game_state.get_available_actions() default_options = [] action_map = { "观察": {"text": "仔细观察周围的环境", "action_type": "EXPLORE"}, "对话": {"text": "与附近的人交谈", "action_type": "TALK"}, "移动": {"text": "前往其他地方", "action_type": "MOVE"}, "休息": {"text": "在此处休息一会儿", "action_type": "REST"}, "战斗": {"text": "准备战斗", "action_type": "ATTACK"}, "搜索": {"text": "搜索这个区域", "action_type": "EXPLORE"}, } for i, action in enumerate(available[:limit], 1): if action in action_map: opt = action_map[action].copy() opt["id"] = i default_options.append(opt) while len(default_options) < limit: default_options.append({ "id": len(default_options) + 1, "text": "继续探索", "action_type": "EXPLORE", }) return default_options[:limit] def _ensure_option_count( self, options: list[dict], *, minimum: int = DEFAULT_OPTION_COUNT, maximum: int = MAX_OPTION_COUNT, ) -> list[dict]: """Normalize option count for the current UI contract.""" normalized_options = list(options[:maximum]) if minimum > 0 and len(normalized_options) < minimum: defaults = self._generate_default_options(limit=maximum) for default_option in defaults: if len(normalized_options) >= minimum: break if not any( existing.get("text") == default_option["text"] for existing in normalized_options if isinstance(existing, dict) ): normalized_options.append(default_option.copy()) while len(normalized_options) < minimum: normalized_options.append({ "id": len(normalized_options) + 1, "text": "继续探索", "action_type": "EXPLORE", }) for i, opt in enumerate(normalized_options[:maximum], 1): opt["id"] = i return normalized_options[:maximum] def _ensure_three_options(self, options: list[dict]) -> list[dict]: """Compatibility wrapper for legacy call sites.""" return self._ensure_option_count( options, minimum=DEFAULT_OPTION_COUNT, maximum=MAX_OPTION_COUNT, ) def _fallback_response( self, player_intent: dict, tick_log: list[str] | None = None, *, fallback_reason: str = "unknown", engine_mode: str = "fallback", ) -> dict: """ 降级响应：当大纲生成完全失败时，提供基本响应。 设计思路： - 直接用单次调用生成简短叙事 + 选项 - 不涉及状态变更（安全保守策略） - 确保游戏不会卡死 Args: player_intent: 玩家意图 tick_log: tick_time 返回的时间流逝日志（可能为 None） """ logger.warning("使用降级响应模式") fallback_prompt = ( f"你是一个 RPG 游戏的叙事引擎。\n" f"当前场景: {self.game_state.world.current_scene}\n" f"玩家说: {player_intent.get('raw_input', '...')}\n\n" f"请写一段简短的过渡叙事（100字以内），并给出 3 个后续选项。\n" f"格式:\n" f"---STORY_TEXT---\n(叙事文本)\n---OPTIONS_JSON---\n" f'[{{"id":1,"text":"选项1","action_type":"EXPLORE"}},' f'{{"id":2,"text":"选项2","action_type":"TALK"}},' f'{{"id":3,"text":"选项3","action_type":"MOVE"}}]' ) messages = [ {"role": "system", "content": fallback_prompt}, {"role": "user", "content": "请继续。"}, ] try: raw_text = call_qwen(messages, model=self.model, temperature=0.7, max_tokens=800) story_text, options = self._parse_story_response(raw_text) except Exception: story_text = "你沉思片刻，思考着下一步该怎么做..." options = self._generate_default_options() fallback_change_log = (tick_log or []) + ["（系统提示：本回合使用了降级响应）"] return { "story_text": story_text, "options": options, "location": "未知之地", "state_changes": {}, "change_log": fallback_change_log, "outline": None, "consistency_issues": [], "telemetry": _build_telemetry( engine_mode=engine_mode, used_fallback=True, fallback_reason=fallback_reason, ), } def _sanitize_state_changes(self, changes: dict, event_type: str = "") -> tuple[dict, list[str]]: """ 清理状态变更，避免非预期的高风险副作用进入正式状态。 规则： - 交易(TRADE)、赠送、丢弃等行为可以移除任何物品 - 其他行为（使用、探索、对话等）只能移除消耗品 - 非战斗事件不能直接造成 NPC 死亡 Args: changes: 状态变更字典 event_type: 事件类型（来自大纲） Returns: (清理后的changes, 警告列表) """ warnings = [] changes = dict(changes) if "new_location" in changes: blocked_location = str(changes.get("new_location")) warnings.append(f"忽略模型生成的位置变更: {blocked_location}") logger.warning("阻止模型直接改写位置: %s", blocked_location) changes.pop("new_location", None) if "items_lost" in changes: # 交易/赠送行为可以移除任何物品 trade_events = {"TRADE", "GIVE", "DROP"} if event_type.upper() not in trade_events: # 其他行为：只允许移除消耗品 sanitized_items_lost = [] for item_name in changes["items_lost"]: item_str = str(item_name) if self.game_state.is_item_consumable(item_str): sanitized_items_lost.append(item_name) else: warnings.append( f"物品 '{item_str}' 不是消耗品，使用后仍保留在背包中" ) logger.warning(f"阻止移除非消耗品: {item_str}") if sanitized_items_lost: changes["items_lost"] = sanitized_items_lost else: changes.pop("items_lost", None) changes, item_warnings = self._sanitize_controlled_item_changes(changes, event_type) warnings.extend(item_warnings) changes, npc_warnings = self._sanitize_noncombat_npc_changes(changes, event_type) warnings.extend(npc_warnings) return changes, warnings def _sanitize_controlled_item_changes( self, changes: dict, event_type: str = "", ) -> tuple[dict, list[str]]: """Restrict item gain/equip to registered items and paid trade outcomes.""" sanitized_changes = dict(changes) warnings: list[str] = [] registered_items = set(self.game_state.world.item_registry.keys()) inventory_items = set(self.game_state.player.inventory) equipped_items = { str(item) for item in self.game_state.player.equipment.values() if item and str(item).lower() not in {"none", "null", ""} } allowed_trade_items = { str(item_name) for npc in self.game_state.world.npcs.values() if npc.can_trade for item_name in npc.shop_inventory } original_items_gained = [ str(item) for item in sanitized_changes.get("items_gained", []) ] valid_items_gained: list[str] = [] for item_name in original_items_gained: if item_name not in registered_items: warning = f"移除未注册物品获取: {item_name}" warnings.append(warning) logger.warning(warning) continue if event_type.upper() == "TRADE" and allowed_trade_items and item_name not in allowed_trade_items: warning = f"移除不在商店配置中的交易物品: {item_name}" warnings.append(warning) logger.warning(warning) continue valid_items_gained.append(item_name) if valid_items_gained: sanitized_changes["items_gained"] = valid_items_gained else: sanitized_changes.pop("items_gained", None) if event_type.upper() == "TRADE": gold_change = sanitized_changes.get("gold_change") items_lost = sanitized_changes.get("items_lost", []) is_paid_trade = ( isinstance(gold_change, (int, float)) and gold_change < 0 ) or bool(items_lost) if not is_paid_trade: if "items_gained" in sanitized_changes: warning = "移除未支付的交易物品获取" warnings.append(warning) logger.warning(warning) sanitized_changes.pop("items_gained", None) if "equip" in sanitized_changes: warning = "移除未支付的交易装备变更" warnings.append(warning) logger.warning(warning) sanitized_changes.pop("equip", None) if "equip" in sanitized_changes and isinstance(sanitized_changes["equip"], dict): valid_equippable = inventory_items | equipped_items | set( sanitized_changes.get("items_gained", []) ) sanitized_equip: dict[str, str] = {} for slot, item_name in sanitized_changes["equip"].items(): item_str = str(item_name) if item_str.lower() in {"none", "null", ""}: sanitized_equip[str(slot)] = item_name continue if item_str not in registered_items: warning = f"移除未注册装备变更: {slot} -> {item_str}" warnings.append(warning) logger.warning(warning) continue if event_type.upper() == "TRADE" and allowed_trade_items and item_str not in allowed_trade_items: warning = f"移除不在商店配置中的交易装备: {slot} -> {item_str}" warnings.append(warning) logger.warning(warning) continue if item_str not in valid_equippable: warning = f"移除未持有装备变更: {slot} -> {item_str}" warnings.append(warning) logger.warning(warning) continue sanitized_equip[str(slot)] = item_str if sanitized_equip: sanitized_changes["equip"] = sanitized_equip else: sanitized_changes.pop("equip", None) return sanitized_changes, warnings def _sanitize_noncombat_npc_changes( self, changes: dict, event_type: str = "", ) -> tuple[dict, list[str]]: """移除非战斗事件中意外出现的 NPC 致死变更。""" npc_changes = changes.get("npc_changes") if not isinstance(npc_changes, dict): return changes, [] noncombat_events = { "TRADE", "GIVE", "DROP", "TALK", "DIALOGUE", "MOVE", "EXPLORE", "REST", "DISCOVERY", } if event_type.upper() not in noncombat_events: return changes, [] sanitized_changes = dict(changes) sanitized_npc_changes: dict[str, dict] = {} warnings: list[str] = [] for npc_name, npc_data in npc_changes.items(): if not isinstance(npc_data, dict): sanitized_npc_changes[str(npc_name)] = npc_data continue cleaned_data = dict(npc_data) removed_keys: list[str] = [] if cleaned_data.get("is_alive") is False: cleaned_data.pop("is_alive", None) removed_keys.append("is_alive") hp_delta = cleaned_data.get("hp_change") npc = self.game_state.world.npcs.get(str(npc_name)) try: hp_delta_value = int(hp_delta) if hp_delta is not None else None except (TypeError, ValueError): hp_delta_value = None if ( hp_delta_value is not None and npc is not None and npc.hp + hp_delta_value <= 0 ): cleaned_data.pop("hp_change", None) removed_keys.append("hp_change") if removed_keys: warning = ( f"移除非战斗事件中的 NPC 致死变更: {npc_name} -> " f"{', '.join(removed_keys)}" ) warnings.append(warning) logger.warning(warning) if cleaned_data: sanitized_npc_changes[str(npc_name)] = cleaned_data if sanitized_npc_changes: sanitized_changes["npc_changes"] = sanitized_npc_changes else: sanitized_changes.pop("npc_changes", None) return sanitized_changes, warnings def _fill_missing_consumable_effects( self, state_changes: dict, player_intent: dict, ) -> tuple[dict, list[str]]: """ Fill missing consumable effects from item metadata without overriding model output. """ if str(player_intent.get("intent", "")).upper() != "USE_ITEM": return state_changes, [] consumed_items = [ str(item) for item in state_changes.get("items_lost", []) if self.game_state.is_item_consumable(str(item)) ] if not consumed_items: return state_changes, [] merged_changes = dict(state_changes) applied_notes: list[str] = [] for item_name in consumed_items: inferred = self.game_state.get_consumable_rule_effects(item_name) if not inferred: continue applied_keys = [] for key, value in inferred.items(): if key in merged_changes: continue merged_changes[key] = list(value) if isinstance(value, list) else value applied_keys.append(key) if applied_keys: logger.info("按物品规则补全缺失效果: %s -> %s", item_name, applied_keys) applied_notes.append(f"物品规则补全: {item_name}") return merged_changes, applied_notes def _fill_missing_rest_effects( self, state_changes: dict, player_intent: dict, ) -> tuple[dict, list[str]]: """ Fill missing rest recovery with conservative defaults at rest-enabled locations. """ if str(player_intent.get("intent", "")).upper() != "REST": return state_changes, [] recovery_keys = ( "hp_change", "mp_change", "morale_change", "sanity_change", "hunger_change", ) if any(key in state_changes for key in recovery_keys): return state_changes, [] inferred = self.game_state.get_rest_rule_effects() if not inferred: return state_changes, [] merged_changes = dict(state_changes) merged_changes.update(inferred) logger.info("按休息规则补全缺失效果: %s", inferred) return merged_changes, ["休息规则补全"] def _strip_invalid_item_effects(self, state_changes: dict) -> dict: """ 当 LLM 生成了涉及不存在物品的状态变更时，移除相关效果（安全网）。 例如：LLM 生成了"吃了包子" → hunger_change: +10, items_lost: ["包子"]， 但包子不在背包中。此方法移除 items_lost 和相关的属性效果。 """ changes = dict(state_changes) if "items_lost" not in changes: return changes items_gained = {str(i) for i in changes.get("items_gained", [])} invalid_count = 0 valid_items_lost = [] for item in changes["items_lost"]: item_str = str(item) # 同回合获得的物品可以立即消耗（如拾取后使用） if item_str in items_gained: valid_items_lost.append(item) continue if item_str in self.game_state.player.inventory: valid_items_lost.append(item) else: invalid_count += 1 logger.warning(f"[安全网] 阻止消耗不存在的物品「{item_str}」") if invalid_count > 0: if valid_items_lost: changes["items_lost"] = valid_items_lost else: changes.pop("items_lost", None) # 所有 items_lost 都无效 → 属性变更可能全部源于非法物品使用 for key in ["hunger_change", "hp_change", "mp_change", "morale_change", "sanity_change"]: if key in changes: logger.warning(f"[安全网] 因物品不合法，阻止 {key}: {changes[key]}") changes.pop(key) if "status_effects_added" in changes: logger.warning("[安全网] 因物品不合法，阻止状态效果添加") changes.pop("status_effects_added") return changes def _validate_options( self, options: list[dict], *, enrich: bool = True, minimum: int = DEFAULT_OPTION_COUNT, maximum: int = MAX_OPTION_COUNT, ) -> list[dict]: """ 验证生成的选项：移除引用了玩家不拥有的物品的选项。 检查逻辑： - 收集所有已知物品名（来自 item_registry 和事件日志） - 如果选项文本提及了某个已知物品，但该物品不在当前背包中， 则该选项无效，替换为默认选项 """ inventory = set(self.game_state.player.inventory) # 构建所有已知物品名称集合 known_items: set[str] = set(self.game_state.world.item_registry.keys()) for event in self.game_state.event_log: sc = event.state_changes if isinstance(sc, dict): for item in sc.get("items_gained", []): known_items.add(str(item)) for item in sc.get("items_lost", []): known_items.add(str(item)) # 玩家不拥有的已知物品 unavailable_items = known_items - inventory validated = [] for opt in options: text = opt.get("text", "") action_type = opt.get("action_type", "") is_valid = True # 检查选项是否引用了玩家不拥有的物品 for item_name in unavailable_items: # 只检查名称足够长的物品（避免单字误匹配） if len(item_name) >= 2 and item_name in text: # 额外确认：如果选项是使用类动作，更可能是无效的 # 对于移动/对话类选项，提及物品可能只是描述性的 if action_type in ("USE_ITEM", "EQUIP", "SKILL"): logger.warning( f"移除无效选项（引用了不在背包中的物品 '{item_name}'）: {text}" ) is_valid = False break # 对于其他动作类型，用更严格的判断：是否是“使用XX”“吹XX”“吃XX”等模式 use_patterns = [ f"使用{item_name}", f"吹响{item_name}", f"吹{item_name}", f"吃{item_name}", f"喝{item_name}", f"装备{item_name}", f"拿出{item_name}", f"打开{item_name}", ] if any(pattern in text for pattern in use_patterns): logger.warning( f"移除无效选项（涉及使用不在背包中的物品 '{item_name}'）: {text}" ) is_valid = False break if is_valid: normalized = self._normalize_option(opt) if normalized is not None: validated.append(normalized) if enrich: guided_options = build_goal_directed_actions(self.game_state) contextual_options = build_contextual_actions( self.game_state, recent_gain=self.game_state.last_recent_gain, ) adjacent_options = build_adjacent_actions(self.game_state) merged_options = merge_demo_options( validated, guided_options, contextual_options, adjacent_options, limit=maximum, ) else: merged_options = validated filtered_options = [ option for option in merged_options if not ( option.get("action_type") == "TALK" and option.get("target") == self.game_state.last_interacted_npc and not self._objective_requires_npc(str(option.get("target"))) ) ] return self._ensure_option_count( filtered_options, minimum=minimum, maximum=maximum, ) def process_option_selection(self, option: dict) -> dict: """ 处理玩家点击选项的操作。 将选项转化为意图格式，然后调用 generate_story。 Args: option: 玩家选择的选项 {"id": 1, "text": "...", "action_type": "..."} Returns: generate_story 的返回结果 """ intent = { "intent": option.get("action_type", "EXPLORE"), "target": option.get("target"), "details": option.get("text", ""), "raw_input": option.get("text", ""), } return self.generate_story(intent) # ============================================================ # 流式输出 + 合并式生成（核心优化） # ============================================================ def generate_opening_stream(self): """ 流式生成游戏开场叙事。 ★ 重要设计：流式循环中仅做文本展示（story_chunk）， 选项解析逻辑严格在整个数据流结束后才执行一次。 Yields: {"type": "story_chunk", "text": "累积的故事文本"} — 流式文本更新 {"type": "final", ...} — 最终完整结果（包含必定 3 个选项） """ logger.info("[流式] 生成游戏开场叙事...") prompt = OPENING_NARRATIVE_PROMPT.format( world_state=self.game_state.to_prompt(), player_name=self.game_state.player.name, ) messages = [ {"role": "system", "content": prompt}, {"role": "user", "content": "请开始讲述故事的开场。"}, ] full_text = "" story_started = False story_ended = False try: for chunk in call_qwen_stream(messages, model=self.model, temperature=0.9, max_tokens=2000): full_text += chunk # ★ 流式循环中只做展示用途的标记检测，绝不在此处解析选项 normalized = _normalize_markers(full_text) if not story_started: if "---STORY_TEXT---" in normalized: story_started = True idx = normalized.index("---STORY_TEXT---") + len("---STORY_TEXT---") current_story = normalized[idx:] if "---OPTIONS_JSON---" in current_story: story_ended = True current_story = current_story[:current_story.index("---OPTIONS_JSON---")] display_text = self._clean_story_text(current_story) if display_text.strip(): yield {"type": "story_chunk", "text": display_text.strip()} elif not story_ended: idx = normalized.index("---STORY_TEXT---") + len("---STORY_TEXT---") current_story = normalized[idx:] if "---OPTIONS_JSON---" in current_story: story_ended = True current_story = current_story[:current_story.index("---OPTIONS_JSON---")] display_text = self._clean_story_text(current_story) if display_text.strip(): yield {"type": "story_chunk", "text": display_text.strip()} except Exception as e: logger.error(f"流式开场生成失败: {e}，降级为非流式") try: result = self.generate_opening() result["options"] = self._validate_options(result.get("options", [])) yield {"type": "final", **result} except Exception: yield { "type": "final", "story_text": "你踏上了一段新的旅程...", "options": self._generate_default_options(), "state_changes": {}, "change_log": [], "telemetry": _build_telemetry( engine_mode="opening", used_fallback=True, fallback_reason="opening_stream_exception", ), } return # ★ 如果流式阶段未检测到标记但有累积文本，先 yield 给 UI 显示 if not story_started and full_text.strip(): logger.warning("流式阶段未检测到 STORY_TEXT 标记，直接使用累积文本") # 清理后先 yield 给 UI，确保用户能看到文本 display = self._clean_story_text(full_text.strip()) display = re.sub(r'^(好的[，,]?\s*(这是|以下是|下面是).*?[：:]\s*)', '', display, flags=re.MULTILINE) if display.strip(): yield {"type": "story_chunk", "text": display.strip()} # ★ 核心：只在数据流完全结束后，用完整 full_text 解析（暴力提取策略） story_text, options = self._parse_story_response(full_text) options = self._validate_options(options) logger.info(f"[流式开场] 最终 story_text 长度={len(story_text)}, 选项数={len(options)}") yield { "type": "final", "story_text": story_text, "options": options, "state_changes": {}, "change_log": [], "telemetry": _build_telemetry(engine_mode="opening", used_fallback=False), } def generate_story_stream(self, player_intent: dict): """ 流式生成故事响应（合并大纲+叙事为一次 API 调用）。 使用 MERGED_SYSTEM_PROMPT_TEMPLATE，一次调用完成： - 内部规划（THINKING 区域，不展示） - 故事文本（STORY_TEXT 区域，流式展示） - 状态变更（STATE_JSON 区域，程序解析） - 选项（OPTIONS_JSON 区域，程序解析） Yields: {"type": "story_chunk", "text": "累积的故事文本"} {"type": "final", "story_text": ..., "options": ..., ...} """ logger.info(f"[流式/合并] 生成故事响应，玩家意图: {player_intent}") rule_result = self._maybe_resolve_rule_action(player_intent) if rule_result is not None: yield {"type": "final", **rule_result} return # 推进时间 tick_log = self.game_state.tick_time(player_intent) # 构建合并 Prompt system_prompt = MERGED_SYSTEM_PROMPT_TEMPLATE.format( world_state=self.game_state.to_prompt(), ) user_message = ( f"玩家行动: {player_intent.get('raw_input', '未知行动')}\n" f"解析意图: {player_intent.get('intent', 'UNKNOWN')}\n" f"目标: {player_intent.get('target', '无')}\n" f"细节: {player_intent.get('details', '无')}" ) messages = [ {"role": "system", "content": system_prompt}, {"role": "user", "content": user_message}, ] full_text = "" story_started = False story_ended = False try: for chunk in call_qwen_stream(messages, model=self.model, temperature=0.7, max_tokens=3000): full_text += chunk # 使用标准化标记检测，处理 LLM 输出的常见变体格式 normalized = _normalize_markers(full_text) # 跳过 THINKING 区域，从 STORY_TEXT 开始流式输出 if not story_started: if "---STORY_TEXT---" in normalized: story_started = True idx = normalized.index("---STORY_TEXT---") + len("---STORY_TEXT---") current_story = normalized[idx:] # 检查是否已经到了 STATE_JSON if "---STATE_JSON---" in current_story: story_ended = True current_story = current_story[:current_story.index("---STATE_JSON---")] display_text = self._clean_story_text(current_story) if display_text.strip(): yield {"type": "story_chunk", "text": display_text.strip()} elif not story_ended: idx = normalized.index("---STORY_TEXT---") + len("---STORY_TEXT---") current_story = normalized[idx:] if "---STATE_JSON---" in current_story: story_ended = True current_story = current_story[:current_story.index("---STATE_JSON---")] display_text = self._clean_story_text(current_story) if display_text.strip(): yield {"type": "story_chunk", "text": display_text.strip()} except Exception as e: logger.error(f"流式合并生成失败: {e}，降级为非流式两阶段") try: result = self.generate_story(player_intent) result["options"] = self._validate_options(result.get("options", [])) yield {"type": "final", **result} except Exception: fallback = self._fallback_response( player_intent, tick_log, fallback_reason="stream_exception", engine_mode="stream_merged", ) fallback["options"] = self._validate_options(fallback.get("options", [])) yield {"type": "final", **fallback} return # ★ 如果流式阶段未检测到标记但有累积文本，先 yield 给 UI 显示 if not story_started and full_text.strip(): logger.warning("[流式合并] 未检测到 STORY_TEXT 标记，直接使用累积文本") display = self._clean_story_text(full_text.strip()) display = re.sub(r'^(好的[，,]?\s*(这是|以下是|下面是).*?[：:]\s*)', '', display, flags=re.MULTILINE) if display.strip(): yield {"type": "story_chunk", "text": display.strip()} # 解析合并响应 outline, story_text, options = self._parse_merged_response(full_text) extracted_location = "未知之地" if isinstance(outline, dict): extracted_location = str(outline.get("location") or "").strip() or "未知之地" if outline is None: # ★ outline 为空不代表 story_text 也为空！ # 如果 story_text 已成功提取，继续走正常流程而非丢弃 if story_text and story_text.strip(): logger.warning("大纲(STATE_JSON)解析失败，但故事文本已提取，跳过状态更新继续") options = self._validate_options(options) yield { "type": "final", "story_text": story_text, "options": options, "location": extracted_location, "state_changes": {}, "change_log": tick_log + ["（系统提示：本回合状态解析失败，未更新状态）"], "outline": None, "consistency_issues": [], "telemetry": _build_telemetry( engine_mode="stream_merged", used_fallback=True, fallback_reason="state_parse_failed", ), } return else: logger.error("合并响应解析完全失败，使用降级") fallback = self._fallback_response( player_intent, tick_log, fallback_reason="merged_response_parse_failed", engine_mode="stream_merged", ) yield {"type": "final", **fallback} return # 处理时间冲突 state_changes = outline.get("state_changes", {}) if state_changes.get("time_change"): tick_log = [line for line in tick_log if not line.startswith("时间流逝:")] # 一致性检查 consistency_issues = self.game_state.check_consistency(state_changes) if consistency_issues: logger.warning(f"发现一致性问题: {consistency_issues}") # 移除与非法物品相关的状态变更（安全网） state_changes = self._strip_invalid_item_effects(state_changes) # 清理状态变更 event_type = outline.get("event_type", "") state_changes, sanitize_warnings = self._sanitize_state_changes(state_changes, event_type) state_changes, item_rule_notes = self._fill_missing_consumable_effects(state_changes, player_intent) state_changes, rest_rule_notes = self._fill_missing_rest_effects(state_changes, player_intent) # 应用状态变更 change_log = self.game_state.apply_changes(state_changes) # 校验状态合法性 is_valid, validation_issues = self.game_state.validate() # 记录事件 self.game_state.log_event( event_type=outline.get("event_type", "UNKNOWN"), description=outline.get("event_summary", ""), player_action=player_intent.get("raw_input", ""), involved_npcs=outline.get("involved_npcs", []), state_changes=state_changes, consequence_tags=outline.get("consequence_tags", []), is_reversible=outline.get("is_reversible", True), ) # 检查游戏结束 if self.game_state.is_game_over(): death_result = self._generate_death_narrative() yield {"type": "final", **death_result} return # 验证选项 options = self._validate_options(options) # 合并日志 merged_log = _merge_change_logs( tick_log, change_log + validation_issues + item_rule_notes + rest_rule_notes, ) yield { "type": "final", "story_text": story_text, "options": options, "location": extracted_location, "state_changes": state_changes, "change_log": merged_log, "outline": outline, "consistency_issues": consistency_issues, "telemetry": _build_telemetry( engine_mode="stream_merged", used_fallback=False, consistency_issues_count=len(consistency_issues), validation_issues_count=len(validation_issues), ), } def process_option_selection_stream(self, option: dict): """ 流式处理玩家点击选项。 将选项转化为意图格式，然后调用 generate_story_stream。 """ intent = { "intent": option.get("action_type", "EXPLORE"), "target": option.get("target"), "details": option.get("text", ""), "raw_input": option.get("text", ""), } yield from self.generate_story_stream(intent) def _parse_merged_response(self, raw_text: str) -> tuple: """ 解析合并格式的响应（THINKING + STORY_TEXT + STATE_JSON + OPTIONS_JSON）。 Returns: (outline_dict, story_text, options_list) outline_dict 可能为 None（解析失败时） """ # ★ 调试日志：记录原始 API 返回内容 logger.debug(f"原始 API 返回内容（合并模式）: {raw_text}") outline = None story_text = "" options = [] # 标准化标记格式 normalized = _normalize_markers(raw_text) # 解析 STORY_TEXT if "---STORY_TEXT---" in normalized: story_start = normalized.index("---STORY_TEXT---") + len("---STORY_TEXT---") # STORY_TEXT 的结尾可能是 STATE_JSON 或 OPTIONS_JSON story_end = len(normalized) if "---STATE_JSON---" in normalized: story_end = normalized.index("---STATE_JSON---") elif "---OPTIONS_JSON---" in normalized: story_end = normalized.index("---OPTIONS_JSON---") story_text = normalized[story_start:story_end].strip() # 解析 STATE_JSON if "---STATE_JSON---" in normalized: state_start = normalized.index("---STATE_JSON---") + len("---STATE_JSON---") state_end = len(normalized) if "---OPTIONS_JSON---" in normalized: state_end = normalized.index("---OPTIONS_JSON---") state_part = normalized[state_start:state_end].strip() outline = extract_json_from_text(state_part) # 解析 OPTIONS_JSON if "---OPTIONS_JSON---" in normalized: options_start = normalized.index("---OPTIONS_JSON---") + len("---OPTIONS_JSON---") options_part = normalized[options_start:].strip() parsed = extract_json_from_text(options_part) if isinstance(parsed, list): options = parsed if not options: options = self._generate_default_options() options = self._ensure_option_count(options) if not story_text and outline is None: # 完全解析失败 → 使用暴力提取策略 logger.warning("合并格式解析完全失败，使用暴力提取策略") # 先移除 THINKING 区域 cleaned = re.sub(r'---THINKING---.*?(?=---[A-Z]|$)', '', raw_text, flags=re.DOTALL | re.IGNORECASE).strip() if not cleaned: cleaned = raw_text story_text, options_fallback = self._brute_force_extract(cleaned) if options_fallback: options = self._ensure_option_count(options_fallback) elif not story_text.strip() and raw_text.strip(): # story_text 为空但有原始文本 → 暴力提取 logger.warning("story_text 为空，使用暴力提取") cleaned = re.sub(r'---THINKING---.*?(?=---[A-Z]|$)', '', raw_text, flags=re.DOTALL | re.IGNORECASE).strip() if not cleaned: cleaned = raw_text story_text, options_fallback = self._brute_force_extract(cleaned) if options_fallback: options = self._ensure_option_count(options_fallback) # 清理 story_text 中可能残留的 JSON 和标记 story_text = self._clean_story_text(story_text) # 移除常见 AI 前缀 story_text = re.sub(r'^(好的[，,]?\s*(这是|以下是|下面是).*?[：:]\s*)', '', story_text, flags=re.MULTILINE) # 确保 story_text 不为空 if not story_text.strip(): story_text = "你环顾四周，思考着接下来该做什么..." # 确保选项 if not options: options = self._generate_default_options() options = self._ensure_option_count(options) return outline, story_text, options