Spaces:
Running on Zero
Running on Zero
| """ | |
| Baseline keyword parser for Text-to-SQL. | |
| Rule-based model (no ML) that maps natural language questions to SQL using | |
| keyword matching and fuzzy string matching. Exists solely to provide a | |
| comparison point for the fine-tuned T5 model. | |
| Expected accuracy: ~10-20% exact match — by design. | |
| """ | |
| import json | |
| import os | |
| import re | |
| from difflib import get_close_matches | |
| # ── Aggregation keyword map ────────────────────────────────────────────────── | |
| # Multi-word phrases are checked first (sorted by length, descending). | |
| AGG_KEYWORDS = { | |
| "average": "AVG", | |
| "mean": "AVG", | |
| "how many": "COUNT", | |
| "count": "COUNT", | |
| "number of": "COUNT", | |
| "maximum": "MAX", | |
| "highest": "MAX", | |
| "most": "MAX", | |
| "largest": "MAX", | |
| "greatest": "MAX", | |
| "biggest": "MAX", | |
| "minimum": "MIN", | |
| "lowest": "MIN", | |
| "least": "MIN", | |
| "smallest": "MIN", | |
| "fewest": "MIN", | |
| "total": "SUM", | |
| "sum": "SUM", | |
| } | |
| # Pre-sort so multi-word phrases are checked before single words | |
| _AGG_PHRASES_SORTED = sorted(AGG_KEYWORDS, key=len, reverse=True) | |
| # ── Internal helpers ───────────────────────────────────────────────────────── | |
| def _detect_aggregation(question: str) -> str | None: | |
| """Return SQL aggregation keyword (AVG, COUNT, …) or None for plain SELECT.""" | |
| q_lower = question.lower() | |
| for phrase in _AGG_PHRASES_SORTED: | |
| if phrase in q_lower: | |
| return AGG_KEYWORDS[phrase] | |
| return None | |
| def _fuzzy_match_column(text: str, columns: list[str], | |
| cutoff: float = 0.4) -> str | None: | |
| """Find the best-matching column name for a piece of text. | |
| Strategy: | |
| 1. Exact substring — if a full column name appears inside the text, | |
| pick the longest match (avoids "No." matching before "No. of goals"). | |
| 2. Fuzzy word-level — for each column, check if any of its words | |
| fuzzy-match any word in the text. | |
| """ | |
| text_lower = text.lower() | |
| # --- exact substring pass --- | |
| best_exact = None | |
| best_exact_len = 0 | |
| for col in columns: | |
| cl = col.lower() | |
| if cl in text_lower and len(cl) > best_exact_len: | |
| best_exact = col | |
| best_exact_len = len(cl) | |
| if best_exact: | |
| return best_exact | |
| # --- fuzzy word-level pass --- | |
| text_words = re.findall(r"[a-z0-9]+", text_lower) | |
| best_score = 0.0 | |
| best_col = None | |
| for col in columns: | |
| col_words = re.findall(r"[a-z0-9]+", col.lower()) | |
| for cw in col_words: | |
| if len(cw) <= 2: # skip tiny words like "no", "id" | |
| continue | |
| matches = get_close_matches(cw, text_words, n=1, cutoff=cutoff) | |
| if matches: | |
| # use ratio as a rough score | |
| from difflib import SequenceMatcher | |
| score = SequenceMatcher(None, cw, matches[0]).ratio() | |
| if score > best_score: | |
| best_score = score | |
| best_col = col | |
| return best_col | |
| def _extract_where_condition(question: str, columns: list[str], | |
| sel_column: str) -> tuple[str, str] | None: | |
| """Try to extract a (column, value) pair for a WHERE clause. | |
| Very crude — scans for column names other than the SELECT column | |
| appearing in the question, then grabs the surrounding text as the value. | |
| Falls back to fuzzy matching if no exact column substring is found. | |
| """ | |
| q_lower = question.lower() | |
| remaining = [c for c in columns if c != sel_column] | |
| # Pass 1: exact column-name substring match in the question | |
| for col in remaining: | |
| cl = col.lower() | |
| pos = q_lower.find(cl) | |
| if pos == -1: | |
| continue | |
| # Grab everything *after* the column name as candidate value | |
| after = question[pos + len(col):].strip() | |
| after = re.sub(r"^[\s,=:]+", "", after).rstrip("?.!,").strip() | |
| # Grab everything *before* the column name as alternate candidate | |
| before = question[:pos].strip() | |
| before = re.sub(r"[\s,=:]+$", "", before).rstrip("?.!,").strip() | |
| # Prefer "after" if non-empty, else try "before" tail | |
| if after: | |
| return (col, after) | |
| if before: | |
| words = before.split() | |
| value = " ".join(words[-4:]) if len(words) > 4 else before | |
| return (col, value) | |
| # Pass 2: pattern-based extraction | |
| # Look for common question structures that imply a WHERE condition: | |
| # "What is the X where/when/for/of/with Y" | |
| # "... that were/was/is/are Y" | |
| signal_patterns = [ | |
| r"(?:where|when|if|that)\s+(?:the\s+)?(.+?)(?:\?|$)", | |
| r"\bfor\s+(?:the\s+|a\s+)?(.+?)(?:\?|$)", | |
| r"\bwith\s+(?:the\s+|a\s+)?(.+?)(?:\?|$)", | |
| r"\bof\s+(?:the\s+|a\s+)?(.+?)(?:\?|$)", | |
| r"\bhad\s+(.+?)(?:\?|$)", | |
| ] | |
| for pattern in signal_patterns: | |
| m = re.search(pattern, q_lower) | |
| if not m: | |
| continue | |
| fragment = m.group(1).strip().rstrip("?.!,") | |
| if not fragment or len(fragment) < 2: | |
| continue | |
| # Try to fuzzy-match a column name within this fragment | |
| cond_col = _fuzzy_match_column(fragment, remaining, cutoff=0.5) | |
| if cond_col: | |
| # Value = everything in the fragment that isn't the column name | |
| cl = cond_col.lower() | |
| frag_lower = fragment.lower() | |
| col_pos = frag_lower.find(cl) | |
| if col_pos != -1: | |
| value = fragment[col_pos + len(cl):].strip() | |
| value = re.sub(r"^[\s,=:]+", "", value).rstrip("?.!,").strip() | |
| if value: | |
| return (cond_col, value) | |
| # Column matched but couldn't isolate value — skip | |
| continue | |
| # No column match in fragment — if there's a "=" or "is" pattern, | |
| # try to split on it | |
| is_split = re.split(r"\s+(?:is|are|was|were|=)\s+", fragment, maxsplit=1) | |
| if len(is_split) == 2: | |
| left, right = is_split | |
| left_col = _fuzzy_match_column(left, remaining, cutoff=0.5) | |
| if left_col and right.strip(): | |
| return (left_col, right.strip().rstrip("?.!,")) | |
| return None | |
| # ── Public API ─────────────────────────────────────────────────────────────── | |
| def baseline_predict(question: str, columns: list[str], | |
| types: list[str]) -> str: | |
| """Generate SQL from a question using keyword + fuzzy matching. | |
| Parameters | |
| ---------- | |
| question : str | |
| Natural-language question. | |
| columns : list[str] | |
| Column names from the table header. | |
| types : list[str] | |
| Column types (``"real"`` or ``"text"``). Currently unused by the | |
| baseline but kept in the signature so it mirrors the T5 prompt | |
| builder interface. | |
| Returns | |
| ------- | |
| str | |
| SQL string in WikiSQL ``human_readable`` format. | |
| """ | |
| # 1 — aggregation | |
| agg = _detect_aggregation(question) | |
| # 2 — SELECT column | |
| sel_col = _fuzzy_match_column(question, columns) or columns[0] | |
| # 3 — WHERE condition (best-effort) | |
| where = _extract_where_condition(question, columns, sel_col) | |
| # 4 — assemble | |
| if agg: | |
| sql = f"SELECT {agg} {sel_col} FROM table" | |
| else: | |
| sql = f"SELECT {sel_col} FROM table" | |
| if where: | |
| cond_col, cond_val = where | |
| sql += f" WHERE {cond_col} = {cond_val}" | |
| return sql | |
| # ── Manual sanity test ─────────────────────────────────────────────────────── | |
| if __name__ == "__main__": | |
| DATA_DIR = os.path.join(os.path.dirname(__file__), "..", "data", "wikisql_export") | |
| TEST_FILE = os.path.join(DATA_DIR, "test.json") | |
| # Load first 20 test samples | |
| samples = [] | |
| with open(TEST_FILE, "r", encoding="utf-8") as f: | |
| for i, line in enumerate(f): | |
| if i >= 20: | |
| break | |
| samples.append(json.loads(line.strip())) | |
| exact_matches = 0 | |
| total = len(samples) | |
| print("=" * 90) | |
| print("BASELINE KEYWORD PARSER — Manual Sanity Check (first 20 test samples)") | |
| print("=" * 90) | |
| for idx, sample in enumerate(samples): | |
| question = sample["question"] | |
| columns = sample["table"]["header"] | |
| types = sample["table"]["types"] | |
| expected = sample["sql"]["human_readable"] | |
| predicted = baseline_predict(question, columns, types) | |
| match = predicted.strip() == expected.strip() | |
| if match: | |
| exact_matches += 1 | |
| icon = "✅" if match else "❌" | |
| print(f"\n--- Sample {idx + 1} ---") | |
| print(f" Question: {question}") | |
| print(f" Expected: {expected}") | |
| print(f" Got: {predicted}") | |
| print(f" Match: {icon}") | |
| print("\n" + "=" * 90) | |
| print(f"Exact matches: {exact_matches}/{total} ({exact_matches/total:.0%})") | |
| print("=" * 90) | |
| print("\nNote: Low accuracy is EXPECTED. This is a dumb baseline.") | |
| print("The real model (T5) will be evaluated against these same metrics in Phase 4.") | |