| """Cevap düzeyinde soru-cevap değerlendirmesi (altın SQL gerektirmez). |
| |
| Pilot, stres ve ağır LLM kümeleri bu modülle çalışır; arayüzdeki "Model karşılaştırma" bölümü ve |
| pilot/pilot_calistir.py aynı kodu kullanır. |
| |
| Değerlendirme kuralları |
| sayi : sonucun ilk satırındaki sayılardan biri |a−b| ≤ 0,01 + 10⁻³|b| ile eşleşirse doğru |
| (isaret_serbest: mutlak değerler de denenir) |
| metin : ilk satırdaki bir değer Türkçe katlama sonrası beklenene eşitse doğru; tarih/dönem ISO'ya çevrilir. |
| 'kabul' listesi varsa (ör. kurum kısa adı) değerin içinde geçmesi yeterlidir |
| liste : ilk sütundaki değer kümesi beklenen kümeye eşitse doğru |
| cevapsiz : sistem cevap üretmezse doğru |
| """ |
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import re |
| import time |
| from dataclasses import dataclass |
| from pathlib import Path |
|
|
| from .agent import Text2SQLAgent, format_answer |
| from .ai import AIHelper |
| from .llm import USAGE |
| from .pipeline import build_database |
| from .schema import connect |
| from .trnum import fold, parse_date |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| TOL_A, TOL_R = 0.01, 1e-3 |
|
|
|
|
| |
| @dataclass |
| class QASet: |
| key: str |
| belgeler: Path |
| sorular: Path |
| ham: Path | None |
| belge_basina: bool |
| filtre: str | None = None |
|
|
| def questions(self) -> list[dict]: |
| spec = json.loads(self.sorular.read_text(encoding="utf-8")) |
| qs = spec["sorular"] if isinstance(spec, dict) else spec |
| if self.filtre: |
| qs = [q for q in qs if q.get(self.filtre)] |
| return qs |
|
|
| def available(self) -> bool: |
| return self.belgeler.exists() and self.sorular.exists() |
|
|
| def docs(self) -> list[Path]: |
| return sorted(p for p in self.belgeler.iterdir() if p.suffix.lower() in (".pdf", ".html", ".htm")) |
|
|
|
|
| SETS = { |
| "agir": QASet("agir", ROOT / "agir_kume" / "belgeler", ROOT / "agir_kume" / "sorular.json", |
| ROOT / "agir_kume" / "ham_degerler.json", False), |
| "agir_hizli": QASet("agir_hizli", ROOT / "agir_kume" / "belgeler", ROOT / "agir_kume" / "sorular.json", |
| ROOT / "agir_kume" / "ham_degerler.json", False, "hizli"), |
| "stres": QASet("stres", ROOT / "stres_kumesi" / "belgeler", ROOT / "stres_kumesi" / "sorular.json", |
| ROOT / "stres_kumesi" / "ham_degerler.json", True), |
| "pilot": QASet("pilot", ROOT / "pilot" / "belgeler", ROOT / "pilot" / "sorular.json", |
| ROOT / "pilot" / "ham_degerler.json", False), |
| } |
|
|
|
|
| def code_hash() -> str: |
| return hashlib.sha256(b"".join(p.read_bytes() for p in sorted((ROOT / "t2s").glob("*.py")))).hexdigest()[:16] |
|
|
|
|
| def extraction_hash() -> str: |
| """Çıkarımı etkileyen modüllerin özeti: değişmedikçe kurulu veritabanı yeniden kullanılır.""" |
| names = ("extract.py", "pipeline.py", "schema.py", "trnum.py") |
| return hashlib.sha256(b"".join((ROOT / "t2s" / n).read_bytes() for n in names)).hexdigest()[:16] |
|
|
|
|
| |
| def _num(v): |
| if isinstance(v, bool): |
| return None |
| if isinstance(v, (int, float)): |
| return float(v) |
| return None |
|
|
|
|
| def _txt(v): |
| if isinstance(v, float) and v.is_integer(): |
| return str(int(v)) |
| return str(v) |
|
|
|
|
| def _key(v) -> str: |
| """Metin karşılaştırma anahtarı: Türkçe katlama; tarih/dönem ise ISO ('Ekim 2025' == '2025-10' == '2025/10').""" |
| s = str(v).strip() |
| m = re.fullmatch(r"(\d{4})[/.](\d{1,2})", s) |
| if m: |
| return f"{m.group(1)}-{int(m.group(2)):02d}" |
| return parse_date(s) or fold(s) |
|
|
|
|
| def judge(q: dict, res) -> tuple[bool, str]: |
| tip = q["tip"] |
| if tip == "cevapsiz": |
| return res.abstained, "red" if res.abstained else "cevap üretti" |
| if res.abstained or not res.rows: |
| return False, f"red/boş ({res.status})" |
| first = res.rows[0] |
| if tip == "sayi": |
| g = float(q["cevap"]) |
| vals = [x for x in (_num(v) for v in first) if x is not None] |
| if q.get("isaret_serbest"): |
| vals += [abs(x) for x in vals] |
| ok = any(abs(x - g) <= TOL_A + TOL_R * abs(g) for x in vals) |
| return ok, f"bulunan {vals[:3]}" |
| if tip == "metin" and q.get("kabul"): |
| keys = [fold(str(k)) for k in q["kabul"]] + [fold(str(q["cevap"]))] |
| ok = any(v is not None and any(k in fold(_txt(v)) for k in keys) for v in first) |
| return ok, f"bulunan {[v for v in first if v is not None][:3]}" |
| if tip == "metin": |
| g = _key(q["cevap"]) |
| ok = any(v is not None and _key(_txt(v)) == g for v in first) |
| return ok, f"bulunan {[v for v in first if v is not None][:3]}" |
| if tip == "liste": |
| got = {_key(_txt(r[0])) for r in res.rows if r and r[0] is not None} |
| exp = {_key(x) for x in q["cevap"]} |
| return got == exp, f"{len(got & exp)}/{len(exp)} ortak, {len(got - exp)} fazla" |
| return False, "bilinmeyen tip" |
|
|
|
|
| def value_recall(con, doc: str, raw_values: list[float]) -> dict: |
| """Belgenin ham sayısal değerlerinden kaçı, o belgeden çıkarılan tablolarda bir sayı olarak bulunuyor?""" |
| tabs = [t for (t,) in con.execute("SELECT tablo FROM _tablolar WHERE kaynak=?", (doc,))] |
| found = set() |
| for t in tabs: |
| for (col, typ) in con.execute(f'SELECT name, type FROM pragma_table_info("{t}")'): |
| if typ.upper() not in ("REAL", "INTEGER"): |
| continue |
| for (v,) in con.execute(f'SELECT "{col}" FROM "{t}" WHERE "{col}" IS NOT NULL'): |
| found.add(round(float(v), 4)) |
| need = [round(float(x), 4) for x in raw_values] |
| hit = sum(1 for x in need if x in found or -x in found) |
| return {"tablo": len(tabs), "ham_deger": len(need), "bulunan": hit, |
| "oran": round(100 * hit / len(need), 1) if need else None, "tablolar": tabs} |
|
|
|
|
| |
| def prepare_dbs(qs: QASet, out_dir: Path, docs: list[Path] | None = None, reuse: bool = True) -> list[tuple[Path, list[Path]]]: |
| """[(veritabanı, belgeler)] döndürür. Çıkarım kodu ve belgeler değişmedikçe kurulu veritabanı yeniden kullanılır.""" |
| docs = docs or qs.docs() |
| out_dir.mkdir(parents=True, exist_ok=True) |
| groups = [(out_dir / f"{d.stem}.db", [d]) for d in docs] if qs.belge_basina else [(out_dir / f"{out_dir.name}.db", docs)] |
| sig = extraction_hash() + "|" + hashlib.sha256(b"".join(d.read_bytes()[:4096] + str(d.stat().st_size).encode() |
| for d in docs)).hexdigest()[:12] |
| stamp = out_dir / ".imza" |
| fresh = reuse and stamp.exists() and stamp.read_text() == sig and all(p.exists() for p, _ in groups) |
| if not fresh: |
| for db, gdocs in groups: |
| db.unlink(missing_ok=True) |
| build_database(db, docs=gdocs, mode="tr") |
| stamp.write_text(sig) |
| return groups |
|
|
|
|
| |
| def run_set(qs: QASet, backend, out_dir: Path, *, questions: list[dict] | None = None, max_attempts: int = 3, |
| self_correct: bool = True, use_ai_tables: bool = True, progress=None, limit: int | None = None, |
| reuse_db: bool = True) -> dict: |
| """Bir test kümesini bir arka uçla çalıştırır; {'ozet':…, 'kayitlar':…} döndürür ve out_dir'e yazar.""" |
| t0 = time.time() |
| u0 = (USAGE.calls, USAGE.input_tokens, USAGE.output_tokens) |
| questions = questions if questions is not None else qs.questions() |
| if limit: |
| questions = questions[:limit] |
| dbkey = qs.key.replace("_hizli", "") |
| if qs.key == "ozel": |
| dbkey = "ozel_" + hashlib.sha256(str(qs.belgeler.resolve()).encode()).hexdigest()[:8] |
| groups = prepare_dbs(qs, ROOT / "outputs" / "qa" / "db" / dbkey, reuse=reuse_db) |
| ham = json.loads(qs.ham.read_text(encoding="utf-8")) if qs.ham and qs.ham.exists() else {} |
| llm = backend != "kural" |
| ai = AIHelper(backend, cache=False) if (llm and use_ai_tables) else None |
| recs, cikarim, done, n = [], {}, 0, len(questions) |
| for db, gdocs in groups: |
| names = {d.name for d in gdocs} |
| mine = [q for q in questions if (not qs.belge_basina) or q["belge"] in names] |
| con = connect(db, readonly=True) |
| for d in gdocs: |
| info = {"yazilan_tablo": [t for (t,) in con.execute("SELECT tablo FROM _tablolar WHERE kaynak=?", (d.name,))]} |
| if d.name in ham: |
| info["deger_geri_cagirma"] = value_recall(con, d.name, ham[d.name]) |
| cikarim[d.name] = info |
| if not mine: |
| con.close() |
| continue |
| agent = Text2SQLAgent(con, backend, max_attempts=max_attempts, self_correct=self_correct, |
| fallback_rule=False, log_source="qa_eval", ai=ai) |
| for q in mine: |
| try: |
| res = agent.run(q["soru"]) |
| ok, why = judge(q, res) |
| rec = {**q, "dogru": ok, "aciklama": why, "durum": res.status, "sql": res.sql, |
| "sonuc": [list(r) for r in res.rows[:5]], "cevap_metni": format_answer(res, con), |
| "deneme": res.attempts, "sure": round(res.duration, 2)} |
| except Exception as e: |
| rec = {**q, "dogru": False, "aciklama": f"istisna: {str(e)[:200]}", "durum": "istisna", "sql": None, |
| "sonuc": [], "cevap_metni": "", "deneme": 0, "sure": 0} |
| recs.append(rec) |
| done += 1 |
| if progress: |
| progress(done, n, rec) |
| con.close() |
| summary = summarize(recs) |
| summary.update({"arka_uc": "kural" if not llm else backend.name, "kume": qs.key, "cikarim": cikarim, |
| "sure_sn": round(time.time() - t0, 1), "kod_ozeti": code_hash(), |
| "llm_cagri": USAGE.calls - u0[0], "girdi_belirteci": USAGE.input_tokens - u0[1], |
| "cikti_belirteci": USAGE.output_tokens - u0[2], |
| "zaman": time.strftime("%Y-%m-%dT%H:%M:%S")}) |
| out_dir.mkdir(parents=True, exist_ok=True) |
| (out_dir / "sonuclar.json").write_text(json.dumps({"ozet": summary, "kayitlar": recs}, ensure_ascii=False, |
| indent=1, default=str), encoding="utf-8") |
| return {"ozet": summary, "kayitlar": recs} |
|
|
|
|
| def _rate(rs): |
| return round(100 * sum(r["dogru"] for r in rs) / len(rs), 1) if rs else None |
|
|
|
|
| def summarize(recs: list[dict]) -> dict: |
| def by(field): |
| ks = sorted({str(r.get(field)) for r in recs if r.get(field) is not None}) |
| return {k: {"n": sum(1 for r in recs if str(r.get(field)) == k), |
| "basari": _rate([r for r in recs if str(r.get(field)) == k])} for k in ks} |
|
|
| ans = [r for r in recs if r["tip"] != "cevapsiz"] |
| una = [r for r in recs if r["tip"] == "cevapsiz"] |
| return { |
| "soru": len(recs), "dogru": sum(r["dogru"] for r in recs), "basari": _rate(recs), |
| "cevaplanabilir_basari": _rate(ans), "cevapsiz_dogru_red": _rate(una), |
| "yanlis_red": round(100 * sum(r["durum"] != "cevap" for r in ans) / len(ans), 1) if ans else None, |
| "sessiz_yanlis": round(100 * sum((not r["dogru"]) and r["durum"] == "cevap" for r in recs) / len(recs), 1) |
| if recs else None, |
| "seviyeye_gore": by("seviye"), "dile_gore": by("dil"), "aileye_gore": by("aile"), "tipe_gore": by("tip"), |
| "cizime_gore": by("cizim"), "belge_turune_gore": by("belge_turu"), "zorluga_gore": by("zorluk"), |
| } |
|
|
