ROMFASTSQL/proxmox/lxc171-claude-agent/maria-whatsapp-bridge/ops/calibrate-rank.py

#!/usr/bin/env python3
"""Calibreaza pragurile de escaladare pe indexul REAL, nu pe intuitie.

De ce exista: pragul care decide "avem raspunsul in documente" nu se poate alege
din burta. Marginea dintre o intrebare buna si una straina e de cateva sutimi de
cosinus si se MUTA cand se schimba documentele — un prag bun azi poate escalada
maine jumatate din intrebarile legitime.

Cum se foloseste:

    MARIA_BRIDGE_DIR=$HOME/.maria-bridge ~/.maria-bridge/venv/bin/python \\
        ops/calibrate-rank.py                 # raport pe setul de cazuri
    ... ops/calibrate-rank.py --sweep          # matura grila de praguri
    ... ops/calibrate-rank.py --cases fisier.json

Embeddings-urile intrebarilor se tin in cache pe disc (`~/.maria-bridge/
calibrare-cache.json`): pe CPU, o intrebare costa secunde, iar maturarea grilei
le-ar recalcula de zeci de ori degeaba.

Setul de cazuri e deliberat mic si scris de mana. Nu e un benchmark, e o plasa de
siguranta: daca un prag nou strica un caz care mergea, se vede imediat.
"""

from __future__ import annotations

import argparse
import json
import pathlib
import sys

sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent / "rag"))

import config  # noqa: E402
import consumer  # noqa: E402
import rank  # noqa: E402

# (intrebare, are raspuns in documente?)
CAZURI: list[tuple[str, bool]] = [
    ("token efactura expirat", True),
    ("Nu exista nici un CIF pentru care sa aveti drept in SPV", True),
    ("cum trimit declaratia D406 SAF-T", True),
    ("eroare la semnatura electronica in SPV", True),
    ("nu pot incarca factura in SPV, imi da eroare de certificat", True),
    ("cum inregistrez o factura de achizitie in ROA", True),
    ("bilant contabil", True),
    ("import de date", True),
    ("care e capitala Frantei", False),
    ("cat costa un bilet de avion la Paris", False),
    ("imi da eroare la imprimanta HP LaserJet, nu trage hartia", False),
    ("cum schimb uleiul la masina", False),
    ("vremea de maine in Bucuresti", False),
    ("cum imi resetez parola de la Windows", False),
    ("imi da eroare 0x80070005 la instalarea unui joc pe Steam", False),
]

CACHE = config.STATE_DIR / "calibrare-cache.json"


def incarca_cache() -> dict[str, list[float]]:
    try:
        return json.loads(CACHE.read_text(encoding="utf-8"))
    except (OSError, ValueError):
        return {}


def pregateste(cazuri: list[tuple[str, bool]]) -> tuple[consumer.Index, list[dict]]:
    index = consumer.load_index()
    if not len(index):
        sys.exit(f"index gol sau lipsa: {config.INDEX_FILE}")
    cache = incarca_cache()
    date = []
    nou = 0
    for intrebare, asteptat in cazuri:
        vec = cache.get(intrebare)
        if vec is None:
            print(f"  embed: {intrebare[:50]}…", file=sys.stderr)
            vec = consumer.embed(intrebare)
            cache[intrebare] = vec
            nou += 1
        cos = [consumer.cosine(vec, e["embedding"]) for e in index.entries]
        bm = index.bm25.scores(intrebare)
        ordine = rank.rank(cos, bm)[: config.get_int("TOP_K", 3)]
        date.append({
            "intrebare": intrebare,
            "asteptat": asteptat,
            "best": max(cos),
            "top_texts": [index.texts[i] for i in ordine],
            "top_surse": [index.sources[i] for i in ordine],
        })
    if nou:
        CACHE.write_text(json.dumps(cache), encoding="utf-8")
        print(f"  ({nou} embeddings noi puse in cache)", file=sys.stderr)
    return index, date


def evalueaza(index: consumer.Index, date: list[dict], strong: float, weak: float,
              min_ratio: float) -> tuple[int, list[str]]:
    corecte, gresite = 0, []
    for d in date:
        ok, _ = rank.assess(
            d["intrebare"], d["best"], d["top_texts"], index.bm25,
            strong=strong, weak=weak, min_ratio=min_ratio,
        )
        if ok == d["asteptat"]:
            corecte += 1
        else:
            gresite.append(f"{'ar fi raspuns' if ok else 'ar fi escaladat'}: {d['intrebare']}")
    return corecte, gresite


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--sweep", action="store_true", help="matura grila de praguri")
    ap.add_argument("--cases", help="JSON cu [[intrebare, true/false], ...]")
    args = ap.parse_args()

    cazuri = CAZURI
    if args.cases:
        cazuri = [(q, bool(a)) for q, a in json.loads(pathlib.Path(args.cases).read_text())]

    index, date = pregateste(cazuri)
    print(f"index: {len(index)} chunk-uri | cazuri: {len(date)}\n")

    if args.sweep:
        rezultate = []
        for strong in (0.62, 0.66, 0.68, 0.70, 0.72, 0.75):
            for weak in (0.54, 0.58, 0.60):
                for ratio in (0.0, 0.34, 0.5, 0.67, 1.0):
                    if weak >= strong:
                        continue
                    corecte, _ = evalueaza(index, date, strong, weak, ratio)
                    rezultate.append((corecte, strong, weak, ratio))
        rezultate.sort(key=lambda r: (-r[0], r[1]))
        print(f"{'corecte':>8}  strong  weak  ratio")
        for corecte, strong, weak, ratio in rezultate[:15]:
            print(f"{corecte:>5}/{len(date)}   {strong:.2f}  {weak:.2f}  {ratio:.2f}")
        return

    strong, weak = rank.strong_cosine(), rank.weak_cosine()
    ratio = rank.min_rare_ratio()
    print(f"praguri curente: strong={strong} weak={weak} ratio={ratio}\n")
    for d in date:
        ok, motiv = rank.assess(d["intrebare"], d["best"], d["top_texts"], index.bm25)
        semn = "OK    " if ok == d["asteptat"] else "GRESIT"
        stare = "ACOPERIT " if ok else "ESCALADAT"
        print(f"{semn} [{stare}] {d['intrebare'][:44]:46} {motiv}")
    corecte, gresite = evalueaza(index, date, strong, weak, ratio)
    print(f"\n{corecte}/{len(date)} corecte")
    for g in gresite:
        print(f"  - {g}")


if __name__ == "__main__":
    main()