- 扩展 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>
90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
"""行动点系统 - 管理用户行动点的消耗与恢复"""
|
|
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
|
|
)
|