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

update prompt

Browse Source
This commit is contained in:
bridge
2025-09-07 20:12:04 +08:00
parent 4947284f29
commit d83379e3a5
8 changed files with 115 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

73
src/classes/ai.py
View File

@@ -22,9 +22,23 @@ class AI(ABC):
"""
AI的基类
"""
pass
class SingleAI(AI):
"""
单个角色的AI
"""
def __init__(self, avatar: Avatar):
self.avatar = avatar
@abstractmethod
async def _decide(self, world: World) -> tuple[ACTION_NAME, ACTION_PARAMS, str]:
"""
决策逻辑：决定执行什么动作和参数
由子类实现具体的决策逻辑
"""
pass
async def decide(self, world: World) -> tuple[ACTION_NAME, ACTION_PARAMS, str, Event]:
"""
决定做什么，同时生成对应的事件
@@ -38,15 +52,37 @@ class AI(ABC):
return action_name, action_params, avatar_thinking, event
@abstractmethod
async def _decide(self, world: World) -> tuple[ACTION_NAME, ACTION_PARAMS, str]:
class GroupAI(AI):
"""
多个角色的AI
"""
def __init__(self):
self.avatars = []
async def _decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, tuple[ACTION_NAME, ACTION_PARAMS, str]]:
"""
决策逻辑：决定执行什么动作和参数
由子类实现具体的决策逻辑
"""
pass
class RuleAI(AI):
async def decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, tuple[ACTION_NAME, ACTION_PARAMS, str, Event]]:
"""
决定做什么，同时生成对应的事件
"""
# 先决定动作和参数
results = await self._decide(world, avatars_to_decide)
for avatar, result in results.items():
action_name, action_params, avatar_thinking = result
action = avatar.create_action(action_name)
event = action.get_event(**action_params)
results[avatar] = (action_name, action_params, avatar_thinking, event)
# 获取动作对象并生成事件
return results
class RuleAI(SingleAI):
"""
规则AI
"""
@@ -80,7 +116,7 @@ class RuleAI(AI):
region_with_best_essence = max(regions, key=lambda region: region.essence.get_density(essence_type))
return region_with_best_essence
class LLMAI(AI):
class LLMAI(GroupAI):
"""
LLM AI
一些思考：
@@ -90,20 +126,23 @@ class LLMAI(AI):
不能每个单步step都调用一次LLM来决定下一步做什么。这样子一方面动作一直乱变，另一方面也太费token了。
decide的作用是，拉取既有的动作链（如果没有了就call_llm），再根据动作链决定动作，以及动作之间的衔接。
"""
async def _decide(self, world: World) -> tuple[ACTION_NAME, ACTION_PARAMS, str]:
async def _decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, tuple[ACTION_NAME, ACTION_PARAMS, str]]:
"""
异步决策逻辑：通过LLM决定执行什么动作和参数
"""
action_space_str = self.avatar.get_action_space_str()
avatar_infos_str = str(self.avatar)
regions_str = "\n".join([str(region) for region in world.map.regions.values()])
avatar_persona = self.avatar.persona.prompt
dict_info = {
"action_space": action_space_str,
"avatar_infos": avatar_infos_str,
"regions": regions_str,
"avatar_persona": avatar_persona
global_info = world.get_prompt()
avatar_infos = {avatar.id: avatar.get_prompt() for avatar in avatars_to_decide}
info = {
"avatar_infos": avatar_infos,
"global_info": global_info
}
res = await get_ai_prompt_and_call_llm_async(dict_info)
action_name, action_params, avatar_thinking = res["action_name"], res["action_params"], res["avatar_thinking"]
return action_name, action_params, avatar_thinking
res = await get_ai_prompt_and_call_llm_async(info)
results = {}
for avatar in avatars_to_decide:
action_name, action_params, avatar_thinking = res[avatar.id]["action_name"], res[avatar.id]["action_params"], res[avatar.id]["avatar_thinking"]
results[avatar] = (action_name, action_params, avatar_thinking)
return results
llm_ai = LLMAI()
# rule_ai = RuleAI()
rule_ai = None

41
src/classes/avatar.py
View File

@@ -14,7 +14,6 @@ from src.classes.age import Age
from src.classes.event import NULL_EVENT
from src.classes.typings import ACTION_NAME, ACTION_PARAMS, ACTION_PAIR
from src.classes.ai import AI, RuleAI, LLMAI
from src.classes.persona import Persona, personas_by_id
from src.utils.id_generator import get_avatar_id
from src.utils.config import CONFIG
@@ -53,28 +52,29 @@ class Avatar:
root: Root = field(default_factory=lambda: random.choice(list(Root)))
persona: Persona = field(default_factory=lambda: random.choice(list(personas_by_id.values())))
ai: AI = None
cur_action_pair: Optional[ACTION_PAIR] = None
history_action_pairs: list[ACTION_PAIR] = field(default_factory=list)
thinking: str = ""
def __post_init__(self):
"""
在Avatar创建后自动初始化tile和AI
在Avatar创建后自动初始化tile
"""
self.tile = self.world.map.get_tile(self.pos_x, self.pos_y)
if CONFIG.ai.mode == "llm":
self.ai = LLMAI(self)
else:
self.ai = RuleAI(self)
def __str__(self) -> str:
def __hash__(self) -> int:
return hash(self.id)
def get_info(self) -> str:
"""
获取avatar的详细信息
尽量多打一些，因为会用来给LLM进行决策
"""
return f"Avatar(id={self.id}, 性别={self.gender}, 年龄={self.age}, name={self.name}, 区域={self.tile.region.name}, 灵根={self.root.value}, 境界={self.cultivation_progress})"
def __str__(self) -> str:
return self.get_info()
def create_action(self, action_name: ACTION_NAME) -> Action:
"""
根据动作名称创建新的action实例
@@ -95,21 +95,17 @@ class Avatar:
raise ValueError(f"未找到名为 '{action_name}' 的动作类")
def load_decide_result(self, action_name: ACTION_NAME, action_args: ACTION_PARAMS, avatar_thinking: str):
action = self.create_action(action_name)
self.thinking = avatar_thinking
self.cur_action_pair = (action, action_args)
async def act(self):
"""
角色执行动作。
实际上分为两步：决定做什么（decide）和实际去做（do）
注意这里只负责执行，不负责决定做什么动作。
事件只在决定动作时产生，执行过程不产生事件
"""
event = NULL_EVENT
if self.cur_action_pair is None:
# 决定动作时生成事件
action_name, action_args, avatar_thinking, event = await self.ai.decide(self.world)
action = self.create_action(action_name)
self.thinking = avatar_thinking
self.cur_action_pair = (action, action_args)
# 纯粹执行动作，不产生事件
action, action_params = self.cur_action_pair
@@ -119,7 +115,7 @@ class Avatar:
# 将完成的动作对添加到历史记录中
self._add_to_history(self.cur_action_pair)
return event
return
def _add_to_history(self, action_pair: ACTION_PAIR) -> None:
"""
@@ -208,6 +204,15 @@ class Avatar:
action_space = [{"action": action.__class__.__name__, "params": action.PARAMS, "comment": action.COMMENT} for action in doable_actions]
return action_space
def get_prompt(self) -> str:
"""
获取角色提示词
"""
info = self.get_info()
persona = self.persona.prompt
action_space = self.get_action_space_str()
return f"{info}\n其个性为：{persona}\n决策时需参考这个角色的个性。\n该角色的动作空间及其参数为：{action_space}"
def get_new_avatar_from_ordinary(world: World, current_month_stamp: MonthStamp, name: str, age: Age):
"""
从凡人中来的新修士

