bench/bench_mathe.py

#!/usr/bin/env python3
"""MATH-500 und GSM8K gegen einen OpenAI-kompatiblen Server.

Warum Mathe zusaetzlich zu HumanEval+: dort erreichten alle sieben Varianten
88-91 Prozent und waren statistisch ununterscheidbar. Mathematik trennt
staerker, vor allem bei den hoeheren Schwierigkeitsstufen, und ist der Ort,
an dem sich zeigt, ob ein Denkblock etwas bringt.

Die Bewertung ist die eigentliche Arbeit: MATH-500 erwartet LaTeX-Antworten
(`\\frac{3}{4}`, `\\left( 3, \\frac{\\pi}{2} \\right)`), und dieselbe Zahl kann
in einem Dutzend Schreibweisen kommen. Deshalb wird normalisiert statt
stur verglichen — und wo Normalisierung nicht reicht, zusaetzlich numerisch
ausgewertet.
"""
import argparse
import json
import os
import re
import sys
import time
import urllib.request
from fractions import Fraction

DS = "/mnt/models/qwen38-prep/datasets"
REASONING = False
EFFORT = ""  # nur wirksam wenn REASONING; "xhigh" ist die tiefste Stufe
MODELL_ID = ""  # gesetzt: Feld "model" wird mitgeschickt (fuer PropellerA & Co.)
DRY = 0.0       # DRY-Multiplikator gegen Wiederholungsschleifen, 0 = aus
DRY_LAENGE = 8  # erst ab dieser Wiederholungslaenge bestrafen


def lade(satz, grenze):
    zeilen = [json.loads(z) for z in open(os.path.join(DS, f"{satz}.jsonl"))
              if z.strip()]
    if not grenze or grenze >= len(zeilen):
        return zeilen
    # Gleichmaessig ueber den Datensatz greifen statt die ersten N: MATH-500
    # ist nach Thema sortiert, die ersten 150 waeren fast nur Precalculus.
    schritt = len(zeilen) / grenze
    return [zeilen[int(i * schritt)] for i in range(grenze)]


def bau_prompt(satz, a):
    if satz.startswith("math"):
        return (a["problem"] + "\n\nLöse die Aufgabe. Schreibe die "
                "endgültige Antwort ganz zum Schluss in der Form "
                "\\boxed{ANTWORT}.")
    return (a["question"] + "\n\nLöse die Aufgabe Schritt für Schritt. "
            "Schreibe die endgültige Zahl ganz zum Schluss in der Form "
            "#### ZAHL.")


def frage(port, prompt, max_tokens, timeout):
    b = {"messages": [{"role": "user", "content": prompt}],
         "max_tokens": max_tokens, "cache_prompt": False,
         "temperature": 0.2, "top_p": 0.9}
    # Ein blanker llama-server braucht kein "model", eine Fassade davor schon.
    # Damit laesst sich derselbe Messlauf gegen die Produktion fahren, ohne
    # den Stack fuer einen Testserver abzuraeumen.
    if MODELL_ID:
        b["model"] = MODELL_ID
    # DRY bestraft wiederholte Tokenfolgen und ist das einzige Mittel gegen den
    # Kollaps vom 17.08. (' ? x= ? x= …' hunderte Male, danach leere Antwort).
    # Die erlaubte Laenge steht bewusst hoch: in einem Rechenweg wiederholen
    # sich kurze Folgen voellig zu Recht. Bestraft werden soll der Zyklus, nicht
    # die Rechnung.
    if DRY > 0:
        b["dry_multiplier"] = DRY
        b["dry_base"] = 1.75
        b["dry_allowed_length"] = DRY_LAENGE
        b["dry_penalty_last_n"] = -1
    if not REASONING:
        b["chat_template_kwargs"] = {"enable_thinking": False}
    elif EFFORT:
        # Qwen3.8 steuert die Denktiefe ueber reasoning_effort; xhigh ist die
        # Voreinstellung und die tiefste Stufe. Wir setzen sie trotzdem
        # ausdruecklich: die Voreinstellung steckt in der Chat-Vorlage, und
        # eine andere Vorlage aendert sie still. Was gemessen wurde, soll in
        # der Anfrage stehen und nicht in einer Datei daneben.
        b["chat_template_kwargs"] = {"reasoning_effort": EFFORT}
    req = urllib.request.Request(
        f"http://127.0.0.1:{port}/v1/chat/completions",
        data=json.dumps(b).encode(),
        headers={"Content-Type": "application/json"})
    t0 = time.time()
    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            d = json.load(r)
    except Exception as e:
        return "", dict(tps=0.0, tokens=0, sekunden=time.time() - t0,
                        abbruch="fehler"), f"{type(e).__name__}: {e}"
    m = d["choices"][0]["message"]
    t = d.get("timings", {})
    # vLLM liefert keine timings — dann aus usage und Wanduhr rechnen.
    sek = time.time() - t0
    tok = t.get("predicted_n") or d.get("usage", {}).get("completion_tokens") or 0
    tps = t.get("predicted_per_second") or (tok / sek if sek > 0 else 0.0)
    return (m.get("content") or ""), dict(
        tps=tps, tokens=tok,
        denk_zeichen=len(m.get("reasoning_content") or ""),
        abbruch=d["choices"][0].get("finish_reason") or "",
        sekunden=round(sek, 2)), ""


