digital-garbageandClaude Opus 5.5
ce2de22b2e
STAS (arXiv 2603.17825) sets massive-activation channels on frame 0 and frame edges to alpha x peak at one block over the first 40% of steps. Rating-learned dials now share rated_dial.py. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
91 lines
3.5 KiB
Python
91 lines
3.5 KiB
Python
"""One number per run, learned from Liked/Disliked ratings.
|
|
|
|
Each run tries a value near the key's current centre; the rating of that run pulls the
|
|
centre toward the tried value (liked) or pushes it away (disliked). The tried value is
|
|
only pending once the run finishes, so an interrupted run never takes a finished run's
|
|
rating, and a keyed run with the feature off clears it. Stored as
|
|
`<key>.<name>.json`: {"history": [{"v", "reward"}], "pending": v}.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import random
|
|
|
|
MODES = ("off", "learned", "manual")
|
|
|
|
|
|
class Dial:
|
|
def __init__(self, name, start, lo, hi, explore, lr=0.5):
|
|
self.name, self.start, self.lo, self.hi = name, start, lo, hi
|
|
self.explore, self.lr = explore, lr
|
|
|
|
def path(self, refinement_key):
|
|
try:
|
|
from .conditioning import refinement_state_path
|
|
except ImportError:
|
|
from conditioning import refinement_state_path
|
|
return refinement_state_path(refinement_key, self.name, prefix="refine_v2",
|
|
extension="json")
|
|
|
|
def read(self, refinement_key):
|
|
try:
|
|
with open(self.path(refinement_key), "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
return data if isinstance(data, dict) else {}
|
|
except (FileNotFoundError, ValueError):
|
|
return {}
|
|
|
|
def _write(self, refinement_key, data):
|
|
path = self.path(refinement_key)
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
tmp = path + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(data, f)
|
|
os.replace(tmp, path)
|
|
|
|
def learned(self, history):
|
|
m = self.start
|
|
for h in history:
|
|
s = (h["reward"] > 0) - (h["reward"] < 0)
|
|
m += self.lr * s * (float(h["v"]) - m)
|
|
m = min(max(m, self.lo), self.hi)
|
|
return m
|
|
|
|
def choose(self, refinement_key, mode, manual, rng=random):
|
|
"""-> (value, note); `None` value = off this run."""
|
|
if mode == "manual":
|
|
return float(manual), f"manual {float(manual):.2f}"
|
|
if mode != "learned" or not refinement_key:
|
|
return None, ""
|
|
history = self.read(refinement_key).get("history") or []
|
|
centre = self.learned(history)
|
|
v = min(max(centre + rng.gauss(0.0, self.explore), self.lo), self.hi)
|
|
return v, f"learned {centre:.2f} from {len(history)} rating(s), trying {v:.2f}"
|
|
|
|
def save_pending(self, refinement_key, v):
|
|
if refinement_key:
|
|
data = self.read(refinement_key)
|
|
data["pending"] = float(v)
|
|
self._write(refinement_key, data)
|
|
|
|
def clear_pending(self, refinement_key):
|
|
"""A keyed run not learning this: its rating must not score an older run's value."""
|
|
if refinement_key and "pending" in self.read(refinement_key):
|
|
data = self.read(refinement_key)
|
|
data.pop("pending", None)
|
|
self._write(refinement_key, data)
|
|
|
|
def commit(self, refinement_key, reward):
|
|
"""Pair the pending value with its run's rating. -> ratings held, or None."""
|
|
if not refinement_key:
|
|
return None
|
|
data = self.read(refinement_key)
|
|
v = data.pop("pending", None)
|
|
if v is None:
|
|
return None
|
|
if reward:
|
|
data["history"] = ((data.get("history") or [])
|
|
+ [{"v": float(v), "reward": float(reward)}])[-200:]
|
|
self._write(refinement_key, data)
|
|
return len(data["history"]) if reward else None
|