yexuejc/cultivation-world-simulator
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

add mutual action

Browse Source
This commit is contained in:
bridge
2025-09-21 18:44:34 +08:00
parent 0e49362e0f
commit 3b21d99718
8 changed files with 416 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
src/classes/actions.py Normal file
View File

@@ -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)

2
src/classes/ai.py
View File

@@ -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

68
src/classes/avatar.py
View File

@@ -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)
# 返回并清空待派发事件
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:
"""

47
src/classes/battle.py Normal file
View File

@@ -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

235
src/classes/mutual_action.py Normal file
View File

@@ -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

6
src/run/run.py
View File

@@ -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)

4
src/sim/simulator.py
View File

@@ -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()}岁")

16
static/templates/mutual_action.txt Normal file
View File

@@ -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
}}
}}