File size: 6,754 Bytes
4764138 bf1c911 4764138 f244c2b 4764138 bf1c911 4764138 f244c2b 4764138 f244c2b 4764138 f244c2b 4764138 bf1c911 4764138 f244c2b 4764138 bf1c911 f244c2b 4764138 f244c2b 4764138 f244c2b 4764138 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 | """Policies of the hidden-state arms (EXPERIMENT_LOG.md 2026-10-19, tasks 3 and 4a); eval venv (numpy, requests).
HidPolicy (MC_POLICY / rollout kind "hid"): the controller's input is [feature (4096), camera history (40)] -> MLP
(train/hid_bc.py Net: 256 -> 256 -> A's three heads, exported as ctrl_policy's npz layout, read by ctrl_policy.logits).
- feature: train/hid_server.py's Qwen3.5-9B hidden state for (the current frame, the current target name), asked on
A's perception schedule (the episode's first step, when the target name changes, PERCEIVE_EVERY steps after the
last call) and held unchanged in between;
- camera history: A's window camera part, the previous camera action [yaw/64, pitch/16] of the last 20 steps oldest
first (zeros before the episode start; not reset on a target change), rounded to 4 decimals as A's x.
Memory (curiosity, EXPERIMENT_LOG.md 2026-10-25): an MLP whose w0 has 16 rows more than [feature, camera history] also
gets ctrl_policy.memory() of the episode's poses so far after the camera history, as ctrl_policy.Policy does with 116
rows; the logged cam then carries those 16 values after the 40 history values (the non-feature part of x).
Sampling as ctrl_policy.Policy (three heads, numpy Generator seeded by seed). url: the feature server (a list: one
picked per process). Options: log_feat logs the feature (base64 bf16 bits) on asking steps; frames marks asking
steps with "frame" (train/grpo_rollout.py then saves the PNG); ref (an npz) asks the server for the base feature too
and logs ref, the 45 log-probs of that controller on [base feature, camera history] (the 4a KL reference); attack (an
npz of train/mcu_attack.py, MCU) is a second policy model on the same input whose 2 outputs (no, yes) decide attack:
pressed when P(yes) > 0.5 (breaking a block needs it held without a gap), logged as patk; cam_temp divides the
yaw and pitch logits before sampling (MCU, EXPERIMENT_LOG.md 2026-10-29: steadier aim while breaking a block).
Per-step log: target, bins, logp, cam (the 40 history values), held (0 on asking steps), feat_s on asking steps
(seconds of the call), truth as Policy; plus feat / frame / ref as above.
Teacher (kind "hid_teacher"): A's controller (ctrl_policy.Policy, gt perception) whose log marks with "frame" the
steps where A's vlm schedule would ask (the BC demonstrations, train/hid_bc.py collect).
"""
import base64
from collections import deque
import os
import time
import numpy as np
import requests
from ctrl_policy import (CAM_SCALE, PERCEIVE_EVERY, WINDOW, Policy, action, cell, current_target, gt_features,
log_softmax, logits, memory, sector, view)
FEAT = 4096
def decode(s):
"""base64 bf16 bits -> float32 vector (exact)."""
return (np.frombuffer(base64.b64decode(s), np.uint16).astype(np.uint32) << 16).view(np.float32)
class HidPolicy:
def __init__(self, npz, seed, url, log_feat=False, frames=False, ref=None, attack=None, cam_temp=1.0):
self.w = dict(np.load(npz))
self.ref = dict(np.load(ref)) if ref else None
self.wa = dict(np.load(attack)) if attack else None
self.rng = np.random.default_rng(seed)
urls = url if isinstance(url, list) else [url]
self.url = urls[os.getpid() % len(urls)]
self.log_feat, self.frames = log_feat, frames
self.temps = [1.0, cam_temp, cam_temp, 1.0] # move, yaw, pitch, (the npz attack head)
self.cams = deque((np.zeros(2) for _ in range(WINDOW)), maxlen=WINDOW)
self.cam = [0.0, 0.0]
self.asked, self.held, self.feat, self.base = None, 0, None, None
self.memory = len(self.w["w0"]) == FEAT + 2 * WINDOW + 16
self.sectors, self.cells = set(), set()
def act(self, frame, info, checker, presatisfied):
index, target, name = current_target(checker, presatisfied)
dist, h, v = view(info["player_pos"], target)
a, log = self.step(frame, info["player_pos"], name)
log.update(target=index, truth={"seen": bool(gt_features(h, v)[0]), "dist": round(dist, 2),
"h": round(h, 2), "v": round(v, 2)})
return a, log
def step(self, frame, pose, name):
"""One step on frame (base64 PNG) at pose for the target name (MCU, train/mcu.py: the task's fixed name)."""
log = {}
if self.asked == name and self.held < PERCEIVE_EVERY - 1:
self.held += 1
else:
t = time.monotonic()
r = requests.post(self.url + "/feat", json={"frame": frame, "name": name}, timeout=600)
r.raise_for_status()
r = r.json()
self.feat = decode(r["feat"])
self.base = decode(r["base"]) if self.ref else None
self.asked, self.held = name, 0
log["feat_s"] = round(time.monotonic() - t, 3)
if self.log_feat:
log["feat"] = r["feat"]
if self.frames:
log["frame"] = True
self.cams.append(np.array(self.cam)[::-1] / CAM_SCALE[::-1])
cam = np.round(np.concatenate(self.cams), 4)
if self.memory:
self.sectors.add(sector(pose["yaw"]))
self.cells.add(cell(pose["x"], pose["z"]))
cam = np.concatenate([cam, memory(self.sectors, self.cells, pose)])
x = np.concatenate([self.feat, cam])
heads = [log_softmax(z / t) for z, t in zip(logits(self.w, x), self.temps)]
bins = [int(self.rng.choice(len(lp), p=np.exp(lp))) for lp in heads]
a = action(*bins)
if self.wa:
p = float(np.exp(log_softmax(logits(self.wa, x)[0])[1])) # 2 outputs: logits()' first split is all of them
a["attack"] = int(p > 0.5)
log["patk"] = round(p, 3)
self.cam = a["camera"]
if self.ref:
log["ref"] = np.round(np.concatenate([log_softmax(z) for z in logits(self.ref, np.concatenate(
[self.base, cam]))]), 4).tolist()
log.update(bins=bins, logp=float(sum(lp[b] for lp, b in zip(heads, bins))), cam=cam.tolist(), held=self.held)
return a, log
class Teacher(Policy):
def __init__(self, npz, seed):
super().__init__(npz, seed, "gt")
self.asked_name, self.since = None, 0
def act(self, frame, info, checker, presatisfied):
name = current_target(checker, presatisfied)[2]
a, log = super().act(frame, info, checker, presatisfied)
if self.asked_name == name and self.since < PERCEIVE_EVERY - 1:
self.since += 1
else:
self.asked_name, self.since = name, 0
log["frame"] = True
return a, log
def make(kind, **kwargs):
return Teacher(**kwargs) if kind == "hid_teacher" else HidPolicy(**kwargs)
|