Agentic Data Engineering: NL-to-SQL Agents
Source: mortalapps.com- NL-to-SQL agents autonomously translate natural language queries into executable database queries, introspecting schemas and executing them in real-time.
- Solves the brittleness of static text-to-SQL pipelines by introducing iterative error correction, schema discovery, and self-debugging loops.
- Critical for enterprise self-service analytics, automated data engineering pipelines, and real-time operational reporting.
- Implements strict runtime sandboxing, read-only transaction boundaries, and query validation to mitigate SQL injection and data leakage.
- Requires careful context management to avoid token exhaustion when dealing with large database schemas.
Why This Matters
Traditional business intelligence (BI) architectures rely on rigid semantic layers, pre-calculated aggregates, and static dashboards. While these systems provide predictable performance, they fail to support ad-hoc, exploratory data analysis for non-technical users, creating a persistent bottleneck for data-driven organizations. The emergence of the natural language to sql ai agent represents a paradigm shift in data engineering, enabling dynamic, conversational access to relational databases. Unlike legacy text-to-SQL models that perform simple pattern matching and fail when schemas evolve, an agentic system dynamically introspects the database catalog, resolves schema ambiguities, and autonomously recovers from execution errors.
If ignored in production, naive text-to-SQL implementations present severe security and operational risks. A single unvalidated query can lead to SQL injection, unauthorized data exposure, or database denial-of-service (DoS) caused by unindexed cartesian joins. Furthermore, static prompts fail to adapt to schema migrations, leading to silent failures or corrupted data payloads.
An agentic architecture mitigates these risks by wrapping the LLM in a closed-loop system of verification and execution. By using Abstract Syntax Tree (AST) parsers, read-only transaction boundaries, and iterative self-correction loops, the agent can safely explore schemas, generate precise queries, and debug its own syntax errors before presenting results to the user. This approach should be deployed when users require unstructured, real-time access to complex relational databases. It is superior to vector-search RAG when queries require precise mathematical aggregation, filtering, or multi-table joins, and superior to static dashboards when the analytical questions cannot be predicted in advance.
Core Concepts
To build a production-grade NL-to-SQL agent, you must understand several core architectural concepts:
- Schema Introspection: The process of dynamically querying database catalogs (such as
information_schema) to retrieve table definitions, column types, primary keys, and foreign keys at runtime. - Iterative Query Correction Loop: A feedback mechanism where database execution errors (e.g., syntax errors, missing columns) are captured, formatted, and fed back to the LLM to regenerate and correct the SQL query.
- Schema Injection: A security vulnerability where malicious user input manipulates the agent into exposing, modifying, or deleting the database schema itself.
- Read-Only Transaction Boundary: A database-level constraint (e.g.,
SET TRANSACTION READ ONLY) that prevents the agent from executing mutating queries likeINSERT,UPDATE,DELETE, orDROP. - AST Parsing (Abstract Syntax Tree): Analyzing the structure of a generated SQL query before execution to programmatically verify that it complies with security policies and does not contain destructive commands.
- Semantic Schema Mapping: A metadata layer that maps natural language terms to specific database tables and columns, reducing ambiguity for the LLM.
How It Works
An agentic NL-to-SQL workflow operates as a closed-loop system, moving from user intent to validated execution through the following phases:
Phase 1: Intent Parsing and Schema Pruning
When a user submits a natural language query, the agent first parses the intent to identify which database tables and columns are required. Instead of injecting the entire database schema into the LLM context window - which causes token exhaustion and model confusion - the agent queries a metadata index (often stored in a vector database or a lightweight lookup table) to retrieve only the relevant table schemas and column descriptions.
Phase 2: SQL Generation and AST Validation
Using the pruned schema and the user's query, the agent generates a candidate SQL statement. Before this statement is sent to the database, it is intercepted by an Abstract Syntax Tree (AST) parser. The parser analyzes the query structure to ensure it is a read-only SELECT statement and does not target restricted system tables or administrative functions. If validation fails, the query is rejected immediately without hitting the database.
Phase 3: Execution and Error Capture
If the query passes AST validation, it is executed against the target database using a dedicated, low-privilege database connection. The connection is wrapped in a read-only transaction block. If the query executes successfully, the raw tabular data is returned. If the database returns an error (e.g., a syntax error, a non-existent column, or a type mismatch), the execution engine captures the raw error message and database state.
Phase 4: Iterative Self-Correction
Upon capturing an execution error, the agent enters a self-correction loop. It constructs a new prompt containing the original user query, the pruned schema, the failed SQL statement, and the exact database error message. The LLM analyzes why the query failed, generates a corrected SQL statement, and passes it back to Phase 2. This loop repeats until the query executes successfully or the maximum retry budget (typically 3 attempts) is exhausted.
Architecture
The architecture of an NL-to-SQL agent consists of seven primary components. Execution begins when a user submits a natural language query to the Agent Orchestrator. The Orchestrator first calls the Schema Retriever, which queries a metadata store to return a pruned schema containing only the tables relevant to the user's request. The Orchestrator then passes the user query and pruned schema to the SQL Generator.
The SQL Generator outputs a candidate SQL string, which is passed directly to the AST Guardrail Parser. If the parser detects any non-select statements or unauthorized table access, it rejects the query and sends the validation error back to the SQL Generator. If the query passes validation, it is sent to the Database Executor.
The Database Executor runs the query against the target database using a read-only connection pool. If the database returns an execution error, the Error Handler formats the error and routes it back to the SQL Generator to trigger the self-correction loop. If execution succeeds, the raw dataset is returned to the Orchestrator, which formats the data and returns the final answer to the user.
Schema Pruning and Context Management
In enterprise databases with hundreds of tables and thousands of columns, injecting the entire schema into the LLM prompt is impossible due to context window limits and performance degradation. Production systems must implement a two-stage schema retrieval pipeline.
Stage 1 uses semantic search to find candidate tables. Table names, column names, and descriptions are indexed in a vector database. When a user query arrives, the system retrieves the top $N$ (typically 3 to 5) most relevant tables. Stage 2 dynamically constructs a minimal DDL (Data Definition Language) representation of only these selected tables, including primary and foreign key relationships to assist the LLM in generating correct joins.
AST Parsing and SQL Injection Prevention
Preventing SQL injection and unauthorized data access is the most critical requirement of an NL-to-SQL agent. Relying on the LLM to follow system instructions (e.g., "only write SELECT queries") is insufficient, as prompt injection can easily bypass these guardrails.
To guarantee security, the system must parse the generated SQL into an Abstract Syntax Tree (AST) using libraries like sqlglot or sqlparse before execution. The parser traverses the AST and enforces the following rules:
- Statement Type Verification: Ensure the root node of the AST is a
SELECTstatement. Reject any AST containingINSERT,UPDATE,DELETE,DROP,ALTER, orCREATEnodes. - Table Whitelisting: Verify that all tables referenced in the query exist within the user's authorized schema boundary.
- Function Blacklisting: Block database-specific administrative functions (e.g.,
pg_sleep(),current_setting(), or metadata functions) that could be used for side-channel attacks or schema harvesting.
The Iterative Self-Correction Loop
When a generated SQL query fails during execution, the agent must not fail silently or return a generic error to the user. Instead, it must leverage the database's compiler feedback to correct its own mistake.
The self-correction prompt must be highly structured. It should contain:
- The original user intent.
- The schema definition provided during generation.
- The exact SQL query that failed.
- The database engine's error message (e.g.,
Relation "users" does not existorColumn "created_at" is ambiguous).
By analyzing this feedback, the LLM can identify if it made a join error, misspelled a column name, or used an incompatible data type, and output a corrected query. To prevent infinite loops and runaway token costs, the orchestrator must enforce a strict recursion limit (typically 3 attempts) and log each iteration for observability.
Code Example
import os
import sqlite3
import logging
from typing import Dict, Any, List, Tuple
# Using standard library or simple mock to ensure 100% runnability without external API keys
# If OPENAI_API_KEY is present, it can be integrated here.
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger("NLtoSQLAgent")
class DatabaseSandbox:
def __init__(self):
# Initialize an in-memory SQLite database for demonstration
self.conn = sqlite3.connect(":memory:")
self.setup_mock_data()
def setup_mock_data(self):
cursor = self.conn.cursor()
cursor.execute("""
CREATE TABLE departments (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL
)
""")
cursor.execute("""
CREATE TABLE sales (
id INTEGER PRIMARY KEY,
department_id INTEGER,
amount REAL NOT NULL,
sale_date TEXT NOT NULL,
FOREIGN KEY(department_id) REFERENCES departments(id)
)
""")
# Insert sample data
cursor.executemany("INSERT INTO departments VALUES (?, ?)", [(1, "Engineering"), (2, "Sales")])
cursor.executemany("INSERT INTO sales VALUES (?, ?, ?, ?)", [
(1, 1, 1500.0, "2026-03-01"),
(2, 1, 2500.0, "2026-03-02"),
(3, 2, 5000.0, "2026-03-01")
])
self.conn.commit()
def get_schema(self) -> str:
cursor = self.conn.cursor()
cursor.execute("SELECT name, sql FROM sqlite_master WHERE type='table'")
tables = cursor.fetchall()
schema_desc = []
for table_name, sql in tables:
schema_desc.append(f"Table: {table_name}
DDL: {sql}")
return "
".join(schema_desc)
def execute_query(self, sql: str) -> Tuple[List[Tuple], List[str], str]:
"""Executes query in a read-only transaction and returns (results, columns, error_message)"""
cursor = self.conn.cursor()
try:
# SQLite doesn't have a strict read-only mode, but we can simulate it
# or rely on AST validation to prevent mutations.
cursor.execute(sql)
results = cursor.fetchall()
columns = [desc[0] for desc in cursor.description] if cursor.description else []
return results, columns, ""
except sqlite3.Error as e:
return [], [], str(e)
class ASTValidator:
@staticmethod
def is_safe_query(sql: str) -> bool:
"""Simple AST-like validation to ensure query is read-only and safe"""
sql_upper = sql.upper().strip()
# Block multiple statements
if ";" in sql_upper and not sql_upper.endswith(";"):
return False
# Block write operations
forbidden_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE", "REPLACE", "TRUNCATE"]
for keyword in forbidden_keywords:
if f" {keyword} " in f" {sql_upper} " or sql_upper.startswith(keyword):
return False
# Ensure it starts with SELECT
return sql_upper.startswith("SELECT")
class MockLLM:
"""Simulates LLM behavior, including an initial error to demonstrate self-correction"""
def __init__(self):
self.call_count = 0
def generate_sql(self, prompt: str) -> str:
self.call_count += 1
# Simulate an initial syntax error on the first call to show the self-correction loop
if self.call_count == 1:
logger.info("Mock LLM generating initial query with a deliberate column name error...")
return "SELECT name, SUM(amount) FROM departments JOIN sales ON departments.id = sales.dept_id GROUP BY name;"
else:
logger.info("Mock LLM generating corrected query...")
return "SELECT name, SUM(amount) FROM departments JOIN sales ON departments.id = sales.department_id GROUP BY name;"
class NLtoSQLAgent:
def __init__(self, sandbox: DatabaseSandbox, validator: ASTValidator, llm: MockLLM):
self.sandbox = sandbox
self.validator = validator
self.llm = llm
self.max_retries = 3
def run(self, user_query: str) -> Dict[str, Any]:
schema = self.sandbox.get_schema()
current_prompt = f"User Query: {user_query}
Database Schema:
{schema}
Generate a valid SQLite query to answer the user query."
for attempt in range(1, self.max_retries + 1):
logger.info(f"Attempt {attempt} of {self.max_retries}")
# Generate SQL
sql = self.llm.generate_sql(current_prompt)
logger.info(f"Generated SQL: {sql}")
# Validate SQL
if not self.validator.is_safe_query(sql):
error_msg = "Security Validation Failed: Non-SELECT or unsafe query detected."
logger.warning(error_msg)
current_prompt += f"
Failed SQL: {sql}
Error: {error_msg}. Please generate a strictly read-only SELECT query."
continue
# Execute SQL
results, columns, db_error = self.sandbox.execute_query(sql)
if db_error:
logger.warning(f"Database Execution Error: {db_error}")
current_prompt += f"
Failed SQL: {sql}
Database Error: {db_error}. Please correct the SQL query and try again."
continue
# Success
logger.info("Query executed successfully!")
return {
"status": "success",
"sql": sql,
"columns": columns,
"results": results
}
return {
"status": "failed",
"error": "Maximum retries reached without successful execution."
}
if __name__ == "__main__":
sandbox = DatabaseSandbox()
validator = ASTValidator()
llm = MockLLM()
agent = NLtoSQLAgent(sandbox, validator, llm)
user_request = "What is the total sales for each department?"
logger.info(f"Starting agent run for request: '{user_request}'")
result = agent.run(user_request)
print("
Final Result:", result)INFO - Starting agent run for request: 'What is the total sales for each department?'
INFO - Attempt 1 of 3
INFO - Mock LLM generating initial query with a deliberate column name error...
INFO - Generated SQL: SELECT name, SUM(amount) FROM departments JOIN sales ON departments.id = sales.dept_id GROUP BY name;
WARNING - Database Execution Error: no such column: sales.dept_id
INFO - Attempt 2 of 3
INFO - Mock LLM generating corrected query...
INFO - Generated SQL: SELECT name, SUM(amount) FROM departments JOIN sales ON departments.id = sales.department_id GROUP BY name;
INFO - Query executed successfully!
Final Result: {'status': 'success', 'sql': 'SELECT name, SUM(amount) FROM departments JOIN sales ON departments.id = sales.department_id GROUP BY name;', 'columns': ['name', 'SUM(amount)'], 'results': [('Engineering', 4000.0), ('Sales', 5000.0)]}