6
src/classes/world.py
View File

@@ -6,4 +6,8 @@ from src.classes.calendar import Year, Month, MonthStamp
@dataclass
class World():
map: Map
month_stamp: MonthStamp
month_stamp: MonthStamp
def get_prompt(self) -> str:
regions_str = "\n".join([str(region) for region in self.map.regions.values()])
return f"世界地图上存在的区域为：{regions_str}"

19
src/sim/simulator.py
View File

@@ -5,6 +5,7 @@ from src.classes.avatar import Avatar, get_new_avatar_from_ordinary, Gender
from src.classes.age import Age
from src.classes.world import World
from src.classes.event import Event, is_null_event
from src.classes.ai import llm_ai, rule_ai
from src.utils.names import get_random_name
from src.utils.config import CONFIG
@@ -25,11 +26,23 @@ class Simulator:
events = [] # list of Event
death_avatar_ids = [] # list of str
# 决定动作行为
avatars_to_decide = [avatar for avatar in list(self.avatars.values()) if avatar.cur_action_pair is None]
if CONFIG.ai.mode == "llm":
ai = llm_ai
else:
ai = rule_ai
if avatars_to_decide:
decide_results = await ai.decide(self.world, avatars_to_decide)
for avatar, result in decide_results.items():
action_name, action_args, avatar_thinking, event = result
avatar.load_decide_result(action_name, action_args, avatar_thinking)
if not is_null_event(event):
events.append(event)
# 结算角色行为
for avatar_id, avatar in self.avatars.items():
event = await avatar.act()
if not is_null_event(event):
events.append(event)
await avatar.act()
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()}岁")

3
src/utils/llm.py
View File

@@ -80,8 +80,9 @@ async def get_prompt_and_call_llm_async(template_path: Path, infos: dict) -> str
template = read_txt(template_path)
prompt = get_prompt(template, infos)
res = await call_llm_async(prompt)
# print(f"res = {res}")
json_res = parse_llm_response(res)
# print(f"prompt = {prompt}")
# print(f"json_res = {json_res}")
return json_res
def get_ai_prompt_and_call_llm(infos: dict) -> dict: