""" Bộ Điều Phối Trí Tuệ Trung Tâm (Agentic Coordinator - Façade Pattern) Đóng vai trò cầu nối duy nhất kết nối toàn bộ 8 phân hệ Agentic với Tool hiện tại (control_center.py, overlay_gui.py, core_shm_bridge.py, log_watcher.py). Đảm bảo 100% thread-safe, non-blocking và tương thích ngược hoàn hảo. """ import logging import threading from typing import Any, Callable, Dict, List, Optional from src.agentic_engine.bandit_advisor import LinUCBTacticalAdvisor from src.agentic_engine.blackbox import BlackboxRecorder from src.agentic_engine.dps_meter import DPSAccumulator from src.agentic_engine.item_lifecycle import ( ItemLifecycleManager, PreIdTriageResult, ) from src.agentic_engine.reflection_critic import ReflectionCritic from src.agentic_engine.reflex_tuner import ReflexParameterTuner from src.agentic_engine.run_tracker import RunTracker from src.agentic_engine.what_is_better import ( ItemComparisonResult, ItemStats, WhatIsBetterEvaluator, ) from src.assistant_tool.core_shm_bridge import CoreTelemetrySnapshot from src.assistant_tool.log_watcher import LogEvent logger = logging.getLogger("AgenticCoordinator") class AgenticCoordinator: """ Đầu não điều phối thống nhất toàn bộ hệ thống Agentic Self-Learning của AutoPOE2. """ _instance: Optional["AgenticCoordinator"] = None @classmethod def get_instance(cls) -> "AgenticCoordinator": if cls._instance is None: cls._instance = AgenticCoordinator() return cls._instance def __init__(self, enabled: bool = True): self.enabled = enabled self._lock = threading.Lock() # Khởi tạo các phân hệ cốt lõi self.run_tracker = RunTracker() self.dps_accumulator = DPSAccumulator(window_seconds=3.0) self.blackbox_recorder = BlackboxRecorder(capacity=120) self.bandit_advisor = LinUCBTacticalAdvisor() self.reflex_tuner = ReflexParameterTuner() self.reflection_critic = ReflectionCritic() self.evaluator = WhatIsBetterEvaluator() self.item_lifecycle = ItemLifecycleManager(self.evaluator) # Trạng thái thời gian thực self.current_dps: float = 0.0 self.latest_tactical_action: str = "HOLD_POSITION" self.latest_ucb_scores: Dict[str, float] = {} self.latest_advice_headline: str = "Hệ thống tự học đang sẵn sàng" self.unidentified_items_count: int = 0 self.hooded_one_notice: Optional[str] = None # Danh sách callback cập nhật giao diện người dùng (UI update listeners) self._ui_listeners: List[Callable[[Dict[str, Any]], None]] = [] def register_ui_listener(self, cb: Callable[[Dict[str, Any]], None]) -> None: self._ui_listeners.append(cb) def set_enabled(self, enabled: bool) -> None: self.enabled = enabled logger.info(f"[AgenticCoordinator] Trạng thái Agentic Brain: {'BẬT' if enabled else 'TẮT'}") # ========================================================== # CẦU NỐI SỰ KIỆN 1: TELEMETRY SNAPSHOT (30Hz TỪ SHMBRIDGE) # ========================================================== def on_telemetry_snapshot(self, snapshot: CoreTelemetrySnapshot) -> None: """Xử lý snapshot telemetry gửi từ CoreShmBridge.""" if not self.enabled or not snapshot.is_online: return with self._lock: # 1. Ghi nhận snapshot vào Hộp Đen (Blackbox) self.blackbox_recorder.record_snapshot(snapshot) # 2. Phát hiện nguy cấp (Near-Death) if snapshot.player.hp_percentage < 25.0 and snapshot.player.current_hp > 0: self.reflex_tuner.record_near_death() # 3. Tính toán Realized Combat DPS # (Trích xuất quái vật từ snapshot nếu có cấu trúc) monsters = [] if hasattr(snapshot, "monsters_count") and snapshot.monsters_count > 0: # Tạo mock monster array dựa trên count nếu telemetry chưa giải mã chi tiết từng con monsters = [{"id": i, "hp": 1000} for i in range(min(snapshot.monsters_count, 10))] self.current_dps = self.dps_accumulator.update_monsters_snapshot(monsters) # Cập nhật peak dps vào run hiện tại self.run_tracker.update_run_stats(peak_dps=self.current_dps) # 4. Contextual Bandit đưa ra khuyến nghị hành động tối ưu ctx = self.bandit_advisor.extract_context( hp_pct=snapshot.player.hp_percentage, es_pct=100.0, monster_count=snapshot.monsters_count, boss_poise_pct=0.0, flask_charges_pct=80.0, map_tier=snapshot.area_level, recent_dps=self.current_dps, damage_taken_rate=0.0 ) action, score, all_scores = self.bandit_advisor.recommend_action(ctx) self.latest_tactical_action = action self.latest_ucb_scores = all_scores # ========================================================== # CẦU NỐI SỰ KIỆN 2: LOG WATCHER EVENTS (LatestClient.txt) # ========================================================== def on_log_event(self, event: LogEvent) -> None: """Xử lý sự kiện nhật ký từ LogWatcher.""" if not self.enabled: return with self._lock: # SỰ KIỆN 1: CHUYỂN BẢN ĐỒ / KHU VỰC if event.event_type == "ZONE_CHANGE": # Bóc tách tên bản đồ zone_name = event.message.replace("Đã chuyển đến khu vực: ", "").strip() completed_run = self.run_tracker.on_zone_change(zone_name) # Nếu vừa hoàn thành một map -> Phản tư & tự học if completed_run and completed_run.status == "COMPLETED": self.reflection_critic.analyze_map_completion( area_name=completed_run.area_name, duration_s=completed_run.duration_seconds, peak_dps=completed_run.peak_dps ) # Nếu chạy an toàn không chết, nới lỏng nhẹ tham số phản xạ if completed_run.deaths_count == 0: self.reflex_tuner.record_clean_run() # Kiểm tra tương tác The Hooded One hooded_check = self.item_lifecycle.check_the_hooded_one_interaction( current_area_name=zone_name, unidentified_items_in_inventory=self.unidentified_items_count ) if hooded_check.get("can_interact"): self.hooded_one_notice = hooded_check.get("action_guidance") else: self.hooded_one_notice = None self._notify_ui_update() # SỰ KIỆN 2: NGƯỜI CHƠI TỬ TRẬN (DEATH EVENT) elif event.event_type == "PLAYER_DEATH": logger.warning(f"[AgenticCoordinator] Phát hiện người chơi tử trận: {event.message}") self.run_tracker.record_death() self.reflex_tuner.record_death() # Khóa và xuất Hộp Đen 120 ticks ticks = self.blackbox_recorder.get_recent_ticks() area_name = ticks[-1].get("area_name", "Unknown") if ticks else "Unknown" area_lvl = ticks[-1].get("area_level", 0) if ticks else 0 # Reflection Critic phân tích nguyên nhân gốc rễ lesson = self.reflection_critic.analyze_death_event( area_name=area_name, area_level=area_lvl, killer_info=event.sender, blackbox_ticks=ticks ) # Cập nhật phản hồi tiêu cực vào Bandit if ticks: ctx = self.bandit_advisor.extract_context( hp_pct=0.0, es_pct=0.0, monster_count=ticks[-1].get("monsters_count", 0), boss_poise_pct=0.0, flask_charges_pct=0.0, map_tier=area_lvl, recent_dps=self.current_dps, damage_taken_rate=100.0 ) reward = self.bandit_advisor.compute_combat_reward( self.latest_tactical_action, 0.0, 0.0, 0, is_player_slain=True ) self.bandit_advisor.update_feedback(self.latest_tactical_action, ctx, reward) self.latest_advice_headline = f"Hậu kiểm tử vong: {lesson.get('local_diagnosis', {}).get('summary', '')}" self._notify_ui_update() # SỰ KIỆN 3: GIÁM ĐỊNH ĐỒ (ITEMS IDENTIFIED) elif "Items identified" in event.message or "Item identified" in event.message: try: parts = event.message.strip().split() count = int(parts[1]) if len(parts) > 1 and parts[1].isdigit() else 1 except Exception: count = 1 if self.unidentified_items_count >= count: self.unidentified_items_count -= count else: self.unidentified_items_count = 0 logger.info(f"[AgenticCoordinator] Người chơi đã giám định {count} món đồ. Còn lại trong túi: {self.unidentified_items_count}") self._notify_ui_update() # SỰ KIỆN 4: NHẶT ĐỒ / RỚT ĐỒ CHƯA GIÁM ĐỊNH elif event.event_type in ("ITEM_DROP", "ITEM_PICKUP"): self.unidentified_items_count += 1 self._notify_ui_update() # ========================================================== # CÁC API TRẢ LỜI CÂU HỎI "WHAT IS BETTER" # ========================================================== def appraise_item_comparison( self, current_equipped: ItemStats, candidate_item: ItemStats, player_resists: Optional[Dict[str, int]] = None ) -> ItemComparisonResult: """Trả lời câu hỏi: Đồ mới nhặt có tốt hơn đồ đang mặc không?""" return self.item_lifecycle.process_identified_item( item_id="item_query", revealed_stats=candidate_item, current_equipped=current_equipped, player_resists=player_resists ) def triage_unidentified_item( self, base_name: str, item_level: int, rarity: str, slot: str, scrolls_available: int = 40, in_combat_zone: bool = True ) -> PreIdTriageResult: """Đánh giá tiền giám định: Đồ này có đáng nhặt hay tốn Scroll không?""" return self.item_lifecycle.triage_unidentified_item( base_name=base_name, item_level=item_level, rarity=rarity, slot=slot, scrolls_available=scrolls_available, in_combat_zone=in_combat_zone ) def simulate_sample_item_appraisal(self) -> ItemComparisonResult: """ Thực hiện một ca thẩm định trang bị mẫu cho Monk / Martial Artist: So sánh găng tay đang đeo (Iron Gauntlets) với găng tay mới nhặt (Spiked Fists). """ curr = ItemStats( name="Iron Gauntlets of the Bear", slot="gloves", item_level=30, flat_phys=12.0, flat_cold=0.0, attack_speed_pct=6.0, max_life=42, fire_res=18, cold_res=15 ) cand = ItemStats( name="Spiked Fists of the Polar Vortex", slot="gloves", item_level=75, flat_phys=32.0, flat_cold=58.0, attack_speed_pct=18.0, crit_multiplier_pct=28.0, max_life=88, fire_res=32, cold_res=36, chaos_res=22, spell_suppression_pct=14 ) res = self.appraise_item_comparison(curr, cand, {"fire": 75, "cold": 75, "lightning": 75}) self.latest_advice_headline = f"Thẩm định đồ: {res.headline_verdict} ({res.decision})" self._notify_ui_update() return res def get_dashboard_state(self) -> Dict[str, Any]: """Tổng hợp toàn bộ trạng thái của phân hệ Agentic để hiển thị lên GUI.""" session_stats = self.run_tracker.get_session_summary() reflex_params = self.reflex_tuner.get_current_parameters() recent_lessons = self.reflection_critic.get_recent_lessons(limit=3) return { "enabled": self.enabled, "current_dps": self.current_dps, "recommended_action": self.latest_tactical_action, "advice_headline": self.latest_advice_headline, "hooded_one_notice": self.hooded_one_notice, "unidentified_items_count": self.unidentified_items_count, "session_runs": session_stats.get("total_runs", 0), "avg_clear_time": session_stats.get("avg_clear_time", 0.0), "total_deaths": session_stats.get("total_deaths", 0), "fastest_clear": session_stats.get("fastest_clear", 0.0), "tuned_flask_hp": reflex_params.get("flask_hp_threshold", 0.55), "tuned_dodge_burst": reflex_params.get("burst_dodge_threshold", 0.22), "tuned_kiting_dist": reflex_params.get("kiting_distance", 35.0), "recent_lessons": recent_lessons } def _notify_ui_update(self) -> None: state = self.get_dashboard_state() for cb in self._ui_listeners: try: cb(state) except Exception as e: logger.error(f"[AgenticCoordinator] Lỗi UI listener: {e}")