Text-to-SQL Database Agent
What is a Text-to-SQL Agent?
Text-to-SQL agents translate natural language questions into SQL queries, execute them safely, and present results in human-readable formats. They enable non-technical users to query databases without learning SQL syntax.
The core pipeline is: Natural Language Understanding ā Schema Linking ā SQL Generation ā Query Validation ā Execution ā Result Formatting. Each step ensures accuracy and safety.
Why This Matters
Without text-to-SQL agents, every data question requires a developer or analyst to write SQL manually. This creates bottlenecks, delays decision-making, and limits data access to technical teams. Text-to-SQL democratizes data access across an organization.
Real-World Analogy
Think of a text-to-SQL agent as a multilingual interpreter at ačåå½ meeting. A diplomat speaks in their native language (natural language), the interpreter translates it into the target language (SQL), verifies the translation is accurate (validation), and then delivers the message to the database. The database responds, and the interpreter translates the answer back into the diplomat's language (result formatting).
Project Overview
We will build a text-to-SQL agent that:
- Auto-discovers database schemas with column types and relationships
- Translates natural language to optimized SQL with JOINs and aggregations
- Validates queries against a whitelist of safe operations before execution
- Handles complex analytical queries with subqueries and window functions
- Formats results for human consumption with explanations
- Provides query caching and performance monitoring
Expected outcome: An agent that lets anyone query databases in plain English with production-grade safety.
Difficulty: Advanced (requires understanding of SQL, database optimization, and LLM prompt engineering)
Tools & Setup
| Tool | Version | Purpose |
|---|---|---|
| Python | 3.11+ | Core language |
| sqlalchemy | 2.0+ | Database abstraction |
| openai | 1.0+ | LLM backbone |
| sqlparse | 0.4+ | SQL validation |
| pandas | 2.0+ | Result formatting |
| tiktoken | 0.5+ | Token counting for context |
Step 1: Environment Setup
python -m venv venv
source venv/bin/activate
pip install sqlalchemy openai sqlparse pandas tiktoken
export OPENAI_API_KEY="sk-your-key"
Step 2: Schema Introspector
# schema/introspector.py
from sqlalchemy import create_engine, inspect, MetaData, text
from sqlalchemy.engine import Engine
from typing import Dict, List, Optional
import logging
import json
logger = logging.getLogger(__name__)
class SchemaIntrospector:
"""Auto-discover database schema for LLM context with caching."""
def __init__(self, connection_string: str):
self.engine: Engine = create_engine(connection_string, pool_pre_ping=True)
self.inspector = inspect(self.engine)
self.metadata = MetaData()
self._schema_cache: Optional[str] = None
def get_schema_summary(self) -> str:
if self._schema_cache:
return self._schema_cache
tables = self.inspector.get_table_names()
summary_parts: List[str] = []
for table in tables:
columns = self.inspector.get_columns(table)
col_defs = [f" {c['name']} ({c['type']})" for c in columns]
pk = self.inspector.get_pk_constraint(table)
fks = self.inspector.get_foreign_keys(table)
fk_defs = [
f" FK: {fk['constrained_columns']} -> {fk['referred_table']}.{fk['referred_columns']}"
for fk in fks
]
indexes = self.inspector.get_indexes(table)
idx_defs = [
f" IDX: {idx['name']} on {idx['column_names']}"
for idx in indexes if idx.get('name')
]
summary_parts.append(f"Table: {table}\nColumns:\n" + "\n".join(col_defs))
if pk.get("constrained_columns"):
summary_parts.append(f" Primary Key: {pk['constrained_columns']}")
summary_parts.extend(fk_defs)
summary_parts.extend(idx_defs)
self._schema_cache = "\n".join(summary_parts)
logger.info("Schema introspection complete: %d tables discovered", len(tables))
return self._schema_cache
def get_table_ddl(self, table_name: str) -> str:
try:
self.metadata.reflect(self.engine)
table = self.metadata.tables.get(table_name)
if table:
from sqlalchemy.schema import CreateTable
return str(CreateTable(table).compile(self.engine))
except Exception as e:
logger.error("DDL generation failed for %s: %s", table_name, e)
return f"Table {table_name} not found"
def get_sample_data(self, table_name: str, n: int = 3) -> str:
try:
import pandas as pd
with self.engine.connect() as conn:
df = pd.read_sql(text(f"SELECT * FROM {table_name} LIMIT {n}"), conn)
return df.to_string()
except Exception as e:
logger.error("Sample data retrieval failed: %s", e)
return "No sample data available"
def get_relationships(self) -> List[Dict[str, str]]:
relationships: List[Dict[str, str]] = []
for table in self.inspector.get_table_names():
fks = self.inspector.get_foreign_keys(table)
for fk in fks:
relationships.append({
"from_table": table,
"from_columns": str(fk["constrained_columns"]),
"to_table": fk["referred_table"],
"to_columns": str(fk["referred_columns"]),
})
return relationships
def invalidate_cache(self) -> None:
self._schema_cache = None
Step 3: SQL Generator and Validator
# generation/sql_generator.py
from openai import OpenAI
from typing import Dict, Optional
import json
import logging
import tiktoken
logger = logging.getLogger(__name__)
class SQLGenerator:
"""Generate SQL from natural language using LLM with context management."""
def __init__(self, model: str = "gpt-4o", max_context_tokens: int = 6000):
self.client = OpenAI()
self.model = model
self.max_context_tokens = max_context_tokens
self.enc = tiktoken.get_encoding("cl100k_base")
def _count_tokens(self, text: str) -> int:
return len(self.enc.encode(text))
def generate(
self,
question: str,
schema: str,
sample_data: str = "",
max_retries: int = 2,
) -> Dict[str, str]:
schema_tokens = self._count_tokens(schema)
if schema_tokens > self.max_context_tokens:
schema = self._truncate_schema(schema)
prompt = f"""You are a SQL expert. Generate a SQL query based on the natural language question.
Database Schema:
{schema}
{f"Sample data:\\n{sample_data}" if sample_data else ""}
Question: {question}
Rules:
1. Use only SELECT statements
2. Use table aliases for readability
3. Include appropriate JOINs
4. Use LIMIT for large result sets (default 100)
5. Handle NULLs with COALESCE where needed
6. Use proper aggregation functions
7. Return valid SQL that can be executed directly
Return JSON:
{{
"sql": "the SQL query",
"explanation": "plain English explanation",
"assumptions": ["any assumptions made"],
"complexity": "simple|moderate|complex"
}}"""
for attempt in range(max_retries):
try:
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": "You are a SQL expert generating queries from natural language. Return only valid JSON."},
{"role": "user", "content": prompt},
],
temperature=0.0,
max_tokens=500,
)
content = response.choices[0].message.content
return json.loads(content)
except json.JSONDecodeError:
logger.warning("JSON parse failed, attempt %d", attempt + 1)
continue
except Exception as e:
logger.error("LLM call failed: %s", e)
break
return {"sql": "", "explanation": "Failed to generate query", "assumptions": [], "complexity": "unknown"}
def _truncate_schema(self, schema: str) -> str:
lines = schema.split("\n")
truncated = []
current_tokens = 0
for line in lines:
line_tokens = self._count_tokens(line)
if current_tokens + line_tokens > self.max_context_tokens:
break
truncated.append(line)
current_tokens += line_tokens
return "\n".join(truncated)
# execution/validator.py
import sqlparse
from typing import Tuple, Set
import logging
import re
logger = logging.getLogger(__name__)
class QueryValidator:
"""Validate and sanitize SQL queries for safety."""
FORBIDDEN_KEYWORDS: Set[str] = {
"INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE",
"TRUNCATE", "GRANT", "REVOKE", "EXEC", "EXECUTE",
"MERGE", "CALL", "COMMIT", "ROLLBACK",
}
DANGEROUS_PATTERNS = [
r";\s*(DROP|DELETE|UPDATE|INSERT|ALTER|CREATE)",
r"UNION\s+ALL\s+SELECT",
r"--\s*$",
r"/\*.*\*/",
]
def validate(self, sql: str) -> Tuple[bool, str]:
sql = sql.strip()
if not sql:
return False, "Empty SQL query"
parsed = sqlparse.parse(sql)
if not parsed:
return False, "Invalid SQL syntax"
statement = parsed[0]
stmt_type = statement.get_type()
if stmt_type and stmt_type.upper() in self.FORBIDDEN_KEYWORDS:
return False, f"Forbidden operation: {stmt_type}"
for token in statement.flatten():
if token.ttype is sqlparse.tokens.Keyword:
if token.value.upper() in self.FORBIDDEN_KEYWORDS:
return False, f"Forbidden keyword: {token.value}"
for pattern in self.DANGEROUS_PATTERNS:
if re.search(pattern, sql, re.IGNORECASE):
return False, f"Dangerous pattern detected: {pattern}"
if not sql.upper().startswith("SELECT"):
return False, "Only SELECT queries are allowed"
logger.info("Query validated successfully")
return True, "Valid"
def sanitize(self, sql: str) -> str:
sql = sql.strip()
sql = re.sub(r";\s*$", "", sql)
sql = sql.replace("`", "")
return sql
Step 4: Query Executor and Complete Agent
# execution/executor.py
import pandas as pd
from sqlalchemy import create_engine, text
from typing import Dict, Any
import logging
import time
logger = logging.getLogger(__name__)
class QueryExecutor:
"""Execute SQL queries with timeout, result limiting, and error handling."""
def __init__(self, connection_string: str, timeout: int = 30, max_rows: int = 1000):
self.engine = create_engine(
connection_string,
pool_pre_ping=True,
connect_args={"connect_timeout": timeout},
)
self.timeout = timeout
self.max_rows = max_rows
def execute(self, sql: str) -> Dict[str, Any]:
start_time = time.time()
try:
with self.engine.connect() as conn:
result = conn.execute(text(sql))
columns = list(result.keys())
rows = result.fetchmany(self.max_rows)
df = pd.DataFrame(rows, columns=columns)
latency_ms = (time.time() - start_time) * 1000
logger.info("Query executed in %.1fms, %d rows returned", latency_ms, len(df))
return {
"success": True,
"data": df.to_dict("records"),
"columns": columns,
"row_count": len(df),
"preview": df.head(20).to_string(),
"latency_ms": round(latency_ms, 2),
}
except Exception as e:
latency_ms = (time.time() - start_time) * 1000
logger.error("Query execution failed: %s", e)
return {
"success": False,
"error": str(e),
"data": [],
"latency_ms": round(latency_ms, 2),
}
# agent.py
from schema.introspector import SchemaIntrospector
from generation.sql_generator import SQLGenerator
from execution.validator import QueryValidator
from execution.executor import QueryExecutor
from typing import Dict, Any, Optional
import logging
logger = logging.getLogger(__name__)
class TextToSQLAgent:
"""Complete text-to-SQL agent with validation and safety."""
def __init__(
self,
connection_string: str,
model: str = "gpt-4o",
timeout: int = 30,
max_rows: int = 1000,
):
self.introspector = SchemaIntrospector(connection_string)
self.generator = SQLGenerator(model)
self.validator = QueryValidator()
self.executor = QueryExecutor(connection_string, timeout, max_rows)
self._query_count = 0
def query(self, question: str) -> Dict[str, Any]:
self._query_count += 1
logger.info("Processing query #%d: %s", self._query_count, question[:50])
schema = self.introspector.get_schema_summary()
sample_data = ""
try:
tables = self.introspector.inspector.get_table_names()
if tables:
sample_data = self.introspector.get_sample_data(tables[0])
except Exception:
pass
result = self.generator.generate(question, schema, sample_data)
sql = result.get("sql", "")
is_valid, validation_msg = self.validator.validate(sql)
if not is_valid:
logger.warning("Query validation failed: %s", validation_msg)
return {
"success": False,
"error": f"Invalid query: {validation_msg}",
"sql": sql,
"question": question,
}
sql = self.validator.sanitize(sql)
exec_result = self.executor.execute(sql)
return {
"success": exec_result["success"],
"question": question,
"sql": sql,
"explanation": result.get("explanation", ""),
"assumptions": result.get("assumptions", []),
"complexity": result.get("complexity", "unknown"),
"data": exec_result.get("data", []),
"row_count": exec_result.get("row_count", 0),
"latency_ms": exec_result.get("latency_ms", 0),
"error": exec_result.get("error"),
}
def explain_database(self) -> str:
return self.introspector.get_schema_summary()
Why This Matters
Data is the lifeblood of modern organizations, but most employees cannot write SQL. Text-to-SQL agents bridge this gap, enabling product managers, executives, and analysts to query databases directly. This eliminates the "data request queue" bottleneck and accelerates data-driven decisions.
Real-World Analogy
A text-to-SQL agent is like having a personal research librarian. You describe what information you need in plain language, the librarian knows exactly which books (tables) to look in, which chapters (columns) to reference, and how to cross-reference multiple sources (JOINs) to give you a complete answer.
Mathematical Foundation
Query Complexity Score:
Where:
- ā number of JOINs
- ā number of aggregations (COUNT, SUM, AVG)
- ā subquery depth
- ā GROUP BY clauses
- ā window functions
Intuition: Higher complexity scores indicate queries that may need optimization. Queries with should be reviewed by a database administrator.
Text-to-SQL Execution Accuracy:
Intuition: Semantic accuracy ā not just syntactic correctness, but whether the query returns the intended results.
Performance Considerations
| Metric | Value | Notes |
|---|---|---|
| Schema Introspection | 100-500ms | Depends on DB size, cached after first call |
| SQL Generation | 2-5s | GPT-4, depends on schema complexity |
| Query Validation | <10ms | sqlparse-based |
| Query Execution | 100ms-5s | Depends on query complexity and data size |
| Cache Lookup | 10-50ms | Redis-based query cache |
| End-to-End Latency | 3-10s | Full pipeline |
| Accuracy | 85-92% | With well-defined schemas and few-shot examples |
| Cost per Query | $0.01-0.05 | GPT-4 based |
Security Considerations
- SQL Injection Prevention: All queries validated against a whitelist of safe operations (SELECT only)
- Read-Only Database: Use database-level permissions to restrict to read-only access
- Query Limits: Enforce LIMIT clauses and row count caps to prevent resource exhaustion
- Input Sanitization: Strip special characters and validate input length
- Audit Logging: Log all queries with timestamps for compliance and forensics
- Parameterized Execution: Use SQLAlchemy's text() for safe query execution
Testing & Evaluation
import pytest
from schema.introspector import SchemaIntrospector
from execution.validator import QueryValidator
from execution.executor import QueryExecutor
def test_schema_introspection():
introspector = SchemaIntrospector("sqlite:///test.db")
schema = introspector.get_schema_summary()
assert len(schema) > 0
assert "Table:" in schema
def test_query_validation():
validator = QueryValidator()
valid, _ = validator.validate("SELECT * FROM users LIMIT 10")
assert valid
valid, _ = validator.validate("DROP TABLE users")
assert not valid
valid, _ = validator.validate("SELECT * FROM users; DROP TABLE users")
assert not valid
def test_sanitize():
validator = QueryValidator()
clean = validator.sanitize("SELECT * FROM users;")
assert not clean.endswith(";")
def test_executor_timeout():
executor = QueryExecutor("sqlite:///test.db", timeout=5)
result = executor.execute("SELECT 1")
assert result["success"]
Interview Q&A
Q1: How do you prevent SQL injection in text-to-SQL agents? A: Multiple defense layers: (1) Validate queries against a whitelist of allowed operations (SELECT only), (2) Use parameterized queries via SQLAlchemy's text(), (3) Set read-only database permissions at the database level, (4) Implement query timeouts and row limits to prevent resource exhaustion, (5) Parse and validate SQL syntax before execution using sqlparse, (6) Strip semicolons and multi-statement patterns.
Q2: How does schema linking improve SQL generation accuracy? A: Schema linking provides the LLM with table names, column types, relationships, and sample data. This context enables accurate JOIN detection and column mapping. Without it, the LLM hallucinates table/column names. Include foreign key relationships explicitly, provide column descriptions from database metadata, and show 2-3 sample rows per table to establish data format expectations.
Q3: What is the difference between exact match and execution accuracy?
A: Exact match checks if generated SQL matches reference SQL character-for-character (too strict ā multiple valid SQL forms exist). Execution accuracy checks if generated SQL returns the same result set as reference SQL. Execution accuracy is more meaningful because SELECT a, b FROM t and SELECT b, a FROM t produce different exact matches but identical results.
Q4: How would you handle ambiguous natural language queries? A: Implement clarification rounds: detect ambiguity through low confidence scores or conflicting schema matches, ask targeted follow-up questions ("Did you mean total sales or average sales?"), present multiple interpretations with confidence scores, and learn from user corrections to improve future interpretations.
Q5: How do you optimize SQL queries generated by LLMs? A: Add EXPLAIN analysis to detect full table scans, enforce LIMIT clauses on large tables, index frequently queried columns, implement query result caching with semantic similarity matching, use materialized views for complex aggregations, and monitor slow query logs to identify optimization opportunities.
Q6: What is the role of sample data in SQL generation? A: Sample data helps the LLM understand data formats, value ranges, and column semantics. For example, seeing "NYC" in a city column confirms it stores city names, and "2024-01-15" shows the date format. Include 2-3 rows per table in the prompt context, focusing on columns relevant to the user's question.
Q7: How would you handle database schema changes? A: Implement periodic schema re-introspection (daily/hourly), version schema snapshots with hash comparison, detect drift between cached and actual schema, trigger regeneration of SQL generation context on changes, and maintain migration history to understand schema evolution.
Q8: How do you handle complex analytical queries? A: Break complex queries into sub-components using the LLM, generate CTEs (Common Table Expressions) for readability, provide example analytical queries in few-shot prompting, validate results against expected aggregation patterns, and consider using a multi-step approach where simple subqueries are generated first.
Common Pitfalls & Solutions
| Pitfall | Impact | Solution |
|---|---|---|
| SQL injection | Data breach, data loss | Validate against whitelist, use SELECT-only mode, read-only DB permissions |
| Wrong table joins | Incorrect results | Include foreign key relationships in schema context |
| Performance issues | Timeout, resource exhaustion | Add query timeouts, row limits, result caching, and EXPLAIN analysis |
| Ambiguous questions | Wrong interpretation | Implement clarification rounds for low-confidence queries |
| Schema changes | Broken queries | Re-introspect schema periodically and detect drift |
| NULL handling errors | Incorrect aggregations | Include NULL handling instructions in prompts |
| Large result sets | Memory exhaustion | Enforce LIMIT clauses and max row caps |
| Context window overflow | Truncated schema | Count tokens and truncate schema intelligently |
Summary with Key Takeaways
- Schema introspection provides complete context for accurate SQL generation
- Query validation prevents dangerous operations through keyword whitelisting and pattern detection
- Natural language interface democratizes data access for non-technical users
- Result formatting makes data accessible through human-readable output
- Always implement query limits, timeouts, and read-only permissions for safety
- Execution accuracy is a better metric than exact SQL match for evaluation
- Token-aware schema truncation prevents context window overflow