File size: 4,537 Bytes
6a82282 | 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 128 | """Programmatic Mellea probe — hit the agent stream N times and dump
per-requirement pass/fail to a CSV so we can see which invariant keeps
failing and decide how to fix.
Requires the local server running on http://127.0.0.1:7860.
Usage:
uv run python scripts/probe_mellea.py --query Hollis --runs 5
"""
from __future__ import annotations
import argparse
import csv
import json
import time
from pathlib import Path
from urllib.parse import quote
import httpx
def stream_one(query: str, base: str, timeout_s: float) -> dict:
url = f"{base}/api/agent/stream?q={quote(query)}"
t0 = time.time()
final = None
intent = None
attempts = [] # list of {attempt, passed, failed} from mellea_attempt
with httpx.stream("GET", url, timeout=timeout_s) as r:
r.raise_for_status()
ev = None
buf = []
for line in r.iter_lines():
if line.startswith("event:"):
ev = line.split(":", 1)[1].strip()
elif line.startswith("data:"):
buf.append(line[5:].lstrip())
elif line == "":
if ev and buf:
data = "\n".join(buf)
buf = []
if ev == "plan":
try:
intent = json.loads(data).get("intent")
except json.JSONDecodeError:
pass
elif ev == "mellea_attempt":
try:
attempts.append(json.loads(data))
except json.JSONDecodeError:
pass
elif ev == "final":
try:
final = json.loads(data)
except json.JSONDecodeError:
final = {"_raw": data}
ev = None
dt = round(time.time() - t0, 2)
return {"final": final or {}, "elapsed_s": dt, "intent": intent,
"attempts": attempts}
def main():
ap = argparse.ArgumentParser()
ap.add_argument("--query", required=True)
ap.add_argument("--runs", type=int, default=5)
ap.add_argument("--base", default="http://127.0.0.1:7860")
ap.add_argument("--timeout", type=float, default=600.0)
ap.add_argument("--out", default="outputs/mellea_probe.csv")
args = ap.parse_args()
out = Path(args.out)
out.parent.mkdir(parents=True, exist_ok=True)
rows = []
for i in range(args.runs):
try:
r = stream_one(args.query, args.base, args.timeout)
except Exception as e:
print(f"[{i+1}/{args.runs}] ERROR: {e!r}")
continue
f = r["final"]
m = f.get("mellea") or {}
passed = m.get("requirements_passed", [])
failed = m.get("requirements_failed", [])
para = f.get("paragraph", "")
row = {
"run": i + 1,
"intent": r.get("intent"),
"elapsed_s": r["elapsed_s"],
"rerolls": m.get("rerolls"),
"n_attempts": m.get("n_attempts"),
"passed_count": len(passed),
"failed_count": len(failed),
"failed": ",".join(failed),
"passed": ",".join(passed),
"para_chars": len(para),
"paragraph": para.replace("\n", " "),
}
# Add per-attempt detail.
for a in r.get("attempts", []):
row[f"attempt{a.get('attempt')}_failed"] = ",".join(a.get("failed", []))
rows.append(row)
atts = r.get("attempts", [])
att_summary = " | ".join(
f"#{a.get('attempt')}={'✓' if not a.get('failed') else 'fail:'+','.join(a.get('failed', []))}"
for a in atts
) or "no attempts"
print(f"[{i+1}/{args.runs}] {r['elapsed_s']:6.1f}s final={len(passed)}/4 attempts: {att_summary}")
if rows:
with out.open("w", newline="") as f:
w = csv.DictWriter(f, fieldnames=list(rows[0].keys()))
w.writeheader()
w.writerows(rows)
print(f"\nWrote {out}")
print("Pass-rate distribution: " +
json.dumps({n: sum(1 for r in rows if r['passed_count'] == n)
for n in range(5)}))
# Show the failed paragraphs for inspection.
for r in rows:
if r['failed_count']:
print(f"\n--- run {r['run']} failed [{r['failed']}] ---")
print(r['paragraph'][:600])
if __name__ == "__main__":
main()
|