docs(diagnostics): fiction-wing retrieval probe harness (R42 + #389 gate)
Self-contained, re-runnable probe requested by brokkr-smithy-dev for R42 (fiction-wing retrieval characterization) and the standing #389 ranking acceptance gate. Two paths kept separate by noise property: search_library (mimir, fixed-string, deterministic — ranking arm) and reference_knowledge (donut, captures her reformulated tool_query — the query- formulation/arm-4 surface). Scoring conventions baked in: high/medium/low RRF buckets (0.030/0.016), on-target = a row whose excerpt names the subject, bucket-distribution over N runs. Carries the frozen artifact yardstick (4 source-verified items + epithet-dropped variants). Config from env (no secrets). Smoke-verified live: reproduces the Crown-HIT / other-three-MISS baseline and the near-floor bucket flips.
This commit is contained in:
@@ -0,0 +1,170 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Fiction-wing retrieval probe harness — the recipe R42 (brokkr-smithy-dev) builds against
|
||||
and the re-runnable #389 acceptance gate.
|
||||
|
||||
Two retrieval paths, kept SEPARATE because they have different noise properties:
|
||||
|
||||
* search_library (raw, RANKING-clean): drive the `mimir` foundational agent (all-wing
|
||||
librarian) with a FIXED query string. Deterministic against a fixed index — use it for
|
||||
ranking baselines (R42 arm-2). No LLM in the query loop.
|
||||
* reference_knowledge (the Tier-3 consumer path): drive `ratatoskr:donut`; her reasoning
|
||||
REFORMULATES the query each turn, so this path carries QUERY-FORMULATION variance
|
||||
(the arm-4 signal), attributable via the captured tool_start query. NOT for ranking numbers.
|
||||
|
||||
Scoring conventions (identical across both paths and all R42 arms):
|
||||
* confidence BUCKET vs WT's RRF thresholds: high >= 0.030, medium >= 0.016, low < 0.016.
|
||||
* ON-TARGET (load-bearing): a returned row is on-target iff its excerpt actually NAMES or
|
||||
describes the queried subject (keyword match on the subject's distinctive tokens). The
|
||||
failure signature "10 hits / MEDIUM / 0 on-target" = present-by-topic, subject absent —
|
||||
the split that separated #384 (packaging) / #387 (coverage) / #389 (ranking).
|
||||
* MISS = no on-target row in the returned top-k.
|
||||
|
||||
Noise floor: freeze the generation (pin the b-tag) to remove extraction variance; fixed-string
|
||||
search_library is deterministic (no CI needed); reference_knowledge variance is query-
|
||||
formulation, not floor noise. Residual = bucket-boundary sensitivity at 0.016/0.030 — so probe
|
||||
N>=3-5 times per term and report the bucket DISTRIBUTION, never a single-run point label.
|
||||
|
||||
Config from env (source ratatoskr's env.sh): WORLDTREE_API_URL, WORLDTREE_API_KEY,
|
||||
RATATOSKR_END_USER_ID. No secrets are stored here.
|
||||
|
||||
Usage:
|
||||
uv run python docs/diagnostics/fiction_wing_probe.py # run the artifact yardstick
|
||||
uv run python docs/diagnostics/fiction_wing_probe.py --runs 5 # N repeats -> bucket distribution
|
||||
uv run python docs/diagnostics/fiction_wing_probe.py --term "Enhanced Pet Biscuit" --keywords biscuit
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
from collections import Counter
|
||||
|
||||
import httpx
|
||||
|
||||
HIGH, MEDIUM = 0.030, 0.016 # WT RRF confidence thresholds
|
||||
|
||||
|
||||
def _cfg() -> tuple[str, dict, str]:
|
||||
base = os.environ.get("WORLDTREE_API_URL", "http://10.250.50.152:8081")
|
||||
key = os.environ.get("WORLDTREE_API_KEY")
|
||||
if not key:
|
||||
raise SystemExit("WORLDTREE_API_KEY unset — source env.sh first.")
|
||||
end_user = os.environ.get("RATATOSKR_END_USER_ID", "ratatoskr-tui")
|
||||
return base, {"Authorization": f"Bearer {key}"}, end_user
|
||||
|
||||
|
||||
def _bucket(score: float | None) -> str:
|
||||
if score is None:
|
||||
return "none"
|
||||
return "high" if score >= HIGH else "medium" if score >= MEDIUM else "low"
|
||||
|
||||
|
||||
def _on_target(excerpt: str, keywords: list[str]) -> bool:
|
||||
ex = (excerpt or "").lower()
|
||||
return any(k.lower() in ex for k in keywords)
|
||||
|
||||
|
||||
def _session(base: str, headers: dict, agent_id: str, end_user: str) -> str:
|
||||
r = httpx.post(f"{base}/sessions", json={"agent_id": agent_id, "end_user_id": end_user},
|
||||
headers=headers, timeout=30)
|
||||
return r.json()["session_id"]
|
||||
|
||||
|
||||
def _drive(base: str, headers: dict, sid: str, content: str) -> tuple[str | None, dict]:
|
||||
"""POST a turn, return (tool_query, tool_result_dict). tool_result is the first tool packet."""
|
||||
tool_query, result = None, {}
|
||||
with httpx.stream("POST", f"{base}/sessions/{sid}/messages", json={"content": content},
|
||||
headers=headers, timeout=120) as r:
|
||||
for line in r.iter_lines():
|
||||
if not line.startswith("data: "):
|
||||
continue
|
||||
ev = json.loads(line[6:])
|
||||
t = ev.get("type")
|
||||
if t == "tool_start" and tool_query is None:
|
||||
tool_query = (ev.get("arguments") or {}).get("query")
|
||||
elif t == "tool_result" and not result:
|
||||
result = ev.get("result") if isinstance(ev.get("result"), dict) else {}
|
||||
elif t == "done":
|
||||
break
|
||||
return tool_query, result
|
||||
|
||||
|
||||
def search_library(base, headers, msid, term, keywords):
|
||||
"""RANKING-clean path: fixed-string search over the mimir librarian. Deterministic."""
|
||||
_, res = _drive(base, headers, msid, f"Use search_library to find: {term}")
|
||||
rows = res.get("results", []) if isinstance(res, dict) else []
|
||||
on = [h for h in rows if isinstance(h, dict) and _on_target(h.get("excerpt", ""), keywords)]
|
||||
top = on[0] if on else None
|
||||
score = round(top["score"], 4) if top else None
|
||||
return {"n": len(rows), "on_target": len(on), "hit": bool(on),
|
||||
"score": score, "bucket": _bucket(score),
|
||||
"excerpt": (top.get("excerpt", "")[:140] if top else None)}
|
||||
|
||||
|
||||
def reference_knowledge(base, headers, dsid, question, keywords):
|
||||
"""Consumer path: Donut reformulates -> capture her tool_query. NOT for ranking numbers."""
|
||||
q, res = _drive(base, headers, dsid, question)
|
||||
hits = res.get("hits", []) if isinstance(res, dict) else []
|
||||
on = [h for h in hits if isinstance(h, dict) and _on_target(h.get("excerpt", ""), keywords)]
|
||||
return {"tool_query": q, "n": len(hits), "on_target": len(on),
|
||||
"confidence": res.get("confidence") if isinstance(res, dict) else None}
|
||||
|
||||
|
||||
# Artifact yardstick — worldtree-dev grep-confirmed in DCC book-1. Frozen arm-2 baseline.
|
||||
YARDSTICK = [
|
||||
("Enchanted Crown of the Sepsis Whore", "Crown of the Sepsis Whore", ["sepsis", "crown"]),
|
||||
("Enhanced Pet Biscuit", "Pet Biscuit", ["biscuit"]),
|
||||
("Enchanted BigBoi Boxers", "BigBoi Boxers", ["boxers", "bigboi"]),
|
||||
("Enchanted Toe Ring of the Splatter Skunk", "Toe Ring of the Splatter Skunk",
|
||||
["toe ring", "splatter", "skunk"]),
|
||||
]
|
||||
|
||||
|
||||
def run_yardstick(runs: int) -> None:
|
||||
base, headers, end_user = _cfg()
|
||||
msid = _session(base, headers, "mimir", end_user)
|
||||
print(f"# Fiction-wing ranking yardstick (search_library, {runs} run(s) per name)\n")
|
||||
for full, partial, kw in YARDSTICK:
|
||||
for label, term in (("full ", full), ("part ", partial)):
|
||||
buckets, hits = Counter(), 0
|
||||
for _ in range(runs):
|
||||
r = search_library(base, headers, msid, term, kw)
|
||||
buckets[r["bucket"]] += 1
|
||||
hits += r["hit"]
|
||||
dist = " ".join(f"{b}:{c}" for b, c in buckets.most_common())
|
||||
print(f" [{label}] {term:<42} hit {hits}/{runs} buckets({dist})")
|
||||
print()
|
||||
|
||||
|
||||
def run_term(term: str, keywords: list[str], runs: int) -> None:
|
||||
base, headers, end_user = _cfg()
|
||||
msid = _session(base, headers, "mimir", end_user)
|
||||
dsid = _session(base, headers, "ratatoskr:donut", end_user)
|
||||
print(f"# Probe: {term!r} ({runs} run(s))\n")
|
||||
sl_buckets, sl_hits = Counter(), 0
|
||||
for _ in range(runs):
|
||||
r = search_library(base, headers, msid, term, keywords)
|
||||
sl_buckets[r["bucket"]] += 1
|
||||
sl_hits += r["hit"]
|
||||
print(f" search_library : hit {sl_hits}/{runs} buckets({dict(sl_buckets)})")
|
||||
for _ in range(runs):
|
||||
rk = reference_knowledge(base, headers, dsid, f"Tell me about the {term}.", keywords)
|
||||
print(f" reference_knowledge: conf={rk['confidence']} on_target={rk['on_target']}"
|
||||
f" (donut query: {rk['tool_query']!r})")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
ap = argparse.ArgumentParser(description=__doc__)
|
||||
ap.add_argument("--runs", type=int, default=1, help="repeats per term (>=3-5 near the floor)")
|
||||
ap.add_argument("--term", help="probe a single term instead of the yardstick")
|
||||
ap.add_argument("--keywords", nargs="*", default=[], help="on-target keywords for --term")
|
||||
ns = ap.parse_args()
|
||||
if ns.term:
|
||||
run_term(ns.term, ns.keywords or [ns.term.split()[-1]], ns.runs)
|
||||
else:
|
||||
run_yardstick(ns.runs)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user