From 5ae63d9df98ad78a8bfb165ea4962f34c85218f5 Mon Sep 17 00:00:00 2001 From: empty Date: Tue, 30 Dec 2025 13:48:33 +0800 Subject: [PATCH] =?UTF-8?q?feat(engine):=20=E5=AE=9E=E7=8E=B0=E8=A1=8C?= =?UTF-8?q?=E5=8A=A8=E7=82=B9=E7=B3=BB=E7=BB=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 扩展 AgentState 添加 action_points, max_action_points, last_action_tick 字段 - 新增 ActionFeedback 模型用于返回行动执行结果 - 创建 action_points.py 模块实现行动点消耗与恢复逻辑 - 行动消耗表: vote=1, trigger_skill=2, influence=2, comment/support/chaos=0 - 每 tick 恢复 1 点行动点（不超过 max） - 行动点不足时拒绝执行并返回失败反馈 - 新增 7 个测试用例，全部 37 个测试通过 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.5 --- engine-python/app/action_points.py | 89 +++++++++++++++++++ engine-python/app/engine.py | 26 +++++- engine-python/app/main.py | 7 +- engine-python/app/models.py | 13 +++ engine-python/app/state.py | 6 ++ engine-python/tests/test_step.py | 133 +++++++++++++++++++++++++++++ 6 files changed, 267 insertions(+), 7 deletions(-) create mode 100644 engine-python/app/action_points.py diff --git a/engine-python/app/action_points.py b/engine-python/app/action_points.py new file mode 100644 index 0000000..8d8adbe --- /dev/null +++ b/engine-python/app/action_points.py @@ -0,0 +1,89 @@ +"""行动点系统 - 管理用户行动点的消耗与恢复""" +from typing import List, Tuple, Dict +from .models import WorldState, Event, ActionFeedback + +# 行动消耗表 +ACTION_COST: Dict[str, int] = { + "vote": 1, + "trigger_skill": 2, + "influence": 2, + "comment": 0, + "support": 0, + "chaos": 0, +} +DEFAULT_COST = 0 + + +def get_action_cost(event_type: str) -> int: + """获取行动消耗的点数""" + return ACTION_COST.get(event_type, DEFAULT_COST) + + +def check_action_points(state: WorldState, user: str, cost: int) -> bool: + """检查用户是否有足够的行动点""" + if user not in state.agents: + return True # 非 agent 用户不受限制 + return state.agents[user].action_points >= cost + + +def consume_action_points( + state: WorldState, user: str, cost: int +) -> None: + """消耗行动点""" + if user not in state.agents: + return + agent = state.agents[user] + agent.action_points = max(0, agent.action_points - cost) + agent.last_action_tick = state.tick + + +def regenerate_action_points(state: WorldState) -> None: + """每 tick 恢复行动点""" + for agent_id, agent in state.agents.items(): + if state.tick - agent.last_action_tick >= 1: + if agent.action_points < agent.max_action_points: + agent.action_points = min( + agent.max_action_points, + agent.action_points + 1 + ) + + +def process_event_with_ap( + state: WorldState, event: Event +) -> Tuple[bool, ActionFeedback]: + """处理单个事件的行动点检查 + + 返回: (是否允许执行, 反馈信息) + """ + user = event.user + cost = get_action_cost(event.type) + + # 0 消耗的行动不需要检查 + if cost == 0: + return True, ActionFeedback( + success=True, + reason="action applied", + remaining_ap=state.agents[user].action_points if user in state.agents else 0, + user=user + ) + + # 检查行动点 + if not check_action_points(state, user, cost): + remaining = state.agents[user].action_points if user in state.agents else 0 + return False, ActionFeedback( + success=False, + reason=f"insufficient action points (need {cost}, have {remaining})", + remaining_ap=remaining, + user=user + ) + + # 消耗行动点 + consume_action_points(state, user, cost) + remaining = state.agents[user].action_points if user in state.agents else 0 + + return True, ActionFeedback( + success=True, + reason="action applied", + remaining_ap=remaining, + user=user + ) diff --git a/engine-python/app/engine.py b/engine-python/app/engine.py index e596a82..551d52d 100644 --- a/engine-python/app/engine.py +++ b/engine-python/app/engine.py @@ -2,7 +2,7 @@ from __future__ import annotations from typing import List, Optional, Tuple from .models import ( WorldState, AgentState, Event, Action, Emotion, Weather, - GlobalEventResult, GlobalEventInfo, FactionEventResult + GlobalEventResult, GlobalEventInfo, FactionEventResult, ActionFeedback ) from .global_events import GLOBAL_EVENT_POOL, GlobalEvent from .opinions import generate_opinions @@ -12,6 +12,9 @@ from .factions import ( ) from .voting import process_votes, apply_votes_to_factions from .faction_skills import check_and_unlock_skills, apply_skill_effects +from .action_points import ( + regenerate_action_points, process_event_with_ap, get_action_cost +) MAX_EVENTS = 20 MAX_MEMORY_PER_AGENT = 3 @@ -40,11 +43,16 @@ def update_world_effects(state: WorldState) -> None: state.world_effects = active_effects -def process_events(state: WorldState, events: List[Event]) -> WorldState: - """处理事件并更新世界状态""" +def process_events( + state: WorldState, events: List[Event] +) -> Tuple[WorldState, List[ActionFeedback]]: + """处理事件并更新世界状态，返回行动反馈列表""" # tick 递增 state.tick += 1 + # ★ 行动点恢复 + regenerate_action_points(state) + # ★ 在 tick 开始时：将上一轮投票累加到阵营能量 apply_votes_to_factions(state) @@ -58,7 +66,17 @@ def process_events(state: WorldState, events: List[Event]) -> WorldState: # ★ 处理投票事件（记录到 votes，下一 tick 生效） process_votes(state, events) + # 收集行动反馈 + action_feedbacks: List[ActionFeedback] = [] + for event in events: + # ★ 行动点检查 + allowed, feedback = process_event_with_ap(state, event) + action_feedbacks.append(feedback) + + if not allowed: + continue # 行动点不足，跳过此事件 + text = event.text # 累计能量 @@ -113,7 +131,7 @@ def process_events(state: WorldState, events: List[Event]) -> WorldState: # 应用已解锁技能效果 apply_skill_effects(state) - return state + return state, action_feedbacks def check_and_trigger_global_event( diff --git a/engine-python/app/main.py b/engine-python/app/main.py index 0316705..2155d80 100644 --- a/engine-python/app/main.py +++ b/engine-python/app/main.py @@ -14,8 +14,8 @@ def step(request: StepRequest) -> StepResponse: # 加载当前状态 state = load_state() - # 处理事件并更新状态 - state = process_events(state, request.events) + # 处理事件并更新状态，获取行动反馈 + state, action_feedbacks = process_events(state, request.events) # 检查并触发世界级事件 state, global_event_result = check_and_trigger_global_event(state) @@ -40,7 +40,8 @@ def step(request: StepRequest) -> StepResponse: actions=actions, global_event=global_event_result, triggered_faction_event=faction_event_result, - story_event=story_event_result + story_event=story_event_result, + action_feedbacks=action_feedbacks ) diff --git a/engine-python/app/models.py b/engine-python/app/models.py index 0f20917..7de637f 100644 --- a/engine-python/app/models.py +++ b/engine-python/app/models.py @@ -108,6 +108,14 @@ class StoryEventResult(BaseModel): description: Optional[str] = None +class ActionFeedback(BaseModel): + """行动反馈""" + success: bool = True + reason: str = "" + remaining_ap: int = 0 + user: str = "" + + class AgentState(BaseModel): emotion: Emotion = Emotion.CALM goal: str = "" @@ -119,6 +127,10 @@ class AgentState(BaseModel): stance: Stance = Field(default_factory=Stance) # 所属阵营 faction: str = "neutral" # "optimists" | "fearful" | "neutral" + # 行动点系统 + action_points: int = Field(default=3, ge=0) + max_action_points: int = Field(default=3, ge=1) + last_action_tick: int = 0 class WorldState(BaseModel): @@ -171,3 +183,4 @@ class StepResponse(BaseModel): global_event: GlobalEventResult = Field(default_factory=GlobalEventResult) triggered_faction_event: FactionEventResult = Field(default_factory=FactionEventResult) story_event: StoryEventResult = Field(default_factory=StoryEventResult) + action_feedbacks: List[ActionFeedback] = Field(default_factory=list) diff --git a/engine-python/app/state.py b/engine-python/app/state.py index 74075c7..c1aa4a6 100644 --- a/engine-python/app/state.py +++ b/engine-python/app/state.py @@ -20,6 +20,9 @@ def get_default_state() -> WorldState: memory=[], stance=Stance(optimism=0.6, fear=0.4), faction="neutral", + action_points=3, + max_action_points=3, + last_action_tick=0, ), "bob": AgentState( emotion="calm", @@ -27,6 +30,9 @@ def get_default_state() -> WorldState: memory=[], stance=Stance(optimism=0.4, fear=0.6), faction="neutral", + action_points=3, + max_action_points=3, + last_action_tick=0, ), }, events=[], diff --git a/engine-python/tests/test_step.py b/engine-python/tests/test_step.py index 5f15ed9..8995da5 100644 --- a/engine-python/tests/test_step.py +++ b/engine-python/tests/test_step.py @@ -541,3 +541,136 @@ def test_skill_requires_prerequisite(client): assert skills["unity"]["requires"] == ["festival_boost"] # 初始状态下 unity 未解锁 assert skills["unity"]["unlocked"] is False + + +# ==================== 行动点系统测试 ==================== + +def test_action_points_exists_in_agent(client): + """测试 agent 包含行动点字段""" + resp = client.post("/step", json={"events": []}) + data = resp.json() + + for agent_id, agent in data["world_state"]["agents"].items(): + assert "action_points" in agent + assert "max_action_points" in agent + assert "last_action_tick" in agent + assert agent["action_points"] >= 0 + assert agent["max_action_points"] >= 1 + + +def test_action_feedbacks_in_response(client): + """测试响应包含 action_feedbacks 字段""" + resp = client.post("/step", json={"events": []}) + data = resp.json() + + assert "action_feedbacks" in data + assert isinstance(data["action_feedbacks"], list) + + +def test_vote_consumes_action_points(client): + """测试投票消耗行动点""" + # alice 投票（消耗 1 点） + events = [ + {"type": "vote", "faction": "optimists", "text": "", "user": "alice", "ts": 1} + ] + resp = client.post("/step", json={"events": events}) + data = resp.json() + + # 验证行动反馈 + assert len(data["action_feedbacks"]) == 1 + feedback = data["action_feedbacks"][0] + assert feedback["success"] is True + assert feedback["user"] == "alice" + # 初始 3 点，消耗 1 点，恢复 1 点 = 3 点 + # 但消耗在恢复之后，所以是 3 - 1 = 2 + assert feedback["remaining_ap"] == 2 + + +def test_insufficient_action_points_rejected(client): + """测试行动点不足时拒绝执行""" + # 重置状态，设置 alice 行动点为 0，last_action_tick 为当前 tick + # 这样在下一个 tick 恢复后只有 1 点 + state = get_default_state() + state.tick = 10 + state.agents["alice"].action_points = 0 + state.agents["alice"].last_action_tick = 10 # 刚行动过，不会恢复 + save_state(state) + + # 尝试投票（需要 1 点），但恢复后有 1 点 + # 为了测试不足，我们需要让 last_action_tick = tick，这样不会恢复 + # 但 regenerate 检查的是 tick - last_action_tick >= 1 + # tick 会先 +1 变成 11，11 - 10 = 1 >= 1，所以会恢复 + + # 重新设计：设置 last_action_tick 为未来的 tick + state = get_default_state() + state.tick = 10 + state.agents["alice"].action_points = 0 + state.agents["alice"].last_action_tick = 11 # 未来 tick，不会恢复 + save_state(state) + + events = [ + {"type": "vote", "faction": "optimists", "text": "", "user": "alice", "ts": 10} + ] + resp = client.post("/step", json={"events": events}) + data = resp.json() + + # 验证被拒绝 + assert len(data["action_feedbacks"]) == 1 + feedback = data["action_feedbacks"][0] + assert feedback["success"] is False + assert "insufficient" in feedback["reason"] + + +def test_action_points_regenerate_per_tick(client): + """测试行动点每 tick 恢复""" + # 设置 alice 行动点为 1 + state = get_default_state() + state.agents["alice"].action_points = 1 + state.agents["alice"].last_action_tick = 0 + save_state(state) + + # 执行一个 tick（无事件） + resp = client.post("/step", json={"events": []}) + data = resp.json() + + # 行动点应该恢复到 2（1 + 1） + assert data["world_state"]["agents"]["alice"]["action_points"] == 2 + + +def test_action_points_not_exceed_max(client): + """测试行动点不超过最大值""" + # 设置 alice 行动点为 3（已满） + state = get_default_state() + state.agents["alice"].action_points = 3 + state.agents["alice"].max_action_points = 3 + save_state(state) + + # 执行多个 tick + for _ in range(3): + client.post("/step", json={"events": []}) + + resp = client.post("/step", json={"events": []}) + data = resp.json() + + # 行动点不应超过 max + assert data["world_state"]["agents"]["alice"]["action_points"] <= 3 + + +def test_zero_cost_action_always_allowed(client): + """测试 0 消耗的行动始终允许""" + # 设置 alice 行动点为 0 + state = get_default_state() + state.agents["alice"].action_points = 0 + save_state(state) + + # comment 类型消耗 0 点 + events = [ + {"type": "comment", "text": "hello", "user": "alice", "ts": 1} + ] + resp = client.post("/step", json={"events": events}) + data = resp.json() + + # 应该成功执行 + assert len(data["action_feedbacks"]) == 1 + feedback = data["action_feedbacks"][0] + assert feedback["success"] is True