scripts/make_eval_set.py

"""Generate a labelled grading eval set (T3.2).

Two paths:

  1. Synthetic (default, no dependencies, always runs): builds valid, gold
     graded answers with claims, criteria, and correct per-criterion scores.
     These are the "clean" (negative) class for the checker evaluation and
     the ground truth for scorer agreement. Deterministic given --seed.

  2. Giskard RAGET (optional): if `giskard` is installed and an LLM key is
     set, generate question/reference/context triples from a knowledge base.
     Kept as a documented function; it needs network + keys, so the synthetic
     path is the runnable default that unblocks scripts/eval.py immediately.

Each record (one JSON object per line):
    {
      "id": str, "question": str, "max_marks": float,
      "criteria": [ {name, points, conditions, ...} ],
      "claims":   [ {step, content, concern} ],          # what the student wrote
      "gold_per_criterion": [ {name, awarded, max, reasoning, claim_refs} ]
    }

Usage:
    python scripts/make_eval_set.py --out data/eval/set.jsonl --n 12 --seed 7
"""
from __future__ import annotations

import argparse
import json
import random
from pathlib import Path


# Base problems: each has claims (numbered by step) and criteria whose points
# are grounded in specific claims. gold_per_criterion is the correct grading.
_BASES = [
    {
        "question": "Find the antiderivative F(x) of f(x)=2x and evaluate F(2)-F(0).",
        "max_marks": 3.0,
        "criteria": [
            {"name": "Antiderivative", "points": 1.0, "conditions": "F(x)=x^2 (+C)"},
            {"name": "Evaluation", "points": 1.0, "conditions": "F(2)-F(0)=4"},
            {"name": "Constant handled", "points": 1.0, "conditions": "constant cancels or noted"},
        ],
        "claims": [
            {"step": 1, "content": "F(x) = x^2 + C", "concern": ""},
            {"step": 2, "content": "F(2) = 4 + C, F(0) = C", "concern": ""},
            {"step": 3, "content": "F(2) - F(0) = 4", "concern": ""},
        ],
        "gold": [
            {"name": "Antiderivative", "awarded": 1.0, "max": 1.0,
             "reasoning": "claim 1 gives F(x)=x^2+C", "claim_refs": [1]},
            {"name": "Evaluation", "awarded": 1.0, "max": 1.0,
             "reasoning": "claims 2 and 3 evaluate the difference to 4", "claim_refs": [2, 3]},
            {"name": "Constant handled", "awarded": 1.0, "max": 1.0,
             "reasoning": "claim 2 shows C cancels", "claim_refs": [2]},
        ],
    },
    {
        "question": "Differentiate g(x)=x^3 and give g'(2).",
        "max_marks": 2.0,
        "criteria": [
            {"name": "Derivative", "points": 1.0, "conditions": "g'(x)=3x^2"},
            {"name": "Value", "points": 1.0, "conditions": "g'(2)=12"},
        ],
        "claims": [
            {"step": 1, "content": "g'(x) = 3x^2", "concern": ""},
            {"step": 2, "content": "g'(2) = 3*4 = 12", "concern": ""},
        ],
        "gold": [
            {"name": "Derivative", "awarded": 1.0, "max": 1.0,
             "reasoning": "claim 1 gives 3x^2", "claim_refs": [1]},
            {"name": "Value", "awarded": 1.0, "max": 1.0,
             "reasoning": "claim 2 gives 12", "claim_refs": [2]},
        ],
    },
    {
        "question": "Solve 2x+3=11 for x.",
        "max_marks": 2.0,
        "criteria": [
            {"name": "Isolate", "points": 1.0, "conditions": "2x=8"},
            {"name": "Solve", "points": 1.0, "conditions": "x=4"},
        ],
        "claims": [
            {"step": 1, "content": "2x = 11 - 3 = 8", "concern": ""},
            {"step": 2, "content": "x = 4", "concern": ""},
        ],
        "gold": [
            {"name": "Isolate", "awarded": 1.0, "max": 1.0,
             "reasoning": "claim 1 isolates 2x=8", "claim_refs": [1]},
            {"name": "Solve", "awarded": 1.0, "max": 1.0,
             "reasoning": "claim 2 gives x=4", "claim_refs": [2]},
        ],
    },
]


