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

fix long events calling

Browse Source
This commit is contained in:
bridge
2025-11-19 01:06:42 +08:00
parent d2cf568154
commit c341a1fddd
13 changed files with 93 additions and 124 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/classes/ai.py
View File

@@ -63,7 +63,7 @@ class LLMAI(AI):
avatar_infos = {}
for avatar in avatars_to_decide:
observed = world.get_observable_avatars(avatar)
avatar_infos[avatar.name] = avatar.get_prompt_info(observed)
avatar_infos[avatar.name] = avatar.get_expanded_info(observed)
general_action_infos = ACTION_INFOS_STR
info = {
"avatar_infos": avatar_infos,

30
src/classes/avatar.py
View File

@@ -547,11 +547,21 @@ class Avatar(AvatarSaveMixin, AvatarLoadMixin):
action_space = [action.name for action in doable_actions]
return action_space
def get_prompt_info(self, co_region_avatars: Optional[List["Avatar"]] = None) -> dict:
def get_expanded_info(
self,
co_region_avatars: Optional[List["Avatar"]] = None,
other_avatar: Optional["Avatar"] = None,
detailed: bool = False
) -> dict:
"""
获取角色提示词信息，返回 dict。
获取角色的扩展信息，包含基础信息、观察到的角色和事件历史。
Args:
co_region_avatars: 同区域的其他角色列表，用于"观察到的角色"字段
other_avatar: 另一个角色，如果提供则返回两人共同经历的事件，否则返回单人事件
detailed: 是否返回详细信息
"""
info = self.get_info(detailed=False)
info = self.get_info(detailed=detailed)
observed: list[str] = []
if co_region_avatars:
@@ -562,15 +572,21 @@ class Avatar(AvatarSaveMixin, AvatarLoadMixin):
em = self.world.event_manager
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)
# 根据是否提供 other_avatar 决定获取单人事件还是双人共同事件
if other_avatar is not None:
major_events = em.get_major_events_between(self.id, other_avatar.id, limit=major_limit)
minor_events = em.get_minor_events_between(self.id, other_avatar.id, limit=minor_limit)
else:
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["长期记忆"] = major_list
info["短期记忆"] = minor_list
info["重大事件"] = major_list
info["短期事件"] = minor_list
info["长期目标"] = self.long_term_objective.content if self.long_term_objective else "无"
return info

20
src/classes/long_term_objective.py
View File

