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

add long time memory

Browse Source
This commit is contained in:
bridge
2025-11-18 00:39:06 +08:00
parent 5ae2538d86
commit 307ad405ac
13 changed files with 167 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

29
src/classes/action/action.py
View File

@@ -126,7 +126,34 @@ class ActualActionMixin():
不一定是多个step，也有可能就一个step。
新接口：子类必须实现 can_start/start/step/finish。
类变量：
- IS_MAJOR: 是否为大事（长期记忆），默认False（小事/短期记忆）
"""
# 默认为小事（短期记忆）
IS_MAJOR: bool = False
def create_event(self, content: str, related_avatars=None) -> Event:
"""
创建事件的辅助方法，自动带上is_major属性
Args:
content: 事件内容
related_avatars: 相关角色ID列表
Returns:
Event对象，is_major根据当前Action的IS_MAJOR类变量设置
"""
from src.classes.action.action import Action
# 获取当前类的IS_MAJOR属性
is_major = self.__class__.IS_MAJOR if hasattr(self.__class__, 'IS_MAJOR') else False
return Event(
month_stamp=self.world.month_stamp,
content=content,
related_avatars=related_avatars,
is_major=is_major
)
@abstractmethod
def can_start(self, **params) -> tuple[bool, str]:
@@ -164,7 +191,7 @@ class TimedAction(DefineAction, ActualActionMixin):
duration_months: int = 1
def step(self, **params) -> ActionResult:
if getattr(self, 'start_monthstamp', None) is None:
if not hasattr(self, 'start_monthstamp') or self.start_monthstamp is None:
self.start_monthstamp = self.world.month_stamp
self._execute(**params)
done = (self.world.month_stamp - self.start_monthstamp) >= (self.duration_months - 1)

10
src/classes/action/battle.py
View File

@@ -15,6 +15,8 @@ class Battle(InstantAction):
STORY_PROMPT: str | None = (
"不要出现具体血量、伤害点数或任何数值表达。"
)
# 战斗是大事（长期记忆）
IS_MAJOR: bool = True
def _get_target(self, avatar_name: str):
"""
@@ -62,7 +64,7 @@ class Battle(InstantAction):
rel_ids.append(target.id)
except Exception:
pass
event = Event(self.world.month_stamp, f"{self.avatar.name} 对 {target_name} 发起战斗（战斗力：{self.avatar.name} {int(s_att)} vs {target_name} {int(s_def)}）", related_avatars=rel_ids)
event = Event(self.world.month_stamp, f"{self.avatar.name} 对 {target_name} 发起战斗（战斗力：{self.avatar.name} {int(s_att)} vs {target_name} {int(s_def)}）", related_avatars=rel_ids, is_major=True)
# 记录开始事件内容，供故事生成使用
self._start_event_content = event.content
return event
@@ -83,13 +85,13 @@ class Battle(InstantAction):
rel_ids.append(t.id)
except Exception:
pass
result_event = Event(self.world.month_stamp, result_text, related_avatars=rel_ids)
result_event = Event(self.world.month_stamp, result_text, related_avatars=rel_ids, is_major=True)
# 生成战斗小故事（同步调用，与其他动作保持一致）
target = self._get_target(avatar_name)
start_text = getattr(self, "_start_event_content", "") or result_event.content
start_text = self._start_event_content if hasattr(self, '_start_event_content') else result_event.content
story = StoryTeller.tell_from_actors(start_text, result_event.content, self.avatar, target, prompt=self.STORY_PROMPT)
story_event = Event(self.world.month_stamp, story, related_avatars=rel_ids)
story_event = Event(self.world.month_stamp, story, related_avatars=rel_ids, is_story=True)
return [result_event, story_event]

38
src/classes/action/breakthrough.py
View File

