| |
| """FULL-418 head-to-head: run every coverage item (all 6 scorers, no OBJECTIVE filter) against an |
| Ollama-API model, scored IDENTICALLY to eval_coverage.py (the harness r9/ConeML were scored with). |
| Matched decoding: temp 0 + repeat_penalty 1.15, full budget (num_predict 2048). Saves generations. |
| Usage: --model qwen2.5:0.5b --host http://127.0.0.1:11434 --out <json>""" |
| import sys, json, argparse, re, subprocess, tempfile, os, urllib.request |
|
|
| BLEED_RE = re.compile( |
| r"\b(scope|scoped|scoping|dependency order|validation|stop condition|acceptance criteria|" |
| r"tool use|tool-use|artifact|deliverable|implementation plan|agentic|route handler)\b", re.I) |
|
|
|
|
| def s_math(it, g): |
| m = re.search(r"####\s*([-\d,\.]+)", g) |
| if not m: |
| cu = re.findall(r"(?:answer|final|total|result|equals?|is)\D{0,15}(-?\d[\d,]*\.?\d*)", g, re.I) |
| pred = cu[-1] if cu else (re.findall(r"-?\d[\d,]*\.?\d*", g.replace(",", "")) or [None])[-1] |
| else: |
| pred = m.group(1) |
| try: return abs(float(str(pred).replace(",", "")) - float(str(it["expected"]))) < 1e-6, pred |
| except Exception: return False, pred |
|
|
|
|
| def s_code(it, g): |
| m = re.search(r"```(?:python)?\n(.*?)```", g, re.S) |
| body = m.group(1) if m else g |
| cands = [] |
| if "def " in body and body.lstrip().startswith("def "): |
| cands.append(body) |
| else: |
| pc = it.get("prompt_code", "") |
| cands.append(pc + body) |
| indented = "\n".join((" " + ln if ln.strip() and not ln[:1].isspace() else ln) for ln in body.splitlines()) |
| cands.append(pc + indented) |
| wrap = (f"\ncheck({it['entry_point']})\n" if it.get("entry_point") else "\n") |
| last = "" |
| for cand in cands: |
| prog = cand + "\n" + it["tests"] + wrap |
| try: |
| with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f: |
| f.write(prog); p = f.name |
| r = subprocess.run([sys.executable, p], capture_output=True, timeout=12, text=True) |
| os.unlink(p) |
| if r.returncode == 0: return True, "ok" |
| last = (r.stderr or "")[-200:] |
| except Exception as e: last = str(e)[-200:] |
| return False, last |
|
|
|
|
| def s_terms_all(it, g): |
| lo = g.lower(); return all(t.lower() in lo for t in it["expected_terms"]), None |
|
|
|
|
| def s_terms_any(it, g): |
| lo = g.lower(); return any(t.lower() in lo for t in it["expected_terms"]), None |
|
|
|
|
| def degenerate(g): |
| lines = [x.strip() for x in g.splitlines() if x.strip()] |
| if lines and max({l: lines.count(l) for l in set(lines)}.values()) / len(lines) > 0.4 and len(lines) > 4: |
| return True |
| words = g.split() |
| return len(words) > 12 and len(set(words)) / len(words) < 0.35 |
|
|
|
|
| def s_rubric(it, g): |
| ok = len(re.findall(r"[A-Za-z']+", g)) >= it.get("min_len", 30) and not degenerate(g) |
| if it.get("expected_terms"): |
| ok = ok and any(t.lower() in g.lower() for t in it["expected_terms"]) |
| return ok, None |
|
|
|
|
| def s_bleed(it, g): |
| bleed = sorted(set(m.group(0).lower() for m in BLEED_RE.finditer(g))) |
| term_ok = (not it.get("expected_terms")) or any(t.lower() in g.lower() for t in it["expected_terms"]) |
| return (term_ok and not bleed), (",".join(bleed) if bleed else None) |
|
|
|
|
| SC = {"math_exact": s_math, "code_exec": s_code, "sql_shape": s_terms_all, |
| "factual_terms": s_terms_any, "rubric": s_rubric, "bleed": s_bleed} |
|
|
|
|
| def chat(host, model, prompt, timeout=180): |
| body = json.dumps({"model": model, "messages": [{"role": "user", "content": prompt}], |
| "stream": False, "options": {"temperature": 0, "repeat_penalty": 1.15, "num_predict": 2048}}).encode() |
| req = urllib.request.Request(host.rstrip("/") + "/api/chat", data=body, headers={"Content-Type": "application/json"}) |
| with urllib.request.urlopen(req, timeout=timeout) as r: |
| return json.loads(r.read())["message"]["content"] |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--model", required=True) |
| ap.add_argument("--host", default="http://127.0.0.1:11434") |
| ap.add_argument("--out", required=True) |
| ap.add_argument("--batteries", nargs="+", default=["data/eval_fixed/coverage/all.jsonl"]) |
| a = ap.parse_args() |
| items = [] |
| for b in a.batteries: |
| for l in open(b): |
| if l.strip(): |
| it = json.loads(l) |
| if it["scorer"] in SC: |
| items.append(it) |
| rows = []; agg = {} |
| for i, it in enumerate(items): |
| try: |
| g = chat(a.host, a.model, it["prompt"]) |
| except Exception as e: |
| g = f"[API_ERROR:{e}]" |
| ok, info = SC[it["scorer"]](it, g) |
| t = it["type"]; agg.setdefault(t, {"ok": 0, "n": 0}) |
| agg[t]["ok"] += int(ok); agg[t]["n"] += 1 |
| rows.append({"id": it["id"], "type": t, "scorer": it["scorer"], "ok": ok, "info": info, "gen": g[:1500]}) |
| if (i + 1) % 25 == 0: print(f" {i+1}/{len(items)}", flush=True) |
| summary = {t: {"ok": v["ok"], "n": v["n"], "rate": round(v["ok"]/v["n"], 3)} for t, v in sorted(agg.items())} |
| tot = sum(v["ok"] for v in agg.values()); n = sum(v["n"] for v in agg.values()) |
| os.makedirs(os.path.dirname(a.out), exist_ok=True) |
| json.dump({"model": a.model, "overall": {"ok": tot, "n": n}, "by_type": summary, "rows": rows}, open(a.out, "w"), indent=1) |
| print(f"\n{a.model} FULL {tot}/{n}") |
| for t, v in summary.items(): print(f" {t:24s} {v['ok']:3d}/{v['n']:<3d} = {100*v['rate']:5.1f}%") |
| print("->", a.out) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