# --- Antwortextraktion -----------------------------------------------------

def hole_boxed(text):
    """Inhalt des LETZTEN \\boxed{...}. Klammern werden mitgezaehlt, weil
    verschachtelte Ausdruecke wie \\boxed{\\frac{1}{2}} sonst abbrechen."""
    stelle = text.rfind("\\boxed")
    if stelle < 0:
        return None
    i = text.find("{", stelle)
    if i < 0:
        return None
    tiefe, j = 0, i
    while j < len(text):
        if text[j] == "{":
            tiefe += 1
        elif text[j] == "}":
            tiefe -= 1
            if tiefe == 0:
                return text[i + 1:j]
        j += 1
    return None


def hole_gsm(text):
    m = re.findall(r"####\s*([-\d.,]+)", text)
    if m:
        return m[-1]
    # Kein Marker: die letzte Zahl im Text nehmen.
    z = re.findall(r"-?\d[\d.,]*", text)
    return z[-1] if z else None


def klammere_frac(s):
    """Bringt jedes \\frac auf die Form \\frac{..}{..}.

    Vorkommende Schreibweisen: \\frac12, \\frac{270}7, \\frac2{15}. Ohne diese
    Vereinheitlichung vergleicht man Klammersetzung statt Zahlen.
    """
    muster = re.compile(r"\\frac(\{[^{}]*\}|[^{}])(\{[^{}]*\}|[^{}])")

    def ersetze(m):
        teile = []
        for g in (m.group(1), m.group(2)):
            teile.append(g if g.startswith("{") else "{" + g + "}")
        return "\\frac" + "".join(teile)

    vorher = None
    while vorher != s:
        vorher = s
        s = muster.sub(ersetze, s)
    return s


