| """Kendini düzelten çok adımlı Text-to-SQL ajanı. |
| |
| Akış (her adım AgentResult.steps içinde izlenir): |
| 1. Şema bağlama : ilgili tablolar ve sütun açıklamaları (orijinal başlık, birim, örnek değer) |
| 2. SQL üretimi : LLM veya kural tabanlı üretici |
| 3. Statik denetim : sözdizimi, salt-okuma, TR sayı/tarih literali, toplam satırı tuzağı |
| 4. Yürütme : salt-okunur bağlantı + zaman aşımı |
| 5. Sonuç denetimi : boş sonuç -> değer eşleme önerileri |
| 6. Geri besleme : hata tipi + öneriler modele geri verilir, en fazla N deneme |
| 7. Cevap : Türkçe sayı biçimi + birim; cevap yoksa bunu AÇIKÇA söyler |
| """ |
| from __future__ import annotations |
|
|
| import re |
| import sqlite3 |
| import time |
| from dataclasses import dataclass, field |
|
|
| from .errors import ErrorEvent, ErrorLog, check_result, classify_exception, static_checks |
| from .llm import SYSTEM_PROMPT, LLMBackend, build_user_prompt, parse_json_reply |
| from .rulebased import RuleBasedTextToSQL |
| from .i18n import abstain_prefix, answer_texts, format_num, translate_reason |
| from .schema import describe_schema, list_tables |
|
|
| ABSTAIN_PREFIX = abstain_prefix("tr") |
|
|
|
|
| @dataclass |
| class AgentResult: |
| question: str |
| backend: str |
| status: str = "hata" |
| sql: str | None = None |
| columns: list[str] = field(default_factory=list) |
| rows: list[tuple] = field(default_factory=list) |
| answer: str = "" |
| steps: list[dict] = field(default_factory=list) |
| events: list[ErrorEvent] = field(default_factory=list) |
| attempts: int = 0 |
| first_sql: str | None = None |
| first_rows: list[tuple] | None = None |
| first_error: str | None = None |
| exec_error: str | None = None |
| duration: float = 0.0 |
| short_answer: str = "" |
| fallback: str | None = None |
|
|
| @property |
| def abstained(self) -> bool: |
| return self.status != "cevap" |
|
|
|
|
| def _exec(con, sql: str, timeout: float = 5.0, max_rows: int = 1000): |
| deadline = time.time() + timeout |
| con.set_progress_handler(lambda: 1 if time.time() > deadline else 0, 10000) |
| try: |
| cur = con.execute(sql) |
| cols = [d[0] for d in cur.description] if cur.description else [] |
| rows = cur.fetchmany(max_rows) |
| return cols, [tuple(r) for r in rows] |
| finally: |
| con.set_progress_handler(None, 0) |
|
|
|
|
| def result_unit(sql: str, con) -> str | None: |
| """Tek sütunlu sonuçların birimini meta veriden kestirir.""" |
| try: |
| import sqlglot |
| from sqlglot import exp |
| tree = sqlglot.parse_one(sql, read="sqlite") |
| sel = tree.find(exp.Select) |
| if sel is None or len(sel.expressions) != 1: |
| return None |
| e = sel.expressions[0] |
| e = e.this if isinstance(e, exp.Alias) else e |
| if isinstance(e, exp.Count): |
| return None |
| text = e.sql().lower() |
| cols = [c.name for c in e.find_all(exp.Column)] |
| if re.search(r"\*\s*100(\.0)?\b(?!0)", text): |
| return "%" |
| tables = [t.name for t in tree.find_all(exp.Table)] |
| units = set() |
| for c in cols: |
| for t in tables: |
| r = con.execute("SELECT birim FROM _sutunlar WHERE tablo=? AND sutun=?", (t, c)).fetchone() |
| if r and r[0]: |
| units.add(r[0]) |
| if len(units) != 1: |
| return None |
| u = units.pop() |
| if re.search(r"\*\s*1000\b", text) and u.startswith("bin "): |
| return u[4:] |
| if re.search(r"\*\s*1000\b", text) and u == "m³": |
| return "litre" |
| if re.search(r"/\s*1000000", text) and u == "TL": |
| return "milyon TL" |
| if re.search(r"[*/]\s*\d", text) or ("-" in text and len(cols) > 1 and u == "%"): |
| return None if u != "%" else "puan" |
| return u |
| except Exception: |
| return None |
|
|
|
|
| def format_answer(res: AgentResult, con, lang: str = "tr") -> str: |
| rows, cols = res.rows, res.columns |
| T = answer_texts(lang) |
| if not rows: |
| return T["no_rows"] |
| if len(rows) == 1 and len(cols) == 1: |
| v = rows[0][0] |
| unit = result_unit(res.sql or "", con) |
| if isinstance(v, float): |
| txt = format_num(round(v, 2), lang, 2 if abs(v) < 1e6 or not v.is_integer() else 0) |
| else: |
| txt = format_num(v, lang) |
| if lang == "en" and unit: |
| unit = {"bin TL": "thousand TL", "milyon TL": "million TL", "milyar TL": "billion TL", |
| "bin MWh": "thousand MWh", "kişi": "persons", "puan": "points", "litre": "litres"}.get(unit, unit) |
| if unit == "%": |
| return f"{T['answer']}: %{txt}" if lang == "tr" else f"{T['answer']}: {txt}%" |
| return f"{T['answer']}: {txt}" + (f" {unit}" if unit else "") |
| if len(cols) == 1: |
| vals = [format_num(r[0], lang) if not isinstance(r[0], str) else r[0] for r in rows] |
| head = ", ".join(vals[:15]) + (" …" if len(vals) > 15 else "") |
| return f"{T['n_results'].format(n=len(rows))}: {head}" |
| return T["table_below"].format(n=len(rows)) |
|
|
|
|
| class Text2SQLAgent: |
| def __init__(self, con: sqlite3.Connection, backend: str | LLMBackend = "kural", max_attempts: int = 3, |
| self_correct: bool = True, static_check: bool = True, error_log: ErrorLog | None = None, |
| log_source: str = "arayuz", verbalize: bool = False, lang: str = "tr", ai=None, |
| fallback_rule: bool = True): |
| self.con = con |
| self.lang = lang |
| self.ai = ai if (ai is not None and getattr(ai, "active", False)) else None |
| self.fallback_rule = fallback_rule |
| self.backend = backend |
| self.max_attempts = max_attempts if self_correct else 1 |
| self.self_correct = self_correct |
| self.static_check = static_check |
| self.error_log = error_log |
| self.log_source = log_source |
| self.verbalize = verbalize |
| self._rule = RuleBasedTextToSQL(con) if backend == "kural" else None |
| self.backend_name = "kural-tabanli" if backend == "kural" else backend.name |
|
|
| def refresh(self): |
| if self._rule: |
| self._rule.refresh() |
|
|
| |
| def _log(self, res: AgentResult, ev: ErrorEvent): |
| ev.soru = res.question |
| res.events.append(ev) |
| if self.error_log: |
| self.error_log.add(ev, kaynak=self.log_source, arka_uc=self.backend_name) |
|
|
| def _schema_for(self, question: str) -> tuple[str, list[str]]: |
| tabs = list_tables(self.con) |
| if len(tabs) > 8: |
| picked = None |
| if self.ai is not None: |
| try: |
| |
| caps = dict(self.con.execute("SELECT tablo, COALESCE(baslik,'') || ' ' || COALESCE(ozet,'') || " |
| "' [belge: ' || COALESCE(kaynak,'') || ']' FROM _tablolar").fetchall()) |
| except sqlite3.Error: |
| caps = {t: t for t in tabs} |
| picked = self.ai.select_tables(question, {t: caps.get(t, t) for t in tabs}) |
| if not picked: |
| rb = self._rule or RuleBasedTextToSQL(self.con) |
| r = rb.generate(question) |
| picked = [n for _, n in r.trace.get("tablo_puanlari", [])][:5] |
| tabs = picked or tabs[:8] |
| return describe_schema(self.con, tabs), tabs |
|
|
| |
| def run(self, question: str, evidence: str | None = None) -> AgentResult: |
| t0 = time.time() |
| res = AgentResult(question, self.backend_name) |
| schema, tabs = self._schema_for(question) |
| res.steps.append({"tur": "sema", "baslik": "Şema bağlama", "tablolar": tabs, |
| "icerik": f"{len(tabs)} tablo bağlama alındı: {', '.join(tabs)}", "sema": schema}) |
|
|
| candidates, rule_reason = None, "" |
| messages: list[dict] = [] |
| if self._rule is not None: |
| rr = self._rule.generate(question) |
| candidates = rr.candidates |
| rule_reason = rr.reason |
| res.steps.append({"tur": "analiz", "baslik": "Soru çözümleme (kural tabanlı)", |
| "icerik": f"Tablo puanları: {rr.trace.get('tablo_puanlari', [])[:3]} | " |
| f"ölçü: {rr.trace.get('olcu')} | sayılar: {rr.trace.get('sayilar')} | " |
| f"tarihler: {rr.trace.get('tarihler')}"}) |
| else: |
| messages = [{"role": "user", "content": build_user_prompt(question, schema, evidence)}] |
|
|
| last_event = None |
| for attempt in range(1, self.max_attempts + 1): |
| res.attempts = attempt |
| |
| if self._rule is not None: |
| if not candidates or attempt > len(candidates): |
| if attempt == 1: |
| reason_l = translate_reason(rule_reason, self.lang) |
| res.status = "cevapsiz" |
| res.answer = f"{abstain_prefix(self.lang)}. {reason_l}" |
| res.steps.append({"tur": "cevapsiz", "baslik": "Cevaplanamaz", "icerik": reason_l}) |
| elif last_event is not None and last_event.tip in ("BOS_SONUC", "DEGER_ESLEME"): |
| res.status = "bulunamadi" |
| break |
| sql, answerable, reason = candidates[attempt - 1], True, "" |
| else: |
| try: |
| raw = self.backend.chat(SYSTEM_PROMPT, messages) |
| except Exception as e: |
| res.steps.append({"tur": "hata", "baslik": "Model hatası", "asama": "model", "icerik": str(e)[:400]}) |
| if self.fallback_rule: |
| return self._fallback(res, question, t0, str(e)) |
| res.status = "hata" |
| res.answer = f"{abstain_prefix(self.lang)}: " + answer_texts(self.lang)["llm_down"].format(e=e) |
| break |
| messages.append({"role": "assistant", "content": raw}) |
| d = parse_json_reply(raw) |
| sql, answerable, reason = (d.get("sql") or "").strip(), bool(d.get("cevaplanabilir")), d.get("gerekce", "") |
| res.steps.append({"tur": "uretim", "baslik": f"Deneme {attempt}: model çıktısı", "k": attempt, |
| "icerik": d.get("dusunce", ""), "sql": sql}) |
| if not answerable or not sql: |
| res.status = "cevapsiz" |
| res.answer = (f"{abstain_prefix(self.lang)}. " |
| f"{reason or translate_reason('Model soruyu tablolarla cevaplanamaz buldu.', self.lang)}") |
| res.steps.append({"tur": "cevapsiz", "baslik": "Model: cevaplanamaz", "icerik": reason}) |
| break |
| if self._rule is not None: |
| res.steps.append({"tur": "uretim", "baslik": f"Deneme {attempt}: aday SQL", "k": attempt, |
| "icerik": "", "sql": sql}) |
| if attempt == 1: |
| res.first_sql = sql |
|
|
| |
| evs = static_checks(sql, self.con) if self.static_check else [] |
| blocking = [e for e in evs if e.tip == "SOZDIZIMI"] |
| warnings = [e for e in evs if e.tip != "SOZDIZIMI"] |
| for e in evs: |
| self._log(res, e) |
| res.steps.append({"tur": "uyari" if e in warnings else "hata", "baslik": f"Statik denetim: {e.tip}", |
| "asama": "statik", "icerik": e.feedback(), "hata_tipi": e.tip}) |
| can_retry = self.self_correct and attempt < self.max_attempts |
| if blocking or (warnings and can_retry and self._rule is None): |
| last_event = (blocking or warnings)[0] |
| if attempt == 1: |
| res.first_error = last_event.tip |
| if can_retry: |
| if self._rule is None: |
| messages.append({"role": "user", "content": "Sorgun denetimden geçmedi:\n" + "\n".join( |
| e.feedback() for e in (blocking or warnings)) + "\nDüzeltilmiş JSON'u döndür."}) |
| res.steps.append({"tur": "geribildirim", "baslik": "Geri besleme gönderildi", |
| "icerik": "Hata tipi ve öneriler bir sonraki denemeye aktarıldı."}) |
| continue |
| if blocking: |
| res.exec_error = blocking[0].mesaj |
| break |
|
|
| |
| try: |
| cols, rows = _exec(self.con, sql) |
| except Exception as exc: |
| ev = classify_exception(exc, sql, self.con) |
| self._log(res, ev) |
| res.exec_error = str(exc) |
| if attempt == 1: |
| res.first_error = ev.tip |
| last_event = ev |
| res.steps.append({"tur": "hata", "baslik": f"Yürütme hatası: {ev.tip}", "icerik": ev.feedback(), |
| "asama": "yurutme", "hata_tipi": ev.tip}) |
| if can_retry and self._rule is None: |
| messages.append({"role": "user", "content": "Sorgu çalıştırılırken hata oluştu:\n" |
| + ev.feedback() + "\nDüzeltilmiş JSON'u döndür."}) |
| continue |
| res.exec_error = None |
| if attempt == 1: |
| res.first_rows = rows |
| |
| rev = check_result(sql, rows, self.con) |
| res.steps.append({"tur": "sonuc", "baslik": f"Yürütme: {len(rows)} satır", "icerik": "", "n": len(rows), |
| "sql": sql, "satir": rows[:20], "sutun": cols}) |
| if rev: |
| ev = rev[0] |
| self._log(res, ev) |
| last_event = ev |
| if attempt == 1: |
| res.first_error = ev.tip |
| res.steps.append({"tur": "uyari", "baslik": f"Sonuç denetimi: {ev.tip}", "icerik": ev.feedback(), |
| "asama": "sonuc", "hata_tipi": ev.tip}) |
| res.sql, res.columns, res.rows = sql, cols, rows |
| if can_retry: |
| if self._rule is None: |
| messages.append({"role": "user", "content": "Sorgu boş/NULL sonuç döndürdü:\n" |
| + ev.feedback() + "\nGerekirse düzelt; cevap tablolarda yoksa " |
| "cevaplanabilir=false döndür. JSON ver."}) |
| continue |
| res.status = "bulunamadi" |
| break |
| res.sql, res.columns, res.rows = sql, cols, rows |
| res.status = "cevap" |
| break |
|
|
| T = answer_texts(self.lang) |
| if res.status == "cevap": |
| res.answer = format_answer(res, self.con, self.lang) |
| if self.verbalize: |
| v = self._verbalize(res) |
| if v: |
| res.short_answer = res.answer |
| res.answer = v |
| res.steps.append({"tur": "ai", "baslik": "AI ile cümleleştirildi", "icerik": v}) |
| elif res.status == "bulunamadi": |
| res.answer = f"{abstain_prefix(self.lang)}: {T['notfound']}" |
| res.rows = [] |
| elif res.status == "hata" and not res.answer: |
| tip = last_event.tip if last_event else "?" |
| res.answer = f"{abstain_prefix(self.lang)}: " + T["failed"].format(k=res.attempts, tip=tip) |
| res.steps.append({"tur": "cevap", "baslik": "Sonuç", "icerik": res.answer}) |
| res.duration = time.time() - t0 |
| return res |
|
|
| def _fallback(self, res: AgentResult, question: str, t0: float, err: str) -> AgentResult: |
| """LLM'e ulaşılamazsa aynı soruyu kural tabanlı üreticiyle yanıtla (hibrit çalışma).""" |
| alt = Text2SQLAgent(self.con, "kural", max_attempts=self.max_attempts, self_correct=self.self_correct, |
| error_log=self.error_log, log_source=self.log_source, lang=self.lang) |
| r2 = alt.run(question) |
| r2.backend = f"{self.backend_name} → kural-tabanli" |
| r2.fallback = err |
| r2.steps = res.steps + [{"tur": "geridusme", "baslik": "Kural tabanlıya geri düşüldü", "icerik": err[:300]}] \ |
| + r2.steps[1:] |
| r2.duration = time.time() - t0 |
| return r2 |
|
|
| def _verbalize(self, res: AgentResult) -> str | None: |
| if self.ai is not None: |
| return self.ai.verbalize(res.question, res.sql or "", res.columns, res.rows, |
| result_unit(res.sql or "", self.con), res.answer) |
| if self._rule is not None: |
| return None |
| try: |
| unit = result_unit(res.sql or "", self.con) |
| txt = self.backend.chat( |
| "Kısa ve kesin Türkçe cevap ver. Sayıları Türkçe biçimde yaz (1.234,56). Uydurma bilgi ekleme.", |
| [{"role": "user", "content": f"Soru: {res.question}\nSQL: {res.sql}\nSütunlar: {res.columns}\n" |
| f"Sonuç: {res.rows[:20]}\nBirim: {unit or '-'}\nTek-iki cümlelik cevap yaz."}], max_tokens=200) |
| return txt.strip() |
| except Exception: |
| return None |
|
|
