make action chain

This commit is contained in:
bridge
2025-09-13 22:45:46 +08:00
parent 50da151334
commit 244e428d81
9 changed files with 200 additions and 59 deletions

View File

@@ -65,6 +65,15 @@ class Action(ABC):
def execute(self) -> None:
pass
@property
def name(self) -> str:
"""
获取动作名称
"""
return str(self.__class__.__name__)
class DefineAction(Action):
def __init__(self, avatar: Avatar, world: World):
"""
@@ -173,6 +182,7 @@ class MoveToRegion(DefineAction, ActualActionMixin):
移动到某个region
"""
COMMENT = "移动到某个区域"
DOABLES_REQUIREMENTS = "任何时候都可以执行"
PARAMS = {"region": "region_name"}
def _execute(self, region: Region|str) -> None:
"""
@@ -228,6 +238,7 @@ class Cultivate(DefineAction, ActualActionMixin):
修炼动作,可以增加修仙进度。
"""
COMMENT = "修炼,增进修为"
DOABLES_REQUIREMENTS = "在修炼区域中,角色不可以突破"
PARAMS = {}
def _execute(self) -> None:
"""
@@ -272,6 +283,7 @@ class Breakthrough(DefineAction, ActualActionMixin):
突破境界
"""
COMMENT = "尝试突破境界"
DOABLES_REQUIREMENTS = "角色可以突破时"
PARAMS = {}
def calc_success_rate(self) -> float:
"""
@@ -307,6 +319,7 @@ class Play(DefineAction, ActualActionMixin):
游戏娱乐动作,持续半年时间
"""
COMMENT = "游戏娱乐,放松身心"
DOABLES_REQUIREMENTS = "任何时候都可以执行"
PARAMS = {}
def _execute(self) -> None:
@@ -335,6 +348,7 @@ class Hunt(DefineAction, ActualActionMixin):
可以获得动物对应的物品
"""
COMMENT = "在当前区域狩猎动物,获取动物材料"
DOABLES_REQUIREMENTS = "在有动物的普通区域且avatar的境界必须大于等于动物的境界"
PARAMS = {}
def _execute(self) -> None:
@@ -390,6 +404,7 @@ class Harvest(DefineAction, ActualActionMixin):
可以获得植物对应的物品
"""
COMMENT = "在当前区域采集植物,获取植物材料"
DOABLES_REQUIREMENTS = "在有植物的普通区域且avatar的境界必须大于等于植物的境界"
PARAMS = {}
def _execute(self) -> None:
@@ -445,6 +460,7 @@ class Sold(DefineAction, ActualActionMixin):
收益为 item_price * item_num动作耗时1个月。
"""
COMMENT = "在城镇出售持有的某类物品的全部"
DOABLES_REQUIREMENTS = "在城镇且背包非空"
PARAMS = {"item_name": "str"}
def _execute(self, item_name: str) -> None:
@@ -486,4 +502,13 @@ class Sold(DefineAction, ActualActionMixin):
ALL_ACTION_CLASSES = [Move, Cultivate, Breakthrough, MoveToRegion, Play, Hunt, Harvest, Sold]
ALL_ACTUAL_ACTION_CLASSES = [Cultivate, Breakthrough, MoveToRegion, Play, Hunt, Harvest, Sold]
ALL_ACTION_NAMES = ["Move", "Cultivate", "Breakthrough", "MoveToRegion", "Play", "Hunt", "Harvest", "Sold"]
ALL_ACTUAL_ACTION_NAMES = ["Cultivate", "Breakthrough", "MoveToRegion", "Play", "Hunt", "Harvest", "Sold"]
ALL_ACTUAL_ACTION_NAMES = ["Cultivate", "Breakthrough", "MoveToRegion", "Play", "Hunt", "Harvest", "Sold"]
ACTION_INFOS = {
action.__name__: {
"comment": action.COMMENT,
"doable_requirements": action.DOABLES_REQUIREMENTS,
"params": action.PARAMS,
} for action in ALL_ACTUAL_ACTION_CLASSES
}
ACTION_INFOS_STR = json.dumps(ACTION_INFOS, ensure_ascii=False)

View File

@@ -13,8 +13,9 @@ from src.classes.region import Region
from src.classes.root import corres_essence_type
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
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
if TYPE_CHECKING:
from src.classes.avatar import Avatar
@@ -27,10 +28,10 @@ class AI(ABC):
"""
@abstractmethod
async def _decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, tuple[ACTION_NAME, ACTION_PARAMS, str]]:
async def _decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, tuple]:
pass
async def decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, tuple[ACTION_NAME, ACTION_PARAMS, str, Event]]:
async def decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, tuple[ACTION_NAME_PARAMS_PAIRS, str, str, Event]]:
"""
决定做什么,同时生成对应的事件。
一个ai支持批量生成多个avatar的动作。
@@ -42,10 +43,17 @@ class AI(ABC):
results.update(await self._decide(world, avatars_to_decide[i:i+max_decide_num]))
for avatar, result in list(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)
# 兼容RuleAI 返回单动作LLMAI 返回动作链
if result and isinstance(result[0], list):
action_name_params_pairs, avatar_thinking, objective = result # type: ignore
else:
action_name, action_params, avatar_thinking, objective = result # type: ignore
action_name_params_pairs = [(action_name, action_params)]
# 只为队列中的第一个动作生成事件
first_action_name, first_action_params = action_name_params_pairs[0]
action = avatar.create_action(first_action_name)
event = action.get_event(**first_action_params)
results[avatar] = (action_name_params_pairs, avatar_thinking, objective, event)
return results
@@ -54,7 +62,7 @@ class RuleAI(AI):
规则AI批量接口内部逐个决策
"""
def __decide(self, world: World, avatar: "Avatar", regions: list[Region]) -> tuple[ACTION_NAME, ACTION_PARAMS, str]:
def __decide(self, world: World, avatar: "Avatar", regions: list[Region]) -> tuple[ACTION_NAME, ACTION_PARAMS, str, str]:
"""
单个 Avatar 的决策逻辑。
先做一个简单的:
@@ -65,24 +73,24 @@ class RuleAI(AI):
5. 如果需要突破境界了,则突破境界
"""
if random.random() < 0.1:
return ("Play", {}, "")
return ("Play", {}, "", "放松一下,缓解修行压力")
best_region = self.get_best_region_for_avatar(avatar, regions)
if avatar.is_in_region(best_region):
if avatar.cultivation_progress.can_break_through():
return ("Breakthrough", {}, "")
return ("Breakthrough", {}, "", "尽快突破到更高境界")
else:
return ("Cultivate", {}, "")
return ("Cultivate", {}, "", "稳步提升修为")
else:
return ("MoveToRegion", {"region": best_region.name}, "")
return ("MoveToRegion", {"region": best_region.name}, "", f"前往{best_region.name}修行")
async def _decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, 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, str]]:
"""
决策逻辑:批量接口的实现上,逐个 Avatar 调用 __decide 进行独立决策,
以保持规则AI的可控性与可测试性。
"""
results: dict[Avatar, tuple[ACTION_NAME, ACTION_PARAMS, str]] = {}
results: dict[Avatar, tuple[ACTION_NAME, ACTION_PARAMS, str, str]] = {}
regions: list[Region] = list(world.map.regions.values())
for avatar in avatars_to_decide:
@@ -109,23 +117,40 @@ class LLMAI(AI):
2. 突发应对动作比如突然有人要攻击NPC这个时候的反应
"""
async def _decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, tuple[ACTION_NAME, ACTION_PARAMS, str]]:
async def _decide(self, world: World, avatars_to_decide: list[Avatar]) -> dict[Avatar, tuple[ACTION_NAME_PARAMS_PAIRS, str, str]]:
"""
异步决策逻辑通过LLM决定执行什么动作和参数
"""
global_info = world.get_info()
avatar_infos = {avatar.name: avatar.get_prompt() for avatar in avatars_to_decide}
avatar_infos = {avatar.name: avatar.get_prompt_info() for avatar in avatars_to_decide}
general_action_infos = ACTION_INFOS_STR
info = {
"avatar_infos": avatar_infos,
"global_info": global_info,
"general_action_infos": general_action_infos,
}
res = await get_ai_prompt_and_call_llm_async(info)
results: dict[Avatar, tuple[ACTION_NAME, ACTION_PARAMS, str]] = {}
results: dict[Avatar, tuple[ACTION_NAME_PARAMS_PAIRS, str, str]] = {}
for avatar in avatars_to_decide:
action_name = res[avatar.name]["action_name"]
action_params = res[avatar.name]["action_params"]
avatar_thinking = res[avatar.name]["avatar_thinking"]
results[avatar] = (action_name, action_params, avatar_thinking)
r = res[avatar.name]
# 仅接受 action_name_params_pairs不再支持单个 action_name/action_params
raw_pairs = r["action_name_params_pairs"]
pairs: ACTION_NAME_PARAMS_PAIRS = []
for p in raw_pairs:
if isinstance(p, list) and len(p) == 2:
pairs.append((p[0], p[1]))
elif isinstance(p, dict) and "action_name" in p and "action_params" in p:
pairs.append((p["action_name"], p["action_params"]))
else:
# 跳过无法解析的项
continue
# 至少有一个
if not pairs:
raise ValueError(f"LLM未返回有效的action_name_params_pairs: {r}")
avatar_thinking = r.get("avatar_thinking", r.get("thinking", ""))
objective = r.get("objective", "")
results[avatar] = (pairs, avatar_thinking, objective)
return results
llm_ai = LLMAI()

View File

@@ -12,8 +12,8 @@ from src.classes.region import Region
from src.classes.cultivation import CultivationProgress
from src.classes.root import Root
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.event import NULL_EVENT, Event
from src.classes.typings import ACTION_NAME, ACTION_PARAMS, ACTION_PAIR, ACTION_NAME_PARAMS_PAIRS, ACTION_NAME_PARAMS_PAIR
from src.classes.persona import Persona, personas_by_id
from src.classes.item import Item
@@ -57,7 +57,9 @@ class Avatar:
persona: Persona = field(default_factory=lambda: random.choice(list(personas_by_id.values())))
cur_action_pair: Optional[ACTION_PAIR] = None
history_action_pairs: list[ACTION_PAIR] = field(default_factory=list)
next_actions: ACTION_NAME_PARAMS_PAIRS = field(default_factory=list)
thinking: str = ""
objective: str = ""
magic_stone: MagicStone = field(default_factory=lambda: MagicStone(0)) # 灵石,即货币
items: dict[Item, int] = field(default_factory=dict)
@@ -100,10 +102,61 @@ 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)
def load_decide_result_chain(self, action_name_params_pairs: ACTION_NAME_PARAMS_PAIRS, avatar_thinking: str, objective: str):
"""
加载AI的决策结果动作链立即设置第一个为当前动作其余进入队列。
"""
if not action_name_params_pairs:
return
first_action_name, first_action_params = action_name_params_pairs[0]
action = self.create_action(first_action_name)
self.thinking = avatar_thinking
self.cur_action_pair = (action, action_args)
self.objective = objective
self.cur_action_pair = (action, first_action_params)
# 余下的动作进入队列
if len(action_name_params_pairs) > 1:
self.next_actions.extend(action_name_params_pairs[1:])
def has_next_actions(self) -> bool:
return len(self.next_actions) > 0
def pop_next_action_and_set_current(self) -> Optional[Event]:
"""
从队列中取出下一个动作并设置为当前动作,同时返回开始事件。
若队列为空则返回None。
"""
if not self.next_actions:
return None
action_name, action_params = self.next_actions.pop(0)
action = self.create_action(action_name)
self.cur_action_pair = (action, action_params)
try:
event = action.get_event(**action_params)
except TypeError:
# 兼容无参数的 get_event 定义
event = action.get_event()
return event
def peek_next_action(self) -> Optional[ACTION_NAME_PARAMS_PAIR]:
"""
查看下一个动作但不弹出。
"""
if not self.next_actions:
return None
return self.next_actions[0]
def is_next_action_doable(self) -> bool:
"""
判断队列中的下一个动作当前是否可执行。
若没有下一个动作返回False。
"""
pair = self.peek_next_action()
if pair is None:
return False
action_name, _ = pair
action = self.create_action(action_name)
# 动作的 is_doable 定义为 @property
return bool(getattr(action, "is_doable", True))
async def act(self):
"""
@@ -262,7 +315,7 @@ class Avatar:
"""
获取历史动作对的字符串
"""
return "\n".join([f"{action.__class__.__name__}: {action_params}" for action, action_params in self.history_action_pairs])
return "\n".join([f"{action.name}: {action_params}" for action, action_params in self.history_action_pairs])
def get_action_space_str(self) -> str:
action_space = self.get_action_space()
@@ -275,12 +328,12 @@ class Avatar:
"""
actual_actions = [self.create_action(action_cls_name) for action_cls_name in ALL_ACTUAL_ACTION_NAMES]
doable_actions = [action for action in actual_actions if action.is_doable]
action_space = [{"action": action.__class__.__name__, "params": action.PARAMS, "comment": action.COMMENT} for action in doable_actions]
action_space = [action.name for action in doable_actions]
return action_space
def get_prompt(self) -> str:
def get_prompt_info(self) -> str:
"""
获取角色提示词
获取角色提示词信息
"""
info = self.get_info()
persona = self.persona.prompt
@@ -295,14 +348,14 @@ class Avatar:
else:
items_info = "物品持有情况:无"
return f"{info}\n其个性为:{persona}\n{magic_stone_info}\n{items_info}\n决策时需参考这个角色的个性。\n该角色的动作空间及其参数为:{action_space}"
return f"{info}\n其个性为:{persona}\n{magic_stone_info}\n{items_info}\n决策时需参考这个角色的个性。\n该角色的目前暂时的合法动作为:{action_space}"
@property
def move_step_length(self) -> int:
"""
获取角色的移动步长
"""
return int(self.cultivation_progress.realm.value)
return self.cultivation_progress.get_month_step()
def get_new_avatar_from_ordinary(world: World, current_month_stamp: MonthStamp, name: str, age: Age):
"""

