diff --git a/scripts/farmhouse_decision_eval.py b/scripts/farmhouse_decision_eval.py new file mode 100644 index 0000000..b1afe56 --- /dev/null +++ b/scripts/farmhouse_decision_eval.py @@ -0,0 +1,75 @@ +"""How often does a decision model (Laya / Kev) pick the same action as the LLM did, on game states the app has already recorded? + + DECISION_URL=http://laya:8080 python scripts/farmhouse_decision_eval.py [decisions_dir] [model ...] + +Reads the brain's own decision logs (logs/decisions/*.jsonl: the snapshot and the LLM's answer), asks the decision model the same +question for each, and prints agreement overall, at several confidence levels, per action, the "always the most common action" +baseline, and the time per answer. The LLM is the teacher here, not the truth: this measures how well a fast model can stand in +for the slow one, not how well either plays. +""" +import collections +import glob +import importlib.util +import json +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +spec = importlib.util.spec_from_file_location("farmhouse_decide", os.path.join(HERE, "..", "src", "commander_ai_lab", "sim", "farmhouse_decide.py")) +D = importlib.util.module_from_spec(spec) +spec.loader.exec_module(D) + + +def load(dirpath): + rows = [] + for f in sorted(glob.glob(os.path.join(dirpath, "*.jsonl"))): + for line in open(f, encoding="utf-8"): + try: + r = json.loads(line) + except ValueError: + continue + if (r.get("decision") or {}).get("source") == "llm" and r.get("game_state") and r["decision"].get("action") in D.ACTION_HELP: + rows.append(r) + return rows + + +def main(): + dirpath = sys.argv[1] if len(sys.argv) > 1 else os.path.join(HERE, "..", "logs", "decisions") + models = sys.argv[2:] or [os.environ.get("DECISION_MODEL", "")] + rows = load(dirpath) + if not rows: + print("no recorded LLM decisions in", dirpath) + return 1 + truth = [r["decision"]["action"] for r in rows] + common, n_common = collections.Counter(truth).most_common(1)[0] + print("%d recorded decisions; LLM action mix: %s" % (len(rows), dict(collections.Counter(truth)))) + print("baseline 'always %s': %.1f%%\n" % (common, 100.0 * n_common / len(rows))) + for model in models: + os.environ["DECISION_MODEL"] = model + D.STATS.update({"asked": 0, "agreed": 0, "answered_conf": 0, "errors": 0, "ms_total": 0.0, "by_action": {}}) + got = [] + for r in rows: + a = D.ask(r["game_state"]) + got.append(a) + D.record(a, r["decision"]["action"]) + ok = [(g, t) for g, t in zip(got, truth) if g] + print("== model %r: answered %d of %d, %.0f ms each" % (model or "auto", len(ok), len(rows), D.stats()["avg_ms"] or 0)) + if not ok: + continue + print(" agrees with the LLM on %.1f%% (baseline %.1f%%)" % (100.0 * sum(g["choice"] == t for g, t in ok) / len(ok), 100.0 * n_common / len(rows))) + for lo in (0.5, 0.7, 0.85): + sel = [(g, t) for g, t in ok if g["confidence"] >= lo] + if sel: + print(" when it is >= %.2f sure (%d answers, %.0f%%): agrees %.1f%%" % ( + lo, len(sel), 100.0 * len(sel) / len(ok), 100.0 * sum(g["choice"] == t for g, t in sel) / len(sel))) + by = collections.defaultdict(lambda: [0, 0]) + for g, t in ok: + by[t][0] += 1 + by[t][1] += g["choice"] == t + print(" per LLM action:", {k: "%d/%d" % (v[1], v[0]) for k, v in sorted(by.items())}) + print(" it picked:", dict(collections.Counter(g["choice"] for g, t in ok))) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/commander_ai_lab/sim/deepseek_brain.py b/src/commander_ai_lab/sim/deepseek_brain.py index 048e10d..390a604 100644 --- a/src/commander_ai_lab/sim/deepseek_brain.py +++ b/src/commander_ai_lab/sim/deepseek_brain.py @@ -254,6 +254,7 @@ def build_game_state_snapshot( "my_life": me.life, "opp_life": opp.life, "my_mana_total": mana_total, + "land_play_available": getattr(me, "_lands_played_this_turn", 0) < getattr(me, "_land_plays_this_turn_limit", 1), "my_mana_by_color": mana_colors, "my_hand": my_hand, "my_board": my_board, @@ -268,6 +269,8 @@ def build_game_state_snapshot( "opp_graveyard": opp_gy, "my_library_count": cards_remaining, "opp_library_count": len(opp.library), + "opp_hand_count": len(opp.hand), + "stack_supported": False, "deck_drawn_pct": pct_drawn, } @@ -578,6 +581,11 @@ class DeepSeekBrain: if available_mana is not None: snapshot["my_mana_available"] = available_mana + from commander_ai_lab.sim.strategic_context import turn_context + advice = turn_context(snapshot, (deck_context or {}).get("strategic_preferences")) + snapshot["strategic_advice"] = advice + candidate_actions = advice["candidate_actions"] + if self.config.cache_enabled: cache_key = self._cache_key(snapshot) if cache_key in self._cache: @@ -587,6 +595,18 @@ class DeepSeekBrain: cached["latency_ms"] = round((time.time() - t_start) * 1000, 1) return cached + from commander_ai_lab.sim import farmhouse_decide as _fd # FARMHOUSE_DECISION + _fd_mode = _fd.mode() + if _fd_mode == "assist": + _fd_ans = _fd.ask(snapshot, actions=candidate_actions) + if (_fd_ans and _fd_ans["choice"] in candidate_actions + and _fd_ans["confidence"] >= float(os.environ.get("DECISION_MIN_CONF", "0.7"))): + result = _fd.as_decision(_fd_ans, t_start) + result["strategic_advice"] = advice + if self.config.log_decisions: + self._log_decision(snapshot, result, player_index) + return result + if not self._connected: return self._fallback_action(snapshot, t_start, "not_connected") @@ -595,7 +615,10 @@ class DeepSeekBrain: try: result = self._call_llm(snapshot) result["source"] = "llm" + if _fd_mode == "shadow": # FARMHOUSE_DECISION_SHADOW + _fd.record(_fd.ask(snapshot, actions=candidate_actions), result.get("action")) result["latency_ms"] = round((time.time() - t_start) * 1000, 1) + result["strategic_advice"] = advice if self.config.cache_enabled: self._put_cache(cache_key, result) @@ -882,4 +905,5 @@ class DeepSeekBrain: "api_base": self.config.api_base, "pending_log_entries": len(self._decision_log), "current_game_id": self._current_game_id, + "decision": __import__("commander_ai_lab.sim.farmhouse_decide", fromlist=["stats"]).stats(), # FARMHOUSE_DECISION_STATS } diff --git a/src/commander_ai_lab/sim/farmhouse_decide.py b/src/commander_ai_lab/sim/farmhouse_decide.py new file mode 100644 index 0000000..feb42e6 --- /dev/null +++ b/src/commander_ai_lab/sim/farmhouse_decide.py @@ -0,0 +1,158 @@ +"""A fast "which action?" answer from a decision model (Laya, or Kev: both serve POST /v1/systemone). + +The LLM brain answers the same question about 25 times a game, each taking seconds. A decision model reads a short text and picks ONE of +a fixed list of options in a single forward pass (tens of milliseconds on a GPU, about a second on a small CPU) and never writes text. +It cannot choose a target card; the engine's heuristics do that, exactly as when the LLM names no target. + +Modes (DECISION_MODE; with no DECISION_URL nothing here runs at all): + off do nothing + shadow ask it too and RECORD whether it agreed with the LLM; the game is played with the LLM's answer (the safe way to measure) + assist ask it first; use its answer when it is at least DECISION_MIN_CONF sure, otherwise ask the LLM as before + +Settings: DECISION_URL (e.g. http://laya:8080), DECISION_MODEL (a Laya checkpoint name such as english or typed-decisions), +DECISION_MIN_CONF (default 0.7, on the probability of the chosen option), DECISION_TIMEOUT (seconds, default 5), DECISION_API_KEY. +After three failures in a row the model is not asked again for a minute, so a down server costs the game nothing. +""" +import json +import os +import threading +import time +import urllib.request + +ACTION_HELP = { + "play_land": "play a land from hand (always do this first)", + "cast_creature": "cast a creature spell", + "cast_removal": "use removal on the opponent's best threat", + "cast_board_wipe": "wipe the board when behind on creatures", + "cast_ramp": "cast mana ramp to accelerate", + "cast_spell": "cast any other spell such as an enchantment or artifact", + "attack_all": "attack with all untapped creatures", + "attack_safe": "attack only with evasive or large creatures", + "hold": "pass and save mana", +} + +_lock = threading.Lock() +_fail = {"n": 0, "until": 0.0} +STATS = {"asked": 0, "agreed": 0, "answered_conf": 0, "errors": 0, "ms_total": 0.0, "by_action": {}} + + +def mode(): + m = os.environ.get("DECISION_MODE", "off").strip().lower() + return m if m in ("shadow", "assist") and os.environ.get("DECISION_URL") else "off" + + +def _card_text(c): + if isinstance(c, str): + return c + bits = [str(c.get("type") or "").strip()] + if c.get("cmc") is not None: + bits.append("cmc %s" % c["cmc"]) + if c.get("pt"): + bits.append(str(c["pt"])) + if c.get("castable") is not None: + bits.append("castable" if c["castable"] else "not castable") + roles = c.get("roles") + if roles: + bits.append("/".join(roles) if isinstance(roles, (list, tuple)) else str(roles)) + return "%s (%s)" % (c.get("name", "?"), ", ".join(b for b in bits if b)) + + +def compact_state(snap): + """The snapshot as a short paragraph (a decision model reads about 500 tokens, the LLM's indented JSON is two to three times that).""" + hand = snap.get("my_hand") or [] + mine = [c for c in (snap.get("my_board") or []) if isinstance(c, dict)] + theirs = [c for c in (snap.get("opp_board") or []) if isinstance(c, dict)] + my_creatures = [c for c in mine if "creature" in str(c.get("type", "")).lower()] + opp_creatures = [c for c in theirs if "creature" in str(c.get("type", "")).lower()] + threat = snap.get("opp_threat") or {} + cmdr = snap.get("commander") or {} + parts = ["Turn %s. Your life %s, opponent life %s. Mana available %s." % ( + snap.get("turn"), snap.get("my_life"), snap.get("opp_life"), snap.get("my_mana_total", snap.get("my_mana_available")))] + parts.append("Opponent threat %s, total power %s." % (threat.get("level", "?"), threat.get("total_power", "?"))) + if cmdr: + parts.append("Your commander %s is in the %s." % (cmdr.get("name", "?"), cmdr.get("zone", "?"))) + parts.append("Hand: %s." % ("; ".join(_card_text(c) for c in hand[:10]) if hand else "empty")) + parts.append("Your creatures: %s." % (", ".join(_card_text(c) for c in my_creatures[:8]) if my_creatures else "none")) + parts.append("Your other permanents: %d." % (len(mine) - len(my_creatures))) + parts.append("Opponent creatures: %s." % (", ".join(_card_text(c) for c in opp_creatures[:8]) if opp_creatures else "none")) + parts.append("Opponent other permanents: %d." % (len(theirs) - len(opp_creatures))) + advice = snap.get("strategic_advice") or {} + ranking = (advice.get("ranking") or {}).get("ranked") or [] + if ranking: + parts.append("CSIE relative ranking: %s." % "; ".join( + "%s %.2f (%s)" % (r["id"], r["score"], r["explanation"]) + for r in ranking[:5])) + if snap.get("stack_supported") and snap.get("stack"): + parts.append("Visible stack: %s." % json.dumps(snap["stack"], separators=(",", ":"))) + return " ".join(parts) + + +def ask(snap, actions=None): + """{'choice', 'confidence', 'probabilities', 'ms', 'model'} or None (not configured, backing off, or an error).""" + url = os.environ.get("DECISION_URL") + if not url: + return None + now = time.time() + with _lock: + if _fail["until"] > now: + return None + acts = [a for a in (actions or ACTION_HELP) if a in ACTION_HELP] + body = {"state": {"body": compact_state(snap)}, + "questions": {"action": {"type": "choice", "instructions": "Which action is best for this turn?", + "criteria": {a: ACTION_HELP[a] for a in acts}}}} + model = os.environ.get("DECISION_MODEL") + if model: + body["model"] = model + headers = {"content-type": "application/json"} + if os.environ.get("DECISION_API_KEY"): + headers["authorization"] = "Bearer " + os.environ["DECISION_API_KEY"] + t0 = time.time() + try: + req = urllib.request.Request(url.rstrip("/") + "/v1/systemone", data=json.dumps(body).encode(), headers=headers) + out = json.load(urllib.request.urlopen(req, timeout=float(os.environ.get("DECISION_TIMEOUT", "5")))) + ans = out["answers"]["action"] + probs = ans.get("probabilities") or {} + choice = ans["choice"] + if choice not in acts: + raise ValueError("Decision model selected an action outside the supplied candidates") + with _lock: + _fail["n"] = 0 + return {"choice": choice, "confidence": float(probs.get(choice, ans.get("answer_confidence", 0.0))), + "probabilities": probs, "ms": round((time.time() - t0) * 1000, 1), "model": model or "auto"} + except Exception: + with _lock: + _fail["n"] += 1 + STATS["errors"] += 1 + if _fail["n"] >= 3: + _fail["until"] = time.time() + 60 + _fail["n"] = 0 + return None + + +def as_decision(answer, t_start): + """The brain's own result dict for a decision model's answer (no target card: the engine's heuristics pick it).""" + return {"action": answer["choice"], "target_card": "", "source": "decision", + "reasoning": "decision model %s, p=%.2f" % (answer["model"], answer["confidence"]), + "latency_ms": round((time.time() - t_start) * 1000, 1)} + + +def record(answer, llm_action): + """Count one shadow comparison.""" + if not answer: + return + with _lock: + STATS["asked"] += 1 + STATS["ms_total"] += answer["ms"] + agree = answer["choice"] == llm_action + STATS["agreed"] += 1 if agree else 0 + if answer["confidence"] >= float(os.environ.get("DECISION_MIN_CONF", "0.7")): + STATS["answered_conf"] += 1 + row = STATS["by_action"].setdefault(llm_action, {"n": 0, "agreed": 0}) + row["n"] += 1 + row["agreed"] += 1 if agree else 0 + + +def stats(): + with _lock: + n = STATS["asked"] + return dict(STATS, agreement=round(STATS["agreed"] / n, 3) if n else None, avg_ms=round(STATS["ms_total"] / n, 1) if n else None) diff --git a/tests/farmhouse/test_farmhouse_decide.py b/tests/farmhouse/test_farmhouse_decide.py new file mode 100644 index 0000000..f695a65 --- /dev/null +++ b/tests/farmhouse/test_farmhouse_decide.py @@ -0,0 +1,111 @@ +"""Run: python -I -X utf8 tests/farmhouse/test_farmhouse_decide.py (offline: a stub /v1/systemone server on localhost)""" +import json +import os +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, HTTPServer + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +from _load import load # noqa: E402 + +D = load("farmhouse_decide", "src/commander_ai_lab/sim/farmhouse_decide.py") +FAILS = 0 + + +def ck(name, cond, detail=""): + global FAILS + print(" %s %s%s" % ("ok " if cond else "FAIL", name, (" -- " + str(detail)[:200]) if detail and not cond else "")) + if not cond: + FAILS += 1 + + +SEEN = [] +BEHAVIOUR = {"mode": "ok"} + + +class Stub(BaseHTTPRequestHandler): + def log_message(self, *a): + pass + + def do_POST(self): + body = json.loads(self.rfile.read(int(self.headers.get("Content-Length") or 0))) + SEEN.append((self.path, body, self.headers.get("authorization"))) + if BEHAVIOUR["mode"] == "down": + self.send_response(500) + self.end_headers() + return + crit = list(body["questions"]["action"]["criteria"]) + pick = "cast_creature" if "cast_creature" in crit else crit[0] + probs = {k: (0.8 if k == pick else 0.2 / max(len(crit) - 1, 1)) for k in crit} + out = json.dumps({"answers": {"action": {"type": "choice", "choice": pick, "probabilities": probs, "confidence": 0.5}}}).encode() + self.send_response(200) + self.send_header("content-type", "application/json") + self.send_header("content-length", str(len(out))) + self.end_headers() + self.wfile.write(out) + + +srv = HTTPServer(("127.0.0.1", 0), Stub) +threading.Thread(target=srv.serve_forever, daemon=True).start() +URL = "http://127.0.0.1:%d" % srv.server_port + +SNAP = {"turn": 4, "my_life": 31, "opp_life": 20, "my_mana_total": 4, + "my_hand": [{"name": "Forest", "type": "land", "cmc": 0, "castable": True}, + {"name": "Llanowar Elves", "type": "creature", "cmc": 1, "castable": True, "roles": ["ramp"]}, "Sol Ring"], + "my_board": [{"name": "Bear", "type": "creature", "pt": "2/2"}, {"name": "Rock", "type": "artifact"}], + "opp_board": [{"name": "Wurm", "type": "creature", "pt": "6/6"}], + "opp_threat": {"level": "HIGH", "total_power": 6}, "commander": {"name": "Atraxa", "zone": "command zone"}} + +# --- the text it reads +t = D.compact_state(SNAP) +ck("the text has the facts the choice depends on", all(x in t for x in ("Turn 4", "life 31", "Llanowar Elves", "Bear (creature, 2/2)", "Wurm", "HIGH")), t) +ck("short enough for a 512-token reader", len(D.compact_state(dict(SNAP, my_hand=[{"name": "Card %d" % i, "type": "creature", "cmc": 3} for i in range(40)]))) < 1900) +ck("an empty snapshot does not raise", isinstance(D.compact_state({}), str)) + +# --- modes +for k in ("DECISION_URL", "DECISION_MODE", "DECISION_MODEL", "DECISION_API_KEY"): + os.environ.pop(k, None) +ck("no URL: off, and ask() does nothing", D.mode() == "off" and D.ask(SNAP) is None and not SEEN) +os.environ["DECISION_URL"] = URL +ck("a URL alone is still off (the mode must be chosen)", D.mode() == "off") +os.environ["DECISION_MODE"] = "shadow" +ck("shadow", D.mode() == "shadow") +os.environ["DECISION_MODE"] = "assist" +ck("assist", D.mode() == "assist") +os.environ["DECISION_MODE"] = "bogus" +ck("an unknown mode is off", D.mode() == "off") +os.environ["DECISION_MODE"] = "shadow" + +# --- asking +os.environ["DECISION_MODEL"] = "typed-decisions" +os.environ["DECISION_API_KEY"] = "k" +a = D.ask(SNAP, actions=["play_land", "cast_creature", "hold", "not_an_action"]) +path, body, auth = SEEN[-1] +ck("it posts to /v1/systemone with the model, key and only the known actions", + path == "/v1/systemone" and body["model"] == "typed-decisions" and auth == "Bearer k" + and list(body["questions"]["action"]["criteria"]) == ["play_land", "cast_creature", "hold"], (path, body, auth)) +ck("the answer carries the choice, its probability and timing", a and a["choice"] == "cast_creature" and abs(a["confidence"] - 0.8) < 1e-9 and a["ms"] >= 0, a) +d = D.as_decision(a, time.time()) +ck("as_decision is a brain result with no target (the engine picks one)", d["action"] == "cast_creature" and d["target_card"] == "" and d["source"] == "decision", d) + +# --- shadow bookkeeping +D.record(a, "cast_creature") +D.record(a, "hold") +D.record(None, "hold") +s = D.stats() +ck("agreement is counted over answered questions only", s["asked"] == 2 and s["agreed"] == 1 and s["agreement"] == 0.5 and s["answered_conf"] == 2, s) +ck("per-action tallies", s["by_action"]["hold"] == {"n": 1, "agreed": 0} and s["by_action"]["cast_creature"]["agreed"] == 1) + +# --- a server that is down costs nothing after three failures +BEHAVIOUR["mode"] = "down" +n0 = len(SEEN) +res = [D.ask(SNAP) for _ in range(6)] +ck("errors give None, never raise", all(r is None for r in res)) +ck("after three failures it stops asking for a while", len(SEEN) - n0 == 3, len(SEEN) - n0) +BEHAVIOUR["mode"] = "ok" +ck("still backing off even though the server is back", D.ask(SNAP) is None) +D._fail["until"] = 0 +ck("and asks again once the pause ends", D.ask(SNAP) is not None) +print("\nFAILURES: %d" % FAILS) +sys.exit(1 if FAILS else 0)