Text-to-SQL / t2s /agent.py
SevdanurGenc's picture
Upload 466 files
a72d1cf verified
Raw History Blame Contribute Delete
17.6 kB
"""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" # cevap | cevapsiz | bulunamadi | 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()
# ------------------------------------------------------------ yardımcılar
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: # büyük şemalarda bağlamı daralt: önce AI, olmazsa kural tabanlı tablo puanlaması
picked = None
if self.ai is not None:
try:
# kaynak belge adı da verilir: çok belgeli veritabanında aynı başlıklı tablolar kurumla ayrışır
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
# ------------------------------------------------------------ ana döngü
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
# ---- üretim
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
# ---- statik denetim
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
# ---- yürütme
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
# ---- sonuç denetimi
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