File size: 5,048 Bytes
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 | """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.
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).
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, current_target, gt_features, log_softmax,
logits, view)
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):
self.w = dict(np.load(npz))
self.ref = dict(np.load(ref)) if ref 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.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
def act(self, frame, info, checker, presatisfied):
index, target, name = current_target(checker, presatisfied)
pose = info["player_pos"]
dist, h, v = view(pose, target)
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)
heads = [log_softmax(z) for z in logits(self.w, np.concatenate([self.feat, cam]))]
bins = [int(self.rng.choice(len(lp), p=np.exp(lp))) for lp in heads]
a = action(*bins)
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(target=index, bins=bins, logp=float(sum(lp[b] for lp, b in zip(heads, bins))), cam=cam.tolist(),
held=self.held, truth={"seen": bool(gt_features(h, v)[0]), "dist": round(dist, 2),
"h": round(h, 2), "v": round(v, 2)})
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)
|