Data Analyst Agent (Text-to-SQL)
"Talk to your database." A non-technical user asks a question in plain English; the agent writes SQL against your schema, runs it read-only, and explains the result in words (and a chart). A top enterprise ask in 2026 — and a superb showcase of grounding, tool use, and a strict safety boundary.
What this project teaches you to design
- Grounding the model in a database schema so it writes correct SQL.
- The read-only boundary that makes running generated SQL safe.
- A generate → validate → run → explain loop that self-corrects on errors.
- Text-to-SQL evals: does the query return the right answer?
The brief advanced
"Let anyone answer their own data questions without waiting on the data team." Every "can you pull the numbers for…" is a ticket to a busy analyst. An agent that turns the question into correct SQL, runs it safely, and explains the answer democratizes data — while the analysts focus on the hard modeling work.
1 · Discovery — where does the time go? advanced
| Where time goes | Agent leverage |
|---|---|
| Analysts writing ad-hoc SQL for others' questions | ⭐⭐⭐ high — the core value |
| Non-technical users blocked waiting for numbers | ⭐⭐⭐ high — self-serve |
| Remembering table/column names & joins | ⭐⭐ medium — schema grounding |
| Deep statistical / causal analysis, data modeling | ⭐ low — humans; agent does the retrieval |
2 · Architecture advanced
This picture is the whole agent as a pipeline: an English question flows left to right, becomes SQL, is checked and run safely, and comes back as a plain-English answer. Follow the blue arrows.
- Question (English), far left, is what the user types. It feeds the Analyst agent box in the middle, whose job is to write SQL.
- Above the agent, schema + examples (RAG grounding) feeds in the table/column definitions and example queries — this is what stops the model from inventing column names.
- The agent's SQL goes to validate SQL (the read-only check) and then run (read-only) against the database — the two safety controls from Step 2.
- The error? / retry-fix box is the self-correcting loop: the pink dashed arrow curving back to 'validate SQL' shows an error being fed back so the agent can try again.
- Finally Explain (words + chart), far right, turns the result rows into a sentence (and optionally a chart) for the user.
In short: Left-to-right is the happy path (question → SQL → validate → run → explain); the one dashed arrow looping backward is the agent fixing its own mistakes.
A Chapter 4 loop: grounded in the DB schema (RAG), the agent writes SQL, a validator checks it's read-only and well-formed, it runs against a read-only connection, and — critically — if the query errors, the error goes back to the agent to self-correct. Then it explains the result.
3 · Risk & safety model advanced
Running model-generated SQL against a real database sounds scary — the read-only boundary is what makes it safe, and it's enforced in infrastructure, not the prompt.
| Risk | Control |
|---|---|
| 🔴 Destructive SQL (DROP/DELETE/UPDATE) | Connect with a read-only database user — writes are impossible at the DB level. Also validate the query is a single SELECT before running. Belt + suspenders. |
| 🔴 Reading data the user shouldn't see | The read-only user only has access to permitted tables/rows; enforce row-level security or per-user views (never trust the model) |
| 🟠 Runaway / expensive query | Statement timeout + auto LIMIT; run against a replica, not the primary |
| 🟠 Wrong-but-plausible answer | Show the generated SQL to the user for transparency; verify with evals; explain assumptions |
| 🟠 SQL injection via the question | The model writes parameterized/validated SQL; the user's text is never concatenated into a query blindly |
4 · Tool surface advanced
| Tool | Does | Risk |
|---|---|---|
get_schema | Return table/column definitions (+ descriptions) | 🟢 read-only |
sample_rows | A few example rows so the model sees real values/formats | 🟢 read-only |
validate_sql | Parse; confirm it's a single read-only SELECT; reject otherwise | 🟢 deterministic |
run_query | Execute against the read-only replica, with timeout + LIMIT | 🟡 read-only exec (bounded) |
make_chart | Render results as a chart (e.g. via code execution / matplotlib) | 🟢 sandboxed |
run_query returns a SQL error ("column x does not exist"), feed that error back as a tool result. The agent reads it, fixes the query, and retries — often getting it right on the second try. This is the Ch 4 agentic loop doing real work, and it dramatically improves accuracy.5 · Schema grounding — the accuracy lever advanced
Text-to-SQL lives or dies on the model knowing your schema. Ground it (RAG, Ch 3) with:
- Table & column definitions with human descriptions ("
status: one of active/churned/trial"). - Relationships — foreign keys and how tables join.
- A few sample rows per table — so it sees real value formats (dates, enums, units).
- Curated example question→SQL pairs — few-shot examples of your team's common queries and idioms.
6 · Evaluation expert
| Eval | Measures |
|---|---|
| Execution accuracy | Does the generated query return the same result as a hand-written gold query? (the key metric — compare result sets, not SQL text) |
| Valid-SQL rate | % of queries that run without error (after self-correction) |
| Read-only compliance (deterministic) | The agent never produces a non-SELECT that passes the validator — hard-fail |
| Explanation faithfulness | Does the plain-English answer match what the query actually returned? (LLM-judge) |
run_query modify data or escape read-only?" Must be zero — enforced by the read-only credential and the validator, and tested. Same shape as the DevOps agent's "never exceed your rung" eval.7 · Phased rollout expert
Skills & course map expert
| Skill | Learn it in |
|---|---|
| Schema grounding / relevant-table retrieval | Ch 3 |
| Generate→run→self-correct loop with tools | Ch 4 |
| Read-only boundary (creds + validator) | Ch 8 safety model + Ch 6 |
| Execution-accuracy evals | Ch 5 |
| Charts via sandboxed code execution | Ch 6 / tool concepts |
| Robust result handling & explanation | Python P2 (data), P1 (formatting) |
Build-along plan expert
- Set up a read-only SQLite/Postgres with a sample schema (e.g. orders/customers) — and a read-only user.
- Ground the model (Ch 3): a
get_schematool + a few example question→SQL pairs. - The loop (Ch 4): agent writes SQL →
validate_sql(must be a single SELECT) →run_query→ on error, feed it back to self-correct → explain. - Read-only hard boundary: connect as a read-only user; assert the validator rejects any non-SELECT (the hard-fail eval).
- Evals (Ch 5): a set of question + gold-SQL pairs; compare result sets for execution accuracy.
- Polish (Ch 6): add
make_chart, timeouts/LIMIT, and always surface the SQL.
Learning objectives
- Ground the model in a DB schema so it writes correct SQL.
- Enforce read-only two ways: a SELECT-only validator + a read-only connection.
- Build a loop that feeds SQL errors back so the model self-corrects.
- Measure execution accuracy against gold queries.
What you'll build expert
Ask "how many US customers?" and get the answer — the agent writes SQL, runs it read-only, self-corrects on errors, and explains the result. Running model-generated SQL is safe because writes are impossible at the connection level.
llm-course-starter/data-analyst/. Design rationale in the design chapter.Step 1 · A tiny real database expert
A local SQLite DB with customers and orders. get_schema() returns the CREATE statements to ground the model.
agent/db.pySCHEMA = """
CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, country TEXT, plan TEXT);
CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, amount REAL,
status TEXT, created TEXT);
"""
def init_db(): ... # create + seed sample rows
def get_schema(): ... # return the CREATE TABLE text for grounding
terminalpython agent/db.py # builds the db, no API key needed
Before the agent can write SQL it needs to know what tables and columns exist. This file builds a tiny local SQLite database and exposes its shape so we can hand that shape to the model. No API key is needed to run it.
SCHEMAis just a block of text holding twoCREATE TABLEstatements —customersandorders. It describes the database's structure in plain SQL.init_db()creates those tables and fills them with a few sample rows, so there is real data to query.get_schema()returns thatCREATE TABLEtext. This is the grounding step: we paste this into the model's prompt so it knows the exact table and column names and never has to guess them.
What the output means: Running it builds the database file on disk. Nothing is printed to you here — it just prepares the data the later steps will query.
Try this: Open the database and add a third table (say products) to SCHEMA. Because get_schema() returns whatever is in SCHEMA, the model would immediately be able to write queries against the new table too.
Step 2 · The read-only guard (the safety boundary) expert
Setup to run this snippet
_DB_PATH = 5 # demo constant (a hard cap)
class _re_t:
match = 'demo'
search = 'demo'
def match(self, *a, **k): return 'demo'
def search(self, *a, **k): return 'demo'
def __getattr__(self, k): return 'demo'
re = _re_t()
import sqlite3agent/db.pydef validate_sql(sql):
s = sql.strip().rstrip(";").strip()
if ";" in s: return False, "multiple statements not allowed"
if not re.match(r"(?is)^\s*select\b", s): return False, "SELECT only"
if re.search(r"(?i)\b(insert|update|delete|drop|alter|create|pragma)\b", s):
return False, "write/DDL not allowed"
return True, "ok"
def run_query(sql, limit=100):
ok, reason = validate_sql(sql)
if not ok: raise ValueError(f"blocked: {reason}")
con = sqlite3.connect(f"file:{_DB_PATH}?mode=ro", uri=True) # read-only!
...
This is the safety boundary — the single most important code in the project. It decides whether a piece of SQL is allowed to run at all. The rule: only a single, read-only SELECT gets through; everything else is refused before it touches the database.
validate_sql(sql)first cleans up the text (strips spaces and a trailing;). If a semicolon still remains inside, there are two statements glued together — a classic injection trick — so it returnsFalsewith a reason. Every check returns a(True/False, reason)pair.- It then requires the query to start with
select(re.matchchecks the beginning), and separately scans the whole string for dangerous words likeinsert,update,delete,drop. Either failure blocks the query. run_query(sql)calls the validator first and raises if it said no — so nothing unsafe ever runs. Only then does it connect.- The connection string ends in
?mode=rowithuri=True— that opens the database read-only. This is a second, independent guard: even if a bad query slipped past the validator, the engine itself refuses to write.
What the output means: For a normal SELECT you get (True, "ok") and the query runs. For a DROP or DELETE you get (False, ...) and it never executes.
Try this: Picture handing this a DROP TABLE customers. The word-scan catches drop and returns False, "write/DDL not allowed" — and even if it didn't, mode=ro would block it at the engine. Two independent locks: that is 'belt and suspenders'.
mode=ro), so even if the validator missed something, a DELETE raises at the SQLite engine level. This is the DevOps capstone's "guardrails in infrastructure, not the prompt" principle, applied to a DB. Verified: python agent/db.py shows DROP blocked.Step 3 · The self-correcting loop expert
Setup to run this snippet
MAX_TRIES = 5 # demo constant (a hard cap)
class _Any:
'''stands in for any undefined demo value; supports call/attr/index/
iteration and basic arithmetic (as 0.7) so demo snippets run.'''
def __call__(self, *a, **k): return _Any()
def __getattr__(self, k): return _Any()
def __getitem__(self, k): return _Any()
def __iter__(self): return iter([])
def __len__(self): return 0
def __contains__(self, o): return True
def __enter__(self, *a): return _Any()
def __exit__(self, *a): return False
def __float__(self): return 0.7
def __int__(self): return 1
def __lt__(self, o): return True
def __gt__(self, o): return False
def __le__(self, o): return True
def __ge__(self, o): return False
def __add__(self, o): return o
def __radd__(self, o): return o
def __bool__(self): return True
def __repr__(self): return 'demo'
def __str__(self): return 'demo'
def _explain(*a, **k): # demo stub
return _Any()
def _write_sql(*a, **k): # demo stub
return _Any()
class _db_t:
get_schema = 'demo'
run_query = 'demo'
validate_sql = 'demo'
def get_schema(self, *a, **k): return 'demo'
def run_query(self, *a, **k): return 'demo'
def validate_sql(self, *a, **k): return 'demo'
def __getattr__(self, k): return 'demo'
db = _db_t()agent/engine.pydef ask(question):
schema = db.get_schema()
error = None
for attempt in range(MAX_TRIES):
q = _write_sql(question, schema, error) # model writes SQL
ok, reason = db.validate_sql(q.sql)
if not ok: error = reason; continue # feed back: wrong kind of SQL
try:
cols, rows = db.run_query(q.sql)
except Exception as e:
error = str(e); continue # feed back: SQL error -> self-correct
return {"sql": q.sql, "cols": cols, "rows": rows,
"explanation": _explain(question, q.sql, cols, rows)}
return {"error": f"no valid query after {MAX_TRIES} tries"}
This is the heart of the agent — the loop that lets it fix its own mistakes. Instead of writing SQL once and giving up if it's wrong, the agent tries up to MAX_TRIES times, feeding each error back so the next attempt can correct it.
- It grabs the schema once (
db.get_schema()) and starts witherror = None. Thefor attempt in range(MAX_TRIES)loop gives it a limited number of tries so it can't loop forever. - Each pass,
_write_sql(question, schema, error)asks the model for SQL. Crucially it also passes the previous error — so the model knows what went wrong last time and can fix it. db.validate_sql(q.sql)runs the Step 2 safety check. If it fails, we save the reason intoerrorandcontinue— jump straight to the next attempt with that feedback.- If it passes,
db.run_query(q.sql)executes it. A database error (like a misspelled column) is caught byexcept, stored inerror, and fed back on the next loop. On success it returns the SQL, the rows, and a plain-English_explain(...). If all tries fail, it returns an error dict.
What the output means: Normally you get a dictionary with sql, cols, rows and explanation. The self-correcting part means a first-try mistake often becomes a correct answer by the second try.
Try this: Imagine attempt 1 produces SELECT revenue FROM orders but there's no revenue column. run_query raises, the error text "no such column: revenue" is fed back, and attempt 2 writes SELECT amount ... instead. That feedback loop is the whole trick.
Step 4 · Explain the result expert
terminalpython agent/engine.py # needs key
Q: How many customers are in the US?
SQL: SELECT count(*) FROM customers WHERE country='US'
-> There are 2 customers in the US.
Q: What's total paid revenue by country?
SQL: SELECT c.country, sum(o.amount) FROM orders o JOIN customers c
ON o.customer_id=c.id WHERE o.status='paid' GROUP BY c.country
-> US leads with $700 in paid revenue, followed by DE ($60)...This runs the finished agent end-to-end (this one needs an API key, because a real model writes the SQL). It shows what a user actually sees: their English question, the SQL the agent generated, and a plain-English answer.
- Each block starts with
Q:— the question typed in plain English, e.g. "How many customers are in the US?". - The
SQL:line is the query the agent wrote itself from the schema. Showing it is deliberate — the user (or an analyst) can verify the logic. - The
->line is the explanation: the model turns the raw result rows back into a sentence a non-technical person can read.
What the output means: Two answered questions. The second is more advanced — the agent wrote a JOIN across orders and customers, filtered to paid orders, grouped by country, and then summarised the numbers in words.
Try this: Notice the SQL is always shown, never hidden. That transparency is a trust feature: if the answer looks off, you can read the query and see exactly how it was computed.
5 · Tests (no key — the safety boundary) expert
terminalpython -m pytest tests/ -v
test_select_is_allowed PASSED
test_writes_are_blocked PASSED
test_multiple_statements_blocked PASSED
test_readonly_connection_rejects_writes PASSED
test_a_real_query_returns_rows PASSED
5 passedThese are the automated tests that prove the safety boundary actually works — and they run with no API key, because the safety logic is plain Python, not the model. Green PASSED lines mean each guarantee held.
test_select_is_allowed— a legitimateSELECTpasses the validator, so we haven't blocked real work.test_writes_are_blockedandtest_multiple_statements_blocked— the validator refuses every write (DROP/DELETE/UPDATE/INSERT) and refuses two-statements-in-one injection.test_readonly_connection_rejects_writes— the second lock: even if a write somehow reached the database, the read-only connection rejects it at the engine.test_a_real_query_returns_rows— confirms the database and query runner actually work end-to-end.
What the output means: 5 passed means all five safety promises are verified. This is the cheap, fast, key-free proof that running model-generated SQL here is safe.
Try this: These tests are your regression net: if someone later loosens the validator regex, one of these turns red immediately. Run them before trusting any change to the safety code.
| Test | Proves |
|---|---|
| SELECT allowed | legit queries pass the validator |
| writes blocked (DROP/DELETE/UPDATE/INSERT) | the validator refuses every write |
| multiple statements blocked | no SELECT 1; DROP ... injection |
| read-only connection rejects writes | even a bypassed validator can't write — engine blocks it |
| real query returns rows | the DB + runner work end-to-end |
6 · Evals (needs key) expert
terminalpython evals.py
[PASS] How many customers are in the US?
[PASS] What is total paid revenue?
execution accuracy: 2/2
✅ evals passedExecution accuracy compares the agent's result set to a hand-written gold query's — not the SQL text, since many queries yield the same answer. That's the right metric for text-to-SQL.
Safety tests prove the agent can't do harm; evals prove it gives the right answer. This measures execution accuracy — the correct metric for text-to-SQL.
- Each
[PASS]line is one question where the agent's answer matched a hand-written 'gold' query's answer. - It compares the result set (the rows returned), not the SQL text — because many different queries can produce the same correct answer. What matters is whether the numbers are right.
execution accuracy: 2/2is the score: 2 of 2 questions returned the right result.
What the output means: A per-question pass/fail list plus an overall accuracy fraction, then a final ✅ evals passed. This is how you'd track quality as you add more question types.
Try this: Add a new (question, gold-SQL) pair to the eval set and re-run. If accuracy drops, you've found a question type the agent isn't reliable on yet — exactly the signal used to decide which questions are safe to self-serve.
Troubleshooting expert
| Symptom | Fix |
|---|---|
| Agent invents column/table names | Ensure get_schema() is in the prompt; add sample_rows so it sees real values |
| Query errors it can't fix | Raise MAX_TRIES; make sure the error string is fed back in _write_sql |
| Wrong answer, valid SQL | Ambiguous question or wrong join; show the SQL, add example question→SQL pairs (Ch 3) |
| Validator blocks a legit query | A CTE/subquery with a keyword substring? Refine the regex, but keep it strict — false-block beats false-allow |
OperationalError: readonly database | That's correct behavior on a write — the read-only guard working |
🪜 Practice ladder beginner → industry
Six graded exercises, easy to real-world. Try each before opening its solution.
Context: Text-to-SQL accuracy starts with grounding. Give the model the exact CREATE TABLE text and it stops inventing columns like revenue that don't exist.
Your task: Build a get_schema() that returns the real CREATE TABLE text the prompt will carry, plus a helper that extracts a table's column names from it.
Requirements:
- A schema string holding the actual CREATE TABLE statements for the tables
get_schema()returns that text verbatim for the prompt- A helper parses a CREATE TABLE statement into its list of column names
- The columns returned match the real schema exactly
- Explain that grounding on the real schema is what stops invented columns
💡 Hint: Parse column names by slicing between the parentheses of the CREATE TABLE line and taking the first token of each comma-separated field.
Show solution
Schema grounding — the accuracy lever (pure stdlib, runnable):
SCHEMA = """
CREATE TABLE customers (id INTEGER PRIMARY KEY, name TEXT, country TEXT, plan TEXT);
CREATE TABLE orders (id INTEGER PRIMARY KEY, customer_id INTEGER, amount REAL,
status TEXT, created TEXT);
"""
def get_schema():
return SCHEMA.strip()
def columns_of(table):
for line in get_schema().splitlines():
if line.strip().lower().startswith(f"create table {table}"):
inside = line[line.index("(")+1 : line.rindex(")")]
return [c.strip().split()[0] for c in inside.split(",")]
return []
print(columns_of("customers")) # ['id', 'name', 'country', 'plan']Handing the model the real schema is what stops it inventing a revenue column that doesn't exist. Grounding in exact table and column names is the single biggest accuracy lever for text-to-SQL — the model writes against reality, not a guess.
Context: The first half of the safety boundary is a validator that runs before any SQL touches the database. It accepts a single SELECT and rejects everything else — writes, DDL, and stacked statements.
Your task: Implement validate_sql() that accepts exactly one SELECT and rejects multiple statements, writes, and DDL, then prove it blocks the dangerous cases.
Requirements:
- Reject input containing more than one statement (a semicolon separating statements)
- Require the query to start with SELECT, case-insensitively
- Reject any write/DDL keyword (insert, update, delete, drop, alter, create, pragma)
- Return a pass/fail plus a human-readable reason
- Demonstrate a valid SELECT passing and each dangerous case being blocked
💡 Hint: Regex with word boundaries and case-insensitive flags handles both the SELECT-prefix check and the forbidden-keyword scan; this is the "belt" of belt-and-suspenders.
Show solution
The SQL validator — deterministic, not the model's judgment (pure stdlib, runnable):
import re
def validate_sql(sql):
s = sql.strip().rstrip(";").strip()
if ";" in s:
return False, "multiple statements not allowed" # block stacked injection
if not re.match(r"(?is)^\s*select\b", s):
return False, "SELECT only"
if re.search(r"(?i)\b(insert|update|delete|drop|alter|create|pragma)\b", s):
return False, "write/DDL not allowed"
return True, "ok"
print(validate_sql("SELECT * FROM customers")) # (True, 'ok')
print(validate_sql("DROP TABLE customers")) # (False, ...)
print(validate_sql("SELECT 1; DROP TABLE customers")) # (False, multiple)The validator is a deterministic parser check — it accepts exactly one SELECT and rejects writes, DDL, and stacked statements. This is the "belt": the safety boundary lives in plain Python, never in a prompt the model could be tricked into ignoring.
Context: Defense-in-depth means one layer failing is not a breach. Even if the validator misses something, opening the database read-only makes writes impossible at the engine level.
Your task: Open a read-only SQLite connection and show it serving a SELECT while rejecting a write, demonstrating the second redundant guard beneath the validator.
Requirements:
- Create a throwaway database and seed it with a normal connection
- Reopen it read-only using the SQLite URI mode
- A SELECT succeeds on the read-only connection
- A write (e.g. DELETE) raises an operational error
- Explain why the connection is the "suspenders" that back up the validator "belt"
💡 Hint: SQLite's file:...?mode=ro URI (with URI parsing enabled) is the whole mechanism — the engine itself refuses the write.
Show solution
The read-only connection — the second, independent guard (pure stdlib, runnable):
import sqlite3, tempfile, os
# build a tiny throwaway DB
path = os.path.join(tempfile.gettempdir(), "analyst_demo.db")
w = sqlite3.connect(path); w.execute("CREATE TABLE IF NOT EXISTS t(x)")
w.execute("INSERT INTO t VALUES (1)"); w.commit(); w.close()
# open READ-ONLY: writes are forbidden by the engine, not by our code
con = sqlite3.connect(f"file:{path}?mode=ro", uri=True)
print("read works:", con.execute("SELECT count(*) FROM t").fetchone())
try:
con.execute("DELETE FROM t")
print("write got through -- BAD")
except sqlite3.OperationalError as e:
print("write blocked by mode=ro:", str(e)[:40])Opening the connection as mode=ro makes writes impossible at the database engine, so even a validator miss cannot mutate data. Two independent guards — validator plus read-only connection — is defense in depth; in production the read-only credential also points at a replica, never the primary.
Context: This is where the analyst becomes an agent. When a query fails validation or execution, feed the error back to the model and retry — bounded — so a wrong-column first attempt gets fixed on the next try.
Your task: Model the agentic loop: on a failed validation or execution, pass the error back to the SQL-writer and retry up to MAX_TRIES, returning the result and the number of tries once it succeeds.
Requirements:
- A retry loop bounded by
MAX_TRIES - The SQL writer receives the previous error (None on the first attempt)
- A validation failure feeds its reason back and retries
- An execution exception feeds its message back and retries
- On success, return the SQL, the rows, and the number of tries
- Show a wrong-column first attempt corrected on the retry
💡 Hint: Carry an error string across iterations and pass it into the writer each time; the loop exits on the first successful run_query.
Show solution
The self-correcting loop — errors become feedback (pure stdlib, runnable):
import re
MAX_TRIES = 5
def validate_sql(sql):
return (bool(re.match(r"(?is)^\s*select\b", sql.strip())), "SELECT only")
def ask(question, write_sql, run_query):
error = None
for attempt in range(MAX_TRIES):
sql = write_sql(question, error) # model sees the last error
ok, reason = validate_sql(sql)
if not ok:
error = reason; continue
try:
rows = run_query(sql)
except Exception as e:
error = str(e); continue # feed SQL error back and retry
return {"sql": sql, "rows": rows, "tries": attempt + 1}
return {"error": f"no valid query after {MAX_TRIES} tries"}
# stub model: first tries a bad column, then fixes it once it sees the error
def write_sql(q, error):
return "SELECT revenue FROM orders" if not error else "SELECT amount FROM orders"
def run_query(sql):
if "revenue" in sql: raise Exception("no such column: revenue")
return [(30,), (60,)]
print(ask("total amount", write_sql, run_query)) # corrected on 2nd tryWhen run_query raises "no such column", that error string is fed back so the model fixes its own query on the next iteration — the agentic self-correction loop from Chapter 4. The MAX_TRIES cap keeps it from looping forever on a genuinely impossible request.
Context: Scoring text-to-SQL on the SQL string is wrong — many different queries return the same answer. The right metric compares the agent's result set to a gold query's result set.
Your task: Implement execution-accuracy scoring over a golden set: run both the agent's SQL and the gold SQL and count a pass when their result sets match, not when their SQL text matches.
Requirements:
- A golden set of
(question, gold_sql)pairs against a seeded database - Execute both the agent's query and the gold query and fetch all rows
- Count a pass when the two result sets are equal
- Report accuracy as passed-over-total
- Include a case where different SQL (e.g.
count(*)vscount(id)) yields the same rows and still passes
💡 Hint: Compare fetchall() lists, not query strings — equivalent queries with different text must both count as correct.
Show solution
Execution accuracy — compare results, not SQL strings (pure stdlib, runnable):
import sqlite3
con = sqlite3.connect(":memory:")
con.executescript("""
CREATE TABLE customers(id, country);
INSERT INTO customers VALUES (1,'US'),(2,'US'),(3,'DE');
""")
def run(sql):
return con.execute(sql).fetchall()
GOLD = [
{"q": "how many US customers",
"gold_sql": "SELECT count(*) FROM customers WHERE country='US'"},
]
# stub agent produces a different-but-equivalent query
def agent_sql(q):
return "SELECT count(id) FROM customers WHERE country = 'US'"
passed = 0
for case in GOLD:
if run(agent_sql(case["q"])) == run(case["gold_sql"]):
passed += 1
print(f"execution accuracy: {passed}/{len(GOLD)}") # 1/1 -- same result setTwo different queries (count(*) vs count(id)) can be equally correct, so scoring on SQL text would wrongly fail the agent. Execution accuracy — do the result sets match? — is the honest text-to-SQL metric, alongside a hard-fail check that no non-SELECT ever reaches the database.
Context: As owner, the production risk is running model-generated SQL against real data. The milestone composes every guardrail into one gate and enumerates a phased rollout that keeps humans in front of it.
Your task: Compose the full safety boundary — read-only replica credential, validator, read-only connection, timeout, and an auto-applied LIMIT — into a single safe_execute gate, and enumerate the phased rollout.
Requirements:
- Validate the SQL is a single SELECT and reject write/DDL keywords
- Auto-append a LIMIT when the query lacks one
- Target a read-only replica with a read-only connection mode
- Apply a query timeout as a runaway guard
- Return the config that was enforced (mode, timeout, limit)
- Enumerate a phased rollout: analysts verify first, then vetted question types self-serve, always showing the SQL, never touching primary
💡 Hint: Layer the controls so each is independent insurance — credential, validator, connection mode, timeout, LIMIT — and the rollout earns autonomy one vetted question-type at a time.
Show solution
The composed safety gate for production (pure stdlib, runnable):
import re
def safe_execute(sql, run_ro, limit=100, timeout_s=5):
# 1. validator (belt)
s = sql.strip().rstrip(";")
if ";" in s or not re.match(r"(?is)^\s*select\b", s):
return {"blocked": "not a single SELECT"}
if re.search(r"(?i)\b(insert|update|delete|drop|alter|create)\b", s):
return {"blocked": "write/DDL"}
# 2. always bound the query
if "limit" not in s.lower():
s += f" LIMIT {limit}"
# 3. run on a read-only replica connection (suspenders), with a timeout
return {"ran": s, "timeout_s": timeout_s, "connection": "replica?mode=ro"}
print(safe_execute("SELECT country, sum(amount) FROM orders GROUP BY country",
run_ro=lambda s: None))
print(safe_execute("DELETE FROM orders", run_ro=lambda s: None)) # blockedIndustry scenario: a "talk to your database" tool for non-technical staff. Safety is layered infrastructure, not a prompt: a read-only credential on a replica, a validator, a read-only connection, and a timeout+LIMIT. The rollout is phased — analysts verify SQL first, then vetted question-types go self-serve (always showing the SQL) — and the tool never writes, never touches the primary, and never hides the query.
✓ Checkpoint — done when…
- The validator blocks every write and multi-statement; the read-only connection blocks writes at the engine.
- The agent answers English questions with correct SQL, self-correcting on errors.
- All safety tests pass with no API key.
- Execution accuracy matches the gold queries.
| Dimension | Meets the bar | Above the bar (staff-level) |
|---|---|---|
| Read-only safety | Writes are impossible two ways: a SELECT-only validator rejects mutating statements and the connection itself is read-only. | The validator resists obfuscation (CTEs, stacked statements, comments hiding a write); a bypass attempt fails a dedicated adversarial test. |
| Schema grounding | The model is grounded in the real schema, so it uses actual table/column names and correct joins rather than inventing them. | Grounding scales — only relevant tables are fed for large schemas — and the agent asks/abstains when a question can't be answered from the schema. |
| Execution accuracy | Generated queries are checked against gold queries for the right answer, not just for running without error. | Accuracy is measured on result equivalence (row sets match), catching queries that run cleanly but compute the wrong number. |
| Self-correction loop | SQL errors are fed back so the model repairs the query; the loop is bounded so it can't retry forever. | The loop distinguishes recoverable errors (typo, wrong column) from unanswerable questions and stops with an honest can't-answer rather than looping. |
| Answer explanation | The agent explains the answer in plain language tied to the query it ran, so a non-analyst can trust it. | The explanation surfaces assumptions (date range, filters, what was excluded) so a subtly-misread question is caught by the reader. |
| Cost & blast radius | Query cost/latency is bounded (row limits, timeouts) so a broad question can't run a table-scan that hammers the warehouse. | Expensive queries are estimated and capped before running; the agent runs against a replica/limited role, not the primary. |
Score each row 0 (missing) / 1 (meets) / 2 (above). A passing analyst build is 9+/12 with read-only safety at 2 — any path that can mutate data, or that runs an unbounded query against the primary, is an automatic fail regardless of how accurate the SQL is.
Knowledge check check yourself
Why is the read-only boundary enforced with a read-only database credential on a replica rather than by instructing the model not to write?
Show answer
What role does feeding SQL errors back to the agent play, and why does it improve accuracy?
Show answer
run_query returns an error like "column x does not exist", that error is returned as a tool result so the agent can read it, fix the query, and retry. This self-correction loop (Ch 4 agentic loop) often gets it right on the second try, dramatically improving accuracy.