feat: implement autonomous exploration mode with LLM orchestration
- Add LLMOrchestrator for LLM-first decision making - Create autonomous_mode.py as standalone entry point - Implement ExplorationSession with structured logging - Add GameContext for complete game state awareness - LLM decides strategy: explore|combat|quest|social|grinding|retreat - Automatic discovery logging (rooms, NPCs, dangers, items) - Learning memory for dangerous rooms and valuable NPCs - Session logs saved to data/autonomous_sessions/ as JSON - Real-time console output with emojis and reasoning - Fallback to rule-based when LLM unavailable - Safety overrides for low HP and high danger Usage: python -m src.autonomous_mode --duration 60 The bot now explores autonomously, learns patterns, makes strategic decisions, and saves detailed session logs for review.
This commit is contained in:
1 parent
43a1006223
commit
a1a0ad5840
3 files changed
+782
No files matched your search
@@ -0,0 +1,379 @@
|
||||
"""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()
|
||||
@@ -11,6 +11,7 @@ behavior_states Concrete behavior states (Idle, Explore, Combat, Loot).
|
||||
combat_ai Intelligent combat decision-making.
|
||||
engine Central decision engine / state machine.
|
||||
goal_manager High-level goal tracking and prioritisation.
|
||||
llm_orchestrator LLM-first autonomous exploration mode.
|
||||
"""
|
||||
|
||||
from .engine import DecisionEngine
|
||||
@@ -24,6 +25,7 @@ from .behavior_states import (
|
||||
CombatState,
|
||||
LootState,
|
||||
)
|
||||
from .llm_orchestrator import LLMOrchestrator, ExplorationSession, GameContext
|
||||
|
||||
__all__ = [
|
||||
"DecisionEngine",
|
||||
@@ -41,4 +43,7 @@ __all__ = [
|
||||
"ExploreState",
|
||||
"CombatState",
|
||||
"LootState",
|
||||
"LLMOrchestrator",
|
||||
"ExplorationSession",
|
||||
"GameContext",
|
||||
]
|
||||
@@ -0,0 +1,398 @@
|
||||
"""LLM Orchestrator — modalità autonoma con LLM-first decision making.
|
||||
|
||||
In questa modalità, l'LLM diventa il decision maker principale, non solo un fallback.
|
||||
Il bot esplora autonomamente, mappa il mondo, impara dai comportamenti e cerca di crescere.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Dict, List, Optional, Any
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from ..llm.client import LLMClient, LLMConfig, render_template
|
||||
from ..llm.brain import LLMBrain
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
@dataclass
|
||||
class ExplorationSession:
|
||||
"""Rappresenta una sessione di esplorazione autonoma."""
|
||||
session_id: str
|
||||
start_time: datetime
|
||||
character_name: str
|
||||
character_class: str
|
||||
goals: List[Dict] = field(default_factory=list)
|
||||
discoveries: List[Dict] = field(default_factory=list)
|
||||
decisions: List[Dict] = field(default_factory=list)
|
||||
stats: Dict = field(default_factory=dict)
|
||||
|
||||
def to_dict(self) -> Dict:
|
||||
return {
|
||||
"session_id": self.session_id,
|
||||
"start_time": self.start_time.isoformat(),
|
||||
"character_name": self.character_name,
|
||||
"character_class": self.character_class,
|
||||
"goals": self.goals,
|
||||
"discoveries": self.discoveries,
|
||||
"decisions": self.decisions,
|
||||
"stats": self.stats,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class GameContext:
|
||||
"""Contesto completo del gioco per l'LLM."""
|
||||
current_room: Dict
|
||||
character_stats: Dict
|
||||
inventory: List[str]
|
||||
recent_events: List[str]
|
||||
map_stats: Dict
|
||||
exploration_queue: List[Dict]
|
||||
known_npcs: List[str]
|
||||
known_items: List[str]
|
||||
danger_level: str # low, medium, high
|
||||
current_goal: Optional[str] = None
|
||||
|
||||
|
||||
class LLMOrchestrator:
|
||||
"""Orchestratore LLM-first per esplorazione autonoma.
|
||||
|
||||
L'LLM prende decisioni strategiche basate sul contesto completo del gioco.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
llm_brain: LLMBrain,
|
||||
session_log_dir: str = "data/autonomous_sessions",
|
||||
):
|
||||
self.llm_brain = llm_brain
|
||||
self.session_log_dir = Path(session_log_dir)
|
||||
self.session_log_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
self.current_session: Optional[ExplorationSession] = None
|
||||
self.exploration_history: List[Dict] = []
|
||||
self.learning_memory: Dict[str, Any] = {
|
||||
"dangerous_rooms": [],
|
||||
"valuable_npcs": [],
|
||||
"good_equipment": [],
|
||||
"efficient_paths": {},
|
||||
"combat_strategies": {},
|
||||
}
|
||||
|
||||
async def start_session(
|
||||
self,
|
||||
character_name: str,
|
||||
character_class: str,
|
||||
initial_goals: Optional[List[str]] = None,
|
||||
) -> ExplorationSession:
|
||||
"""Avvia una nuova sessione di esplorazione autonoma."""
|
||||
session_id = f"session_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
|
||||
|
||||
self.current_session = ExplorationSession(
|
||||
session_id=session_id,
|
||||
start_time=datetime.now(),
|
||||
character_name=character_name,
|
||||
character_class=character_class,
|
||||
goals=[{"description": g, "status": "active"} for g in (initial_goals or [])],
|
||||
)
|
||||
|
||||
logger.info(f"🚀 Avviata sessione autonoma: {session_id}")
|
||||
logger.info(f" Personaggio: {character_name} ({character_class})")
|
||||
logger.info(f" Obiettivi: {initial_goals or 'Esplorazione libera'}")
|
||||
|
||||
return self.current_session
|
||||
|
||||
async def make_decision(
|
||||
self,
|
||||
game_context: GameContext,
|
||||
world_mapper,
|
||||
game_state,
|
||||
) -> Dict:
|
||||
"""L'LLM prende una decisione strategica basata sul contesto."""
|
||||
if not self.llm_brain.is_available:
|
||||
logger.warning("LLM non disponibile, uso logica rule-based")
|
||||
return self._fallback_decision(game_context)
|
||||
|
||||
# Costruisci prompt contestuale
|
||||
prompt = self._build_decision_prompt(game_context)
|
||||
|
||||
# Chiedi decisione all'LLM
|
||||
response = await self.llm_brain.client.generate(prompt, json_mode=True)
|
||||
|
||||
if not response.success:
|
||||
logger.error(f"LLM decision failed: {response.error}")
|
||||
return self._fallback_decision(game_context)
|
||||
|
||||
try:
|
||||
decision = json.loads(response.text)
|
||||
|
||||
# Valida e arricchisci decisione
|
||||
decision = self._validate_decision(decision, game_context)
|
||||
|
||||
# Log della decisione
|
||||
self._log_decision(decision, game_context)
|
||||
|
||||
return decision
|
||||
|
||||
except json.JSONDecodeError as e:
|
||||
logger.error(f"Invalid JSON from LLM: {e}")
|
||||
return self._fallback_decision(game_context)
|
||||
|
||||
def _build_decision_prompt(self, context: GameContext) -> str:
|
||||
"""Costruisce il prompt per la decisione LLM."""
|
||||
return f"""Sei un AI player autonomo in un MUD. Devi decidere la prossima azione.
|
||||
|
||||
## STATO ATTUALE
|
||||
|
||||
### Personaggio
|
||||
- Nome: {context.character_stats.get('name', 'Unknown')}
|
||||
- Classe: {context.character_stats.get('class', 'Unknown')}
|
||||
- Livello: {context.character_stats.get('level', 1)}
|
||||
- HP: {context.character_stats.get('hp', 0)}/{context.character_stats.get('max_hp', 100)}
|
||||
- MP: {context.character_stats.get('mp', 0)}/{context.character_stats.get('max_mp', 100)}
|
||||
|
||||
### Posizione Attuale
|
||||
- Stanza: {context.current_room.get('title', 'Unknown')}
|
||||
- Descrizione: {context.current_room.get('description', 'N/A')[:200]}
|
||||
- Uscite: {', '.join(context.current_room.get('exits', []))}
|
||||
- NPC presenti: {context.current_room.get('npcs', [])}
|
||||
- Oggetti: {context.current_room.get('items', [])}
|
||||
- Livello pericolo: {context.danger_level}
|
||||
|
||||
### Inventario
|
||||
{context.inventory[:10] if context.inventory else ['Vuoto']}
|
||||
|
||||
### Statistiche Mappa
|
||||
- Stanze scoperte: {context.map_stats.get('total_rooms', 0)}
|
||||
- Uscite mappate: {context.map_stats.get('total_exits', 0)}
|
||||
- Uscite da esplorare: {len(context.exploration_queue)}
|
||||
|
||||
### Memoria Apprendimento
|
||||
- Stanze pericolose: {self.learning_memory['dangerous_rooms'][-5:]}
|
||||
- NPC utili: {self.learning_memory['valuable_npcs'][-5:]}
|
||||
- Eventi recenti: {context.recent_events[-5:]}
|
||||
|
||||
## ISTRUZIONI
|
||||
|
||||
Decidi la prossima azione. Rispondi in JSON:
|
||||
|
||||
```json
|
||||
{{
|
||||
"action": "comando_da_inviare",
|
||||
"reasoning": "breve spiegazione della decisione",
|
||||
"strategy": "explore|combat|quest|social|grinding|retreat",
|
||||
"expected_outcome": "cosa ti aspetti succeda",
|
||||
"risk_level": "low|medium|high",
|
||||
"learning_note": "cosa hai imparato da questa situazione (opzionale)"
|
||||
}}
|
||||
```
|
||||
|
||||
Strategie consigliate:
|
||||
- **explore**: Muoviti verso uscite non esplorate se sicuro
|
||||
- **combat**: Attacca solo se hai vantaggio
|
||||
- **quest**: Interagisci con NPC per informazioni
|
||||
- **social**: Rispondi ad altri giocatori
|
||||
- **grinding**: Torna a zone sicure per exp
|
||||
- **retreat**: Scappa se HP bassi o pericolo elevato
|
||||
|
||||
REGOLE CRITICHE:
|
||||
1. Se HP < 30% → PRIORITÀ: curati o scappa
|
||||
2. Se pericolo alto → Esplora stanze alternative
|
||||
3. Se NPC hostili → Valora combattere o fuggire
|
||||
4. Mantieni obiettivi a lungo termine
|
||||
5. Annota pattern per apprendimento futuro"""
|
||||
|
||||
def _validate_decision(
|
||||
self,
|
||||
decision: Dict,
|
||||
context: GameContext,
|
||||
) -> Dict:
|
||||
"""Valida e corregge la decisione dell'LLM."""
|
||||
# Assicura campi obbligatori
|
||||
if "action" not in decision:
|
||||
decision["action"] = "look"
|
||||
if "strategy" not in decision:
|
||||
decision["strategy"] = "explore"
|
||||
if "reasoning" not in decision:
|
||||
decision["reasoning"] = "Nessuna spiegazione fornita"
|
||||
|
||||
# Sanitizza comando
|
||||
decision["action"] = decision["action"].strip().lower()
|
||||
|
||||
# Override di sicurezza
|
||||
hp_percent = context.character_stats.get('hp', 100) / max(context.character_stats.get('max_hp', 100), 1)
|
||||
if hp_percent < 0.3 and decision["strategy"] != "retreat":
|
||||
logger.warning(f"Override decisione: HP critico ({hp_percent:.0%}), forzo ritirata")
|
||||
decision["action"] = self._get_flee_command(context)
|
||||
decision["strategy"] = "retreat"
|
||||
decision["reasoning"] += " [OVERRIDE: HP critico]"
|
||||
|
||||
return decision
|
||||
|
||||
def _get_flee_command(self, context: GameContext) -> str:
|
||||
"""Trova direzione di fuga."""
|
||||
exits = context.current_room.get('exits', [])
|
||||
return exits[0] if exits else "flee"
|
||||
|
||||
def _fallback_decision(self, context: GameContext) -> Dict:
|
||||
"""Decisione di fallback quando LLM non disponibile."""
|
||||
exits = context.current_room.get('exits', [])
|
||||
|
||||
# Logica semplice: esplora o curati
|
||||
hp_percent = context.character_stats.get('hp', 100) / max(context.character_stats.get('max_hp', 100), 1)
|
||||
|
||||
if hp_percent < 0.5:
|
||||
return {
|
||||
"action": "rest",
|
||||
"reasoning": "HP basso, recupero energia (fallback mode)",
|
||||
"strategy": "retreat",
|
||||
"expected_outcome": "Recupero HP",
|
||||
"risk_level": "low",
|
||||
}
|
||||
elif exits and context.exploration_queue:
|
||||
# Esplora direzione non mappata
|
||||
unexplored = context.exploration_queue[0] if context.exploration_queue else None
|
||||
if unexplored:
|
||||
return {
|
||||
"action": unexplored.get('direction', exits[0]),
|
||||
"reasoning": "Esplorazione automatica (fallback mode)",
|
||||
"strategy": "explore",
|
||||
"expected_outcome": "Scoperta nuova stanza",
|
||||
"risk_level": "medium",
|
||||
}
|
||||
|
||||
return {
|
||||
"action": "look",
|
||||
"reasoning": "Osservazione ambiente (fallback mode)",
|
||||
"strategy": "explore",
|
||||
"expected_outcome": "Raccolta informazioni",
|
||||
"risk_level": "low",
|
||||
}
|
||||
|
||||
def record_discovery(
|
||||
self,
|
||||
discovery_type: str,
|
||||
description: str,
|
||||
metadata: Optional[Dict] = None,
|
||||
):
|
||||
"""Registra una nuova scoperta durante l'esplorazione."""
|
||||
if not self.current_session:
|
||||
return
|
||||
|
||||
discovery = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"type": discovery_type, # room, npc, item, quest, danger
|
||||
"description": description,
|
||||
"metadata": metadata or {},
|
||||
}
|
||||
|
||||
self.current_session.discoveries.append(discovery)
|
||||
|
||||
# Aggiorna memoria apprendimento
|
||||
if discovery_type == "danger":
|
||||
self.learning_memory["dangerous_rooms"].append(description)
|
||||
elif discovery_type == "valuable_npc":
|
||||
self.learning_memory["valuable_npcs"].append(description)
|
||||
elif discovery_type == "good_equipment":
|
||||
self.learning_memory["good_equipment"].append(description)
|
||||
|
||||
logger.info(f"🔍 Scoperta: [{discovery_type}] {description}")
|
||||
|
||||
def _log_decision(self, decision: Dict, context: GameContext):
|
||||
"""Logga la decisione presa."""
|
||||
if not self.current_session:
|
||||
return
|
||||
|
||||
log_entry = {
|
||||
"timestamp": datetime.now().isoformat(),
|
||||
"decision": decision,
|
||||
"context": {
|
||||
"room": context.current_room.get('title'),
|
||||
"hp_percent": context.character_stats.get('hp', 0) / max(context.character_stats.get('max_hp', 1), 1),
|
||||
"danger": context.danger_level,
|
||||
},
|
||||
}
|
||||
|
||||
self.current_session.decisions.append(log_entry)
|
||||
|
||||
logger.info(f"🧠 Decisione LLM: [{decision['strategy']}] {decision['action']}")
|
||||
logger.info(f" Ragionamento: {decision['reasoning']}")
|
||||
|
||||
async def end_session(self, reason: str = "manual") -> Path:
|
||||
"""Termina la sessione e salva il log."""
|
||||
if not self.current_session:
|
||||
return None
|
||||
|
||||
self.current_session.stats["end_time"] = datetime.now().isoformat()
|
||||
self.current_session.stats["duration_minutes"] = (
|
||||
datetime.now() - self.current_session.start_time
|
||||
).total_seconds() / 60
|
||||
self.current_session.stats["total_discoveries"] = len(self.current_session.discoveries)
|
||||
self.current_session.stats["total_decisions"] = len(self.current_session.decisions)
|
||||
self.current_session.stats["end_reason"] = reason
|
||||
|
||||
# Salva su file
|
||||
log_file = self.session_log_dir / f"{self.current_session.session_id}.json"
|
||||
with open(log_file, 'w', encoding='utf-8') as f:
|
||||
json.dump(self.current_session.to_dict(), f, indent=2, ensure_ascii=False)
|
||||
|
||||
logger.info(f"💾 Sessione salvata: {log_file}")
|
||||
logger.info(f" Durata: {self.current_session.stats['duration_minutes']:.1f} minuti")
|
||||
logger.info(f" Scoperte: {self.current_session.stats['total_discoveries']}")
|
||||
logger.info(f" Decisioni: {self.current_session.stats['total_decisions']}")
|
||||
|
||||
return log_file
|
||||
|
||||
def get_session_summary(self) -> str:
|
||||
"""Genera un riepilogo testuale della sessione."""
|
||||
if not self.current_session:
|
||||
return "Nessuna sessione attiva"
|
||||
|
||||
lines = [
|
||||
f"\n{'='*60}",
|
||||
f"📊 RIEPILOGO SESSIONE: {self.current_session.session_id}",
|
||||
f"{'='*60}",
|
||||
f"Personaggio: {self.current_session.character_name} ({self.current_session.character_class})",
|
||||
f"Inizio: {self.current_session.start_time.strftime('%Y-%m-%d %H:%M:%S')}",
|
||||
f"",
|
||||
f"🎯 OBIETTIVI:",
|
||||
]
|
||||
|
||||
for goal in self.current_session.goals:
|
||||
status_icon = "✅" if goal['status'] == 'completed' else "🔄" if goal['status'] == 'active' else "⏸️"
|
||||
lines.append(f" {status_icon} {goal['description']}")
|
||||
|
||||
lines.extend([
|
||||
f"",
|
||||
f"🔍 SCOPERTE PRINCIPALI ({len(self.current_session.discoveries)} totali):",
|
||||
])
|
||||
|
||||
# Raggruppa per tipo
|
||||
discoveries_by_type = {}
|
||||
for d in self.current_session.discoveries[-20:]: # Ultime 20
|
||||
t = d['type']
|
||||
if t not in discoveries_by_type:
|
||||
discoveries_by_type[t] = []
|
||||
discoveries_by_type[t].append(d['description'])
|
||||
|
||||
for dtype, items in discoveries_by_type.items():
|
||||
lines.append(f" {dtype.upper()}:")
|
||||
for item in items[:3]: # Max 3 per tipo
|
||||
lines.append(f" - {item}")
|
||||
|
||||
lines.extend([
|
||||
f"",
|
||||
f"🧠 APPRENDIMENTO:",
|
||||
f" Stanze pericolose: {len(self.learning_memory['dangerous_rooms'])}",
|
||||
f" NPC utili: {len(self.learning_memory['valuable_npcs'])}",
|
||||
f" Equip. buono: {len(self.learning_memory['good_equipment'])}",
|
||||
f"",
|
||||
f"{'='*60}",
|
||||
])
|
||||
|
||||
return "\n".join(lines)
|
||||
Reference in new issue
Block a user