def normalisiere(s):
    """Bringt LaTeX-Antworten auf eine vergleichbare Form.

    Die Faelle hier stammen aus echten Fehlschlaegen des ersten Durchgangs:
    von acht als falsch gewerteten Antworten war genau eine wirklich falsch,
    die uebrigen sieben waren Schreibweisen. `\\frac43` gegen `\\frac{4}{3}`,
    `.35625` gegen `0,35625`, `x=5` gegen `5`. Wer das nicht abfaengt, misst
    LaTeX-Konventionen statt Mathematik.
    """
    if s is None:
        return None
    s = s.strip()
    # \frac43 und \frac 34 -> \frac{4}{3}: die Kurzform ohne Klammern ist in
    # MATH-500 haeufig und sonst nicht mit der Langform vergleichbar.
    s = re.sub(r"\\(d|t)?frac\s*(\d)\s*(\d)", r"\\frac{\2}{\3}", s)
    # ... und der gemischte Fall \frac{270}7, den die Regel oben nicht trifft,
    # weil sie zwei nackte Ziffern verlangt. Gefunden am 17.08. in der
    # Bedarfsmessung: erwartet '\frac{270}7', geliefert '\frac{270}{7}' —
    # als falsch gewertet, obwohl dieselbe Zahl.
    s = klammere_frac(s)
    for weg in ("\\left", "\\right", "\\!", "\\,", "\\;", "\\ ", "$", " "):
        s = s.replace(weg, "")
    s = s.replace("dfrac", "frac").replace("tfrac", "frac")
    s = s.replace("^{\\circ}", "").replace("^\\circ", "")
    s = re.sub(r"\\text\{([^}]*)\}", r"\1", s)
    s = re.sub(r"\\mbox\{([^}]*)\}", r"\1", s)
    # Zuweisungspraefixe: "x=5" und "5" sind dieselbe Antwort, ebenso
    # "x\in[-2,7]" und "[-2,7]". Nur bei einem einzelnen Buchstaben links,
    # damit Gleichungen als Antwort nicht zerstoert werden.
    s = re.sub(r"^[a-zA-Z]\s*(=|\\in)\s*", "", s)
    # Multiple-Choice: "(B)" und "B" sind dasselbe.
    m = re.fullmatch(r"\(([a-eA-E])\)", s)
    if m:
        s = m.group(1)
    # Komma in Zahlen: Tausendertrenner oder Dezimaltrenner? Unterschieden
    # wird an der Stellenzahl — "1,000" ist Tausender (englische Konvention
    # des Datensatzes), "0,35625" kann es nicht sein. Die Dezimalpruefung
    # MUSS zuerst kommen, sonst frisst die Tausenderregel das Komma weg.
    if re.fullmatch(r"-?\d*,\d+", s) and not re.fullmatch(r"-?\d+,\d{3}", s):
        s = s.replace(",", ".")
    elif re.fullmatch(r"-?[\d,]+(\.\d+)?", s):
        s = s.replace(",", "")
    # Fuehrende Null: ".5" und "0.5" sind dieselbe Zahl.
    s = re.sub(r"^(-?)\.(\d)", r"\g<1>0.\2", s)
    s = s.rstrip(".")
    # Einheiten und Prozentzeichen am Ende sind fuer die Gleichheit egal.
    s = re.sub(r"(\\%|%)$", "", s)
    return s.lower()


def als_zahl(s):
    """Versucht, einen Ausdruck numerisch auszuwerten — nur die Formen, die in
    diesen Datensaetzen vorkommen. Kein eval, das waere ein Sicherheitsloch."""
    if s is None:
        return None
    s = s.strip()
    m = re.fullmatch(r"\\frac\{(-?[\d.]+)\}\{(-?[\d.]+)\}", s)
    if m:
        try:
            return float(Fraction(m.group(1)) / Fraction(m.group(2)))
        except Exception:
            return None
    m = re.fullmatch(r"(-?[\d.]+)/(-?[\d.]+)", s)
    if m:
        try:
            return float(Fraction(m.group(1)) / Fraction(m.group(2)))
        except Exception:
            return None
    try:
        return float(s)
    except ValueError:
        return None


def ohne_latex_abstand(s):
    """Entfernt den optionalen Zeilenabstand nach \\\\ in Matrizen.

    Das Modell schreibt '\\\\[2mm]', die Musterloesung nur '\\\\'. Reine
    Typografie, kein Rechenunterschied.
    """
    return re.sub(r"\\\\\s*\[[^\]]*\]", r"\\\\", s)


def frac_zu_slash(s):
    """\\frac{16}{49} -> 16/49.

    Innerhalb einer Matrix greift die Zahlenumrechnung nicht, weil der
    Gesamtausdruck keine Zahl ist. Dann bleibt nur, beide Seiten auf dieselbe
    Bruchschreibweise zu bringen.
    """
    return re.sub(r"\\d?frac\{([^{}]*)\}\{([^{}]*)\}", r"\1/\2", klammere_frac(s))