View File

@@ -11,30 +11,30 @@ class Stage(Enum):
Middle_Stage = "中期"
Late_Stage = "后期"
levels_per_realm = 30
levels_per_stage = 10
LEVELS_PER_REALM = 30
LEVELS_PER_STAGE = 10
level_to_realm = {
LEVEL_TO_REALM = {
0: Realm.Qi_Refinement,
30: Realm.Foundation_Establishment,
60: Realm.Core_Formation,
90: Realm.Nascent_Soul,
}
level_to_stage = {
LEVEL_TO_STAGE = {
0: Stage.Early_Stage,
10: Stage.Middle_Stage,
20: Stage.Late_Stage,
}
# realm_id到Realm的映射用于物品等级系统
realm_id_to_realm = {
REALM_ID_TO_REALM = {
1: Realm.Qi_Refinement,
2: Realm.Foundation_Establishment,
3: Realm.Core_Formation,
4: Realm.Nascent_Soul,
}
level_to_break_through = {
LEVEL_TO_BREAK_THROUGH = {
30: Realm.Foundation_Establishment,
60: Realm.Core_Formation,
90: Realm.Nascent_Soul,
@@ -60,21 +60,31 @@ class CultivationProgress:
def get_realm(self, level: int) -> str:
"""获取境界"""
for level_threshold, realm in reversed(list(level_to_realm.items())):
for level_threshold, realm in reversed(list(LEVEL_TO_REALM.items())):
if level >= level_threshold:
return realm
return Realm.Qi_Refinement
def get_stage(self, level: int) -> Stage:
"""获取阶段"""
_level = level % levels_per_realm
for level_threshold, stage in reversed(list(level_to_stage.items())):
_level = level % LEVELS_PER_REALM
for level_threshold, stage in reversed(list(LEVEL_TO_STAGE.items())):
if _level >= level_threshold:
return stage
return Stage.Early_Stage
def get_month_step(self) -> int:
"""
每月能够移动的距离,
练气筑基为1
金丹元婴为2
"""
return int(self.level // LEVELS_PER_REALM * 2) + 1
def __str__(self) -> str:
return f"{self.realm.value}{self.stage.value}({self.level}级)"
can_break_through = self.can_break_through()
can_break_through_str = "可以突破" if can_break_through else "不可以突破"
return f"{self.realm.value}{self.stage.value}({self.level}级){can_break_through_str}"
def get_exp_required(self) -> int:
"""
@@ -146,7 +156,7 @@ class CultivationProgress:
"""
检查是否可以突破
"""
return self.level in level_to_break_through.keys()
return self.level in LEVEL_TO_BREAK_THROUGH.keys()
def can_cultivate(self) -> bool:
"""
@@ -180,15 +190,15 @@ def _realm_from_id(cls, realm_id: int) -> Realm:
Raises:
ValueError: 如果realm_id不存在
"""
if realm_id not in realm_id_to_realm:
if realm_id not in REALM_ID_TO_REALM:
raise ValueError(f"Unknown realm_id: {realm_id}")
return realm_id_to_realm[realm_id]
return REALM_ID_TO_REALM[realm_id]
# 将from_id方法绑定到Realm类
Realm.from_id = classmethod(_realm_from_id)
# 境界顺序映射
_realm_order = {
_REALM_ORDER = {
Realm.Qi_Refinement: 1,
Realm.Foundation_Establishment: 2,
Realm.Core_Formation: 3,
@@ -200,25 +210,25 @@ def _realm_ge(self, other):
"""大于等于比较"""
if not isinstance(other, Realm):
return NotImplemented
return _realm_order[self] >= _realm_order[other]
return _REALM_ORDER[self] >= _REALM_ORDER[other]
def _realm_le(self, other):
"""小于等于比较"""
if not isinstance(other, Realm):
return NotImplemented
return _realm_order[self] <= _realm_order[other]
return _REALM_ORDER[self] <= _REALM_ORDER[other]
def _realm_gt(self, other):
"""大于比较"""
if not isinstance(other, Realm):
return NotImplemented
return _realm_order[self] > _realm_order[other]
return _REALM_ORDER[self] > _REALM_ORDER[other]
def _realm_lt(self, other):
"""小于比较"""
if not isinstance(other, Realm):
return NotImplemented
return _realm_order[self] < _realm_order[other]
return _REALM_ORDER[self] < _REALM_ORDER[other]
# 将比较方法绑定到Realm类
Realm.__ge__ = _realm_ge

View File

@@ -2,4 +2,6 @@ from src.classes.action import Action
ACTION_NAME = str
ACTION_PARAMS = dict
ACTION_PAIR = tuple[Action, ACTION_PARAMS]
ACTION_PAIR = tuple[Action, ACTION_PARAMS]
ACTION_NAME_PARAMS_PAIR = tuple[ACTION_NAME, ACTION_PARAMS]
ACTION_NAME_PARAMS_PAIRS = list[ACTION_NAME_PARAMS_PAIR]

View File

@@ -143,6 +143,11 @@ def draw_tooltip_for_avatar(pygame_mod, screen, colors, font, avatar: Avatar):
lines.append("思考:")
thinking_lines = wrap_text(avatar.thinking, 20)
lines.extend(thinking_lines)
if getattr(avatar, "objective", None):
lines.append("")
lines.append("目标:")
objective_lines = wrap_text(avatar.objective, 20)
lines.extend(objective_lines)
draw_tooltip(pygame_mod, screen, colors, lines, *pygame_mod.mouse.get_pos(), font)

View File

@@ -27,7 +27,20 @@ class Simulator:
death_avatar_ids = [] # list of str
# 决定动作行为
avatars_to_decide = [avatar for avatar in list(self.avatars.values()) if avatar.cur_action_pair is None]
avatars_to_decide = []
for avatar in list(self.avatars.values()):
if avatar.cur_action_pair is None:
# 若有排队动作但当前不可执行:丢弃之后的所有动作
if avatar.has_next_actions():
if not avatar.is_next_action_doable():
avatar.next_actions.clear()
avatars_to_decide.append(avatar)
else:
event = avatar.pop_next_action_and_set_current()
if event is not None and not is_null_event(event):
events.append(event)
else:
avatars_to_decide.append(avatar)
if CONFIG.ai.mode == "llm":
ai = llm_ai
else:
@@ -35,8 +48,8 @@ class Simulator:
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)
action_name_params_pairs, avatar_thinking, objective, event = result
avatar.load_decide_result_chain(action_name_params_pairs, avatar_thinking, objective)
if not is_null_event(event):
events.append(event)

View File

@@ -9,3 +9,4 @@ id,name,prompt
7,采药,你是一个热爱采集的人,喜欢在山林中寻找各种奇花异草和灵药,对植物有着敏锐的直觉和深厚的兴趣。你认为大自然的恩赐需要用心去发现和珍惜。
8,猎者,你是一个热爱狩猎的人,享受在野外追踪猎物的刺激感,对各种动物的习性了如指掌。你相信通过狩猎能够磨练自己的意志和技能,获得更强大的力量。
9,爱财,你嗜财如命,对灵石和财富有着强烈的渴望。
10,沉思,你是一个沉思的人,你总是会深思熟虑,思考问题比较有哲理。
1 id name prompt
9 7 采药 你是一个热爱采集的人,喜欢在山林中寻找各种奇花异草和灵药,对植物有着敏锐的直觉和深厚的兴趣。你认为大自然的恩赐需要用心去发现和珍惜。
10 8 猎者 你是一个热爱狩猎的人,享受在野外追踪猎物的刺激感,对各种动物的习性了如指掌。你相信通过狩猎能够磨练自己的意志和技能,获得更强大的力量。
11 9 爱财 你嗜财如命,对灵石和财富有着强烈的渴望。
12 10 沉思 你是一个沉思的人,你总是会深思熟虑,思考问题比较有哲理。

View File

@@ -2,14 +2,21 @@
{global_info}
你需要进行决策的NPC的dict[AvatarName, info]为
{avatar_infos}
通用的动作说明为:
{general_action_infos}
注意只返回json格式的结果。
分Avatar进行返回格式为
{{
AvatarName: {{
"thinking": ..., // 简单思考应该怎么决策
"action_name": ...,
"action_params": ...,
"avatar_thinking": ..., // 从角色角度,以第一人称视角,描述心态,符合世界观
"objective": ..., // 角色接下来一段时间的目标
// 基于objective一次性决定未来的3~8个动作按顺序执行
"action_name_params_pairs": list[Tuple[action_name, action_params]],
"avatar_thinking": ... // 从角色角度以第一人称视角基于action_name_params_pairs描述想法
}}
}}
}}
要求与约束:
- 若需要先移动再修炼,请将 "MoveToRegion" 放在前面,随后接 "Cultivate"。
- 若当前可突破,可在合适时机插入 "Breakthrough"。