#!/usr/bin/env python3 """Score the question-answering service against a held-out question set. Purpose: run every question in qa-questions.json through the same pipeline the service uses, score each answer three ways - did it refuse when it should have, does it contain the facts the reference answer contains, did it cite the right document - and optionally add a judge model's rating. Writes one JSON line per question plus one summary line to the lab notebook, so two runs a week apart can be compared. Platform: all (pure Python over HTTP) Minimum memory: 8 GB on the machine running the models Assumes: Python 3.9 or later, `sqlite-vec` and `pydantic` installed, ask.py in the same directory, an index built by ingest.py, and the same servers ask.py needs. The judge is optional: pass --no-judge to score without one, or --judge-model to name a different and preferably larger model than the one being scored. Usage: python3 eval-qa.py --db qa-index.db --model qwen3-4b --judge-model qwen3-8b \ --questions qa-questions.json --labbook labbook.md python3 eval-qa.py --db qa-index.db --model qwen3-8b --no-judge """ from __future__ import annotations import argparse import json import re import sys import time from collections import Counter from pathlib import Path from typing import Optional try: import ask except ImportError: # pragma: no cover - environment check sys.exit("ask.py must be in the same directory as this script; run it from there.") JUDGE_SYSTEM = """You grade one answer against a reference answer. Reply with JSON only. Scale: 5 every fact in the reference is present and nothing is added that the reference does not support 4 correct, with a small omission or a harmless addition 3 partly correct: one required fact is missing or wrong 2 mostly wrong, or answers a different question 1 wrong, or asserts something the reference contradicts Grade the content, not the length or the style. A short correct answer scores higher than a long one that buries the same fact. If the reference says the documents do not contain the answer, then a refusal scores 5 and any confident answer scores 1.""" def normalise(text: str) -> str: return re.sub(r"[^a-z0-9]+", " ", text.lower()).strip() def contains_all(answer: str, needles) -> bool: haystack = normalise(answer) return all(normalise(n) in haystack for n in needles) def judge(question: str, reference: str, candidate: str, args) -> Optional[int]: """One judge call. Returns 1-5, or None when the judge itself failed.""" endpoint = args.base_url.rstrip("/") + "/chat/completions" user = ( f"Question: {question}\n\n" f"Reference answer: {reference}\n\n" f"Answer to grade: {candidate or '(the system refused to answer)'}\n\n" 'Reply as {"score": <1-5>, "why": ""}' ) payload = { "model": args.judge_model, "messages": [{"role": "system", "content": JUDGE_SYSTEM}, {"role": "user", "content": user}], "temperature": 0.0, "max_tokens": 200, "seed": args.seed, "response_format": { "type": "json_schema", "json_schema": { "name": "grade", "schema": { "type": "object", "properties": { "score": {"type": "integer", "minimum": 1, "maximum": 5}, "why": {"type": "string"}, }, "required": ["score", "why"], "additionalProperties": False, }, "strict": True, }, }, } try: body = ask.post_json(endpoint, payload, args.api_key, args.timeout) content = body["choices"][0]["message"]["content"] return int(json.loads(content)["score"]) except (RuntimeError, KeyError, ValueError, TypeError) as exc: print(f" judge failed: {exc}", file=sys.stderr) return None def main() -> None: parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) ask.add_common_arguments(parser) parser.add_argument("--questions", default="qa-questions.json") parser.add_argument("--judge-model", default=None, help="model that grades the answers; defaults to --model, which is worse") parser.add_argument("--no-judge", action="store_true", help="score without a judge model") parser.add_argument("--labbook", default=None, help="append JSON lines to this file") args = parser.parse_args() if args.judge_model is None: args.judge_model = args.model if not args.no_judge and args.judge_model == args.model: print("note: the judge and the model under test are the same. Self-preference bias " "makes that score optimistic; the lesson page explains why.", file=sys.stderr) spec = json.loads(Path(args.questions).read_text(encoding="utf-8")) questions = spec["questions"] db = ask.open_index(args.db) meta = ask.index_meta(db) if meta.get("embedding_model") and meta["embedding_model"] != args.embed_model: sys.exit(f"this index was built with {meta['embedding_model']}, not {args.embed_model}.") started = time.time() rows = [] tallies = Counter() by_category = {} for item in questions: record = ask.answer_question(item["question"], db, args) expected_answerable = bool(item["answerable"]) refusal_correct = record["answerable"] == expected_answerable if expected_answerable and record["answerable"]: contains = contains_all(record["answer"], item.get("must_contain", [])) exact = normalise(record["answer"]) == normalise(item["reference"]) cited_sources = {s["source"] for s in record["sources"]} source_correct = item.get("expected_source") in cited_sources else: contains = exact = source_correct = False score = None if not args.no_judge: score = judge(item["question"], item["reference"], record["answer"], args) row = { "lab": "part-10/project-private-document-qa/eval", "id": item["id"], "category": item["category"], "expected_answerable": expected_answerable, "answered": record["answerable"], "refusal_correct": refusal_correct, "contains": contains, "exact": exact, "source_correct": source_correct, "invalid_citations": record.get("invalid_citations", []), "judge_score": score, "seconds": record["seconds"], "model": args.model, "judge_model": None if args.no_judge else args.judge_model, "embed_model": args.embed_model, "reranked": not args.no_rerank, "candidates": args.candidates, "top_k": args.top_k, "temperature": args.temperature, "seed": args.seed, "date": time.strftime("%Y-%m-%d"), } rows.append(row) tallies["total"] += 1 tallies["refusal_correct"] += int(refusal_correct) tallies["contains"] += int(contains) tallies["source_correct"] += int(source_correct) tallies["fabricated_citation"] += int(bool(row["invalid_citations"])) bucket = by_category.setdefault(item["category"], Counter()) bucket["total"] += 1 bucket["ok"] += int(refusal_correct and (contains or not expected_answerable)) flag = "ok " if refusal_correct and (contains or not expected_answerable) else "BAD" print(f" {flag} {item['id']} [{item['category']:11s}] " f"judge={score if score is not None else '-'} {item['question'][:56]}") elapsed = time.time() - started scored = [r["judge_score"] for r in rows if r["judge_score"] is not None] summary = { "lab": "part-10/project-private-document-qa/eval-summary", "questions": tallies["total"], "refusal_correct": tallies["refusal_correct"], "contains": tallies["contains"], "source_correct": tallies["source_correct"], "fabricated_citations": tallies["fabricated_citation"], "judge_mean": round(sum(scored) / len(scored), 2) if scored else None, "by_category": {k: dict(v) for k, v in by_category.items()}, "model": args.model, "judge_model": None if args.no_judge else args.judge_model, "embed_model": args.embed_model, "reranked": not args.no_rerank, "candidates": args.candidates, "top_k": args.top_k, "temperature": args.temperature, "seed": args.seed, "seconds": round(elapsed, 1), "date": time.strftime("%Y-%m-%d"), } print("\n" + json.dumps(summary, indent=2)) if args.labbook: with Path(args.labbook).open("a", encoding="utf-8") as handle: for row in rows: handle.write(json.dumps(row) + "\n") handle.write(json.dumps(summary) + "\n") print(f"\nrecorded {len(rows) + 1} line(s) in {args.labbook}") # A fabricated citation is a correctness failure, not a warning. sys.exit(1 if tallies["fabricated_citation"] else 0) if __name__ == "__main__": main()