def pm_ausschreiben(s):
    """'1 \\pm \\sqrt{19}' -> beide Reihenfolgen von '1-\\sqrt{19},1+\\sqrt{19}'.

    Ein Modell listet die zwei Loesungen oft einzeln auf, die Musterloesung
    fasst sie mit \\pm zusammen. Beide Reihenfolgen, weil ungeordnete
    Loesungsmengen keine feste Reihenfolge haben — im Gegensatz zu Koordinaten,
    die hier bewusst NICHT angefasst werden (es wird nur \\pm ersetzt, nie
    ein Komma umsortiert).
    """
    if s.count("\\pm") != 1:
        return set()
    links, rechts = s.split("\\pm", 1)
    links, rechts = links.strip(), rechts.strip()
    if not links or not rechts:
        return set()
    return {f"{links}-{rechts},{links}+{rechts}",
            f"{links}+{rechts},{links}-{rechts}"}


def varianten(s):
    """Alle zulaessigen Normalformen einer Antwort.

    Warum eine Menge statt einer einzigen Form: MATH-500 haengt Einheiten mal
    an die Musterloesung an und mal nicht — '\\frac{270}7\\text{ degrees}'
    gegen '\\frac{270}{7}'. Beides ist dieselbe Antwort. Abgeschnitten wird
    ausschliesslich ein \\text{..} bzw. \\mbox{..} ganz am Ende; ein Modell,
    das nur 'degrees' antwortet, verliert dadurch nichts, weil der Rest dann
    leer waere und die Variante entfaellt.
    """
    if s is None:
        return set()
    roh = s.strip()
    grundformen = {roh}
    ohne_einheit = re.sub(r"(\\text|\\mbox)\s*\{[^{}]*\}\s*$", "", roh).strip()
    if ohne_einheit:
        grundformen.add(ohne_einheit)

    formen = set()
    for g in grundformen:
        # Jede Grundform zusaetzlich ohne Typografie-Abstand und mit Bruechen
        # als Schraegstrich — beides sind Schreibweisen, keine Rechenschritte.
        for h in (g, ohne_latex_abstand(g)):
            formen.add(normalisiere(h))
            formen.add(normalisiere(frac_zu_slash(h)))
            for p in pm_ausschreiben(h):
                formen.add(normalisiere(p))
    return {f for f in formen if f}


