refactor: remediation plan Fase 0-4 completata
- Fase 0: env var allineate, SYSTEM_PROMPT deduplicato, .gitignore pulito - Fase 1: GameContext isolato, ciclo DM unificato, provider thread-safe (contextvars) - Fase 2: DB schema esteso (current_area_id, combat_state, character_items) - Fase 3: thread-local connection pool + WAL + busy_timeout - Fase 4: test suite (32 test, 100% pass) - heuristic_fallback come parser di riserva (non secondo pilota)
This commit is contained in:
truetype committed
1 parent
90f823fdb9
commit
cb45172717
16 files changed
+906
-221
No files matched your search
@@ -1,14 +1,13 @@
|
||||
from src.engine.config import ProviderConfig, load_from_env, ProviderKind
|
||||
from src.engine.dm import dm_cycle, interactive
|
||||
from src.engine.llm import set_provider
|
||||
from src.engine.state_machine import GameContext, process_turn
|
||||
from src.engine.game_context import GameContext
|
||||
from src.engine.tools import run as run_tool
|
||||
|
||||
__all__ = [
|
||||
"dm_cycle",
|
||||
"interactive",
|
||||
"GameContext",
|
||||
"process_turn",
|
||||
"run_tool",
|
||||
"ProviderConfig",
|
||||
"ProviderKind",
|
||||
|
||||
@@ -13,6 +13,7 @@ class ProviderConfig:
|
||||
ollama_model: str = "gemma2:9b"
|
||||
openrouter_api_key: str = ""
|
||||
openrouter_model: str = "openrouter/free"
|
||||
openrouter_key_env: str = "DCC_OPENROUTER_KEY"
|
||||
openrouter_free_fallback: bool = True
|
||||
local_base_url: str = "http://192.168.0.240/v1"
|
||||
local_api_key: str = ""
|
||||
@@ -25,7 +26,7 @@ def load_from_env() -> ProviderConfig:
|
||||
kind=os.getenv("DCC_LLM_PROVIDER", "openrouter"), # type: ignore
|
||||
ollama_base_url=os.getenv("DCC_OLLAMA_URL", "http://192.168.0.50:11434"),
|
||||
ollama_model=os.getenv("DCC_OLLAMA_MODEL", "gemma2:9b"),
|
||||
openrouter_api_key=os.getenv("OPENROUTER_API_KEY", ""),
|
||||
openrouter_api_key=os.getenv("DCC_OPENROUTER_KEY", os.getenv("OPENROUTER_API_KEY", "")),
|
||||
openrouter_model=os.getenv("DCC_OPENROUTER_MODEL", "openrouter/free"),
|
||||
openrouter_free_fallback=os.getenv("DCC_OR_FREE_FALLBACK", "1") == "1",
|
||||
local_base_url=os.getenv("DCC_LOCAL_URL", "http://192.168.0.240/v1"),
|
||||
|
||||
+164
-8
@@ -7,19 +7,33 @@ Gestisce schema SQLite e popolazione iniziale dati.
|
||||
import sqlite3
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
DB_PATH = Path(__file__).resolve().parent.parent.parent / "data" / "dcc_rpg.db"
|
||||
DATA_DIR = Path(__file__).resolve().parent.parent.parent / "data"
|
||||
|
||||
_conn_lock = threading.Lock()
|
||||
_local = threading.local()
|
||||
|
||||
|
||||
def get_conn() -> sqlite3.Connection:
|
||||
"""Restituisce connessione al database, creandolo se non esiste."""
|
||||
"""Restituisce connessione thread-local al database.
|
||||
|
||||
Usa WAL mode + busy_timeout per supportare accessi concorrenti.
|
||||
La connessione è cached per-thread via threading.local().
|
||||
"""
|
||||
conn = getattr(_local, "conn", None)
|
||||
if conn is not None:
|
||||
return conn
|
||||
|
||||
DB_PATH.parent.mkdir(parents=True, exist_ok=True)
|
||||
conn = sqlite3.connect(str(DB_PATH))
|
||||
conn = sqlite3.connect(str(DB_PATH), check_same_thread=False)
|
||||
conn.execute("PRAGMA journal_mode=WAL")
|
||||
conn.execute("PRAGMA busy_timeout=5000")
|
||||
conn.execute("PRAGMA foreign_keys=ON")
|
||||
conn.row_factory = sqlite3.Row
|
||||
_local.conn = conn
|
||||
return conn
|
||||
|
||||
|
||||
@@ -58,11 +72,13 @@ def init_schema(conn: sqlite3.Connection) -> None:
|
||||
-- ============================================================
|
||||
|
||||
CREATE TABLE IF NOT EXISTS campaigns (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT NOT NULL,
|
||||
adventure TEXT,
|
||||
turn INTEGER NOT NULL DEFAULT 0,
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
name TEXT NOT NULL,
|
||||
adventure TEXT,
|
||||
turn INTEGER NOT NULL DEFAULT 0,
|
||||
current_area_id TEXT DEFAULT '',
|
||||
current_area_name TEXT DEFAULT '',
|
||||
created_at TEXT NOT NULL DEFAULT (datetime('now'))
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS characters (
|
||||
@@ -105,6 +121,28 @@ def init_schema(conn: sqlite3.Connection) -> None:
|
||||
CREATE UNIQUE INDEX IF NOT EXISTS idx_area_campaign
|
||||
ON adventure_state(campaign_id, area_id);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS combat_state (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
campaign_id INTEGER NOT NULL REFERENCES campaigns(id),
|
||||
monster_id TEXT NOT NULL,
|
||||
monster_name TEXT NOT NULL DEFAULT '',
|
||||
current_hp INTEGER NOT NULL,
|
||||
max_hp INTEGER NOT NULL,
|
||||
status TEXT NOT NULL DEFAULT 'active'
|
||||
CHECK(status IN ('active', 'defeated', 'fled')),
|
||||
UNIQUE(campaign_id, monster_id)
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS character_items (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
character_id INTEGER NOT NULL REFERENCES characters(id),
|
||||
item_def_id TEXT,
|
||||
name TEXT NOT NULL,
|
||||
quantity INTEGER DEFAULT 1,
|
||||
equipped INTEGER DEFAULT 0,
|
||||
properties TEXT DEFAULT '{}'
|
||||
);
|
||||
|
||||
CREATE TABLE IF NOT EXISTS quests (
|
||||
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
||||
campaign_id INTEGER NOT NULL REFERENCES campaigns(id),
|
||||
@@ -178,10 +216,25 @@ def populate_from_json(conn: sqlite3.Connection) -> None:
|
||||
conn.commit()
|
||||
|
||||
|
||||
def _migrate_schema(conn: sqlite3.Connection) -> None:
|
||||
"""Applica migrazioni incrementali allo schema esistente."""
|
||||
migrations = [
|
||||
"ALTER TABLE campaigns ADD COLUMN current_area_id TEXT DEFAULT ''",
|
||||
"ALTER TABLE campaigns ADD COLUMN current_area_name TEXT DEFAULT ''",
|
||||
]
|
||||
for sql in migrations:
|
||||
try:
|
||||
conn.execute(sql)
|
||||
except sqlite3.OperationalError:
|
||||
pass
|
||||
conn.commit()
|
||||
|
||||
|
||||
def init_db() -> sqlite3.Connection:
|
||||
"""Inizializza database completo. Chiamata all'avvio."""
|
||||
conn = get_conn()
|
||||
init_schema(conn)
|
||||
_migrate_schema(conn)
|
||||
populate_from_json(conn)
|
||||
return conn
|
||||
|
||||
@@ -273,7 +326,8 @@ def get_all_tables() -> list[dict]:
|
||||
def create_campaign(name: str, adventure: str = "") -> int:
|
||||
conn = get_conn()
|
||||
cur = conn.execute(
|
||||
"INSERT INTO campaigns (name, adventure) VALUES (?, ?)", (name, adventure)
|
||||
"INSERT INTO campaigns (name, adventure, current_area_id, current_area_name) VALUES (?, ?, '', '')",
|
||||
(name, adventure),
|
||||
)
|
||||
conn.commit()
|
||||
return cur.lastrowid or 0
|
||||
@@ -408,6 +462,108 @@ def get_quests(campaign_id: int) -> list[dict]:
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Area / Combat / Inventory helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def set_current_area(campaign_id: int, area_id: str, area_name: str = "") -> None:
|
||||
conn = get_conn()
|
||||
conn.execute(
|
||||
"UPDATE campaigns SET current_area_id = ?, current_area_name = ? WHERE id = ?",
|
||||
(area_id, area_name, campaign_id),
|
||||
)
|
||||
conn.commit()
|
||||
set_area_state(campaign_id, area_id, visited=True)
|
||||
|
||||
|
||||
def get_current_area(campaign_id: int) -> dict:
|
||||
conn = get_conn()
|
||||
row = conn.execute(
|
||||
"SELECT current_area_id, current_area_name FROM campaigns WHERE id = ?",
|
||||
(campaign_id,),
|
||||
).fetchone()
|
||||
if row and row["current_area_id"]:
|
||||
area = get_area_by_id(row["current_area_id"])
|
||||
return {
|
||||
"area_id": row["current_area_id"],
|
||||
"area_name": row["current_area_name"] or (area["name"] if area else ""),
|
||||
}
|
||||
return {"area_id": "", "area_name": ""}
|
||||
|
||||
|
||||
def set_combat_state(campaign_id: int, monster_id: str, monster_name: str, hp: int, status: str = "active") -> None:
|
||||
conn = get_conn()
|
||||
conn.execute(
|
||||
"""INSERT INTO combat_state (campaign_id, monster_id, monster_name, current_hp, max_hp, status)
|
||||
VALUES (?, ?, ?, ?, ?, ?)
|
||||
ON CONFLICT(campaign_id, monster_id) DO UPDATE SET
|
||||
current_hp = excluded.current_hp,
|
||||
status = excluded.status""",
|
||||
(campaign_id, monster_id, monster_name, hp, hp, status),
|
||||
)
|
||||
conn.commit()
|
||||
|
||||
|
||||
def update_combat_hp(campaign_id: int, monster_id: str, damage: int) -> dict | None:
|
||||
conn = get_conn()
|
||||
row = conn.execute(
|
||||
"SELECT * FROM combat_state WHERE campaign_id = ? AND monster_id = ?",
|
||||
(campaign_id, monster_id),
|
||||
).fetchone()
|
||||
if not row:
|
||||
return None
|
||||
new_hp = max(0, row["current_hp"] - damage)
|
||||
new_status = "defeated" if new_hp <= 0 else row["status"]
|
||||
conn.execute(
|
||||
"UPDATE combat_state SET current_hp = ?, status = ? WHERE id = ?",
|
||||
(new_hp, new_status, row["id"]),
|
||||
)
|
||||
conn.commit()
|
||||
return {"monster_id": monster_id, "previous_hp": row["current_hp"], "new_hp": new_hp, "damage": damage, "status": new_status}
|
||||
|
||||
|
||||
def get_combat_state(campaign_id: int) -> list[dict]:
|
||||
conn = get_conn()
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM combat_state WHERE campaign_id = ? ORDER BY id", (campaign_id,)
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def add_character_item(character_id: int, name: str, item_def_id: str = "", quantity: int = 1) -> int:
|
||||
conn = get_conn()
|
||||
cur = conn.execute(
|
||||
"INSERT INTO character_items (character_id, name, item_def_id, quantity) VALUES (?, ?, ?, ?)",
|
||||
(character_id, name, item_def_id, quantity),
|
||||
)
|
||||
conn.commit()
|
||||
return cur.lastrowid or 0
|
||||
|
||||
|
||||
def remove_character_item(item_id: int) -> bool:
|
||||
conn = get_conn()
|
||||
conn.execute("DELETE FROM character_items WHERE id = ?", (item_id,))
|
||||
conn.commit()
|
||||
return conn.total_changes > 0
|
||||
|
||||
|
||||
def get_character_items(character_id: int) -> list[dict]:
|
||||
conn = get_conn()
|
||||
rows = conn.execute(
|
||||
"SELECT * FROM character_items WHERE character_id = ? ORDER BY id",
|
||||
(character_id,),
|
||||
).fetchall()
|
||||
return [dict(r) for r in rows]
|
||||
|
||||
|
||||
def equip_character_item(item_id: int, equipped: bool = True) -> bool:
|
||||
conn = get_conn()
|
||||
conn.execute("UPDATE character_items SET equipped = ? WHERE id = ?", (int(equipped), item_id))
|
||||
conn.commit()
|
||||
return conn.total_changes > 0
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Adventure data loading (from JSON)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
+8
-7
@@ -4,20 +4,21 @@ import logging
|
||||
|
||||
from src.engine import db, tools
|
||||
from src.engine.config import load_from_env
|
||||
from src.engine.game_context import GameContext
|
||||
from src.engine.heuristic_fallback import parse_action as fallback_parse
|
||||
from src.engine.llm import ask as llm_ask, parse_tool_call
|
||||
from src.engine.prompts import SYSTEM_PROMPT, build_prompt
|
||||
from src.engine.state_machine import GameContext, _parse_action
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _llm_decision(ctx: GameContext, player_action: str) -> tuple[str, dict, str]:
|
||||
prompt = build_prompt(ctx, player_action)
|
||||
def _llm_decision(ctx: GameContext, player_action: str, current_area_id: str = "") -> tuple[str, dict, str]:
|
||||
prompt = build_prompt(ctx, player_action, current_area_id=current_area_id)
|
||||
try:
|
||||
raw = llm_ask(prompt, system=SYSTEM_PROMPT)
|
||||
except Exception as e:
|
||||
log.warning("LLM call failed (%s), fallback a _parse_action", e)
|
||||
tool, params = _parse_action(player_action, ctx)
|
||||
log.warning("LLM call failed (%s), fallback a heuristic parser", e)
|
||||
tool, params = fallback_parse(player_action, ctx.characters)
|
||||
return tool, params, ""
|
||||
|
||||
decision = parse_tool_call(raw)
|
||||
@@ -28,7 +29,7 @@ def _llm_decision(ctx: GameContext, player_action: str) -> tuple[str, dict, str]
|
||||
|
||||
if not tool:
|
||||
log.info("LLM non ha prodotto tool valido, fallback heuristico")
|
||||
tool, params = _parse_action(player_action, ctx)
|
||||
tool, params = fallback_parse(player_action, ctx.characters)
|
||||
|
||||
return tool, params, narration
|
||||
|
||||
@@ -36,7 +37,7 @@ def _llm_decision(ctx: GameContext, player_action: str) -> tuple[str, dict, str]
|
||||
def dm_cycle(campaign_id: int, player_action: str) -> dict:
|
||||
ctx = GameContext.load(campaign_id)
|
||||
|
||||
tool_name, tool_params, narration = _llm_decision(ctx, player_action)
|
||||
tool_name, tool_params, narration = _llm_decision(ctx, player_action, current_area_id=ctx.current_area_id)
|
||||
|
||||
result = tools.run(tool_name, **tool_params)
|
||||
|
||||
|
||||
@@ -0,0 +1,57 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from src.engine import db
|
||||
|
||||
|
||||
@dataclass
|
||||
class GameContext:
|
||||
campaign_id: int
|
||||
turn: int
|
||||
campaign: dict | None = None
|
||||
characters: list[dict] = field(default_factory=list)
|
||||
log: list[dict] = field(default_factory=list)
|
||||
quests: list[dict] = field(default_factory=list)
|
||||
areas: list[dict] = field(default_factory=list)
|
||||
current_area_id: str = ""
|
||||
current_area_name: str = ""
|
||||
|
||||
@classmethod
|
||||
def load(cls, campaign_id: int) -> GameContext:
|
||||
camp = db.get_campaign(campaign_id)
|
||||
if not camp:
|
||||
raise ValueError(f"Campagna #{campaign_id} non trovata.")
|
||||
chars = db.get_characters(campaign_id)
|
||||
log = db.get_log(campaign_id, limit=10)
|
||||
quests = db.get_quests(campaign_id)
|
||||
areas = db.get_area_state(campaign_id)
|
||||
area_info = db.get_current_area(campaign_id)
|
||||
return cls(
|
||||
campaign_id=campaign_id,
|
||||
turn=camp.get("turn", 0),
|
||||
campaign=camp,
|
||||
characters=chars,
|
||||
log=log,
|
||||
quests=quests,
|
||||
areas=areas,
|
||||
current_area_id=area_info.get("area_id", ""),
|
||||
current_area_name=area_info.get("area_name", ""),
|
||||
)
|
||||
|
||||
def summary(self) -> str:
|
||||
camp = self.campaign or {"name": "?", "turn": 0}
|
||||
lines = [f"Campagna: {camp['name']} | Turno {self.turn}"]
|
||||
if self.current_area_id:
|
||||
lines.append(f"Area corrente: {self.current_area_name or self.current_area_id}")
|
||||
for c in self.characters:
|
||||
cls = c.get("char_class") or ""
|
||||
lines.append(
|
||||
f" [{c['id']}] {c['name']} — Lv{c['level']} {cls} "
|
||||
f"PF {c['hp_current']}/{c['hp_max']} PE {c.get('xp', 0)}"
|
||||
)
|
||||
if self.quests:
|
||||
lines.append("Missioni:")
|
||||
for q in self.quests:
|
||||
lines.append(f" [{q['status']}] {q['name']}")
|
||||
return "\n".join(lines)
|
||||
@@ -0,0 +1,77 @@
|
||||
"""
|
||||
Heuristic fallback parser per azioni giocatore.
|
||||
|
||||
Usato SOLO quando il parsing LLM fallisce o restituisce tool vuoto.
|
||||
Non è un secondo pilota — è un salvagente.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
|
||||
def parse_action(raw_action: str, chars: list[dict]) -> tuple[str, dict]:
|
||||
"""Parsing euristico azione testuale → (tool_name, params).
|
||||
|
||||
Chiamato SOLO come fallback quando parse_tool_call() fallisce.
|
||||
"""
|
||||
a = raw_action.lower().strip()
|
||||
cd = re.search(r'(?:cd|dc)\s*(\d+)', a)
|
||||
char_id = chars[0]["id"] if chars else 0
|
||||
|
||||
if any(k in a for k in ["tempra", "riflessi", "volontà", "salvezza"]):
|
||||
for key in ["tempra", "riflessi", "volontà"]:
|
||||
if key in a:
|
||||
return "saving_throw", {
|
||||
"character_id": char_id,
|
||||
"save_type": key.capitalize(),
|
||||
"dc": int(cd.group(1)) if cd else 15,
|
||||
}
|
||||
return "saving_throw", {
|
||||
"character_id": char_id,
|
||||
"save_type": "Tempra",
|
||||
"dc": int(cd.group(1)) if cd else 15,
|
||||
}
|
||||
|
||||
if any(k in a for k in ["abilità", "prova", "percepire", "scassinare", "nascondersi"]):
|
||||
return "ability_check", {
|
||||
"character_id": char_id,
|
||||
"dc": int(cd.group(1)) if cd else 12,
|
||||
}
|
||||
|
||||
if any(k in a for k in ["attacco", "attaccare", "colpire", "combattimento", "combat", "mischia"]):
|
||||
mod = re.search(r'([+-]\s*\d+)', a)
|
||||
ac = re.search(r'(?:ca|ac)\s*(\d+)', a)
|
||||
return "combat", {
|
||||
"character_id": char_id,
|
||||
"target_ac": int(ac.group(1)) if ac else 12,
|
||||
"attack_bonus": int(mod.group(1).replace(" ", "")) if mod else 0,
|
||||
}
|
||||
|
||||
if any(k in a for k in ["muovi", "muovo", "movimento", "esplora", "cammina", "cammino", "corri", "corro"]):
|
||||
dist = re.search(r'(\d+)\s*m', a)
|
||||
return "movement", {"distance_m": int(dist.group(1)) if dist else 10}
|
||||
|
||||
if any(k in a for k in ["incantesimo", "magia", "lancio", "spell"]):
|
||||
return "spellcasting", {
|
||||
"character_id": char_id,
|
||||
"spell_name": a,
|
||||
"spell_check": 10,
|
||||
}
|
||||
|
||||
if any(k in a for k in ["inventario", "prendi", "prendo", "lascia", "lascio", "usa", "uso", "raccogli", "raccolgo"]):
|
||||
return "update_inventory", {
|
||||
"character_id": char_id,
|
||||
"item": raw_action.split()[-1],
|
||||
"action": "add",
|
||||
}
|
||||
|
||||
if any(k in a for k in ["xp", "esperienza", "livello"]):
|
||||
return "award_xp", {"character_id": char_id, "xp": 5}
|
||||
|
||||
die_match = re.search(r'\bd(?:20|%|3|4|5|6|7|8|10|12|14|16|24|30)\b', a)
|
||||
if die_match:
|
||||
return "roll_dice", {"die": die_match.group()}
|
||||
|
||||
return "dm_override", {"command": "narrate", "params": {"text": raw_action}}
|
||||
+12
-44
@@ -1,62 +1,29 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import contextvars
|
||||
import json
|
||||
import re
|
||||
|
||||
from src.engine.config import load_from_env
|
||||
from src.engine.providers import LLMProvider, create_provider
|
||||
|
||||
DEFAULT_MODEL = "gemma2:9b"
|
||||
|
||||
_KNOWN_TOOLS = {
|
||||
"roll_dice": {"die": "str", "count": "int (default 1)", "modifier": "int (default 0)"},
|
||||
"dice_chain": {"die": "str", "steps": "int"},
|
||||
"combat": {"character_id": "int", "target_ac": "int", "attack_bonus": "int (default 0)",
|
||||
"damage_die": "str (default d8)", "damage_bonus": "int (default 0)",
|
||||
"mighty_deed_die": "str (default '')", "luck_mod": "int (default 0)",
|
||||
"armor_type": "str (none/light/medium/heavy)"},
|
||||
"ability_check": {"character_id": "int", "dc": "int", "modifier": "int (default 0)"},
|
||||
"saving_throw": {"character_id": "int", "save_type": "str (Tempra/Riflessi/Volontà)", "dc": "int", "modifier": "int (default 0)"},
|
||||
"movement": {"distance_m": "int", "terrain": "str (normale/difficile/molto difficile)", "stealth": "bool (default false)"},
|
||||
"spellcasting": {"character_id": "int", "spell_name": "str", "spell_check": "int", "dc": "int (default 12)"},
|
||||
"update_inventory": {"character_id": "int", "item": "str", "action": "str (add/remove)"},
|
||||
"award_xp": {"character_id": "int", "xp": "int"},
|
||||
"dm_override": {"command": "str", "params": "dict (opzionale)"},
|
||||
}
|
||||
|
||||
SYSTEM_PROMPT = f"""Sei il Dungeon Master di una partita di DCC RPG (Dungeon Crawl Classics).
|
||||
Il giocatore scrive un'azione e TU decidi quale strumento chiamare.
|
||||
|
||||
Strumenti disponibili:
|
||||
{json.dumps(_KNOWN_TOOLS, indent=2)}
|
||||
|
||||
Regole:
|
||||
1. Scegli SEMPRE UNO strumento per rispondere all'azione del giocatore.
|
||||
2. I parametri DEVONO essere realistici per il contesto.
|
||||
3. Se il giocatore fa qualcosa di narrativo, usa dm_override con command="narrate".
|
||||
4. Tira sempre i dadi quando il giocatore tenta un'azione rischiosa.
|
||||
5. Rispondi SOLO con JSON in questo formato:
|
||||
{{"tool": "nome_tool", "params": {{...}}, "narration": "breve descrizione di cosa succede"}}
|
||||
|
||||
|
||||
"""
|
||||
|
||||
_provider: LLMProvider | None = None
|
||||
_provider_var: contextvars.ContextVar[LLMProvider | None] = contextvars.ContextVar("provider", default=None)
|
||||
|
||||
|
||||
def get_provider() -> LLMProvider:
|
||||
global _provider
|
||||
if _provider is None:
|
||||
_provider = create_provider(load_from_env())
|
||||
return _provider
|
||||
provider = _provider_var.get()
|
||||
if provider is None:
|
||||
from src.engine.config import load_from_env
|
||||
provider = create_provider(load_from_env())
|
||||
_provider_var.set(provider)
|
||||
return provider
|
||||
|
||||
|
||||
def set_provider(provider: LLMProvider) -> None:
|
||||
global _provider
|
||||
_provider = provider
|
||||
_provider_var.set(provider)
|
||||
|
||||
|
||||
def ask(prompt: str, system: str = SYSTEM_PROMPT) -> str:
|
||||
def ask(prompt: str, system: str = "") -> str:
|
||||
return get_provider().ask(prompt, system=system)
|
||||
|
||||
|
||||
@@ -100,10 +67,11 @@ if __name__ == "__main__":
|
||||
"Attacco l'orco con la mia ascia!",
|
||||
"Esploro il corridoio buio",
|
||||
]
|
||||
from src.engine.prompts import SYSTEM_PROMPT as _DM_PROMPT
|
||||
for a in test_actions:
|
||||
print(f"\n>>> Giocatore: {a}")
|
||||
try:
|
||||
resp = ask(a)
|
||||
resp = ask(a, system=_DM_PROMPT)
|
||||
print(f"LLM: {resp[:300]}")
|
||||
parsed = parse_tool_call(resp)
|
||||
print(f"Parser: {json.dumps(parsed, indent=2)}")
|
||||
|
||||
@@ -4,7 +4,7 @@ import json
|
||||
from typing import Any
|
||||
|
||||
from src.engine import db as db_module
|
||||
from src.engine.state_machine import GameContext
|
||||
from src.engine.game_context import GameContext
|
||||
|
||||
_KNOWN_TOOLS: dict[str, dict[str, str]] = {
|
||||
"roll_dice": {"die": "tipo dado (d3..d30, d%)", "count": "numero dadi (default 1)", "modifier": "modificatore (default 0)"},
|
||||
|
||||
@@ -131,9 +131,9 @@ def create_provider(cfg: ProviderConfig | None = None) -> LLMProvider:
|
||||
if cfg.kind == "ollama":
|
||||
return cls(cfg.ollama_base_url, cfg.ollama_model, cfg.timeout)
|
||||
elif cfg.kind == "openrouter":
|
||||
key = cfg.openrouter_api_key or os.getenv("OPENROUTER_API_KEY", "")
|
||||
key = cfg.openrouter_api_key or os.getenv("DCC_OPENROUTER_KEY", os.getenv("OPENROUTER_API_KEY", ""))
|
||||
if not key:
|
||||
raise ValueError("OPENROUTER_API_KEY non configurata")
|
||||
raise ValueError("DCC_OPENROUTER_KEY non configurata")
|
||||
return cls(key, cfg.openrouter_model, cfg.timeout, cfg.openrouter_free_fallback)
|
||||
elif cfg.kind == "local":
|
||||
return cls(cfg.local_base_url, cfg.local_model, cfg.local_api_key, cfg.timeout)
|
||||
|
||||
+8
-155
@@ -1,160 +1,13 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any
|
||||
import warnings
|
||||
|
||||
from src.engine import db, tools
|
||||
from src.engine.game_context import GameContext
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Stato del gioco (immutabile dentro un turno)
|
||||
# ---------------------------------------------------------------------------
|
||||
warnings.warn(
|
||||
"state_machine module is deprecated. Import GameContext from src.engine.game_context instead.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class GameContext:
|
||||
campaign_id: int
|
||||
turn: int
|
||||
campaign: dict | None = None
|
||||
characters: list[dict] = field(default_factory=list)
|
||||
log: list[dict] = field(default_factory=list)
|
||||
quests: list[dict] = field(default_factory=list)
|
||||
areas: list[dict] = field(default_factory=list)
|
||||
|
||||
@classmethod
|
||||
def load(cls, campaign_id: int) -> GameContext:
|
||||
camp = db.get_campaign(campaign_id)
|
||||
if not camp:
|
||||
raise ValueError(f"Campagna #{campaign_id} non trovata.")
|
||||
chars = db.get_characters(campaign_id)
|
||||
log = db.get_log(campaign_id, limit=10)
|
||||
quests = db.get_quests(campaign_id)
|
||||
areas = db.get_area_state(campaign_id)
|
||||
return cls(
|
||||
campaign_id=campaign_id,
|
||||
turn=camp["turn"],
|
||||
campaign=camp,
|
||||
characters=chars,
|
||||
log=log,
|
||||
quests=quests,
|
||||
areas=areas,
|
||||
)
|
||||
|
||||
def summary(self) -> str:
|
||||
camp = self.campaign or {"name": "?", "turn": 0}
|
||||
lines = [f"Campagna: {camp['name']} | Turno {self.turn}"]
|
||||
for c in self.characters:
|
||||
cls = c.get("char_class") or ""
|
||||
lines.append(
|
||||
f" [{c['id']}] {c['name']} — Lv{c['level']} {cls} "
|
||||
f"PF {c['hp_current']}/{c['hp_max']} PE {c.get('xp', 0)}"
|
||||
)
|
||||
if self.quests:
|
||||
lines.append("Missioni:")
|
||||
for q in self.quests:
|
||||
lines.append(f" [{q['status']}] {q['name']}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Decisore — sceglie quale tool chiamare in base all'azione testuale
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
# Mappa euristica: parole chiave → tool
|
||||
def _parse_action(raw_action: str, ctx: GameContext) -> tuple[str, dict]:
|
||||
a = raw_action.lower().strip()
|
||||
cd = re.search(r'(?:cd|dc)\s*(\d+)', a)
|
||||
char_id = ctx.characters[0]["id"] if ctx.characters else 0
|
||||
|
||||
if any(k in a for k in ["tempra", "riflessi", "volontà", "salvezza"]):
|
||||
for key in ["tempra", "riflessi", "volontà"]:
|
||||
if key in a:
|
||||
return "saving_throw", {"character_id": char_id, "save_type": key.capitalize(), "dc": int(cd.group(1)) if cd else 15}
|
||||
return "saving_throw", {"character_id": char_id, "save_type": "Tempra", "dc": int(cd.group(1)) if cd else 15}
|
||||
|
||||
if any(k in a for k in ["abilità", "prova", "percepire", "scassinare", "nascondersi"]):
|
||||
return "ability_check", {"character_id": char_id, "dc": int(cd.group(1)) if cd else 12}
|
||||
|
||||
if any(k in a for k in ["attacco", "attaccare", "colpire", "combattimento", "combat", "mischia"]):
|
||||
mod = re.search(r'([+-]\s*\d+)', a)
|
||||
ac = re.search(r'(?:ca|ac)\s*(\d+)', a)
|
||||
return "combat", {
|
||||
"character_id": char_id,
|
||||
"target_ac": int(ac.group(1)) if ac else 12,
|
||||
"attack_bonus": int(mod.group(1).replace(" ", "")) if mod else 0,
|
||||
}
|
||||
|
||||
if any(k in a for k in ["muovi", "muovo", "movimento", "esplora", "cammina", "cammino", "corri", "corro"]):
|
||||
dist = re.search(r'(\d+)\s*m', a)
|
||||
return "movement", {"distance_m": int(dist.group(1)) if dist else 10}
|
||||
|
||||
if any(k in a for k in ["incantesimo", "magia", "lancio", "spell"]):
|
||||
return "spellcasting", {"character_id": char_id, "spell_name": a, "spell_check": 10}
|
||||
|
||||
if any(k in a for k in ["inventario", "prendi", "prendo", "lascia", "lascio", "usa", "uso", "raccogli", "raccolgo"]):
|
||||
return "update_inventory", {"character_id": char_id, "item": raw_action.split()[-1], "action": "add"}
|
||||
|
||||
if any(k in a for k in ["xp", "esperienza", "livello"]):
|
||||
return "award_xp", {"character_id": char_id, "xp": 5}
|
||||
|
||||
die_match = re.search(r'\bd(?:20|%|3|4|5|6|7|8|10|12|14|16|24|30)\b', a)
|
||||
if die_match:
|
||||
return "roll_dice", {"die": die_match.group()}
|
||||
|
||||
return "dm_override", {"command": "narrate", "params": {"text": raw_action}}
|
||||
|
||||
|
||||
def process_turn(campaign_id: int, raw_action: str) -> dict:
|
||||
ctx = GameContext.load(campaign_id)
|
||||
tool_name, tool_params = _parse_action(raw_action, ctx)
|
||||
|
||||
result = tools.run(tool_name, **tool_params)
|
||||
|
||||
turn = db.advance_turn(campaign_id)
|
||||
db.log_turn(campaign_id, turn, raw_action, result.get("message", ""))
|
||||
|
||||
return {
|
||||
"turn": turn,
|
||||
"action": raw_action,
|
||||
"tool": tool_name,
|
||||
"result": result,
|
||||
"context": ctx.summary(),
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
if __name__ == "__main__":
|
||||
conn = db.init_db()
|
||||
|
||||
camp_id = db.create_campaign("Avventura Test", "Tomba del Mago-Guerriero")
|
||||
char_id = db.create_character(camp_id, "Marco il Fidato", "guerriero", "soldato")
|
||||
db.update_character(char_id,
|
||||
level=1, str=15, agi=11, sta=14,
|
||||
per=9, int=8, lck=10,
|
||||
hp_current=12, hp_max=12, ac=15,
|
||||
armor_type="light"
|
||||
)
|
||||
|
||||
actions = [
|
||||
"Tiro un d20 per iniziativa",
|
||||
"Attacco il guardiano con la spada",
|
||||
"Prova di percezione CD 12",
|
||||
"Tiro salvezza su Tempra CD 15",
|
||||
"Mi muovo verso la porta",
|
||||
"Lancio l'incantesimo Dardo Incantato",
|
||||
"Prendo la pozione dalla mensola",
|
||||
"dm forzo la scena",
|
||||
]
|
||||
|
||||
for action in actions:
|
||||
print(f"\n>>> {action}")
|
||||
out = process_turn(camp_id, action)
|
||||
print(f" Tool: {out['tool']} | Msg: {out['result'].get('message', '?')[:100]}")
|
||||
|
||||
print(f"\n=== Cronologia ({len(actions)} turni) ===")
|
||||
ctx = GameContext.load(camp_id)
|
||||
print(ctx.summary())
|
||||
__all__ = ["GameContext"]
|
||||
+1
-1
@@ -12,7 +12,7 @@ from fastapi.staticfiles import StaticFiles
|
||||
from src.engine import db
|
||||
from src.engine.config import load_from_env
|
||||
from src.engine.dm import dm_cycle
|
||||
from src.engine.state_machine import GameContext
|
||||
from src.engine.game_context import GameContext
|
||||
|
||||
log = logging.getLogger(__name__)
|
||||
|
||||
|
||||
Reference in new issue
Block a user