Every developer has been there. Your product manager walks over with a "quick question" about user metrics. Your sales team needs data for a presentation in 30 minutes. Your support team wants to check order statuses without bugging engineering. Each time, someone needs to write SQL, understand the schema, and format the results.
What if users could just ask "How many users signed up last week?" and get an answer immediately?
This is exactly what natural language to SQL chatbots solve. They turn plain English questions into executable SQL queries, democratizing data access across your organization. In this tutorial, we'll build one from scratch.
What Is an NL-to-SQL Chatbot?
An NL-to-SQL chatbot is an application that accepts questions in natural language and converts them into SQL queries. Instead of requiring users to understand database schemas, JOIN syntax, or aggregation functions, they can ask questions conversationally.
Here's a simple example:
User input: "Show me all orders from yesterday"
Generated SQL:
SELECT * FROM orders
WHERE DATE(created_at) = CURRENT_DATE - INTERVAL '1 day';
The chatbot interprets the user's intent, maps it to the appropriate tables and columns, constructs a valid SQL query, executes it, and returns formatted results.
High-Level Architecture
Before diving into code, let's understand the components:
User Input Processing: Receive and normalize the natural language question
Schema Context: Provide the model with information about available tables, columns, and relationships
SQL Generation: Use an LLM or fine-tuned model to generate SQL from the question and schema
Validation Layer: Check the generated SQL for safety and correctness
Execution Engine: Run the query against the database with appropriate permissions
Response Formatting: Convert query results into human-readable responses
The key insight is that the model needs context about your database structure to generate accurate queries.
Tech Stack
For this tutorial, we'll use:
Python 3.9+: Backend language with excellent data handling libraries
LangChain: Framework for building LLM applications with SQL capabilities
SQLAlchemy: ORM for database interaction and query validation
PostgreSQL: Database (but MySQL, SQLite work similarly)
OpenAI API: For the language model (you can substitute with other providers)
FastAPI: Simple API server for the chatbot endpoint
This stack is production-ready and relatively easy to understand. You can swap components based on your requirements.
Step-by-Step Implementation
Project Setup
First, create a new Python project and install dependencies:
mkdir nl-to-sql-chatbot
cd nl-to-sql-chatbot
python -m venv venv
source venv/bin/activate
pip install langchain langchain-openai sqlalchemy psycopg2-binary fastapi uvicorn python-dotenv
Create a .env file for configuration:
DATABASE_URL=postgresql://username:password@localhost:5432/your_database
OPENAI_API_KEY=your_api_key_here
Never commit this file. Add it to .gitignore immediately.
Database Schema Awareness
The chatbot needs to understand your database structure. Let's create a schema inspector:
CODE FILE: schema_inspector.py
from sqlalchemy import create_engine, inspect
import os
from dotenv import load_dotenv
load_dotenv()
class SchemaInspector:
def init(self, database_url):
self.engine = create_engine(database_url)
self.inspector = inspect(self.engine)
def get_schema_info(self):
schema_description = []
for table_name in self.inspector.get_table_names():
columns = self.inspector.get_columns(table_name)
column_info = []
for col in columns:
col_desc = f"{col['name']} ({col['type']})"
column_info.append(col_desc)
table_desc = f"Table: {table_name}\nColumns: {', '.join(column_info)}"
schema_description.append(table_desc)
return "\n\n".join(schema_description)
def get_sample_rows(self, table_name, limit=3):
query = f"SELECT * FROM {table_name} LIMIT {limit}"
with self.engine.connect() as conn:
result = conn.execute(query)
return result.fetchall()
This class introspects your database and creates a text representation that we'll feed to the LLM.
SQL Query Generation
Now let's build the core query generation logic:
CODE FILE: query_generator.py
from langchain_openai import ChatOpenAI
from langchain.prompts import PromptTemplate
from langchain.chains import LLMChain
import os
class SQLQueryGenerator:
def init(self, schema_info):
self.llm = ChatOpenAI(
temperature=0,
model="gpt-4",
api_key=os.getenv("OPENAI_API_KEY")
)
self.schema_info = schema_info
self.prompt_template = PromptTemplate(
input_variables=["schema", "question"],
template="""You are a SQL expert. Given the database schema below, write a SQL query to answer the user's question.
Database Schema:
{schema}
Rules:
Only use tables and columns from the schema above
Write valid PostgreSQL syntax
Use appropriate JOINs when needed
Return only the SQL query, no explanations
Use proper date/time functions
Limit results to 100 rows unless specifically asked for more
User Question: {question}
SQL Query:"""
)
def generate_sql(self, question):
chain = LLMChain(llm=self.llm, prompt=self.prompt_template)
response = chain.run(
schema=self.schema_info,
question=question
)
sql_query = response.strip()
if sql_query.startswith("
sql_query = sql_query.split("```
")[1]
if sql_query.startswith("sql"):
sql_query = sql_query[3:]
sql_query = sql_query.strip()
return sql_query
The prompt engineering here is crucial. We're giving the model clear constraints and the schema context it needs. If you're building chatbots for other use cases beyond SQL generation, understanding how to analyze [customer queries](https://chatboq.com/blogs/analyzing-customer-queries) becomes essential for improving accuracy.
SQL Safety Validation
Before executing any generated SQL, we need validation:
CODE FILE: query_validator.py
import sqlparse
from sqlparse.sql import IdentifierList, Identifier, Where
from sqlparse.tokens import Keyword, DML
class QueryValidator:
DANGEROUS_KEYWORDS = ['DROP', 'DELETE', 'TRUNCATE', 'ALTER', 'CREATE', 'INSERT', 'UPDATE']
@staticmethod
def is_safe(sql_query):
parsed = sqlparse.parse(sql_query)
if not parsed:
return False, "Invalid SQL syntax"
statement = parsed[0]
sql_upper = sql_query.upper()
for keyword in QueryValidator.DANGEROUS_KEYWORDS:
if keyword in sql_upper:
return False, f"Dangerous operation detected: {keyword}"
if statement.get_type() != 'SELECT':
return False, "Only SELECT queries are allowed"
return True, "Query is safe"
@staticmethod
def validate_syntax(sql_query):
try:
sqlparse.parse(sql_query)
return True, "Syntax valid"
except Exception as e:
return False, f"Syntax error: {str(e)}"
@staticmethod
def is_safe(sql_query):
parsed = sqlparse.parse(sql_query)
if not parsed:
return False, "Invalid SQL syntax"
statement = parsed[0]
sql_upper = sql_query.upper()
for keyword in QueryValidator.DANGEROUS_KEYWORDS:
if keyword in sql_upper:
return False, f"Dangerous operation detected: {keyword}"
if statement.get_type() != 'SELECT':
return False, "Only SELECT queries are allowed"
return True, "Query is safe"
@staticmethod
def validate_syntax(sql_query):
try:
sqlparse.parse(sql_query)
return True, "Syntax valid"
except Exception as e:
return False, f"Syntax error: {str(e)}"
This validator ensures we only execute read-only SELECT queries and blocks any modification operations.
Query Execution
Now let's execute validated queries safely:
CODE FILE: query_executor.py
from sqlalchemy import create_engine, text
import pandas as pd
class QueryExecutor:
def init(self, database_url):
self.engine = create_engine(
database_url,
pool_pre_ping=True,
connect_args={"options": "-c default_transaction_read_only=on"}
)
def execute_query(self, sql_query):
try:
with self.engine.connect() as conn:
result = conn.execute(text(sql_query))
df = pd.DataFrame(result.fetchall(), columns=result.keys())
return {
"success": True,
"data": df.to_dict('records'),
"row_count": len(df),
"columns": list(df.columns)
}
except Exception as e:
return {
"success": False,
"error": str(e),
"data": None
}
Note the read-only transaction mode for extra safety.
Response Formatting
Finally, let's format results into natural language:
CODE FILE: response_formatter.py
class ResponseFormatter:
@staticmethod
def format_response(question, sql_query, execution_result):
if not execution_result["success"]:
return {
"answer": f"I encountered an error: {execution_result['error']}",
"sql": sql_query,
"success": False
}
data = execution_result["data"]
row_count = execution_result["row_count"]
if row_count == 0:
answer = "No results found for your query."
elif row_count == 1:
answer = "Here's what I found:\n"
for key, value in data[0].items():
answer += f"- {key}: {value}\n"
else:
answer = f"Found {row_count} results. "
if row_count <= 10:
answer += "Here they are:\n" + str(data)
else:
answer += f"Showing first 10:\n" + str(data[:10])
return {
"answer": answer,
"sql": sql_query,
"data": data,
"row_count": row_count,
"success": True
}
### Putting It All Together
Create the main chatbot class:
CODE FILE: chatbot.py
from schema_inspector import SchemaInspector
from query_generator import SQLQueryGenerator
from query_validator import QueryValidator
from query_executor import QueryExecutor
from response_formatter import ResponseFormatter
import os
class SQLChatbot:
def init(self):
database_url = os.getenv("DATABASE_URL")
inspector = SchemaInspector(database_url)
self.schema_info = inspector.get_schema_info()
self.generator = SQLQueryGenerator(self.schema_info)
self.validator = QueryValidator()
self.executor = QueryExecutor(database_url)
self.formatter = ResponseFormatter()
def ask(self, question):
sql_query = self.generator.generate_sql(question)
is_safe, safety_msg = self.validator.is_safe(sql_query)
if not is_safe:
return {
"success": False,
"error": safety_msg,
"sql": sql_query
}
is_valid, syntax_msg = self.validator.validate_syntax(sql_query)
if not is_valid:
return {
"success": False,
"error": syntax_msg,
"sql": sql_query
}
result = self.executor.execute_query(sql_query)
return self.formatter.format_response(question, sql_query, result)
### Creating an API Endpoint
Wrap this in a FastAPI server:
CODE FILE: main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from chatbot import SQLChatbot
app = FastAPI()
chatbot = SQLChatbot()
class Question(BaseModel):
question: str
@app.post("/ask")
async def ask_question(q: Question):
try:
response = chatbot.ask(q.question)
return response
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
if name == "main":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Run the server:
python main.
## py
Example Chatbot Flow
Let's see it in action with a sample e-commerce database:
User Question: "How many orders were placed last month?"
Generated SQL:
SELECT COUNT(*) as total_orders
FROM orders
WHERE created_at >= DATE_TRUNC('month', CURRENT_DATE - INTERVAL '1 month')
AND created_at < DATE_TRUNC('month', CURRENT_DATE);
Execution Result:
{
"success": true,
"answer": "Here's what I found:\n- total_orders: 1247\n",
"row_count": 1
}
## Handling Errors and Edge Cases
Real-world usage requires robust error handling:
Ambiguous Questions
When a question is unclear, prompt for clarification:
def handle_ambiguous_question(self, question):
ambiguous_keywords = ['it', 'them', 'those', 'that']
if any(keyword in question.lower().split() for keyword in ambiguous_keywords):
return {
"success": False,
"clarification_needed": True,
"message": "Could you be more specific? Which table or data are you asking about?"
}
## Dangerous Queries
We already block DELETE and DROP, but also monitor for resource-intensive queries:
def check_query_complexity(self, sql_query):
join_count = sql_query.upper().count('JOIN')
referenced_tables = self.extract_table_names(parsed)
for table in referenced_tables:
if table not in self.valid_tables:
return False, f"Table '{table}' does not exist"
return True, "Schema references valid"
## Fallback Responses
When generation fails completely:
def get_fallback_response(self, question):
return {
"success": False,
"message": "I couldn't generate a SQL query for that question.",
"suggestions": [
"Try rephrasing your question",
"Be more specific about which data you need",
"Use table and column names if you know them"
]
}
Security Considerations
Security is critical when executing dynamic SQL:
SQL Injection Prevention
Always use parameterized queries and validation:
from sqlalchemy import text
def safe_execute(self, base_query, params):
query = text(base_query)
result = self.engine.execute(query, params)
return result.fetchall()
Read-Only Database Roles
Create a dedicated read-only database user:
CREATE ROLE chatbot_readonly;
GRANT CONNECT ON DATABASE your_db TO chatbot_readonly;
GRANT USAGE ON SCHEMA public TO chatbot_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO chatbot_readonly;
Query Whitelisting
For sensitive environments, maintain an allow-list:
class WhitelistValidator:
ALLOWED_TABLES = {'orders', 'users', 'products', 'order_items'}
ALLOWED_OPERATIONS = {'SELECT', 'COUNT', 'AVG', 'SUM', 'MIN', 'MAX'}
context += "\n\nSample Data:\n"
for table in self.tables:
samples = self.get_sample_rows(table, 2)
context += f"{table}: {samples}\n"
context += "\n\nRelationships:\n"
context += "orders.user_id -> users.id\n"
context += "order_items.order_id -> orders.id\n"
return context
Few-Shot Examples
Include example question-SQL pairs in your prompt template.
Query Correction Loops
If a query fails, try to fix it:
def attempt_correction(self, original_query, error_message):
correction_prompt = f"""
The following SQL query failed:
{original_query}
Error: {error_message}
Please provide a corrected version.
"""
corrected_query = self.generator.generate_sql(correction_prompt)
return corrected_query
Feedback-Based Learning
Store successful queries for future reference.
Real-World Use Cases
This chatbot architecture works well for:
BI Dashboards
Allow business users to ask ad-hoc questions:
"What percentage of users completed onboarding this week?"
"Show me revenue breakdown by product category"
"Which marketing channel has the best conversion rate?"
Internal Admin Panels
Support teams can quickly look up customer data:
"Find all orders for customer email john@example.com"
"Show me failed payments from yesterday"
"List users who haven't logged in for 30 days"
Customer Analytics Tools
Product managers can explore user behavior:
"What features do power users engage with most?"
"Show me the funnel drop-off points"
"Which user segments have the highest retention?"
Non-Technical Team Data Access
Empower teams to answer their own questions:
"How many support tickets were resolved today?"
"What's our current inventory for product SKU ABC123?"
"Show me this quarter's sales vs last quarter"
For eCommerce businesses specifically, implementing AI chatbots for [customer support](https://chatboq.com/blogs/ecommerce-customer-support) can significantly reduce the burden on your team while providing instant data access.
##
Limitations and Trade-of**fs
Be aware of these constraints:**
### Model Hallucinations
LLMs can generate plausible-looking but incorrect SQL. Always validate results and provide ways for users to verify the generated query.
Performance Concerns
Each query requires an LLM API call, adding latency. Consider:
### Caching common queries
Using smaller, faster models for simple questions
Implementing query result caching
### Cost Implications
API calls add up quickly. Monitor usage and implement:
### Rate limiting per user
Query complexity budgets
Caching strategies
### Complex Joins
Multi-table queries with complex relationships may generate incorrect JOINs. Provide clear schema relationship information and consider limiting join depth.
## Building Production-Ready Chatbot Systems
When scaling this solution for production use, you'll need to consider additional factors beyond SQL generation. Professional chatbot [development services](https://chatboq.com/blogs/chatbot-development-services) typically include monitoring, error tracking, user analytics, and continuous improvement workflows that ensure your chatbot remains accurate and helpful over time.
##
Conclusion
You now have a working natural language to SQL chatbot that can:
Accept plain English questions
Generate safe, validated SQL queries
Execute queries against your database
Return formatted, readable results
This approach democratizes data access while maintaining security through validation layers and read-only access.
To take this further, consider:
Adding query explanation capabilities
Implementing result visualization
Supporting query refinement through conversation
Building a feedback loop to improve accuracy over time
Adding support for multiple databases
The key to success is thorough testing with real user questions, continuous prompt refinement, and robust validation. Start small with a limited schema, validate thoroughly, and expand gradually as you build confidence in the system.
Now go build something useful and let your team ask questions in plain English.
Top comments (0)