feat(engine): 实现行动点系统
- 扩展 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 <noreply@anthropic.com>
This commit is contained in:
89
engine-python/app/action_points.py
Normal file
89
engine-python/app/action_points.py
Normal file
@@ -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
|
||||
)
|
||||
@@ -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(
|
||||
|
||||
@@ -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
|
||||
)
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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=[],
|
||||
|
||||
Reference in New Issue
Block a user