| """Verification step 1b (2026-10-13): presence scores of the base Qwen3.5-9B (no LoRA) from token logprobs, plus the |
| box temporal check of PLAN_DECOUPLED_20261013.md 2.4 point 3. Thinking off, temperature 0, top 20 logprobs. |
| |
| 1b, on step 1's 500 items (runs/diag_20261013/perception_direction/items.jsonl: 300 geometric positives, 100 behind, |
| 100 wrong-name), two scores: |
| - s_yes: B's question, max_tokens 1; P(yes) / (P(yes) + P(no)), case and leading-space variants summed; |
| - s_box: step 1's grounding question with max_tokens 48 (the controller's call). Answers either open a code fence |
| ("```", "json", "\\n", "[", "\\n", "\\t", '{"', ...) or refuse ("There are no objects belonging to ..."); after the fence |
| the list token is "[" with P("[]") ~ 0 (stored per item as p_list_empty), so empty vs non-empty is decided at the |
| first token: s_box = P(first token "```"). |
| AUC: 300 positives vs 200 negatives, and step 1's hand-labelled contact-sheet positives (37 visible vs 11 not). |
| Temporal check on GRPO v2 rounds 1-5: disjoint windows of 5 consecutive frames turning >= 5 deg at each step. |
| real = a target in the geometric field of view (|h| <= 51.2, |v| <= 35) 3-16 blocks away in all 5 frames; |
| wrong = the same windows asked about a wrong name (step 1's rule); behind = a target 3-16 blocks away with |h| >= 120 |
| in all 5 frames (spread as step 1, up to 100). Per step, the first box centre's angle change is compared with |
| -delta yaw, the change a fixed world point shows. |
| Writes items.jsonl, windows.jsonl, summary.md. |
| Usage: diag_perception_presence.py --url http://localhost:9041/v1 --model NAME [--out DIR] |
| """ |
| import argparse |
| import base64 |
| from concurrent.futures import ThreadPoolExecutor |
| import json |
| import math |
| from pathlib import Path |
| import random |
| import statistics |
|
|
| from openai import OpenAI |
|
|
| from diag_perception import BENCH, HALF_VFOV, ROOT, episodes, spread, view |
| from diag_perception_direction import BOX, GROUND, SIDES, TAN_HALF_HFOV, side |
|
|
| STEP1 = ROOT / "runs/diag_20261013/perception_direction/items.jsonl" |
| HIDDEN = {11, 17, 19, 20, 27, 29, 33, 36, 43, 45, 47} |
| HALF_HFOV = math.degrees(math.atan(TAN_HALF_HFOV)) |
| YESNO = "Is there a {0} in this image? Answer yes or no." |
|
|
|
|
| def windows(rng): |
| """(real, wrong, behind) windows: start frame, the 4 yaw changes, geometric (dist, h, v) per frame.""" |
| real, behind, names, placed = [], [], {}, {} |
| for ep, scene, _, poses, targets in episodes(): |
| placed[scene] = " ".join(json.loads((BENCH / scene / "multi-agent/metadata.json").read_text())["commands"]) |
| turn = [(b["yaw"] - a["yaw"] + 180) % 360 - 180 for a, b in zip(poses, poses[1:])] |
| for mid, task, target in targets: |
| if not task.startswith("find "): |
| continue |
| name = task[5:].removeprefix("a ").replace("_", " ") |
| names.setdefault(scene, set()).add(name) |
| free = {"real": 0, "behind": 0} |
| for s in range(len(poses) - 4): |
| if not all(abs(d) >= 5 for d in turn[s:s + 4]): |
| continue |
| geo = [[round(a, 1) for a in view(poses[s + j], target)] for j in range(5)] |
| kind = ("real" if all(3 <= d <= 16 and abs(h) <= HALF_HFOV and abs(v) <= HALF_VFOV for d, h, v in geo) |
| else "behind" if all(3 <= d <= 16 and abs(h) >= 120 for d, h, v in geo) else None) |
| if kind and s >= free[kind]: |
| free[kind] = s + 5 |
| (real if kind == "real" else behind).append( |
| {"kind": kind, "ep": str(ep.relative_to(ROOT)), "scene": scene, "mid": mid, "frame": s, |
| "name": name, "asked": name, "turn": [round(d, 2) for d in turn[s:s + 4]], "geo": geo}) |
| every = sorted(set().union(*names.values())) |
| wrong = [{**w, "kind": "wrong", "asked": rng.choice([x for x in every if x not in names[w["scene"]] |
| and x.replace(" ", "_") not in placed[w["scene"]]])} |
| for w in real] |
| return real, wrong, spread(behind, 100, rng) |
|
|
|
|
| def auc(pos, neg): |
| return sum((a > b) + 0.5 * (a == b) for a in pos for b in neg) / (len(pos) * len(neg)) |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--url", required=True) |
| ap.add_argument("--model", required=True) |
| ap.add_argument("--out", type=Path, default=ROOT / "runs/diag_20261013/perception_presence") |
| args = ap.parse_args() |
| args.out.mkdir(parents=True, exist_ok=True) |
| step1 = [json.loads(line) for line in STEP1.read_text().splitlines()] |
| real, wrong, behind = windows(random.Random(0)) |
| print(f"windows: real {len(real)} ({len({w['scene'] for w in real})} scenes, {len({w['ep'] for w in real})} " |
| f"episodes), wrong {len(wrong)}, behind {len(behind)} ({len({w['scene'] for w in behind})} scenes)", flush=True) |
|
|
| client = OpenAI(api_key="EMPTY", base_url=args.url) |
|
|
| def chat(key, prompt, max_tokens): |
| ep, frame, asked = key |
| png = (ROOT / ep / "frames" / f"{frame:04d}.png").read_bytes() |
| r = client.chat.completions.create( |
| model=args.model, temperature=0, max_tokens=max_tokens, logprobs=True, top_logprobs=20, |
| messages=[{"role": "user", "content": [ |
| {"type": "image_url", "image_url": {"url": "data:image/png;base64," + base64.b64encode(png).decode()}}, |
| {"type": "text", "text": prompt.format(asked)}]}], |
| extra_body={"chat_template_kwargs": {"enable_thinking": False}}) |
| return r.choices[0].message.content, r.choices[0].logprobs.content |
|
|
| def ground(key): |
| text, lp = chat(key, GROUND, 48) |
| boxes = [[float(c) for c in b] for b in BOX.findall(text)] |
| out = {"ground": text, "boxes": boxes, |
| "s_box": sum(math.exp(t.logprob) for t in lp[0].top_logprobs if t.token.strip() == "```"), |
| "p_list_empty": sum(math.exp(t.logprob) for t in lp[3].top_logprobs if t.token == "[]") |
| if text.startswith("```json\n[") else None} |
| if boxes: |
| x1, _, x2, _ = boxes[0] |
| out["h_box"] = round(math.degrees(math.atan(((x1 + x2) / 1000 - 1) * TAN_HALF_HFOV)), 1) |
| return out |
|
|
| def yes(key): |
| text, lp = chat(key, YESNO, 1) |
| p = {w: sum(math.exp(t.logprob) for t in lp[0].top_logprobs if t.token.strip().lower() == w) |
| for w in ("yes", "no")} |
| return {"answer": text, "s_yes": p["yes"] / (p["yes"] + p["no"])} |
|
|
| keys1 = [(x["ep"], x["frame"], x["asked"]) for x in step1] |
| keys = list(dict.fromkeys(keys1 + [(w["ep"], w["frame"] + j, w["asked"]) for w in real + wrong + behind |
| for j in range(5)])) |
| with ThreadPoolExecutor(32) as pool: |
| g = dict(zip(keys, pool.map(ground, keys))) |
| y = dict(zip(keys1, pool.map(yes, keys1))) |
| items = [{**{k: x[k] for k in ("kind", "ep", "scene", "mid", "frame", "name", "asked", "dist", "h", "v", "pitch")}, |
| **y[key], **g[key]} for x, key in zip(step1, keys1)] |
| wins = [{**w, "ground": [g[(w["ep"], w["frame"] + j, w["asked"])] for j in range(5)]} |
| for w in real + wrong + behind] |
| real, wrong, behind = ([w for w in wins if w["kind"] == k] for k in ("real", "wrong", "behind")) |
| with (args.out / "items.jsonl").open("w") as f: |
| f.writelines(json.dumps(x, ensure_ascii=False) + "\n" for x in items) |
| with (args.out / "windows.jsonl").open("w") as f: |
| f.writelines(json.dumps(w, ensure_ascii=False) + "\n" for w in wins) |
|
|
| by = {k: [x for x in items if x["kind"] == k] for k in ("pos", "behind", "wrong")} |
| shown = [x for s in SIDES for x in [x for x in by["pos"] if side(x["h"]) == s][:16]] |
| vis = [x for k, x in enumerate(shown) if k not in HIDDEN] |
| hid = [x for k, x in enumerate(shown) if k in HIDDEN] |
| neg = by["behind"] + by["wrong"] |
| aucs = ["| 比较 | n | s_yes AUC | s_box AUC |", "|---|---|---:|---:|"] |
| aucs += [f"| {label} | {len(a)} 对 {len(b)} | {auc([x['s_yes'] for x in a], [x['s_yes'] for x in b]):.3f} | " |
| f"{auc([x['s_box'] for x in a], [x['s_box'] for x in b]):.3f} |" |
| for label, a, b in (("正例对负例（背后 + 错名）", by["pos"], neg), ("正例对背后", by["pos"], by["behind"]), |
| ("正例对错名", by["pos"], by["wrong"]), ("人工：可见对看不见", vis, hid), |
| ("人工可见对负例", vis, neg))] |
| med = lambda xs: f"{statistics.median(xs):.3f}" |
| scores = ["| 组 | n | s_yes 中位数 | 答 yes | s_box 中位数 | 给框（48 token） |", "|---|---:|---:|---:|---:|---:|"] |
| scores += [f"| {label} | {len(xs)} | {med([x['s_yes'] for x in xs])} | " |
| f"{sum(x['answer'].strip().lower().startswith('yes') for x in xs)} | {med([x['s_box'] for x in xs])} | " |
| f"{sum(bool(x['boxes']) for x in xs)} |" |
| for label, xs in (("正例", by["pos"]), ("背后", by["behind"]), ("错名", by["wrong"]), |
| ("人工可见", vis), ("人工看不见", hid))] |
| fence = [x["p_list_empty"] for x in items if x["p_list_empty"] is not None] |
| starts = sorted({x["ground"][:12] for x in items if x["p_list_empty"] is None}) |
| scores += ["", f"代码块开头的回答 {len(fence)} 条，列表 token 上 P(\"[]\") 最大 {max(fence):.1e}；" |
| f"其余 {len(items) - len(fence)} 条开头为 {starts}。"] |
|
|
| def row(label, seqs): |
| """seqs: (4 yaw changes, 5 box centre angles, None = no box).""" |
| had = [(-d, a, b) for turns, hs in seqs for d, a, b in zip(turns, hs, hs[1:]) if a is not None] |
| both = [(x, b - a) for x, a, b in had if b is not None] |
| slope, icpt = statistics.linear_regression(*zip(*both)) |
| rms = math.sqrt(statistics.fmean((y - slope * x - icpt) ** 2 for x, y in both)) |
| frames = [h for _, hs in seqs for h in hs] |
| big = [y for x, y in both if abs(x) > 10] |
| pct = lambda k, n: f"{100 * k / n:.0f}%" |
| return (f"| {label} | {len(seqs)} | {pct(sum(h is not None for h in frames), len(frames))} | " |
| f"{pct(sum(b is None for _, _, b in had), len(had))} | {len(both)} | " |
| f"{statistics.median(abs(x) for x, _ in both):.1f} | {slope:.2f} | {icpt:+.1f} | {rms:.1f} | " |
| f"{pct(sum(abs(y - x) <= 5 for x, y in both), len(both))} | {pct(sum(abs(y) <= 5 for y in big), len(big))} | " |
| f"{pct(sum(min(abs(y - x), abs(y)) > 10 for x, y in both), len(both))} | " |
| f"{statistics.median(abs(y) for _, y in both):.1f} |") |
|
|
| boxes = lambda ws: [(w["turn"], [f.get("h_box") for f in w["ground"]]) for w in ws] |
| rng = random.Random(0) |
| pool = [f["h_box"] for w in wrong + behind for f in w["ground"] if "h_box" in f] |
| errs = [x["h_box"] - x["h"] for x in by["pos"] if "h_box" in x] |
|
|
| def fake_sim(redraw, world): |
| """Fake box angles over the fake windows' turns (20 repeats): first from the observed fake boxes; each step |
| redrawn with probability redraw, else kept (world = 1: shifted by -delta yaw like a fixed world point).""" |
| seqs = [] |
| for w in (wrong + behind) * 20: |
| hs = [rng.choice(pool)] |
| for d in w["turn"]: |
| hs.append(rng.choice(pool) if rng.random() < redraw else hs[-1] - world * d) |
| seqs.append((w["turn"], hs)) |
| return seqs |
|
|
| def real_sim(redraw): |
| """True target angle + an error from the 300 positives (plan 2.4 point 1), redrawn with probability redraw.""" |
| seqs = [] |
| for w in real * 20: |
| e, hs = rng.choice(errs), [] |
| for _, h, _ in w["geo"]: |
| e = rng.choice(errs) if hs and rng.random() < redraw else e |
| hs.append(h + e) |
| seqs.append((w["turn"], hs)) |
| return seqs |
|
|
| temporal = ["| 窗口 | 窗口数 | 帧给框 | 消失（上一帧有框，这一帧没有） | 相邻两帧都有框的步 | \\|Δyaw\\| 中位数 | 斜率 | 截距 | " |
| "残差 RMS | 随世界点（\\|Δh_box + Δyaw\\| ≤ 5°） | 画面不动（\\|Δh_box\\| ≤ 5°，\\|Δyaw\\| > 10° 的步） | " |
| "跳（两者都偏 >10°） | \\|Δh_box\\| 中位数 |", |
| "|---|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|---:|", |
| row("真目标：几何真值 h（参照）", [(w["turn"], [g[1] for g in w["geo"]]) for w in real]), |
| row("真目标：框", boxes(real)), row("假框：错名（同一批窗口）", boxes(wrong)), |
| row("假框：背后", boxes(behind)), row("假框合计", boxes(wrong + behind)), |
| row("模拟假框 (1) 每步独立", fake_sim(1, 0)), row("模拟假框 (2) 固定世界点", fake_sim(0, 1)), |
| row("模拟假框 (3) 画面里不动", fake_sim(0, 0)), row("模拟假框 (3)，每步以 0.5 概率重抽", fake_sim(0.5, 0)), |
| row("模拟真框：每步独立误差（2.4 第 1 点原样）", real_sim(1)), row("模拟真框：误差保持，每步以 0.5 概率重抽", real_sim(0.5))] |
| (args.out / "summary.md").write_text( |
| "基座 Qwen3.5-9B（无 LoRA），思考关闭，温度 0，取前 20 个 token 的 logprob。\n\n" |
| "## 1b 存在性分数\n\n" |
| "同第 1 步的 500 条（runs/diag_20261013/perception_direction/items.jsonl：几何正例 300，背后负例 100，错名负例 100）。" |
| f"s_yes：B 的问题 \"{YESNO.format('<name>')}\"，max_tokens 1，P(yes)/(P(yes)+P(no))，大小写和前导空格变体合计。" |
| "s_box：第 1 步的定位问题，max_tokens 48（2.2 里控制器的调用）。回答只有两种开头：代码块（```、json、换行、[、换行、制表符、{\"…）" |
| "或拒答（There are no objects belonging to …）；进了代码块后列表 token 是 [，P(\"[]\") 约为 0，" |
| "所以空和非空在第一个 token 上决定：s_box = P(第一个 token 是 ```)。人工标注取第 1 步 summary.md 接触表核对：" |
| "48 个正例里序号 11、17、19、20、27、29、33、36、43、45、47 看不见，其余 37 个可见。\n\n" |
| + "\n".join(aucs) + "\n\n" + "\n".join(scores) + "\n\n" |
| "## 框的时间行为\n\n" |
| "GRPO v2 第 1 到 5 轮 rollout，5 帧连续窗口，4 步每步 |Δyaw| ≥ 5°，同一回合同一目标的窗口不重叠。" |
| f"真目标：目标在全部 5 帧都在几何视场内（|h| ≤ {HALF_HFOV:.1f}°，|v| ≤ 35°）、距离 3 到 16 格，全部取用 {len(real)} 个" |
| f"（{len({w['scene'] for w in real})} 个场景、{len({w['ep'] for w in real})} 个回合，这批 rollout 里满足条件的只有这么多）；" |
| "错名：同一批窗口问别的场景的目标名（第 1 步的规则）；背后：目标在全部 5 帧都在 3 到 16 格、|h| ≥ 120°，按第 1 步的 spread 取" |
| f" {len(behind)} 个（{len({w['scene'] for w in behind})} 个场景）。定位问题同 1b（max_tokens 48），取第一个框中心换算水平角 h_box。" |
| "每步把 Δh_box 对 −Δyaw（固定世界点在画面里的水平角变化）做最小二乘：固定世界点斜率 1，画面里不动斜率 0。" |
| "参照行用目标点的几何 h 代替框，差别来自走路的视差。跳：框中心变化离固定世界点的预测和画面不动都超过 10°。" |
| "模拟行在观测窗口的转头序列上各重复 20 次（seed 0）：假框的位置从观测到的假框 h_box 里抽，真框 = 目标几何 h + 从 1b 的 300 个正例里抽的" |
| "框误差；“重抽”指这一步换一个新抽样，否则沿用上一步（假框 (2) 再按 −Δyaw 平移）。\n\n" |
| + "\n".join(temporal) + "\n\n" |
| "逐条结果：items.jsonl（1b 的 500 条），windows.jsonl（每个窗口 5 帧的回答、框、h_box、s_box）。\n") |
| print((args.out / "summary.md").read_text(), flush=True) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
