"""Autonomous Mode — esplorazione autonoma con LLM-first decision making. Questo modulo estende il main.py per supportare una modalità completamente autonoma dove il bot esplora, impara e cresce da solo, orchestrato dall'LLM. Usage: python -m src.autonomous_mode --config config.yaml --duration 60 La sessione viene salvata in data/autonomous_sessions/ come JSON dettagliato. """ import asyncio import argparse import sys from pathlib import Path from datetime import datetime, timedelta sys.path.insert(0, str(Path(__file__).parent)) from core.telnet_client import TelnetClient from core.ansi_parser import ANSIParser from core.game_state import GameState from core.parsers import ParserFactory, OutputBuffer, EventType from config.settings import load_config from utils.logger import setup_logger from utils.human_behavior import HumanBehavior from modules.decision import LLMOrchestrator, GameContext from modules.decision.combat_ai import CombatConfig from modules.mapper import WorldMapper from modules.character import CharacterCreator, CreatorConfig from modules.quest import QuestManager from modules.progression import ProgressionOptimizer from modules.llm import LLMBrain, LLMConfig, SocialAI from modules.web import MonitoringServer logger = None class AutonomousPlayer: """Player autonomo che esplora e impara guidato dall'LLM.""" def __init__(self, config, duration_minutes: int = 60): self.config = config self.duration = timedelta(minutes=duration_minutes) self.orchestrator: LLMOrchestrator = None self.session_start: datetime = None self.recent_events: list = [] self.max_events = 20 async def run(self): """Esegue la sessione autonoma completa.""" global logger # Inizializza componenti telnet = TelnetClient(self.config.mud.host, self.config.mud.port) ansi_parser = ANSIParser() game_state = GameState() human_behavior = HumanBehavior( self.config.behavior.reaction_time_min_ms, self.config.behavior.reaction_time_max_ms, self.config.behavior.pause_probability, ) # Parser MUD-specifico mud_parser = ParserFactory.create(self.config.mud.mud_type) output_buffer = OutputBuffer(mud_parser) # Mapper world_mapper = WorldMapper( db_path=self.config.mapper.active_storage.db_path, archive_dir=self.config.mapper.historical_storage.archive_dir, enable_sync=True, room_count_trigger=self.config.mapper.historical_storage.snapshot_triggers.room_count, time_interval_hours=self.config.mapper.historical_storage.snapshot_triggers.time_interval_hours, embedding_model=self.config.mapper.semantic_search.model, ) # LLM Brain (essenziale per modalità autonoma) llm_config = LLMConfig( base_url=self.config.ai.llm.base_url, model=self.config.ai.llm.model, temperature=self.config.ai.llm.temperature, max_tokens=self.config.ai.llm.max_tokens, timeout_seconds=self.config.ai.llm.timeout_seconds, max_retries=self.config.ai.llm.max_retries, system_prompt=self.config.ai.llm.system_prompt, ) llm_brain = LLMBrain(llm_config, memory_db_path=self.config.ai.llm.memory_db_path) if not self.config.ai.llm.enabled: logger.error("❌ Modalità autonoma richiede LLM abilitato!") return await llm_brain.initialize() if not llm_brain.is_available: logger.error("❌ LLM non disponibile. Avvia Ollama prima!") return logger.info("🧠 LLM Brain online — Modalità autonoma pronta") # Inizializza orchestratore self.orchestrator = LLMOrchestrator(llm_brain) # Avvia sessione await self.orchestrator.start_session( character_name=self.config.character.name, character_class=self.config.character.preferred_class, initial_goals=[ "Esplora e mappa il mondo", "Impara dai pattern del gioco", "Cresci di livello in sicurezza", "Interagisci socialmente con altri giocatori", ], ) self.session_start = datetime.now() # Connetti al MUD logger.info(f"🔌 Connessione a {self.config.mud.host}:{self.config.mud.port}") if not await telnet.connect(): logger.error("❌ Connessione fallita") return # Stato login login_state = "username" # Callback ricezione async def on_receive(data: str): nonlocal login_state clean_text = ansi_parser.strip_ansi(data) # Login automatico if login_state == "username": if "login" in clean_text.lower() or "name" in clean_text.lower(): await human_behavior.delay() await telnet.send(self.config.mud.username) login_state = "password" return elif login_state == "password": if "password" in clean_text.lower(): await human_behavior.delay() await telnet.send(self.config.mud.password) login_state = "logged_in" return if "take over" in clean_text.lower() or "copy" in clean_text.lower(): await human_behavior.delay() await telnet.send("y") login_state = "logged_in" return if login_state != "logged_in": return # Processa blocchi di output blocks = output_buffer.feed(clean_text) for block in blocks: events = mud_parser.parse_block(block) for event in events: if event.event_type == EventType.PROMPT and event.prompt: game_state.update_from_prompt(event.prompt) elif event.event_type == EventType.ROOM and event.room: game_state.update_room(event.room) # Aggiorna mapper world_mapper.on_room_enter( title=event.room.title or "", description=event.room.description or "", exits=event.room.exits, npcs=event.room.npcs, items=event.room.items, ) # Registra scoperta self.orchestrator.record_discovery( discovery_type="room", description=f"{event.room.title}: {event.room.description[:100]}...", metadata={ "exits": event.room.exits, "npcs": event.room.npcs, "items": event.room.items, } ) # Aggiungi a eventi recenti self._add_event(f"Entrato in: {event.room.title}") elif event.event_type == EventType.COMBAT and event.combat: game_state.update_combat(event.combat) self._add_event(f"Combattimento: {event.combat.action}") if event.combat.action in ["hit", "attack"]: self.orchestrator.record_discovery( discovery_type="danger", description=f"Combattimento in {game_state.current_room.title if game_state.current_room else 'unknown'}", ) # Social AI (sottoinsieme dell'orchestratore) # Per ora usiamo la logica esistente di SocialAI telnet.set_receive_callback(on_receive) receive_task = asyncio.create_task(telnet.receive_loop()) try: logger.info("🎮 Inizio esplorazione autonoma...") logger.info(f"⏱️ Durata massima: {self.duration.total_seconds()/60:.0f} minuti") last_decision_time = asyncio.get_running_loop().time() decision_interval = 3.0 # Secondi tra decisioni while telnet.connected: # Controlla durata sessione elapsed = datetime.now() - self.session_start if elapsed > self.duration: logger.info("⏰ Durata massima raggiunta") break current_time = asyncio.get_running_loop().time() # Prendi decisione ogni N secondi if login_state == "logged_in" and (current_time - last_decision_time) >= decision_interval: # Costruisci contesto context = self._build_game_context(game_state, world_mapper) # Chiedi decisione all'LLM decision = await self.orchestrator.make_decision( context, world_mapper, game_state ) # Esegui azione action = decision.get("action", "look") logger.info(f"🎯 Azione: {action}") logger.info(f" Strategia: {decision.get('strategy', 'unknown')}") logger.info(f" Ragionamento: {decision.get('reasoning', 'N/A')[:100]}") # Notifica mapper se è un movimento if decision.get("strategy") == "explore" and action in ["n", "s", "e", "w", "u", "d", "north", "south", "east", "west", "up", "down"]: world_mapper.on_move(action) await human_behavior.delay() await telnet.send(action) last_decision_time = current_time await asyncio.sleep(0.5) except KeyboardInterrupt: logger.info("⚠️ Interrotto dall'utente") finally: # Salva sessione log_file = await self.orchestrator.end_session(reason="completed") # Stampa riepilogo summary = self.orchestrator.get_session_summary() print(summary) if log_file: logger.info(f"📄 Log completo: {log_file}") # Cleanup world_mapper.close() await telnet.disconnect() receive_task.cancel() def _add_event(self, event: str): """Aggiunge evento alla lista recente.""" self.recent_events.append(f"{datetime.now().strftime('%H:%M:%S')} - {event}") if len(self.recent_events) > self.max_events: self.recent_events.pop(0) def _build_game_context(self, game_state: GameState, world_mapper) -> GameContext: """Costruisce il contesto completo per l'LLM.""" # Calcola livello pericolo danger_level = "low" if game_state.current_room and game_state.current_room.npcs: danger_level = "medium" if game_state.is_in_combat(): danger_level = "high" hp_percent = game_state.character.hp / max(game_state.character.max_hp, 1) if hp_percent < 0.3: danger_level = "high" elif hp_percent < 0.6: danger_level = "medium" if danger_level == "low" else danger_level return GameContext( current_room={ "title": game_state.current_room.title if game_state.current_room else "Unknown", "description": game_state.current_room.description if game_state.current_room else "", "exits": list(game_state.current_room.exits.keys()) if game_state.current_room else [], "npcs": game_state.current_room.npcs if game_state.current_room else [], "items": game_state.current_room.items if game_state.current_room else [], }, character_stats={ "name": game_state.character.name, "class": game_state.character.class_type, "level": game_state.character.level, "hp": game_state.character.hp, "max_hp": game_state.character.max_hp, "mp": game_state.character.mp, "max_mp": game_state.character.max_mp, }, inventory=game_state.inventory, recent_events=self.recent_events.copy(), map_stats={ "total_rooms": world_mapper.total_rooms, "total_exits": world_mapper.total_exits, }, exploration_queue=[], # Da implementare known_npcs=[], known_items=[], danger_level=danger_level, ) def main(): """Entry point per modalità autonoma.""" global logger parser = argparse.ArgumentParser(description="MUD AI Player - Autonomous Mode") parser.add_argument("--config", default="config.yaml", help="Path to config file") parser.add_argument("--duration", type=int, default=60, help="Session duration in minutes") args = parser.parse_args() # Carica configurazione try: config = load_config(args.config) except FileNotFoundError: print(f"❌ Configurazione non trovata: {args.config}") sys.exit(1) except Exception as e: print(f"❌ Errore caricamento config: {e}") sys.exit(1) # Setup logging logger = setup_logger( level=config.logging.level, log_file=config.logging.file, console=config.logging.console, max_file_size_mb=config.logging.max_file_size_mb, backup_count=config.logging.backup_count, ) logger.info("=" * 60) logger.info("🤖 MUD AI PLAYER - AUTONOMOUS MODE") logger.info("=" * 60) logger.info(f"⏱️ Durata: {args.duration} minuti") logger.info(f"🧠 LLM: {config.ai.llm.model}") logger.info(f"🎭 Personaggio: {config.character.name} ({config.character.preferred_class})") logger.info("=" * 60) # Verifica LLM abilitato if not config.ai.llm.enabled: logger.error("❌ Errore: LLM non abilitato in config.yaml!") logger.error(" Imposta ai.llm.enabled: true") sys.exit(1) # Avvia player autonomo player = AutonomousPlayer(config, duration_minutes=args.duration) try: asyncio.run(player.run()) except Exception as e: logger.exception(f"❌ Errore fatale: {e}") sys.exit(1) logger.info("✅ Sessione completata") if __name__ == "__main__": main()