""" Autonomous Agent Decision Core for FreeExile. Translates real-time ServerEngineLoop perception into authoritative game actions: Movement, Skill Execution (with Five Elements counter), Phantom Evasion i-frame, and Ground Loot Collection. """ from enum import Enum from typing import Dict, List, Optional, Any from dataclasses import dataclass, field import math import time from world.server_engine_loop import ServerEngineLoop, ClientMoveCommand, ClientSkillCommand from world.combat_engine import CombatEngine, CombatActor from world.martial_matrix import FiveElements from agent.llm_provider_client import ( AgentActionType, LLMProviderType, LLMProviderConfig, TacticalActionDecision, LLMProviderClientRouter ) class TacticalStance(str, Enum): BALANCED = "BALANCED" # Balance between leveling and survival AGGRESSIVE = "AGGRESSIVE" # Relentless offensive pursuit DEFENSIVE = "DEFENSIVE" # High caution, retreat early TREASURE_HUNTER = "TREASURE_HUNTER" # Prioritize rare spirit stones above all else @dataclass(slots=True, frozen=True) class PerceivedEntity: entity_id: int x: float y: float distance: float hp: float max_hp: float element: FiveElements @dataclass(slots=True, frozen=True) class PerceivedDrop: drop_id: str item_id: str x: float y: float distance: float rarity: int @dataclass(slots=True, frozen=True) class AgentPerceptionSnapshot: player_id: int player_x: float player_y: float player_z: float current_hp: float max_hp: float hp_ratio: float element: FiveElements nearby_entities: List[PerceivedEntity] nearby_drops: List[PerceivedDrop] @dataclass(slots=True, frozen=True) class AgentActionPlan: action_type: AgentActionType target_entity_id: Optional[int] = None skill_id: Optional[int] = None target_x: Optional[float] = None target_y: Optional[float] = None target_drop_id: Optional[str] = None reasoning: str = "" class AgentDecisionCore: """ Tactical Brain driving an autonomous character. Features dual-layer cognition: - Layer 1: High-level tactical evaluation & stance control. - Layer 2: 30Hz Authoritative reflex dispatcher. """ # Ngũ Hành Counter Lookups: Defender Element -> Element that counters it ELEMENT_COUNTERS = { FiveElements.MOC: FiveElements.KIM, # Kim khắc Mộc FiveElements.THO: FiveElements.MOC, # Mộc khắc Thổ FiveElements.THUY: FiveElements.THO, # Thổ khắc Thủy FiveElements.HOA: FiveElements.THUY, # Thủy khắc Hỏa FiveElements.KIM: FiveElements.HOA # Hỏa khắc Kim } def __init__( self, player_id: int, provider_config: Optional[LLMProviderConfig] = None, stance: TacticalStance = TacticalStance.BALANCED ) -> None: self.player_id: int = player_id self.provider_config: LLMProviderConfig = provider_config or LLMProviderConfig( provider=LLMProviderType.LOCAL_TACTICAL ) self.stance: TacticalStance = stance self.ground_drops: Dict[str, Dict[str, Any]] = {} self.sequence_counter: int = 0 self.llm_router: LLMProviderClientRouter = LLMProviderClientRouter() self.last_thought: str = "Khởi tạo thần niệm phân thân thành công." def register_ground_drop(self, drop_id: str, item_id: str, x: float, y: float, rarity: int = 1) -> None: self.ground_drops[drop_id] = { "drop_id": drop_id, "item_id": item_id, "x": x, "y": y, "rarity": rarity } def remove_ground_drop(self, drop_id: str) -> None: self.ground_drops.pop(drop_id, None) def build_perception(self, engine: ServerEngineLoop) -> AgentPerceptionSnapshot: """Extracts spatial and tactical context from the authoritative world.""" player_entity = engine.active_entities.get(self.player_id) combat_actor = engine.combat_engine.actors.get(self.player_id) px = player_entity.x if player_entity else 0.0 py = player_entity.y if player_entity else 0.0 pz = engine.entity_z_levels.get(self.player_id, 0.0) cur_hp = combat_actor.current_hp if combat_actor else 1000.0 max_hp = combat_actor.max_hp if combat_actor else 1000.0 hp_ratio = cur_hp / max_hp if max_hp > 0 else 0.0 elem = combat_actor.element if combat_actor else FiveElements.KIM # Perceived nearby entities perceived_entities: List[PerceivedEntity] = [] for ent_id, ent in engine.active_entities.items(): if ent_id == self.player_id: continue actor = engine.combat_engine.actors.get(ent_id) if not actor or actor.current_hp <= 0: continue dx = ent.x - px dy = ent.y - py dist = math.hypot(dx, dy) perceived_entities.append(PerceivedEntity( entity_id=ent_id, x=ent.x, y=ent.y, distance=dist, hp=actor.current_hp, max_hp=actor.max_hp, element=actor.element )) # Sort entities by distance ascending perceived_entities.sort(key=lambda e: e.distance) # Perceived nearby drops perceived_drops: List[PerceivedDrop] = [] for d_id, d in self.ground_drops.items(): dx = d["x"] - px dy = d["y"] - py dist = math.hypot(dx, dy) perceived_drops.append(PerceivedDrop( drop_id=d_id, item_id=d["item_id"], x=d["x"], y=d["y"], distance=dist, rarity=d["rarity"] )) # Sort drops: prioritize rarity descending, then distance ascending perceived_drops.sort(key=lambda d: (-d.rarity, d.distance)) return AgentPerceptionSnapshot( player_id=self.player_id, player_x=px, player_y=py, player_z=pz, current_hp=cur_hp, max_hp=max_hp, hp_ratio=hp_ratio, element=elem, nearby_entities=perceived_entities, nearby_drops=perceived_drops ) def evaluate_tick_action(self, perception: AgentPerceptionSnapshot) -> AgentActionPlan: """ Determines the immediate authoritative action for this tick. Follows priority: 1. Emergency survival reflex (HP < 25%) 2. Treasure Hunter stance high-value loot priority 3. Combat engagement (approaching target or casting skill) 4. Normal Loot pickup 5. Idle / Patrol """ # 1. Emergency survival reflex if perception.hp_ratio < 0.25: if perception.nearby_entities and perception.nearby_entities[0].distance < 4.0: reason = "Sinh mệnh nguy kịch (<25%). Kích hoạt Huyễn Ảnh Bộ né tránh sát thương tức thì!" self.last_thought = reason return AgentActionPlan( action_type=AgentActionType.DODGE, reasoning=reason ) reason = "Sinh mệnh thấp. Tự động phục dụng Hoàn Hồn Đan hồi phục khí huyết." self.last_thought = reason return AgentActionPlan( action_type=AgentActionType.USE_POTION, reasoning=reason ) # 2. Treasure Hunter stance: Prioritize rare drops if self.stance == TacticalStance.TREASURE_HUNTER and perception.nearby_drops: top_drop = perception.nearby_drops[0] reason = f"Phát hiện thần thạch {top_drop.item_id}. Ưu tiên tiến tới thu thập vào Kỳ Trân Các." self.last_thought = reason return AgentActionPlan( action_type=AgentActionType.LOOT, target_drop_id=top_drop.drop_id, target_x=top_drop.x, target_y=top_drop.y, reasoning=reason ) # 3. Combat Engagement if perception.nearby_entities: target = perception.nearby_entities[0] # Check attack range (melee range ~ 3.5 units) if target.distance <= 3.5: counter_elem = self.ELEMENT_COUNTERS.get(target.element, FiveElements.KIM) reason = ( f"Mục tiêu #{target.entity_id} thuộc tính {target.element.name}. " f"Vận chuyển tâm pháp {counter_elem.name} xuất chiêu khắc chế." ) self.last_thought = reason return AgentActionPlan( action_type=AgentActionType.ATTACK_SKILL, target_entity_id=target.entity_id, skill_id=1, reasoning=reason ) else: reason = f"Tiếp cận quái #{target.entity_id} (khoảng cách {target.distance:.1f}m)." self.last_thought = reason return AgentActionPlan( action_type=AgentActionType.MOVE_TOWARDS, target_entity_id=target.entity_id, target_x=target.x, target_y=target.y, reasoning=reason ) # 4. Standard Loot pickup if drops exist if perception.nearby_drops: top_drop = perception.nearby_drops[0] reason = f"Nhặt vật phẩm rơi: {top_drop.item_id}." self.last_thought = reason return AgentActionPlan( action_type=AgentActionType.LOOT, target_drop_id=top_drop.drop_id, target_x=top_drop.x, target_y=top_drop.y, reasoning=reason ) # 5. Default Idle self.last_thought = "Khu vực an toàn. Thần niệm duy trì tuần tra cảnh giới." return AgentActionPlan( action_type=AgentActionType.IDLE, reasoning=self.last_thought ) def dispatch_plan_to_server( self, plan: AgentActionPlan, engine: ServerEngineLoop, dt: float = 0.0333 ) -> None: """Transforms the high-level plan into authoritative server queue commands.""" self.sequence_counter += 1 now_ms = int(time.time() * 1000) if plan.action_type in [AgentActionType.MOVE_TOWARDS, AgentActionType.MOVE_AWAY]: player_ent = engine.active_entities.get(self.player_id) if not player_ent or plan.target_x is None or plan.target_y is None: return dx = plan.target_x - player_ent.x dy = plan.target_y - player_ent.y dist = math.hypot(dx, dy) if dist > 0.001: multiplier = -1.0 if plan.action_type == AgentActionType.MOVE_AWAY else 1.0 dir_x = (dx / dist) * multiplier dir_y = (dy / dist) * multiplier cmd = ClientMoveCommand( entity_id=self.player_id, dir_x=dir_x, dir_y=dir_y, sequence=self.sequence_counter, timestamp_ms=now_ms ) engine.queue_move_command(cmd) elif plan.action_type == AgentActionType.ATTACK_SKILL: if plan.target_entity_id is not None: combat_actor = engine.combat_engine.actors.get(self.player_id) elem = combat_actor.element if combat_actor else FiveElements.KIM base_atk = combat_actor.base_attack if combat_actor else 50.0 skill_cmd = ClientSkillCommand( attacker_id=self.player_id, target_id=plan.target_entity_id, skill_id=plan.skill_id or 1, raw_damage=base_atk * 1.5, element=elem, sequence=self.sequence_counter, timestamp_ms=now_ms ) engine.queue_skill_command(skill_cmd) elif plan.action_type == AgentActionType.DODGE: engine.queue_evasion(entity_id=self.player_id, timestamp_ms=now_ms) elif plan.action_type == AgentActionType.LOOT: if plan.target_drop_id: self.remove_ground_drop(plan.target_drop_id)