""" AutoPOE2 Tactical Brain - Bộ Não Chiến Thuật Điều Phối Cấp Cao Tích hợp Google Antigravity SDK & Gemini 3.8/3.7 Flash Phân tầng: Tầng 1 (C++ Reflex 1000Hz) <-> Tầng 2 (Antigravity Brain 1-5Hz) """ import asyncio import logging import os from enum import Enum from typing import Any, List, Optional from pydantic import BaseModel, Field # Kiểm tra thư viện Google Antigravity SDK và Google GenAI try: from google.antigravity import Agent, LocalAgentConfig # type: ignore[import-not-found] from google.antigravity import types as agy_types # type: ignore[import-not-found] ANTIGRAVITY_AVAILABLE = True except ImportError: ANTIGRAVITY_AVAILABLE = False try: from google import genai from google.genai import types as genai_types GENAI_AVAILABLE = True except ImportError: GENAI_AVAILABLE = False from src.assistant_tool.core_shm_bridge import CoreShmBridge, CoreTelemetrySnapshot from src.assistant_tool.screen_capturer import ScreenCapturer logger = logging.getLogger("TacticalBrain") # ========================================================== # 1. CÁC SCHEMA DỮ LIỆU ĐẦU RA CHIẾN THUẬT (PYDANTIC) # ========================================================== class TacticalDirective(str, Enum): KITE = "KITE" # Thả diều vừa lùi vừa đánh RETREAT_TO_SAFE_ZONE = "RETREAT_TO_SAFE_ZONE" # Rút lui về khu vực đã dọn sạch FOCUS_ELITE = "FOCUS_ELITE" # Khóa mục tiêu dồn dame Boss/Rare EXECUTE_COMBO_STAGGER = "EXECUTE_COMBO_STAGGER" # Xả Heavy Finisher khi quái Stagger REFILL_FLASKS = "REFILL_FLASKS" # Hết bình máu, rút lui nạp bình PORTAL_OUT = "PORTAL_OUT" # Mở portal thoát hiểm chiến thuật CLEAR_TRASH = "CLEAR_TRASH" # Dọn dẹp quái thường xung quanh LOOT_SAFETY_WINDOW = "LOOT_SAFETY_WINDOW" # An toàn, cho phép nhặt đồ HOLD_POSITION = "HOLD_POSITION" # Giữ vị trí xả sát thương tối đa class DangerLevel(str, Enum): LOW = "LOW" MODERATE = "MODERATE" HIGH = "HIGH" CRITICAL = "CRITICAL" class TacticalRecommendation(BaseModel): """Cấu trúc khuyến nghị chiến thuật chuẩn hóa từ Gemini/Antigravity Agent""" primary_directive: TacticalDirective = Field( description="Chỉ thị chiến thuật cấp cao ưu tiên số 1 hiện tại" ) danger_level: DangerLevel = Field( description="Đánh giá mức độ nguy hiểm của tình thế hiện tại" ) action_summary: str = Field( description="Câu trả lời ngắn gọn (dưới 20 từ) trả lời cho câu hỏi: 'Tốt nhất bây giờ mình nên làm gì?'" ) detailed_reasoning: str = Field( description="Phân tích chi tiết lý do dựa trên máu, vị trí quái, mod của boss và tài nguyên bình máu" ) suggested_target_id: Optional[int] = Field( default=0, description="ID thực thể mục tiêu cần focus hoặc tránh né (nếu có)" ) suggested_target_xyz: Optional[List[float]] = Field( default_factory=lambda: [0.0, 0.0, 0.0], description="Tọa độ [X, Y, Z] khuyến nghị di chuyển tới hoặc xả chiêu" ) macro_opcode_action: int = Field( default=0, description="Mã MacroOpCode gửi xuống C++ Core (0: IDLE, 1: NAV, 2: ENGAGE, 4: PORTAL, 99: PANIC)" ) # ========================================================== # 2. CUSTOM TOOLS CỦA ANTIGRAVITY AGENT (QUAN SÁT & TÍNH TOÁN) # ========================================================== def get_threat_assessment(telemetry_summary: str) -> str: """Đánh giá bản đồ mối đe dọa không gian dựa trên telemetry hiện tại.""" return f"Threat analysis compiled: {telemetry_summary}" def find_safe_escape_vector(player_x: float, player_y: float, player_z: float) -> str: """Tính toán vector rút lui an toàn nhất dựa trên NavMesh và mật độ quái.""" retreat_x = player_x - 25.0 retreat_y = player_y - 25.0 return f"Safe vector calculated: Target X={retreat_x:.1f}, Y={retreat_y:.1f}, Z={player_z:.1f}" # ========================================================== # 3. CLASS BỘ NÃO CHIẾN THUẬT CHÍNH (TACTICAL BRAIN) # ========================================================== class TacticalBrain: """ Bộ Điều Phối Chiến Thuật Cấp Cao cho AutoPOE2. Kết hợp Google Antigravity SDK với C++ Reflex IPC. """ def __init__(self, shm_bridge: Optional[CoreShmBridge] = None, api_key: Optional[str] = None): self.shm_bridge = shm_bridge or CoreShmBridge() self.api_key = api_key or os.getenv("GEMINI_API_KEY", "") self.model_name = os.getenv("GEMINI_MODEL", "gemini-3.7-flash") self.screen_capturer = ScreenCapturer(output_dir="captures") self.last_recommendation: Optional[TacticalRecommendation] = None self._is_evaluating = False self._autonomous_running = False # Khởi tạo Engine tương thích kép (Antigravity SDK / Google GenAI) self._init_ai_client() def _init_ai_client(self): self.genai_client = None if GENAI_AVAILABLE and self.api_key: try: self.genai_client = genai.Client(api_key=self.api_key) logger.info(f"[TacticalBrain] Đã kết nối Google GenAI Engine ({self.model_name})") except Exception as e: logger.error(f"[TacticalBrain] Lỗi khởi tạo GenAI Client: {e}") # ---------------------------------------------------------- # CHỨC NĂNG TRỌNG TÂM: "TỐT NHẤT BÂY GIỜ MÌNH NÊN LÀM GÌ?" # ---------------------------------------------------------- def advise_best_action_sync(self, include_screen: bool = True) -> TacticalRecommendation: """Phiên bản đồng bộ tiện lợi cho GUI Event loop.""" try: loop = asyncio.get_event_loop() if loop.is_running(): return self._rule_based_tactical_fallback(self.shm_bridge.last_snapshot) return loop.run_until_complete(self.advise_best_action(include_screen)) except Exception: return asyncio.run(self.advise_best_action(include_screen)) async def advise_best_action(self, include_screen: bool = True) -> TacticalRecommendation: """ Trả lời tức thời câu hỏi trọng tâm của game thủ: 'Tốt nhất bây giờ mình nên làm gì?' Đọc snapshot telemetry thời gian thực + chụp ảnh màn hình để đưa ra khuyến nghị. """ if self._is_evaluating: if self.last_recommendation: return self.last_recommendation return self._create_fallback_recommendation("Đang phân tích bối cảnh, giữ vị trí an toàn!") self._is_evaluating = True try: snapshot = self.shm_bridge.last_snapshot telemetry_context = self._build_telemetry_prompt(snapshot) # Chụp màn hình bối cảnh nếu có yêu cầu image_bytes = None if include_screen: img_path = self.screen_capturer.capture_screenshot("TACTICAL_QUERY", {"hp": snapshot.player.hp_percentage}) if img_path and os.path.exists(img_path): with open(img_path, "rb") as f: image_bytes = f.read() recommendation = await self._query_antigravity_agent(telemetry_context, image_bytes) self.last_recommendation = recommendation # Tự động gửi chỉ thị xuống C++ Core nếu là tình huống cấp bách if recommendation.macro_opcode_action > 0: self._dispatch_directive_to_core(recommendation) return recommendation except Exception as e: logger.error(f"[TacticalBrain] Lỗi phân tích chiến thuật: {e}") return self._create_fallback_recommendation(f"Duy trì cảnh giác! (Lỗi: {e})") finally: self._is_evaluating = False def _build_telemetry_prompt(self, snap: CoreTelemetrySnapshot) -> str: p = snap.player return ( f"=== BỐI CẢNH CHIẾN ĐẤU POE2 (v0.5.5) ===\n" f"- Khu vực: {snap.area_name or 'Unknown Area'} (Level {snap.area_level})\n" f"- Máu (HP): {p.current_hp}/{p.max_hp} ({p.hp_percentage:.1f}%)\n" f"- Khiên (ES): {p.current_es}/{p.max_es}\n" f"- Mana: {p.current_mana}/{p.max_mana}\n" f"- Spirit: {p.current_spirit}/{p.max_spirit}\n" f"- Tọa độ hiện tại: X={p.pos_x:.1f}, Y={p.pos_y:.1f}, Z={p.pos_z:.1f}\n" f"- Radar xung quanh: {snap.monsters_count} Quái vật, {snap.items_count} Đồ rơi, {snap.portals_count} Cổng Portal\n" f"- Tổng số thực thể đang nạp: {snap.entity_count}\n" ) async def _query_antigravity_agent(self, telemetry_prompt: str, image_bytes: Optional[bytes] = None) -> TacticalRecommendation: """ Sử dụng Antigravity SDK hoặc Gemini GenAI với Structured Outputs Pydantic. """ system_instruction = ( "Bạn là Antigravity Tactical Commander - Bộ Não Chiến Thuật Tối Cao của game thủ Path of Exile 2. " "Nhiệm vụ của bạn là trả lời câu hỏi: 'Tốt nhất bây giờ mình nên làm gì?' (What is the best tactical action right now?).\n" "Quy tắc chiến thuật POE2:\n" "1. Tầng siêu tốc C++ Reflex đã tự động lo việc bơm máu gấp khi HP < 50% và Iframe Dodge Roll khi bị burst sát thương.\n" "2. Nhiệm vụ của bạn là đưa ra chỉ thị vĩ mô: KITE (thả diều khi quái đông/áp sát), RETREAT_TO_SAFE_ZONE (rút lui về vị trí đã dọn sạch), " "FOCUS_ELITE (dồn dame boss), REFILL_FLASKS (hết bình máu cần rút về thành), PORTAL_OUT (mở cổng thoát hiểm khi quá nguy cấp), " "hoặc LOOT_SAFETY_WINDOW (nhặt đồ khi đã an toàn).\n" "3. Câu trả lời action_summary phải cực kỳ súc tích, dứt khoát, mang phong cách chỉ huy chiến trường, giải thích ngắn gọn nguyên nhân." ) user_prompt = f"{telemetry_prompt}\nHãy đưa ra quyết định chiến thuật tối ưu nhất ngay lúc này." # KỊCH BẢN 1: Dùng Google Antigravity SDK nếu đã nạp if ANTIGRAVITY_AVAILABLE: try: config = LocalAgentConfig( model=self.model_name, system_instructions=system_instruction, response_schema=TacticalRecommendation, tools=[get_threat_assessment, find_safe_escape_vector], capabilities=agy_types.CapabilitiesConfig( agent_behavior=agy_types.AgentBehavior.INTERACTIVE ) ) async with Agent(config=config) as agent: resp = await agent.chat(user_prompt) data = await resp.structured_output() if data: return TacticalRecommendation.model_validate(data) except Exception as e: logger.warning(f"[TacticalBrain] Antigravity SDK runtime fallback sang GenAI Client: {e}") client = self.genai_client if client: try: contents: list[Any] = [] if image_bytes: contents.append(genai_types.Part.from_bytes(data=image_bytes, mime_type="image/png")) contents.append(user_prompt) loop = asyncio.get_event_loop() response = await loop.run_in_executor( None, lambda: client.models.generate_content( model=self.model_name, contents=contents, config=genai_types.GenerateContentConfig( system_instruction=system_instruction, response_mime_type="application/json", response_schema=TacticalRecommendation, temperature=0.1, ) ) ) if response.text: return TacticalRecommendation.model_validate_json(response.text) except Exception as e: logger.warning(f"[TacticalBrain] GenAI API error, fallback to heuristics: {e}") return self._rule_based_tactical_fallback(self.shm_bridge.last_snapshot) # ---------------------------------------------------------- # CHUYỂN TIẾP CHỈ THỊ CHIẾN THUẬT VÀO C++ CORE RING BUFFER # ---------------------------------------------------------- def _dispatch_directive_to_core(self, rec: TacticalRecommendation): """Đẩy lệnh điều khiển vĩ mô vào Command Ring Buffer (Offset 0x300000).""" opcode = rec.macro_opcode_action if opcode == 0: return tx, ty, tz = rec.suggested_target_xyz if rec.suggested_target_xyz else (0.0, 0.0, 0.0) target_id = rec.suggested_target_id or 0 logger.info(f"[TacticalBrain -> Core] Phát lệnh MacroOpCode={opcode} tới mục tiêu XYZ=({tx:.1f}, {ty:.1f}, {tz:.1f}) (Target ID: {target_id})") try: if opcode == 99: self.shm_bridge.send_panic_stop() except Exception as e: logger.error(f"[TacticalBrain] Lỗi đẩy lệnh vào Command Buffer: {e}") def _rule_based_tactical_fallback(self, snap: CoreTelemetrySnapshot) -> TacticalRecommendation: """Hệ thống luật chiến thuật dự phòng độ trễ micro-giây (Zero Network Latency).""" hp_pct = snap.player.hp_percentage monsters = snap.monsters_count if hp_pct < 25.0: return TacticalRecommendation( primary_directive=TacticalDirective.PORTAL_OUT, danger_level=DangerLevel.CRITICAL, action_summary="Mở Portal rút lui khẩn cấp về làng ngay lập tức!", detailed_reasoning=f"Máu tụt nguy hiểm ({hp_pct:.1f}%), có {monsters} quái đang bao vây.", macro_opcode_action=4 ) elif hp_pct < 45.0 and monsters > 0: return TacticalRecommendation( primary_directive=TacticalDirective.RETREAT_TO_SAFE_ZONE, danger_level=DangerLevel.HIGH, action_summary="Lùi gấp ra khỏi tầm cào của quái, chờ bình máu hồi đầy!", detailed_reasoning=f"Máu còn {hp_pct:.1f}%, cần tạo khoảng cách để tránh dính combo đòn đánh.", macro_opcode_action=1 ) elif monsters >= 4: return TacticalRecommendation( primary_directive=TacticalDirective.KITE, danger_level=DangerLevel.HIGH, action_summary="Thả diều lùi từng bước, xả kỹ năng diện rộng giải tỏa quái!", detailed_reasoning=f"Mật độ quái cao ({monsters} con quây kín). Cần vừa đi vừa bắn.", macro_opcode_action=1 ) elif monsters > 0: return TacticalRecommendation( primary_directive=TacticalDirective.FOCUS_ELITE, danger_level=DangerLevel.MODERATE, action_summary="Dồn sát thương dứt điểm quái mục tiêu gần nhất!", detailed_reasoning="Số lượng quái trong tầm kiểm soát an toàn.", macro_opcode_action=2 ) elif snap.items_count > 0: return TacticalRecommendation( primary_directive=TacticalDirective.LOOT_SAFETY_WINDOW, danger_level=DangerLevel.LOW, action_summary="Khu vực sạch bóng quái, nhanh chóng nhặt đồ rơi!", detailed_reasoning=f"Không có quái vật xung quanh, có {snap.items_count} vật phẩm trên mặt đất.", macro_opcode_action=3 ) else: return TacticalRecommendation( primary_directive=TacticalDirective.HOLD_POSITION, danger_level=DangerLevel.LOW, action_summary="Khu vực an toàn, tiếp tục di chuyển khám phá bản đồ!", detailed_reasoning="Không có mối đe dọa trực tiếp.", macro_opcode_action=0 ) def _create_fallback_recommendation(self, msg: str) -> TacticalRecommendation: return TacticalRecommendation( primary_directive=TacticalDirective.HOLD_POSITION, danger_level=DangerLevel.MODERATE, action_summary=msg, detailed_reasoning="Khuyến nghị an toàn tức thời.", macro_opcode_action=0 )