| """Uygulama genelindeki yapay zekâ yardımcıları. |
| |
| Bir LLM arka ucu seçildiğinde (Claude, OpenAI uyumlu, Ollama) şu adımlar AI ile zenginleşir: |
| enrich_table : çıkarılan tabloya anlamlı ad, özet, sütun açıklaması/birim önerisi ve ayrıştırma uyarıları |
| suggest_questions : veritabanı şemasına göre örnek soru önerileri |
| verbalize : SQL sonucunu tek-iki cümlelik doğal dil cevaba dönüştürme |
| explain_error : hata kaydını kullanıcıya açıklama ve onarım önerme |
| summarize_failures : değerlendirmedeki başarısız örnekleri temalara ayırma |
| select_tables : çok tablolu veritabanlarında soruya ilgili tabloları seçme (şema daraltma) |
| (SQL üretimi ve kendini düzeltme Text2SQLAgent içinde yapılır.) |
| |
| Tasarım ilkeleri: AI yalnızca öneri üretir; deterministik çıkarım/ayrıştırma her zaman birincil kaynaktır. |
| Tüm çağrılar başarısız olursa None döner ve uygulama AI'sız çalışmaya devam eder. Yanıtlar diskte önbelleğe |
| alınır (aynı girdi için tekrar ücret ödenmez). |
| """ |
| from __future__ import annotations |
|
|
| import hashlib |
| import json |
| import re |
| from pathlib import Path |
|
|
| from .llm import LLMBackend, parse_json_reply |
| from .trnum import slugify |
|
|
| ROOT = Path(__file__).resolve().parent.parent |
| CACHE = ROOT / "outputs" / ".ai_onbellek.json" |
|
|
| LANG_NAME = {"tr": "Türkçe", "en": "English"} |
|
|
|
|
| class AIHelper: |
| def __init__(self, backend: LLMBackend | None, lang: str = "tr", cache: bool = True): |
| self.backend = backend if isinstance(backend, LLMBackend) else None |
| self.lang = lang |
| self.use_cache = cache |
| self.last_error: str | None = None |
| self._mem = self._load() |
|
|
| |
| @property |
| def active(self) -> bool: |
| return self.backend is not None |
|
|
| @property |
| def label(self) -> str: |
| return self.backend.name if self.backend else "" |
|
|
| def _load(self) -> dict: |
| try: |
| return json.loads(CACHE.read_text(encoding="utf-8")) if self.use_cache and CACHE.exists() else {} |
| except Exception: |
| return {} |
|
|
| def _save(self): |
| if not self.use_cache: |
| return |
| try: |
| CACHE.parent.mkdir(parents=True, exist_ok=True) |
| items = list(self._mem.items())[-500:] |
| CACHE.write_text(json.dumps(dict(items), ensure_ascii=False), encoding="utf-8") |
| except Exception: |
| pass |
|
|
| def _call(self, system: str, user: str, max_tokens: int = 900, as_json: bool = False): |
| if not self.active: |
| return None |
| key = hashlib.sha256(f"{self.backend.name}|{system}|{user}".encode()).hexdigest()[:24] |
| if key in self._mem: |
| raw = self._mem[key] |
| else: |
| try: |
| raw = self.backend.chat(system, [{"role": "user", "content": user}], max_tokens=max_tokens) |
| self.last_error = None |
| except Exception as e: |
| self.last_error = str(e)[:300] |
| return None |
| self._mem[key] = raw |
| self._save() |
| if not as_json: |
| return (raw or "").strip() |
| return _parse_any_json(raw) |
|
|
| def _lang_rule(self) -> str: |
| return f"Yanıt dili: {LANG_NAME.get(self.lang, 'Türkçe')}." |
|
|
| |
| def enrich_table(self, caption: str, columns: list[dict], sample_rows: list[list], context: str = "") -> dict | None: |
| """columns: [{"ad","orijinal","tip","anlam","birim","ayristirilamayan"}]""" |
| system = ("[GÖREV:TABLO_ZENGINLESTIR]\nSen Türkçe kamu ve finans belgelerindeki tabloları veritabanına aktaran bir " |
| "veri mühendisisin. Sana belgeden otomatik çıkarılmış bir tablonun başlığı, sütunları ve örnek satırları " |
| "verilecek. Görevin: (1) tabloya kısa, anlamlı, snake_case ve yalnızca ASCII karakterli bir ad önermek " |
| "(en çok 4 sözcük, Türkçe kökler, ör. 'il_elektrik_tuketimi'); (2) tabloyu bir-iki cümleyle özetlemek; " |
| "(3) her sütun için kısa bir açıklama ve (biliniyorsa) birim yazmak; (4) ayrıştırmada şüpheli gördüğün " |
| "durumları (ör. sayısal olması gerekirken metin kalan sütun, tutarsız birim, toplam satırı) uyarı olarak " |
| "listelemek. Sütun adlarını DEĞİŞTİRME; yalnızca verilen adları anahtar olarak kullan. Uydurma bilgi " |
| f"ekleme. {self._lang_rule()}\n" |
| 'Yalnızca şu JSON\'u döndür: {"tablo_adi": "...", "ozet": "...", "sutunlar": {"<sütun_adı>": ' |
| '{"aciklama": "...", "birim": "..."}}, "uyarilar": ["..."]}') |
| user = json.dumps({"belgedeki_baslik": caption, "baglam": context[-400:], "sutunlar": columns, |
| "ornek_satirlar": sample_rows[:5]}, ensure_ascii=False, default=str) |
| d = self._call(system, user, max_tokens=1200, as_json=True) |
| if not isinstance(d, dict): |
| return None |
| names = {c["ad"] for c in columns} |
| cols = {k: {"aciklama": str((v or {}).get("aciklama", ""))[:240], "birim": str((v or {}).get("birim", "") or "")[:30]} |
| for k, v in (d.get("sutunlar") or {}).items() if k in names and isinstance(v, dict)} |
| name = slugify(str(d.get("tablo_adi", "")), 200) |
| if len(name) > 40: |
| name = "_".join(name[:41].split("_")[:-1]) or name[:40] |
| return {"tablo_adi": name if name and name != "sutun" else None, |
| "ozet": str(d.get("ozet", ""))[:600], "sutunlar": cols, |
| "uyarilar": [str(x)[:240] for x in (d.get("uyarilar") or [])][:6]} |
|
|
| def suggest_questions(self, schema: str, n: int = 6, seed: int = 0) -> list[str]: |
| system = ("[GÖREV:SORU_ONER]\nSana bir SQLite veritabanının şeması (tablo/sütun açıklamaları ve örnek değerler) " |
| f"verilecek. Bu tablolardan kesin cevabı bulunabilen, çeşitli türlerde (seçim, toplam/ortalama, en " |
| f"yüksek, yüzde değişim, filtre, tarih) {n} kısa soru yaz. Soruda tablolarda gerçekten geçen değerleri " |
| f"kullan. {self._lang_rule()} Yalnızca JSON dizi döndür: [\"soru 1\", ...]") |
| out = self._call(system, f"(öneri turu {seed})\n{schema[:9000]}", max_tokens=700, as_json=True) |
| if isinstance(out, dict): |
| out = out.get("sorular") or out.get("questions") or [] |
| return [str(q).strip() for q in (out or []) if str(q).strip()][:n] |
|
|
| def verbalize(self, question: str, sql: str, columns: list, rows: list, unit: str | None, fallback: str) -> str | None: |
| num_rule = ("Sayıları Türkçe biçimde yaz (1.234,56)." if self.lang == "tr" |
| else "Write numbers in English format (1,234.56).") |
| system = (f"[GÖREV:CEVAP]\nKullanıcının sorusunu yalnızca verilen SQL sonucuna dayanarak bir-iki cümleyle yanıtla. " |
| f"{num_rule} Birim verilmişse kullan. Sonuçta olmayan hiçbir bilgiyi ekleme. {self._lang_rule()}") |
| user = (f"Soru: {question}\nSQL: {sql}\nSütunlar: {columns}\nSonuç (ilk 20 satır): {rows[:20]}\n" |
| f"Birim: {unit or '-'}\nBiçimlenmiş kısa cevap: {fallback}") |
| return self._call(system, user, max_tokens=250) |
|
|
| def explain_error(self, record: dict) -> str | None: |
| system = ("[GÖREV:HATA_ACIKLA]\nBir Text-to-SQL sisteminin hata kaydını teknik olmayan bir kullanıcıya açıkla. " |
| "Üç kısa başlık kullan: Ne oldu? Neden oldu? Nasıl düzeltilir? Hata Türkçe sayı/tarih biçimi, birim, " |
| "İ/ı büyük-küçük harf veya belgedeki 'Toplam' satırıyla ilgiliyse bunu özellikle belirt. " |
| f"Kısa tut (en çok 120 sözcük). Markdown kullan. {self._lang_rule()}") |
| keep = {k: record.get(k) for k in ("tip", "asama", "soru", "mesaj", "sql", "oneriler")} |
| return self._call(system, json.dumps(keep, ensure_ascii=False, default=str), max_tokens=500) |
|
|
| def summarize_failures(self, records: list[dict]) -> str | None: |
| bad = [{k: r.get(k) for k in ("id", "soru", "hata_tipi", "tahmin_sql", "altin_sql", "cevap")} |
| for r in records if not r.get("dogru")][:40] |
| if not bad: |
| return None |
| system = ("[GÖREV:DEGERLENDIRME_OZETI]\nBir Text-to-SQL değerlendirmesindeki başarısız örnekleri analiz et. " |
| "Hataları 3-5 temaya grupla; her tema için kaç örnek olduğunu, tipik nedenini ve somut bir iyileştirme " |
| f"önerisini yaz. Örnek ID'lerini parantez içinde ver. Markdown kullan. {self._lang_rule()}") |
| return self._call(system, json.dumps(bad, ensure_ascii=False, default=str), max_tokens=900) |
|
|
| def describe_table(self, caption: str, schema_snippet: str) -> str | None: |
| system = ("[GÖREV:TABLO_OZETI]\nVeritabanındaki bir tabloyu iki cümleyle özetle: ne ölçüyor, hangi dönemleri ve " |
| f"birimleri kapsıyor, nasıl sorular sorulabilir. {self._lang_rule()}") |
| return self._call(system, f"Başlık: {caption}\n{schema_snippet[:4000]}", max_tokens=250) |
|
|
| def select_tables(self, question: str, tables: dict[str, str], k: int = 5) -> list[str] | None: |
| """tables: {ad: başlık/özet}. Soruya en ilgili en fazla k tabloyu döndürür.""" |
| system = ("[GÖREV:TABLO_SEC]\nSoru ve veritabanındaki tabloların listesi verilecek. Soruyu cevaplamak için gereken " |
| f"tabloları seç (en çok {k}). Yalnızca listede bulunan adları kullan. Yalnızca JSON dizi döndür: [\"ad\", ...]") |
| user = f"Soru: {question}\nTablolar:\n" + "\n".join(f"- {n}: {d}" for n, d in tables.items()) |
| out = self._call(system, user, max_tokens=200, as_json=True) |
| if isinstance(out, dict): |
| out = out.get("tablolar") or out.get("tables") or [] |
| picked = [x for x in (out or []) if isinstance(x, str) and x in tables][:k] |
| return picked or None |
|
|
|
|
| def _parse_any_json(raw: str): |
| if raw is None: |
| return None |
| txt = raw.strip() |
| m = re.search(r"```(?:json)?\s*(.+?)```", txt, re.S) |
| if m: |
| txt = m.group(1).strip() |
| for cand in (txt, _slice(txt, "[", "]"), _slice(txt, "{", "}")): |
| if not cand: |
| continue |
| try: |
| return json.loads(cand) |
| except json.JSONDecodeError: |
| continue |
| d = parse_json_reply(raw) |
| return d if d.get("sql") else None |
|
|
|
|
| def _slice(s: str, a: str, b: str) -> str | None: |
| i, j = s.find(a), s.rfind(b) |
| return s[i:j + 1] if 0 <= i < j else None |
|
|
