""" state_manager.py - StoryWeaver 状态管理器 职责： 1. 定义游戏世界的完整数据模型（Pydantic BaseModel） 2. 维护全局状态的唯一真相来源 (Single Source of Truth) 3. 提供状态变更、校验、一致性检查、序列化等核心方法 4. 记录事件日志用于一致性维护 设计思路： - 所有数据结构使用 Pydantic BaseModel，天然支持 JSON 序列化/反序列化 - GameState 是顶层容器，包含 PlayerState、WorldState、EventLog - event_log 是一致性维护的灵魂：每次操作都记录快照，用于矛盾检测 - to_prompt() 方法将结构化数据转为自然语言，注入 LLM 的 System Prompt """ from __future__ import annotations import copy import logging import random import re from typing import Any, Optional from pydantic import BaseModel, Field from demo_rules import OVERNIGHT_REST_LOCATIONS, action_time_cost_minutes, build_scene_actions from utils import clamp logger = logging.getLogger("StoryWeaver") # ============================================================ # 辅助数据模型 # ============================================================ class StatusEffect(BaseModel): """ 状态效果模型（Buff / Debuff） 设计思路： - 每个状态效果有持续时间和属性修正，每回合自动结算 - source 记录来源，便于在 Prompt 中说明"你身上中了哥布林的毒" - stackable 控制是否可叠加，防止无限叠加 bug """ name: str # 效果名（中毒、祝福、隐身…） effect_type: str = "debuff" # buff / debuff / neutral stat_modifiers: dict[str, int] = Field(default_factory=dict) # 属性修正 {"attack": -3, "defense": +2} duration: int = 3 # 剩余持续回合数（-1 = 永久） description: str = "" # 效果描述 source: str = "" # 来源（"哥布林的毒刃"） stackable: bool = False # 是否可叠加 class ItemInfo(BaseModel): """ 物品详情模型 设计思路： - item_type 区分装备/消耗品/任务道具等，不同类型有不同交互逻辑 - rarity 影响掉落概率和商店价格 - quest_related 标记任务道具，防止玩家丢弃关键物品 - lore_text 提供物品背景，丰富生成文本的细节 """ name: str # 物品名称 item_type: str = "misc" # weapon / armor / consumable / quest_item / material / key / misc description: str = "" # 物品描述 rarity: str = "common" # common / uncommon / rare / epic / legendary stat_bonus: dict[str, int] = Field(default_factory=dict) # 装备时属性加成 {"attack": +5} usable: bool = False # 是否可主动使用 use_effect: str = "" # 使用效果描述（如"恢复 30 HP"） value: int = 0 # 商店价值（金币） quest_related: bool = False # 是否为任务道具 lore_text: str = "" # 物品背景故事 class NPCState(BaseModel): """ NPC 状态模型 设计思路： - npc_type 决定交互方式（商人可交易、任务NPC可接任务、敌人可战斗） - memory 是一致性维护的关键：NPC"记住"与玩家的互动历史 - schedule 模拟 NPC 日常行为，不同时间段出现在不同地点 - relationship_level 影响对话态度和任务可用性 """ name: str # NPC 名称 npc_type: str = "civilian" # civilian / merchant / quest_giver / enemy / companion / boss location: str = "" # 所在地点 attitude: str = "neutral" # friendly / neutral / cautious / hostile is_alive: bool = True # 是否存活 description: str = "" # 外观描述 race: str = "人类" # 种族 occupation: str = "" # 职业（铁匠、旅店老板、守卫…） faction: str = "" # 所属阵营 # --- 交互相关 --- relationship_level: int = 0 # 与玩家好感度（-100 ~ 100） dialogue_tags: list[str] = Field(default_factory=list) # 已触发的对话标签（防止重复触发） can_trade: bool = False # 是否可交易 shop_inventory: list[str] = Field(default_factory=list) # 商店物品（如果是商人） can_give_quest: bool = False # 是否可发布任务 available_quests: list[str] = Field(default_factory=list) # 可发布的任务 ID # --- 战斗相关（敌人/Boss） --- hp: int = 0 max_hp: int = 0 attack: int = 0 defense: int = 0 loot_table: list[str] = Field(default_factory=list) # 击败后掉落物品 weakness: str = "" # 弱点（火、光…） special_ability: str = "" # 特殊能力 # --- 记忆与行为 --- memory: list[str] = Field(default_factory=list) # NPC 记住的关键事件 schedule: dict[str, str] = Field(default_factory=dict) # 时间行为表 {"清晨": "在市场摆摊"} backstory: str = "" # 背景故事 class QuestRewards(BaseModel): """ 任务奖励模型 设计思路： - 奖励类型丰富，覆盖经济、声望、技能、称号等多维度 - 每种奖励都可选，通过组合实现多样化的奖励体验 """ gold: int = 0 # 金币奖励 experience: int = 0 # 经验值奖励 items: list[str] = Field(default_factory=list) # 奖励物品 reputation_changes: dict[str, int] = Field(default_factory=dict) # 声望变化 {"精灵族": +10} karma_change: int = 0 # 善恶值变化 unlock_location: str = "" # 解锁新地点 unlock_skill: str = "" # 解锁新技能 title: str = "" # 解锁称号 class QuestState(BaseModel): """ 任务状态模型 设计思路： - quest_type 区分主线/支线/隐藏任务，影响 UI 展示和优先级 - objectives 是任务子目标字典，每个子目标独立追踪 - branching_choices 支持任务内分支（如"放走囚犯"导向不同结局） - time_limit / turns_remaining 支持限时任务机制 - prerequisites 保证任务链的逻辑顺序 """ quest_id: str # 任务唯一 ID title: str # 任务名称 description: str # 任务描述 quest_type: str = "main" # main / side / hidden / daily status: str = "active" # active / completed / failed / expired giver_npc: str = "" # 任务发布者 NPC # --- 目标 --- objectives: dict[str, bool] = Field(default_factory=dict) # 子目标 {"找到钥匙": False, "打开宝箱": False} # --- 奖励 --- rewards: QuestRewards = Field(default_factory=QuestRewards) # --- 约束 --- time_limit: int = -1 # 限时回合数（-1 = 无限） turns_remaining: int = -1 # 剩余回合数 prerequisites: list[str] = Field(default_factory=list) # 前置任务 ID level_requirement: int = 1 # 等级要求 karma_requirement: Optional[int] = None # 善恶值要求 # --- 分支 --- branching_choices: dict[str, str] = Field(default_factory=dict) # 关键选择 {"放走囚犯": "mercy_path"} chosen_path: str = "" # 已选择的路线 consequences: list[str] = Field(default_factory=list) # 完成后的剧情后果描述 class LocationInfo(BaseModel): """ 地点详情模型 设计思路： - connected_to 构成游戏地图的拓扑结构，控制玩家可移动范围 - danger_level 影响遭遇概率和 NPC 行为 - is_accessible + required_item 实现"锁门/钥匙"机制 - ambient_description 用于丰富 LLM 生成的场景描写 - special_events 支持地点触发式事件 """ name: str # 地点名称 location_type: str = "town" # town / dungeon / wilderness / shop / special description: str = "" # 地点描述 connected_to: list[str] = Field(default_factory=list) # 可前往的相邻地点 npcs_present: list[str] = Field(default_factory=list) # 当前在该地点的 NPC available_items: list[str] = Field(default_factory=list) # 可拾取/发现的物品 enemies: list[str] = Field(default_factory=list) # 可能遭遇的敌人 danger_level: int = 0 # 危险等级 (0=安全, 10=极度危险) weather: str = "晴朗" # 当前天气 is_discovered: bool = False # 是否已被玩家发现 is_accessible: bool = True # 是否可进入 required_item: str = "" # 进入所需道具 ambient_description: str = "" # 环境氛围描述 special_events: list[str] = Field(default_factory=list) # 该地点可触发的特殊事件 rest_available: bool = False # 是否可以休息恢复 shop_available: bool = False # 是否有商店 # ============================================================ # 玩家状态 # ============================================================ class PlayerState(BaseModel): """ 玩家角色状态（RPG 核心属性） 设计思路： - 基础属性 + 战斗属性 + 装备栏 + 社交属性 构成完整的角色模型 - reputation / karma / relationships 影响 NPC 态度和剧情分支 - morale / sanity / hunger 增加生存维度，丰富游戏体验 - known_lore 记录玩家获得的情报，影响可用对话选项 - death_count 支持"轮回"类剧情彩蛋 """ # --- 基础属性 --- name: str = "旅人" # 玩家名称 title: str = "无名冒险者" # 称号（随剧情解锁） level: int = 1 # 等级 experience: int = 0 # 当前经验值 exp_to_next_level: int = 100 # 升级所需经验 # --- 战斗属性 --- hp: int = 100 # 当前生命值 max_hp: int = 100 # 最大生命值 mp: int = 50 # 魔力值 max_mp: int = 50 # 最大魔力值 attack: int = 10 # 攻击力 defense: int = 5 # 防御力 attack_power: int = 10 # 实战攻击力（基础攻击+装备加成） defense_power: int = 5 # 实战防御力（基础防御+装备加成） stamina: int = 100 # 体力值 max_stamina: int = 100 # 最大体力值 speed: int = 8 # 速度（影响行动顺序） luck: int = 5 # 幸运（影响暴击、掉落） perception: int = 5 # 感知（影响探索发现、陷阱识别） # --- 装备栏 --- equipment: dict[str, Optional[str]] = Field(default_factory=lambda: { "weapon": None, # 武器 "armor": None, # 护甲 "accessory": None, # 饰品 "helmet": None, # 头盔 "boots": None, # 靴子 }) # --- 状态 --- location: str = "村庄" # 当前所在地点 inventory: list[str] = Field(default_factory=list) # 背包物品列表 skills: list[str] = Field(default_factory=list) # 已习得技能列表 status_effects: list[StatusEffect] = Field(default_factory=list) # 状态效果列表 gold: int = 50 # 金币 reputation: dict[str, int] = Field(default_factory=dict) # 阵营声望 {"精灵族": 10} morale: int = 100 # 士气（0=崩溃, 100=高昂） sanity: int = 100 # 理智值（探索黑暗区域消耗） hunger: int = 100 # 饱食度（0=饥饿惩罚） karma: int = 0 # 善恶值（正=善, 负=恶） known_lore: list[str] = Field(default_factory=list) # 已知传说/情报片段 relationships: dict[str, int] = Field(default_factory=dict) # 与特定 NPC 的好感度 death_count: int = 0 # 累计死亡次数 # ============================================================ # 世界状态 # ============================================================ class WorldState(BaseModel): """ 世界状态容器 设计思路： - 包含所有非玩家的世界数据：地图、NPC、任务、物品注册表 - time_of_day + day_count + weather + season 构成动态环境系统 - global_flags 是灵活的剧情标记系统，支持分支判断 - rumors / active_threats 丰富 NPC 对话内容 - faction_relations 支持阵营间动态关系 """ current_scene: str = "村庄广场" # 当前场景名称 time_of_day: str = "清晨" # 清晨 / 上午 / 正午 / 下午 / 黄昏 / 夜晚 / 深夜 day_count: int = 1 # 当前天数 weather: str = "晴朗" # 晴朗 / 多云 / 小雨 / 暴风雨 / 大雪 / 浓雾 light_level: str = "明亮" # 明亮 / 柔和 / 昏暗 / 幽暗 / 漆黑 time_progress_units: int = 0 # 当前时段内累积的动作耗时点数 last_weather_change_minutes: int = -999999 # 上次天气变化时的累计分钟数 season: str = "春" # 春 / 夏 / 秋 / 冬 # --- 地图 --- locations: dict[str, LocationInfo] = Field(default_factory=dict) discovered_locations: list[str] = Field(default_factory=list) # --- NPC --- npcs: dict[str, NPCState] = Field(default_factory=dict) # --- 任务 --- quests: dict[str, QuestState] = Field(default_factory=dict) # --- 物品注册表 --- item_registry: dict[str, ItemInfo] = Field(default_factory=dict) # --- 全局标记 --- global_flags: dict[str, bool] = Field(default_factory=dict) world_events: list[str] = Field(default_factory=list) # 已发生的全局事件 recent_environment_events: list["EnvironmentEvent"] = Field(default_factory=list) active_threats: list[str] = Field(default_factory=list) # 当前全局威胁 rumors: list[str] = Field(default_factory=list) # 流传的传闻 faction_relations: dict[str, dict[str, str]] = Field(default_factory=dict) # 阵营间关系 # ============================================================ # 事件日志 # ============================================================ class GameEvent(BaseModel): """ 事件日志模型（一致性维护的关键） 设计思路： - 每次状态变更都记录为一个事件，包含完整的上下文信息 - state_changes 记录该事件引发的状态变更快照 - consequence_tags 用于后续一致性检查（如 "killed_goblin_king"） - is_reversible 标记不可逆事件，LLM 生成时需特别注意 - involved_npcs + location 便于按维度检索历史事件 """ turn: int # 发生在第几回合 day: int = 1 # 发生在第几天 time_of_day: str = "" # 发生时的时段 event_type: str = "" # COMBAT / DIALOGUE / MOVE / ITEM / QUEST / TRADE / REST / DISCOVERY / DEATH / LEVEL_UP description: str = "" # 事件简述 location: str = "" # 事件发生地点 involved_npcs: list[str] = Field(default_factory=list) # 涉及的 NPC state_changes: dict = Field(default_factory=dict) # 状态变更快照 player_action: str = "" # 触发该事件的玩家操作 consequence_tags: list[str] = Field(default_factory=list) # 后果标签 is_reversible: bool = True # 是否可逆 class EnvironmentEvent(BaseModel): """Structured environment event used by UI, logs, and prompt injection.""" event_id: str category: str = "environment" # weather / light / environment title: str = "" description: str = "" location: str = "" time_of_day: str = "" weather: str = "" light_level: str = "" severity: str = "low" # low / medium / high state_changes: dict[str, Any] = Field(default_factory=dict) prompt_hint: str = "" WorldState.model_rebuild() # ============================================================ # 游戏主控类 # ============================================================ class GameState: """ 游戏全局状态管理器 —— 项目的灵魂 职责： 1. 持有并管理 PlayerState、WorldState、EventLog 2. 提供状态变更、校验、一致性检查的统一入口 3. 将结构化状态序列化为自然语言 Prompt 4. 每回合自动结算状态效果、时间推进、任务超时等 核心设计原则： - 所有状态修改必须通过 apply_changes() 进入 - 每次修改都伴随 validate() 校验和 log_event() 记录 - check_consistency() 在生成前检测可能的矛盾 """ def __init__(self, player_name: str = "旅人"): """初始化游戏状态，创建默认的起始世界""" self.player = PlayerState(name=player_name) self.world = WorldState() self.event_log: list[GameEvent] = [] self.turn: int = 0 self.game_mode: str = "exploration" # exploration / combat / dialogue / cutscene / game_over self.difficulty: str = "normal" # easy / normal / hard self.story_arc: str = "序章" # 当前故事章节 self.ending_flags: dict[str, bool] = {} # 结局条件追踪 self.combat_log: list[str] = [] # 最近战斗记录 self.achievement_list: list[str] = [] # 已解锁成就 self.elapsed_minutes_total: int = 0 self.last_recent_gain: str | None = None self.last_interacted_npc: str | None = None # 初始化起始世界 self._init_starting_world() self.refresh_combat_stats() # 純文本地图渲染使用的“当前位置 + 足迹历史” # current_location 必须始终与 self.player.location 保持一致。 self.current_location: str = str(self.player.location) self.location_history: list[str] = [] self.world.time_progress_units = 36 self.pending_environment_event: EnvironmentEvent | None = None self._sync_world_clock() self.world.light_level = self._determine_light_level() def _init_starting_world(self): """ 创建游戏的起始世界设定。 包含初始地点、NPC、任务和物品，为故事提供起点。 """ # --- 初始地点 --- self.world.locations = { "村庄广场": LocationInfo( name="村庄广场", location_type="town", description="一个宁静的小村庄中心广场，阳光温暖地照耀着鹅卵石路面。周围有几家商铺和一口古老的水井。", connected_to=["村庄铁匠铺", "村庄旅店", "村口小路", "村庄杂货铺"], npcs_present=["村长老伯"], danger_level=0, is_discovered=True, rest_available=False, ambient_description="阳光斑驳地洒在广场上，远处传来铁匠铺叮叮当当的锤声。", ), "村庄铁匠铺": LocationInfo( name="村庄铁匠铺", location_type="shop", description="一间热气腾腾的铁匠铺，炉火正旺。墙上挂满了各式武器和护甲。", connected_to=["村庄广场"], npcs_present=["铁匠格林"], danger_level=0, is_discovered=True, shop_available=True, ambient_description="炉火映红了铁匠粗犷的脸庞，空气中弥漫着金属和碳的气味。", ), "村庄旅店": LocationInfo( name="村庄旅店", location_type="shop", description="一家温馨的小旅店，空气中弥漫着烤肉和麦酒的香气。壁炉里的火焰跳动着。", connected_to=["村庄广场"], npcs_present=["旅店老板娘莉娜"], danger_level=0, is_discovered=True, rest_available=True, shop_available=True, ambient_description="壁炉噼啪作响，几位旅客正低声交谈，空气温暖而舒适。", ), "村庄杂货铺": LocationInfo( name="村庄杂货铺", location_type="shop", description="一家琳琅满目的杂货铺，从草药到绳索应有尽有。老板是个精明的商人。", connected_to=["村庄广场"], npcs_present=["杂货商人阿尔"], danger_level=0, is_discovered=True, shop_available=True, ambient_description="货架上摆满了稀奇古怪的商品，柜台后的商人正用算盘噼里啪啦地算账。", ), "村口小路": LocationInfo( name="村口小路", location_type="wilderness", description="通往村外的一条泥泞小路，两旁长满了野草。远处隐约可见黑暗森林的轮廓。", connected_to=["村庄广场", "黑暗森林入口"], danger_level=2, is_discovered=True, ambient_description="微风拂过野草，远处的森林在薄雾中若隐若现，传来不知名鸟兽的叫声。", ), "黑暗森林入口": LocationInfo( name="黑暗森林入口", location_type="wilderness", description="森林的入口处，参天大树遮蔽了阳光，地面覆盖着厚厚的落叶。一股不祥的气息扑面而来。", connected_to=["村口小路", "森林深处", "溪边营地"], enemies=["哥布林", "野狼"], danger_level=4, is_discovered=False, ambient_description="树冠密集得几乎遮蔽了所有阳光，偶尔传来树枝折断的声音，不知道是风还是别的什么。", ), "森林深处": LocationInfo( name="森林深处", location_type="dungeon", description="森林的最深处，古树盘根错节。空气中弥漫着腐朽和魔力的气息，据说这里住着森林的主人。", connected_to=["黑暗森林入口"], enemies=["哥布林巫师", "巨型蜘蛛", "森林巨魔"], danger_level=7, is_discovered=False, is_accessible=True, ambient_description="黑暗几乎吞噬了一切，只有奇异的荧光苔藓发出微弱的光。远处传来低沉的咆哮。", ), "溪边营地": LocationInfo( name="溪边营地", location_type="wilderness", description="森林中一处难得的开阔地带，一条清澈的小溪从旁流过。适合扎营休息。", connected_to=["黑暗森林入口"], danger_level=2, is_discovered=False, rest_available=True, ambient_description="溪水潺潺流淌，偶有鸟鸣声在林间回荡，这里是森林中的一片安宁绿洲。", ), # -------- 扩展地点 -------- "河边渡口": LocationInfo( name="河边渡口", location_type="wilderness", description="一座破旧的木制渡口，宽阔的河流在此缓缓流淌。一艘半沉的渡船拴在码头桩上。", connected_to=["村口小路", "废弃矿洞入口", "山麓盗贼营"], npcs_present=["渡口老渔夫"], danger_level=3, is_discovered=False, ambient_description="河水拍打着朽烂的木桩，远处有鹰在盘旋，对岸隐约可见矿洞的轮廓。", ), "废弃矿洞入口": LocationInfo( name="废弃矿洞入口", location_type="dungeon", description="荒废多年的铁矿洞，入口被蛛网和碎石半堵。矿道里传来金属碰撞的回声。", connected_to=["河边渡口", "矿洞深层"], enemies=["骷髅兵", "矿洞蝙蝠群", "锈铁傀儡"], danger_level=5, is_discovered=False, ambient_description="腐朽的矿车轨道延伸向黑暗深处，空气里弥漫着铁锈和硫磺的气味。", ), "矿洞深层": LocationInfo( name="矿洞深层", location_type="dungeon", description="矿洞最深处，一个巨大的地下空间。墙壁上嵌着发光的矿石，中央有一座被遗忘的祭坛。", connected_to=["废弃矿洞入口"], enemies=["亡灵矿工", "岩石巨像"], danger_level=8, is_discovered=False, is_accessible=False, required_item="矿工旧钥匙", ambient_description="发光矿石将洞穴映成幽蓝色，祭坛上刻着无人能读的文字，隐约有低沉的嗡鸣。", ), "山麓盗贼营": LocationInfo( name="山麓盗贼营", location_type="wilderness", description="藏在山脚灌木丛后的盗贼据点，几顶破帐篷围着一堆余烬。看起来已被匆忙弃置。", connected_to=["河边渡口", "精灵遗迹"], enemies=["盗贼斥候", "盗贼头目"], danger_level=5, is_discovered=False, ambient_description="地上散落着翻倒的酒桶和吃了一半的干粮，有人走得很匆忙。", ), "精灵遗迹": LocationInfo( name="精灵遗迹", location_type="special", description="一片被藤蔓覆盖的古老石柱林，精灵文字在月光下隐约发光。空气中有淡淡的魔力涌动。", connected_to=["山麓盗贼营"], npcs_present=["遗迹守护者"], danger_level=4, is_discovered=False, ambient_description="石柱上的符文随风明灭，仿佛在回应某种古老的感召。脚下的青苔异常柔软。", ), "古塔废墟": LocationInfo( name="古塔废墟", location_type="dungeon", description="一座半坍塌的石塔，据说曾是某位法师的研究所。顶层似乎还有东西在闪烁。", connected_to=["村口小路"], enemies=["石像鬼", "游荡幽灵"], danger_level=6, is_discovered=False, ambient_description="风从塔身的裂缝中呼啸而过，残破的阶梯上覆满了青苔和鸟粪。", ), } self.world.discovered_locations = ["村庄广场", "村庄铁匠铺", "村庄旅店", "村庄杂货铺", "村口小路"] # 扩展村口小路的连接 —— 链接到新区域 self.world.locations["村口小路"].connected_to = [ "村庄广场", "黑暗森林入口", "河边渡口", "古塔废墟", ] # --- 初始 NPC --- self.world.npcs = { "村长老伯": NPCState( name="村长老伯", npc_type="quest_giver", location="村庄广场", attitude="friendly", description="一位白发苍苍但精神矍铄的老人，是这个村庄的领导者。他的眼中带着忧虑。", race="人类", occupation="村长", relationship_level=20, can_give_quest=True, available_quests=["main_quest_01"], memory=[], schedule={"清晨": "村庄广场", "上午": "村庄广场", "正午": "村庄广场", "下午": "村庄广场", "黄昏": "村庄广场", "夜晚": "村庄旅店"}, backstory="在这个村庄生活了七十年的老者，见证过上一次暗潮来袭，深知森林中潜伏的危险。", ), "铁匠格林": NPCState( name="铁匠格林", npc_type="merchant", location="村庄铁匠铺", attitude="neutral", description="一个肌肉发达的中年矮人，手臂上布满烧伤痕迹。沉默寡言但手艺精湛。", race="矮人", occupation="铁匠", can_trade=True, shop_inventory=["小刀", "短剑", "铁剑", "皮甲", "木盾"], relationship_level=0, schedule={"清晨": "村庄铁匠铺", "上午": "村庄铁匠铺", "正午": "村庄铁匠铺", "下午": "村庄铁匠铺", "黄昏": "村庄铁匠铺", "夜晚": "村庄旅店"}, backstory="曾经是王都的御用铁匠，因一场变故隐居此地。对远方的怪物有独到的了解。", ), "旅店老板娘莉娜": NPCState( name="旅店老板娘莉娜", npc_type="merchant", location="村庄旅店", attitude="friendly", description="一位热情开朗的红发女子，笑容温暖。她的旅店是村里情报的集散地。", race="人类", occupation="旅店老板", can_trade=True, shop_inventory=["面包", "烤肉", "麦酒", "草药包"], relationship_level=10, schedule={"清晨": "村庄旅店", "上午": "村庄旅店", "正午": "村庄旅店", "下午": "村庄旅店", "黄昏": "村庄旅店", "夜晚": "村庄旅店"}, backstory="年轻时曾是一名冒险者，后来受伤退役经营旅店。对旅行者总是格外关照。", ), "杂货商人阿尔": NPCState( name="杂货商人阿尔", npc_type="merchant", location="村庄杂货铺", attitude="neutral", description="一个精明的瘦长男子，鹰钩鼻上架着一副圆框眼镜。善于讨价还价。", race="人类", occupation="商人", can_trade=True, shop_inventory=["火把", "绳索", "解毒药水", "小型治疗药水"], relationship_level=-5, schedule={"清晨": "村庄杂货铺", "上午": "村庄杂货铺", "正午": "村庄广场", "下午": "村庄杂货铺", "黄昏": "村庄杂货铺", "夜晚": "村庄杂货铺"}, backstory="来自远方的行商，在村中定居多年。对各地的传闻消息灵通，但消息总是要收费的。", ), "神秘旅人": NPCState( name="神秘旅人", npc_type="quest_giver", location="村庄旅店", attitude="cautious", description="一个身披灰色斗篷的旅人，面容隐藏在兜帽之下，只露出锐利的双眼。", race="未知", occupation="旅人", relationship_level=-10, can_give_quest=True, available_quests=["side_quest_01"], memory=[], schedule={"清晨": "村庄旅店", "夜晚": "村口小路"}, backstory="似乎在寻找什么。偶尔从斗篷下露出的手指上有奇异的魔法纹路。", ), # -------- 扩展 NPC -------- "渡口老渔夫": NPCState( name="渡口老渔夫", npc_type="quest_giver", location="河边渡口", attitude="friendly", description="一个皮肤黝黑、满脸皱纹的老人，正坐在码头上修补渔网。", race="人类", occupation="渔夫", relationship_level=5, can_give_quest=True, available_quests=["side_quest_02"], memory=[], schedule={"清晨": "河边渡口", "上午": "河边渡口", "正午": "河边渡口", "下午": "河边渡口", "黄昏": "河边渡口", "夜晚": "村庄旅店"}, backstory="在这条河边住了四十年，对河流两岸的地形了如指掌。最近总念叨对岸矿洞里的怪响。", ), "遗迹守护者": NPCState( name="遗迹守护者", npc_type="quest_giver", location="精灵遗迹", attitude="cautious", description="一个身形消瘦的半精灵，穿着褪色的绿袍，眼神中有深深的疲惫。", race="半精灵", occupation="守护者", relationship_level=-5, can_give_quest=True, available_quests=["side_quest_03"], memory=[], schedule={"清晨": "精灵遗迹", "正午": "精灵遗迹", "夜晚": "精灵遗迹"}, backstory="最后一位遗迹守护者，独自守护这片先祖的圣地已有三十年。对外来者充满警惕，但内心渴望帮助。", ), } # --- 初始任务 --- self.world.quests = { "main_quest_01": QuestState( quest_id="main_quest_01", title="森林中的阴影", description="村长老伯告诉你，最近森林中频繁出现怪物袭击事件。他请求你前往调查。", quest_type="main", status="active", giver_npc="村长老伯", objectives={ "与村长对话了解情况": False, "前往黑暗森林入口调查": False, "击败森林中的怪物": False, "调查怪物活动的原因": False, "与村长老伯对话汇报发现": False, }, rewards=QuestRewards( gold=100, experience=50, items=["森林之钥"], reputation_changes={"村庄": 20}, karma_change=5, ), ), "main_quest_02": QuestState( quest_id="main_quest_02", title="森林深处的咆哮", description="村长老伯确认森林巨魔已经苏醒，并命你持森林之钥深入黑暗森林，将这头怪物彻底斩杀。", quest_type="main", status="inactive", giver_npc="村长老伯", objectives={ "前往森林深处": False, "击败森林巨魔": False, }, rewards=QuestRewards( gold=0, experience=90, reputation_changes={"村庄": 30}, karma_change=10, ), prerequisites=["main_quest_01"], ), "side_quest_01": QuestState( quest_id="side_quest_01", title="失落的传承", description="神秘旅人似乎在寻找一件古老的遗物。也许帮助他能得到意想不到的回报。", quest_type="side", status="inactive", giver_npc="神秘旅人", objectives={ "与神秘旅人交谈": False, "找到古老遗物的线索": False, }, rewards=QuestRewards( experience=30, items=["神秘卷轴"], unlock_skill="暗影感知", ), prerequisites=[], ), # -------- 扩展任务 -------- "side_quest_02": QuestState( quest_id="side_quest_02", title="河底的秘密", description="渡口老渔夫说他最近总在河里捞到奇怪的骨头，而且对岸矿洞方向夜里总是有光。他请你去查明真相。", quest_type="side", status="inactive", giver_npc="渡口老渔夫", objectives={ "与渡口老渔夫交谈": False, "前往废弃矿洞调查": False, "找到矿洞异常的原因": False, }, rewards=QuestRewards( gold=60, experience=40, items=["矿工旧钥匙"], reputation_changes={"村庄": 10}, ), prerequisites=[], ), "side_quest_03": QuestState( quest_id="side_quest_03", title="守护者的试炼", description="精灵遗迹的守护者提出一个交换条件：通过她设下的试炼，就能获得古老的精灵祝福。", quest_type="side", status="inactive", giver_npc="遗迹守护者", objectives={ "与遗迹守护者交谈": False, "通过守护者的试炼": False, }, rewards=QuestRewards( experience=50, unlock_skill="精灵祝福", karma_change=10, title="遗迹认可者", ), prerequisites=[], ), } # --- 初始物品注册表 --- self.world.item_registry = { "小刀": ItemInfo(name="小刀", item_type="weapon", description="一把朴素但实用的小刀，便于近身防身。", rarity="common", stat_bonus={"attack": 2}, value=5), "短剑": ItemInfo(name="短剑", item_type="weapon", description="一把适合新手携带的短剑，轻便易用。", rarity="common", stat_bonus={"attack": 3}, value=10), "铁剑": ItemInfo(name="铁剑", item_type="weapon", description="一把标准的铁制长剑，刀锋锐利。", rarity="common", stat_bonus={"attack": 5}, value=30), "皮甲": ItemInfo(name="皮甲", item_type="armor", description="硬化皮革制成的轻甲，兼顾防护与灵活性。", rarity="common", stat_bonus={"defense": 3}, value=25), "木盾": ItemInfo(name="木盾", item_type="armor", description="坚硬橡木制成的盾牌，可以抵挡基础攻击。", rarity="common", stat_bonus={"defense": 2}, value=15), "面包": ItemInfo(name="面包", item_type="consumable", description="新鲜烤制的面包，香气扑鼻。", usable=True, use_effect="恢复 10 饱食度", value=5), "烤肉": ItemInfo(name="烤肉", item_type="consumable", description="多汁的烤肉，令人食指大动。", usable=True, use_effect="恢复 25 饱食度", value=10), "麦酒": ItemInfo(name="麦酒", item_type="consumable", description="村庄特产的麦酒，味道醇厚。", usable=True, use_effect="恢复 10 士气，降低 5 理智", value=8), "草药包": ItemInfo(name="草药包", item_type="consumable", description="采集的新鲜草药，可以制作简单药剂。", usable=True, use_effect="恢复 20 HP", value=15), "火把": ItemInfo(name="火把", item_type="misc", description="浸过油脂的火把，可在黑暗中照明。", usable=True, use_effect="照亮周围区域", value=3), "绳索": ItemInfo(name="绳索", item_type="misc", description="结实的麻绳，在探险中很实用。", value=5), "解毒药水": ItemInfo(name="解毒药水", item_type="consumable", description="散发着清苦气味的药水，可以解除中毒状态。", usable=True, use_effect="解除中毒状态", value=20), "小型治疗药水": ItemInfo(name="小型治疗药水", item_type="consumable", description="泛着淡红色光芒的药水。", usable=True, use_effect="恢复 30 HP", value=25), "村庄地图": ItemInfo(name="村庄地图", item_type="quest_item", description="一张画着村庄与周边道路的实用地图，边角处有村长留下的简短记号。", quest_related=True, value=0, lore_text="这张地图把村庄广场、店铺和村口小路都标得很清楚，显然是给初次上路的人准备的。"), "黑暗森林地图": ItemInfo(name="黑暗森林地图", item_type="quest_item", description="一张补全了森林入口、溪边营地和深处路径的地图。", quest_related=True, value=0, lore_text="地图边缘沾着泥水和血迹，像是刚从危险地带抢出来的。"), "山麓地图": ItemInfo(name="山麓地图", item_type="quest_item", description="记着渡口、盗贼营和遗迹路径的山麓地图。", quest_related=True, value=0, lore_text="粗糙的炭笔线条标出了山道、渡口和盗贼常走的隐蔽小径。"), "古塔地图": ItemInfo(name="古塔地图", item_type="quest_item", description="一张标记古塔废墟出入口和危险区域的旧图。", quest_related=True, value=0, lore_text="纸面上反复描重的几处塔层，似乎都是前人特意警告的危险位置。"), "森林之钥": ItemInfo(name="森林之钥", item_type="key", description="一把散发着微弱绿光的古老钥匙，似乎能打开森林深处的某个入口。", rarity="rare", quest_related=True, value=0, lore_text="钥匙上刻着精灵文字，翻译过来是：'唯有勇者可通行'。"), "神秘卷轴": ItemInfo(name="神秘卷轴", item_type="quest_item", description="记载着古老知识的卷轴，散发着微弱的魔力波动。", rarity="rare", quest_related=True, value=0), # -------- 扩展物品 -------- "矿工旧钥匙": ItemInfo(name="矿工旧钥匙", item_type="key", description="一把生锈的铜钥匙，上面刻着一个矿镐图案。", rarity="uncommon", quest_related=True, value=0, lore_text="钥匙柄上隐约可见'B-7采掘区'的刻字。"), "骷髅碎骨": ItemInfo(name="骷髅碎骨", item_type="material", description="从骷髅兵身上掉落的骨头碎片，泛着不自然的寒光。", rarity="common", value=5), "盗贼日志": ItemInfo(name="盗贼日志", item_type="quest_item", description="一本沾满泥渍的笔记本，记录着盗贼团伙近期的行动计划。", rarity="uncommon", quest_related=True, value=0), "精灵护符": ItemInfo(name="精灵护符", item_type="accessory", description="由精灵遗迹守护者亲手制作的小型护符，散发着柔和的绿色微光。", rarity="rare", stat_bonus={"perception": 3, "sanity": 5}, value=50, lore_text="佩戴者能感受到来自远古精灵的庇佑。"), "锈蚀铁锤": ItemInfo(name="锈蚀铁锤", item_type="weapon", description="矿洞里发现的旧铁锤，虽然锈迹斑斑但依然沉重有力。", rarity="common", stat_bonus={"attack": 4}, value=15), "荧光苔藓": ItemInfo(name="荧光苔藓", item_type="consumable", description="矿洞深处生长的发光苔藓，据说有微弱的疗伤效果。", usable=True, use_effect="恢复 15 HP，恢复 5 理智", rarity="uncommon", value=12), "古塔法师笔记": ItemInfo(name="古塔法师笔记", item_type="quest_item", description="在古塔废墟中找到的残破笔记，记载着某种仪式的片段。", rarity="rare", quest_related=True, value=0, lore_text="字迹已经模糊，但仍能辨认出几个关键的魔法符号。"), } # --- 初始传闻 --- self.world.rumors = [ "最近森林里的哥布林越来越嚣张了，好几个猎人都不敢进去了。", "听说铁匠格林以前在王都待过，不知道为什么来了这个小村子。", "旅店里来了个奇怪的旅人，整天把自己裹得严严实实的。", "河对岸的旧矿洞晚上闹鬼，渡口的老渔夫说他亲眼看见过蓝色的火光。", "山脚下好像有一伙盗贼扎了营，最近有商队被劫的消息。", "村子东边的古塔里据说住过一个法师，后来不知为何法师消失了，塔也荒废了。", "有人在精灵遗迹附近见到过一个穿绿袍的身影，不知是人是鬼。", ] # --- 显式环境事件模板 --- self.environment_event_pool: list[dict[str, Any]] = [ { "event_id": "lanterns_dim", "category": "light", "title": "灯火忽暗", "description": "屋内的灯火突然暗了一截，墙角的影子被拉得细长。", "location_types": ["town", "shop"], "time_slots": ["黄昏", "夜晚", "深夜"], "severity": "medium", "state_changes": {"sanity_change": -2}, "prompt_hint": "环境光线明显变暗，叙事里要体现角色对阴影和氛围变化的反应。", }, { "event_id": "cold_gust", "category": "environment", "title": "冷风穿林", "description": "一阵带着湿意的冷风从林间掠过，让人本能地绷紧肩背。", "location_types": ["wilderness", "dungeon"], "time_slots": ["下午", "黄昏", "夜晚", "深夜"], "severity": "low", "state_changes": {"morale_change": -3}, "prompt_hint": "风声和体感温度都发生了变化，适合让玩家察觉环境压力正在上升。", }, { "event_id": "forest_rustle", "category": "environment", "title": "林影骚动", "description": "黑暗树影间传来急促的窸窣声，仿佛刚有什么东西贴着边缘掠过。", "location_types": ["wilderness", "dungeon"], "time_slots": ["黄昏", "夜晚", "深夜"], "min_danger": 3, "severity": "medium", "state_changes": {"sanity_change": -3}, "prompt_hint": "这是偏悬疑的环境扰动，至少一个后续选项应允许玩家追查或回避。", }, { "event_id": "fireplace_relief", "category": "environment", "title": "火光回暖", "description": "炉火和热气驱散了紧绷感，让呼吸也慢慢平稳下来。", "location_types": ["shop", "town"], "requires_rest_available": True, "time_slots": ["黄昏", "夜晚", "深夜"], "severity": "low", "state_changes": {"morale_change": 4, "sanity_change": 2}, "prompt_hint": "这是偏正向的氛围事件，叙事里可以体现安全感和短暂放松。", }, { "event_id": "fog_pressures_in", "category": "environment", "title": "雾气压近", "description": "潮湿的雾从地面漫上来，视野被一点点吞没，声音也变得含混。", "location_types": ["wilderness", "dungeon"], "time_slots": ["清晨", "黄昏", "夜晚", "深夜"], "weathers": ["浓雾", "小雨"], "severity": "medium", "state_changes": {"sanity_change": -2}, "prompt_hint": "视野受限且不安感上升，叙事里应弱化远景、强调近距离感官细节。", }, ] # --- 玩家初始装备 --- self.player.inventory = ["面包", "面包", "小型治疗药水"] self.player.location = self.world.current_scene # ============================================================ # 核心方法 # ============================================================ def update_location(self, new_location: str) -> None: """ 更新当前位置并维护足迹历史。 规则： - 当 new_location 与当前地点不同：把旧地点写入 location_history，然后更新 current_location - 当 new_location 相同：不追加历史 - 同步更新 self.player.location 与 self.world.current_scene，确保一致性 """ target = str(new_location or "").strip() if not target: return if target == self.current_location: return old_location = self.current_location if old_location: self.location_history.append(old_location) # 让游戏状态和地图状态始终一致 self.current_location = target self.player.location = target self.world.current_scene = target # ============================================================ # 状态变更应用 # ============================================================ def apply_changes(self, changes: dict) -> list[str]: """ 接收 Qwen 返回的状态变更 JSON，校验并应用到当前状态。 设计思路： - LLM 返回的变更是增量式的（如 hp_change: -10），而非绝对值 - 逐字段解析和应用，确保每个变更都经过校验 - 返回变更日志列表，方便 UI 展示 Args: changes: Qwen 输出中解析出的状态变更字典 Returns: 变更描述列表 ["HP: 100 → 90", "位置: 村庄 → 森林"] """ change_log: list[str] = [] # --- 过滤 None 值：LLM 可能将 null 字段返回为 None，全部跳过 --- _filtered = {} for k, v in changes.items(): if v is None: continue # 字符串 "None" / "null" 也视为空 if isinstance(v, str) and v.strip().lower() in ("none", "null", ""): continue # 数值 0 的 change 字段无意义，也跳过 if isinstance(v, (int, float)) and v == 0 and k.endswith("_change"): continue # 空列表 / 空字典跳过 if isinstance(v, (list, dict)) and len(v) == 0: continue _filtered[k] = v changes = _filtered # --- 玩家属性变更 --- if "hp_change" in changes: old_hp = self.player.hp self.player.hp = clamp( self.player.hp + int(changes["hp_change"]), 0, self.player.max_hp, ) if self.player.hp != old_hp: change_log.append(f"HP: {old_hp} → {self.player.hp}") if "mp_change" in changes: old_mp = self.player.mp self.player.mp = clamp( self.player.mp + int(changes["mp_change"]), 0, self.player.max_mp, ) if self.player.mp != old_mp: change_log.append(f"MP: {old_mp} → {self.player.mp}") if "gold_change" in changes: old_gold = self.player.gold self.player.gold = max(0, self.player.gold + int(changes["gold_change"])) if self.player.gold != old_gold: change_log.append(f"金币: {old_gold} → {self.player.gold}") if "exp_change" in changes: old_exp = self.player.experience self.player.experience += int(changes["exp_change"]) change_log.append(f"经验: {old_exp} → {self.player.experience}") # 检查是否升级 while self.player.experience >= self.player.exp_to_next_level: self._level_up() change_log.append(f"升级！当前等级: {self.player.level}") if "morale_change" in changes: old_morale = self.player.morale self.player.morale = clamp( self.player.morale + int(changes["morale_change"]), 0, 100, ) if self.player.morale != old_morale: change_log.append(f"士气: {old_morale} → {self.player.morale}") if "sanity_change" in changes: old_sanity = self.player.sanity self.player.sanity = clamp( self.player.sanity + int(changes["sanity_change"]), 0, 100, ) if self.player.sanity != old_sanity: change_log.append(f"理智: {old_sanity} → {self.player.sanity}") if "hunger_change" in changes: old_hunger = self.player.hunger self.player.hunger = clamp( self.player.hunger + int(changes["hunger_change"]), 0, 100, ) if self.player.hunger != old_hunger: change_log.append(f"饱食度: {old_hunger} → {self.player.hunger}") if "stamina_change" in changes: old_stamina = self.player.stamina self.player.stamina = clamp( self.player.stamina + int(changes["stamina_change"]), 0, self.player.max_stamina, ) if self.player.stamina != old_stamina: change_log.append(f"体力: {old_stamina} → {self.player.stamina}") if "karma_change" in changes: old_karma = self.player.karma self.player.karma += int(changes["karma_change"]) if self.player.karma != old_karma: change_log.append(f"善恶值: {old_karma} → {self.player.karma}") # --- 位置变更 --- if "new_location" in changes: old_loc = self.player.location new_loc = str(changes["new_location"]) if new_loc.strip().lower() not in ("", "none", "null") and new_loc != old_loc: current_loc = self.world.locations.get(old_loc) target_loc = self.world.locations.get(new_loc) if target_loc is None: change_log.append(f"忽略非法位置变更: {new_loc}") elif current_loc and new_loc not in current_loc.connected_to: change_log.append(f"忽略非法位置变更: {old_loc} → {new_loc}") elif ( not target_loc.is_accessible and target_loc.required_item and target_loc.required_item not in self.player.inventory ): change_log.append(f"忽略未解锁地点: {new_loc}") else: self.update_location(new_loc) change_log.append(f"位置: {old_loc} → {new_loc}") # 发现新地点 if new_loc not in self.world.discovered_locations: self.world.discovered_locations.append(new_loc) change_log.append(f"发现新地点: {new_loc}") if new_loc in self.world.locations: self.world.locations[new_loc].is_discovered = True # --- 物品变更 --- # 货币关键词列表：这些物品不进背包，而是直接转换为金币 _CURRENCY_KEYWORDS = ["铜币", "银币", "铜钱", "银两", "金币", "货币", "钱袋", "钱币", "硬币"] if "items_gained" in changes: for item in changes["items_gained"]: item_str = str(item) # 检查是否为货币类物品 —— 如果是，跳过入背包（金币已通过 gold_change 处理） is_currency = any(kw in item_str for kw in _CURRENCY_KEYWORDS) if is_currency: # 如果 gold_change 没有设置，尝试自动补偿少量金币 if "gold_change" not in changes: old_gold = self.player.gold self.player.gold += 3 # 默认少量金币 change_log.append(f"金币: {old_gold} → {self.player.gold}") logger.info(f"货币物品 '{item_str}' 已转换为金币，不放入背包") continue self.player.inventory.append(item_str) self.last_recent_gain = item_str change_log.append(f"获得物品: {item}") if "items_lost" in changes: for item in changes["items_lost"]: item_str = str(item) # 货币类物品也不需要从背包移除 is_currency = any(kw in item_str for kw in _CURRENCY_KEYWORDS) if is_currency: continue if item_str in self.player.inventory: self.player.inventory.remove(item_str) change_log.append(f"失去物品: {item}") # --- 技能变更 --- if "skills_gained" in changes: for skill in changes["skills_gained"]: skill_str = str(skill) if skill_str not in self.player.skills: self.player.skills.append(skill_str) change_log.append(f"习得技能: {skill}") # --- 状态效果 --- if "status_effects_added" in changes: for effect_data in changes["status_effects_added"]: if isinstance(effect_data, dict): effect = StatusEffect(**effect_data) self.player.status_effects.append(effect) # 构建详细的状态效果日志 parts = [f"获得状态: {effect.name}"] if effect.description: parts.append(f"（{effect.description}）") if effect.stat_modifiers: mod_strs = [] _STAT_CN = { "hp": "生命", "mp": "魔力", "attack": "攻击力", "defense": "防御力", "speed": "速度", "luck": "幸运", "perception": "感知", "sanity": "理智", "hunger": "饱食度", "morale": "士气", "gold": "金币", "karma": "善恶值", "experience": "经验", } for stat, val in effect.stat_modifiers.items(): cn = _STAT_CN.get(stat, stat) sign = "+" if val > 0 else "" mod_strs.append(f"{cn}{sign}{val}/回合") parts.append(f"[{', '.join(mod_strs)}]") if effect.duration > 0: parts.append(f"持续{effect.duration}回合") elif effect.duration == -1: parts.append("永久") change_log.append(" ".join(parts)) elif isinstance(effect_data, str): effect = StatusEffect(name=effect_data) self.player.status_effects.append(effect) change_log.append(f"获得状态: {effect_data}") if "status_effects_removed" in changes: for name in changes["status_effects_removed"]: self.player.status_effects = [ e for e in self.player.status_effects if e.name != str(name) ] change_log.append(f"移除状态: {name}") # --- NPC 相关变更 --- if "npc_changes" in changes: for npc_name, npc_data in changes["npc_changes"].items(): if npc_name in self.world.npcs: npc = self.world.npcs[npc_name] if "attitude" in npc_data: new_attitude = str(npc_data["attitude"]) if npc.attitude != new_attitude: npc.attitude = new_attitude change_log.append(f"NPC {npc_name} 态度变为: {npc.attitude}") if "is_alive" in npc_data: new_is_alive = bool(npc_data["is_alive"]) was_alive = npc.is_alive if npc.is_alive != new_is_alive: npc.is_alive = new_is_alive if was_alive and not npc.is_alive: change_log.append(f"NPC {npc_name} 已死亡") if "relationship_change" in npc_data: old_rel = npc.relationship_level npc.relationship_level = clamp( npc.relationship_level + int(npc_data["relationship_change"]), -100, 100, ) if npc.relationship_level != old_rel: change_log.append( f"NPC {npc_name} 好感度: {old_rel} → {npc.relationship_level}" ) if "hp_change" in npc_data: old_hp = npc.hp npc.hp = max(0, npc.hp + int(npc_data["hp_change"])) if npc.hp <= 0: npc.is_alive = False change_log.append(f"NPC {npc_name} 被击败") elif npc.hp != old_hp: change_log.append(f"NPC {npc_name} HP: {old_hp} → {npc.hp}") if "memory_add" in npc_data: npc.memory.append(str(npc_data["memory_add"])) # --- 任务变更 --- if "quest_updates" in changes: for quest_id, quest_data in changes["quest_updates"].items(): if quest_id in self.world.quests: quest = self.world.quests[quest_id] if "objectives_completed" in quest_data: for obj in quest_data["objectives_completed"]: if str(obj) in quest.objectives: quest.objectives[str(obj)] = True change_log.append(f"完成目标: {obj}") if "status" in quest_data: quest.status = str(quest_data["status"]) _QUEST_STATUS_CN = { "active": "进行中", "in_progress": "进行中", "IN_PROGRESS": "进行中", "ACTIVE": "进行中", "completed": "已完成", "COMPLETED": "已完成", "failed": "已失败", "FAILED": "已失败", "expired": "已过期", "EXPIRED": "已过期", } status_cn = _QUEST_STATUS_CN.get(quest.status, quest.status) change_log.append(f"任务【{quest.title}】进展至\"{status_cn}\"") # --- 世界状态变更 --- if "weather_change" in changes: valid_weathers = {"晴朗", "多云", "小雨", "暴风雨", "大雪", "浓雾"} new_weather = str(changes["weather_change"]) if new_weather in valid_weathers: if self.world.weather != new_weather: self.world.weather = new_weather self.world.last_weather_change_minutes = self.elapsed_minutes_total change_log.append(f"天气变为: {self.world.weather}") else: logger.warning(f"无效的 weather_change 值 '{new_weather}'，已忽略。") if "time_change" in changes: valid_times = ["清晨", "上午", "正午", "下午", "黄昏", "夜晚", "深夜"] new_time = str(changes["time_change"]) if new_time in valid_times: old_time = self.world.time_of_day if new_time != old_time: self.world.time_of_day = new_time change_log.append(f"时间流逝: {old_time} → {self.world.time_of_day}") else: logger.warning(f"无效的 time_change 值 '{new_time}'，已忽略。合法值: {valid_times}") if "weather_change" in changes or "time_change" in changes: old_light = self.world.light_level self.world.light_level = self._determine_light_level() if old_light != self.world.light_level: change_log.append(f"光照变化: {old_light} → {self.world.light_level}") if "global_flags_set" in changes: for flag, value in changes["global_flags_set"].items(): self.world.global_flags[flag] = bool(value) # 全局标记仅内部使用，不展示给用户 logger.info(f"全局标记设置: {flag} = {value}") if "world_event" in changes: world_event = str(changes["world_event"]) if not self.world.world_events or self.world.world_events[-1] != world_event: self.world.world_events.append(world_event) change_log.append(f"世界事件: {world_event}") # --- 装备变更 --- if "equip" in changes: for slot, item_name in changes["equip"].items(): if slot in self.player.equipment: old_item = self.player.equipment[slot] new_item = item_name if item_name and str(item_name).lower() not in ("none", "null", "") else None # 1. 如果旧装备栏有物品，卸下时放回背包 if old_item and old_item != "无": if old_item not in self.player.inventory: self.player.inventory.append(old_item) logger.info(f"卸下装备 '{old_item}' 放回背包") # 2. 如果要装备新物品，从背包中移除 if new_item: new_item_str = str(new_item) if new_item_str in self.player.inventory: self.player.inventory.remove(new_item_str) logger.info(f"从背包取出 '{new_item_str}' 装备到 [{slot}]") self.player.equipment[slot] = new_item display_old = old_item or "无" display_new = new_item or "无" change_log.append(f"装备 [{slot}]: {display_old} → {display_new}") # --- 玩家称号变更 --- if "title_change" in changes: old_title = self.player.title self.player.title = str(changes["title_change"]) change_log.append(f"称号: {old_title} → {self.player.title}") # 战斗派生属性需要与装备和基础属性保持同步 self.refresh_combat_stats() if change_log: logger.info(f"状态变更: {change_log}") return change_log def validate(self) -> tuple[bool, list[str]]: """ 校验当前状态的合法性。 设计思路： - 检查所有数值是否在合法范围内 - HP <= 0 时标记游戏结束 - 理智过低时施加特殊效果 - 返回 (是否合法, 问题列表) Returns: (is_valid, issues): 合法性标志和问题描述列表 """ issues: list[str] = [] # HP 校验 —— 核心逻辑：HP <= 0 触发死亡 if self.player.hp <= 0: self.player.hp = 0 self.game_mode = "game_over" self.player.death_count += 1 issues.append("玩家生命值归零，触发死亡结局！") # MP 范围校验 self.player.mp = clamp(self.player.mp, 0, self.player.max_mp) self.player.stamina = clamp(self.player.stamina, 0, self.player.max_stamina) # 饱食度惩罚 if self.player.hunger <= 0: self.player.hunger = 0 issues.append("玩家极度饥饿，攻击力和防御力下降！") # 理智值校验 if self.player.sanity <= 0: self.player.sanity = 0 self.game_mode = "game_over" issues.append("玩家理智归零，陷入疯狂！触发疯狂结局！") # 士气校验 if self.player.morale <= 10: issues.append("玩家士气极低，行动效率降低。") # 金币不能为负 if self.player.gold < 0: self.player.gold = 0 issues.append("金币不足。") is_valid = self.game_mode != "game_over" return is_valid, issues def get_equipment_stat_bonuses(self) -> dict[str, int]: """Aggregate stat bonuses from currently equipped items.""" bonuses: dict[str, int] = {} for item_name in self.player.equipment.values(): if not item_name: continue item_info = self.world.item_registry.get(str(item_name)) if item_info is None: continue for stat_name, amount in item_info.stat_bonus.items(): bonuses[stat_name] = bonuses.get(stat_name, 0) + int(amount) return bonuses def refresh_combat_stats(self) -> None: """Refresh deterministic combat stats from base values + equipment bonuses.""" bonuses = self.get_equipment_stat_bonuses() self.player.attack_power = max(1, int(self.player.attack) + int(bonuses.get("attack", 0))) self.player.defense_power = max(0, int(self.player.defense) + int(bonuses.get("defense", 0))) def _status_multiplier(self, value: int) -> float: if value >= 90: return 1.5 if value >= 80: return 1.2 if value >= 30: return 1.0 if value >= 20: return 0.9 if value >= 10: return 0.7 if value >= 5: return 0.6 return 0.3 def get_survival_state_snapshot(self) -> dict[str, Any]: hunger_multiplier = self._status_multiplier(self.player.hunger) sanity_multiplier = self._status_multiplier(self.player.sanity) morale_multiplier = self._status_multiplier(self.player.morale) # 体力也纳入战斗乘数：低体力会显著降低战斗能力 stamina_multiplier = self._status_multiplier(self.player.stamina) combined_multiplier = min( hunger_multiplier, sanity_multiplier, morale_multiplier, stamina_multiplier, ) peak_state = all( value >= 80 for value in (self.player.hunger, self.player.sanity, self.player.morale, self.player.stamina) ) near_death = sum( value < 10 for value in (self.player.hunger, self.player.sanity, self.player.morale) ) >= 2 if peak_state: combined_multiplier *= 1.1 if near_death: combined_multiplier *= 0.5 return { "hunger_multiplier": hunger_multiplier, "sanity_multiplier": sanity_multiplier, "morale_multiplier": morale_multiplier, "stamina_multiplier": stamina_multiplier, "combined_multiplier": round(combined_multiplier, 3), "peak_state": peak_state, "near_death": near_death, } def get_effective_player_stats(self) -> dict[str, int]: """Return display-oriented effective stats after equipment bonuses.""" bonuses = self.get_equipment_stat_bonuses() tracked_stats = ("attack", "defense", "speed", "luck", "perception", "stamina") state_snapshot = self.get_survival_state_snapshot() multiplier = float(state_snapshot["combined_multiplier"]) effective_stats: dict[str, int] = {} for stat_name in tracked_stats: base_value = int(getattr(self.player, stat_name)) bonus_value = int(bonuses.get(stat_name, 0)) boosted_value = int(round((base_value + bonus_value) * multiplier)) cap = self.player.max_stamina if stat_name == "stamina" else max(base_value + bonus_value, 1) if multiplier >= 1: effective_stats[stat_name] = max(base_value + bonus_value, boosted_value) else: effective_stats[stat_name] = clamp(boosted_value, 1, max(cap, boosted_value)) return effective_stats def get_clock_minutes(self) -> int: return int(self.world.time_progress_units) * 10 def get_minute_of_day(self) -> int: return self.get_clock_minutes() % (24 * 60) def get_clock_display(self) -> str: total_minutes = self.get_clock_minutes() % (24 * 60) hours = total_minutes // 60 minutes = total_minutes % 60 return f"{hours:02d}:{minutes:02d}" def _time_of_day_from_minutes(self, total_minutes: int) -> str: minute_of_day = total_minutes % (24 * 60) if 300 <= minute_of_day < 480: return "清晨" if 480 <= minute_of_day < 720: return "上午" if 720 <= minute_of_day < 840: return "正午" if 840 <= minute_of_day < 1080: return "下午" if 1080 <= minute_of_day < 1260: return "黄昏" if 1260 <= minute_of_day < 1440: return "夜晚" return "深夜" def _sync_world_clock(self): self.world.time_of_day = self._time_of_day_from_minutes(self.get_clock_minutes()) def can_overnight_rest(self) -> bool: current_loc = self.world.locations.get(self.player.location) if current_loc is None or not current_loc.rest_available: return False if self.player.location not in OVERNIGHT_REST_LOCATIONS: return False return self.get_minute_of_day() >= 19 * 60 def prepare_overnight_rest(self) -> tuple[list[str], dict[str, int]]: """Advance to next morning and return full-recovery deltas for overnight rest.""" if not self.can_overnight_rest(): return [], {} old_clock = self.get_clock_display() old_time_of_day = self.world.time_of_day old_day_count = self.world.day_count old_light = self.world.light_level minute_of_day = self.get_minute_of_day() minutes_until_midnight = (24 * 60) - minute_of_day target_elapsed_minutes = self.elapsed_minutes_total + minutes_until_midnight + 6 * 60 self.elapsed_minutes_total = target_elapsed_minutes self.world.day_count = target_elapsed_minutes // (24 * 60) + 1 self.world.time_progress_units = (target_elapsed_minutes % (24 * 60)) // 10 self._sync_world_clock() self.world.light_level = self._determine_light_level() tick_log: list[str] = [] if self.world.day_count != old_day_count: tick_log.append(f"新的一天！第{self.world.day_count}天") new_clock = self.get_clock_display() if new_clock != old_clock: tick_log.append(f"时间流逝: {old_clock} → {new_clock}") if self.world.time_of_day != old_time_of_day: tick_log.append(f"时段变化: {old_time_of_day} → {self.world.time_of_day}") if self.world.light_level != old_light: tick_log.append(f"光照变化: {old_light} → {self.world.light_level}") hunger_cost = 20 if self.player.location == "村庄旅店" else 25 recovery_changes: dict[str, int] = { "hp_change": self.player.max_hp - self.player.hp, "mp_change": self.player.max_mp - self.player.mp, "stamina_change": self.player.max_stamina - self.player.stamina, "morale_change": 100 - self.player.morale, "sanity_change": 100 - self.player.sanity, "hunger_change": -hunger_cost, } return tick_log, { key: value for key, value in recovery_changes.items() if int(value) != 0 } def to_prompt(self) -> str: """ 将当前完整状态序列化为自然语言描述，注入 System Prompt。 设计思路（需求文档核心要求）： - System Prompt 必须包含当前状态描述 - 描述要全面但简洁，避免 token 浪费 - 包括：场景、玩家状态、已发生的重要事件、NPC 信息 - 加入一致性约束指令，提醒 LLM 不要产生矛盾 """ # 1. 场景与环境 scene_desc = ( f"【当前场景】{self.world.current_scene}\n" f"【时间】第{self.world.day_count}天 {self.world.time_of_day}\n" f"【天气】{self.world.weather}\n" f"【季节】{self.world.season}" ) # 2. 玩家状态 effects_str = "、".join(e.name for e in self.player.status_effects) if self.player.status_effects else "无" equipped = {k: (v or "无") for k, v in self.player.equipment.items()} equip_str = "、".join(f"{k}={v}" for k, v in equipped.items()) # 背包物品标注消耗品/可重复使用 if self.player.inventory: inv_items = [] for item_name in self.player.inventory: if self.is_item_consumable(item_name): inv_items.append(f"{item_name}[消耗品]") else: inv_items.append(f"{item_name}[可重复使用]") inventory_str = "、".join(inv_items) else: inventory_str = "空" skills_str = "、".join(self.player.skills) if self.player.skills else "无" player_desc = ( f"【玩家】{self.player.name}（{self.player.title}）\n" f" 等级: {self.player.level} | 经验: {self.player.experience}/{self.player.exp_to_next_level}\n" f" HP: {self.player.hp}/{self.player.max_hp}\n" f" MP: {self.player.mp}/{self.player.max_mp}\n" f" 攻击: {self.player.attack} | 防御: {self.player.defense} | 实战攻击: {self.player.attack_power} | 实战防御: {self.player.defense_power}\n" f" 速度: {self.player.speed} | 幸运: {self.player.luck} | 感知: {self.player.perception}\n" f" 金币: {self.player.gold} | 善恶值: {self.player.karma}\n" f" 士气: {self.player.morale} | 理智: {self.player.sanity} | 饱食度: {self.player.hunger}\n" f" 装备: {equip_str}\n" f" 背包: {inventory_str}\n" f" 技能: {skills_str}\n" f" 状态效果: {effects_str}\n" f" 所在位置: {self.player.location}" ) # 3. 当前场景中的 NPC current_npcs = [ npc for npc in self.world.npcs.values() if npc.location == self.player.location and npc.is_alive ] if current_npcs: npc_lines = [] for npc in current_npcs: mem = "；".join(npc.memory[-3:]) if npc.memory else "无记忆" npc_lines.append( f" - {npc.name}（{npc.occupation}, {npc.race}, 态度: {npc.attitude}, " f"好感度: {npc.relationship_level}, 记忆: {mem}）" ) npc_desc = "【场景中的NPC】\n" + "\n".join(npc_lines) else: npc_desc = "【场景中的NPC】无" # 4. 当前活跃任务 active_quests = [q for q in self.world.quests.values() if q.status == "active"] if active_quests: quest_lines = [] for q in active_quests: objectives = "；".join( f"{'✅' if done else '❌'}{obj}" for obj, done in q.objectives.items() ) time_info = f"（剩余 {q.turns_remaining} 回合）" if q.turns_remaining > 0 else "" quest_lines.append(f" - [{q.quest_type.upper()}] {q.title}: {objectives}{time_info}") quest_desc = "【活跃任务】\n" + "\n".join(quest_lines) else: quest_desc = "【活跃任务】无" # 5. 已发现地点的连接关系 loc_info = self.world.locations.get(self.player.location) if loc_info: accessible = [ name for name in loc_info.connected_to if name in self.world.locations and self.world.locations[name].is_accessible ] blocked = [ f"{name}（需要: {self.world.locations[name].required_item}）" for name in loc_info.connected_to if name in self.world.locations and not self.world.locations[name].is_accessible ] move_desc = f"【可前往的地点】{'、'.join(accessible) if accessible else '无'}" if blocked: move_desc += f"\n【被阻挡的地点】{'、'.join(blocked)}" else: move_desc = "【可前往的地点】未知" # 6. 近期事件（最近 5 条） if self.event_log: recent = self.event_log[-5:] event_lines = [f" - [回合{e.turn}] {e.description}" for e in recent] event_desc = "【近期事件】\n" + "\n".join(event_lines) else: event_desc = "【近期事件】无" # 7. 传闻 rumors_desc = "" if self.world.rumors: rumors_desc = "\n【流传的传闻】\n" + "\n".join(f" - {r}" for r in self.world.rumors[-3:]) # 8. 显式环境事件 environment_event_desc = "" if self.pending_environment_event: env = self.pending_environment_event environment_event_desc = ( f"\n【本回合环境事件 —— 必须融入本次叙事】\n" f"[{env.category.upper()}|{env.severity}] {env.title}\n" f"{env.description}\n" f"{env.prompt_hint}\n" f"请将此事件自然地融入剧情描写中，作为本回合可感知的环境变化。" f"玩家可以选择回应、调查、规避或忽视它。至少一个选项应与此事件相关。" ) self.pending_environment_event = None # 用后清除 # 9. 一致性约束指令 consistency_rules = ( "\n【一致性约束 —— 你必须严格遵守】\n" "1. 已死亡的NPC不可再出现或对话（除非有特殊复活剧情）。\n" "2. 玩家背包中没有的物品不可使用或赠送。\n" "3. 玩家不可到达未连接的地点，被阻挡的地点需要对应物品才能进入。\n" "4. 时间线不可回退，已发生的事件不可矛盾。\n" "5. NPC的态度和记忆应与历史事件一致。\n" "6. 战斗伤害应考虑攻击力和防御力的差值，结果要合理。\n" "7. 所有状态变更必须在 state_changes 字段中明确输出。\n" "8. 每次生成的文本描写必须使用全新的比喻和意象，严禁重复之前回合用过的修辞和句式。\n" "9. 【物品消耗规则】只有消耗品（药水、食物等一次性物品）在使用后才会消失，应放入 items_lost。" "非消耗品（哨子、武器、工具、乐器、钥匙等可重复使用的物品）使用后仍然保留在背包中，" "绝对不要将它们放入 items_lost。例如：吹响哨子后哨子仍在背包中；使用火把照明后火把仍在。\n" "10. 【选项物品约束】生成的选项中如果涉及使用某个物品，该物品必须当前在玩家背包中。" "不要生成使用玩家不拥有的物品的选项。\n" '11. 【货币规则】游戏内货币统一为"金币"，对应 gold_change 字段。严禁使用"铜币""银币""银两"等名称。' "任何钱财/财物类收获（如击败怪物掉落的钱币、交易获得的货款等）必须通过 gold_change 表达，" "严禁将任何种类的钱币放入 items_gained。\n" '12. 【装备规则】装备物品时必须使用 equip 字段指定槽位和物品名称（如 "weapon": "小刀"）。' "系统会自动将装备的物品从背包移到装备栏，并将旧装备放回背包。" "因此装备操作时不要在 items_lost/items_gained 中重复处理该物品。" "合法槽位：weapon / armor / accessory / helmet / boots。卸下装备时将对应槽位设为 null。" ) # 组合完整 Prompt full_prompt = "\n\n".join([ scene_desc, player_desc, npc_desc, quest_desc, move_desc, event_desc, rumors_desc, environment_event_desc, consistency_rules, ]) return full_prompt def log_event( self, event_type: str, description: str, player_action: str = "", involved_npcs: list[str] | None = None, state_changes: dict | None = None, consequence_tags: list[str] | None = None, is_reversible: bool = True, ): """ 记录一条事件到 event_log。 每次状态变更都应该调用此方法，确保完整的历史记录。 事件日志是一致性维护的基石。 """ event = GameEvent( turn=self.turn, day=self.world.day_count, time_of_day=self.world.time_of_day, event_type=event_type, description=description, location=self.player.location, involved_npcs=involved_npcs or [], state_changes=state_changes or {}, player_action=player_action, consequence_tags=consequence_tags or [], is_reversible=is_reversible, ) self.event_log.append(event) logger.info(f"事件记录: [{event_type}] {description}") def check_consistency(self, proposed_changes: dict) -> list[str]: """ 对比事件日志和当前状态，检测拟议变更中的矛盾。 设计思路： - 在 apply_changes 之前调用，预防性检测 - 返回所有发现的矛盾描述列表 - 空列表 = 无矛盾，可以安全应用 检测维度： 1. 已死亡 NPC 是否被重新引用 2. 不存在的物品是否被消耗 3. 不可达的地点是否被移动到 4. 任务目标是否已经跳跃完成 """ contradictions: list[str] = [] # 检测1: 已死亡NPC是否被引用 if "npc_changes" in proposed_changes: for npc_name in proposed_changes["npc_changes"]: if npc_name in self.world.npcs and not self.world.npcs[npc_name].is_alive: contradictions.append( f"矛盾: 试图与已死亡的NPC '{npc_name}' 交互" ) # 检测2: 不存在的物品是否被消耗 if "items_lost" in proposed_changes: for item in proposed_changes["items_lost"]: if str(item) not in self.player.inventory: contradictions.append( f"矛盾: 试图消耗不在背包中的物品 '{item}'" ) elif not self.is_item_consumable(str(item)): # 非消耗品不应因使用而消失（交易/丢弃除外，由引擎层判断） contradictions.append( f"矛盾: 物品 '{item}' 不是消耗品，使用后不应消失。请将其从 items_lost 中移除。" ) # 检测3: 位置移动是否合法 if "new_location" in proposed_changes: target = str(proposed_changes["new_location"]) if target.strip().lower() not in ("", "none", "null") and target != self.player.location: current_loc = self.world.locations.get(self.player.location) target_loc = self.world.locations.get(target) if target_loc is None: contradictions.append( f"矛盾: 试图移动到未注册的地点 '{target}'" ) if current_loc and target not in current_loc.connected_to: contradictions.append( f"矛盾: 试图移动到不相邻的地点 '{target}'（当前位置: {self.player.location}）" ) if target_loc and not target_loc.is_accessible: if target_loc.required_item and target_loc.required_item not in self.player.inventory: contradictions.append( f"矛盾: 地点 '{target}' 被锁定，需要 '{target_loc.required_item}'" ) # 检测4: 金币是否足够（如果是消费操作） if "gold_change" in proposed_changes: change = int(proposed_changes["gold_change"]) if change < 0 and self.player.gold + change < 0: contradictions.append( f"矛盾: 金币不足（当前: {self.player.gold}，需要: {abs(change)}）" ) if contradictions: logger.warning(f"一致性检查发现矛盾: {contradictions}") return contradictions def pre_validate_action(self, intent: dict) -> tuple[bool, str]: """ 预校验玩家意图的合法性（在 LLM 调用前立即拦截非法操作）。 设计思路： - 在任何 API 调用之前就检测明显违反一致性的操作 - 不合法时立即驳回，避免浪费 API 调用和回合 - 检测维度：物品是否在背包/装备中、技能是否已习得、 raw_input 中是否提及使用不存在的物品 Returns: (is_valid, rejection_reason): 合法返回 (True, ""), 否则返回 (False, "拒绝原因") """ action = intent.get("intent", "") raw_target = intent.get("target") target = intent.get("target", "") or "" details = intent.get("details", "") or "" raw_input = intent.get("raw_input", "") or "" action_upper = str(action or "").upper() inventory = list(self.player.inventory) equipped_items = [v for v in self.player.equipment.values() if v] all_owned = set(inventory) | set(equipped_items) def normalize_item_phrase(text: str) -> str: cleaned = str(text or "").strip() cleaned = re.sub( r"^(?:喝掉|吃掉|使用|服用|装备|穿上|戴上|拿出|掏出|拔出|举起|喝|吃|用|掉)", "", cleaned, ) cleaned = re.sub(r"^(?:一瓶|一杯|一口|一个|一份|一块|一把)", "", cleaned) cleaned = re.sub(r"(?:照明|攻击|挥舞|挥动|一下|试试)$", "", cleaned) return cleaned.strip() normalized_target = normalize_item_phrase(target) def _resolve_dialogue_npc() -> NPCState | None: """Resolve TALK target from explicit target or alias mentions in free text.""" text_blob = f"{target} {details} {raw_input}".strip() if not text_blob: return None # 1) Exact NPC name match first. explicit_target = str(target or "").strip() if explicit_target and explicit_target in self.world.npcs: npc = self.world.npcs.get(explicit_target) if npc and npc.is_alive: return npc # 2) Name / occupation fuzzy match from free text (e.g. "和村长聊天"). alive_npcs = [npc for npc in self.world.npcs.values() if npc.is_alive] ranked_candidates: list[tuple[int, NPCState]] = [] for npc in alive_npcs: score = 0 if npc.name and npc.name in text_blob: score += 3 if explicit_target and ( (npc.name and explicit_target in npc.name) or (npc.name and npc.name in explicit_target) ): score += 2 if npc.occupation and npc.occupation in text_blob: score += 2 if explicit_target and npc.occupation and ( explicit_target in npc.occupation or npc.occupation in explicit_target ): score += 1 if score > 0: ranked_candidates.append((score, npc)) if ranked_candidates: ranked_candidates.sort(key=lambda item: item[0], reverse=True) return ranked_candidates[0][1] return None # --- 检测 1: USE_ITEM / EQUIP: target 必须在背包或装备中 --- if action_upper in ("USE_ITEM", "EQUIP") and target: if target not in all_owned: return False, f"你的背包中没有「{target}」，无法使用或装备。" if action_upper == "EQUIP" and target not in inventory: if target in equipped_items: return False, f"「{target}」已经装备在身上了。" return False, f"你的背包中没有「{target}」，无法装备。" # 体力耗尽时禁止移动和战斗 if action_upper in ("MOVE", "ATTACK", "COMBAT") and self.player.stamina <= 0: return False, "你精疲力竭，体力耗尽，无法行动。需要先在旅店或营地休息恢复体力。" if action_upper == "TRADE": if not isinstance(raw_target, dict): return False, "交易指令缺少商品信息，请从商店列表中选择要购买的物品。" merchant_name = str(raw_target.get("merchant") or "") item_name = str(raw_target.get("item") or raw_target.get("item_name") or "") if not merchant_name or not item_name: return False, "交易信息不完整，请重新从商店列表选择商品。" if action_upper in ("ATTACK", "COMBAT"): scene_actions = build_scene_actions(self, self.player.location) attack_targets = [ str(option.get("target")) for option in scene_actions if str(option.get("action_type", "")).upper() == "ATTACK" and isinstance(option.get("target"), str) and str(option.get("target")).strip() ] if target: if target not in attack_targets: if attack_targets: return ( False, f"当前无法攻击「{target}」。你现在可攻击的目标只有：{'、'.join(attack_targets)}。", ) return False, "当前场景没有可攻击目标，先观察环境或移动到有敌人的地点。" elif not attack_targets: return False, "当前场景没有可攻击目标，先观察环境或移动到有敌人的地点。" if action_upper == "MOVE" and target: current_loc = self.world.locations.get(self.player.location) target_loc = self.world.locations.get(str(target)) if current_loc is None or target_loc is None: return False, "你无法前往一个未注册的地点。" if str(target) not in current_loc.connected_to: return False, f"当前位置只能前往相邻地点，不能直接前往「{target}」。" if not target_loc.is_accessible: if target_loc.required_item and target_loc.required_item not in all_owned: return False, f"「{target}」尚未解锁，需要「{target_loc.required_item}」。" return False, f"「{target}」当前无法进入。" if action_upper == "VIEW_MAP": if not any("地图" in item for item in all_owned): return False, "你还没有获得可查看的地图。" # --- 检测 2: TALK: 对话对象必须在当前地点 --- if action_upper == "TALK": dialogue_npc = _resolve_dialogue_npc() if dialogue_npc is not None: if dialogue_npc.location != self.player.location: return ( False, f"「{dialogue_npc.name}」目前在「{dialogue_npc.location}」，你现在在「{self.player.location}」。" f"无法隔空对话，请先前往对方所在地点。" ) else: local_alive_npcs = [ npc.name for npc in self.world.npcs.values() if npc.is_alive and npc.location == self.player.location ] if not local_alive_npcs: return False, "这里没有可对话的角色，无法进行聊天。" # --- 检测 2: SKILL: 必须已习得 --- if action_upper == "SKILL" and target: if target not in self.player.skills: return False, f"你尚未习得技能「{target}」。" wants_overnight_rest = action_upper == "OVERNIGHT_REST" or ( action_upper == "REST" and any(keyword in f"{details} {raw_input}" for keyword in ("过夜", "睡一晚", "住一晚", "睡到天亮")) ) # --- 检测 3: REST/OVERNIGHT_REST: 当前位置必须允许休息 --- if action_upper in ("REST", "OVERNIGHT_REST"): current_loc = self.world.locations.get(self.player.location) if current_loc is None or not current_loc.rest_available: return False, "这里不适合休息，试着前往旅店或营地。" if wants_overnight_rest and not self.can_overnight_rest(): return False, "现在还不能在这里过夜。晚上七点后，且仅限旅店或溪边营地。" # --- 检测 4: 扫描 raw_input 中是否在使用上下文中引用了未拥有的已知物品 --- known_items: set[str] = set(self.world.item_registry.keys()) for event in self.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_known = { item for item in known_items if item not in all_owned and len(item) >= 2 } use_verbs = [ "使用", "用", "吃", "喝", "装备", "穿上", "戴上", "拿出", "掏出", "挥舞", "举起", "服用", "食用", "拔出", "拿起", ] for item_name in unavailable_known: if item_name not in raw_input: continue for verb in use_verbs: if verb + item_name in raw_input: return False, f"你的背包中没有「{item_name}」，无法{verb}。" # --- 检测 5: 检查 raw_input 中是否提及使用完全未知的物品 --- # 匹配常见的"使用物品"语句模式，提取物品名称并校验 extraction_patterns = [ (r'(?:掏出|拿出|拔出|举起)(.{2,8}?)(?:$|[，。！？,\s来])', "使用"), (r'吃(?:一个|一块|一份|了个|了一个)?(.{2,8}?)(?:$|[，。！？,\s来])', "吃"), (r'喝(?:一瓶|一杯|一口|了一瓶|了)?(.{2,8}?)(?:$|[，。！？,\s来])', "喝"), (r'用(.{2,6}?)(?:打|攻击|砍|刺|射|劈|挡|切|割)', "使用"), ] non_item_words = { "拳头", "双手", "手", "脚", "头", "身体", "魔法", "技能", "力量", "勇气", "智慧", "办法", "方法", "速度", "周围", "四周", } full_text = raw_input + " " + details for pattern, verb_desc in extraction_patterns: match = re.search(pattern, full_text) if match: mentioned = normalize_item_phrase(match.group(1).strip()) if not mentioned or mentioned in non_item_words: continue if mentioned not in all_owned: if normalized_target and ( mentioned in normalized_target or normalized_target in mentioned ): continue # 模糊匹配："剑" 可能是 "铁剑" 的简称 fuzzy_match = any( mentioned in normalize_item_phrase(owned) or normalize_item_phrase(owned) in mentioned for owned in all_owned ) if not fuzzy_match: return False, f"你并没有「{mentioned}」，请检查你的背包。" return True, "" def is_game_over(self) -> bool: """ 判断游戏是否结束。 结束条件： 1. HP <= 0（死亡） 2. 理智 <= 0（疯狂） 3. 触发终局标记 """ if self.player.hp <= 0: return True if self.player.sanity <= 0: return True if self.game_mode == "game_over": return True # 检查终局标记 if self.ending_flags.get("game_complete", False): return True return False def tick_time(self, player_intent: Optional[dict] = None) -> list[str]: """ 按动作消耗推进游戏时间。 设计思路： - 回合数递增 - 不同动作消耗不同的时间点数 - 累积点数达到阈值时，时间段按固定顺序轮转 - 每过一个完整日夜循环，天数+1 - 自动减少饱食度，模拟饥饿机制 - 结算状态效果持续时间 - 检查限时任务 Returns: tick_log: 本回合时间流逝引起的状态变化描述列表 """ tick_log: list[str] = [] self.turn += 1 action_units = self._estimate_time_cost_units(player_intent) if action_units <= 0: return tick_log old_clock = self.get_clock_display() old_time_of_day = self.world.time_of_day previous_units = self.world.time_progress_units total_units = previous_units + action_units day_rollovers = total_units // 144 self.world.time_progress_units = total_units % 144 if day_rollovers > 0: self.world.day_count += day_rollovers tick_log.append(f"新的一天！第 {self.world.day_count} 天") action_minutes = action_units * 10 self.elapsed_minutes_total += action_minutes self._sync_world_clock() new_clock = self.get_clock_display() if new_clock != old_clock: tick_log.append(f"时间流逝: {old_clock} → {new_clock}") if self.world.time_of_day != old_time_of_day: tick_log.append(f"时段变化: {old_time_of_day} → {self.world.time_of_day}") intent_name = str((player_intent or {}).get("intent", "")).upper() previous_elapsed_minutes = self.elapsed_minutes_total - action_minutes crossed_half_hours = ( self.elapsed_minutes_total // 30 - previous_elapsed_minutes // 30 ) hunger_delta = -int(max(crossed_half_hours, 0)) thirty_minute_blocks = max(1, (action_minutes + 29) // 30) if intent_name in {"MOVE"}: stamina_delta = -3 * thirty_minute_blocks elif intent_name in {"ATTACK", "COMBAT"}: stamina_delta = -5 * thirty_minute_blocks elif intent_name == "REST": stamina_delta = 12 * thirty_minute_blocks else: stamina_delta = 0 old_hunger = self.player.hunger self.player.hunger = clamp(self.player.hunger + hunger_delta, 0, 100) if self.player.hunger != old_hunger: tick_log.append(f"饱食度: {old_hunger} → {self.player.hunger}") old_stamina = self.player.stamina self.player.stamina = clamp(self.player.stamina + stamina_delta, 0, self.player.max_stamina) if self.player.stamina != old_stamina: tick_log.append(f"体力: {old_stamina} → {self.player.stamina}") crossed_half_days = self.elapsed_minutes_total // 720 - (self.elapsed_minutes_total - action_minutes) // 720 for _ in range(max(crossed_half_days, 0)): if self.player.hunger <= 0: old_hp = self.player.hp hp_loss = max(1, int(round(self.player.max_hp * 0.1))) self.player.hp = max(0, self.player.hp - hp_loss) tick_log.append(f"饥饿伤害: {old_hp} → {self.player.hp}") elif self.player.hunger > 80: old_hp = self.player.hp hp_gain = max(1, int(round(self.player.max_hp * 0.03))) self.player.hp = min(self.player.max_hp, self.player.hp + hp_gain) if self.player.hp != old_hp: tick_log.append(f"充足补给恢复: {old_hp} → {self.player.hp}") effect_log = self._apply_status_effects() tick_log.extend(effect_log) self._check_quest_deadlines() self._update_npc_schedules() self._update_environment_cycle(tick_log) return tick_log def _estimate_time_cost_units(self, player_intent: Optional[dict] = None) -> int: """Estimate how much in-world time a player action should consume.""" if not isinstance(player_intent, dict): return 3 action_type = str(player_intent.get("intent", "")).upper() return max(1, action_time_cost_minutes(action_type) // 10) def _determine_light_level(self) -> str: """Derive current light level from time of day and weather.""" base_levels = { "清晨": "柔和", "上午": "明亮", "正午": "明亮", "下午": "柔和", "黄昏": "昏暗", "夜晚": "幽暗", "深夜": "漆黑", } ordered_levels = ["明亮", "柔和", "昏暗", "幽暗", "漆黑"] weather_penalty = { "晴朗": 0, "多云": 0, "小雨": 1, "大雪": 1, "浓雾": 1, "暴风雨": 2, } base_level = base_levels.get(self.world.time_of_day, "柔和") current_index = ordered_levels.index(base_level) darker_by = weather_penalty.get(self.world.weather, 0) next_index = min(len(ordered_levels) - 1, current_index + darker_by) return ordered_levels[next_index] def _update_environment_cycle(self, tick_log: list[str]): """Advance light/weather and roll explicit environment events.""" self.pending_environment_event = None self._update_light_level_event(tick_log) self._maybe_shift_weather(tick_log) self._update_light_level_event(tick_log) self._roll_environment_event(tick_log) def _update_light_level_event(self, tick_log: list[str]): old_light = self.world.light_level new_light = self._determine_light_level() if new_light == old_light: return self.world.light_level = new_light event = EnvironmentEvent( event_id=f"light-{self.turn}", category="light", title=f"光照转为{new_light}", description=f"随着时间与天气变化，周围环境现在呈现出{new_light}的光照状态。", location=self.player.location, time_of_day=self.world.time_of_day, weather=self.world.weather, light_level=new_light, severity="medium" if new_light in {"幽暗", "漆黑"} else "low", prompt_hint="请在叙事中体现能见度、阴影和角色主观感受的变化。", ) self._register_environment_event( event, tick_log, inject_prompt=new_light in {"昏暗", "幽暗", "漆黑"}, ) def _maybe_shift_weather(self, tick_log: list[str]): """Occasionally shift weather to keep the environment dynamic.""" if self.elapsed_minutes_total - int(self.world.last_weather_change_minutes) < 180: return chance = 0.18 if self.world.time_of_day in {"清晨", "黄昏"} else 0.08 if random.random() >= chance: return weather_transitions = { "晴朗": ["多云", "小雨"], "多云": ["晴朗", "小雨", "浓雾"], "小雨": ["多云", "暴风雨", "浓雾"], "浓雾": ["多云", "小雨", "晴朗"], "暴风雨": ["小雨", "多云"], "大雪": ["多云"], } next_candidates = weather_transitions.get(self.world.weather, ["晴朗", "多云"]) new_weather = random.choice(next_candidates) if new_weather == self.world.weather: return loc = self.world.locations.get(self.player.location) weather_effects: dict[str, Any] = {"weather_change": new_weather} if loc and loc.location_type in {"wilderness", "dungeon"}: if new_weather == "暴风雨": weather_effects.update({"morale_change": -3, "sanity_change": -2}) elif new_weather == "浓雾": weather_effects.update({"sanity_change": -1}) elif new_weather == "晴朗" and self.world.weather in {"小雨", "浓雾", "暴风雨"}: weather_effects.update({"morale_change": 2}) event = EnvironmentEvent( event_id=f"weather-{self.turn}", category="weather", title=f"天气转为{new_weather}", description=f"周围的天象正在变化，空气与视野都随着天气转向{new_weather}。", location=self.player.location, time_of_day=self.world.time_of_day, weather=new_weather, light_level=self.world.light_level, severity="medium" if new_weather in {"暴风雨", "浓雾"} else "low", state_changes=weather_effects, prompt_hint="请把天气变化作为当前回合的重要氛围来源，影响角色观察和选择。", ) self._register_environment_event(event, tick_log, inject_prompt=True) self.world.light_level = self._determine_light_level() def _roll_environment_event(self, tick_log: list[str]): """Roll a structured environment event using explicit template filters.""" loc = self.world.locations.get(self.player.location) if loc is None or not self.environment_event_pool: return # 提高基础触发率，让环境事件更常进入叙事与决策反馈 chance = 0.15 if loc.danger_level >= 3: chance += 0.1 if self.world.light_level in {"幽暗", "漆黑"}: chance += 0.08 if self.world.weather in {"暴风雨", "浓雾"}: chance += 0.06 if random.random() >= chance: return candidates: list[dict[str, Any]] = [] for template in self.environment_event_pool: if template.get("location_types") and loc.location_type not in template["location_types"]: continue if template.get("time_slots") and self.world.time_of_day not in template["time_slots"]: continue if template.get("weathers") and self.world.weather not in template["weathers"]: continue if template.get("requires_rest_available") and not loc.rest_available: continue if loc.danger_level < int(template.get("min_danger", 0)): continue candidates.append(template) if not candidates: return template = random.choice(candidates) event = EnvironmentEvent( event_id=f"{template['event_id']}-{self.turn}", category=str(template.get("category", "environment")), title=str(template.get("title", "环境异动")), description=str(template.get("description", "")), location=self.player.location, time_of_day=self.world.time_of_day, weather=self.world.weather, light_level=self.world.light_level, severity=str(template.get("severity", "low")), state_changes=copy.deepcopy(template.get("state_changes", {})), prompt_hint=str(template.get("prompt_hint", "")), ) self._register_environment_event(event, tick_log, inject_prompt=True) def _register_environment_event( self, event: EnvironmentEvent, tick_log: list[str], *, inject_prompt: bool, ): """Persist an environment event, optionally inject it into the next prompt, and apply effects.""" self.world.recent_environment_events.append(event) self.world.recent_environment_events = self.world.recent_environment_events[-8:] if inject_prompt: self.pending_environment_event = event if event.category == "light": tick_log.append(f"光照变化: {event.title}") changes = copy.deepcopy(event.state_changes) changes.setdefault("world_event", event.title) change_log = self.apply_changes(changes) tick_log.extend(change_log) logger.info("环境事件触发: %s", event.title) def _apply_status_effects(self) -> list[str]: """每回合结算状态效果：应用修正、递减持续时间、移除过期效果 Returns: effect_log: 状态效果结算引起的变化描述列表 """ effect_log: list[str] = [] expired = [] for effect in self.player.status_effects: # 应用属性修正（每回合） if "hp" in effect.stat_modifiers: old_hp = self.player.hp self.player.hp = clamp( self.player.hp + effect.stat_modifiers["hp"], 0, self.player.max_hp, ) if old_hp != self.player.hp: effect_log.append(f"{effect.name}: HP {old_hp} → {self.player.hp}") if "mp" in effect.stat_modifiers: old_mp = self.player.mp self.player.mp = clamp( self.player.mp + effect.stat_modifiers["mp"], 0, self.player.max_mp, ) if old_mp != self.player.mp: effect_log.append(f"{effect.name}: MP {old_mp} → {self.player.mp}") if "sanity" in effect.stat_modifiers: old_sanity = self.player.sanity self.player.sanity = clamp( self.player.sanity + effect.stat_modifiers["sanity"], 0, 100, ) if old_sanity != self.player.sanity: effect_log.append(f"{effect.name}: 理智 {old_sanity} → {self.player.sanity}") # 感知、攻击、防御、速度、幸运、士气、饱食度等直接加减属性 for stat_key, stat_cn in [("perception", "感知"), ("attack", "攻击力"), ("defense", "防御力"), ("speed", "速度"), ("luck", "幸运"), ("morale", "士气"), ("hunger", "饱食度")]: if stat_key in effect.stat_modifiers: old_val = getattr(self.player, stat_key) max_val = 100 if stat_key in ("morale", "hunger") else None new_val = old_val + effect.stat_modifiers[stat_key] if max_val is not None: new_val = clamp(new_val, 0, max_val) setattr(self.player, stat_key, new_val) if old_val != new_val: effect_log.append(f"{effect.name}: {stat_cn} {old_val} → {new_val}") # 递减持续时间 if effect.duration > 0: effect.duration -= 1 if effect.duration <= 0: expired.append(effect) # duration == -1 表示永久效果，不递减 # 移除过期效果 for effect in expired: self.player.status_effects.remove(effect) effect_log.append(f"状态效果 '{effect.name}' 已过期并移除") logger.info(f"状态效果 '{effect.name}' 已过期") return effect_log def _check_quest_deadlines(self): """检查限时任务是否过期""" for quest in self.world.quests.values(): if quest.status == "active" and quest.turns_remaining > 0: quest.turns_remaining -= 1 if quest.turns_remaining <= 0: quest.status = "failed" logger.info(f"任务 '{quest.title}' 已超时失败！") def _update_npc_schedules(self): """根据当前时间段更新 NPC 位置""" for npc in self.world.npcs.values(): if not npc.is_alive: continue if self.world.time_of_day in npc.schedule: old_loc = npc.location new_loc = npc.schedule[self.world.time_of_day] if old_loc != new_loc: npc.location = new_loc # 更新地点的 NPC 列表 if old_loc in self.world.locations: loc = self.world.locations[old_loc] if npc.name in loc.npcs_present: loc.npcs_present.remove(npc.name) if new_loc in self.world.locations: loc = self.world.locations[new_loc] if npc.name not in loc.npcs_present: loc.npcs_present.append(npc.name) def _level_up(self): """ 角色升级逻辑。 每次升级： - 等级+1 - 扣除当前升级所需经验 - 下次升级所需经验提升 50% - 属性随机增长 - HP/MP 完全恢复 """ self.player.experience -= self.player.exp_to_next_level self.player.level += 1 self.player.exp_to_next_level = int(self.player.exp_to_next_level * 1.5) # 属性提升 self.player.max_hp += 10 self.player.max_mp += 5 self.player.attack += 2 self.player.defense += 1 self.player.speed += 1 self.player.perception += 1 # 升级后满血满蓝 self.player.hp = self.player.max_hp self.player.mp = self.player.max_mp logger.info( f"升级！等级: {self.player.level}, " f"HP: {self.player.max_hp}, MP: {self.player.max_mp}, " f"ATK: {self.player.attack}, DEF: {self.player.defense}" ) def get_death_narrative_context(self) -> str: """生成死亡结局的上下文信息（供 story_engine 使用）""" cause = "生命值归零" if self.player.hp <= 0 else "理智崩溃" last_event = self.event_log[-1].description if self.event_log else "未知" return ( f"玩家 {self.player.name} 因{cause}而倒下。\n" f"最后发生的事件: {last_event}\n" f"死亡次数: {self.player.death_count}\n" f"存活天数: {self.world.day_count}\n" f"最终善恶值: {self.player.karma}" ) def is_item_consumable(self, item_name: str) -> bool: """ 判断物品是否为消耗品（使用后会消失）。 规则： - item_registry 中 item_type == "consumable" 的物品是消耗品 - item_type == "material" 的物品也视为消耗品（合成材料，用完即消失） - 其他类型（weapon, armor, key, quest_item, misc 等）为可重复使用物品 - 未注册物品默认为非消耗品（更安全，避免误删） """ if item_name in self.world.item_registry: item_info = self.world.item_registry[item_name] return item_info.item_type in ("consumable", "material") # 对未注册物品，用关键词启发式判断 consumable_keywords = ["药水", "药剂", "食物", "面包", "烤肉", "麦酒", "草药", "卷轴", "炸弹", "手雷", "箭矢", "弹药", "丹药", "果实", "干粮", "肉干", "饮料", "汤", "符咒", "一次性"] for keyword in consumable_keywords: if keyword in item_name: return True return False # 默认为非消耗品 def get_available_actions(self) -> list[str]: """根据当前场景和状态返回可用的行动类型""" actions = ["观察", "对话", "移动"] # 当前场景信息 loc = self.world.locations.get(self.player.location) if loc: if loc.rest_available: actions.append("休息") if loc.shop_available: actions.append("交易") if loc.enemies: actions.append("战斗") if loc.available_items: actions.append("搜索") # 背包中有可用物品 if self.player.inventory: actions.append("使用物品") # 有技能可用 if self.player.skills: actions.append("使用技能") return actions def get_scene_summary(self) -> str: """获取当前场景的简短摘要（用于 UI 展示）""" loc = self.world.locations.get(self.player.location) desc = loc.description if loc else "未知区域" ambient = loc.ambient_description if loc else "" npcs = [ npc.name for npc in self.world.npcs.values() if npc.location == self.player.location and npc.is_alive ] npc_str = f"可见NPC: {'、'.join(npcs)}" if npcs else "" return f"{desc}\n{ambient}\n{npc_str}".strip() def get_consumable_rule_effects(self, item_name: str) -> dict[str, Any]: """Parse deterministic consumable effects from item metadata.""" item_info = self.world.item_registry.get(str(item_name)) if item_info is None or not item_info.usable: return {} effect_text = str(item_info.use_effect or "").strip() if not effect_text: return {} changes: dict[str, Any] = {} numeric_rules = [ (r"恢复\s*(\d+)\s*HP", "hp_change", 1), (r"恢复\s*(\d+)\s*MP", "mp_change", 1), (r"恢复\s*(\d+)\s*饱食度", "hunger_change", 1), (r"恢复\s*(\d+)\s*士气", "morale_change", 1), (r"恢复\s*(\d+)\s*理智", "sanity_change", 1), (r"降低\s*(\d+)\s*HP", "hp_change", -1), (r"降低\s*(\d+)\s*MP", "mp_change", -1), (r"降低\s*(\d+)\s*饱食度", "hunger_change", -1), (r"降低\s*(\d+)\s*士气", "morale_change", -1), (r"降低\s*(\d+)\s*理智", "sanity_change", -1), ] for pattern, key, sign in numeric_rules: for match in re.finditer(pattern, effect_text): amount = int(match.group(1)) * sign changes[key] = int(changes.get(key, 0)) + amount if "解除中毒状态" in effect_text: changes["status_effects_removed"] = ["中毒"] return changes def get_rest_rule_effects(self) -> dict[str, int]: """Return conservative default recovery when resting at a valid location.""" loc = self.world.locations.get(self.player.location) if loc is None or not loc.rest_available: return {} if loc.shop_available: base_recovery = { "hp_change": 20, "mp_change": 10, "morale_change": 10, "sanity_change": 6, } else: base_recovery = { "hp_change": 12, "mp_change": 6, "morale_change": 6, "sanity_change": 4, } filtered: dict[str, int] = {} if self.player.hp < self.player.max_hp: filtered["hp_change"] = base_recovery["hp_change"] if self.player.mp < self.player.max_mp: filtered["mp_change"] = base_recovery["mp_change"] if self.player.morale < 100: filtered["morale_change"] = base_recovery["morale_change"] if self.player.sanity < 100: filtered["sanity_change"] = base_recovery["sanity_change"] return filtered def get_environment_snapshot(self, limit: int = 3) -> dict[str, Any]: """Return a compact environment summary for UI and logs.""" loc = self.world.locations.get(self.player.location) recent_events = [ event.model_dump() for event in self.world.recent_environment_events[-limit:] ] return { "weather": self.world.weather, "light_level": self.world.light_level, "time_of_day": self.world.time_of_day, "season": self.world.season, "location_type": loc.location_type if loc else "unknown", "danger_level": loc.danger_level if loc else 0, "rest_available": bool(loc.rest_available) if loc else False, "shop_available": bool(loc.shop_available) if loc else False, "recent_events": recent_events, }