Text-to-SQL

"Just ask your database in English." Text-to-SQL makes it real — and it's a masterclass in LLM + guardrails engineering.

▶ Watch this reel

What you'll learn

How it works
Schema linking & semantic layer
SQL safety
Validation & self-correction

Remember this

Text-to-SQL = pipeline: schema linking prunes the warehouse, generation writes SQL, validation gates execution — the model is one stage of four
Semantic layer + column annotations + 3–5 gold example pairs beat any amount of instruction
Safety = read-only role + AST allow-list + LIMIT/timeout wrapping + capped self-correction loop with actionable errors

Pipeline

1. Question (ambiguous NL) 2. Schema linking — prune to relevant tables/columns 3. Generation — schema + annotations + examples 4. Validate → execute read-only → verify result

Schema linking & semantic layer

Annotate: meaning, units, caveats per column.
Semantic layer: versioned business definitions (churn, MRR, region).
3–5 gold question→SQL pairs per domain (demonstration > instruction).

SQL safety

Read-only DB role (deep fence) · AST allow-list (sqlglot) · LIMIT/timeout wrap · replica/masked data.

Validation & self-correction

DB errors back to model as actionable observations; retry ≤3.
Result checks: non-empty, shape, plausibility.
Persistent failure → ask a clarifying question (abstention).

Code: Text-to-SQL, the full guarded pipeline

SEMANTIC_LAYER = """
churned: no transaction in 90 days
mrr: recurring revenue from subscriptions only
region: customer's billing country
"""

async def ask_db(question: str) -> str:
    # 1 · schema linking: embed question → match table/column descriptions
    schema = link_schema(question, catalog)          # ~5 relevant tables

    for attempt in range(3):
        # 2 · generate with meaning + examples
        sql = llm(GEN_PROMPT, schema=schema,
                  semantics=SEMANTIC_LAYER,
                  examples=gold_pairs_for(question))
        try:
            rows = safe_execute(sql, readonly_conn)   # fenced (slide 4)
        except (SqlRejected, DbError) as e:
            history.append((sql, str(e)))             # actionable error → retry
            continue
        if not rows or not plausible(question, rows):
            history.append((sql, "result empty/implausible"))
            continue
        return render_answer(question, rows, sql)      # answer + SQL shown

    return "I couldn't answer that reliably — can you clarify the timeframe or metric?"