| """LLM deneyi: makaledeki hakem eleştirilerini kapatmak için tek komutluk, tekrarlanabilir protokol. |
| |
| Çalıştırılan koşullar (aynı TR-TabloSQL soruları, aynı veritabanları, sıcaklık 0, hibrit geri düşme kapalı): |
| <etiket>_tr Türkçe normalizasyon + statik denetim + geri beslemeli döngü (K = 3) -> ana sonuç |
| <etiket>_tr_K1 aynı, döngü kapalı (tek deneme) -> döngünün katkısı |
| <etiket>_tr_statiksiz aynı, statik denetim kapalı (yalnızca yürütme/sonuç geri beslemesi) -> statik denetimin katkısı |
| <etiket>_metin sayılar metin olarak saklanmış veritabanı -> LLM metni kurtarabilir mi? |
| <etiket>_en İngilizce yerel ayar varsayımlı veritabanı -> geri döndürülemez kayıp |
| stres testi pilot/belgeler (görülmemiş düzenler), cevap düzeyinde değerlendirme |
| |
| Kullanım (model adını sabit sürüm olarak verin; 'auto' yalnızca --auto-izin ile): |
| python scripts/llm_deneyi.py --saglayici claude --model claude-sonnet-4-5-20250929 --kuru # maliyet tahmini |
| python scripts/llm_deneyi.py --saglayici claude --model claude-sonnet-4-5-20250929 |
| python scripts/llm_deneyi.py --saglayici openai --model gpt-4o-2024-08-06 |
| python scripts/llm_deneyi.py --saglayici ollama --model qwen2.5-coder:7b # yerel, ücretsiz |
| |
| Anahtarlar .env dosyasından ya da ortam değişkenlerinden okunur (ANTHROPIC_API_KEY, OPENAI_API_KEY). |
| Çıktılar: outputs/llm/<etiket>/ozet.json, rapor.md, sonuclar_*.jsonl, stres/pilot_sonuclari.json |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import re |
| import subprocess |
| import sys |
| import time |
| import warnings |
| from math import comb |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| sys.path.insert(0, str(ROOT)) |
| warnings.filterwarnings("ignore") |
|
|
| from t2s.errors import ErrorLog |
| from t2s.evaluate import SystemSpec, evaluate_system, load_benchmark |
| from t2s.llm import USAGE, load_dotenv, make_backend |
| from t2s.pipeline import build_database, build_reference_db |
|
|
| DB = ROOT / "outputs" / "db" |
|
|
|
|
| def wilson(k, n, z=1.96): |
| if not n: |
| return None |
| p = k / n |
| d = 1 + z * z / n |
| c = (p + z * z / (2 * n)) / d |
| h = z * ((p * (1 - p) / n + z * z / (4 * n * n)) ** 0.5) / d |
| return [round(100 * (c - h), 1), round(100 * (c + h), 1)] |
|
|
|
|
| def mcnemar(b, c): |
| n = b + c |
| if n == 0: |
| return 1.0 |
| k = min(b, c) |
| return min(1.0, 2 * sum(comb(n, i) for i in range(k + 1)) / 2 ** n) |
|
|
|
|
| def load_recs(path: Path) -> dict: |
| return {r["id"]: r for r in (json.loads(l) for l in path.read_text(encoding="utf-8").splitlines() if l.strip())} |
|
|
|
|
| def compare(a: dict, b: dict): |
| ids = sorted(set(a) & set(b)) |
| bb = sum(1 for i in ids if a[i]["dogru"] and not b[i]["dogru"]) |
| cc = sum(1 for i in ids if b[i]["dogru"] and not a[i]["dogru"]) |
| return bb, cc, mcnemar(bb, cc) |
|
|
|
|
| def dry_run(items, n_systems): |
| from t2s.llm import SYSTEM_PROMPT, build_user_prompt |
| from t2s.schema import connect, describe_schema |
| con = connect(DB / "sistem_tr.db", readonly=True) |
| schema = describe_schema(con) |
| chars = sum(len(SYSTEM_PROMPT) + len(build_user_prompt(it["soru"], schema, it.get("kanit"))) for it in items) |
| con.close() |
| tok_in = chars / 3.2 * 1.35 * n_systems |
| tok_out = len(items) * 220 * 1.35 * n_systems |
| return int(tok_in), int(tok_out) |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--saglayici", required=True, choices=["claude", "openai", "ollama"]) |
| ap.add_argument("--model", required=True, help="sabit model sürümü (ör. claude-sonnet-4-5-20250929)") |
| ap.add_argument("--anahtar", default=None) |
| ap.add_argument("--taban-url", default=None) |
| ap.add_argument("--etiket", default=None, help="çıktı klasörü adı (varsayılan: sağlayıcı_model)") |
| ap.add_argument("--yalniz", choices=["hepsi", "ana", "stres"], default="hepsi") |
| ap.add_argument("--kuru", action="store_true", help="çağrı yapmadan token/maliyet tahmini") |
| ap.add_argument("--auto-izin", action="store_true", help="model adı 'auto' olabilir (önerilmez)") |
| args = ap.parse_args() |
| load_dotenv(ROOT / ".env") |
|
|
| if args.model == "auto" and not args.auto_izin: |
| raise SystemExit("Tekrarlanabilirlik için sabit bir model sürümü verin (ör. claude-sonnet-4-5-20250929).") |
| etiket = args.etiket or re.sub(r"[^A-Za-z0-9._-]+", "_", f"{args.saglayici}_{args.model}") |
| out = ROOT / "outputs" / "llm" / etiket |
|
|
| items = load_benchmark() |
| for m in ("tr", "en", "text"): |
| if not (DB / f"sistem_{m}.db").exists(): |
| build_database(DB / f"sistem_{m}.db", mode=m) |
| if not (DB / "referans.db").exists(): |
| build_reference_db(DB / "referans.db") |
| db_paths = {m: DB / f"sistem_{m}.db" for m in ("tr", "en", "text")} |
|
|
| p = args.saglayici |
| specs_def = [(f"{p}_tr", "tr", True, True), (f"{p}_tr_K1", "tr", False, True), |
| (f"{p}_tr_statiksiz", "tr", True, False), (f"{p}_metin", "text", True, True), |
| (f"{p}_en", "en", True, True)] |
|
|
| if args.kuru: |
| ti, to = dry_run(items, len(specs_def)) |
| pilot_q = len(json.loads((ROOT / "pilot" / "sorular.json").read_text(encoding="utf-8"))["sorular"]) |
| print(f"Tahmini çağrı sayısı : ~{int(len(items) * 1.35 * len(specs_def)) + int(pilot_q * 1.35)}") |
| print(f"Tahmini giriş tokenı : ~{ti:,} (+ stres testi ~{int(ti / len(specs_def) * pilot_q / len(items)):,})") |
| print(f"Tahmini çıkış tokenı : ~{to:,}") |
| print("Ek cevap düzeyinde kümeler: stres kümesi (307 soru) ve ağır küme hızlı alt kümesi (200 soru) için") |
| print(" yaklaşık 2,5 milyon giriş tokenı daha (tablo seçimi ve düzeltme denemeleriyle değişir).") |
| print("Maliyet = giriş × (sağlayıcının giriş fiyatı) + çıkış × (çıkış fiyatı); Ollama ücretsizdir.") |
| return |
|
|
| out.mkdir(parents=True, exist_ok=True) |
| |
| from t2s.settings import Settings |
| cfg = Settings() |
| key = args.anahtar or cfg.key(p) or None |
| base = args.taban_url or cfg.base(p) |
| backend = make_backend(p, args.model, key, base, |
| device=cfg.conn("ollama").get("cihaz") if p == "ollama" else None) |
| print("Bağlantı denemesi…", end=" ", flush=True) |
| backend.chat("Kısa yanıt ver.", [{"role": "user", "content": "ping"}], max_tokens=5) |
| print("tamam") |
|
|
| meta = {"saglayici": p, "model": args.model, "sicaklik": 0.0, "K": 3, "hibrit_geri_dusme": False, |
| "baslangic": time.strftime("%Y-%m-%dT%H:%M:%S"), "kosullar": {}} |
| if args.yalniz in ("hepsi", "ana"): |
| elog = ErrorLog(out / "hata_gunlugu.jsonl") |
| for name, mode, sc, st in specs_def: |
| t0 = time.time() |
| spec = SystemSpec(name, backend, mode, self_correct=sc, static_check=st) |
| s = evaluate_system(spec, items, db_paths, DB / "referans.db", out, elog, |
| progress=lambda k, n, r: print(f"\r {name}: {k}/{n}", end="", flush=True)) |
| n = s["n"] |
| k = round(s["EX"] * n / 100) |
| s["EX_ga95"] = wilson(k, n) |
| s["sure_dk"] = round((time.time() - t0) / 60, 1) |
| meta["kosullar"][name] = s |
| print(f"\r {name:24} EX=%{s['EX']} test=%{s['bolum'].get('test')} ilk=%{s['EX_ilk_deneme']} " |
| f"red=%{s['cevapsiz_dogru_red']} yanlış red=%{s['yanlis_red_orani']} ({s['sure_dk']} dk)") |
|
|
| rule = ROOT / "outputs" / "sonuclar" / "sonuclar_kural_tr.jsonl" |
| R = {name: load_recs(out / f"sonuclar_{name}.jsonl") for name, *_ in specs_def} |
| pairs = [(f"{p}_tr", f"{p}_tr_K1", "geri beslemeli döngünün katkısı"), |
| (f"{p}_tr", f"{p}_tr_statiksiz", "statik denetimin katkısı"), |
| (f"{p}_tr", f"{p}_metin", "Türkçe normalizasyonun katkısı (LLM, metin kipine karşı)"), |
| (f"{p}_tr", f"{p}_en", "Türkçe normalizasyonun katkısı (LLM, en kipine karşı)")] |
| if rule.exists(): |
| R["kural_tr"] = load_recs(rule) |
| pairs.append((f"{p}_tr", "kural_tr", "LLM ve kural tabanlı üretici")) |
| meta["mcnemar"] = [{"a": a, "b": b, "anlam": t, **dict(zip(("a_dogru_b_yanlis", "b_dogru_a_yanlis", "p"), |
| compare(R[a], R[b])))} for a, b, t in pairs] |
|
|
| if args.yalniz in ("hepsi", "stres"): |
| flag = {"claude": "--claude", "openai": "--openai", "ollama": "--ollama"}[p] |
| extra = (["--anahtar", args.anahtar] if args.anahtar else []) + \ |
| (["--taban-url", args.taban_url] if args.taban_url else []) |
| for kume, ad in (("pilot", "stres"), ("stres", "stres_kumesi"), ("agir_hizli", "agir_hizli")): |
| cmd = [sys.executable, str(ROOT / "pilot" / "pilot_calistir.py"), "--kume", kume, flag, args.model, |
| "--cikti", str(out / ad)] + extra |
| print(f"Cevap düzeyinde küme: {kume}…") |
| subprocess.run(cmd, check=True) |
| meta[ad] = json.loads((out / ad / "sonuclar.json").read_text(encoding="utf-8"))["ozet"] |
| meta[ad].pop("cikarim", None) |
|
|
| meta["kullanim"] = {"cagri": USAGE.calls, "hata": USAGE.errors, "giris_token": USAGE.input_tokens, |
| "cikis_token": USAGE.output_tokens} |
| meta["bitis"] = time.strftime("%Y-%m-%dT%H:%M:%S") |
| (out / "ozet.json").write_text(json.dumps(meta, ensure_ascii=False, indent=1, default=str), encoding="utf-8") |
|
|
| L = [f"# LLM deneyi: {p} / {args.model}", "", f"Sıcaklık 0 · K = 3 · hibrit geri düşme kapalı · {meta['baslangic']}", ""] |
| if meta["kosullar"]: |
| L += ["| Koşul | EX (%95 GA) | Ayrılmış test | İlk deneme | Cevapsızda red | Yanlış red | Ort. deneme |", |
| "|---|---|---|---|---|---|---|"] |
| for n_, s in meta["kosullar"].items(): |
| L.append(f"| {n_} | {s['EX']} ({s['EX_ga95'][0]}–{s['EX_ga95'][1]}) | {s['bolum'].get('test')} | " |
| f"{s['EX_ilk_deneme']} | {s['cevapsiz_dogru_red']} | {s['yanlis_red_orani']} | {s['ort_deneme']} |") |
| L += ["", "| Karşılaştırma | A doğru B yanlış | B doğru A yanlış | McNemar p |", "|---|---|---|---|"] |
| for m in meta["mcnemar"]: |
| L.append(f"| {m['a']} – {m['b']} ({m['anlam']}) | {m['a_dogru_b_yanlis']} | {m['b_dogru_a_yanlis']} | " |
| f"{m['p']:.3g} |") |
| for ad, baslik in (("stres", "Pilot (3 belge)"), ("stres_kumesi", "Stres kümesi (33 belge)"), |
| ("agir_hizli", "Ağır küme, hızlı alt küme (tek veritabanı)")): |
| if ad in meta: |
| st = meta[ad] |
| lv = st.get("seviyeye_gore") or {} |
| lvs = " · ".join(f"{k} %{v['basari']}" for k, v in sorted(lv.items())) |
| L += ["", f"**{baslik}:** genel %{st['basari']}, cevaplanabilir %{st['cevaplanabilir_basari']}, " |
| f"cevapsızda red %{st['cevapsiz_dogru_red']}, yanlış red %{st['yanlis_red']}, " |
| f"sessiz yanlış %{st.get('sessiz_yanlis')}" + (f" ({lvs})" if lvs else "")] |
| u = meta["kullanim"] |
| L += ["", f"Kullanım: {u['cagri']} çağrı, {u['giris_token']:,} giriş / {u['cikis_token']:,} çıkış tokenı, " |
| f"{u['hata']} hata."] |
| (out / "rapor.md").write_text("\n".join(L) + "\n", encoding="utf-8") |
| print("\n".join(L)) |
| print(f"\nÇıktılar: {out}") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
