"""App-level chat operational evaluation — focused on the specificity head + working-memory pinning. Two phases (run both): PHASE 1 — ARCHITECTURE scorecard (deterministic, NO LLM, seconds). Replays a scripted session through the SAME primitives turn() uses (intent_of / specific_spans / retrieve_personal / command branch) and asserts ROUTING + PIN + RETRIEVAL. This is the part the specificity head changes, and it's where we demand correctness (per the metric-rigor rule: assert architecture). PHASE 2 — FULL generation (real ChatSession.turn, the actual answers). Reported, not graded on the 1.5B's arithmetic/world-knowledge (that's the base ceiling, out of scope). Scenario stresses exactly what the head is for: * a VALUE stated inside a MATH turn ($120/day) -> pinned though never a logged fact -> later recompute (#7) AND later recall-via-pin-fallback (#14) * the pin must NOT derail a genuine recall of real facts ('budget + how long', #9) * genuine logged-fact recall still works (deposit #10, seat code #13) * all 6 intents routed; a correction (#6) Run: python3.12 evals/app_eval2.py # phase 1 then phase 2 python3.12 evals/app_eval2.py --arch # phase 1 only (fast) """ import sys, os sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import tiered_rag_mlx as T # (msg, expected_intent, ack_only_fact?, [arch checks on (src,chunks,pins)]) # checks: ("pin", substr) value must be in pins; ("ctx", substr) substr in retrieved chunks; # ("noctx", substr) substr must NOT be in chunks; ("src", substr) substr in src label. SESSION = [ ("hey, I'm planning a trip", "chitchat", False, []), ("I'm going to Kyoto for 3 days", "fact", True, []), ("my budget is $800", "fact", True, []), ("the hotel deposit was $250", "fact", True, [("pin", "250")]), ("if I spend $120 a day on food for 3 days, what's the total?", "math", False, [("pin", "120")]), ("actually make it 4 days", "fact", True, []), ("recompute the food total", "command", False, [("ctx", "120"), ("src", "pin")]), ("who is the current emperor of Japan?", "lookup", False, [("src", "web")]), ("remind me my budget and how long I'm staying", "recall", False, [("ctx", "800"), ("ctx", "4 days"), ("noctx", "food")]), ("what was the hotel deposit?", "recall", False, [("ctx", "250")]), ("I'm vegetarian", "fact", True, []), ("my flight is DL472 seat 14C", "fact", True, [("pin", "14C")]), ("what's my seat again?", "recall", False, [("ctx", "14C")]), # classifier routes this to 'lookup' (no 'my', sounds generic) — the WM safety-net is exactly what # salvages this recall/lookup boundary error, surfacing the pin-only $120. Functionally correct. ("what was the food budget per day?", "lookup", False, [("ctx", "120"), ("src", "pin")]), ("thanks, that's all!", "chitchat", False, []), ] def route(mem, msg): """Mirror ChatSession.turn()'s routing+pin logic EXACTLY (no generation) -> (intent, src, chunks).""" intent = T.intent_of(msg) if intent not in ("recall", "lookup") and T.specific_spans(msg): mem.pin(msg) if intent == "recall": src, chunks = mem.retrieve_personal(msg) elif intent == "lookup": wm_only = [p for p in mem.pins if p not in mem.session] mp = T._sem_matches(msg, wm_only, min_sim=0.5) if wm_only else None if mp: src, chunks = "WM·pins", mp else: src, chunks = "L3·web(not-run-in-arch)", [""] # arch phase doesn't hit the network elif intent == "command" and (mem.session or mem.pins): log = mem.session[-mem.LOGCAP:] chunks = log + [c for c in mem.pins if c not in log] src = "L1·same-session" + ("+WM·pins" if mem.pins else "") else: src, chunks = None, [] return intent, src, chunks def phase1(): print("=" * 72); print("PHASE 1 — ARCHITECTURE scorecard (no LLM)"); print("=" * 72) mem = T.TieredMemory("/tmp/app_eval2_arch.jsonl") if os.path.exists(mem.path): os.remove(mem.path); mem = T.TieredMemory(mem.path) npass = nfail = 0 for msg, exp_intent, ack, checks in SESSION: intent, src, chunks = route(mem, msg) if ack and intent == "fact": mem.remember_session(msg) # fact -> logged (turn() does this) elif intent == "fact": mem.remember_session(msg) line_ok = True notes = [] if intent != exp_intent: line_ok = False; notes.append(f"INTENT {intent}!={exp_intent}") for kind, sub in checks: chunks = chunks or [] in_ctx = any(sub.lower() in c.lower() for c in chunks) in_pin = any(sub.lower() in p.lower() for p in mem.pins) if kind == "pin" and not in_pin: line_ok = False; notes.append(f"!pin '{sub}'") if kind == "ctx" and not in_ctx: line_ok = False; notes.append(f"!ctx '{sub}'") if kind == "noctx" and in_ctx: line_ok = False; notes.append(f"LEAK '{sub}'") if kind == "src" and sub.lower() not in (src or "").lower(): line_ok = False; notes.append(f"!src '{sub}' (got {src})") npass += line_ok; nfail += not line_ok print(f" {'OK ' if line_ok else 'XX '} [{intent:8s}] {msg[:46]:46s} " f"src={src or '—'}" + (f" << {', '.join(notes)}" if notes else "")) print(f"\nPHASE1: {npass}/{npass+nfail} architecture checks pass") return nfail == 0 def phase2(): import time print("\n" + "=" * 72); print("PHASE 2 — FULL generation (real answers)"); print("=" * 72) P = "/tmp/app_eval2_full.jsonl" if os.path.exists(P): os.remove(P) chat = T.ChatSession(T.TieredMemory(P)) for msg, exp_intent, ack, checks in SESSION: t0 = time.time() intent = T.intent_of(msg) # REAL app policy (cf. app_session.py): a fact -> instant ack + log; everything else -> store='none' # (turn() still auto-logs intent=='fact'; non-fact turns must NOT be logged, or a math-turn value # stops being 'working-memory-only' and the lookup->pins safety net can't fire for it). if intent == "fact": ans, src, ch = chat.turn(msg, store="session", ack_only=True) else: ans, src, ch = chat.turn(msg, store="none") dt = time.time() - t0 print("=" * 72) print(f"USER: {msg}") print(f" [intent={intent} · src={src or '—'} · {dt:.0f}s] pins={chat.mem.pins[-3:]}") print(f"ASSISTANT: {ans[:220]}") print("\nAPP_EVAL2_DONE") if __name__ == "__main__": ok1 = phase1() print(f"\n>>> PHASE 1 {'PASS' if ok1 else 'FAIL'}") if "--arch" not in sys.argv: phase2()