diff --git a/src/classes/actions.py b/src/classes/actions.py new file mode 100644 index 0000000..0373f96 --- /dev/null +++ b/src/classes/actions.py @@ -0,0 +1,75 @@ +from __future__ import annotations + +import json + +from src.classes.action import ( + DefineAction, + ActualActionMixin, + Move, + Cultivate, + Breakthrough, + MoveToRegion, + MoveToAvatar, + Play, + Hunt, + Harvest, + Sold, +) +from src.classes.mutual_action import ( + MutualAction, + DriveAway, + AttackInteract, + MoveAwayFromAvatar, + MoveAwayFromRegion, + Battle, +) + + +ALL_ACTION_CLASSES = [ + Move, + Cultivate, + Breakthrough, + MoveToRegion, + MoveToAvatar, + Play, + Hunt, + Harvest, + Sold, + # 互动相关动作（实际执行的反馈动作也纳入） + DriveAway, + AttackInteract, + MoveAwayFromAvatar, + MoveAwayFromRegion, + Battle, +] + +ALL_ACTUAL_ACTION_CLASSES = [ + Cultivate, + Breakthrough, + MoveToRegion, + MoveToAvatar, + Play, + Hunt, + Harvest, + Sold, + DriveAway, + AttackInteract, + MoveAwayFromAvatar, + MoveAwayFromRegion, + Battle, +] + +ALL_ACTION_NAMES = [action.__name__ for action in ALL_ACTION_CLASSES] +ALL_ACTUAL_ACTION_NAMES = [action.__name__ for action in ALL_ACTUAL_ACTION_CLASSES] + +ACTION_INFOS = { + action.__name__: { + "comment": getattr(action, "COMMENT", ""), + "doable_requirements": getattr(action, "DOABLES_REQUIREMENTS", ""), + "params": getattr(action, "PARAMS", {}), + } + for action in ALL_ACTUAL_ACTION_CLASSES +} +ACTION_INFOS_STR = json.dumps(ACTION_INFOS, ensure_ascii=False) + + diff --git a/src/classes/ai.py b/src/classes/ai.py index cccd2f8..0306aed 100644 --- a/src/classes/ai.py +++ b/src/classes/ai.py @@ -15,7 +15,7 @@ from src.classes.event import Event, NULL_EVENT from src.utils.llm import get_ai_prompt_and_call_llm_async from src.classes.typings import ACTION_NAME, ACTION_PARAMS, ACTION_PAIR, ACTION_NAME_PARAMS_PAIRS from src.utils.config import CONFIG -from src.classes.action import ACTION_INFOS_STR +from src.classes.actions import ACTION_INFOS_STR if TYPE_CHECKING: from src.classes.avatar import Avatar diff --git a/src/classes/avatar.py b/src/classes/avatar.py index dfaef93..9230b37 100644 --- a/src/classes/avatar.py +++ b/src/classes/avatar.py @@ -5,7 +5,8 @@ from typing import Optional, List import json from src.classes.calendar import MonthStamp -from src.classes.action import Action, ALL_ACTUAL_ACTION_CLASSES, ALL_ACTION_CLASSES, ALL_ACTUAL_ACTION_NAMES +from src.classes.action import Action +from src.classes.actions import ALL_ACTUAL_ACTION_CLASSES, ALL_ACTION_CLASSES, ALL_ACTUAL_ACTION_NAMES from src.classes.world import World from src.classes.tile import Tile from src.classes.region import Region @@ -37,8 +38,8 @@ gender_strs = { Gender.FEMALE: "女", } -# 历史动作对的最大数量 -MAX_HISTORY_ACTIONS = 3 +# 历史事件的最大数量 +MAX_HISTORY_EVENTS = 10 @dataclass class Avatar: @@ -60,7 +61,8 @@ class Avatar: root: Root = field(default_factory=lambda: random.choice(list(Root))) personas: List[Persona] = field(default_factory=list) cur_action_pair: Optional[ACTION_PAIR] = None - history_action_pairs: list[ACTION_PAIR] = field(default_factory=list) + history_events: List[Event] = field(default_factory=list) + _pending_events: List[Event] = field(default_factory=list) next_actions: ACTION_NAME_PARAMS_PAIRS = field(default_factory=list) thinking: str = "" objective: str = "" @@ -135,6 +137,12 @@ class Avatar: if len(action_name_params_pairs) > 1: self.next_actions.extend(action_name_params_pairs[1:]) + def clear_next_actions(self) -> None: + """ + 清空后续动作队列（不影响当前动作）。 + """ + self.next_actions.clear() + def has_next_actions(self) -> bool: return len(self.next_actions) > 0 @@ -176,41 +184,23 @@ class Avatar: # 动作的 is_doable 定义为 @property return bool(getattr(action, "is_doable", True)) - async def act(self): + async def act(self) -> List[Event]: """ 角色执行动作。 注意这里只负责执行，不负责决定做什么动作。 事件只在决定动作时产生，执行过程不产生事件 """ - # 纯粹执行动作，不产生事件 + # 纯粹执行动作。具体事件由决定阶段或动作内部通过 add_event 添加 action, action_params = self.cur_action_pair action.execute(**action_params) if action.is_finished(**action_params): - # 将完成的动作对添加到历史记录中 - self._add_to_history(self.cur_action_pair) - - return - - def _add_to_history(self, action_pair: ACTION_PAIR) -> None: - """ - 将完成的动作对添加到历史记录中 - - Args: - action_pair: 要添加的动作对 - - 注意: - - 如果历史记录达到上限，会丢弃最老的记录 - - 新的记录会被添加到列表末尾 - """ - # 添加新的动作对到历史记录 - self.history_action_pairs.append(action_pair) - self.cur_action_pair = None - - # 如果超过上限，移除最老的记录 - if len(self.history_action_pairs) > MAX_HISTORY_ACTIONS: - self.history_action_pairs.pop(0) + # 完成后清空当前动作 + self.cur_action_pair = None + # 返回并清空待派发事件 + events, self._pending_events = self._pending_events, [] + return events def update_cultivation(self, new_level: int): """ @@ -329,11 +319,18 @@ class Avatar: """ return self.items.get(item, 0) - def get_history_action_pairs_str(self) -> str: + def add_event(self, event: Event, *, to_sidebar: bool = True, to_history: bool = True) -> None: """ - 获取历史动作对的字符串 + 添加事件： + - to_sidebar: 是否进入全局侧边栏（通过 Simulator 收集） + - to_history: 是否进入本角色的历史事件（最多保留 MAX_HISTORY_EVENTS 条） """ - return "\n".join([f"{action.name}: {action_params}" for action, action_params in self.history_action_pairs]) + if to_sidebar: + self._pending_events.append(event) + if to_history: + self.history_events.append(event) + if len(self.history_events) > MAX_HISTORY_EVENTS: + self.history_events = self.history_events[-MAX_HISTORY_EVENTS:] def get_action_space_str(self) -> str: action_space = self.get_action_space() @@ -382,7 +379,14 @@ class Avatar: # 关系摘要 relations_summary = self._get_relations_summary_str() - return f"{info}\n{personas_info}\n{magic_stone_info}\n{items_info}\n关系：{relations_summary}\n{co_region_info}\n该角色的目前暂时的合法动作为：{action_space}" + # 历史事件摘要 + if self.history_events: + history_lines = "；".join([str(e) for e in self.history_events[-8:]]) + history_info = f"历史事件：{history_lines}" + else: + history_info = "历史事件：无" + + return f"{info}\n{personas_info}\n{magic_stone_info}\n{items_info}\n{history_info}\n关系：{relations_summary}\n{co_region_info}\n该角色的目前合法动作为：{action_space}" def set_relation(self, other: "Avatar", relation: Relation) -> None: """ diff --git a/src/classes/battle.py b/src/classes/battle.py new file mode 100644 index 0000000..be702ee --- /dev/null +++ b/src/classes/battle.py @@ -0,0 +1,47 @@ +from __future__ import annotations + +import random +from typing import Tuple, TYPE_CHECKING + +from src.classes.cultivation import Realm + +if TYPE_CHECKING: + from src.classes.avatar import Avatar + + +def _realm_order(realm: Realm) -> int: + """ + 将境界映射为数值顺序，用于胜率计算。 + """ + order_map = { + Realm.Qi_Refinement: 1, + Realm.Foundation_Establishment: 2, + Realm.Core_Formation: 3, + Realm.Nascent_Soul: 4, + } + return order_map.get(realm, 1) + + +def calc_win_rate(attacker: "Avatar", defender: "Avatar") -> float: + """ + 根据双方境界粗略计算进攻方胜率。 + 基准50%，每高一个大境界+15%，限制在[0.1, 0.9]。 + """ + atk_order = _realm_order(attacker.cultivation_progress.realm) + def_order = _realm_order(defender.cultivation_progress.realm) + delta = atk_order - def_order + base = 0.5 + 0.15 * delta + return max(0.1, min(0.9, base)) + + +def decide_battle(attacker: "Avatar", defender: "Avatar") -> Tuple["Avatar", "Avatar", float]: + """ + 结算一场战斗，返回(胜者, 败者, 进攻方胜率)。 + 仅做结果判定，不做数值伤害结算。 + """ + p = calc_win_rate(attacker, defender) + if random.random() < p: + return attacker, defender, p + else: + return defender, attacker, p + diff --git a/src/classes/mutual_action.py b/src/classes/mutual_action.py new file mode 100644 index 0000000..e84b25c --- /dev/null +++ b/src/classes/mutual_action.py @@ -0,0 +1,235 @@ +from __future__ import annotations + +from pathlib import Path +from typing import TYPE_CHECKING + +from src.classes.action import DefineAction, ActualActionMixin, LLMAction +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.battle import decide_battle + +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 = getattr(self, "FEEDBACK_ACTIONS", []) + return { + "avatar_infos": avatar_infos, + "avatar_name_1": avatar_name_1, + "avatar_name_2": avatar_name_2, + "action_name": getattr(self, "ACTION_NAME", self.name), + "action_info": getattr(self, "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: + """ + 将反馈决定落地为目标角色的立即动作（清空后加载单步动作链）。 + """ + # 使用已有的加载动作链接口，立即设置为当前动作 + target_avatar.load_decide_result_chain([(action_name, action_params)], target_avatar.thinking, "") + + 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 _execute(self, target_avatar: "Avatar|str") -> None: + # 允许传入名字字符串 + if isinstance(target_avatar, str): + name = target_avatar + target_avatar = None + for v in self.world.avatar_manager.avatars.values(): + if v.name == name: + target_avatar = v + break + 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 + # 发起事件（进入侧边栏与双方历史） + start_event = self.get_event(target_avatar) + self.avatar.add_event(start_event) + target_avatar.add_event(start_event) + # 1) 先清空目标后续动作（仅清空队列，不动当前动作） + if hasattr(target_avatar, "clear_next_actions"): + target_avatar.clear_next_actions() + # 2) 再结算反馈映射为对应动作 + self._settle_feedback(target_avatar, feedback) + # 3) 反馈事件（进入侧边栏与双方历史） + feedback_event = Event(self.world.month_stamp, f"{target_avatar.name} 对 {self.avatar.name} 的反馈：{feedback}") + self.avatar.add_event(feedback_event) + target_avatar.add_event(feedback_event) + # 4) 记录历史（文本记录） + self._apply_feedback(target_avatar, feedback) + + # 互动力一般是一次性的即时动作 + def is_finished(self, target_avatar: "Avatar") -> bool: # type: ignore[override] + return True + + def get_event(self, target_avatar: "Avatar|str") -> Event: # type: ignore[override] + target_name = target_avatar if isinstance(target_avatar, str) else target_avatar.name + return Event(self.world.month_stamp, f"{self.avatar.name} 对 {target_name} 发起 {getattr(self, 'ACTION_NAME', self.name)}") + + @property + def is_doable(self) -> bool: # type: ignore[override] + # 一般来讲，必须和对象avatar在同一区域 + return True + + +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 AttackInteract(MutualAction, ActualActionMixin): + """攻击互动：被攻击者的反馈。""" + ACTION_NAME = "攻击" + COMMENT = "对目标进行攻击。" + DOABLES_REQUIREMENTS = "与目标处于同一区域" + PARAMS = {"target_avatar": "AvatarName"} + FEEDBACK_ACTIONS = ["MoveAwayFromAvatar", "Battle"] + + def _settle_feedback(self, target_avatar: "Avatar", feedback_name: str) -> None: + fb = str(feedback_name).strip() + if fb == "MoveAwayFromAvatar": + 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 MoveAwayFromAvatar(DefineAction, ActualActionMixin): + COMMENT = "远离指定角色" + DOABLES_REQUIREMENTS = "任何时候都可以执行" + PARAMS = {"avatar_name": "AvatarName"} + def _execute(self, avatar_name: str) -> None: + target = None + for v in self.world.avatar_manager.avatars.values(): + if v.name == avatar_name: + target = v + break + 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 is_finished(self, avatar_name: str) -> bool: + return True + def get_event(self, avatar_name: str) -> Event: + return Event(self.world.month_stamp, f"{self.avatar.name} 远离 {avatar_name}") + @property + def is_doable(self) -> bool: + return True + + +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 is_finished(self, region: str) -> bool: + return True + def get_event(self, region: str) -> Event: + return Event(self.world.month_stamp, f"{self.avatar.name} 离开 {region}") + @property + def is_doable(self) -> bool: + return True + + +class Battle(DefineAction, ActualActionMixin): + COMMENT = "与目标进行对战，判定胜负" + DOABLES_REQUIREMENTS = "任何时候都可以执行" + PARAMS = {"avatar_name": "AvatarName"} + def _execute(self, avatar_name: str) -> None: + target = None + for v in self.world.avatar_manager.avatars.values(): + if v.name == avatar_name: + target = v + break + if target is None: + return + winner, loser, _ = decide_battle(self.avatar, target) + # 简化：失败者HP小额扣减 + if hasattr(loser, "hp"): + loser.hp.reduce(10) + def is_finished(self, avatar_name: str) -> bool: + return True + def get_event(self, avatar_name: str) -> Event: + return Event(self.world.month_stamp, f"{self.avatar.name} 与 {avatar_name} 进行对战") + @property + def is_doable(self) -> bool: + return True diff --git a/src/run/run.py b/src/run/run.py index 819de21..d62fe2a 100644 --- a/src/run/run.py +++ b/src/run/run.py @@ -89,11 +89,9 @@ def make_avatars(world: World, count: int = 12, current_month_stamp: MonthStamp # —— 为演示添加少量示例关系 —— avatar_list = list(avatars.values()) if len(avatar_list) >= 2: - # 朋友 - avatar_list[0].set_relation(avatar_list[1], Relation.FRIEND) + avatar_list[0].set_relation(avatar_list[1], Relation.ENEMY) if len(avatar_list) >= 4: - # 仇人 - avatar_list[2].set_relation(avatar_list[3], Relation.ENEMY) + avatar_list[2].set_relation(avatar_list[3], Relation.FRIEND) if len(avatar_list) >= 6: # 师徒（随意指派方向，关系对称） avatar_list[4].set_relation(avatar_list[5], Relation.MASTER_APPRENTICE) diff --git a/src/sim/simulator.py b/src/sim/simulator.py index 41600ee..03d7da5 100644 --- a/src/sim/simulator.py +++ b/src/sim/simulator.py @@ -54,7 +54,9 @@ class Simulator: # 结算角色行为 for avatar_id, avatar in self.world.avatar_manager.avatars.items(): - await avatar.act() + new_events = await avatar.act() + if new_events: + events.extend(new_events) if avatar.death_by_old_age(): death_avatar_ids.append(avatar_id) event = Event(self.world.month_stamp, f"{avatar.name} 老死了，时年{avatar.age.get_age()}岁") diff --git a/static/templates/mutual_action.txt b/static/templates/mutual_action.txt new file mode 100644 index 0000000..7923501 --- /dev/null +++ b/static/templates/mutual_action.txt @@ -0,0 +1,16 @@ +你是一个决策者，这是一个修仙的仙侠世界，你负责来决定两个NPC的下一步行为。 + +你需要进行决策的NPC的dict[AvatarName, info]为 +{avatar_infos} +正在进行的动作为：{avatar_name_1}向{avatar_name_2}发起了动作：{action_name}。这个动作的意味为{action_info} +{avatar_name_2}可以进行的选择为： +{FEEDBACK_ACTIONS} + +注意，只返回json格式的结果。 +只返回{avatar_name_2}的行动，格式为： +{{ + {avatar_name_2}: {{ + "thinking": ..., // 简单思考应该怎么决策 + "feedback": ... // 面对{avatar_name_1}的行为的合法feedback action name + }} +}} \ No newline at end of file