Text-to-SQL Agents: Converting Natural Language Questions to Database Queries

The Core Challenge of Text-to-SQL

Converting "Show me all customers who placed more than 3 orders last month" into valid SQL sounds simple, but it requires understanding the schema (which table has orders? what is the date column called?), resolving ambiguity ("last month" relative to when?), and generating syntactically correct SQL for the target database dialect. Text-to-SQL agents solve this by combining schema awareness, LLM reasoning, and an iterative error correction loop.

The difference between a naive text-to-SQL prompt and an agent approach is the feedback loop. A prompt generates one query and hopes it works. An agent generates a query, attempts execution, reads any error, and refines until it succeeds.

Schema Understanding: The Foundation

The agent must know your database intimately. Simply listing table and column names is not enough — it needs types, relationships, and sample data:

flowchart LR
    INPUT(["User intent"])
    PARSE["Parse plus<br/>classify"]
    PLAN["Plan and tool<br/>selection"]
    AGENT["Agent loop<br/>LLM plus tools"]
    GUARD{"Guardrails<br/>and policy"}
    EXEC["Execute and<br/>verify result"]
    OBS[("Trace and metrics")]
    OUT(["Outcome plus<br/>next action"])
    INPUT --> PARSE --> PLAN --> AGENT --> GUARD
    GUARD -->|Pass| EXEC --> OUT
    GUARD -->|Fail| AGENT
    AGENT --> OBS
    style AGENT fill:#4f46e5,stroke:#4338ca,color:#fff
    style GUARD fill:#f59e0b,stroke:#d97706,color:#1f2937
    style OBS fill:#ede9fe,stroke:#7c3aed,color:#1e1b4b
    style OUT fill:#059669,stroke:#047857,color:#fff

import sqlite3
from agents import Agent, Runner, function_tool

DB_PATH = "ecommerce.db"

@function_tool
def get_schema_details() -> str:
    """Return comprehensive schema information including tables, columns,
    types, foreign keys, and sample values."""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()

    cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
    tables = [r[0] for r in cursor.fetchall()]

    parts = []
    for table in tables:
        cursor.execute(f"PRAGMA table_info({table})")
        columns = cursor.fetchall()
        col_defs = []
        for col in columns:
            pk = " PRIMARY KEY" if col[5] else ""
            nullable = "" if col[3] else " NOT NULL"
            col_defs.append(f"    {col[1]} {col[2]}{pk}{nullable}")

        cursor.execute(f"PRAGMA foreign_key_list({table})")
        fks = cursor.fetchall()
        fk_defs = [f"    FK: {fk[3]} -> {fk[2]}.{fk[4]}" for fk in fks]

        cursor.execute(f"SELECT * FROM {table} LIMIT 3")
        sample_rows = cursor.fetchall()
        header = [col[1] for col in columns]
        sample_str = "\n".join(
            "    " + " | ".join(str(v) for v in row) for row in sample_rows
        )

        section = f"Table: {table}\n"
        section += "  Columns:\n" + "\n".join(col_defs)
        if fk_defs:
            section += "\n  Foreign Keys:\n" + "\n".join(fk_defs)
        section += f"\n  Sample ({' | '.join(header)}):\n{sample_str}"
        parts.append(section)

    conn.close()
    return "\n\n".join(parts)

Including sample rows is critical. When the agent sees that the status column contains values like "shipped", "pending", and "cancelled", it avoids generating a query that filters on "Shipped" with a capital S.

Query Validation Before Execution

Before running a query against the database, validate its structure:

import re

@function_tool
def validate_sql(query: str) -> str:
    """Validate a SQL query for safety and basic correctness."""
    normalized = query.strip().upper()

    # Safety: only allow SELECT
    if not normalized.startswith("SELECT"):
        return "REJECTED: Only SELECT statements are allowed."

    # Check for dangerous patterns
    dangerous = ["DROP", "DELETE", "UPDATE", "INSERT", "ALTER", "TRUNCATE", "GRANT"]
    for keyword in dangerous:
        pattern = rf"\b{keyword}\b"
        if re.search(pattern, normalized):
            return f"REJECTED: Query contains forbidden keyword '{keyword}'."

    # Basic syntax checks
    if normalized.count("(") != normalized.count(")"):
        return "WARNING: Mismatched parentheses detected."

    if "SELECT" in normalized and "FROM" not in normalized:
        if "SELECT 1" not in normalized and "SELECT COUNT" not in normalized:
            return "WARNING: SELECT without FROM clause."

    return "VALID: Query passed all checks."

SQL Executor with Error Context

When a query fails, the error message needs to be informative enough for the LLM to fix the issue:

@function_tool
def execute_query(query: str, max_rows: int = 25) -> str:
    """Execute a validated SQL query and return results as a formatted table."""
    conn = sqlite3.connect(DB_PATH)
    conn.row_factory = sqlite3.Row
    cursor = conn.cursor()

    try:
        cursor.execute(query)
        rows = cursor.fetchmany(max_rows)
        total = cursor.fetchall()
        total_count = len(rows) + len(total)

        if not rows:
            return "Query executed successfully but returned 0 rows."

        headers = list(rows[0].keys())
        lines = [" | ".join(headers)]
        lines.append("-" * len(lines[0]))
        for row in rows:
            lines.append(" | ".join(str(row[h]) for h in headers))

        result = "\n".join(lines)
        if total_count > max_rows:
            result += f"\n... showing {max_rows} of {total_count} total rows"
        return result

    except sqlite3.OperationalError as e:
        error_msg = str(e)
        hint = ""
        if "no such column" in error_msg:
            col_match = re.search(r"no such column: (\w+)", error_msg)
            if col_match:
                hint = f"\nHint: Column '{col_match.group(1)}' was not found. Call get_schema_details to check available columns."
        elif "no such table" in error_msg:
            hint = "\nHint: Table not found. Call get_schema_details to see available tables."
        elif "ambiguous column" in error_msg:
            hint = "\nHint: Use table_name.column_name to disambiguate."
        return f"SQL Error: {error_msg}{hint}"

    finally:
        conn.close()

The enriched error messages with hints guide the LLM toward the correct fix rather than leaving it to guess.

The Agent with Iterative Refinement

text_to_sql_agent = Agent(
    name="Text-to-SQL Agent",
    instructions="""You convert natural language questions into SQL queries. Follow this process:

1. ALWAYS call get_schema_details first to understand the database structure.
2. Write a SQL query that answers the user's question.
3. Call validate_sql to check the query before running it.
4. If validation passes, call execute_query.
5. If execution returns an error, read the error message carefully, fix the
   query, and try again. You may attempt up to 3 retries.
6. Present the results in a clear format and explain what the data shows.

Rules:
- Use explicit column names, never SELECT *.
- Always qualify column names with table names in JOINs.
- Use ISO date format for date comparisons.
- Add ORDER BY for meaningful result ordering.
- Use LIMIT to keep results manageable.""",
    tools=[get_schema_details, validate_sql, execute_query],
)

Handling Ambiguous Questions

Natural language is inherently ambiguous. "What are our best products?" could mean highest revenue, most units sold, best ratings, or highest margin. The agent should ask for clarification or state its interpretation:

result = Runner.run_sync(
    text_to_sql_agent,
    "What are our best-selling products this quarter?",
)

A well-instructed agent will respond with: "I interpreted 'best-selling' as highest total units sold. Here are the top 10 products by unit volume for Q1 2026..." and include the SQL it generated for transparency.

Multi-Dialect Support

Different databases use different SQL dialects. Add dialect awareness to the agent:

dialect_instructions = {
    "sqlite": "Use SQLite syntax. Date functions: date(), strftime(). String concat: ||",
    "postgresql": "Use PostgreSQL syntax. Date functions: date_trunc(), EXTRACT(). String concat: || or CONCAT()",
    "mysql": "Use MySQL syntax. Date functions: DATE_FORMAT(), YEAR(). String concat: CONCAT(). Use backticks for identifiers.",
}

def create_sql_agent(dialect: str = "sqlite"):
    base_instructions = text_to_sql_agent.instructions
    dialect_note = dialect_instructions.get(dialect, "")
    return Agent(
        name="Text-to-SQL Agent",
        instructions=f"{base_instructions}\n\nSQL Dialect: {dialect}. {dialect_note}",
        tools=[get_schema_details, validate_sql, execute_query],
    )

FAQ

How accurate are LLM-generated SQL queries?

On standard benchmarks like Spider, modern LLMs achieve 70-85% accuracy on the first attempt. With the iterative refinement loop (schema inspection, validation, error-driven retry), practical accuracy rises above 90% for common query patterns.

How do I handle very large schemas with hundreds of tables?

Inject only the relevant subset of the schema into the agent context. Build a schema search tool that takes the user's question, identifies relevant tables using keyword matching or embeddings, and returns only those table definitions.

Can this approach work with NoSQL databases like MongoDB?

Yes, but you replace SQL generation with MongoDB query syntax. The same agent pattern applies: inspect the collection schema, generate a query (find(), aggregate()), execute, and iterate on errors. The LLM needs examples of MongoDB query syntax in its instructions.

---

#TexttoSQL #NLP #Database #QueryGeneration #AIAgents #AgenticAI #LearnAI #AIEngineering