"""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() # ------------------------------------------------------------ altyapı @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: # noqa: BLE001 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: # noqa: BLE001 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: # noqa: BLE001 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')}." # ------------------------------------------------------------ özellikler 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": {"": ' '{"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: # sözcük ortasından kesme 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