@@ -30,6 +30,8 @@ class Breakthrough(TimedAction):
PARAMS = {}
# 冷却：突破应当有CD，避免连刷
ACTION_CD_MONTHS: int = 3
# 突破是大事（长期记忆）
IS_MAJOR: bool = True
# 保留类级常量声明，实际读取模块级配置
def calc_success_rate(self) -> float:
@@ -65,8 +67,8 @@ class Breakthrough(TimedAction):
# 记录结果用于 finish 事件
self._last_result = (
"success",
getattr(old_realm, "value", str(old_realm)),
getattr(new_realm, "value", str(new_realm)),
old_realm.value,
new_realm.value,
)
else:
# 突破失败：减少最大寿元上限
@@ -100,12 +102,11 @@ class Breakthrough(TimedAction):
return (ok, "" if ok else "当前不处于瓶颈，无法突破")
def start(self) -> Event:
# 清理状态
# 初始化状态
self._last_result = None
self._success_rate_cached = None
# 预判是否生成故事与选择劫难
old_realm = self.avatar.cultivation_progress.realm
# 仅基于出发境界判断是否生成故事
self._gen_story = old_realm in ALLOW_STORY_FROM_REALMS
if self._gen_story:
self._calamity = TribulationSelector.choose_tribulation(self.avatar)
@@ -113,36 +114,33 @@ class Breakthrough(TimedAction):
else:
self._calamity = None
self._calamity_other = None
return Event(self.world.month_stamp, f"{self.avatar.name} 开始尝试突破境界", related_avatars=[self.avatar.id])
return Event(self.world.month_stamp, f"{self.avatar.name} 开始尝试突破境界", related_avatars=[self.avatar.id], is_major=True)
# TimedAction 已统一 step 逻辑
def finish(self) -> list[Event]:
res = getattr(self, "_last_result", None)
if not (isinstance(res, tuple) and res):
if not self._last_result:
return []
result_ok = res[0] == "success"
if not getattr(self, "_gen_story", False):
result_ok = self._last_result[0] == "success"
if not self._gen_story:
# 不生成故事：不出现劫难，仅简单结果
core_text = f"{self.avatar.name} 突破{'成功' if result_ok else '失败'}"
return [Event(self.world.month_stamp, core_text, related_avatars=[self.avatar.id])]
return [Event(self.world.month_stamp, core_text, related_avatars=[self.avatar.id], is_major=True)]
calamity = getattr(self, "_calamity", "劫难")
calamity = self._calamity
core_text = f"{self.avatar.name} 遭遇了{calamity}劫难，突破{'成功' if result_ok else '失败'}"
rel_ids = [self.avatar.id]
other = getattr(self, "_calamity_other", None)
if other is not None:
if self._calamity_other is not None:
try:
rel_ids.append(other.id)
rel_ids.append(self._calamity_other.id)
except Exception:
pass
events: list[Event] = [Event(self.world.month_stamp, core_text, related_avatars=rel_ids)]
events: list[Event] = [Event(self.world.month_stamp, core_text, related_avatars=rel_ids, is_major=True)]
if True:
# 故事参与者：本体 +（可选）相关角色
prompt = TribulationSelector.get_story_prompt(str(calamity))
story = StoryTeller.tell_from_actors(core_text, ("突破成功" if result_ok else "突破失败"), self.avatar, getattr(self, "_calamity_other", None), prompt=prompt)
events.append(Event(self.world.month_stamp, story, related_avatars=rel_ids))
# 故事参与者：本体 +（可选）相关角色
prompt = TribulationSelector.get_story_prompt(str(calamity))
story = StoryTeller.tell_from_actors(core_text, ("突破成功" if result_ok else "突破失败"), self.avatar, self._calamity_other, prompt=prompt)
events.append(Event(self.world.month_stamp, story, related_avatars=rel_ids, is_story=True))
return events

21
src/classes/avatar.py
View File