@@ -77,31 +77,17 @@ async def generate_long_term_objective(avatar: "Avatar") -> Optional[LongTermObj
Returns:
生成的LongTermObjective对象，失败则返回None
"""
# 准备世界信息
world_info = avatar.world.get_info()
# 准备角色信息
avatar_info = avatar.get_info(detailed=True)
avatar_info_str = "\n".join([f"{k}: {v}" for k, v in avatar_info.items()])
# 获取事件历史
em = avatar.world.event_manager
major_limit = CONFIG.social.major_event_context_num
minor_limit = CONFIG.social.minor_event_context_num
major_events = em.get_major_events_by_avatar(avatar.id, limit=major_limit)
minor_events = em.get_minor_events_by_avatar(avatar.id, limit=minor_limit)
major_events_str = "\n".join([f"- {str(e)}" for e in major_events]) if major_events else "无"
minor_events_str = "\n".join([f"- {str(e)}" for e in minor_events]) if minor_events else "无"
# 获取 expanded_info（包含详细信息和事件历史）
expanded_info = avatar.get_expanded_info(detailed=True)
# 准备模板参数
template_path = CONFIG.paths.templates / "long_term_objective.txt"
infos = {
"world_info": world_info,
"avatar_info": avatar_info_str,
"major_events": major_events_str,
"minor_events": minor_events_str
"avatar_info": expanded_info,
}
# 调用LLM并自动解析JSON（使用fast模型）

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

@@ -43,31 +43,27 @@ class Conversation(MutualAction):
def _build_prompt_infos(self, target_avatar: "Avatar") -> dict:
avatar_name_1 = self.avatar.name
avatar_name_2 = target_avatar.name
# 交谈：使用详细信息，便于生成更丰富对话
# avatar1 使用 expanded_info（包含详细信息和共同事件），避免重复
expanded_info = self.avatar.get_expanded_info(other_avatar=target_avatar, detailed=True)
avatar_infos = {
avatar_name_1: self.avatar.get_info(detailed=True),
avatar_name_1: expanded_info,
avatar_name_2: target_avatar.get_info(detailed=True),
}
# 可能的后天关系（转中文名，给模板阅读）
# 注意：这里计算的是 target 相对于 avatar 的可能关系
possible_new_relations = [relation_display_names[r] for r in get_possible_new_relations(self.avatar, target_avatar)]
# 可能取消的关系
possible_cancel_relations = [relation_display_names[r] for r in get_possible_cancel_relations(target_avatar, self.avatar)]
# 历史上下文：仅双方共同经历的大事和小事
major_limit = CONFIG.social.major_event_context_num
minor_limit = CONFIG.social.minor_event_context_num
em = self.world.event_manager
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,
"avatar_name_2": avatar_name_2,
"possible_new_relations": possible_new_relations,
"possible_cancal_relations": possible_cancel_relations, # 保持模板中的拼写
"recent_events": pair_recent_events,
}
def _can_start(self, target: "Avatar") -> tuple[bool, str]:

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

@@ -60,19 +60,15 @@ class MutualAction(DefineAction, LLMAction, TargetingMixin):
def _build_prompt_infos(self, target_avatar: "Avatar") -> dict:
avatar_name_1 = self.avatar.name
avatar_name_2 = target_avatar.name
# avatar infos 仅放入与两人相关的提示，避免超长
# avatar1 使用 expanded_info（包含非详细信息和共同事件），避免重复
expanded_info = self.avatar.get_expanded_info(other_avatar=target_avatar, detailed=False)
avatar_infos = {
# 决策：使用非详细信息
avatar_name_1: self.avatar.get_info(detailed=False),
avatar_name_1: expanded_info,
avatar_name_2: target_avatar.get_info(detailed=False),
}
# 历史上下文：仅双方共同经历的大事和小事
major_limit = CONFIG.social.major_event_context_num
minor_limit = CONFIG.social.minor_event_context_num
em = self.world.event_manager
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
@@ -83,7 +79,6 @@ class MutualAction(DefineAction, LLMAction, TargetingMixin):
"action_name": action_name,
"action_info": comment,
"feedback_actions": feedback_actions,
"recent_events": pair_recent_events,
}
def _call_llm_feedback(self, infos: dict) -> dict:

19
src/classes/nickname.py
View File

@@ -63,26 +63,13 @@ async def generate_nickname(avatar: "Avatar") -> Optional[str]:
生成的绰号，失败则返回None
"""
try:
# 准备角色信息
avatar_info = avatar.get_info(detailed=True)
avatar_info_str = "\n".join([f"{k}: {v}" for k, v in avatar_info.items()])
# 获取事件历史（根据配置的阈值获取对应数量的事件）
em = avatar.world.event_manager
major_threshold = CONFIG.nickname.major_event_threshold
minor_threshold = CONFIG.nickname.minor_event_threshold
major_events = em.get_major_events_by_avatar(avatar.id, limit=major_threshold)
minor_events = em.get_minor_events_by_avatar(avatar.id, limit=minor_threshold)
major_events_str = "\n".join([f"- {str(e)}" for e in major_events]) if major_events else "无"
minor_events_str = "\n".join([f"- {str(e)}" for e in minor_events]) if minor_events else "无"
# 获取 expanded_info（包含详细信息和事件历史）
expanded_info = avatar.get_expanded_info(detailed=True)
# 准备模板参数
template_path = CONFIG.paths.templates / "nickname.txt"
infos = {
"avatar_info": avatar_info_str,
"major_events": major_events_str,
"minor_events": minor_events_str
"avatar_info": expanded_info,
}
# 调用LLM并自动解析JSON

87
src/classes/story_teller.py
View File

@@ -45,48 +45,37 @@ class StoryTeller:
return infos
@staticmethod
def _collect_recent_events(*actors: "Avatar") -> list[str]:
from src.utils.config import CONFIG as _CONFIG
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:
world = av.world
break
if world is None:
return []
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]
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]
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
def tell_story(avatar_infos: Dict[str, dict], event: str, res: str, STORY_PROMPT: str = "", *, recent_events: list[str] | None = None) -> str:
def tell_story(event: str, res: str, *actors: "Avatar", prompt: str = "") -> str:
"""
基于 `static/templates/story.txt` 模板生成小故事。
始终使用 fast 模式以提升速度。
失败时返回降级版文案，避免中断流程。
Args:
event: 事件描述
res: 结果描述
*actors: 参与的角色（1-2个）
prompt: 可选的故事提示词
"""
# 构建 avatar_infos，第一个 avatar 使用 expanded_info
non_null = [a for a in actors if a is not None]
avatar_infos: Dict[str, dict] = {}
if len(non_null) >= 2:
# 双人故事：第一个用 expanded_info（包含共同事件），第二个用 detailed info
avatar_infos[non_null[0].name] = non_null[0].get_expanded_info(other_avatar=non_null[1], detailed=True)
avatar_infos[non_null[1].name] = non_null[1].get_info(detailed=True)
elif non_null:
# 单人故事：直接用 expanded_info
avatar_infos[non_null[0].name] = non_null[0].get_expanded_info(detailed=True)
template_path = CONFIG.paths.templates / "story.txt"
infos = {
"avatar_infos": avatar_infos,
"event": event,
"res": res,
"style": random.choice(story_styles),
"story_prompt": STORY_PROMPT or "",
"recent_events": (recent_events or []),
"story_prompt": prompt,
}
try:
data = get_prompt_and_call_llm(template_path, infos, mode="fast")
@@ -100,18 +89,35 @@ class StoryTeller:
return f"{event}。{res}。{style}"
@staticmethod
async def tell_story_async(avatar_infos: Dict[str, dict], event: str, res: str, STORY_PROMPT: str = "", *, recent_events: list[str] | None = None) -> str:
async def tell_story_async(event: str, res: str, *actors: "Avatar", prompt: str = "") -> str:
"""
异步版本：生成小故事，失败时返回降级文案。
Args:
event: 事件描述
res: 结果描述
*actors: 参与的角色（1-2个）
prompt: 可选的故事提示词
"""
# 构建 avatar_infos，第一个 avatar 使用 expanded_info
non_null = [a for a in actors if a is not None]
avatar_infos: Dict[str, dict] = {}
if len(non_null) >= 2:
# 双人故事：第一个用 expanded_info（包含共同事件），第二个用 detailed info
avatar_infos[non_null[0].name] = non_null[0].get_expanded_info(other_avatar=non_null[1], detailed=True)
avatar_infos[non_null[1].name] = non_null[1].get_info(detailed=True)
elif non_null:
# 单人故事：直接用 expanded_info
avatar_infos[non_null[0].name] = non_null[0].get_expanded_info(detailed=True)
template_path = CONFIG.paths.templates / "story.txt"
infos = {
"avatar_infos": avatar_infos,
"event": event,
"res": res,
"style": random.choice(story_styles),
"story_prompt": STORY_PROMPT or "",
"recent_events": (recent_events or []),
"story_prompt": prompt,
}
try:
data = await get_prompt_and_call_llm_async(template_path, infos, mode="fast")
@@ -126,17 +132,16 @@ class StoryTeller:
@staticmethod
def tell_from_actors(event: str, res: str, *actors: "Avatar", prompt: str | None = None) -> str:
"""
便捷方法：直接从参与者对象生成 avatar_infos 并讲述故事。
便捷方法别名，保持向后兼容。直接调用 tell_story。
"""
avatar_infos = StoryTeller.build_avatar_infos(*actors)
recent_events = StoryTeller._collect_recent_events(*actors)
return StoryTeller.tell_story(avatar_infos, event, res, prompt or "", recent_events=recent_events)
return StoryTeller.tell_story(event, res, *actors, prompt=prompt or "")
@staticmethod
async def tell_from_actors_async(event: str, res: str, *actors: "Avatar", prompt: str | None = None) -> str:
avatar_infos = StoryTeller.build_avatar_infos(*actors)
recent_events = StoryTeller._collect_recent_events(*actors)
return await StoryTeller.tell_story_async(avatar_infos, event, res, prompt or "", recent_events=recent_events)
"""
便捷方法别名，保持向后兼容。直接调用 tell_story_async。
"""
return await StoryTeller.tell_story_async(event, res, *actors, prompt=prompt or "")
__all__ = ["StoryTeller"]