Building a Production-Ready Conversational Text-to-SQL (NL2SQL) Chatbot System

High-Level System Architecture

A production-ready Conversational Text-to-SQL pipeline consists of seven sequential stages:

Pipeline Stage	Primary Function
1. Session & Memory Tracking	Manages conversation turn history, active schema filters, and session state across user interactions.
2. Contextualization & Entity Resolution	Rewrites ambiguous follow-up questions into standalone queries using chat history and resolves user entity values to exact database literals.
3. Schema Pruning (Graph RAG)	Dynamically retrieves relevant tables, columns, and foreign key junction paths required for the query context.
4. LLM SQL Generation	Prompts the LLM with user context, chat history, schema metadata, and dialect rules to output candidate SQL.
5. AST Parsing & Security Guardrails	Validates query syntax, enforces single-statement read-only rules using AST parsing, and injects row limits.
6. Engine Execution & Agentic Repair	Executes sanitized queries against a read-only database replica; feeds execution errors back to the LLM for automated repair.
7. Answer Synthesis & State Update	Translates raw JSON query outputs into human-readable markdown summaries and appends the turn to dialogue history.

Architectural Data Flow

1. User Prompt & Session State -> Conversational Memory & Query Contextualizer -> Standalone Intent Query
2. Standalone Intent Query -> Schema Graph RAG & Value Resolution -> Injected Context Prompt
3. Injected Context Prompt -> LLM SQL Generator -> Candidate SQL String
4. Candidate SQL String -> AST Parser & Security Sandbox -> Sanitized Dialect SQL
5. Sanitized Dialect SQL -> Read-Only DB Engine -> Query Results / Error Backtrace
6. Query Results / Backtrace -> Self-Correction Loop / Chat Synthesis Engine -> Synthesized Response & Updated Memory

---

Step 1: Schema Representation, Graph RAG, and Entity Resolution

Schema Metadata Definition

To assist LLMs in generating valid SQL, schema descriptions must explicitly include column data types, foreign key relationships, primary keys, and allowable ENUM values.

-- Production Schema Definition (SQLite / PostgreSQL Dialect)
CREATE TABLE categories (
    category_id INTEGER PRIMARY KEY,
    category_name TEXT NOT NULL UNIQUE
);

CREATE TABLE products (
    product_id INTEGER PRIMARY KEY,
    name TEXT NOT NULL,
    category_id INTEGER NOT NULL REFERENCES categories(category_id),
    price REAL NOT NULL,
    stock_quantity INTEGER NOT NULL,
    status TEXT CHECK(status IN ('active', 'discontinued', 'out_of_stock'))
);

CREATE TABLE sales_orders (
    order_id INTEGER PRIMARY KEY,
    product_id INTEGER NOT NULL REFERENCES products(product_id),
    quantity_sold INTEGER NOT NULL,
    total_price REAL NOT NULL,
    order_date DATE NOT NULL
);

Graph RAG for Large Schemas

When managing large enterprise databases, naive vector retrieval often selects endpoint tables while omitting critical bridging (junction) tables needed for SQL JOIN clauses.

Graph RAG models the database schema as a topological graph $G = (V, E)$, where vertices $V$ represent tables and edges $E$ represent foreign key relationships. The retrieval engine expands the candidate set of tables $V_{\text{init}}$ obtained via semantic vector search to include all minimal joining paths $V_{\text{bridge}}$ before injecting schema context into the prompt context window.

Categorical Entity Resolution

When user queries contain informal terms (e.g., "desk items" or "out of stock"), exact value matching fails if the database expects specific literals (e.g., 'out_of_stock'). An entity resolution layer resolves natural language phrases against pre-indexed categorical values using trigram string matching before passing context to the LLM.

---

Step 2: Conversational Memory & Follow-Up Contextualization

In a multi-turn chatbot, users frequently refer back to previous results or ask incremental follow-up questions (e.g., "Show me furniture items", followed by "Which ones cost over $300?").

Directly generating SQL from an isolated follow-up prompt fails because the model lacks the context of prior filters. The system uses a Query Contextualizer step to resolve coreferences and rewrite follow-up questions into complete, self-contained questions before SQL generation.

Contextualization Prompt Example

Given the conversation history and a follow-up user question, rewrite the follow-up question into a standalone, self-contained question that explicitly includes all necessary context, filters, and entities from prior turns.