def _vary(base: dict, rng: random.Random, idx: int) -> dict:
    """Produce a valid variant with a possibly-partial (still grounded) score."""
    gold = [dict(pc) for pc in base["gold"]]
    # Randomly make one criterion partially/zero earned, but keep it grounded:
    # a zero-award criterion cites no claim (which is valid: nothing to ground).
    if rng.random() < 0.5 and gold:
        i = rng.randrange(len(gold))
        if rng.random() < 0.5:
            gold[i]["awarded"] = 0.0
            gold[i]["reasoning"] = "not shown"
            gold[i]["claim_refs"] = []
        else:
            gold[i]["awarded"] = round(gold[i]["max"] * 0.5, 3)
    return {
        "id": f"q{_BASES.index(base)+1}_a{idx}",
        "question": base["question"],
        "max_marks": base["max_marks"],
        "criteria": base["criteria"],
        "claims": base["claims"],
        "gold_per_criterion": gold,
    }


def generate_synthetic(n: int, seed: int = 7) -> list[dict]:
    rng = random.Random(seed)
    out = []
    for i in range(n):
        base = _BASES[i % len(_BASES)]
        out.append(_vary(base, rng, i))
    return out


def generate_with_giskard(knowledge_texts: list[str], n: int = 20) -> list[dict]:
    """Optional: Giskard RAGET generation. Needs `giskard` + an LLM key.

    Returns [] with a printed note if giskard is unavailable, so callers can
    fall back to the synthetic set. This is a stub of the wiring, not run in
    CI; adapt the mapping from RAGET's question/reference/context to the
    record schema above once you point it at your real knowledge base.
    """
    try:
        from giskard.rag import KnowledgeBase, generate_testset  # type: ignore
    except Exception as e:
        print(f"[giskard] unavailable ({e}); use the synthetic set instead")
        return []
    import pandas as pd  # noqa

    kb = KnowledgeBase(pd.DataFrame({"text": knowledge_texts}))
    testset = generate_testset(kb, num_questions=n)
    records = []
    for i, row in enumerate(testset.to_pandas().itertuples()):
        # Map RAGET output to our record schema. Reference/context become the
        # gold answer/claims for a single free-form criterion.
        records.append({
            "id": f"raget_{i}",
            "question": getattr(row, "question", ""),
            "max_marks": 1.0,
            "criteria": [{"name": "Correctness", "points": 1.0,
                          "conditions": "matches the reference answer"}],
            "claims": [{"step": 1, "content": getattr(row, "reference_context", ""),
                        "concern": ""}],
            "gold_per_criterion": [{"name": "Correctness", "awarded": 1.0, "max": 1.0,
                                    "reasoning": "claim 1 matches the reference",
                                    "claim_refs": [1]}],
        })
    return records


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--out", default="data/eval/set.jsonl")
    ap.add_argument("--n", type=int, default=12)
    ap.add_argument("--seed", type=int, default=7)
    args = ap.parse_args()

    records = generate_synthetic(args.n, args.seed)
    out = Path(args.out)
    out.parent.mkdir(parents=True, exist_ok=True)
    with out.open("w", encoding="utf-8") as f:
        for r in records:
            f.write(json.dumps(r) + "\n")
    total_points = sum(c["points"] for r in records for c in r["criteria"])
    print(f"wrote {len(records)} records to {out}")
    print(f"  distinct problems: {len({r['question'] for r in records})}")
    print(f"  total gold points: {total_points}")


if __name__ == "__main__":
    main()