def stimmt(gegeben, erwartet):
    va, vb = varianten(gegeben), varianten(erwartet)
    if not va or not vb:
        return False
    if va & vb:
        return True
    for a in va:
        for b in vb:
            za, zb = als_zahl(a), als_zahl(b)
            if za is not None and zb is not None and abs(za - zb) < 1e-6:
                return True
    a, b = normalisiere(gegeben), normalisiere(erwartet)
    if a is None or b is None:
        return False
    za, zb = als_zahl(a), als_zahl(b)
    if za is not None and zb is not None:
        return abs(za - zb) < 1e-6
    return False


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--port", type=int, default=8291)
    ap.add_argument("--modell", required=True)
    ap.add_argument("--satz", default="math500")
    ap.add_argument("--out", required=True)
    ap.add_argument("--grenze", type=int, default=0)
    ap.add_argument("--max-tokens", type=int, default=1600)
    ap.add_argument("--timeout", type=int, default=600)
    ap.add_argument("--reasoning", action="store_true")
    ap.add_argument("--effort", default="",
                    choices=["", "low", "medium", "xhigh"],
                    help="reasoning_effort, nur mit --reasoning. Qwen3.8 kennt "
                         "genau diese drei Stufen; 'high' gibt es NICHT und "
                         "die Chat-Vorlage verwirft unbekannte Werte still.")
    ap.add_argument("--modell-id", default="",
                    help="Wert fuer das Feld 'model' — noetig hinter einer "
                         "Fassade wie PropellerA, unnoetig bei llama-server")
    ap.add_argument("--dry", type=float, default=0.0,
                    help="DRY-Multiplikator gegen Wiederholungsschleifen "
                         "(0 = aus, 0.8 ist ein ueblicher Wert)")
    ap.add_argument("--dry-laenge", type=int, default=8,
                    help="erlaubte Wiederholungslaenge vor der Bestrafung")
    args = ap.parse_args()

    global REASONING, EFFORT, MODELL_ID, DRY, DRY_LAENGE
    REASONING = args.reasoning
    EFFORT = args.effort
    MODELL_ID = args.modell_id
    DRY = args.dry
    DRY_LAENGE = args.dry_laenge
    aufgaben = lade(args.satz, args.grenze)
    print(f"{args.modell} | {args.satz} | {len(aufgaben)} Aufgaben | "
          f"Reasoning {'AN' if REASONING else 'AUS'}", file=sys.stderr)

    zeilen, ok_n = [], 0
    t0 = time.time()
    for i, a in enumerate(aufgaben, 1):
        erwartet = a.get("answer") if args.satz.startswith("math") \
            else hole_gsm(a.get("answer", ""))
        txt, m, fehler = frage(args.port, bau_prompt(args.satz, a),
                               args.max_tokens, args.timeout)
        if fehler:
            zeilen.append(dict(id=str(a.get("unique_id", i)), ok=False,
                               grund=fehler, **m))
            print(f"  [{i:3d}] SERVERFEHLER {fehler[:50]}", file=sys.stderr)
            continue
        gegeben = hole_boxed(txt) if args.satz.startswith("math") \
            else hole_gsm(txt)
        ok = stimmt(gegeben, erwartet)
        ok_n += ok
        zeilen.append(dict(id=str(a.get("unique_id", i)), ok=ok,
                           level=a.get("level"), thema=a.get("subject"),
                           # 400 statt 60: bei 60 war die pmatrix-Antwort vom
                           # 17.08. im Bericht abgeschnitten und liess sich
                           # nachtraeglich nicht mehr neu bewerten. Verglichen
                           # wird die volle Antwort, gespeichert werden musste
                           # sie eben auch.
                           gegeben=(gegeben or "")[:400],
                           erwartet=(str(erwartet) or "")[:60], **m))
        if i % 20 == 0 or not ok:
            print(f"  [{i:3d}/{len(aufgaben)}] "
                  f"{'OK' if ok else 'FEHL: ' + repr(gegeben)[:26] + ' statt ' + repr(erwartet)[:26]}"
                  f"   (bisher {ok_n}/{i} = {ok_n/i*100:.0f}%)", file=sys.stderr)

    dauer = time.time() - t0
    tps = sorted(x["tps"] for x in zeilen if x["tps"] > 0)
    # Aufschluesselung nach Schwierigkeitsgrad — der eigentliche Mehrwert
    # gegenueber einer einzelnen Prozentzahl.
    nach_level = {}
    for x in zeilen:
        lv = x.get("level")
        if lv is not None:
            d = nach_level.setdefault(str(lv), [0, 0])
            d[1] += 1
            d[0] += 1 if x["ok"] else 0
    bericht = dict(
        modell=args.modell, satz=args.satz, reasoning=REASONING,
        effort=EFFORT or None, max_tokens=args.max_tokens,
        dry=DRY or None, dry_laenge=DRY_LAENGE if DRY else None,
        n=len(aufgaben), bestanden=ok_n,
        quote=round(ok_n / len(aufgaben) * 100, 1) if aufgaben else 0.0,
        tps_median=round(tps[len(tps) // 2], 1) if tps else 0.0,
        tokens_je_aufgabe=round(sum(x["tokens"] for x in zeilen) / len(zeilen), 1)
                          if zeilen else 0,
        am_limit=sum(1 for x in zeilen if x.get("abbruch") == "length"),
        nach_level={k: v for k, v in sorted(nach_level.items())},
        dauer_s=round(dauer, 1), aufgaben=zeilen)
    with open(args.out, "w") as f:
        json.dump(bericht, f, ensure_ascii=False, indent=1)
    lv = "  ".join(f"L{k}: {v[0]}/{v[1]}" for k, v in bericht["nach_level"].items())
    print(f"\n{args.modell} | {args.satz}: {ok_n}/{len(aufgaben)} = "
          f"{bericht['quote']}%  |  {bericht['tps_median']} t/s  |  "
          f"{bericht['tokens_je_aufgabe']:.0f} Token/Aufgabe  |  "
          f"{bericht['am_limit']} abgeschnitten  |  {dauer/60:.1f} min",
          file=sys.stderr)
    if lv:
        print(f"  nach Schwierigkeit: {lv}", file=sys.stderr)


if __name__ == "__main__":
    main()