CONVERSATION HISTORY:
User: Which furniture items are currently active?
Assistant: Here are 5 active furniture items in stock...

FOLLOW-UP QUESTION:
Which ones cost less than 400 dollars?

REWRITTEN STANDALONE QUESTION:
Which active furniture items cost less than 400 dollars?

---

Step 3: System Prompt Engineering for SQL Generation

A structured system prompt enforces dialect constraints, restricts SQL mutation statements, incorporates conversational context, and ensures strict output formatting.

Production System Prompt

You are an expert database engineer specializing in SQLite queries.
Your task is to convert natural language questions into single, valid, optimized SQLite SELECT queries.

DATABASE SCHEMA:
{db_schema}

CATEGORICAL VALUE RESOLUTION:
{resolved_entities}

CONVERSATION HISTORY:
{chat_history}

RULES & CONSTRAINTS:
1. Return ONLY a single valid SQL query enclosed within a markdown code block: ```sql ... ```.
2. Do NOT include any commentary or explanations outside the code block.
3. Executing INSERT, UPDATE, DELETE, DROP, ALTER, CREATE, or TRUNCATE statements is strictly prohibited.
4. Join tables strictly on specified foreign key relationships.
5. Use proper SQLite functions (e.g., COALESCE, LOWER, STRFTIME).
6. Respect filters and constraints established in prior conversation turns unless the user explicitly resets them.

USER QUERY:
{user_query}

---

Step 4: Multi-Layer SQL Validation & AST Guardrails

Executing LLM-generated code directly against database connections poses severe security risks. Production pipelines enforce security across two layers:

1. Infrastructure Security: Read-only database user credentials, query execution timeouts, and execution against read replicas.
2. Abstract Syntax Tree (AST) Validation: Programmatic parsing via sqlglot to verify read-only behavior, detect stacked query injection, and enforce limits.

Security Validation Logic

import re
import sqlglot
from sqlglot import exp, errors

def extract_sql_query(raw_llm_output: str) -> str:
    """Extracts clean SQL code string from raw LLM text response using regex."""
    pattern = r"```(?:sql)?\s*(.*?)\s*```"
    match = re.search(pattern, raw_llm_output, re.DOTALL | re.IGNORECASE)
    if match:
        return match.group(1).strip()
    cleaned = raw_llm_output.strip()
    if cleaned.upper().startswith(("SELECT", "WITH")):
        return cleaned.rstrip(";")
    raise ValueError("No valid SQL code block identified in model output.")

def validate_and_sanitize_sql(sql_string: str, dialect: str = "sqlite", max_limit: int = 100) -> str:
    """
    Validates SQL using sqlglot AST parsing.
    Prevents multi-statement execution, non-SELECT mutations, unparseable command fallbacks,
    and injects dynamic row limits safely.
    """
    try:
        # Parse entire string stream to detect stacked multi-statement execution attempts
        statements = sqlglot.parse(sql_string, read=dialect)
    except errors.ParseError as e:
        raise ValueError(f"Invalid SQL syntax: {e}")

    if not statements:
        raise ValueError("Security Violation: Empty query statement received.")
    if len(statements) > 1:
        raise ValueError("Security Violation: Stacked multi-statement queries are prohibited.")

    ast = statements[0]
    if ast is None:
        raise ValueError("Security Violation: Unable to construct AST node.")

    # Reject unparseable fallback command expressions
    if ast.find(exp.Command):
        raise ValueError("Security Violation: Unrecognized dialect command detected.")

    # Validate read-only root statement type (supports Select, Union, and CTE With queries)
    if not isinstance(ast, (exp.Select, exp.Union)):
        if isinstance(ast, exp.Expression) and ast.find(exp.Select):
            pass
        else:
            raise ValueError(f"Security Violation: Prohibited statement root type ({type(ast).__name__}).")

    # Audit for write/mutation operations anywhere in the AST
    forbidden_nodes = (
        exp.Insert, exp.Update, exp.Delete, exp.Drop, 
        exp.Create, exp.AlterTable, exp.TruncateTable
    )
    for forbidden in ast.find_all(*forbidden_nodes):
        raise ValueError(f"Security Violation: Prohibited mutation node detected ({type(forbidden).__name__}).")

    # Enforce MAX LIMIT clause using dialect-native AST modifier
    limit_node = ast.args.get("limit")
    if not limit_node:
        ast = ast.limit(max_limit)
    else:
        try:
            current_limit = int(limit_node.this.this)
            if current_limit > max_limit:
                ast = ast.limit(max_limit)
        except (AttributeError, ValueError):
            ast = ast.limit(max_limit)

    return ast.sql(dialect=dialect)

