| |
| """Evaluate L2 skill decisions from belief states. |
| |
| The first runnable backends are deterministic baselines. The Gemma backend is a |
| strict-JSON wrapper scaffold; it is intentionally gated behind an explicit model |
| path/endpoint so experiments remain reproducible. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import re |
| import argparse |
| import copy |
| import json |
| import math |
| import os |
| import random |
| import sys |
| import time |
| import urllib.request |
| from collections import Counter, defaultdict |
| from typing import Any, Dict, List, Optional |
|
|
| sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "src")) |
| from layered_belief import ( |
| BOSSES, |
| CATEGORY, |
| build_split_manifest, |
| collect_decision_samples, |
| js_divergence, |
| macro_recall, |
| normalized_entropy, |
| skill_grounded, |
| ) |
|
|
|
|
| def choose_majority(sample: Dict[str, Any], train_majority: Dict[str, str]) -> Dict[str, Any]: |
| boss = sample["boss"] |
| skill = train_majority[boss] |
| return {"skill": skill, "reason": "train-majority skill", "confidence": 1.0} |
|
|
|
|
| def choose_random(sample: Dict[str, Any], rng: random.Random) -> Dict[str, Any]: |
| ready = sample["belief"]["cooldown_ready"] |
| legal = [s for s in sample["legal_skills"] if ready.get(s, True)] |
| skill = rng.choice(legal or sample["legal_skills"]) |
| return {"skill": skill, "reason": "uniform random legal-ready skill", "confidence": 1.0} |
|
|
|
|
| def choose_heuristic(sample: Dict[str, Any]) -> Dict[str, Any]: |
| belief = sample["belief"] |
| ready = belief["cooldown_ready"] |
| dist = belief["player_distance_bin"] |
| style = str(belief.get("director_style", "") or "").lower() |
| legal = [s for s in sample["legal_skills"] if ready.get(s, True)] |
| if not legal: |
| legal = list(sample["legal_skills"]) |
| if any(x in style for x in ("keep_distance", "kite", "defensive", "cautious", "sniper")): |
| preferred_cat = "far" |
| elif "aggressive" in style or "pressure" in style: |
| preferred_cat = "close" |
| else: |
| preferred_cat = "far" if dist == "far" else "close" |
| for s in legal: |
| if CATEGORY.get(s) == preferred_cat: |
| return {"skill": s, "reason": f"{dist} distance -> {preferred_cat} skill", "confidence": 0.7} |
| for s in legal: |
| if CATEGORY.get(s) in {"cd_aoe", "summon"}: |
| return {"skill": s, "reason": "no preferred range skill; use ready special skill", "confidence": 0.55} |
| return {"skill": legal[0], "reason": "fallback first legal-ready skill", "confidence": 0.4} |
|
|
|
|
| def extract_json_object(text: str) -> Dict[str, Any]: |
| text = text.strip() |
| try: |
| return json.loads(text) |
| except json.JSONDecodeError: |
| pass |
| start = text.find("{") |
| end = text.rfind("}") |
| if start >= 0 and end > start: |
| return json.loads(text[start:end + 1]) |
| raise ValueError(f"no JSON object found in model output: {text[:200]!r}") |
|
|
|
|
| _HF_MODEL: Any = None |
| _HF_TOKENIZER: Any = None |
|
|
|
|
| def _is_hp_key(k: Any) -> bool: |
| s = str(k).lower() |
| return ("hp_phase" in s) or ("hp_frac" in s) or s in {"hp", "your_hp"} |
|
|
|
|
| _HP_TEXT = re.compile(r"\s*,?\s*HP[ _]phase\s*,?", re.IGNORECASE) |
|
|
|
|
| def _scrub_hp_text(s: str) -> str: |
| |
| out = _HP_TEXT.sub(", ", s) |
| return out.replace(" ,", ",").replace(", ,", ",").replace(",,", ",") |
|
|
|
|
| def _strip_hp(obj: Any) -> Any: |
| """hp_phase 已从 L2 全面移除（L1 无 hp 头、闭环无来源）。递归剔除任何 hp 相关键 |
| （hp_phase / your_hp_phase / *_hp_frac 等）并擦掉字符串值里的 HP phase 提及， |
| 保证喂给模型的 prompt 里绝不含 hp。""" |
| if isinstance(obj, dict): |
| return {k: _strip_hp(v) for k, v in obj.items() if not _is_hp_key(k)} |
| if isinstance(obj, list): |
| return [_strip_hp(v) for v in obj] |
| if isinstance(obj, str): |
| return _scrub_hp_text(obj) |
| return obj |
|
|
|
|
| def build_gemma_prompt(sample: Dict[str, Any], feedback: str | None = None) -> str: |
| if sample.get("l2_prompt_payload"): |
| payload = _strip_hp(copy.deepcopy(sample["l2_prompt_payload"])) |
| if feedback: |
| payload["previous_error"] = feedback |
| payload["repair_instruction"] = "Return a valid JSON object only. Do not include markdown." |
| return json.dumps(payload, ensure_ascii=False) |
| belief = _strip_hp(sample["belief"]) |
| legal = sample["legal_skills"] |
| ready = belief["cooldown_ready"] |
| director_style = belief.get("director_style") |
| payload = { |
| "task": "Choose the next boss skill for a game NPC.", |
| "constraints": { |
| "output_json_only": True, |
| "schema": {"skill": "string", "reason": "string", "confidence": "number"}, |
| "legal_skills": legal, |
| "cooldown_ready": ready, |
| "confidence_range": [0.0, 1.0], |
| }, |
| "belief": belief, |
| "director_style": director_style, |
| "instruction": ( |
| "Pick exactly one skill from legal_skills. Prefer a skill that is ready, grounded in " |
| "the distance/angle/cooldown belief, and plausible for this boss. Return only JSON." |
| ), |
| } |
| if feedback: |
| payload["previous_error"] = feedback |
| payload["repair_instruction"] = "Return a valid JSON object only. Do not include markdown." |
| return json.dumps(payload, ensure_ascii=False) |
|
|
|
|
| def gemma_openai_payload(user_content: Any, max_tokens: int = 256) -> Dict[str, Any]: |
| payload = { |
| "model": os.environ.get("GEMMA_MODEL", "gemma4-e2b-it"), |
| "temperature": float(os.environ.get("GEMMA_TEMPERATURE", "0")), |
| "max_tokens": int(max_tokens), |
| "messages": [ |
| { |
| "role": "system", |
| "content": "You are a strict JSON decision module for a game boss. Output JSON only." |
| }, |
| {"role": "user", "content": user_content} |
| ] |
| } |
| if os.environ.get("GEMMA_RESPONSE_FORMAT", "1") != "0": |
| payload["response_format"] = {"type": "json_object"} |
| return payload |
|
|
|
|
| def call_openai_compatible(prompt: str) -> str: |
| base = os.environ.get("GEMMA_OPENAI_BASE_URL") |
| if not base: |
| raise RuntimeError( |
| "OpenAI-compatible Gemma backend requires GEMMA_OPENAI_BASE_URL. " |
| "Example: http://127.0.0.1:8000/v1" |
| ) |
| api_key = os.environ.get("GEMMA_API_KEY", "EMPTY") |
| payload = gemma_openai_payload(prompt) |
| req = urllib.request.Request( |
| base.rstrip("/") + "/chat/completions", |
| data=json.dumps(payload).encode("utf-8"), |
| headers={ |
| "Content-Type": "application/json", |
| "Authorization": f"Bearer {api_key}" |
| }, |
| method="POST" |
| ) |
| |
| timeout = float(os.environ.get("GEMMA_TIMEOUT", "600")) |
| attempts = int(os.environ.get("GEMMA_RETRIES", "4")) |
| last_exc = None |
| for attempt in range(attempts): |
| try: |
| with urllib.request.urlopen(req, timeout=timeout) as resp: |
| data = json.loads(resp.read().decode("utf-8")) |
| return data["choices"][0]["message"]["content"] |
| except Exception as exc: |
| last_exc = exc |
| if attempt < attempts - 1: |
| time.sleep(2.0 * (attempt + 1)) |
| raise RuntimeError(f"gemma endpoint failed after {attempts} attempts: {last_exc}") |
|
|
|
|
| def call_hf_local(prompt: str) -> str: |
| global _HF_MODEL, _HF_TOKENIZER |
| model_path = os.environ.get("GEMMA_MODEL_PATH") |
| if not model_path: |
| raise RuntimeError("HF Gemma backend requires GEMMA_MODEL_PATH to point to a local model directory or HF id") |
| if _HF_MODEL is None or _HF_TOKENIZER is None: |
| import torch |
| from transformers import AutoModelForCausalLM, AutoTokenizer |
|
|
| dtype_name = os.environ.get("GEMMA_TORCH_DTYPE", "auto") |
| dtype = "auto" |
| if dtype_name == "bfloat16": |
| dtype = torch.bfloat16 |
| elif dtype_name == "float16": |
| dtype = torch.float16 |
| _HF_TOKENIZER = AutoTokenizer.from_pretrained( |
| model_path, |
| trust_remote_code=os.environ.get("GEMMA_TRUST_REMOTE_CODE", "0") == "1", |
| ) |
| _HF_MODEL = AutoModelForCausalLM.from_pretrained( |
| model_path, |
| device_map=os.environ.get("GEMMA_DEVICE_MAP", "auto"), |
| torch_dtype=dtype, |
| trust_remote_code=os.environ.get("GEMMA_TRUST_REMOTE_CODE", "0") == "1", |
| ) |
| _HF_MODEL.eval() |
| messages = [ |
| {"role": "system", "content": "You are a strict JSON decision module for a game boss. Output JSON only."}, |
| {"role": "user", "content": prompt}, |
| ] |
| if hasattr(_HF_TOKENIZER, "apply_chat_template") and _HF_TOKENIZER.chat_template: |
| text = _HF_TOKENIZER.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) |
| else: |
| text = messages[0]["content"] + "\n" + messages[1]["content"] + "\nJSON:" |
| inputs = _HF_TOKENIZER(text, return_tensors="pt") |
| inputs = {k: v.to(_HF_MODEL.device) for k, v in inputs.items()} |
| temperature = float(os.environ.get("GEMMA_TEMPERATURE", "0")) |
| with __import__("torch").no_grad(): |
| out = _HF_MODEL.generate( |
| **inputs, |
| max_new_tokens=int(os.environ.get("GEMMA_MAX_NEW_TOKENS", "192")), |
| do_sample=temperature > 0, |
| temperature=max(temperature, 1e-6), |
| top_p=float(os.environ.get("GEMMA_TOP_P", "0.95")), |
| pad_token_id=_HF_TOKENIZER.eos_token_id, |
| ) |
| return _HF_TOKENIZER.decode(out[0, inputs["input_ids"].shape[1]:], skip_special_tokens=True) |
|
|
|
|
| def call_gemma_model(prompt: str) -> str: |
| if os.environ.get("GEMMA_OPENAI_BASE_URL"): |
| return call_openai_compatible(prompt) |
| if os.environ.get("GEMMA_MODEL_PATH"): |
| return call_hf_local(prompt) |
| raise RuntimeError( |
| "Gemma backend requires either GEMMA_OPENAI_BASE_URL for an OpenAI-compatible endpoint " |
| "or GEMMA_MODEL_PATH for a local transformers model. Neither is set." |
| ) |
|
|
|
|
| def normalize_decision(out: Dict[str, Any]) -> Dict[str, Any]: |
| skill = out.get("skill") |
| reason = out.get("reason", "") |
| confidence = out.get("confidence", 0.0) |
| try: |
| confidence = float(confidence) |
| except (TypeError, ValueError): |
| confidence = 0.0 |
| if not math.isfinite(confidence): |
| confidence = 0.0 |
| return { |
| "skill": skill, |
| "reason": str(reason), |
| "confidence": max(0.0, min(1.0, confidence)), |
| } |
|
|
|
|
| def choose_gemma(sample: Dict[str, Any]) -> Dict[str, Any]: |
| feedback: Optional[str] = None |
| last_raw = "" |
| last_error = "" |
| retries = int(os.environ.get("GEMMA_PARSE_RETRIES", "1")) |
| for attempt in range(retries + 1): |
| prompt = build_gemma_prompt(sample, feedback) |
| raw = call_gemma_model(prompt) |
| last_raw = raw |
| try: |
| decision = normalize_decision(extract_json_object(raw)) |
| decision["raw_output"] = raw |
| if os.environ.get("GEMMA_LOG_PROMPT", "1") != "0": |
| decision["raw_prompt"] = prompt |
| decision["parse_attempts"] = attempt + 1 |
| return decision |
| except Exception as exc: |
| last_error = str(exc) |
| feedback = f"Attempt {attempt + 1} failed: {last_error}. Raw output: {raw[:500]}" |
| return { |
| "skill": None, |
| "reason": "", |
| "confidence": 0.0, |
| "raw_output": last_raw, |
| "parse_error": last_error or "unknown parse error", |
| "parse_attempts": retries + 1, |
| } |
|
|
|
|
| def decide( |
| sample: Dict[str, Any], |
| backend: str, |
| majority: Dict[str, str], |
| rng: random.Random, |
| ) -> Dict[str, Any]: |
| if backend == "oracle": |
| return {"skill": sample["target_skill"], "reason": "oracle engine target", "confidence": 1.0} |
| if backend == "majority": |
| return choose_majority(sample, majority) |
| if backend == "random": |
| return choose_random(sample, rng) |
| if backend == "heuristic": |
| return choose_heuristic(sample) |
| if backend == "gemma": |
| return choose_gemma(sample) |
| raise ValueError(backend) |
|
|
|
|
| def train_majority(samples: List[Dict[str, Any]]) -> Dict[str, str]: |
| by_boss = defaultdict(Counter) |
| for s in samples: |
| by_boss[s["boss"]][s["target_skill"]] += 1 |
| return {boss: ctr.most_common(1)[0][0] for boss, ctr in by_boss.items()} |
|
|
|
|
| def flatten_nested_l1_belief(row: Dict[str, Any]) -> Dict[str, Any]: |
| belief = row["l1_belief"] |
| geom = belief.get("geometry", {}) or {} |
| player = belief.get("player_state", {}) or {} |
| boss = belief.get("boss_state", {}) or {} |
| resource = belief.get("resource_state", {}) or {} |
| tactical = belief.get("tactical_state", {}) or {} |
| confidence = belief.get("confidence", {}) or {} |
| return { |
| "boss_id": row["boss"], |
| "fight": row.get("fight"), |
| "index": row.get("index"), |
| "player_distance_bin": geom.get("distance_bin"), |
| "player_distance_value": geom.get("distance_value"), |
| "dp_bin": geom.get("dp_bin"), |
| "player_angle_bin": geom.get("angle_bin"), |
| "player_angle_value": geom.get("angle_value"), |
| "front_cone": geom.get("front_cone"), |
| "decision_zone": geom.get("decision_zone"), |
| "tactical_sector": geom.get("tactical_sector"), |
| "behind": geom.get("behind"), |
| "player_action": player.get("action", "unknown"), |
| "prev_boss_skill": boss.get("prev_skill"), |
| "skill_phase": boss.get("skill_phase", "decision"), |
| "skill_finished": boss.get("skill_finished", True), |
| "cooldown_ready": resource.get("cooldown_ready", {}), |
| "cooldown_seconds": resource.get("cooldown_seconds", {}), |
| "hp_phase": resource.get("hp_phase"), |
| "skill_family_prior": tactical.get("skill_family_prior", {}), |
| "top_skill_family": tactical.get("top_skill_family"), |
| "confidence": confidence.get("overall", 0.0) if isinstance(confidence, dict) else confidence, |
| "prediction_source": "nested_l1_handoff", |
| } |
|
|
|
|
| def load_belief_overrides(path: str) -> Dict[tuple, Dict[str, Any]]: |
| overrides = {} |
| with open(path, encoding="utf-8") as f: |
| for line in f: |
| if not line.strip(): |
| continue |
| row = json.loads(line) |
| key = (row["boss"], int(row["fight"]), int(row["index"])) |
| if "belief" in row: |
| overrides[key] = {"belief": row["belief"]} |
| elif "l1_belief" in row: |
| overrides[key] = { |
| "belief": flatten_nested_l1_belief(row), |
| "l1_belief": row.get("l1_belief"), |
| "l2_prompt_payload": row.get("l2_prompt_payload"), |
| } |
| else: |
| raise KeyError(f"{path} row for {key} has neither 'belief' nor 'l1_belief'") |
| return overrides |
|
|
|
|
| def apply_belief_overrides(samples: List[Dict[str, Any]], path: str) -> int: |
| overrides = load_belief_overrides(path) |
| matched = 0 |
| for sample in samples: |
| key = (sample["boss"], int(sample["fight"]), int(sample["index"])) |
| if key in overrides: |
| sample["belief"] = overrides[key]["belief"] |
| if overrides[key].get("l1_belief") is not None: |
| sample["l1_belief"] = overrides[key]["l1_belief"] |
| if overrides[key].get("l2_prompt_payload") is not None: |
| sample["l2_prompt_payload"] = overrides[key]["l2_prompt_payload"] |
| matched += 1 |
| return matched |
|
|
|
|
| def write_prediction_rows(path: str, rows: List[Dict[str, Any]]) -> None: |
| os.makedirs(os.path.dirname(path) or ".", exist_ok=True) |
| tmp = f"{path}.tmp.{os.getpid()}" |
| try: |
| with open(tmp, "w", encoding="utf-8") as f: |
| for row in rows: |
| sample = row["sample"] |
| out = { |
| "boss": sample["boss"], |
| "fight": sample["fight"], |
| "index": sample["index"], |
| "target_skill": sample["target_skill"], |
| "legal_skills": sample["legal_skills"], |
| "belief": sample["belief"], |
| "l1_belief": sample.get("l1_belief"), |
| "l2_prompt_payload": sample.get("l2_prompt_payload"), |
| "decision": row["decision"], |
| "valid": row["valid"], |
| } |
| f.write(json.dumps(out, ensure_ascii=False) + "\n") |
| f.flush() |
| os.fsync(f.fileno()) |
| os.replace(tmp, path) |
| finally: |
| if os.path.exists(tmp): |
| os.unlink(tmp) |
|
|
|
|
| def validate_decision(sample: Dict[str, Any], decision: Dict[str, Any]) -> Dict[str, Any]: |
| json_valid = isinstance(decision, dict) and not decision.get("parse_error") |
| skill = decision.get("skill") if isinstance(decision, dict) else None |
| legal = skill in sample["legal_skills"] |
| ready = bool(sample["belief"]["cooldown_ready"].get(skill, True)) if legal else False |
| grounded = legal and skill_grounded(skill, sample["belief"]) |
| return { |
| "json_valid": json_valid, |
| "schema_valid": json_valid and isinstance(skill, str) and isinstance(decision.get("reason", ""), str), |
| "legal": legal, |
| "cooldown_ok": ready, |
| "grounded": grounded |
| } |
|
|
|
|
| def evaluate( |
| manifest: Dict[str, Any], |
| backend: str, |
| split: str, |
| boss: str | None, |
| seed: int, |
| limit: int | None = None, |
| beliefs_path: str | None = None, |
| predictions_out: str | None = None, |
| ) -> Dict[str, Any]: |
| rng = random.Random(seed) |
| samples = collect_decision_samples(manifest, split, boss) |
| train = collect_decision_samples(manifest, "train", boss) |
| majority = train_majority(train) |
| belief_override_matches = 0 |
| if beliefs_path: |
| belief_override_matches = apply_belief_overrides(samples, beliefs_path) |
| if limit is not None: |
| samples = samples[:limit] |
| rows = [] |
| for i, sample in enumerate(samples): |
| dec = decide(sample, backend, majority, rng) |
| val = validate_decision(sample, dec) |
| rows.append({"sample": sample, "decision": dec, "valid": val}) |
| if predictions_out: |
| write_prediction_rows(predictions_out, rows) |
|
|
| y_true = [r["sample"]["target_skill"] for r in rows] |
| y_pred = [r["decision"].get("skill") for r in rows] |
| labels = sorted({s for r in rows for s in r["sample"]["legal_skills"]}) |
| legal_pred = [p for p, r in zip(y_pred, rows) if r["valid"]["legal"]] |
| result = { |
| "backend": backend, |
| "split": split, |
| "boss": boss or "all", |
| "seed": seed, |
| "belief_source": beliefs_path or "oracle_engine_belief", |
| "belief_override_matches": belief_override_matches, |
| "n": len(rows), |
| "macro_recall": round(macro_recall(y_true, y_pred, labels), 4), |
| "json_valid_rate": round(sum(r["valid"]["json_valid"] for r in rows) / max(1, len(rows)), 4), |
| "schema_valid_rate": round(sum(r["valid"]["schema_valid"] for r in rows) / max(1, len(rows)), 4), |
| "invalid_skill_rate": round(1 - sum(r["valid"]["legal"] for r in rows) / max(1, len(rows)), 4), |
| "rule_cooldown_violation_rate": round(1 - sum(r["valid"]["cooldown_ok"] for r in rows) / max(1, len(rows)), 4), |
| "rule_grounding_violation_rate": round(1 - sum(r["valid"]["grounded"] for r in rows) / max(1, len(rows)), 4), |
| "normalized_entropy": round(normalized_entropy(y_pred, labels), 4), |
| "legal_grounded_entropy": round( |
| normalized_entropy( |
| [r["decision"]["skill"] for r in rows if r["valid"]["legal"] and r["valid"]["grounded"]], |
| labels, |
| ), |
| 4, |
| ), |
| "effective_skill_count": round(math.exp(normalized_entropy(y_pred, labels) * math.log(max(2, len(labels)))), 4), |
| "max_skill_frequency": round(Counter(y_pred).most_common(1)[0][1] / max(1, len(y_pred)), 4), |
| "js_to_real_distribution": round(js_divergence(y_pred, y_true, labels), 4), |
| "prediction_counts": dict(Counter(y_pred).most_common()), |
| } |
| return result |
|
|
|
|
| def main() -> None: |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--manifest", default="out/layered/manifest.json") |
| ap.add_argument("--backend", choices=["oracle", "majority", "random", "heuristic", "gemma"], default="heuristic") |
| ap.add_argument("--split", choices=["train", "val", "test"], default="test") |
| ap.add_argument("--boss", choices=BOSSES, default=None) |
| ap.add_argument("--seed", type=int, default=0) |
| ap.add_argument("--limit", type=int, default=None, help="limit examples for Gemma smoke tests") |
| ap.add_argument("--beliefs", default=None, help="optional JSONL predicted beliefs keyed by boss/fight/index") |
| ap.add_argument("--predictions_out", default=None, help="optional JSONL row-level decisions") |
| ap.add_argument("--out", default=None) |
| args = ap.parse_args() |
|
|
| if os.path.exists(args.manifest): |
| with open(args.manifest, encoding="utf-8") as f: |
| manifest = json.load(f) |
| else: |
| manifest = build_split_manifest() |
| result = evaluate( |
| manifest, |
| args.backend, |
| args.split, |
| args.boss, |
| args.seed, |
| args.limit, |
| args.beliefs, |
| args.predictions_out, |
| ) |
| print(json.dumps(result, ensure_ascii=False, indent=2)) |
| limit_tag = f"_n{args.limit}" if args.limit is not None else "" |
| belief_tag = "" |
| if args.beliefs: |
| stem = os.path.splitext(os.path.basename(args.beliefs))[0] |
| belief_tag = f"_belief-{stem}" |
| out = args.out or f"out/layered/l2_{args.backend}_{args.split}_{args.boss or 'all'}{belief_tag}{limit_tag}.json" |
| os.makedirs(os.path.dirname(out), exist_ok=True) |
| with open(out, "w", encoding="utf-8") as f: |
| json.dump(result, f, ensure_ascii=False, indent=2) |
| print(f"wrote {out}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
