File size: 4,328 Bytes
04558eb | 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 129 130 131 132 133 134 135 136 137 138 | #!/usr/bin/env python3
"""Evaluate fine-tuned models on the NL2Bash holdout set (604 entries).
Compares multiple adapters on the same held-out dictation → bash pairs.
"""
import json
import time
from collections import defaultdict
from mlx_lm import load, generate
SYS = "Reconstruct the intended syntax from the dictated text. Output only the result."
# Load test split
with open("datasets/finetune/bash/test.jsonl") as f:
tests = []
for line in f:
msg = json.loads(line)["messages"]
tests.append({
"dictated": msg[1]["content"],
"expected": msg[2]["content"],
})
print(f"Loaded {len(tests)} held-out bash test entries\n")
configs = [
{
"label": "Qwen 0.5B + LoRA (bash)",
"model": "mlx-community/Qwen2.5-0.5B-Instruct-4bit",
"adapter": "datasets/finetune/adapters/qwen-0.5b-bash-v1",
},
{
"label": "Qwen 1.5B + LoRA (bash)",
"model": "mlx-community/Qwen2.5-1.5B-Instruct-4bit",
"adapter": "datasets/finetune/adapters/qwen-1.5b-bash-v1",
},
]
all_results = {}
for cfg in configs:
print(f"\n{'='*60}")
print(f" {cfg['label']}")
print(f"{'='*60}\n")
try:
model, tokenizer = load(cfg["model"], adapter_path=cfg["adapter"])
except Exception as e:
print(f" SKIPPED — {e}\n")
continue
results = []
total_time = 0
errors_shown = 0
for i, t in enumerate(tests):
messages = [
{"role": "system", "content": SYS},
{"role": "user", "content": t["dictated"]},
]
prompt = tokenizer.apply_chat_template(
messages, tokenize=False, add_generation_prompt=True
)
start = time.perf_counter()
got = generate(model, tokenizer, prompt=prompt, max_tokens=120, verbose=False)
elapsed = time.perf_counter() - start
total_time += elapsed
got = got.strip()
match = got == t["expected"]
results.append({
"match": match,
"got": got,
"expected": t["expected"],
"dictated": t["dictated"],
})
if not match and errors_shown < 30:
icon = "\033[31m✗\033[0m"
print(f"{icon} {i+1:>3}. \"{t['dictated']}\"")
print(f" expected: {t['expected']}")
print(f" got: {got}")
errors_shown += 1
correct = sum(1 for r in results if r["match"])
pct = round(correct / len(results) * 100, 1)
avg_ms = round(total_time / len(results) * 1000)
print(f"\nScore: {correct}/{len(results)} ({pct}%)")
print(f"Avg latency: {avg_ms}ms")
if errors_shown < sum(1 for r in results if not r["match"]):
print(f"({sum(1 for r in results if not r['match'])} total errors, showing first {errors_shown})")
all_results[cfg["label"]] = results
del model, tokenizer
# Summary
print(f"\n{'='*60}")
print(" RESULTS SUMMARY")
print(f"{'='*60}\n")
for label, results in all_results.items():
correct = sum(1 for r in results if r["match"])
pct = round(correct / len(results) * 100, 1)
print(f" {label}: {correct}/{len(results)} ({pct}%)")
# Error analysis: categorize common failure patterns
if all_results:
print(f"\n{'='*60}")
print(" ERROR ANALYSIS (first model)")
print(f"{'='*60}\n")
first_results = list(all_results.values())[0]
errors = [r for r in first_results if not r["match"]]
# Categorize errors
categories = defaultdict(list)
for r in errors:
got, exp = r["got"], r["expected"]
if len(got) > len(exp) * 2:
categories["hallucination"].append(r)
elif got.lower() == exp.lower():
categories["case_mismatch"].append(r)
elif got.replace(" ", "") == exp.replace(" ", ""):
categories["spacing"].append(r)
elif abs(len(got) - len(exp)) <= 3:
categories["minor_diff"].append(r)
else:
categories["structural"].append(r)
for cat, items in sorted(categories.items(), key=lambda x: -len(x[1])):
print(f" {cat}: {len(items)}")
for item in items[:3]:
print(f" \"{item['dictated'][:50]}...\"")
print(f" exp: {item['expected'][:60]}")
print(f" got: {item['got'][:60]}")
|