---

Step 5: Answer Synthesis with Dialogue History

Once sanitized SQL executes, raw JSON result payloads are passed back to the LLM alongside the chat history to construct a natural language conversational response.

Conversational Synthesis Prompt

You are a helpful, factual data analyst chatbot assistant. Synthesize the database query results to directly answer the user's natural language question in the context of the ongoing conversation.

Conversation History:
{chat_history}

User Question: {user_query}
Executed SQL: {executed_sql}
Database Results (JSON): {db_results_json}

GUIDELINES:
1. Provide a direct, conversational response in 1-3 sentences.
2. Format data lists or tables using Markdown where appropriate.
3. Format currency and numeric quantities cleanly ($X,XXX.XX).
4. Maintain context from previous turns (e.g., refer back to previously discussed items smoothly).
5. If database results are empty ([]), state clearly that no records matched the query criteria.

---

Step 6: Complete End-to-End Enterprise Chatbot Implementation

This stateful implementation features a multi-turn chat memory manager, follow-up query contextualizer, dialect-aligned SQL generation, AST security validation, agentic self-correction, database execution, and conversational response synthesis.

import os
import re
import json
import sqlite3
from typing import Dict, Any, List, Tuple
from openai import OpenAI
import sqlglot
from sqlglot import exp, errors

client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

# ------------------------------------------------------------------------------
# 1. Database Setup
# ------------------------------------------------------------------------------

def setup_demo_database() -> sqlite3.Connection:
    conn = sqlite3.connect(":memory:")
    cursor = conn.cursor()
    
    cursor.execute("""
        CREATE TABLE categories (
            category_id INTEGER PRIMARY KEY,
            category_name TEXT NOT NULL UNIQUE
        );
    """)
    cursor.execute("""
        CREATE TABLE products (
            product_id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            category_id INTEGER NOT NULL REFERENCES categories(category_id),
            price REAL NOT NULL,
            stock_quantity INTEGER NOT NULL,
            status TEXT CHECK(status IN ('active', 'discontinued', 'out_of_stock'))
        );
    """)
    cursor.execute("""
        CREATE TABLE sales_orders (
            order_id INTEGER PRIMARY KEY,
            product_id INTEGER NOT NULL REFERENCES products(product_id),
            quantity_sold INTEGER NOT NULL,
            total_price REAL NOT NULL,
            order_date DATE NOT NULL
        );
    """)
    
    cursor.executemany("INSERT INTO categories VALUES (?, ?);", [
        (1, 'Electronics'),
        (2, 'Furniture')
    ])
    cursor.executemany("INSERT INTO products VALUES (?, ?, ?, ?, ?, ?);", [
        (1, 'Laptop Pro 16', 1, 1299.99, 45, 'active'),
        (2, 'Wireless Ergonomic Mouse', 1, 25.50, 200, 'active'),
        (3, 'Executive Standing Desk', 2, 450.00, 12, 'active'),
        (4, 'Ergonomic Mesh Chair', 2, 299.00, 0, 'out_of_stock')
    ])
    cursor.executemany("INSERT INTO sales_orders VALUES (?, ?, ?, ?, ?);", [
        (101, 1, 2, 2599.98, '2026-01-15'),
        (102, 2, 5, 127.50, '2026-02-01'),
        (103, 3, 1, 450.00, '2026-02-10')
    ])
    conn.commit()
    return conn

# ------------------------------------------------------------------------------
# 2. Conversational State & Contextualization Engine
# ------------------------------------------------------------------------------

