345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""
|
|
update_weights.py
|
|
=================
|
|
Outcome-weighted reinforcement learning for Commander AI card weights.
|
|
|
|
Reads all decisions_*.jsonl files produced by DeepSeekBrain.flush_log(),
|
|
applies a gradient-free reward nudge to each weight key based on the
|
|
normalised reward signal (-1.0 to +1.0) stamped on every decision entry,
|
|
then saves the updated weights to learned_weights.json via save_weights().
|
|
|
|
Usage
|
|
-----
|
|
# Basic run (reads logs/decisions/, updates learned_weights.json)
|
|
python -m ml.scripts.update_weights
|
|
|
|
# Custom paths
|
|
python -m ml.scripts.update_weights \\
|
|
--log-dir logs/decisions \\
|
|
--weights-out src/commander_ai_lab/sim/learned_weights.json \\
|
|
--lr 0.05 \\
|
|
--min-games 10
|
|
|
|
# Dry-run (print proposed changes, don't write)
|
|
python -m ml.scripts.update_weights --dry-run
|
|
|
|
# Reset to defaults then exit
|
|
python -m ml.scripts.update_weights --reset
|
|
|
|
Algorithm
|
|
---------
|
|
For each decision entry that has a non-null game_result:
|
|
|
|
delta = learning_rate * reward * action_weight_for(action)
|
|
|
|
Each weight key is nudged by the mean delta across all decisions that
|
|
touched that key. Keys with no decisions are left unchanged.
|
|
All values are clamped to [-20.0, 20.0] by save_weights().
|
|
|
|
Reward is the normalised life-delta from flush_log():
|
|
reward = (my_final_life - avg_opp_final_life) / 40.0 in [-1.0, 1.0]
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Iterator
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Allow running as a script without installing the package
|
|
# ---------------------------------------------------------------------------
|
|
_REPO_ROOT = Path(__file__).resolve().parents[2]
|
|
if str(_REPO_ROOT / "src") not in sys.path:
|
|
sys.path.insert(0, str(_REPO_ROOT / "src"))
|
|
|
|
from commander_ai_lab.sim.rules import (
|
|
AI_DEFAULT_WEIGHTS,
|
|
load_weights,
|
|
reset_weights,
|
|
save_weights,
|
|
)
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)-8s %(message)s",
|
|
datefmt="%H:%M:%S",
|
|
)
|
|
logger = logging.getLogger("update_weights")
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Action → weight key mapping
|
|
# Maps each VALID_ACTION to the weight(s) it most directly influences.
|
|
# A single action may touch multiple keys (listed as a tuple).
|
|
# ---------------------------------------------------------------------------
|
|
ACTION_WEIGHT_MAP: dict[str, tuple[str, ...]] = {
|
|
"play_land": (), # no weight key — always correct
|
|
"cast_ramp": ("spell_ramp",),
|
|
"cast_removal": ("spell_destroy",),
|
|
"cast_board_wipe": ("spell_counter",), # reuse counter as board-control proxy
|
|
"cast_creature": ("card_ptBonus",),
|
|
"cast_spell": ("spell_enchantment", "spell_draw"),
|
|
"attack_all": ("kw_haste", "kw_flying"), # aggression / evasion proxy
|
|
"attack_safe": ("kw_flying", "kw_trample"),
|
|
"hold": (), # no weight key — passing is neutral
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# JSONL helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _iter_log_entries(log_dir: str) -> Iterator[dict]:
|
|
"""Yield every entry from all decisions_*.jsonl files in log_dir."""
|
|
log_path = Path(log_dir)
|
|
if not log_path.exists():
|
|
logger.warning("Log directory not found: %s", log_dir)
|
|
return
|
|
|
|
files = sorted(log_path.glob("decisions_*.jsonl"))
|
|
if not files:
|
|
logger.warning("No decisions_*.jsonl files found in %s", log_dir)
|
|
return
|
|
|
|
logger.info("Found %d log file(s) in %s", len(files), log_dir)
|
|
for fpath in files:
|
|
with open(fpath, encoding="utf-8") as f:
|
|
for lineno, line in enumerate(f, 1):
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
yield json.loads(line)
|
|
except json.JSONDecodeError as exc:
|
|
logger.warning("Skipping malformed line %d in %s: %s", lineno, fpath.name, exc)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Core learning loop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def compute_weight_updates(
|
|
log_dir: str,
|
|
current_weights: dict[str, float],
|
|
learning_rate: float = 0.05,
|
|
min_games: int = 1,
|
|
) -> tuple[dict[str, float], dict]:
|
|
"""
|
|
Read decision logs and compute updated weights.
|
|
|
|
Returns
|
|
-------
|
|
updated_weights : dict[str, float]
|
|
New weight dict after applying mean reward nudges.
|
|
stats : dict
|
|
Diagnostic info (games seen, decisions processed, keys changed, etc.)
|
|
"""
|
|
# Accumulators: key → list of (reward * lr) deltas
|
|
key_deltas: dict[str, list[float]] = defaultdict(list)
|
|
|
|
total_entries = 0
|
|
labelled_entries = 0 # entries with non-null game_result
|
|
game_ids_seen: set[str] = set()
|
|
action_counts: dict[str, int] = defaultdict(int)
|
|
|
|
for entry in _iter_log_entries(log_dir):
|
|
total_entries += 1
|
|
|
|
game_result = entry.get("game_result")
|
|
if not game_result:
|
|
continue # no outcome stamped — skip (pre-flush or incomplete game)
|
|
|
|
labelled_entries += 1
|
|
game_ids_seen.add(entry.get("game_id", "unknown"))
|
|
|
|
reward: float = float(game_result.get("reward", 0.0))
|
|
action: str = entry.get("decision", {}).get("action", "hold")
|
|
action_counts[action] += 1
|
|
|
|
weight_keys = ACTION_WEIGHT_MAP.get(action, ())
|
|
for key in weight_keys:
|
|
if key in current_weights:
|
|
key_deltas[key].append(learning_rate * reward)
|
|
|
|
num_games = len(game_ids_seen)
|
|
logger.info(
|
|
"Processed %d entries | %d labelled | %d unique games",
|
|
total_entries, labelled_entries, num_games,
|
|
)
|
|
|
|
if num_games < min_games:
|
|
logger.warning(
|
|
"Only %d game(s) found — minimum is %d. Skipping update.",
|
|
num_games, min_games,
|
|
)
|
|
return dict(current_weights), {
|
|
"skipped": True,
|
|
"reason": f"need {min_games} games, have {num_games}",
|
|
"total_entries": total_entries,
|
|
"labelled_entries": labelled_entries,
|
|
"games": num_games,
|
|
}
|
|
|
|
# Apply mean delta per key
|
|
updated = dict(current_weights)
|
|
changed_keys: dict[str, tuple[float, float]] = {} # key → (old, new)
|
|
|
|
for key, deltas in key_deltas.items():
|
|
if not deltas:
|
|
continue
|
|
mean_delta = sum(deltas) / len(deltas)
|
|
old_val = updated[key]
|
|
new_val = round(old_val + mean_delta, 6)
|
|
updated[key] = new_val
|
|
if abs(new_val - old_val) >= 0.0001:
|
|
changed_keys[key] = (old_val, new_val)
|
|
|
|
stats = {
|
|
"skipped": False,
|
|
"total_entries": total_entries,
|
|
"labelled_entries": labelled_entries,
|
|
"games": num_games,
|
|
"keys_changed": len(changed_keys),
|
|
"changes": changed_keys,
|
|
"action_counts": dict(action_counts),
|
|
}
|
|
return updated, stats
|
|
|
|
|
|
def print_diff(stats: dict, current: dict, updated: dict) -> None:
|
|
"""Pretty-print weight changes to stdout."""
|
|
changes = stats.get("changes", {})
|
|
if not changes:
|
|
print(" (no weight changes above threshold)")
|
|
return
|
|
|
|
col_w = max(len(k) for k in changes) + 2
|
|
print(f" {'Key':<{col_w}} {'Before':>10} {'After':>10} {'Delta':>10}")
|
|
print(" " + "-" * (col_w + 36))
|
|
for key, (old, new) in sorted(changes.items(), key=lambda x: abs(x[1][1] - x[1][0]), reverse=True):
|
|
delta = new - old
|
|
arrow = "▲" if delta > 0 else "▼"
|
|
print(f" {key:<{col_w}} {old:>10.4f} {new:>10.4f} {arrow} {abs(delta):>8.4f}")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _default_log_dir() -> str:
|
|
return str(_REPO_ROOT / "logs" / "decisions")
|
|
|
|
|
|
def _default_weights_out() -> str:
|
|
return str(_REPO_ROOT / "src" / "commander_ai_lab" / "sim" / "learned_weights.json")
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Update AI card weights from simulation decision logs."
|
|
)
|
|
parser.add_argument(
|
|
"--log-dir",
|
|
default=_default_log_dir(),
|
|
help="Directory containing decisions_*.jsonl files (default: logs/decisions/)",
|
|
)
|
|
parser.add_argument(
|
|
"--weights-out",
|
|
default=_default_weights_out(),
|
|
help="Path to write learned_weights.json (default: src/.../sim/learned_weights.json)",
|
|
)
|
|
parser.add_argument(
|
|
"--lr",
|
|
type=float,
|
|
default=0.05,
|
|
help="Learning rate — how much each game nudges a weight (default: 0.05)",
|
|
)
|
|
parser.add_argument(
|
|
"--min-games",
|
|
type=int,
|
|
default=10,
|
|
help="Minimum number of completed games required before updating (default: 10)",
|
|
)
|
|
parser.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="Print proposed changes without writing learned_weights.json",
|
|
)
|
|
parser.add_argument(
|
|
"--reset",
|
|
action="store_true",
|
|
help="Delete learned_weights.json and restore AI_DEFAULT_WEIGHTS, then exit",
|
|
)
|
|
parser.add_argument(
|
|
"--verbose",
|
|
action="store_true",
|
|
help="Show action distribution and per-key delta counts",
|
|
)
|
|
args = parser.parse_args(argv)
|
|
|
|
# ── Reset mode ──────────────────────────────────────────────
|
|
if args.reset:
|
|
reset_weights(args.weights_out)
|
|
print(f"Reset: deleted {args.weights_out} (if it existed). Defaults restored.")
|
|
return 0
|
|
|
|
# ── Load current weights ─────────────────────────────────────
|
|
current = load_weights(args.weights_out)
|
|
logger.info("Loaded %d weight keys from %s", len(current), args.weights_out)
|
|
|
|
# ── Compute updates ──────────────────────────────────────────
|
|
updated, stats = compute_weight_updates(
|
|
log_dir=args.log_dir,
|
|
current_weights=current,
|
|
learning_rate=args.lr,
|
|
min_games=args.min_games,
|
|
)
|
|
|
|
# ── Report ───────────────────────────────────────────────────
|
|
print()
|
|
print("=" * 60)
|
|
print(" Commander AI — Weight Update Report")
|
|
print("=" * 60)
|
|
print(f" Log dir : {args.log_dir}")
|
|
print(f" Weights out: {args.weights_out}")
|
|
print(f" Games : {stats['games']}")
|
|
print(f" Entries : {stats['total_entries']} ({stats['labelled_entries']} labelled)")
|
|
print(f" LR : {args.lr}")
|
|
print()
|
|
|
|
if stats.get("skipped"):
|
|
print(f" SKIPPED — {stats['reason']}")
|
|
print()
|
|
return 0
|
|
|
|
print(f" Keys changed: {stats['keys_changed']}")
|
|
print()
|
|
print_diff(stats, current, updated)
|
|
print()
|
|
|
|
if args.verbose:
|
|
print(" Action distribution:")
|
|
for action, count in sorted(stats["action_counts"].items(), key=lambda x: -x[1]):
|
|
print(f" {action:<20} {count:>6} decisions")
|
|
print()
|
|
|
|
# ── Write ────────────────────────────────────────────────────
|
|
if args.dry_run:
|
|
print(" DRY RUN — no file written.")
|
|
else:
|
|
out_path = save_weights(updated, args.weights_out)
|
|
print(f" Saved updated weights to: {out_path}")
|
|
|
|
print("=" * 60)
|
|
print()
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|