"""eval_1.5b_job.py Runs 15 custom quality tests on SakThai 1.5B merged model. Uploads evaluation report to the model repo. """ import os, json, time, re, sys try: import torch from transformers import AutoTokenizer, Qwen2ForCausalLM from huggingface_hub import HfApi except ImportError as e: print(f"❌ Missing dependency: {e}") sys.exit(1) MODEL_ID = "Nanthasit/sakthai-context-1.5b-merged" N_RUNS = 3 # multiple runs for stability OUTPUT = "/tmp/eval-report" # ── Tests ── TESTS = [ {"category": "basic", "name": "greeting", "messages": [{"role": "user", "content": "Hello! What can you do?"}]}, {"category": "basic", "name": "self-identity", "messages": [{"role": "user", "content": "Who are you? Tell me about yourself."}]}, {"category": "multi-turn", "name": "name-recall", "messages": [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "My name is Beer."}, {"role": "assistant", "content": "Nice to meet you, Beer!"}, {"role": "user", "content": "What is my name?"}]}, {"category": "multi-turn", "name": "context-follow", "messages": [{"role": "user", "content": "I like cats and programming."}, {"role": "assistant", "content": "Great! Cats are wonderful pets."}, {"role": "user", "content": "What two things do I like?"}]}, {"category": "multi-turn", "name": "preference-remember", "messages": [{"role": "user", "content": "Set my favorite color to blue."}, {"role": "assistant", "content": "Got it! Your favorite color is blue."}, {"role": "user", "content": "What is my favorite color?"}]}, {"category": "instruction", "name": "json-array-output", "messages": [{"role": "user", "content": "List exactly 3 primary colors. Respond ONLY with a valid JSON array. No other text."}]}, {"category": "instruction", "name": "numbered-list", "messages": [{"role": "user", "content": "Give me exactly 3 steps to make tea. Number them 1, 2, 3."}]}, {"category": "instruction", "name": "concise-output", "messages": [{"role": "user", "content": "Explain what a GPU does in exactly one short sentence."}]}, {"category": "tool-calling", "name": "weather-query", "messages": [{"role": "user", "content": "What's the weather like in Tokyo?"}]}, {"category": "tool-calling", "name": "limitation-awareness", "messages": [{"role": "user", "content": "Send an email to john@example.com saying hello."}]}, {"category": "reasoning", "name": "simple-math", "messages": [{"role": "user", "content": "If a train travels at 120 km/h for 2.5 hours, how far does it go?"}]}, {"category": "reasoning", "name": "coding-query", "messages": [{"role": "user", "content": "Write a Python function that checks if a string is a palindrome."}]}, {"category": "reasoning", "name": "explain-concept", "messages": [{"role": "user", "content": "What is the difference between LoRA and full fine-tuning? Keep it short."}]}, {"category": "format", "name": "json-object", "messages": [{"role": "user", "content": "Create a JSON object with keys: name, age, city. Use John, 30, London."}]}, {"category": "format", "name": "markdown-table", "messages": [{"role": "user", "content": "Create a markdown table comparing Python, JavaScript, and Rust (columns: Language, Typing, Speed)."}]}, ] EVALUATORS = { "json-array-output": lambda t: json_valid_list(t), "json-object": lambda t: json_valid(t), "simple-math": lambda t: "✅ Contains numeric answer" if any(c.isdigit() for c in t) else "⚠️ No digits found", "coding-query": lambda t: "✅ Contains code" if "def " in t and "return" in t else "⚠️ May lack function definition", "numbered-list": lambda t: "✅ Has numbered steps" if (any(f"{i}." in t for i in range(1,4)) or any(f"Step {i}" in t for i in range(1,4))) else f"⚠️ No numbered steps", "name-recall": lambda t: "✅ Recalls name" if "Beer" in t else "⚠️ Name not found", "context-follow": lambda t: "✅ Mentions both" if "cat" in t.lower() and "program" in t.lower() else "⚠️ Doesn't mention both", "preference-remember": lambda t: "✅ Mentions blue" if "blue" in t.lower() else "⚠️ Color not mentioned", "concise-output": lambda t: "✅ Short" if len(t.split()) <= 30 else f"⚠️ {len(t.split())} words", } def extract_json(text): for bracket in ('[', '{'): start = text.find(bracket) if start == -1: continue depth, in_str, esc = 0, False, False for i in range(start, len(text)): ch = text[i] if not in_str and ch == bracket[0]: depth += 1 elif esc: esc = False elif ch == '\\' and in_str: esc = True elif ch == '"' and not esc: in_str = not in_str elif not in_str and ((bracket == '[' and ch == ']') or (bracket == '{' and ch == '}')): depth -= 1 if depth == 0: return text[start:i+1] return text[start:] return text.strip() def json_valid(text): try: json.loads(extract_json(text)); return "✅ Valid JSON" except: return f"❌ Not valid JSON" def json_valid_list(text): try: obj = json.loads(extract_json(text)); return "✅ Valid JSON array" if isinstance(obj, list) else "❌ Not a list" except: return f"❌ Invalid JSON" def strip_role_prefix(text: str) -> str: for pat in [r'^(assistant|system|user|algorithm|tool)\s*\n', r'^(assistant|system|user|algorithm|tool)\s*[:]\s*']: text = re.sub(pat, '', text, count=1) return text.strip() # ── Load model ── device = "cuda" if torch.cuda.is_available() else "cpu" print(f"📥 Loading {MODEL_ID} on {device}", flush=True) model = Qwen2ForCausalLM.from_pretrained( MODEL_ID, torch_dtype=torch.bfloat16 if device == "cuda" else torch.float32, device_map=device, low_cpu_mem_usage=True, ) tokenizer = AutoTokenizer.from_pretrained(MODEL_ID) print(f"✅ Loaded ({sum(p.numel() for p in model.parameters()):,} params)", flush=True) # ── Run tests ── all_results = [] for run in range(N_RUNS): print(f"\n{'='*40}\nRun {run+1}/{N_RUNS}\n{'='*40}", flush=True) run_results = [] for i, test in enumerate(TESTS): try: prompt = tokenizer.apply_chat_template(test["messages"], tokenize=False) inputs = tokenizer(prompt, return_tensors="pt").to(device) t0 = time.time() with torch.no_grad(): outputs = model.generate(**inputs, max_new_tokens=256, temperature=0.1, do_sample=True, pad_token_id=tokenizer.pad_token_id or tokenizer.eos_token_id) elapsed = time.time() - t0 generated = outputs[0][inputs["input_ids"].shape[1]:] response = tokenizer.decode(generated, skip_special_tokens=True).strip() cleaned = strip_role_prefix(response) evaluator = EVALUATORS.get(test["name"], lambda t: "✅ Generated" if len(t) > 5 else "⚠️ Too short") eval_result = evaluator(cleaned) passed = eval_result.startswith("✅") print(f" [{i+1}/{len(TESTS)}] [{test['category']}] {test['name']:25s} {'✅' if passed else '❌'} {elapsed:.1f}s", flush=True) print(f" {cleaned[:100]}", flush=True) run_results.append({"name": test["name"], "passed": passed, "eval": eval_result, "response": cleaned, "time": round(elapsed, 1)}) except Exception as e: print(f" [{i+1}/{len(TESTS)}] {test['name']}: ❌ Error: {e}", flush=True) run_results.append({"name": test["name"], "passed": False, "eval": f"❌ Error: {e[:80]}", "response": "", "time": 0}) all_results.append(run_results) passed = sum(1 for r in run_results if r["passed"]) print(f" Run {run+1}: {passed}/{len(TESTS)} ({passed/len(TESTS)*100:.0f}%)", flush=True) # ── Aggregate ── from statistics import mean per_test = {t["name"]: {"passes": []} for t in TESTS} for run in all_results: for r in run: per_test[r["name"]]["passes"].append(1 if r["passed"] else 0) overall = sum(v for pt in per_test.values() for v in pt["passes"]) total = sum(len(pt["passes"]) for pt in per_test.values()) overall_rate = overall / total * 100 # ── Generate report ── report = f"""# SakThai 1.5B Merged Model — Evaluation Report **Model:** `{MODEL_ID}` **Base:** Qwen/Qwen2.5-1.5B-Instruct **Adapter:** Nanthasit/sakthai-context-1.5b-tools (LoRA r=16, alpha=32, 4 epochs) **Dataset:** Nanthasit/sakthai-combined-v4 **Runs:** {N_RUNS} | **Tests per run:** {len(TESTS)} **Overall:** {overall}/{total} passed ({overall_rate:.1f}%) ## Test-by-Test Results | # | Category | Test | Pass Rate | Avg Time | |---|----------|------|:---------:|:--------:| """ for i, t in enumerate(TESTS): rates = per_test[t["name"]]["passes"] pr = mean(rates) * 100 at = mean([r["time"] for run in all_results for r in run if r["name"] == t["name"] and r["time"] > 0]) report += f"| {i+1} | {t['category']} | {t['name']} | {'✅' if pr >= 80 else '⚠️'} {pr:.0f}% | {at:.1f}s |\n" report += f""" ## Comparison: 0.5B vs 1.5B Pass rates from 0.5B eval (single run): See `eval/EVAL.md` in 0.5b-merged repo. ## Sample Responses """ for r in all_results[-1]: if r["response"]: report += f""" ### {r['name']} > {r['response'][:200]} """ f"Time: {r['time']}s | {r['eval']}" os.makedirs(OUTPUT, exist_ok=True) with open(os.path.join(OUTPUT, "EVAL.md"), "w") as f: f.write(report) print(f"\n✅ Report saved", flush=True) # Upload print(f"☁️ Uploading report...", flush=True) from huggingface_hub import HfApi api = HfApi() api.upload_folder( repo_id=MODEL_ID, folder_path=OUTPUT, path_in_repo="eval", repo_type="model", commit_message=f"eval: {N_RUNS}-run custom eval ({overall}/{total} passed, {overall_rate:.1f}%)", ) print(f"✅ Report at https://huggingface.co/{MODEL_ID}/tree/main/eval", flush=True) print(f"\n{'='*40}\nResult: {overall}/{total} passed ({overall_rate:.1f}%)", flush=True)