class ConversationMemory:
    """Manages chat context history, executed SQL queries, and session state."""
    def __init__(self, max_turns: int = 5):
        self.turns: List[Dict[str, str]] = []
        self.max_turns = max_turns

    def add_turn(self, user_prompt: str, generated_sql: str, assistant_response: str):
        self.turns.append({
            "user": user_prompt,
            "sql": generated_sql,
            "assistant": assistant_response
        })
        if len(self.turns) > self.max_turns:
            self.turns.pop(0)

    def format_history_for_prompt(self) -> str:
        if not self.turns:
            return "No previous conversation history."
        formatted = []
        for i, turn in enumerate(self.turns, 1):
            formatted.append(f"Turn {i}:\nUser: {turn['user']}\nExecuted SQL: {turn['sql']}\nAssistant: {turn['assistant']}")
        return "\n\n".join(formatted)

def contextualize_user_query(user_query: str, memory: ConversationMemory) -> str:
    """Rewrites follow-up user queries into self-contained standalone questions."""
    if not memory.turns:
        return user_query

    prompt = f"""Given the following conversation history and a follow-up question, rewrite the question to be a fully self-contained question that explicitly preserves entities, table filters, and constraints mentioned previously.

CONVERSATION HISTORY:
{memory.format_history_for_prompt()}

FOLLOW-UP QUESTION: {user_query}

REWRITTEN STANDALONE QUESTION:"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0
    )
    return response.choices[0].message.content.strip()

# ------------------------------------------------------------------------------
# 3. SQL Generation with Agentic Error Self-Correction Loop
# ------------------------------------------------------------------------------

def generate_sql_with_reflection(
    user_query: str, 
    schema: str, 
    memory: ConversationMemory,
    dialect: str = "sqlite", 
    max_retries: int = 2
) -> str:
    system_prompt = f"""You are an expert database engineer specializing in {dialect}.
Convert the natural language user query into a valid, read-only {dialect} query based on the schema and conversation context below.

DATABASE SCHEMA:
{schema}

CONVERSATION HISTORY:
{memory.format_history_for_prompt()}

RULES:
- Return ONLY the SQL query inside a ```sql ``` code block.
- Only SELECT queries are permitted.
- Match joined tables strictly on foreign key paths.
"""
    messages = [
        {"role": "system", "content": system_prompt},
        {"role": "user", "content": user_query}
    ]

    for attempt in range(max_retries + 1):
        response = client.chat.completions.create(
            model="gpt-4o-mini",
            messages=messages,
            temperature=0.0
        )
        raw_content = response.choices[0].message.content
        
        try:
            sql_candidate = extract_sql_query(raw_content)
            validated_sql = validate_and_sanitize_sql(sql_candidate, dialect=dialect)
            return validated_sql
        except ValueError as err:
            if attempt == max_retries:
                raise RuntimeError(f"Pipeline failed to generate valid SQL after {max_retries} attempts: {err}")
            
            # Reflection step: Feed validation error back into model context
            messages.append({"role": "assistant", "content": raw_content})
            messages.append({
                "role": "user", 
                "content": f"The generated SQL failed validation with error: '{err}'. Please correct the query."
            })

# ------------------------------------------------------------------------------
# 4. Answer Synthesis & Chatbot Orchestration
# ------------------------------------------------------------------------------

def synthesize_chatbot_response(
    user_query: str, 
    sql_query: str, 
    results: List[Dict[str, Any]], 
    memory: ConversationMemory
) -> str:
    synthesis_prompt = f"""Synthesize database query results into a concise, natural conversational chatbot response.

Conversation Context:
{memory.format_history_for_prompt()}

Current User Question: {user_query}
Executed SQL: {sql_query}
Database Results: {json.dumps(results, default=str)}

