AI SQL Generation from Text Queries (Text-to-SQL)
Text-to-SQL allows business analysts and managers to ask database questions directly without SQL knowledge. The key technical challenge is not SQL syntax generation (LLM handles that easily), but providing precise schema context: what tables exist, how they relate, what values are valid in enum fields.
Text-to-SQL System Architecture
from anthropic import Anthropic
import psycopg2
import json
from typing import Optional
from dataclasses import dataclass
client = Anthropic()
@dataclass
class QueryResult:
sql: str
explanation: str
rows: list[dict]
error: Optional[str] = None
class TextToSQLEngine:
def __init__(self, connection_string: str):
self.conn = psycopg2.connect(connection_string)
self.schema_cache: dict = {}
def get_schema(self, tables: list[str] = None) -> str:
"""Retrieves DDL schema from PostgreSQL"""
query = """
SELECT
t.table_name,
c.column_name,
c.data_type,
c.is_nullable,
c.column_default,
tc.constraint_type,
kcu.column_name as fk_column,
ccu.table_name as fk_table
FROM information_schema.tables t
JOIN information_schema.columns c ON t.table_name = c.table_name
LEFT JOIN information_schema.key_column_usage kcu
ON c.table_name = kcu.table_name AND c.column_name = kcu.column_name
LEFT JOIN information_schema.table_constraints tc
ON kcu.constraint_name = tc.constraint_name
LEFT JOIN information_schema.constraint_column_usage ccu
ON tc.constraint_name = ccu.constraint_name
WHERE t.table_schema = 'public'
"""
if tables:
placeholders = ",".join(["%s"] * len(tables))
query += f" AND t.table_name IN ({placeholders})"
with self.conn.cursor() as cur:
cur.execute(query, tables or [])
rows = cur.fetchall()
# Format as DDL
tables_dict = {}
for row in rows:
table_name = row[0]
if table_name not in tables_dict:
tables_dict[table_name] = {"columns": [], "foreign_keys": []}
col_def = f" {row[1]} {row[2].upper()}"
if row[3] == "NO":
col_def += " NOT NULL"
if row[4]:
col_def += f" DEFAULT {row[4]}"
if row[5] == "PRIMARY KEY":
col_def += " PRIMARY KEY"
tables_dict[table_name]["columns"].append(col_def)
if row[5] == "FOREIGN KEY" and row[7]:
tables_dict[table_name]["foreign_keys"].append(
f" FOREIGN KEY ({row[6]}) REFERENCES {row[7]}"
)
ddl_parts = []
for table, info in tables_dict.items():
ddl = f"CREATE TABLE {table} (\n"
ddl += ",\n".join(info["columns"])
if info["foreign_keys"]:
ddl += ",\n" + ",\n".join(info["foreign_keys"])
ddl += "\n);"
ddl_parts.append(ddl)
return "\n\n".join(ddl_parts)
def get_sample_values(self, important_columns: dict[str, list[str]]) -> str:
"""Gets sample values for enum/category fields"""
samples = []
with self.conn.cursor() as cur:
for table_col, _ in important_columns.items():
table, col = table_col.split(".")
try:
cur.execute(
f"SELECT DISTINCT {col} FROM {table} LIMIT 10"
)
values = [str(row[0]) for row in cur.fetchall()]
samples.append(f"-- {table}.{col}: {', '.join(values)}")
except Exception:
pass
return "\n".join(samples)
def generate_sql(self, question: str, context_tables: list[str] = None) -> QueryResult:
"""Generates SQL from text question"""
schema = self.get_schema(context_tables)
# Additional context: sample values for string fields
sample_values = self._get_relevant_samples(question)
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=2048,
system="""You are a SQL and PostgreSQL expert.
Generate precise, optimized SQL queries based on database schema.
Rules:
- Use only existing tables and columns from schema
- Prefer JOINs over subqueries where possible
- Add LIMIT 1000 for queries without aggregation
- For dates use PostgreSQL functions: DATE_TRUNC, NOW(), EXTRACT
- Always add ORDER BY for predictable results
- If question is ambiguous — choose most likely interpretation
Return JSON:
{
"sql": "<SQL query>",
"explanation": "<what query does, 1-2 sentences>",
"assumptions": ["<assumption 1 if any>"]
}""",
messages=[{
"role": "user",
"content": f"""Question: {question}
Database schema:
```sql
{schema}
{f"Sample values:{chr(10)}{sample_values}" if sample_values else ""}""" }] )
text = response.content[0].text
try:
# Parse JSON response
start = text.find("{")
end = text.rfind("}") + 1
data = json.loads(text[start:end])
sql = data["sql"]
explanation = data.get("explanation", "")
# Execute query
rows = self._execute_safe(sql)
return QueryResult(sql=sql, explanation=explanation, rows=rows)
except Exception as e:
return QueryResult(sql="", explanation="", rows=[], error=str(e))
def _execute_safe(self, sql: str) -> list[dict]:
"""Executes SELECT queries only"""
sql_upper = sql.strip().upper()
if not sql_upper.startswith("SELECT") and not sql_upper.startswith("WITH"):
raise ValueError("Only SELECT queries are allowed")
with self.conn.cursor() as cur:
cur.execute(sql)
columns = [desc[0] for desc in cur.description]
rows = cur.fetchall()
return [dict(zip(columns, row)) for row in rows]
def _get_relevant_samples(self, question: str) -> str:
"""Simple heuristic to identify relevant enum fields"""
# In real system — LLM determines needed fields
return ""
### Self-Correcting Generator
```python
class SelfCorrectingTextToSQL:
"""Iteratively corrects SQL on execution errors"""
def __init__(self, engine: TextToSQLEngine):
self.engine = engine
def query(self, question: str, max_attempts: int = 3) -> QueryResult:
"""Generates SQL with automatic error correction"""
result = self.engine.generate_sql(question)
if not result.error:
return result
# Iteratively correct
messages = [{
"role": "user",
"content": f"Question: {question}\n\nGenerated query:\n```sql\n{result.sql}\n```\n\nError: {result.error}\n\nFix the query."
}]
for attempt in range(max_attempts - 1):
response = client.messages.create(
model="claude-sonnet-4-5",
max_tokens=1024,
system="You are a SQL expert. Fix SQL queries based on execution errors. Return only fixed SQL.",
messages=messages,
)
fixed_sql = response.content[0].text.strip()
if "```sql" in fixed_sql:
fixed_sql = fixed_sql.split("```sql")[1].split("```")[0].strip()
try:
rows = self.engine._execute_safe(fixed_sql)
return QueryResult(sql=fixed_sql, explanation="Auto-corrected", rows=rows)
except Exception as e:
messages.append({"role": "assistant", "content": response.content[0].text})
messages.append({"role": "user", "content": f"Still error: {e}"})
return QueryResult(sql=result.sql, rows=[], error="Max attempts reached", explanation="")
Natural Language Interface with History
class ConversationalDataAnalyst:
"""Conversational interface for data work"""
def __init__(self, connection_string: str):
self.engine = TextToSQLEngine(connection_string)
self.history: list[dict] = []
self.last_sql: str = ""
def ask(self, question: str) -> str:
"""Answers question considering dialog history"""
# Add previous query context
context = ""
if self.last_sql:
context = f"\nPrevious query:\n```sql\n{self.last_sql}\n```"
# Support clarifying questions
if any(word in question.lower() for word in ["and also", "now", "add", "also"]):
enhanced = f"Based on previous query, {question}"
else:
enhanced = question
result = self.engine.generate_sql(enhanced + context)
if result.error:
return f"Query execution error: {result.error}"
self.last_sql = result.sql
self.history.append({"question": question, "sql": result.sql})
# Format result
if not result.rows:
return "Query executed successfully, no data found."
response_text = f"{result.explanation}\n\n"
response_text += f"SQL: `{result.sql}`\n\n"
response_text += f"Results ({len(result.rows)} rows):\n"
# Results table
if result.rows:
headers = list(result.rows[0].keys())
response_text += " | ".join(headers) + "\n"
response_text += " | ".join(["---"] * len(headers)) + "\n"
for row in result.rows[:10]:
response_text += " | ".join(str(v) for v in row.values()) + "\n"
if len(result.rows) > 10:
response_text += f"... and {len(result.rows) - 10} more rows"
return response_text
Practical Case: E-commerce Analytics
Task: product managers submitted tasks to analysts (2–5 day wait) because they didn't know SQL. Database: PostgreSQL, 23 tables, ~50M records.
Implementation:
- Text-to-SQL interface in Slack:
/data <question> - Allowlist of tables for product teams (without personal data)
- Caching frequent questions
Metrics:
- ad-hoc queries from product teams without analyst involvement: 0 → 23 per week
- Time to answer simple question: 2 days → 30 seconds
- Accuracy of generated SQL: 89% (no edits needed)
- 11% of queries required iterative clarification via dialog
Timeline
- Basic engine (schema extraction + SQL generation): 1 week
- Self-correcting generator + history: 1-2 weeks
- Conversational interface with Slack integration: 1 week
- Query caching + performance optimization: 1 week