@@ -241,7 +241,6 @@ class Avatar(AvatarSaveMixin, AvatarLoadMixin):
spirit_animal_info = self.spirit_animal.get_info() if self.spirit_animal is not None else "无"
info_dict = {
"id": self.id,
"名字": self.name,
"性别": str(self.gender),
"年龄": str(self.age),
@@ -509,12 +508,11 @@ class Avatar(AvatarSaveMixin, AvatarLoadMixin):
添加事件：
- to_sidebar: 是否进入全局侧边栏（通过 Avatar._pending_events 暂存）
- to_history: 兼容参数，已废弃（统一改为通过 World.event_manager 查询历史）
注意：事件会先存入_pending_events，统一由Simulator写入event_manager，避免重复
"""
if to_sidebar:
self._pending_events.append(event)
# 侧边栏类事件通常不在 Simulator 的 events 列表里，直接记入全局事件管理器
em = self.world.event_manager
em.add_event(event)
def get_action_space_str(self) -> str:
action_space = self.get_action_space()
@@ -547,14 +545,19 @@ class Avatar(AvatarSaveMixin, AvatarLoadMixin):
for other in co_region_avatars[:8]:
observed.append(f"{other.name}，境界：{other.cultivation_progress.get_info()}")
# 历史事件改为从全局事件管理器查询
n = CONFIG.social.event_context_num
# 历史事件改为从全局事件管理器分类查询
em = self.world.event_manager
events = em.get_events_by_avatar(self.id, limit=n)
history_list = [str(e) for e in events]
major_limit = CONFIG.social.major_event_context_num
minor_limit = CONFIG.social.minor_event_context_num
major_events = em.get_major_events_by_avatar(self.id, limit=major_limit)
minor_events = em.get_minor_events_by_avatar(self.id, limit=minor_limit)
major_list = [str(e) for e in major_events]
minor_list = [str(e) for e in minor_events]
info["观察到的角色"] = observed
info["历史事件"] = history_list
info["长期记忆"] = major_list
info["短期记忆"] = minor_list
return info
def get_hover_info(self) -> list[str]:

12
src/classes/event.py
View File

@@ -12,6 +12,10 @@ class Event:
content: str
# 相关角色ID列表；若与任何角色无关则为 None
related_avatars: Optional[List[str]] = None
# 是否为大事（长期记忆），默认False（小事/短期记忆）
is_major: bool = False
# 是否为故事事件（不进入记忆索引），默认False
is_story: bool = False
def __str__(self) -> str:
year = self.month_stamp.get_year()
@@ -23,7 +27,9 @@ class Event:
return {
"month_stamp": int(self.month_stamp),
"content": self.content,
"related_avatars": self.related_avatars
"related_avatars": self.related_avatars,
"is_major": self.is_major,
"is_story": self.is_story
}
@classmethod
@@ -32,7 +38,9 @@ class Event:
return cls(
month_stamp=MonthStamp(data["month_stamp"]),
content=data["content"],
related_avatars=data.get("related_avatars")
related_avatars=data.get("related_avatars"),
is_major=data.get("is_major", False),
is_story=data.get("is_story", False)
)
class NullEvent:

54
src/classes/event_manager.py
View File

@@ -20,6 +20,12 @@ class EventManager:
self._by_id: Dict[str, Event] = {}
self._by_avatar: Dict[str, deque[Event]] = defaultdict(deque)
self._by_pair: Dict[frozenset[str], deque[Event]] = defaultdict(deque)
# 按角色分类的大事/小事索引
self._by_avatar_major: Dict[str, deque[Event]] = defaultdict(deque)
self._by_avatar_minor: Dict[str, deque[Event]] = defaultdict(deque)
# 按角色对分类的大事/小事索引
self._by_pair_major: Dict[frozenset[str], deque[Event]] = defaultdict(deque)
self._by_pair_minor: Dict[frozenset[str], deque[Event]] = defaultdict(deque)
def _append_with_limit(self, dq: deque, item: Event) -> None:
dq.append(item)
@@ -39,15 +45,29 @@ class EventManager:
self._events.popleft()
# 分索引：按人/人对
rel = getattr(event, "related_avatars", None) or []
rel = event.related_avatars or []
rel_unique = list(dict.fromkeys(rel)) # 去重但保持顺序
for aid in rel_unique:
self._append_with_limit(self._by_avatar[aid], event)
# 仅当且仅当“恰有两位参与者”时建立按人对索引
# 故事事件进入小事索引，不进入大事索引
if event.is_story:
self._append_with_limit(self._by_avatar_minor[aid], event)
elif event.is_major:
self._append_with_limit(self._by_avatar_major[aid], event)
else:
self._append_with_limit(self._by_avatar_minor[aid], event)
# 仅当且仅当"恰有两位参与者"时建立按人对索引
if len(rel_unique) == 2:
a, b = rel_unique[0], rel_unique[1]
pair_key = frozenset([a, b])
self._append_with_limit(self._by_pair[pair_key], event)
# 角色对也建立分类索引
if event.is_story:
self._append_with_limit(self._by_pair_minor[pair_key], event)
elif event.is_major:
self._append_with_limit(self._by_pair_major[pair_key], event)
else:
self._append_with_limit(self._by_pair_minor[pair_key], event)
# —— 查询接口 ——
def get_recent_events(self, limit: int = 100) -> List[Event]:
@@ -68,4 +88,34 @@ class EventManager:
return []
return list(dq)[-limit:]
def get_major_events_by_avatar(self, avatar_id: str, *, limit: int = 10) -> List[Event]:
"""获取角色的大事（长期记忆）"""
dq = self._by_avatar_major.get(avatar_id)
if not dq:
return []
return list(dq)[-limit:]
def get_minor_events_by_avatar(self, avatar_id: str, *, limit: int = 10) -> List[Event]:
"""获取角色的小事（短期记忆）"""
dq = self._by_avatar_minor.get(avatar_id)
if not dq:
return []
return list(dq)[-limit:]
def get_major_events_between(self, avatar_id1: str, avatar_id2: str, *, limit: int = 10) -> List[Event]:
"""获取两个角色之间的大事（长期记忆）"""
key = frozenset([avatar_id1, avatar_id2])
dq = self._by_pair_major.get(key)
if not dq:
return []
return list(dq)[-limit:]
def get_minor_events_between(self, avatar_id1: str, avatar_id2: str, *, limit: int = 10) -> List[Event]:
"""获取两个角色之间的小事（短期记忆）"""
key = frozenset([avatar_id1, avatar_id2])
dq = self._by_pair_minor.get(key)
if not dq:
return []
return list(dq)[-limit:]

4
src/classes/fortune.py
View File

@@ -475,11 +475,11 @@ async def try_trigger_fortune(avatar: Avatar) -> list[Event]:
story_prompt = "请据此写100~150字小故事。"
month_at_finish = avatar.world.month_stamp
base_event = Event(month_at_finish, event_text, related_avatars=related_avatars)
base_event = Event(month_at_finish, event_text, related_avatars=related_avatars, is_major=True)
# 生成故事事件
story = await StoryTeller.tell_from_actors_async(event_text, res_text, *actors_for_story, prompt=story_prompt)
story_event = Event(month_at_finish, story, related_avatars=related_avatars)
story_event = Event(month_at_finish, story, related_avatars=related_avatars, is_story=True)
# 返回基础事件和故事事件
return [base_event, story_event]

2
src/classes/mutual_action/attack.py
View File

@@ -20,6 +20,8 @@ class Attack(MutualAction):
STORY_PROMPT: str = ""
# 攻击冷却：避免同月连刷攻击
ACTION_CD_MONTHS: int = 3
# 攻击是大事（长期记忆）
IS_MAJOR: bool = True
def _can_start(self, target: "Avatar") -> tuple[bool, str]:
"""攻击无额外检查条件"""

9
src/classes/mutual_action/conversation.py
View File

@@ -54,10 +54,13 @@ class Conversation(MutualAction):
# 可能取消的关系
possible_cancel_relations = [relation_display_names[r] for r in get_possible_cancel_relations(target_avatar, self.avatar)]
# 历史上下文：仅双方共同经历的最近事件
n = CONFIG.social.event_context_num
# 历史上下文：仅双方共同经历的大事和小事
major_limit = CONFIG.social.major_event_context_num
minor_limit = CONFIG.social.minor_event_context_num
em = self.world.event_manager
pair_recent_events = [str(e) for e in em.get_events_between(self.avatar.id, target_avatar.id, limit=n)]
major_events = em.get_major_events_between(self.avatar.id, target_avatar.id, limit=major_limit)
minor_events = em.get_minor_events_between(self.avatar.id, target_avatar.id, limit=minor_limit)
pair_recent_events = [str(e) for e in major_events + minor_events]
return {
"avatar_infos": avatar_infos,
"avatar_name_1": avatar_name_1,

12
src/classes/mutual_action/dual_cultivation.py
View File

@@ -22,7 +22,7 @@ class DualCultivation(MutualAction):
- 仅当目标在交互范围内
- 目标可以选择 接受 或 拒绝
- 若接受：发起者获得大量修为（约为修炼的 3~5 倍，随对方等级浮动），目标不获得修为
- 成功进入后生成一段“恋爱/双修”的小故事
- 成功进入后生成一段"恋爱/双修"的小故事
"""
ACTION_NAME = "双修"
@@ -34,6 +34,8 @@ class DualCultivation(MutualAction):
STORY_PROMPT: str | None = "两位修士在双修过程中情愫暗生，以含蓄、雅致的文字描绘一段暧昧而不露骨的双修体验，体现彼此性格、境界差异与甜蜜的恋爱时光。不要体现经验的数值。"
# 双修的社交冷却：避免频繁请求
ACTION_CD_MONTHS: int = 3
# 双修是大事（长期记忆）
IS_MAJOR: bool = True
def _get_template_path(self) -> Path:
# 复用 mutual_action 模板，仅需返回 Accept/Reject
@@ -53,7 +55,7 @@ class DualCultivation(MutualAction):
rel_ids = [self.avatar.id]
if target is not None:
rel_ids.append(target.id)
event = Event(self.world.month_stamp, f"{self.avatar.name} 邀请 {target_name} 进行双修", related_avatars=rel_ids)
event = Event(self.world.month_stamp, f"{self.avatar.name} 邀请 {target_name} 进行双修", related_avatars=rel_ids, is_major=True)
# 仅写入历史
self.avatar.add_event(event, to_sidebar=False)
if target is not None:
@@ -104,17 +106,17 @@ class DualCultivation(MutualAction):
if success:
gain = int(self._dual_exp_gain)
result_text = f"{self.avatar.name} 与 {target.name} 成功双修，{self.avatar.name} 获得修为经验 +{gain} 点"
result_event = Event(self.world.month_stamp, result_text, related_avatars=[self.avatar.id, target.id])
result_event = Event(self.world.month_stamp, result_text, related_avatars=[self.avatar.id, target.id], is_major=True)
events.append(result_event)
# 生成恋爱/双修小故事：使用 StoryTeller 便捷方法
start_text = self._start_event_content or result_event.content
story = StoryTeller.tell_from_actors(start_text, result_event.content, self.avatar, target, prompt=self.STORY_PROMPT)
story_event = Event(self.world.month_stamp, story, related_avatars=[self.avatar.id, target.id])
story_event = Event(self.world.month_stamp, story, related_avatars=[self.avatar.id, target.id], is_story=True)
events.append(story_event)
else:
result_text = f"{target.name} 拒绝了与 {self.avatar.name} 的双修"
result_event = Event(self.world.month_stamp, result_text, related_avatars=[self.avatar.id, target.id])
result_event = Event(self.world.month_stamp, result_text, related_avatars=[self.avatar.id, target.id], is_major=True)
events.append(result_event)
return events

17
src/classes/mutual_action/mutual_action.py
View File

@@ -66,12 +66,13 @@ class MutualAction(DefineAction, LLMAction, TargetingMixin):
avatar_name_1: self.avatar.get_info(detailed=False),
avatar_name_2: target_avatar.get_info(detailed=False),
}
# 历史上下文：仅双方共同经历的最近事件
n = CONFIG.social.event_context_num
pair_recent_events: list[str] = []
# 历史上下文：仅双方共同经历的大事和小事
major_limit = CONFIG.social.major_event_context_num
minor_limit = CONFIG.social.minor_event_context_num
em = self.world.event_manager
pair_recent_events = [str(e) for e in em.get_events_between(self.avatar.id, target_avatar.id, limit=n)]
major_events = em.get_major_events_between(self.avatar.id, target_avatar.id, limit=major_limit)
minor_events = em.get_minor_events_between(self.avatar.id, target_avatar.id, limit=minor_limit)
pair_recent_events = [str(e) for e in major_events + minor_events]
feedback_actions = self.FEEDBACK_ACTIONS
comment = self.COMMENT
action_name = self.ACTION_NAME
@@ -189,11 +190,13 @@ class MutualAction(DefineAction, LLMAction, TargetingMixin):
"""
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)
action_name = self.ACTION_NAME
rel_ids = [self.avatar.id]
if target is not None:
rel_ids.append(target.id)
event = Event(self.world.month_stamp, f"{self.avatar.name} 对 {target_name} 发起 {action_name}", related_avatars=rel_ids)
# 根据IS_MAJOR类变量设置事件类型
is_major = self.__class__.IS_MAJOR if hasattr(self.__class__, 'IS_MAJOR') else False
event = Event(self.world.month_stamp, f"{self.avatar.name} 对 {target_name} 发起 {action_name}", related_avatars=rel_ids, is_major=is_major)
# 仅写入历史，避免与提交阶段重复推送到侧边栏
self.avatar.add_event(event, to_sidebar=False)
if target is not None:

13
src/classes/story_teller.py
View File

@@ -47,7 +47,8 @@ class StoryTeller:
@staticmethod
def _collect_recent_events(*actors: "Avatar") -> list[str]:
from src.utils.config import CONFIG as _CONFIG
n = _CONFIG.social.event_context_num
major_limit = _CONFIG.social.major_event_context_num
minor_limit = _CONFIG.social.minor_event_context_num
world = None
for av in actors:
if av is not None:
@@ -58,11 +59,17 @@ class StoryTeller:
em = world.event_manager
non_null = [a for a in actors if a is not None]
if len(non_null) >= 2:
# 两人故事：获取两人的大事和小事
a1, a2 = non_null[0], non_null[1]
return [str(e) for e in em.get_events_between(a1.id, a2.id, limit=n)]
major_events = em.get_major_events_between(a1.id, a2.id, limit=major_limit)
minor_events = em.get_minor_events_between(a1.id, a2.id, limit=minor_limit)
return [str(e) for e in major_events + minor_events]
if non_null:
# 单人故事：获取单人的大事和小事
a = non_null[0]
return [str(e) for e in em.get_events_by_avatar(a.id, limit=n)]
major_events = em.get_major_events_by_avatar(a.id, limit=major_limit)
minor_events = em.get_minor_events_by_avatar(a.id, limit=minor_limit)
return [str(e) for e in major_events + minor_events]
return []
@staticmethod