diff --git a/src/classes/action.py b/src/classes/action.py deleted file mode 100644 index d585c05..0000000 --- a/src/classes/action.py +++ /dev/null @@ -1,977 +0,0 @@ -from __future__ import annotations -from abc import ABC, abstractmethod -from typing import TYPE_CHECKING -from enum import Enum -import random - -from src.classes.root import Root, get_essence_types_for_root, extra_breakthrough_success_rate -from src.classes.region import Region, CultivateRegion, NormalRegion, CityRegion -from src.classes.alignment import Alignment -from src.classes.event import Event, NULL_EVENT -from src.classes.item import Item, items_by_name -from src.classes.prices import prices -from src.classes.hp_and_mp import HP_MAX_BY_REALM, MP_MAX_BY_REALM -from src.classes.battle import decide_battle, get_escape_success_rate -from src.utils.config import CONFIG - -if TYPE_CHECKING: - from src.classes.avatar import Avatar - from src.classes.world import World - from src.classes.animal import Animal - from src.classes.plant import Plant - - -def long_action(step_month: int): - """ - 长态动作装饰器，用于为动作类自动添加时间管理功能 - - Args: - step_month: 动作需要的月份数 - """ - def decorator(cls): - # 设置类属性，供基类使用 - cls._step_month = step_month - - def is_finished(self, *args, **kwargs) -> bool: - """ - 根据时间差判断动作是否完成 - 接受但忽略额外的参数以保持与其他动作类型的兼容性 - """ - if self.start_monthstamp is None: - return False - # 修正逻辑：使用 >= step_month - 1 而不是 >= step_month - # 这样1个月的动作在第1个月完成（时间差0 >= 0），10个月的动作在第10个月完成（时间差9 >= 9） - # 避免了原来多执行一个月的bug - return (self.world.month_stamp - self.start_monthstamp) >= self.step_month - 1 - - # 只添加 is_finished 方法 - cls.is_finished = is_finished - - return cls - - return decorator - -class Action(ABC): - """ - 角色可以执行的动作。 - 比如，移动、攻击、采集、建造、etc。 - """ - def __init__(self, avatar: Avatar, world: World): - """ - 传一个avatar的ref - 这样子实际执行的时候，可以知道avatar的能力和状态 - 可选传入world；若不传，则尝试从avatar.world获取。 - """ - self.avatar = avatar - self.world = world - - @abstractmethod - def execute(self) -> None: - pass - - - @property - def name(self) -> str: - """ - 获取动作名称 - """ - return str(self.__class__.__name__) - - -class DefineAction(Action): - def __init__(self, avatar: Avatar, world: World): - """ - 初始化动作，处理长态动作的属性设置 - """ - super().__init__(avatar, world) - - # 如果是长态动作，初始化相关属性 - if hasattr(self.__class__, '_step_month'): - self.step_month = self.__class__._step_month - self.start_monthstamp = None - - def execute(self, *args, **kwargs) -> None: - """ - 执行动作，处理时间管理逻辑，然后调用具体的_execute实现 - """ - # 如果是长态动作且第一次执行，记录开始时间 - if hasattr(self, 'step_month') and self.start_monthstamp is None: - self.start_monthstamp = self.world.month_stamp - - self._execute(*args, **kwargs) - - @abstractmethod - def _execute(self, *args, **kwargs) -> None: - """ - 具体的动作执行逻辑，由子类实现 - """ - pass - -class LLMAction(Action): - """ - 基于LLM的action，这种action一般是不需要实际的规则定义。 - 而是一种抽象的，仅有社会层面的后果的定义。 - 比如“折辱”“恶狠狠地盯着”“退婚”等 - 这种action会通过LLM生成并被执行，让NPC记忆并产生后果。 - 但是不需要规则侧做出反应来。 - """ - pass - -class ChunkActionMixin(): - """ - 动作片，可以理解成只是一种切分出来的动作。 - 不能被avatar直接执行，而是成为avatar执行某个动作的步骤。 - """ - pass - -class ActualActionMixin(): - """ - 实际的可以被规则/LLM调用，让avatar去执行的动作。 - 不一定是多个step，也有可能就一个step。 - - 新接口：子类必须实现 can_start/start/step/finish。 - """ - @abstractmethod - def can_start(self, **params) -> bool: - pass - - @abstractmethod - def start(self, **params) -> Event | None: - pass - - @abstractmethod - def step(self, **params) -> tuple["StepStatus", list[Event]]: - pass - - @abstractmethod - def finish(self, **params) -> list[Event]: - pass - - -class StepStatus(Enum): - RUNNING = "running" - COMPLETED = "completed" - - -class Move(DefineAction, ChunkActionMixin): - """ - 最基础的移动动作，在tile之间进行切换。 - """ - COMMENT = "移动到某个相对位置" - PARAMS = {"delta_x": "int", "delta_y": "int"} - def _execute(self, delta_x: int, delta_y: int) -> None: - """ - 移动到某个tile - """ - world = self.world - # 基于境界的移动步长：每轴最多移动 move_step_length 格 - step = getattr(self.avatar, "move_step_length", 1) - clamped_dx = max(-step, min(step, delta_x)) - clamped_dy = max(-step, min(step, delta_y)) - - new_x = self.avatar.pos_x + clamped_dx - new_y = self.avatar.pos_y + clamped_dy - - # 边界检查：越界则不移动 - if world.map.is_in_bounds(new_x, new_y): - self.avatar.pos_x = new_x - self.avatar.pos_y = new_y - target_tile = world.map.get_tile(new_x, new_y) - self.avatar.tile = target_tile - else: - # 超出边界：不改变位置与tile - pass - -class MoveToRegion(DefineAction, ActualActionMixin): - """ - 移动到某个region - """ - COMMENT = "移动到某个区域" - DOABLES_REQUIREMENTS = "任何时候都可以执行" - PARAMS = {"region": "region_name"} - def _execute(self, region: Region|str) -> None: - """ - 移动到某个region - """ - if isinstance(region, str): - from src.classes.region import regions_by_name - region = regions_by_name[region] - cur_loc = (self.avatar.pos_x, self.avatar.pos_y) - region_center_loc = region.center_loc - delta_x = region_center_loc[0] - cur_loc[0] - delta_y = region_center_loc[1] - cur_loc[1] - # 横纵向一次最多移动 move_step_length 格（可以同时横纵移动） - step = getattr(self.avatar, "move_step_length", 1) - delta_x = max(-step, min(step, delta_x)) - delta_y = max(-step, min(step, delta_y)) - Move(self.avatar, self.world).execute(delta_x, delta_y) - - def can_start(self, region: Region|str|None = None) -> bool: - return True - - def start(self, region: Region|str) -> Event: - if isinstance(region, str): - region_name = region - from src.classes.region import regions_by_name - if region in regions_by_name: - region_name = regions_by_name[region].name - elif hasattr(region, 'name'): - region_name = region.name - else: - region_name = str(region) - return Event(self.world.month_stamp, f"{self.avatar.name} 开始移动向 {region_name}") - - def step(self, region: Region|str) -> tuple[StepStatus, list[Event]]: - self.execute(region=region) - # 完成条件：到达目标区域 - if isinstance(region, str): - from src.classes.region import regions_by_name - region = regions_by_name[region] - done = self.avatar.is_in_region(region) - return (StepStatus.COMPLETED if done else StepStatus.RUNNING), [] - - def finish(self, region: Region|str) -> list[Event]: - return [] - - -class MoveToAvatar(DefineAction, ActualActionMixin): - """ - 朝另一个角色当前位置移动。 - """ - COMMENT = "移动到某个角色所在位置" - DOABLES_REQUIREMENTS = "任何时候都可以执行" - PARAMS = {"avatar_name": "str"} - - def _get_target(self, avatar_name: str): - """ - 根据名字查找目标角色；找不到返回 None。 - """ - for v in self.world.avatar_manager.avatars.values(): - if v.name == avatar_name: - return v - return None - - def _execute(self, avatar_name: str) -> None: - target = self._get_target(avatar_name) - if target is None: - return - cur_loc = (self.avatar.pos_x, self.avatar.pos_y) - target_loc = (target.pos_x, target.pos_y) - delta_x = target_loc[0] - cur_loc[0] - delta_y = target_loc[1] - cur_loc[1] - step = getattr(self.avatar, "move_step_length", 1) - delta_x = max(-step, min(step, delta_x)) - delta_y = max(-step, min(step, delta_y)) - Move(self.avatar, self.world).execute(delta_x, delta_y) - - def can_start(self, avatar_name: str|None = None) -> bool: - target = self._get_target(avatar_name) - if target is None: - return False - return True - - def start(self, avatar_name: str) -> Event: - target = self._get_target(avatar_name) - target_name = target.name if target is not None else avatar_name - return Event(self.world.month_stamp, f"{self.avatar.name} 开始移动向 {target_name}") - - def step(self, avatar_name: str) -> tuple[StepStatus, list[Event]]: - self.execute(avatar_name=avatar_name) - target = None - try: - target = self._get_target(avatar_name) - except Exception: - target = None - if target is None: - return StepStatus.COMPLETED, [] - done = self.avatar.tile == target.tile - return (StepStatus.COMPLETED if done else StepStatus.RUNNING), [] - - def finish(self, avatar_name: str) -> list[Event]: - return [] - - -@long_action(step_month=6) -class MoveAwayFromAvatar(DefineAction, ActualActionMixin): - """ - 持续远离指定角色，持续6个月。 - - 规则：每月尝试使与目标的曼哈顿距离增大一步 - - 任何时候都可以启动 - """ - COMMENT = "持续远离指定角色" - DOABLES_REQUIREMENTS = "任何时候都可以执行" - PARAMS = {"avatar_name": "AvatarName"} - - def _find_avatar_by_name(self, name: str) -> "Avatar|None": - for v in self.world.avatar_manager.avatars.values(): - if v.name == name: - return v - return None - - def _execute(self, avatar_name: str) -> None: - target = self._find_avatar_by_name(avatar_name) - if target is None: - return - # 计算远离方向：使曼哈顿距离尽量增大 - dx = 1 if self.avatar.pos_x >= target.pos_x else -1 - dy = 1 if self.avatar.pos_y >= target.pos_y else -1 - nx = self.avatar.pos_x + dx - ny = self.avatar.pos_y + dy - if self.world.map.is_in_bounds(nx, ny): - self.avatar.pos_x = nx - self.avatar.pos_y = ny - self.avatar.tile = self.world.map.get_tile(nx, ny) - - def can_start(self, avatar_name: str | None = None) -> bool: - return True - - def start(self, avatar_name: str) -> Event: - target_name = avatar_name - try: - t = self._find_avatar_by_name(avatar_name) - if t is not None: - target_name = t.name - except Exception: - pass - return Event(self.world.month_stamp, f"{self.avatar.name} 开始远离 {target_name}") - - def step(self, avatar_name: str) -> tuple[StepStatus, list[Event]]: - self.execute(avatar_name=avatar_name) - done = getattr(self, "is_finished")() - return (StepStatus.COMPLETED if done else StepStatus.RUNNING), [] - - def finish(self, avatar_name: str) -> list[Event]: - return [] - - -class MoveAwayFromRegion(DefineAction, ActualActionMixin): - COMMENT = "离开指定区域" - DOABLES_REQUIREMENTS = "任何时候都可以执行" - PARAMS = {"region": "RegionName"} - - def _execute(self, region: str) -> None: - # 简化：向地图边缘移动一步 - dx = 1 if self.avatar.pos_x < self.world.map.width - 1 else -1 - dy = 1 if self.avatar.pos_y < self.world.map.height - 1 else -1 - nx = max(0, min(self.world.map.width - 1, self.avatar.pos_x + dx)) - ny = max(0, min(self.world.map.height - 1, self.avatar.pos_y + dy)) - if self.world.map.is_in_bounds(nx, ny): - self.avatar.pos_x = nx - self.avatar.pos_y = ny - self.avatar.tile = self.world.map.get_tile(nx, ny) - - def can_start(self, region: str | None = None) -> bool: - return True - - def start(self, region: str) -> Event: - return Event(self.world.month_stamp, f"{self.avatar.name} 开始离开 {region}") - - def step(self, region: str) -> tuple[StepStatus, list[Event]]: - self.execute(region=region) - return StepStatus.COMPLETED, [] - - def finish(self, region: str) -> list[Event]: - return [] - - -class Escape(DefineAction, ActualActionMixin): - """ - 逃离：尝试从对方身边脱离（有成功率）。 - 成功：抢占并进入 MoveAwayFromAvatar(6个月)。 - 失败：抢占并进入 Battle。 - """ - COMMENT = "逃离对方（基于成功率判定）" - DOABLES_REQUIREMENTS = "任何时候都可以执行" - PARAMS = {"avatar_name": "AvatarName"} - - def _find_avatar_by_name(self, name: str) -> "Avatar|None": - for v in self.world.avatar_manager.avatars.values(): - if v.name == name: - return v - return None - - def _preempt_avatar(self, avatar: "Avatar") -> None: - avatar.clear_plans() - avatar.current_action = None - - def _add_event_pair(self, event: Event, initiator: "Avatar", target: "Avatar|None") -> None: - initiator.add_event(event) - if target is not None: - target.add_event(event, to_sidebar=False) - - def _execute(self, avatar_name: str) -> None: - target = self._find_avatar_by_name(avatar_name) - if target is None: - return - escape_rate = float(get_escape_success_rate(target, self.avatar)) - import random as _r - success = _r.random() < escape_rate - result_text = "成功" if success else "失败" - result_event = Event(self.world.month_stamp, f"{self.avatar.name} 试图从 {target.name} 逃离：{result_text}") - self._add_event_pair(result_event, initiator=self.avatar, target=target) - if success: - self._preempt_avatar(self.avatar) - self.avatar.load_decide_result_chain([("MoveAwayFromAvatar", {"avatar_name": avatar_name})], self.avatar.thinking, "") - start_event = self.avatar.commit_next_plan() - if start_event is not None: - self._add_event_pair(start_event, initiator=self.avatar, target=target) - else: - self._preempt_avatar(self.avatar) - self.avatar.load_decide_result_chain([("Battle", {"avatar_name": avatar_name})], self.avatar.thinking, "") - start_event = self.avatar.commit_next_plan() - if start_event is not None: - self._add_event_pair(start_event, initiator=self.avatar, target=target) - - def can_start(self, avatar_name: str | None = None) -> bool: - return True - - def start(self, avatar_name: str) -> Event: - target = self._find_avatar_by_name(avatar_name) - target_name = target.name if target is not None else avatar_name - return Event(self.world.month_stamp, f"{self.avatar.name} 尝试从 {target_name} 逃离") - - def step(self, avatar_name: str) -> tuple[StepStatus, list[Event]]: - self.execute(avatar_name=avatar_name) - return StepStatus.COMPLETED, [] - - def finish(self, avatar_name: str) -> list[Event]: - return [] - -@long_action(step_month=10) -class Cultivate(DefineAction, ActualActionMixin): - """ - 修炼动作，可以增加修仙进度。 - """ - COMMENT = "修炼，增进修为" - DOABLES_REQUIREMENTS = "在修炼区域中，修炼区域的灵气为角色的灵根之一，且角色未到瓶颈。" - PARAMS = {} - def _execute(self) -> None: - """ - 修炼 - 获得的exp增加取决于essence的对应灵根的大小。 - """ - root = self.avatar.root - essence = self.avatar.tile.region.essence - # 多元素：取与角色灵根任一匹配元素的最大密度 - essence_types = get_essence_types_for_root(root) - essence_density = max((essence.get_density(et) for et in essence_types), default=0) - exp = self.get_exp(essence_density) - self.avatar.cultivation_progress.add_exp(exp) - - def get_exp(self, essence_density: int) -> int: - """ - 根据essence的密度，计算获得的exp。 - 公式为：base * essence_density - """ - if self.avatar.cultivation_progress.is_in_bottleneck(): - return 0 - base = 100 - return base * essence_density - - def can_start(self) -> bool: - root = self.avatar.root - region = self.avatar.tile.region - essence_types = get_essence_types_for_root(root) - if not self.avatar.cultivation_progress.can_cultivate(): - return False - if not isinstance(region, CultivateRegion): - return False - if all(region.essence.get_density(et) == 0 for et in essence_types): - return False - return True - - def start(self) -> Event: - return Event(self.world.month_stamp, f"{self.avatar.name} 在 {self.avatar.tile.region.name} 开始修炼") - - def step(self) -> tuple[StepStatus, list[Event]]: - self.execute() - # 使用 long_action 注入的 is_finished - done = getattr(self, "is_finished")() - return (StepStatus.COMPLETED if done else StepStatus.RUNNING), [] - - def finish(self) -> list[Event]: - return [] - -# 突破境界class -@long_action(step_month=1) -class Breakthrough(DefineAction, ActualActionMixin): - """ - 突破境界。 - 成功率由 `CultivationProgress.get_breakthrough_success_rate()` 决定； - 失败时按 `CultivationProgress.get_breakthrough_fail_reduce_lifespan()` 减少寿元（年）。 - """ - COMMENT = "尝试突破境界（成功增加寿元上限，失败折损寿元上限；境界越高，成功率越低。）" - DOABLES_REQUIREMENTS = "角色处于瓶颈时" - PARAMS = {} - def calc_success_rate(self) -> float: - """ - 计算突破境界的成功率（由修为进度给出） - """ - base = self.avatar.cultivation_progress.get_breakthrough_success_rate() - bonus = extra_breakthrough_success_rate[self.avatar.root] - # 夹紧到 [0, 1] - return max(0.0, min(1.0, base + bonus)) - - def _execute(self) -> None: - """ - 突破境界 - """ - assert self.avatar.cultivation_progress.can_break_through() - success_rate = self.calc_success_rate() - # 记录本次尝试的基础信息 - self._success_rate_cached = success_rate - if random.random() < success_rate: - old_realm = self.avatar.cultivation_progress.realm - self.avatar.cultivation_progress.break_through() - new_realm = self.avatar.cultivation_progress.realm - - # 突破成功时更新HP和MP的最大值 - if new_realm != old_realm: - self._update_hp_mp_on_breakthrough(new_realm) - # 成功：确保最大寿元至少达到新境界的基线 - self.avatar.age.ensure_max_lifespan_at_least_realm_base(new_realm) - # 记录结果用于 finish 事件 - self._last_result = ("success", getattr(old_realm, "value", str(old_realm)), getattr(new_realm, "value", str(new_realm))) - else: - # 突破失败：减少最大寿元上限 - reduce_years = self.avatar.cultivation_progress.get_breakthrough_fail_reduce_lifespan() - self.avatar.age.decrease_max_lifespan(reduce_years) - # 记录结果用于 finish 事件 - self._last_result = ("fail", int(reduce_years)) - - def _update_hp_mp_on_breakthrough(self, new_realm): - """ - 突破境界时更新HP和MP的最大值并完全恢复 - - Args: - new_realm: 新的境界 - """ - new_max_hp = HP_MAX_BY_REALM.get(new_realm, 100) - new_max_mp = MP_MAX_BY_REALM.get(new_realm, 100) - - # 计算增加的最大值 - hp_increase = new_max_hp - self.avatar.hp.max - mp_increase = new_max_mp - self.avatar.mp.max - - # 更新最大值并恢复相应的当前值 - self.avatar.hp.add_max(hp_increase) - self.avatar.hp.recover(hp_increase) # 突破时完全恢复HP - self.avatar.mp.add_max(mp_increase) - self.avatar.mp.recover(mp_increase) # 突破时完全恢复MP - - def can_start(self) -> bool: - return self.avatar.cultivation_progress.can_break_through() - - def start(self) -> Event: - # 清理上次残留的结果状态（防御性） - self._last_result = None - self._success_rate_cached = None - return Event(self.world.month_stamp, f"{self.avatar.name} 开始尝试突破境界") - - def step(self) -> tuple[StepStatus, list[Event]]: - self.execute() - done = getattr(self, "is_finished")() - return (StepStatus.COMPLETED if done else StepStatus.RUNNING), [] - - def finish(self) -> list[Event]: - # 根据执行阶段记录的 _last_result 生成简洁完成事件 - res = getattr(self, "_last_result", None) - if isinstance(res, tuple) and res: - if res[0] == "success": - return [Event(self.world.month_stamp, f"{self.avatar.name} 突破成功")] - elif res[0] == "fail": - return [Event(self.world.month_stamp, f"{self.avatar.name} 突破失败")] - else: - raise ValueError(f"Unknown result: {res}") - -@long_action(step_month=6) -class Play(DefineAction, ActualActionMixin): - """ - 游戏娱乐动作，持续半年时间 - """ - COMMENT = "游戏娱乐，放松身心" - DOABLES_REQUIREMENTS = "任何时候都可以执行" - PARAMS = {} - - def _execute(self) -> None: - """ - 进行游戏娱乐活动 - """ - # 游戏娱乐的具体逻辑可以在这里实现 - # 比如增加心情值、减少压力等 - pass - - def can_start(self) -> bool: - return True - - def start(self) -> Event: - return Event(self.world.month_stamp, f"{self.avatar.name} 开始玩耍") - - def step(self) -> tuple[StepStatus, list[Event]]: - self.execute() - done = getattr(self, "is_finished")() - return (StepStatus.COMPLETED if done else StepStatus.RUNNING), [] - - def finish(self) -> list[Event]: - return [] - - -@long_action(step_month=6) -class Hunt(DefineAction, ActualActionMixin): - """ - 狩猎动作，在有动物的区域进行狩猎，持续6个月 - 可以获得动物对应的物品 - """ - COMMENT = "在当前区域狩猎动物，获取动物材料" - DOABLES_REQUIREMENTS = "在有动物的普通区域，且avatar的境界必须大于等于动物的境界" - PARAMS = {} - - def get_available_animals(self) -> list[Animal]: - """ - 获取avatar境界足够的动物 - """ - region = self.avatar.tile.region - avatar_realm = self.avatar.cultivation_progress.realm - return [animal for animal in region.animals if avatar_realm >= animal.realm] - - def _execute(self) -> None: - """ - 执行狩猎动作 - """ - success_rate = self.get_success_rate() - available_animals = self.get_available_animals() - if len(available_animals) == 0: - # TODO: 我的doable检查有问题，之后看看问题在哪里 - return - - if random.random() < success_rate: - # 成功狩猎，从avatar境界足够的动物中随机选择一种 - target_animal = random.choice(available_animals) - # 随机选择该动物的一种物品 - item = random.choice(target_animal.items) - self.avatar.add_item(item, 1) - - def get_success_rate(self) -> float: - """ - 获取狩猎成功率，预留接口，目前固定为100% - """ - return 1.0 # 100%成功率 - - def can_start(self) -> bool: - region = self.avatar.tile.region - if not isinstance(region, NormalRegion): - return False - available_animals = self.get_available_animals() - if len(available_animals) == 0: - return False - return True - - def start(self) -> Event: - region = self.avatar.tile.region - return Event(self.world.month_stamp, f"{self.avatar.name} 在 {region.name} 开始狩猎") - - def step(self) -> tuple[StepStatus, list[Event]]: - self.execute() - done = getattr(self, "is_finished")() - return (StepStatus.COMPLETED if done else StepStatus.RUNNING), [] - - def finish(self) -> list[Event]: - return [] - - -@long_action(step_month=6) -class Harvest(DefineAction, ActualActionMixin): - """ - 采集动作，在有植物的区域进行采集，持续6个月 - 可以获得植物对应的物品 - """ - COMMENT = "在当前区域采集植物，获取植物材料" - DOABLES_REQUIREMENTS = "在有植物的普通区域，且avatar的境界必须大于等于植物的境界" - PARAMS = {} - - def get_available_plants(self) -> list[Plant]: - """ - 获取avatar境界足够的植物 - """ - region = self.avatar.tile.region - avatar_realm = self.avatar.cultivation_progress.realm - return [plant for plant in region.plants if avatar_realm >= plant.realm] - - def _execute(self) -> None: - """ - 执行采集动作 - """ - success_rate = self.get_success_rate() - available_plants = self.get_available_plants() - if len(available_plants) == 0: - # TODO: 我的doable检查有问题，之后看看问题在哪里 - return - - if random.random() < success_rate: - # 成功采集，从avatar境界足够的植物中随机选择一种 - target_plant = random.choice(available_plants) - # 随机选择该植物的一种物品 - item = random.choice(target_plant.items) - self.avatar.add_item(item, 1) - - def get_success_rate(self) -> float: - """ - 获取采集成功率，预留接口，目前固定为100% - """ - return 1.0 # 100%成功率 - - def can_start(self) -> bool: - region = self.avatar.tile.region - if not isinstance(region, NormalRegion): - return False - avaliable_plants = self.get_available_plants() - if len(avaliable_plants) == 0: - return False - return True - - def start(self) -> Event: - region = self.avatar.tile.region - return Event(self.world.month_stamp, f"{self.avatar.name} 在 {region.name} 开始采集") - - def step(self) -> tuple[StepStatus, list[Event]]: - self.execute() - done = getattr(self, "is_finished")() - return (StepStatus.COMPLETED if done else StepStatus.RUNNING), [] - - def finish(self) -> list[Event]: - return [] - - -@long_action(step_month=1) -class Sold(DefineAction, ActualActionMixin): - """ - 在城镇出售指定名称的物品，一次性卖出持有的全部数量。 - 收益为 item_price * item_num，动作耗时1个月。 - """ - COMMENT = "在城镇出售持有的某类物品的全部" - DOABLES_REQUIREMENTS = "在城镇且背包非空" - PARAMS = {"item_name": "str"} - - def _execute(self, item_name: str) -> None: - region = self.avatar.tile.region - if not isinstance(region, CityRegion): - return - - # 找到物品 - item = items_by_name.get(item_name) - if item is None: - return - - # 检查持有数量 - quantity = self.avatar.get_item_quantity(item) - if quantity <= 0: - return - - # 计算价格并结算 - price_per = prices.get_price(item) - total_gain = price_per * quantity - - # 扣除物品并增加灵石 - removed = self.avatar.remove_item(item, quantity) - if not removed: - return - - self.avatar.magic_stone = self.avatar.magic_stone + total_gain - - def can_start(self, item_name: str|None = None) -> bool: - region = self.avatar.tile.region - if not isinstance(region, CityRegion): - return False - if item_name is None: - # 用于动作空间：只要背包非空即可 - return bool(self.avatar.items) - item = items_by_name.get(item_name) - if item is None: - return False - return self.avatar.get_item_quantity(item) > 0 - - def start(self, item_name: str) -> Event: - return Event(self.world.month_stamp, f"{self.avatar.name} 在城镇出售 {item_name}") - - def step(self, item_name: str) -> tuple[StepStatus, list[Event]]: - self.execute(item_name=item_name) - # 一次性动作 - return StepStatus.COMPLETED, [] - - def finish(self, item_name: str) -> list[Event]: - return [] - - -class Battle(DefineAction, ActualActionMixin): - COMMENT = "与目标进行对战，判定胜负" - DOABLES_REQUIREMENTS = "任何时候都可以执行" - PARAMS = {"avatar_name": "AvatarName"} - - def _get_target(self, avatar_name: str): - for v in self.world.avatar_manager.avatars.values(): - if v.name == avatar_name: - return v - return None - - def _execute(self, avatar_name: str) -> None: - target = self._get_target(avatar_name) - if target is None: - return - winner, loser, damage = decide_battle(self.avatar, target) - loser.hp.reduce(damage) - self._last_result = (winner.name, loser.name) - - def can_start(self, avatar_name: str | None = None) -> bool: - if avatar_name is None: - return False - return self._get_target(avatar_name) is not None - - def start(self, avatar_name: str) -> Event: - target = self._get_target(avatar_name) - target_name = target.name if target is not None else avatar_name - return Event(self.world.month_stamp, f"{self.avatar.name} 对 {target_name} 发起战斗") - - def step(self, avatar_name: str) -> tuple[StepStatus, list[Event]]: - self.execute(avatar_name=avatar_name) - return StepStatus.COMPLETED, [] - - def finish(self, avatar_name: str) -> list[Event]: - res = self._last_result - if isinstance(res, tuple) and len(res) == 2: - winner, loser = res - return [Event(self.world.month_stamp, f"{winner} 战胜了 {loser}")] - return [] - - -@long_action(step_month=3) -class PlunderMortals(DefineAction, ActualActionMixin): - """ - 在城镇对凡人进行搜刮，获取少量灵石。 - 仅邪阵营可执行。 - """ - COMMENT = "在城镇搜刮凡人，获取少量灵石" - DOABLES_REQUIREMENTS = "仅限城市区域，且角色阵营为‘邪’" - PARAMS = {} - GAIN = 20 - - def _execute(self) -> None: - region = self.avatar.tile.region - if not isinstance(region, CityRegion): - return - gain = self.GAIN - self.avatar.magic_stone = self.avatar.magic_stone + gain - - def can_start(self) -> bool: - region = self.avatar.tile.region - if not isinstance(region, CityRegion): - return False - return self.avatar.alignment == Alignment.EVIL - - def start(self) -> Event: - return Event(self.world.month_stamp, f"{self.avatar.name} 在城镇开始搜刮凡人") - - def step(self) -> tuple[StepStatus, list[Event]]: - self.execute() - return (StepStatus.COMPLETED if getattr(self, "is_finished")() else StepStatus.RUNNING), [] - - def finish(self) -> list[Event]: - return [] - - -@long_action(step_month=3) -class HelpMortals(DefineAction, ActualActionMixin): - """ - 在城镇帮助凡人，消耗少量灵石。 - 仅正阵营可执行。 - """ - COMMENT = "在城镇帮助凡人，消耗少量灵石" - DOABLES_REQUIREMENTS = "仅限城市区域，且角色阵营为‘正’，并且灵石足够" - PARAMS = {} - COST = 10 - - def _execute(self) -> None: - region = self.avatar.tile.region - if not isinstance(region, CityRegion): - return - cost = self.COST - if getattr(self.avatar.magic_stone, "value", 0) >= cost: - self.avatar.magic_stone = self.avatar.magic_stone - cost - - def can_start(self) -> bool: - region = self.avatar.tile.region - if not isinstance(region, CityRegion): - return False - if self.avatar.alignment != Alignment.RIGHTEOUS: - return False - cost = self.COST - return self.avatar.magic_stone >= cost - - def start(self) -> Event: - return Event(self.world.month_stamp, f"{self.avatar.name} 在城镇开始帮助凡人") - - def step(self) -> tuple[StepStatus, list[Event]]: - self.execute() - return (StepStatus.COMPLETED if getattr(self, "is_finished")() else StepStatus.RUNNING), [] - - def finish(self) -> list[Event]: - return [] - - -class Talk(DefineAction, ActualActionMixin): - """ - 攀谈：尝试与同区域内的某个NPC进行交谈。 - - can_start：同区域内存在其他NPC - - 发起后：随机寻找“同一tile”的NPC，若不存在则本次无法攀谈 - - 若找到，则进入 MutualAction: Conversation（允许建立关系） - """ - - COMMENT = "与同区域内的NPC发起攀谈，若同一tile有人则进入交谈" - DOABLES_REQUIREMENTS = "同区域内存在其他NPC" - PARAMS = {} - - def _get_same_region_others(self) -> list["Avatar"]: - return self.world.avatar_manager.get_avatars_in_same_region(self.avatar) - - def _get_same_tile_others(self) -> list["Avatar"]: - same_tile: list["Avatar"] = [] - my_tile = self.avatar.tile - if my_tile is None: - return [] - for v in self.world.avatar_manager.avatars.values(): - if v is self.avatar or v.tile is None: - continue - if v.tile == my_tile: - same_tile.append(v) - return same_tile - - def _execute(self) -> None: - # Talk 本身不做长期效果，主要在 step 中驱动 Conversation - return - - def can_start(self) -> bool: - # 是否同区域存在其他NPC（用于展示在动作空间） - return len(self._get_same_region_others()) > 0 - - def start(self) -> Event: - self.same_region_others = self._get_same_region_others() - # 记录开始事件 - return Event(self.world.month_stamp, f"{self.avatar.name} 尝试与同区域的他人攀谈") - - def step(self) -> tuple[StepStatus, list[Event]]: - import random - target = random.choice(self.same_region_others) - - # 进入交谈：由概率决定本次是否允许建立关系 - from src.classes.mutual_action import Conversation - # 由配置决定本次是否有“有机会进入关系”标记 - prob = CONFIG.social.talk_into_relation_probability - can_into_relation = (random.random() < prob) - - conv = Conversation(self.avatar, self.world) - # 启动事件写入历史，不入侧边栏 - conv.start(target_avatar=target) - conv.step(target_avatar=target, can_into_relation=can_into_relation) - return StepStatus.COMPLETED, [] - - def finish(self) -> list[Event]: - return [] diff --git a/src/classes/action/__init__.py b/src/classes/action/__init__.py new file mode 100644 index 0000000..ff59b69 --- /dev/null +++ b/src/classes/action/__init__.py @@ -0,0 +1,89 @@ +from __future__ import annotations + +# 基类与工具 +from .action import ( + Action, + DefineAction, + LLMAction, + ChunkActionMixin, + ActualActionMixin, + InstantAction, + TimedAction, + long_action, +) +from .registry import register_action + +# 具体动作（按文件拆分） +from .move import Move +from .move_to_region import MoveToRegion +from .move_to_avatar import MoveToAvatar +from .move_away_from_avatar import MoveAwayFromAvatar +from .move_away_from_region import MoveAwayFromRegion +from .escape import Escape +from .cultivate import Cultivate +from .breakthrough import Breakthrough +from .play import Play +from .hunt import Hunt +from .harvest import Harvest +from .sold import SellItems +from .battle import Battle +from .plunder_mortals import PlunderMortals +from .help_mortals import HelpMortals +from .talk import Talk + +# 注册到 ActionRegistry（标注是否为实际可执行动作） +register_action(actual=False)(Action) +register_action(actual=False)(DefineAction) +register_action(actual=False)(LLMAction) +register_action(actual=False)(ChunkActionMixin) +register_action(actual=False)(ActualActionMixin) +register_action(actual=False)(InstantAction) +register_action(actual=False)(TimedAction) + +register_action(actual=False)(Move) +register_action(actual=True)(MoveToRegion) +register_action(actual=True)(MoveToAvatar) +register_action(actual=True)(MoveAwayFromAvatar) +register_action(actual=True)(MoveAwayFromRegion) +register_action(actual=False)(Escape) +register_action(actual=True)(Cultivate) +register_action(actual=True)(Breakthrough) +register_action(actual=True)(Play) +register_action(actual=True)(Hunt) +register_action(actual=True)(Harvest) +register_action(actual=True)(SellItems) +register_action(actual=False)(Battle) +register_action(actual=True)(PlunderMortals) +register_action(actual=True)(HelpMortals) +register_action(actual=True)(Talk) + +__all__ = [ + # 基类 + "Action", + "DefineAction", + "LLMAction", + "ChunkActionMixin", + "ActualActionMixin", + "InstantAction", + "TimedAction", + "long_action", + # 派生类 + "Move", + "MoveToRegion", + "MoveToAvatar", + "MoveAwayFromAvatar", + "MoveAwayFromRegion", + "Escape", + "Cultivate", + "Breakthrough", + "Play", + "Hunt", + "Harvest", + "SellItems", + "Battle", + "PlunderMortals", + "HelpMortals", + "Talk", +] + + diff --git a/src/classes/action/action.py b/src/classes/action/action.py new file mode 100644 index 0000000..96e76b3 --- /dev/null +++ b/src/classes/action/action.py @@ -0,0 +1,173 @@ +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import TYPE_CHECKING + +from src.classes.event import Event +from src.classes.action_runtime import ActionResult, ActionStatus + +if TYPE_CHECKING: + from src.classes.avatar import Avatar + from src.classes.world import World + + +def long_action(step_month: int): + """ + 长态动作装饰器，用于为动作类自动添加时间管理功能 + + Args: + step_month: 动作需要的月份数 + """ + def decorator(cls): + # 设置类属性，供基类使用 + cls._step_month = step_month + + def is_finished(self, *args, **kwargs) -> bool: + """ + 根据时间差判断动作是否完成 + 接受但忽略额外的参数以保持与其他动作类型的兼容性 + """ + if self.start_monthstamp is None: + return False + # 修正逻辑：使用 >= step_month - 1 而不是 >= step_month + # 这样1个月的动作在第1个月完成（时间差0 >= 0），10个月的动作在第10个月完成（时间差9 >= 9） + # 避免了原来多执行一个月的bug + return (self.world.month_stamp - self.start_monthstamp) >= self.step_month - 1 + + # 只添加 is_finished 方法 + cls.is_finished = is_finished + + return cls + + return decorator + + +class Action(ABC): + """ + 角色可以执行的动作。 + 比如，移动、攻击、采集、建造、etc。 + """ + + def __init__(self, avatar: Avatar, world: World): + """ + 传一个avatar的ref + 这样子实际执行的时候，可以知道avatar的能力和状态 + 可选传入world；若不传，则尝试从avatar.world获取。 + """ + self.avatar = avatar + self.world = world + + @abstractmethod + def execute(self) -> None: + pass + + @property + def name(self) -> str: + """ + 获取动作名称 + """ + return str(self.__class__.__name__) + + +class DefineAction(Action): + def __init__(self, avatar: Avatar, world: World): + """ + 初始化动作，处理长态动作的属性设置 + """ + super().__init__(avatar, world) + + # 如果是长态动作，初始化相关属性 + if hasattr(self.__class__, '_step_month'): + self.step_month = self.__class__._step_month + self.start_monthstamp = None + + def execute(self, *args, **kwargs) -> None: + """ + 执行动作，处理时间管理逻辑，然后调用具体的_execute实现 + """ + # 如果是长态动作且第一次执行，记录开始时间 + if hasattr(self, 'step_month') and self.start_monthstamp is None: + self.start_monthstamp = self.world.month_stamp + + self._execute(*args, **kwargs) + + @abstractmethod + def _execute(self, *args, **kwargs) -> None: + """ + 具体的动作执行逻辑，由子类实现 + """ + pass + + +class LLMAction(Action): + """ + 基于LLM的action，这种action一般是不需要实际的规则定义。 + 而是一种抽象的，仅有社会层面的后果的定义。 + 比如“折辱”“恶狠狠地盯着”“退婚”等 + 这种action会通过LLM生成并被执行，让NPC记忆并产生后果。 + 但是不需要规则侧做出反应来。 + """ + + pass + + +class ChunkActionMixin(): + """ + 动作片，可以理解成只是一种切分出来的动作。 + 不能被avatar直接执行，而是成为avatar执行某个动作的步骤。 + """ + + pass + + +class ActualActionMixin(): + """ + 实际的可以被规则/LLM调用，让avatar去执行的动作。 + 不一定是多个step，也有可能就一个step。 + + 新接口：子类必须实现 can_start/start/step/finish。 + """ + + @abstractmethod + def can_start(self, **params) -> bool: + return True + + @abstractmethod + def start(self, **params) -> Event | None: + return None + + @abstractmethod + def step(self, **params) -> ActionResult: + ... + + @abstractmethod + def finish(self, **params) -> list[Event]: + return [] + + +class InstantAction(DefineAction, ActualActionMixin): + """ + 一次性动作：在一次 step 内完成。子类仅需实现 _execute。 + """ + + def step(self, **params) -> ActionResult: + self._execute(**params) + return ActionResult(status=ActionStatus.COMPLETED, events=[]) + + +class TimedAction(DefineAction, ActualActionMixin): + """ + 长态动作：通过类属性 duration_months 控制持续时间。 + 子类实现 _execute 作为每月的执行逻辑。 + """ + + duration_months: int = 1 + + def step(self, **params) -> ActionResult: + if getattr(self, 'start_monthstamp', None) is None: + self.start_monthstamp = self.world.month_stamp + self._execute(**params) + done = (self.world.month_stamp - self.start_monthstamp) >= (self.duration_months - 1) + return ActionResult(status=(ActionStatus.COMPLETED if done else ActionStatus.RUNNING), events=[]) + + diff --git a/src/classes/action/battle.py b/src/classes/action/battle.py new file mode 100644 index 0000000..2f6efef --- /dev/null +++ b/src/classes/action/battle.py @@ -0,0 +1,46 @@ +from __future__ import annotations + +from src.classes.action import InstantAction +from src.classes.event import Event +from src.classes.battle import decide_battle + + +class Battle(InstantAction): + COMMENT = "与目标进行对战，判定胜负" + DOABLES_REQUIREMENTS = "任何时候都可以执行" + PARAMS = {"avatar_name": "AvatarName"} + + def _get_target(self, avatar_name: str): + for v in self.world.avatar_manager.avatars.values(): + if v.name == avatar_name: + return v + return None + + def _execute(self, avatar_name: str) -> None: + target = self._get_target(avatar_name) + if target is None: + return + winner, loser, damage = decide_battle(self.avatar, target) + loser.hp.reduce(damage) + self._last_result = (winner.name, loser.name) + + def can_start(self, avatar_name: str | None = None) -> bool: + if avatar_name is None: + return False + return self._get_target(avatar_name) is not None + + def start(self, avatar_name: str) -> Event: + target = self._get_target(avatar_name) + target_name = target.name if target is not None else avatar_name + return Event(self.world.month_stamp, f"{self.avatar.name} 对 {target_name} 发起战斗") + + # InstantAction 已实现 step 完成 + + def finish(self, avatar_name: str) -> list[Event]: + res = self._last_result + if isinstance(res, tuple) and len(res) == 2: + winner, loser = res + return [Event(self.world.month_stamp, f"{winner} 战胜了 {loser}")] + return [] + + diff --git a/src/classes/action/breakthrough.py b/src/classes/action/breakthrough.py new file mode 100644 index 0000000..398e12c --- /dev/null +++ b/src/classes/action/breakthrough.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import random +from src.classes.action import TimedAction +from src.classes.event import Event +from src.classes.hp_and_mp import HP_MAX_BY_REALM, MP_MAX_BY_REALM +from src.classes.root import extra_breakthrough_success_rate + + +class Breakthrough(TimedAction): + """ + 突破境界。 + 成功率由 `CultivationProgress.get_breakthrough_success_rate()` 决定； + 失败时按 `CultivationProgress.get_breakthrough_fail_reduce_lifespan()` 减少寿元（年）。 + """ + + COMMENT = "尝试突破境界（成功增加寿元上限，失败折损寿元上限；境界越高，成功率越低。）" + DOABLES_REQUIREMENTS = "角色处于瓶颈时" + PARAMS = {} + + def calc_success_rate(self) -> float: + """ + 计算突破境界的成功率（由修为进度给出） + """ + base = self.avatar.cultivation_progress.get_breakthrough_success_rate() + bonus = extra_breakthrough_success_rate[self.avatar.root] + # 夹紧到 [0, 1] + return max(0.0, min(1.0, base + bonus)) + + duration_months = 1 + + def _execute(self) -> None: + """ + 突破境界 + """ + assert self.avatar.cultivation_progress.can_break_through() + success_rate = self.calc_success_rate() + # 记录本次尝试的基础信息 + self._success_rate_cached = success_rate + if random.random() < success_rate: + old_realm = self.avatar.cultivation_progress.realm + self.avatar.cultivation_progress.break_through() + new_realm = self.avatar.cultivation_progress.realm + + # 突破成功时更新HP和MP的最大值 + if new_realm != old_realm: + self._update_hp_mp_on_breakthrough(new_realm) + # 成功：确保最大寿元至少达到新境界的基线 + self.avatar.age.ensure_max_lifespan_at_least_realm_base(new_realm) + # 记录结果用于 finish 事件 + self._last_result = ( + "success", + getattr(old_realm, "value", str(old_realm)), + getattr(new_realm, "value", str(new_realm)), + ) + else: + # 突破失败：减少最大寿元上限 + reduce_years = self.avatar.cultivation_progress.get_breakthrough_fail_reduce_lifespan() + self.avatar.age.decrease_max_lifespan(reduce_years) + # 记录结果用于 finish 事件 + self._last_result = ("fail", int(reduce_years)) + + def _update_hp_mp_on_breakthrough(self, new_realm): + """ + 突破境界时更新HP和MP的最大值并完全恢复 + + Args: + new_realm: 新的境界 + """ + new_max_hp = HP_MAX_BY_REALM.get(new_realm, 100) + new_max_mp = MP_MAX_BY_REALM.get(new_realm, 100) + + # 计算增加的最大值 + hp_increase = new_max_hp - self.avatar.hp.max + mp_increase = new_max_mp - self.avatar.mp.max + + # 更新最大值并恢复相应的当前值 + self.avatar.hp.add_max(hp_increase) + self.avatar.hp.recover(hp_increase) # 突破时完全恢复HP + self.avatar.mp.add_max(mp_increase) + self.avatar.mp.recover(mp_increase) # 突破时完全恢复MP + + def can_start(self) -> bool: + return self.avatar.cultivation_progress.can_break_through() + + def start(self) -> Event: + # 清理上次残留的结果状态（防御性） + self._last_result = None + self._success_rate_cached = None + return Event(self.world.month_stamp, f"{self.avatar.name} 开始尝试突破境界") + + # TimedAction 已统一 step 逻辑 + + def finish(self) -> list[Event]: + # 根据执行阶段记录的 _last_result 生成简洁完成事件 + res = getattr(self, "_last_result", None) + if isinstance(res, tuple) and res: + if res[0] == "success": + return [Event(self.world.month_stamp, f"{self.avatar.name} 突破成功")] + elif res[0] == "fail": + return [Event(self.world.month_stamp, f"{self.avatar.name} 突破失败")] + else: + raise ValueError(f"Unknown result: {res}") + + diff --git a/src/classes/action/config.py b/src/classes/action/config.py new file mode 100644 index 0000000..51a6eaf --- /dev/null +++ b/src/classes/action/config.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Any + +from src.utils.config import CONFIG + + +class ActionConfig: + """ + 读取动作相关配置的轻量封装。 + + 用法： + ActionConfig.get("actions.duration.play", default=6) + 支持以点号分隔的路径逐层 getattr，读取不到时返回 default。 + """ + @staticmethod + def get(path: str, default: Any = None) -> Any: + cur = CONFIG + for part in path.split("."): + cur = getattr(cur, part, None) + if cur is None: + return default + return cur + + diff --git a/src/classes/action/cultivate.py b/src/classes/action/cultivate.py new file mode 100644 index 0000000..4fc40eb --- /dev/null +++ b/src/classes/action/cultivate.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import random +from src.classes.action import TimedAction +from src.classes.event import Event +from src.classes.root import get_essence_types_for_root +from src.classes.region import CultivateRegion + + +class Cultivate(TimedAction): + """ + 修炼动作，可以增加修仙进度。 + """ + + COMMENT = "修炼，增进修为" + DOABLES_REQUIREMENTS = "在修炼区域中，修炼区域的灵气为角色的灵根之一，且角色未到瓶颈。" + PARAMS = {} + + duration_months = 10 + + def _execute(self) -> None: + """ + 修炼 + 获得的exp增加取决于essence的对应灵根的大小。 + """ + root = self.avatar.root + essence = self.avatar.tile.region.essence + # 多元素：取与角色灵根任一匹配元素的最大密度 + essence_types = get_essence_types_for_root(root) + essence_density = max((essence.get_density(et) for et in essence_types), default=0) + exp = self.get_exp(essence_density) + self.avatar.cultivation_progress.add_exp(exp) + + def get_exp(self, essence_density: int) -> int: + """ + 根据essence的密度，计算获得的exp。 + 公式为：base * essence_density + """ + if self.avatar.cultivation_progress.is_in_bottleneck(): + return 0 + base = 100 + return base * essence_density + + def can_start(self) -> bool: + root = self.avatar.root + region = self.avatar.tile.region + essence_types = get_essence_types_for_root(root) + if not self.avatar.cultivation_progress.can_cultivate(): + return False + if not isinstance(region, CultivateRegion): + return False + if all(region.essence.get_density(et) == 0 for et in essence_types): + return False + return True + + def start(self) -> Event: + return Event(self.world.month_stamp, f"{self.avatar.name} 在 {self.avatar.tile.region.name} 开始修炼") + + # TimedAction 已统一 step 逻辑 + + def finish(self) -> list[Event]: + return [] + + diff --git a/src/classes/action/escape.py b/src/classes/action/escape.py new file mode 100644 index 0000000..c481ead --- /dev/null +++ b/src/classes/action/escape.py @@ -0,0 +1,72 @@ +from __future__ import annotations + +from src.classes.action import InstantAction +from src.classes.event import Event +from src.classes.battle import get_escape_success_rate +from src.classes.action.event_helper import EventHelper + + +class Escape(InstantAction): + """ + 逃离：尝试从对方身边脱离（有成功率）。 + 成功：抢占并进入 MoveAwayFromAvatar(6个月)。 + 失败：抢占并进入 Battle。 + """ + + COMMENT = "逃离对方（基于成功率判定）" + DOABLES_REQUIREMENTS = "任何时候都可以执行" + PARAMS = {"avatar_name": "AvatarName"} + + def _find_avatar_by_name(self, name: str) -> "Avatar|None": + for v in self.world.avatar_manager.avatars.values(): + if v.name == name: + return v + return None + + def _preempt_avatar(self, avatar: "Avatar") -> None: + avatar.clear_plans() + avatar.current_action = None + + def _add_event_pair(self, event: Event, initiator: "Avatar", target: "Avatar|None") -> None: + initiator.add_event(event) + if target is not None: + target.add_event(event, to_sidebar=False) + + def _execute(self, avatar_name: str) -> None: + target = self._find_avatar_by_name(avatar_name) + if target is None: + return + escape_rate = float(get_escape_success_rate(target, self.avatar)) + import random as _r + + success = _r.random() < escape_rate + result_text = "成功" if success else "失败" + result_event = Event(self.world.month_stamp, f"{self.avatar.name} 试图从 {target.name} 逃离：{result_text}") + EventHelper.push_pair(result_event, initiator=self.avatar, target=target, to_sidebar_once=True) + if success: + self._preempt_avatar(self.avatar) + self.avatar.load_decide_result_chain([("MoveAwayFromAvatar", {"avatar_name": avatar_name})], self.avatar.thinking, "") + start_event = self.avatar.commit_next_plan() + if start_event is not None: + EventHelper.push_pair(start_event, initiator=self.avatar, target=target, to_sidebar_once=True) + else: + self._preempt_avatar(self.avatar) + self.avatar.load_decide_result_chain([("Battle", {"avatar_name": avatar_name})], self.avatar.thinking, "") + start_event = self.avatar.commit_next_plan() + if start_event is not None: + EventHelper.push_pair(start_event, initiator=self.avatar, target=target, to_sidebar_once=True) + + def can_start(self, avatar_name: str | None = None) -> bool: + return True + + def start(self, avatar_name: str) -> Event: + target = self._find_avatar_by_name(avatar_name) + target_name = target.name if target is not None else avatar_name + return Event(self.world.month_stamp, f"{self.avatar.name} 尝试从 {target_name} 逃离") + + # InstantAction 已实现 step 完成 + + def finish(self, avatar_name: str) -> list[Event]: + return [] + + diff --git a/src/classes/action/event_helper.py b/src/classes/action/event_helper.py new file mode 100644 index 0000000..ff71368 --- /dev/null +++ b/src/classes/action/event_helper.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from typing import Optional + +from src.classes.event import Event + + +class EventHelper: + """ + 事件推送辅助：统一“侧栏只推一次、双方都写历史”的约定。 + + - push_pair: 向发起者与目标同时写入事件；默认仅在发起者侧进入侧栏。 + - push_self: 仅向自身写入事件，可控制是否进入侧栏。 + """ + @staticmethod + def push_pair(event: Event, initiator: "Avatar", target: Optional["Avatar"], *, to_sidebar_once: bool = True) -> None: + initiator.add_event(event, to_sidebar=True) + if target is not None: + target.add_event(event, to_sidebar=(not to_sidebar_once), to_history=True) + + @staticmethod + def push_self(event: Event, avatar: "Avatar", *, to_sidebar: bool = True) -> None: + avatar.add_event(event, to_sidebar=to_sidebar) + + diff --git a/src/classes/action/harvest.py b/src/classes/action/harvest.py new file mode 100644 index 0000000..40c9182 --- /dev/null +++ b/src/classes/action/harvest.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import random +from src.classes.action import TimedAction +from src.classes.event import Event +from src.classes.region import NormalRegion +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from src.classes.plant import Plant + + +class Harvest(TimedAction): + """ + 采集动作，在有植物的区域进行采集，持续6个月 + 可以获得植物对应的物品 + """ + + COMMENT = "在当前区域采集植物，获取植物材料" + DOABLES_REQUIREMENTS = "在有植物的普通区域，且avatar的境界必须大于等于植物的境界" + PARAMS = {} + + def get_available_plants(self) -> list[Plant]: + """ + 获取avatar境界足够的植物 + """ + region = self.avatar.tile.region + avatar_realm = self.avatar.cultivation_progress.realm + return [plant for plant in region.plants if avatar_realm >= plant.realm] + + duration_months = 6 + + def _execute(self) -> None: + """ + 执行采集动作 + """ + success_rate = self.get_success_rate() + available_plants = self.get_available_plants() + if len(available_plants) == 0: + return + + if random.random() < success_rate: + # 成功采集，从avatar境界足够的植物中随机选择一种 + target_plant = random.choice(available_plants) + # 随机选择该植物的一种物品 + item = random.choice(target_plant.items) + self.avatar.add_item(item, 1) + + def get_success_rate(self) -> float: + """ + 获取采集成功率，预留接口，目前固定为100% + """ + return 1.0 # 100%成功率 + + def can_start(self) -> bool: + region = self.avatar.tile.region + if not isinstance(region, NormalRegion): + return False + avaliable_plants = self.get_available_plants() + if len(avaliable_plants) == 0: + return False + return True + + def start(self) -> Event: + region = self.avatar.tile.region + return Event(self.world.month_stamp, f"{self.avatar.name} 在 {region.name} 开始采集") + + # TimedAction 已统一 step 逻辑 + + def finish(self) -> list[Event]: + return [] + + diff --git a/src/classes/action/help_mortals.py b/src/classes/action/help_mortals.py new file mode 100644 index 0000000..f715c41 --- /dev/null +++ b/src/classes/action/help_mortals.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +from src.classes.action import TimedAction +from src.classes.event import Event +from src.classes.region import CityRegion +from src.classes.alignment import Alignment + + +class HelpMortals(TimedAction): + """ + 在城镇帮助凡人，消耗少量灵石。 + 仅正阵营可执行。 + """ + + COMMENT = "在城镇帮助凡人，消耗少量灵石" + DOABLES_REQUIREMENTS = "仅限城市区域，且角色阵营为‘正’，并且灵石足够" + PARAMS = {} + COST = 10 + + duration_months = 3 + + def _execute(self) -> None: + region = self.avatar.tile.region + if not isinstance(region, CityRegion): + return + cost = self.COST + if getattr(self.avatar.magic_stone, "value", 0) >= cost: + self.avatar.magic_stone = self.avatar.magic_stone - cost + + def can_start(self) -> bool: + region = self.avatar.tile.region + if not isinstance(region, CityRegion): + return False + if self.avatar.alignment != Alignment.RIGHTEOUS: + return False + cost = self.COST + return self.avatar.magic_stone >= cost + + def start(self) -> Event: + return Event(self.world.month_stamp, f"{self.avatar.name} 在城镇开始帮助凡人") + + # TimedAction 已统一 step 逻辑 + + def finish(self) -> list[Event]: + return [] + + diff --git a/src/classes/action/hunt.py b/src/classes/action/hunt.py new file mode 100644 index 0000000..27d648e --- /dev/null +++ b/src/classes/action/hunt.py @@ -0,0 +1,73 @@ +from __future__ import annotations + +import random +from src.classes.action import TimedAction +from src.classes.event import Event +from src.classes.region import NormalRegion +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from src.classes.animal import Animal + + +class Hunt(TimedAction): + """ + 狩猎动作，在有动物的区域进行狩猎，持续6个月 + 可以获得动物对应的物品 + """ + + COMMENT = "在当前区域狩猎动物，获取动物材料" + DOABLES_REQUIREMENTS = "在有动物的普通区域，且avatar的境界必须大于等于动物的境界" + PARAMS = {} + + def get_available_animals(self) -> list[Animal]: + """ + 获取avatar境界足够的动物 + """ + region = self.avatar.tile.region + avatar_realm = self.avatar.cultivation_progress.realm + return [animal for animal in region.animals if avatar_realm >= animal.realm] + + duration_months = 6 + + def _execute(self) -> None: + """ + 执行狩猎动作 + """ + success_rate = self.get_success_rate() + available_animals = self.get_available_animals() + if len(available_animals) == 0: + return + + if random.random() < success_rate: + # 成功狩猎，从avatar境界足够的动物中随机选择一种 + target_animal = random.choice(available_animals) + # 随机选择该动物的一种物品 + item = random.choice(target_animal.items) + self.avatar.add_item(item, 1) + + def get_success_rate(self) -> float: + """ + 获取狩猎成功率，预留接口，目前固定为100% + """ + return 1.0 # 100%成功率 + + def can_start(self) -> bool: + region = self.avatar.tile.region + if not isinstance(region, NormalRegion): + return False + available_animals = self.get_available_animals() + if len(available_animals) == 0: + return False + return True + + def start(self) -> Event: + region = self.avatar.tile.region + return Event(self.world.month_stamp, f"{self.avatar.name} 在 {region.name} 开始狩猎") + + # TimedAction 已统一 step 逻辑 + + def finish(self) -> list[Event]: + return [] + + diff --git a/src/classes/action/move.py b/src/classes/action/move.py new file mode 100644 index 0000000..00f560e --- /dev/null +++ b/src/classes/action/move.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from src.classes.action import DefineAction, ChunkActionMixin + + +class Move(DefineAction, ChunkActionMixin): + """ + 最基础的移动动作，在tile之间进行切换。 + """ + + COMMENT = "移动到某个相对位置" + PARAMS = {"delta_x": "int", "delta_y": "int"} + + def _execute(self, delta_x: int, delta_y: int) -> None: + """ + 移动到某个tile + """ + world = self.world + # 基于境界的移动步长：每轴最多移动 move_step_length 格 + step = getattr(self.avatar, "move_step_length", 1) + clamped_dx = max(-step, min(step, delta_x)) + clamped_dy = max(-step, min(step, delta_y)) + + new_x = self.avatar.pos_x + clamped_dx + new_y = self.avatar.pos_y + clamped_dy + + # 边界检查：越界则不移动 + if world.map.is_in_bounds(new_x, new_y): + self.avatar.pos_x = new_x + self.avatar.pos_y = new_y + target_tile = world.map.get_tile(new_x, new_y) + self.avatar.tile = target_tile + else: + # 超出边界：不改变位置与tile + pass + + diff --git a/src/classes/action/move_away_from_avatar.py b/src/classes/action/move_away_from_avatar.py new file mode 100644 index 0000000..9d8ebdb --- /dev/null +++ b/src/classes/action/move_away_from_avatar.py @@ -0,0 +1,58 @@ +from __future__ import annotations + +from src.classes.action import TimedAction +from src.classes.event import Event + + +class MoveAwayFromAvatar(TimedAction): + """ + 持续远离指定角色，持续6个月。 + - 规则：每月尝试使与目标的曼哈顿距离增大一步 + - 任何时候都可以启动 + """ + + COMMENT = "持续远离指定角色" + DOABLES_REQUIREMENTS = "任何时候都可以执行" + PARAMS = {"avatar_name": "AvatarName"} + + def _find_avatar_by_name(self, name: str) -> "Avatar|None": + for v in self.world.avatar_manager.avatars.values(): + if v.name == name: + return v + return None + + duration_months = 6 + + def _execute(self, avatar_name: str) -> None: + target = self._find_avatar_by_name(avatar_name) + if target is None: + return + # 计算远离方向：使曼哈顿距离尽量增大 + dx = 1 if self.avatar.pos_x >= target.pos_x else -1 + dy = 1 if self.avatar.pos_y >= target.pos_y else -1 + nx = self.avatar.pos_x + dx + ny = self.avatar.pos_y + dy + if self.world.map.is_in_bounds(nx, ny): + self.avatar.pos_x = nx + self.avatar.pos_y = ny + self.avatar.tile = self.world.map.get_tile(nx, ny) + + def can_start(self, avatar_name: str | None = None) -> bool: + return True + + def start(self, avatar_name: str) -> Event: + target_name = avatar_name + try: + t = self._find_avatar_by_name(avatar_name) + if t is not None: + target_name = t.name + except Exception: + pass + return Event(self.world.month_stamp, f"{self.avatar.name} 开始远离 {target_name}") + + # TimedAction 已统一 step 逻辑 + + def finish(self, avatar_name: str) -> list[Event]: + return [] + + diff --git a/src/classes/action/move_away_from_region.py b/src/classes/action/move_away_from_region.py new file mode 100644 index 0000000..a51de25 --- /dev/null +++ b/src/classes/action/move_away_from_region.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from src.classes.action import InstantAction +from src.classes.event import Event + + +class MoveAwayFromRegion(InstantAction): + COMMENT = "离开指定区域" + DOABLES_REQUIREMENTS = "任何时候都可以执行" + PARAMS = {"region": "RegionName"} + + def _execute(self, region: str) -> None: + # 简化：向地图边缘移动一步 + dx = 1 if self.avatar.pos_x < self.world.map.width - 1 else -1 + dy = 1 if self.avatar.pos_y < self.world.map.height - 1 else -1 + nx = max(0, min(self.world.map.width - 1, self.avatar.pos_x + dx)) + ny = max(0, min(self.world.map.height - 1, self.avatar.pos_y + dy)) + if self.world.map.is_in_bounds(nx, ny): + self.avatar.pos_x = nx + self.avatar.pos_y = ny + self.avatar.tile = self.world.map.get_tile(nx, ny) + + def can_start(self, region: str | None = None) -> bool: + return True + + def start(self, region: str) -> Event: + return Event(self.world.month_stamp, f"{self.avatar.name} 开始离开 {region}") + + # InstantAction 已实现 step 完成 + + def finish(self, region: str) -> list[Event]: + return [] + + diff --git a/src/classes/action/move_to_avatar.py b/src/classes/action/move_to_avatar.py new file mode 100644 index 0000000..887206b --- /dev/null +++ b/src/classes/action/move_to_avatar.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +from src.classes.action import DefineAction, ActualActionMixin +from src.classes.event import Event +from src.classes.action import Move +from src.classes.action_runtime import ActionResult, ActionStatus + + +class MoveToAvatar(DefineAction, ActualActionMixin): + """ + 朝另一个角色当前位置移动。 + """ + + COMMENT = "移动到某个角色所在位置" + DOABLES_REQUIREMENTS = "任何时候都可以执行" + PARAMS = {"avatar_name": "str"} + + def _get_target(self, avatar_name: str): + """ + 根据名字查找目标角色；找不到返回 None。 + """ + for v in self.world.avatar_manager.avatars.values(): + if v.name == avatar_name: + return v + return None + + def _execute(self, avatar_name: str) -> None: + target = self._get_target(avatar_name) + if target is None: + return + cur_loc = (self.avatar.pos_x, self.avatar.pos_y) + target_loc = (target.pos_x, target.pos_y) + delta_x = target_loc[0] - cur_loc[0] + delta_y = target_loc[1] - cur_loc[1] + step = getattr(self.avatar, "move_step_length", 1) + delta_x = max(-step, min(step, delta_x)) + delta_y = max(-step, min(step, delta_y)) + Move(self.avatar, self.world).execute(delta_x, delta_y) + + def can_start(self, avatar_name: str | None = None) -> bool: + target = self._get_target(avatar_name) + if target is None: + return False + return True + + def start(self, avatar_name: str) -> Event: + target = self._get_target(avatar_name) + target_name = target.name if target is not None else avatar_name + return Event(self.world.month_stamp, f"{self.avatar.name} 开始移动向 {target_name}") + + def step(self, avatar_name: str) -> ActionResult: + self.execute(avatar_name=avatar_name) + target = self._get_target(avatar_name) + if target is None: + return ActionResult(status=ActionStatus.COMPLETED, events=[]) + done = self.avatar.tile == target.tile + return ActionResult(status=(ActionStatus.COMPLETED if done else ActionStatus.RUNNING), events=[]) + + def finish(self, avatar_name: str) -> list[Event]: + return [] + + diff --git a/src/classes/action/move_to_region.py b/src/classes/action/move_to_region.py new file mode 100644 index 0000000..7e7fea9 --- /dev/null +++ b/src/classes/action/move_to_region.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from src.classes.action import DefineAction, ActualActionMixin +from src.classes.event import Event +from src.classes.region import Region +from src.classes.action import Move +from src.classes.action_runtime import ActionResult, ActionStatus + + +class MoveToRegion(DefineAction, ActualActionMixin): + """ + 移动到某个region + """ + + COMMENT = "移动到某个区域" + DOABLES_REQUIREMENTS = "任何时候都可以执行" + PARAMS = {"region": "region_name"} + + def _execute(self, region: Region | str) -> None: + """ + 移动到某个region + """ + if isinstance(region, str): + from src.classes.region import regions_by_name + + region = regions_by_name[region] + cur_loc = (self.avatar.pos_x, self.avatar.pos_y) + region_center_loc = region.center_loc + delta_x = region_center_loc[0] - cur_loc[0] + delta_y = region_center_loc[1] - cur_loc[1] + # 横纵向一次最多移动 move_step_length 格（可以同时横纵移动） + step = getattr(self.avatar, "move_step_length", 1) + delta_x = max(-step, min(step, delta_x)) + delta_y = max(-step, min(step, delta_y)) + Move(self.avatar, self.world).execute(delta_x, delta_y) + + def can_start(self, region: Region | str | None = None) -> bool: + return True + + def start(self, region: Region | str) -> Event: + if isinstance(region, str): + region_name = region + from src.classes.region import regions_by_name + + if region in regions_by_name: + region_name = regions_by_name[region].name + elif hasattr(region, "name"): + region_name = region.name + else: + region_name = str(region) + return Event(self.world.month_stamp, f"{self.avatar.name} 开始移动向 {region_name}") + + def step(self, region: Region | str) -> ActionResult: + self.execute(region=region) + # 完成条件：到达目标区域 + if isinstance(region, str): + from src.classes.region import regions_by_name + + region = regions_by_name[region] + done = self.avatar.is_in_region(region) + return ActionResult(status=(ActionStatus.COMPLETED if done else ActionStatus.RUNNING), events=[]) + + def finish(self, region: Region | str) -> list[Event]: + return [] + + diff --git a/src/classes/action/play.py b/src/classes/action/play.py new file mode 100644 index 0000000..611e276 --- /dev/null +++ b/src/classes/action/play.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from src.classes.action import TimedAction +from src.classes.event import Event + + +class Play(TimedAction): + """ + 游戏娱乐动作，持续半年时间 + """ + + COMMENT = "游戏娱乐，放松身心" + DOABLES_REQUIREMENTS = "任何时候都可以执行" + PARAMS = {} + + duration_months = 6 + + def _execute(self) -> None: + """ + 进行游戏娱乐活动 + """ + # 游戏娱乐的具体逻辑可以在这里实现 + # 比如增加心情值、减少压力等 + pass + + def can_start(self) -> bool: + return True + + def start(self) -> Event: + return Event(self.world.month_stamp, f"{self.avatar.name} 开始玩耍") + + # TimedAction 已统一 step 逻辑 + + def finish(self) -> list[Event]: + return [] + + diff --git a/src/classes/action/plunder_mortals.py b/src/classes/action/plunder_mortals.py new file mode 100644 index 0000000..a8c8caa --- /dev/null +++ b/src/classes/action/plunder_mortals.py @@ -0,0 +1,43 @@ +from __future__ import annotations + +from src.classes.action import TimedAction +from src.classes.event import Event +from src.classes.region import CityRegion +from src.classes.alignment import Alignment + + +class PlunderMortals(TimedAction): + """ + 在城镇对凡人进行搜刮，获取少量灵石。 + 仅邪阵营可执行。 + """ + + COMMENT = "在城镇搜刮凡人，获取少量灵石" + DOABLES_REQUIREMENTS = "仅限城市区域，且角色阵营为‘邪’" + PARAMS = {} + GAIN = 20 + + duration_months = 3 + + def _execute(self) -> None: + region = self.avatar.tile.region + if not isinstance(region, CityRegion): + return + gain = self.GAIN + self.avatar.magic_stone = self.avatar.magic_stone + gain + + def can_start(self) -> bool: + region = self.avatar.tile.region + if not isinstance(region, CityRegion): + return False + return self.avatar.alignment == Alignment.EVIL + + def start(self) -> Event: + return Event(self.world.month_stamp, f"{self.avatar.name} 在城镇开始搜刮凡人") + + # TimedAction 已统一 step 逻辑 + + def finish(self) -> list[Event]: + return [] + + diff --git a/src/classes/action/registry.py b/src/classes/action/registry.py new file mode 100644 index 0000000..c868062 --- /dev/null +++ b/src/classes/action/registry.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from typing import Callable, Dict, Type, Iterable + + +class ActionRegistry: + """ + 动作注册表：维护动作名到类的映射，并标注哪些是“可实际执行”的动作。 + + - register(action_cls, actual): 注册一个动作类 + - get(name): 按名称获取动作类 + - all()/all_actual(): 获取全部/实际可执行的动作类集合 + """ + _name_to_cls: Dict[str, type] = {} + _actual_name_to_cls: Dict[str, type] = {} + + @classmethod + def register(cls, action_cls: type, *, actual: bool) -> None: + name = action_cls.__name__ + cls._name_to_cls[name] = action_cls + cls._name_to_cls[name.lower()] = action_cls # 大小写别名 + if actual: + cls._actual_name_to_cls[name] = action_cls + cls._actual_name_to_cls[name.lower()] = action_cls # 大小写别名 + + @classmethod + def get(cls, name: str) -> type: + return cls._name_to_cls[name] + + @classmethod + def all(cls) -> Iterable[type]: + return cls._name_to_cls.values() + + @classmethod + def all_actual(cls) -> Iterable[type]: + return cls._actual_name_to_cls.values() + + +def register_action(*, actual: bool = True) -> Callable[[type], type]: + def deco(t: type) -> type: + ActionRegistry.register(t, actual=actual) + return t + return deco + + diff --git a/src/classes/action/sold.py b/src/classes/action/sold.py new file mode 100644 index 0000000..372808b --- /dev/null +++ b/src/classes/action/sold.py @@ -0,0 +1,66 @@ +from __future__ import annotations + +from src.classes.action import InstantAction +from src.classes.event import Event +from src.classes.region import CityRegion +from src.classes.item import items_by_name +from src.classes.prices import prices + + +class SellItems(InstantAction): + """ + 在城镇出售指定名称的物品，一次性卖出持有的全部数量。 + 收益为 item_price * item_num，动作耗时1个月。 + """ + + COMMENT = "在城镇出售持有的某类物品的全部" + DOABLES_REQUIREMENTS = "在城镇且背包非空" + PARAMS = {"item_name": "str"} + + def _execute(self, item_name: str) -> None: + region = self.avatar.tile.region + if not isinstance(region, CityRegion): + return + + # 找到物品 + item = items_by_name.get(item_name) + if item is None: + return + + # 检查持有数量 + quantity = self.avatar.get_item_quantity(item) + if quantity <= 0: + return + + # 计算价格并结算 + price_per = prices.get_price(item) + total_gain = price_per * quantity + + # 扣除物品并增加灵石 + removed = self.avatar.remove_item(item, quantity) + if not removed: + return + + self.avatar.magic_stone = self.avatar.magic_stone + total_gain + + def can_start(self, item_name: str | None = None) -> bool: + region = self.avatar.tile.region + if not isinstance(region, CityRegion): + return False + if item_name is None: + # 用于动作空间：只要背包非空即可 + return bool(self.avatar.items) + item = items_by_name.get(item_name) + if item is None: + return False + return self.avatar.get_item_quantity(item) > 0 + + def start(self, item_name: str) -> Event: + return Event(self.world.month_stamp, f"{self.avatar.name} 在城镇出售 {item_name}") + + # InstantAction 已实现 step 完成 + + def finish(self, item_name: str) -> list[Event]: + return [] + + diff --git a/src/classes/action/talk.py b/src/classes/action/talk.py new file mode 100644 index 0000000..dafa34c --- /dev/null +++ b/src/classes/action/talk.py @@ -0,0 +1,69 @@ +from __future__ import annotations + +from src.classes.action import InstantAction +from src.classes.action_runtime import ActionResult, ActionStatus +from src.utils.config import CONFIG +from src.classes.event import Event + + +class Talk(InstantAction): + """ + 攀谈：尝试与同区域内的某个NPC进行交谈。 + - can_start：同区域内存在其他NPC + - 发起后：随机寻找“同一tile”的NPC，若不存在则本次无法攀谈 + - 若找到，则进入 MutualAction: Conversation（允许建立关系） + """ + + COMMENT = "与同区域内的NPC发起攀谈，若同一tile有人则进入交谈" + DOABLES_REQUIREMENTS = "同区域内存在其他NPC" + PARAMS = {} + + def _get_same_region_others(self) -> list["Avatar"]: + return self.world.avatar_manager.get_avatars_in_same_region(self.avatar) + + def _get_same_tile_others(self) -> list["Avatar"]: + same_tile: list["Avatar"] = [] + my_tile = self.avatar.tile + if my_tile is None: + return [] + for v in self.world.avatar_manager.avatars.values(): + if v is self.avatar or v.tile is None: + continue + if v.tile == my_tile: + same_tile.append(v) + return same_tile + + def _execute(self) -> None: + # Talk 本身不做长期效果，主要在 step 中驱动 Conversation + return + + def can_start(self) -> bool: + # 是否同区域存在其他NPC（用于展示在动作空间） + return len(self._get_same_region_others()) > 0 + + def start(self) -> Event: + self.same_region_others = self._get_same_region_others() + # 记录开始事件 + return Event(self.world.month_stamp, f"{self.avatar.name} 尝试与同区域的他人攀谈") + + def step(self) -> ActionResult: + import random + + target = random.choice(self.same_region_others) + + # 进入交谈：由概率决定本次是否允许建立关系 + from src.classes.mutual_action import Conversation + # 由配置决定本次是否有“有机会进入关系”标记 + prob = CONFIG.social.talk_into_relation_probability + can_into_relation = random.random() < prob + + conv = Conversation(self.avatar, self.world) + # 启动事件写入历史，不入侧边栏 + conv.start(target_avatar=target) + conv.step(target_avatar=target, can_into_relation=can_into_relation) + return ActionResult(status=ActionStatus.COMPLETED, events=[]) + + def finish(self) -> list[Event]: + return [] + + diff --git a/src/classes/action/targeting_mixin.py b/src/classes/action/targeting_mixin.py new file mode 100644 index 0000000..ec3af4c --- /dev/null +++ b/src/classes/action/targeting_mixin.py @@ -0,0 +1,48 @@ +from __future__ import annotations + +from typing import Optional, Iterable + +from src.classes.tile import get_avatar_distance + + +class TargetingMixin: + """ + 目标与距离等通用工具：为动作类提供统一的查找与抢占能力。 + + 注意：不做异常吞噬，失败路径返回 None 或 False，由调用方决策。 + """ + def find_avatar_by_name(self, name: str) -> "Avatar|None": + for v in self.world.avatar_manager.avatars.values(): + if v.name == name: + return v + return None + + def avatars_in_same_region(self, avatar: "Avatar") -> list["Avatar"]: + return self.world.avatar_manager.get_avatars_in_same_region(avatar) + + def avatars_on_same_tile(self, avatar: "Avatar") -> list["Avatar"]: + result: list["Avatar"] = [] + my_tile = avatar.tile + if my_tile is None: + return [] + for v in self.world.avatar_manager.avatars.values(): + if v is avatar or v.tile is None: + continue + if v.tile == my_tile: + result.append(v) + return result + + def distance_between(self, a: "Avatar", b: "Avatar") -> int: + return get_avatar_distance(a, b) + + def preempt_avatar(self, avatar: "Avatar") -> None: + """抢占目标：清空其计划并中断当前动作。""" + avatar.clear_plans() + avatar.current_action = None + + def set_immediate_action(self, avatar: "Avatar", action_name: str, params: dict) -> None: + """将动作立即加载并提交为当前动作，触发开始事件。""" + avatar.load_decide_result_chain([(action_name, params)], avatar.thinking, "") + avatar.commit_next_plan() + + diff --git a/src/classes/action_runtime.py b/src/classes/action_runtime.py index e69473b..5887af4 100644 --- a/src/classes/action_runtime.py +++ b/src/classes/action_runtime.py @@ -1,11 +1,43 @@ from __future__ import annotations from dataclasses import dataclass -from typing import Optional, Dict, Any +from typing import Optional, Dict, Any, List +from enum import Enum -# 运行时动作状态（字符串，便于与现有实现对接） -ActionStatus = str # "running" | "completed" | "failed" | "blocked" +class ActionStatus(Enum): + """ + 动作推进过程中的标准状态枚举。 + - RUNNING: 仍在进行中，需要在未来的 tick 中继续推进 + - COMPLETED: 已正常完成 + - FAILED: 执行失败（参数/前置条件等导致） + - CANCELLED: 被外部取消（如被其他动作抢占） + - INTERRUPTED: 运行中被打断（如战斗/事件中断） + """ + + RUNNING = "running" + COMPLETED = "completed" + FAILED = "failed" + CANCELLED = "cancelled" + INTERRUPTED = "interrupted" + + +@dataclass +class ActionResult: + """ + 标准动作返回体。所有 Action.step() 必须返回该类型。 + + Attributes: + status: 当前推进后的状态（见 ActionStatus） + events: 在本次推进过程中生成的事件（通常由 Avatar 收集并展示） + payload: 可选的结构化数据，便于上层消费（如数值、战斗结果等） + next_action: 可选的“建议下一个动作”（名称, 参数）供上层调度策略参考 + """ + + status: ActionStatus + events: List[Any] + payload: Optional[Dict[str, Any]] = None + next_action: Optional[tuple[str, Dict[str, Any]]] = None @dataclass @@ -29,5 +61,5 @@ class ActionInstance: """ action: Any # src.classes.action.Action params: Dict[str, Any] - status: ActionStatus = "running" + status: str = "running" # 遗留字段：Avatar 以字符串记录运行态 diff --git a/src/classes/actions.py b/src/classes/actions.py index b140ca3..1850612 100644 --- a/src/classes/actions.py +++ b/src/classes/actions.py @@ -2,75 +2,13 @@ from __future__ import annotations import json -from src.classes.action import ( - Move, - Cultivate, - Breakthrough, - MoveToRegion, - MoveToAvatar, - Play, - Hunt, - Harvest, - Sold, - Battle, - PlunderMortals, - HelpMortals, - Talk, -) -from src.classes.mutual_action import ( - DriveAway, - Attack, - Conversation, -) -from src.classes.action import MoveAwayFromAvatar, MoveAwayFromRegion, Escape +from src.classes.action.registry import ActionRegistry -ALL_ACTION_CLASSES = [ - Move, - Battle, - Cultivate, - Breakthrough, - MoveToRegion, - MoveToAvatar, - Play, - Hunt, - Harvest, - Sold, - PlunderMortals, - HelpMortals, - Talk, - # 互动相关动作（实际执行的反馈动作也纳入） - DriveAway, - Attack, - MoveAwayFromAvatar, - MoveAwayFromRegion, - Conversation, - Escape, - Conversation, - Escape, -] - -ALL_ACTUAL_ACTION_CLASSES = [ - Cultivate, - Breakthrough, - MoveToRegion, - MoveToAvatar, - Play, - Hunt, - Harvest, - Sold, - PlunderMortals, - HelpMortals, - Talk, - DriveAway, - Attack, - Conversation, - MoveAwayFromAvatar, - MoveAwayFromRegion, -] - -ALL_ACTION_NAMES = [action.__name__ for action in ALL_ACTION_CLASSES] -ALL_ACTUAL_ACTION_NAMES = [action.__name__ for action in ALL_ACTUAL_ACTION_CLASSES] +ALL_ACTION_CLASSES = list(ActionRegistry.all()) +ALL_ACTUAL_ACTION_CLASSES = list(ActionRegistry.all_actual()) +ALL_ACTION_NAMES = [cls.__name__ for cls in ALL_ACTION_CLASSES] +ALL_ACTUAL_ACTION_NAMES = [cls.__name__ for cls in ALL_ACTUAL_ACTION_CLASSES] ACTION_INFOS = { action.__name__: { diff --git a/src/classes/avatar.py b/src/classes/avatar.py index efe703f..60dfd7b 100644 --- a/src/classes/avatar.py +++ b/src/classes/avatar.py @@ -5,8 +5,9 @@ from typing import Optional, List import json from src.classes.calendar import MonthStamp -from src.classes.action import Action, StepStatus -from src.classes.actions import ALL_ACTUAL_ACTION_CLASSES, ALL_ACTION_CLASSES, ALL_ACTUAL_ACTION_NAMES +from src.classes.action import Action +from src.classes.action_runtime import ActionStatus, ActionResult +from src.classes.action.registry import ActionRegistry from src.classes.world import World from src.classes.tile import Tile from src.classes.region import Region @@ -123,12 +124,8 @@ class Avatar: Raises: ValueError: 如果找不到对应的动作类 """ - # 在所有动作类中查找对应的类 - for action_class in ALL_ACTION_CLASSES: - if action_class.__name__ == action_name: - return action_class(self, self.world) - - raise ValueError(f"未找到名为 '{action_name}' 的动作类") + action_cls = ActionRegistry.get(action_name) + return action_cls(self, self.world) def load_decide_result_chain(self, action_name_params_pairs: ACTION_NAME_PARAMS_PAIRS, avatar_thinking: str, objective: str): """ @@ -187,15 +184,9 @@ class Avatar: action_instance_before = self.current_action action = action_instance_before.action params = action_instance_before.params - try: - status, mid_events = action.step(**params) - except TypeError: - status, mid_events = action.step() - if status == StepStatus.COMPLETED: - try: - finish_events = action.finish(**params) - except TypeError: - finish_events = action.finish() + result: ActionResult = action.step(**params) + if result.status == ActionStatus.COMPLETED: + finish_events = action.finish(**params) # 仅当当前动作仍然是刚才执行的那个实例时才清空 # 若在 step() 内部通过“抢占”机制切换了动作（如 Escape 失败立即切到 Battle），不要清空新动作 if self.current_action is action_instance_before: @@ -205,8 +196,8 @@ class Avatar: for e in finish_events: self._pending_events.append(e) # 合并动作返回的事件（通常为空） - if mid_events: - for e in mid_events: + if result.events: + for e in result.events: self._pending_events.append(e) events, self._pending_events = self._pending_events, [] # 本轮已执行过，清除“新设动作”标记 @@ -356,6 +347,7 @@ class Avatar: """ 获取动作空间 """ + from src.classes.actions import ALL_ACTUAL_ACTION_NAMES actual_actions = [self.create_action(action_cls_name) for action_cls_name in ALL_ACTUAL_ACTION_NAMES] doable_actions: list[Action] = [] for action in actual_actions: diff --git a/src/classes/mutual_action.py b/src/classes/mutual_action.py deleted file mode 100644 index ace1299..0000000 --- a/src/classes/mutual_action.py +++ /dev/null @@ -1,328 +0,0 @@ -from __future__ import annotations - -from pathlib import Path -from typing import TYPE_CHECKING - -from src.classes.action import DefineAction, ActualActionMixin, LLMAction, StepStatus -from src.classes.battle import get_escape_success_rate -from src.classes.tile import get_avatar_distance -import random -from src.classes.event import Event -from src.utils.llm import get_prompt_and_call_llm -from src.utils.config import CONFIG -from src.classes.action import long_action -from src.classes.relation import relation_display_names, Relation, get_possible_post_relations - -if TYPE_CHECKING: - from src.classes.avatar import Avatar - -class MutualAction(DefineAction, LLMAction): - """ - 互动动作：A 对 B 发起动作，B 可以给出反馈（由 LLM 决策）。 - 子类需要定义： - - ACTION_NAME: 当前动作名（给模板展示） - - COMMENT: 动作语义说明（给模板展示） - - FEEDBACK_ACTIONS: 反馈可选的 action name 列表（直接可执行） - - PARAMS: 参数，需要包含 target_avatar - - FEEDBACK_ACTIONS: 反馈可选的 action name 列表（直接可执行） - """ - - ACTION_NAME: str = "MutualAction" - COMMENT: str = "" - DOABLES_REQUIREMENTS: str = "同区域内可互动" - PARAMS: dict = {"target_avatar": "Avatar"} - FEEDBACK_ACTIONS: list[str] = [] - - def _get_template_path(self) -> Path: - return CONFIG.paths.templates / "mutual_action.txt" - - def _build_prompt_infos(self, target_avatar: "Avatar") -> dict: - avatar_name_1 = self.avatar.name - avatar_name_2 = target_avatar.name - # avatar infos 仅放入与两人相关的提示，避免超长 - avatar_infos = { - avatar_name_1: self.avatar.cultivation_progress.get_simple_info(), # avatar1只放境界信息 - avatar_name_2: target_avatar.get_prompt_info([]), # avatar2放全量信息 - } - feedback_actions = self.FEEDBACK_ACTIONS - comment = self.COMMENT - action_name = self.ACTION_NAME - return { - "avatar_infos": avatar_infos, - "avatar_name_1": avatar_name_1, - "avatar_name_2": avatar_name_2, - "action_name": action_name, - "action_info": comment, - "feedback_actions": feedback_actions, - } - - def _call_llm_feedback(self, infos: dict) -> dict: - template_path = self._get_template_path() - res = get_prompt_and_call_llm(template_path, infos) - return res - - def _set_target_immediate_action(self, target_avatar: "Avatar", action_name: str, action_params: dict) -> None: - """ - 将反馈决定落地为目标角色的立即动作（清空后加载单步动作链）。 - """ - # 若当前已是同类同参动作，直接跳过，避免重复“发起战斗”等事件刷屏 - try: - cur = target_avatar.current_action - if cur is not None: - cur_name = getattr(cur.action, "__class__", type(cur.action)).__name__ - if cur_name == action_name: - # 简单判断参数等价（键值相等） - if getattr(cur, "params", {}) == dict(action_params): - return - except Exception: - pass - # 抢占：清空后续计划并中断其当前动作 - self._preempt_avatar(target_avatar) - # 先加载为计划 - target_avatar.load_decide_result_chain([(action_name, action_params)], target_avatar.thinking, "") - # 立即提交为当前动作，触发开始事件 - start_event = target_avatar.commit_next_plan() - if start_event is not None: - # 侧边栏仅推送一次（由动作发起方承担），另一侧仅写历史 - self._add_event_pair(start_event, initiator=self.avatar, target=target_avatar) - - def _settle_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: - """ - 子类实现：把反馈映射为具体动作 - """ - pass - - def _apply_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: - # 默认不额外记录，由事件系统承担 - return - - # 通用工具：按名找人 - def _find_avatar_by_name(self, name: str) -> "Avatar|None": - for v in self.world.avatar_manager.avatars.values(): - if v.name == name: - return v - return None - - def _get_target_avatar(self, target_avatar: "Avatar|str") -> "Avatar|None": - if isinstance(target_avatar, str): - return self._find_avatar_by_name(target_avatar) - return target_avatar - - # 通用工具：抢占 avatar 当前动作 - def _preempt_avatar(self, avatar: "Avatar") -> None: - avatar.clear_plans() - avatar.current_action = None - - # 通用工具：向两人推事件（侧栏只推 initiator） - def _add_event_pair(self, event: Event, initiator: "Avatar", target: "Avatar") -> None: - initiator.add_event(event) - if target is not None: - target.add_event(event, to_sidebar=False) - - def _execute(self, target_avatar: "Avatar|str") -> None: - # 允许传入名字字符串 - target_avatar = self._get_target_avatar(target_avatar) - if target_avatar is None: - return - - infos = self._build_prompt_infos(target_avatar) - res = self._call_llm_feedback(infos) - # LLM 只返回 {avatar_name_2: {thinking, feedback}} - r = res.get(infos["avatar_name_2"], {}) - thinking = r.get("thinking", "") - feedback = r.get("feedback", "") - - # 挂到目标的thinking上（面向UI/日志），并执行反馈落地 - target_avatar.thinking = thinking - # 1) 先清空目标后续计划（仅清空队列，不动当前动作） - target_avatar.clear_plans() - # 2) 再结算反馈映射为对应动作 - self._settle_feedback(target_avatar, feedback) - # 3) 反馈事件（进入侧边栏与双方历史，中文化文案） - fb_map = { - "MoveAwayFromAvatar": "试图远离", - "MoveAwayFromRegion": "试图离开区域", - "Escape": "逃离", - "Battle": "战斗", - } - fb_label = fb_map.get(str(feedback).strip(), str(feedback)) - feedback_event = Event(self.world.month_stamp, f"{target_avatar.name} 对 {self.avatar.name} 的反馈：{fb_label}") - # 侧边栏仅推送一次，另一侧仅写入历史，避免重复 - self._add_event_pair(feedback_event, initiator=self.avatar, target=target_avatar) - # 4) 记录历史（文本记录） - self._apply_feedback(target_avatar, feedback) - - # 实现 ActualActionMixin 接口 - def can_start(self, target_avatar: "Avatar|str|None" = None) -> bool: - """ - 检查互动动作能否启动：两个角色距离必须小于等于2 - """ - if target_avatar is None: - return False - target = self._get_target_avatar(target_avatar) - if target is None: - return False - distance = get_avatar_distance(self.avatar, target) - return distance <= 2 - - def start(self, target_avatar: "Avatar|str") -> Event: - """ - 启动互动动作，返回开始事件 - """ - target = self._get_target_avatar(target_avatar) - target_name = target.name if target is not None else str(target_avatar) - action_name = getattr(self, 'ACTION_NAME', self.name) - event = Event(self.world.month_stamp, f"{self.avatar.name} 对 {target_name} 发起 {action_name}") - # 仅写入历史，避免与提交阶段重复推送到侧边栏 - self.avatar.add_event(event, to_sidebar=False) - if target is not None: - target.add_event(event, to_sidebar=False) - return event - - def step(self, target_avatar: "Avatar|str") -> tuple[StepStatus, list[Event]]: - """ - 执行互动动作，互动动作是即时完成的 - """ - self.execute(target_avatar=target_avatar) - return StepStatus.COMPLETED, [] - - def finish(self, target_avatar: "Avatar|str") -> list[Event]: - """ - 完成互动动作，事件已在 step 中处理，无需额外事件 - """ - return [] - - -class DriveAway(MutualAction, ActualActionMixin): - """驱赶：试图让对方离开当前区域。""" - ACTION_NAME = "驱赶" - COMMENT = "以武力威慑对方离开此地。" - DOABLES_REQUIREMENTS = "与目标处于同一区域" - PARAMS = {"target_avatar": "AvatarName"} - FEEDBACK_ACTIONS = ["MoveAwayFromRegion", "Battle"] - def _settle_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: - fb = str(feedback_name).strip() - if fb == "MoveAwayFromRegion": - # 驱赶选择离开：必定成功，不涉及概率 - params = {"region": self.avatar.tile.region.name} - self._set_target_immediate_action(target_avatar, fb, params) - elif fb == "Battle": - params = {"avatar_name": self.avatar.name} - self._set_target_immediate_action(target_avatar, fb, params) - -class Attack(MutualAction, ActualActionMixin): - """攻击另一个NPC""" - ACTION_NAME = "攻击" - COMMENT = "对目标进行攻击。" - DOABLES_REQUIREMENTS = "与目标处于同一区域" - PARAMS = {"target_avatar": "AvatarName"} - FEEDBACK_ACTIONS = ["Escape", "Battle"] - - def _settle_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: - fb = str(feedback_name).strip() - if fb == "Escape": - params = {"avatar_name": self.avatar.name} - self._set_target_immediate_action(target_avatar, fb, params) - elif fb == "Battle": - params = {"avatar_name": self.avatar.name} - self._set_target_immediate_action(target_avatar, fb, params) - - - - - -class Conversation(MutualAction, ActualActionMixin): - """交谈：两名角色在同一区域进行交流。 - - - 可由“攀谈”触发，或直接发起 - - 仅当双方处于同一 Region 时可启动 - - 当 can_into_relation=True 且 LLM 决策返回 into_relation 时，根据返回建立关系 - - 会将对话内容写入事件系统 - """ - - ACTION_NAME = "交谈" - COMMENT = "两人需在同一地区，进行一段交流对话" - DOABLES_REQUIREMENTS = "与目标处于同一区域" - PARAMS = {"target_avatar": "AvatarName"} - FEEDBACK_ACTIONS: list[str] = ["Talk", "Reject"] - - def _get_template_path(self) -> Path: - # 使用 talk.txt 模板，以获取是否接受与对话内容 - return CONFIG.paths.templates / "talk.txt" - - def _build_prompt_infos(self, target_avatar: "Avatar", *, can_into_relation: bool) -> dict: - avatar_name_1 = self.avatar.name - avatar_name_2 = target_avatar.name - # 目标的 get_prompt_info 已含 personas、关系等，信息更充分 - avatar_infos = { - avatar_name_1: self.avatar.get_prompt_info([]), - avatar_name_2: target_avatar.get_prompt_info([]), - } - # 可能的后天关系（转中文名，给模板阅读） - possible_relations = [relation_display_names[r] for r in get_possible_post_relations(self.avatar, target_avatar)] - return { - "avatar_infos": avatar_infos, - "avatar_name_1": avatar_name_1, - "avatar_name_2": avatar_name_2, - "can_into_relation": bool(can_into_relation), - "possible_relations": possible_relations, - } - - # 关系解析由 Relation 提供类方法，仅接受中文关系名，无法解析则跳过 - - def can_start(self, target_avatar: "Avatar|str|None" = None, **kwargs) -> bool: - if target_avatar is None: - return False - target = self._get_target_avatar(target_avatar) - if target is None or target.tile is None or self.avatar.tile is None: - return False - return target.tile.region == self.avatar.tile.region - - def start(self, target_avatar: "Avatar|str", **kwargs) -> Event: - target = self._get_target_avatar(target_avatar) - target_name = target.name if target is not None else str(target_avatar) - event = Event(self.world.month_stamp, f"{self.avatar.name} 与 {target_name} 开始交谈") - # 写入历史即可，内容事件稍后生成 - self.avatar.add_event(event, to_sidebar=False) - if target is not None: - target.add_event(event, to_sidebar=False) - return event - - def step(self, target_avatar: "Avatar|str", can_into_relation: bool = False) -> tuple[StepStatus, list[Event]]: - target = self._get_target_avatar(target_avatar) - if target is None: - return StepStatus.COMPLETED, [] - - infos = self._build_prompt_infos(target, can_into_relation=can_into_relation) - res = self._call_llm_feedback(infos) - r = res.get(infos["avatar_name_2"], {}) - thinking = r.get("thinking", "") - feedback = str(r.get("feedback", "")).strip() - talk_content = str(r.get("talk_content", "")).strip() - into_relation_str = str(r.get("into_relation", "")).strip() - - target.thinking = thinking - - fb = feedback.strip() - # 仅当明确接受时才记录对话与关系；其余一律视为拒绝 - if fb == "Talk": - if talk_content: - content_event = Event(self.world.month_stamp, f"{self.avatar.name} 与 {target.name} 的交谈：{talk_content}") - # 进入侧栏一次，并写入双方历史 - self._add_event_pair(content_event, initiator=self.avatar, target=target) - - if can_into_relation and into_relation_str: - rel = Relation.from_chinese(into_relation_str) - if rel is not None: - self.avatar.set_relation(target, rel) - set_event = Event(self.world.month_stamp, f"{self.avatar.name} 与 {target.name} 的关系变为：{relation_display_names.get(rel, str(rel))}") - self._add_event_pair(set_event, initiator=self.avatar, target=target) - - return StepStatus.COMPLETED, [] - else: - feedback_event = Event(self.world.month_stamp, f"{target.name} 拒绝与 {self.avatar.name} 交谈") - self._add_event_pair(feedback_event, initiator=self.avatar, target=target) - return StepStatus.COMPLETED, [] - - def finish(self, target_avatar: "Avatar|str", **kwargs) -> list[Event]: - return [] \ No newline at end of file diff --git a/src/classes/mutual_action/__init__.py b/src/classes/mutual_action/__init__.py new file mode 100644 index 0000000..404e0bc --- /dev/null +++ b/src/classes/mutual_action/__init__.py @@ -0,0 +1,21 @@ +from __future__ import annotations + +from .mutual_action import MutualAction +from .drive_away import DriveAway +from .attack import Attack +from .conversation import Conversation +from src.classes.action.registry import register_action + +__all__ = [ + "MutualAction", + "DriveAway", + "Attack", + "Conversation", +] + +# 注册 mutual actions（均为实际动作） +register_action(actual=True)(DriveAway) +register_action(actual=True)(Attack) +register_action(actual=True)(Conversation) + + diff --git a/src/classes/mutual_action/attack.py b/src/classes/mutual_action/attack.py new file mode 100644 index 0000000..0d13449 --- /dev/null +++ b/src/classes/mutual_action/attack.py @@ -0,0 +1,24 @@ +from __future__ import annotations + +from .mutual_action import MutualAction + + +class Attack(MutualAction): + """攻击另一个NPC""" + + ACTION_NAME = "攻击" + COMMENT = "对目标进行攻击。" + DOABLES_REQUIREMENTS = "与目标处于同一区域" + PARAMS = {"target_avatar": "AvatarName"} + FEEDBACK_ACTIONS = ["Escape", "Battle"] + + def _settle_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: + fb = str(feedback_name).strip() + if fb == "Escape": + params = {"avatar_name": self.avatar.name} + self._set_target_immediate_action(target_avatar, fb, params) + elif fb == "Battle": + params = {"avatar_name": self.avatar.name} + self._set_target_immediate_action(target_avatar, fb, params) + + diff --git a/src/classes/mutual_action/conversation.py b/src/classes/mutual_action/conversation.py new file mode 100644 index 0000000..d23d913 --- /dev/null +++ b/src/classes/mutual_action/conversation.py @@ -0,0 +1,111 @@ +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +from .mutual_action import MutualAction +from src.classes.relation import relation_display_names, Relation, get_possible_post_relations +from src.classes.event import Event +from src.utils.config import CONFIG +from src.classes.action_runtime import ActionResult, ActionStatus +from src.classes.action.event_helper import EventHelper + +if TYPE_CHECKING: + from src.classes.avatar import Avatar + + +class Conversation(MutualAction): + """交谈：两名角色在同一区域进行交流。 + + - 可由“攀谈”触发，或直接发起 + - 仅当双方处于同一 Region 时可启动 + - 当 can_into_relation=True 且 LLM 决策返回 into_relation 时，根据返回建立关系 + - 会将对话内容写入事件系统 + """ + + ACTION_NAME = "交谈" + COMMENT = "两人需在同一地区，进行一段交流对话" + DOABLES_REQUIREMENTS = "与目标处于同一区域" + PARAMS = {"target_avatar": "AvatarName"} + FEEDBACK_ACTIONS: list[str] = ["Talk", "Reject"] + + def _get_template_path(self) -> Path: + # 使用 talk.txt 模板，以获取是否接受与对话内容 + return CONFIG.paths.templates / "talk.txt" + + def _build_prompt_infos(self, target_avatar: "Avatar", *, can_into_relation: bool) -> dict: + avatar_name_1 = self.avatar.name + avatar_name_2 = target_avatar.name + # 目标的 get_prompt_info 已含 personas、关系等，信息更充分 + avatar_infos = { + avatar_name_1: self.avatar.get_prompt_info([]), + avatar_name_2: target_avatar.get_prompt_info([]), + } + # 可能的后天关系（转中文名，给模板阅读） + possible_relations = [relation_display_names[r] for r in get_possible_post_relations(self.avatar, target_avatar)] + return { + "avatar_infos": avatar_infos, + "avatar_name_1": avatar_name_1, + "avatar_name_2": avatar_name_2, + "can_into_relation": bool(can_into_relation), + "possible_relations": possible_relations, + } + + def can_start(self, target_avatar: "Avatar|str|None" = None, **kwargs) -> bool: + if target_avatar is None: + return False + target = self._get_target_avatar(target_avatar) + if target is None or target.tile is None or self.avatar.tile is None: + return False + return target.tile.region == self.avatar.tile.region + + def start(self, target_avatar: "Avatar|str", **kwargs) -> Event: + target = self._get_target_avatar(target_avatar) + target_name = target.name if target is not None else str(target_avatar) + event = Event(self.world.month_stamp, f"{self.avatar.name} 与 {target_name} 开始交谈") + # 写入历史即可，内容事件稍后生成 + self.avatar.add_event(event, to_sidebar=False) + if target is not None: + target.add_event(event, to_sidebar=False) + return event + + def step(self, target_avatar: "Avatar|str", can_into_relation: bool = False) -> ActionResult: + target = self._get_target_avatar(target_avatar) + if target is None: + return ActionResult(status=ActionStatus.COMPLETED, events=[]) + + infos = self._build_prompt_infos(target, can_into_relation=can_into_relation) + res = self._call_llm_feedback(infos) + r = res.get(infos["avatar_name_2"], {}) + thinking = r.get("thinking", "") + feedback = str(r.get("feedback", "")).strip() + talk_content = str(r.get("talk_content", "")).strip() + into_relation_str = str(r.get("into_relation", "")).strip() + + target.thinking = thinking + + fb = feedback.strip() + # 仅当明确接受时才记录对话与关系；其余一律视为拒绝 + if fb == "Talk": + if talk_content: + content_event = Event(self.world.month_stamp, f"{self.avatar.name} 与 {target.name} 的交谈：{talk_content}") + # 进入侧栏一次，并写入双方历史 + EventHelper.push_pair(content_event, initiator=self.avatar, target=target, to_sidebar_once=True) + + if can_into_relation and into_relation_str: + rel = Relation.from_chinese(into_relation_str) + if rel is not None: + self.avatar.set_relation(target, rel) + set_event = Event(self.world.month_stamp, f"{self.avatar.name} 与 {target.name} 的关系变为：{relation_display_names.get(rel, str(rel))}") + EventHelper.push_pair(set_event, initiator=self.avatar, target=target, to_sidebar_once=True) + + return ActionResult(status=ActionStatus.COMPLETED, events=[]) + else: + feedback_event = Event(self.world.month_stamp, f"{target.name} 拒绝与 {self.avatar.name} 交谈") + EventHelper.push_pair(feedback_event, initiator=self.avatar, target=target, to_sidebar_once=True) + return ActionResult(status=ActionStatus.COMPLETED, events=[]) + + def finish(self, target_avatar: "Avatar|str", **kwargs) -> list[Event]: + return [] + + diff --git a/src/classes/mutual_action/drive_away.py b/src/classes/mutual_action/drive_away.py new file mode 100644 index 0000000..ca5e06c --- /dev/null +++ b/src/classes/mutual_action/drive_away.py @@ -0,0 +1,25 @@ +from __future__ import annotations + +from .mutual_action import MutualAction + + +class DriveAway(MutualAction): + """驱赶：试图让对方离开当前区域。""" + + ACTION_NAME = "驱赶" + COMMENT = "以武力威慑对方离开此地。" + DOABLES_REQUIREMENTS = "与目标处于同一区域" + PARAMS = {"target_avatar": "AvatarName"} + FEEDBACK_ACTIONS = ["MoveAwayFromRegion", "Battle"] + + def _settle_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: + fb = str(feedback_name).strip() + if fb == "MoveAwayFromRegion": + # 驱赶选择离开：必定成功，不涉及概率 + params = {"region": self.avatar.tile.region.name} + self._set_target_immediate_action(target_avatar, fb, params) + elif fb == "Battle": + params = {"avatar_name": self.avatar.name} + self._set_target_immediate_action(target_avatar, fb, params) + + diff --git a/src/classes/mutual_action/mutual_action.py b/src/classes/mutual_action/mutual_action.py new file mode 100644 index 0000000..877245a --- /dev/null +++ b/src/classes/mutual_action/mutual_action.py @@ -0,0 +1,176 @@ +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +from src.classes.action import DefineAction, ActualActionMixin, LLMAction +from src.classes.tile import get_avatar_distance +from src.classes.event import Event +from src.utils.llm import get_prompt_and_call_llm +from src.utils.config import CONFIG +from src.classes.relation import relation_display_names, Relation, get_possible_post_relations +from src.classes.action_runtime import ActionResult, ActionStatus +from src.classes.action.event_helper import EventHelper +from src.classes.action.targeting_mixin import TargetingMixin + +if TYPE_CHECKING: + from src.classes.avatar import Avatar + + +class MutualAction(DefineAction, LLMAction, TargetingMixin): + """ + 互动动作：A 对 B 发起动作，B 可以给出反馈（由 LLM 决策）。 + 子类需要定义： + - ACTION_NAME: 当前动作名（给模板展示） + - COMMENT: 动作语义说明（给模板展示） + - FEEDBACK_ACTIONS: 反馈可选的 action name 列表（直接可执行） + - PARAMS: 参数，需要包含 target_avatar + - FEEDBACK_ACTIONS: 反馈可选的 action name 列表（直接可执行） + """ + + ACTION_NAME: str = "MutualAction" + COMMENT: str = "" + DOABLES_REQUIREMENTS: str = "同区域内可互动" + PARAMS: dict = {"target_avatar": "Avatar"} + FEEDBACK_ACTIONS: list[str] = [] + + def _get_template_path(self) -> Path: + return CONFIG.paths.templates / "mutual_action.txt" + + def _build_prompt_infos(self, target_avatar: "Avatar") -> dict: + avatar_name_1 = self.avatar.name + avatar_name_2 = target_avatar.name + # avatar infos 仅放入与两人相关的提示，避免超长 + avatar_infos = { + avatar_name_1: self.avatar.cultivation_progress.get_simple_info(), + avatar_name_2: target_avatar.get_prompt_info([]), + } + feedback_actions = self.FEEDBACK_ACTIONS + comment = self.COMMENT + action_name = self.ACTION_NAME + return { + "avatar_infos": avatar_infos, + "avatar_name_1": avatar_name_1, + "avatar_name_2": avatar_name_2, + "action_name": action_name, + "action_info": comment, + "feedback_actions": feedback_actions, + } + + def _call_llm_feedback(self, infos: dict) -> dict: + template_path = self._get_template_path() + res = get_prompt_and_call_llm(template_path, infos) + return res + + def _set_target_immediate_action(self, target_avatar: "Avatar", action_name: str, action_params: dict) -> None: + """ + 将反馈决定落地为目标角色的立即动作（清空后加载单步动作链）。 + """ + # 若当前已是同类同参动作，直接跳过，避免重复“发起战斗”等事件刷屏 + try: + cur = target_avatar.current_action + if cur is not None: + cur_name = getattr(cur.action, "__class__", type(cur.action)).__name__ + if cur_name == action_name: + if getattr(cur, "params", {}) == dict(action_params): + return + except Exception: + pass + # 抢占：清空后续计划并中断其当前动作 + self._preempt_avatar(target_avatar) + # 先加载为计划 + target_avatar.load_decide_result_chain([(action_name, action_params)], target_avatar.thinking, "") + # 立即提交为当前动作，触发开始事件 + start_event = target_avatar.commit_next_plan() + if start_event is not None: + # 侧边栏仅推送一次（由动作发起方承担），另一侧仅写历史 + EventHelper.push_pair(start_event, initiator=self.avatar, target=target_avatar, to_sidebar_once=True) + + def _settle_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: + """ + 子类实现：把反馈映射为具体动作 + """ + pass + + def _apply_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: + # 默认不额外记录，由事件系统承担 + return + + def _get_target_avatar(self, target_avatar: "Avatar|str") -> "Avatar|None": + if isinstance(target_avatar, str): + return self.find_avatar_by_name(target_avatar) + return target_avatar + + def _execute(self, target_avatar: "Avatar|str") -> None: + target_avatar = self._get_target_avatar(target_avatar) + if target_avatar is None: + return + + infos = self._build_prompt_infos(target_avatar) + res = self._call_llm_feedback(infos) + # LLM 只返回 {avatar_name_2: {thinking, feedback}} + r = res.get(infos["avatar_name_2"], {}) + thinking = r.get("thinking", "") + feedback = r.get("feedback", "") + + # 挂到目标的thinking上（面向UI/日志），并执行反馈落地 + target_avatar.thinking = thinking + # 1) 先清空目标后续计划（仅清空队列，不动当前动作） + target_avatar.clear_plans() + # 2) 再结算反馈映射为对应动作 + self._settle_feedback(target_avatar, feedback) + # 3) 反馈事件（进入侧边栏与双方历史，中文化文案） + fb_map = { + "MoveAwayFromAvatar": "试图远离", + "MoveAwayFromRegion": "试图离开区域", + "Escape": "逃离", + "Battle": "战斗", + } + fb_label = fb_map.get(str(feedback).strip(), str(feedback)) + feedback_event = Event(self.world.month_stamp, f"{target_avatar.name} 对 {self.avatar.name} 的反馈：{fb_label}") + # 侧边栏仅推送一次，另一侧仅写入历史，避免重复 + EventHelper.push_pair(feedback_event, initiator=self.avatar, target=target_avatar, to_sidebar_once=True) + # 4) 记录历史（文本记录） + self._apply_feedback(target_avatar, feedback) + + # 实现 ActualActionMixin 接口 + def can_start(self, target_avatar: "Avatar|str|None" = None) -> bool: + """ + 检查互动动作能否启动：两个角色距离必须小于等于2 + """ + if target_avatar is None: + return False + target = self._get_target_avatar(target_avatar) + if target is None: + return False + distance = get_avatar_distance(self.avatar, target) + return distance <= 2 + + def start(self, target_avatar: "Avatar|str") -> Event: + """ + 启动互动动作，返回开始事件 + """ + target = self._get_target_avatar(target_avatar) + target_name = target.name if target is not None else str(target_avatar) + action_name = getattr(self, 'ACTION_NAME', self.name) + event = Event(self.world.month_stamp, f"{self.avatar.name} 对 {target_name} 发起 {action_name}") + # 仅写入历史，避免与提交阶段重复推送到侧边栏 + self.avatar.add_event(event, to_sidebar=False) + if target is not None: + target.add_event(event, to_sidebar=False) + return event + + def step(self, target_avatar: "Avatar|str") -> ActionResult: + """ + 执行互动动作，互动动作是即时完成的 + """ + self.execute(target_avatar=target_avatar) + return ActionResult(status=ActionStatus.COMPLETED, events=[]) + + def finish(self, target_avatar: "Avatar|str") -> list[Event]: + """ + 完成互动动作，事件已在 step 中处理，无需额外事件 + """ + return [] + +