| """Yürütme doğruluğu (execution accuracy, EX) tabanlı değerlendirme. |
| |
| EX: sistemin SQL'i kendi (çıkarımla oluşturulmuş) veritabanında çalıştırıldığında dönen sonuç |
| kümesi, altın SQL'in referans veritabanındaki sonucuyla eşleşiyorsa 1. |
| - Satır sırası önemsiz (çoklu küme karşılaştırması) |
| - Sayılar: |a-b| <= 0,01 + 0,001·|b| (yuvarlama farklarına tolerans) |
| - Metinler: Türkçe katlanmış karşılaştırma (İ/ı, büyük/küçük harf) |
| - Sistem fazladan sütun döndürürse, altın sütunlarla eşleşen bir sütun alt kümesi aranır |
| Cevapsız sorularda (altın SQL yok) sistem cevap vermeyi reddederse doğru sayılır; |
| cevaplanabilir sorularda reddetmek yanlıştır. |
| """ |
| from __future__ import annotations |
|
|
| import itertools |
| import json |
| import time |
| from collections import Counter, defaultdict |
| from dataclasses import dataclass |
| from pathlib import Path |
|
|
| from .agent import Text2SQLAgent |
| from .errors import ErrorLog, diagnose_failure |
| from .schema import connect |
| from .trnum import fold |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| BENCH = ROOT / "benchmark" / "tr_tablosoru.json" |
|
|
|
|
| def load_benchmark(path=BENCH) -> list[dict]: |
| return json.loads(Path(path).read_text(encoding="utf-8"))["sorular"] |
|
|
|
|
| def _val_eq(a, b) -> bool: |
| if a is None or b is None: |
| return a is None and b is None |
| na = isinstance(a, (int, float)) and not isinstance(a, bool) |
| nb = isinstance(b, (int, float)) and not isinstance(b, bool) |
| if na and nb: |
| return abs(a - b) <= 0.01 + 1e-3 * abs(b) |
| if na != nb: |
| return False |
| return fold(str(a)) == fold(str(b)) |
|
|
|
|
| def _sort_key(row): |
| return tuple((0, round(v, 1)) if isinstance(v, (int, float)) and not isinstance(v, bool) |
| else (1, fold(str(v))) for v in row) |
|
|
|
|
| def _rows_eq(p, g) -> bool: |
| if len(p) != len(g): |
| return False |
| ps, gs = sorted(p, key=_sort_key), sorted(g, key=_sort_key) |
| if all(all(_val_eq(x, y) for x, y in zip(rp, rg)) for rp, rg in zip(ps, gs)): |
| return True |
| |
| used = [False] * len(gs) |
| for rp in p: |
| for k, rg in enumerate(gs): |
| if not used[k] and all(_val_eq(x, y) for x, y in zip(rp, rg)): |
| used[k] = True |
| break |
| else: |
| return False |
| return True |
|
|
|
|
| def results_match(pred: list[tuple], gold: list[tuple]) -> bool: |
| if pred is None: |
| return False |
| if not gold: |
| return not pred |
| gw = len(gold[0]) |
| if not pred: |
| return False |
| pw = len(pred[0]) |
| if pw == gw and _rows_eq(pred, gold): |
| return True |
| if pw > gw and pw <= 6: |
| for idx in itertools.permutations(range(pw), gw): |
| if _rows_eq([tuple(r[i] for i in idx) for r in pred], gold): |
| return True |
| if pw == gw and gw > 1: |
| for idx in itertools.permutations(range(pw)): |
| if _rows_eq([tuple(r[i] for i in idx) for r in pred], gold): |
| return True |
| return False |
|
|
|
|
| @dataclass |
| class SystemSpec: |
| name: str |
| backend: object |
| db_mode: str = "tr" |
| self_correct: bool = True |
| use_evidence: bool = True |
| max_attempts: int = 3 |
| static_check: bool = True |
|
|
|
|
| def gold_results(items, ref_db) -> dict[str, list]: |
| con = connect(ref_db, readonly=True) |
| out = {} |
| for it in items: |
| out[it["id"]] = [tuple(r) for r in con.execute(it["altin_sql"]).fetchall()] if it["altin_sql"] else None |
| con.close() |
| return out |
|
|
|
|
| def evaluate_system(spec: SystemSpec, items, db_paths: dict, ref_db, out_dir: Path, |
| error_log: ErrorLog | None = None, progress=None) -> dict: |
| gold = gold_results(items, ref_db) |
| con = connect(db_paths[spec.db_mode], readonly=True) |
| agent = Text2SQLAgent(con, spec.backend, max_attempts=spec.max_attempts, self_correct=spec.self_correct, |
| static_check=spec.static_check, |
| error_log=error_log, log_source=f"degerlendirme:{spec.name}", |
| fallback_rule=False) |
| records = [] |
| for k, it in enumerate(items): |
| t0 = time.time() |
| res = agent.run(it["soru"], evidence=it.get("kanit") if spec.use_evidence else None) |
| g = gold[it["id"]] |
| if not it["cevaplanabilir"]: |
| ok = res.abstained |
| first_ok = res.first_sql is None |
| else: |
| ok = (not res.abstained) and results_match(res.rows, g) |
| first_ok = res.first_rows is not None and res.first_error is None and results_match(res.first_rows, g) |
| err = None if ok else diagnose_failure(it, res.sql, res.rows, g or [], res.abstained, res.exec_error, |
| spec.db_mode) |
| rec = {"id": it["id"], "bolum": it.get("bolum", "gelistirme"), "soru": it["soru"], |
| "zorluk": it["zorluk"], "kategoriler": it["kategoriler"], "cevaplanabilir": it["cevaplanabilir"], |
| "altin_sql": it["altin_sql"], "tahmin_sql": res.sql, "ilk_sql": res.first_sql, |
| "durum": res.status, "cevap": res.answer, "dogru": ok, "ilk_deneme_dogru": first_ok, |
| "deneme": res.attempts, "hata_tipi": err, |
| "calisma_hatalari": [e.tip for e in res.events], "sure": round(time.time() - t0, 3), |
| "tahmin_sonuc": [list(map(_jsonable, r)) for r in (res.rows or [])[:10]], |
| "altin_sonuc": [list(map(_jsonable, r)) for r in (g or [])[:10]]} |
| records.append(rec) |
| if progress: |
| progress(k + 1, len(items), rec) |
| con.close() |
| out_dir.mkdir(parents=True, exist_ok=True) |
| (out_dir / f"sonuclar_{spec.name}.jsonl").write_text( |
| "\n".join(json.dumps(r, ensure_ascii=False) for r in records), encoding="utf-8") |
| return summarize(spec, records) |
|
|
|
|
| def _jsonable(v): |
| return v if isinstance(v, (int, float, str)) or v is None else str(v) |
|
|
|
|
| def _ex(recs): |
| return round(100.0 * sum(r["dogru"] for r in recs) / len(recs), 1) if recs else None |
|
|
|
|
| def summarize(spec: SystemSpec, records: list[dict]) -> dict: |
| ans = [r for r in records if r["cevaplanabilir"]] |
| una = [r for r in records if not r["cevaplanabilir"]] |
| by = lambda key: {k: _ex(v) for k, v in sorted(_group(records, key).items())} |
| cats = defaultdict(list) |
| for r in records: |
| for c in r["kategoriler"]: |
| cats[c].append(r) |
| return { |
| "sistem": spec.name, "arka_uc": spec.backend if isinstance(spec.backend, str) else spec.backend.name, |
| "sayi_kipi": spec.db_mode, "kendini_duzeltme": spec.self_correct, "kanit": spec.use_evidence, |
| "n": len(records), "EX": _ex(records), "EX_ilk_deneme": round(100.0 * sum(r["ilk_deneme_dogru"] for r in records) / len(records), 1), |
| "EX_cevaplanabilir": _ex(ans), |
| "cevapsiz_dogru_red": _ex(una), |
| "yanlis_red_orani": round(100.0 * sum(1 for r in ans if r["durum"] != "cevap") / len(ans), 1) if ans else None, |
| "bolum": by("bolum"), "zorluk": by("zorluk"), |
| "kategori": {k: _ex(v) for k, v in sorted(cats.items())}, |
| "kategori_n": {k: len(v) for k, v in sorted(cats.items())}, |
| "hata_tipleri": dict(Counter(r["hata_tipi"] for r in records if r["hata_tipi"]).most_common()), |
| "ort_deneme": round(sum(r["deneme"] for r in records) / len(records), 2), |
| "ort_sure_sn": round(sum(r["sure"] for r in records) / len(records), 3), |
| } |
|
|
|
|
| def _group(records, key): |
| g = defaultdict(list) |
| for r in records: |
| g[r[key]].append(r) |
| return g |
|
|
|
|
| def gold_on_naive_db(items, db_paths: dict, ref_db) -> list[dict]: |
| """Altın SQL'i naif sayı kipleriyle kurulmuş veritabanlarında çalıştırır. |
| Sorgu doğru olsa bile yalnızca sayı biçimi yüzünden kaç cevabın bozulduğunu ölçer.""" |
| gold = gold_results(items, ref_db) |
| out = [] |
| for mode, p in db_paths.items(): |
| con = connect(p, readonly=True) |
| ok = err = 0 |
| broken = [] |
| n = 0 |
| for it in items: |
| if not it["altin_sql"]: |
| continue |
| n += 1 |
| try: |
| rows = [tuple(r) for r in con.execute(it["altin_sql"]).fetchall()] |
| if results_match(rows, gold[it["id"]]): |
| ok += 1 |
| else: |
| broken.append(it["id"]) |
| except Exception: |
| err += 1 |
| broken.append(it["id"]) |
| con.close() |
| out.append({"kip": mode, "n": n, "EX": round(100.0 * ok / n, 1), "yurutme_hatasi": err, |
| "bozulan": broken}) |
| return out |
|
|