GUIDELINES:
1. Answer the question directly in 1-3 conversational sentences.
2. Use markdown formatting for monetary values and lists.
3. Reference prior conversation topics seamlessly when relevant.
"""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are an intelligent data analyst chatbot."},
            {"role": "user", "content": synthesis_prompt}
        ],
        temperature=0.2
    )
    return response.choices[0].message.content

class TextToSQLChatbot:
    """Enterprise Chatbot wrapper encapsulating state, validation, execution, and synthesis."""
    def __init__(self, db_conn: sqlite3.Connection, schema_definition: str):
        self.conn = db_conn
        self.schema = schema_definition
        self.memory = ConversationMemory(max_turns=5)

    def chat(self, raw_user_prompt: str) -> str:
        # Step A: Contextualize user prompt using conversation history
        standalone_query = contextualize_user_query(raw_user_prompt, self.memory)
        
        # Step B: Generate & Validate SQL with Reflection Loop
        validated_sql = generate_sql_with_reflection(
            user_query=standalone_query, 
            schema=self.schema, 
            memory=self.memory,
            dialect="sqlite"
        )
        
        # Step C: Execute SQL
        cursor = self.conn.cursor()
        cursor.execute(validated_sql)
        columns = [desc[0] for desc in cursor.description] if cursor.description else []
        rows = cursor.fetchall()
        results = [dict(zip(columns, row)) for row in rows]
        
        # Step D: Synthesize Chatbot Response
        bot_response = synthesize_chatbot_response(raw_user_prompt, validated_sql, results, self.memory)
        
        # Step E: Commit turn to memory
        self.memory.add_turn(raw_user_prompt, validated_sql, bot_response)
        
        return bot_response

# ------------------------------------------------------------------------------
# 5. Interactive Pipeline Demo
# ------------------------------------------------------------------------------

if __name__ == "__main__":
    schema_def = """
    TABLE categories (
        category_id INTEGER PRIMARY KEY,
        category_name TEXT NOT NULL
    );

    TABLE products (
        product_id INTEGER PRIMARY KEY,
        name TEXT NOT NULL,
        category_id INTEGER NOT NULL REFERENCES categories(category_id),
        price REAL NOT NULL,
        stock_quantity INTEGER NOT NULL,
        status TEXT -- Values: 'active', 'discontinued', 'out_of_stock'
    );

    TABLE sales_orders (
        order_id INTEGER PRIMARY KEY,
        product_id INTEGER NOT NULL REFERENCES products(product_id),
        quantity_sold INTEGER NOT NULL,
        total_price REAL NOT NULL,
        order_date DATE NOT NULL
    );
    """
    
    db_connection = setup_demo_database()
    chatbot = TextToSQLChatbot(db_connection, schema_def)
    
    # Simulate a multi-turn conversation
    turns = [
        "Which furniture items are currently available?",
        "Which ones cost less than 400 dollars?",
        "How many sales orders have been placed for electronics?"
    ]
    
    print("--- Starting Conversational Text-to-SQL Chatbot Session ---\n")
    for user_turn in turns:
        print(f"User: {user_turn}")
        response = chatbot.chat(user_turn)
        print(f"Chatbot: {response}\n" + "-"*50 + "\n")

---

Enterprise Best Practices & Risk Mitigation Matrix

Operational Risk	System Impact	Production Mitigation Strategy
Stacked Query Injection	Unauthorized data mutation or schema deletion	Parse full statement streams using sqlglot.parse() and enforce len(statements) == 1.
Ambiguous Coreferences	Incorrect filters applied to follow-up prompts	Implement a Query Contextualizer step to rewrite follow-up questions before SQL parsing.
Dialect Mismatch	Engine runtime syntax errors	Standardize prompts, AST parsers, and execution engines on a single target SQL dialect.
Subquery / CTE Rejection	False positive validation failures	Expand AST root inspection to support exp.Union and wrapped CTEs (exp.With).
Categorical Value Mismatches	Empty execution results ([])	Deploy categorical value resolution prior to prompt construction.
Conversation Memory Overflow	High API latency and context window exhaustion	Maintain a sliding window of past $N$ dialogue turns (max_turns=5).
Execution Errors	Chatbot crashing mid-dialogue	Implement an agentic reflection loop that feeds DB error traces back to the LLM.

---

Summary Checklist

Component	Target Recommendation
Memory Tracking	Use a sliding window (ConversationMemory) to retain recent dialogue turns, SQL queries, and answers.
Contextualization	Pre-process user prompts with a Query Contextualizer model to resolve pronouns and implicit filters.
Schema Management	Use Graph RAG to retain foreign key topologies; pre-resolve ENUM and categorical literals.
Prompt Engineering	Enforce temperature=0.0, pass conversation history, column types, foreign keys, and ENUM values.
AST Validation	Enforce single-statement sqlglot.parse(), walk trees for mutation nodes, reject exp.Command.
Database Security	Use read-only database connections, read replicas, and strict statement timeouts.
Resilience	Integrate an agentic reflection/retry loop for syntax and execution error repair.
Synthesis	Convert query outputs into concise, conversational markdown responses with clear state tracking.