Introduction to AI Agents for Database Querying
Database querying has traditionally required fluency in SQL or a specific query language, creating a barrier between non-technical stakeholders and the data they need. An AI agent for database querying bridges that gap by translating natural language questions into executable queries, running them against a database, and returning human-readable answers. In this tutorial, you'll learn what these agents are, why they matter, and how to build one from scratch using Python, LangChain, and a SQLite database.
What Is an AI Database Querying Agent?
An AI database querying agent is a system that combines a large language model (LLM) with tools that interact with a database. Unlike a simple text-to-SQL generator, an agent can reason about the question, inspect the database schema, write a query, execute it, observe the results, and refine its approach if something goes wrong. This loop — often called a ReAct (Reasoning + Acting) pattern — makes the agent more robust than a single-shot prompt.
Typical components include:
- LLM: The reasoning engine (e.g., OpenAI GPT-4, Anthropic Claude, or an open-source model).
- Database connection: A driver such as SQLAlchemy that exposes the schema and executes queries.
- Tools: Functions the agent can call, like
list_tables,get_schema, andrun_query. - Orchestrator: A framework like LangChain or LangGraph that manages the reasoning loop.
Why It Matters
Building an AI agent for database querying delivers several concrete benefits:
- Democratized data access: Product managers, marketers, and executives can ask questions without waiting on a data team.
- Faster insights: Ad-hoc analysis that once took hours of back-and-forth now resolves in seconds.
- Reduced boilerplate: Developers can prototype dashboards and reports by describing them in plain English.
- Iterative exploration: Because the agent can inspect results and retry, it handles ambiguous questions better than one-shot generators.
However, these benefits come with real risks — SQL injection, unauthorized data access, and hallucinated queries — which we'll address in the best practices section.
Prerequisites and Setup
Before writing code, install the required Python packages and set up your environment. You'll need Python 3.10 or newer and an OpenAI API key.
pip install langchain langchain-openai langchain-community \
sqlalchemy sqlite3
Create a .env file to store your API key securely:
OPENAI_API_KEY=sk-your-key-here
Load the key in your script with python-dotenv so it never ends up in source control:
from dotenv import load_dotenv
load_dotenv()
Step 1: Create a Sample Database
For this tutorial we'll build a small e-commerce database with three tables: customers, orders, and products. Using SQLite keeps things lightweight — no server required.
import sqlite3
def create_database(db_path: str = "shop.db") -> None:
conn = sqlite3.connect(db_path)
cur = conn.cursor()
cur.executescript("""
DROP TABLE IF EXISTS customers;
DROP TABLE IF EXISTS products;
DROP TABLE IF EXISTS orders;
CREATE TABLE customers (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE NOT NULL,
country TEXT NOT NULL
);
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT NOT NULL,
price REAL NOT NULL
);
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
customer_id INTEGER NOT NULL,
product_id INTEGER NOT NULL,
quantity INTEGER NOT NULL,
order_date TEXT NOT NULL,
FOREIGN KEY (customer_id) REFERENCES customers(id),
FOREIGN KEY (product_id) REFERENCES products(id)
);
""")
cur.executemany(
"INSERT INTO customers (name, email, country) VALUES (?, ?, ?)",
[
("Alice Lee", "alice@example.com", "USA"),
("Bob Chen", "bob@example.com", "Canada"),
("Carla Diaz", "carla@example.com", "Spain"),
("David Kim", "david@example.com", "USA"),
],
)
cur.executemany(
"INSERT INTO products (name, category, price) VALUES (?, ?, ?)",
[
("Wireless Mouse", "Electronics", 25.99),
("Mechanical Keyboard", "Electronics", 89.50),
("Notebook", "Stationery", 4.20),
("Coffee Mug", "Kitchen", 12.00),
],
)
cur.executemany(
"INSERT INTO orders (customer_id, product_id, quantity, order_date) VALUES (?, ?, ?, ?)",
[
(1, 2, 1, "2024-03-12"),
(1, 3, 5, "2024-03-15"),
(2, 1, 2, "2024-04-02"),
(3, 4, 3, "2024-04-10"),
(4, 2, 1, "2024-05-01"),
(4, 1, 1, "2024-05-03"),
],
)
conn.commit()
conn.close()
print("Database created at", db_path)
if __name__ == "__main__":
create_database()
Run the script once to generate shop.db. You now have a realistic schema to query.
Step 2: Connect with SQLAlchemy
LangChain's SQL integrations work best with a SQLAlchemy engine. The engine gives the agent a uniform way to introspect tables and execute queries.
from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session
engine = create_engine("sqlite:///shop.db", echo=False)
def run_sql(query: str) -> str:
"""Execute a read-only SQL query and return rows as a string."""
with Session(engine) as session:
result = session.execute(text(query))
rows = result.fetchall()
columns = result.keys()
if not rows:
return "Query returned no rows."
header = " | ".join(columns)
body = "\n".join(" | ".join(str(v) for v in row) for row in rows)
return f"{header}\n{body}"
Notice we wrap execution in a session and return a formatted string. Returning a string (rather than raw Python objects) keeps the agent's tool output LLM-friendly.
Step 3: Define Agent Tools
Tools are the actions the agent can take. We'll expose three: listing tables, fetching a table's schema, and running a query. Each tool is a Python function decorated with @tool from LangChain.
from langchain_core.tools import tool
from sqlalchemy import inspect
@tool
def list_tables() -> str:
"""Return a comma-separated list of all tables in the database."""
inspector = inspect(engine)
return ", ".join(inspector.get_table_names())
@tool
def get_table_schema(table_name: str) -> str:
"""Return the column definitions for a given table."""
inspector = inspect(engine)
cols = inspector.get_columns(table_name)
lines = [f"{c['name']} ({c['type']})" for c in cols]
return "\n".join(lines)
@tool
def run_query(sql: str) -> str:
"""Execute a SELECT query against the SQLite database and return results.
Only use SELECT statements. Do not modify data."""
if not sql.strip().lower().startswith("select"):
return "Error: only SELECT queries are allowed."
try:
return run_sql(sql)
except Exception as e:
return f"Error executing query: {e}"
tools = [list_tables, get_table_schema, run_query]
The run_query tool includes a guard clause that rejects anything other than SELECT statements. This is a first line of defense against accidental data modification.
Step 4: Build the Agent
With tools defined, we can construct the agent. LangChain's create_tool_calling_agent works with models that support function calling, such as GPT-4o. The agent uses a prompt template that explains its role and the workflow it should follow.
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain.agents import create_tool_calling_agent, AgentExecutor
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
prompt = ChatPromptTemplate.from_messages([
("system", """You are a helpful database analyst.
You have access to a SQLite e-commerce database.
Always follow this workflow:
1. Call list_tables to see what tables exist.
2. Call get_table_schema for any table you plan to query.
3. Write a SELECT query and call run_query.
4. If the query fails, read the error, fix the SQL, and retry.
5. Summarize the results for the user in plain English.
Never run INSERT, UPDATE, DELETE, DROP, or ALTER statements.
If a question cannot be answered with the available tables, say so."""),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
Setting temperature=0 reduces randomness, which is desirable for query generation where precision matters. The system prompt encodes the agent's workflow, making its behavior predictable and debuggable.
Step 5: Query the Agent
Now you can ask questions in natural language and let the agent handle the rest.
questions = [
"Which country has the most customers?",
"What is the total revenue from electronics products?",
"List the top 2 customers by number of orders.",
]
for q in questions:
print("\n" + "=" * 60)
print("Question:", q)
response = agent_executor.invoke({"input": q})
print("Answer:", response["output"])
A typical run for the first question produces output like:
Question: Which country has the most customers?
> Entering AgentExecutor chain...
> Calling list_tables...
> Calling get_table_schema with table_name='customers'...
> Calling run_query with sql='SELECT country, COUNT(*) ...'
Answer: The USA has the most customers, with 2.
The agent autonomously discovered the schema, wrote the correct aggregation query, executed it, and translated the result into a sentence.
Step 6: Adding Memory for Follow-up Questions
Real users rarely ask isolated questions. They follow up: "What about last month?" or "Break that down by category." To support this, add conversational memory to the prompt.
from langchain_core.prompts import MessagesPlaceholder
prompt_with_memory = ChatPromptTemplate.from_messages([
("system", """You are a helpful database analyst..."""),
MessagesPlaceholder(variable_name="chat_history"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
])
agent = create_tool_calling_agent(llm, tools, prompt_with_memory)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
chat_history = []
def ask(question: str) -> str:
response = agent_executor.invoke(
{"input": question, "chat_history": chat_history}
)
chat_history.append(("human", question))
chat_history.append(("ai", response["output"]))
return response["output"]
print(ask("How many orders did Alice Lee place?"))
print(ask("What products did she buy?"))
The second question relies on context from the first. Because we append each exchange to chat_history, the agent knows "she" refers to Alice Lee.
Best Practices
Restrict Permissions at the Database Level
Never rely solely on prompt instructions to keep an agent safe. Create a dedicated database user with read-only privileges and connect the agent through that account. In PostgreSQL, for example:
CREATE ROLE agent_ro WITH LOGIN PASSWORD 'strongpass';
GRANT USAGE ON SCHEMA public TO agent_ro;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO agent_ro;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
GRANT SELECT ON TABLES TO agent_ro;
Even if the agent somehow generates a DROP TABLE statement, the database will reject it.
Limit Result Sets
LLMs have token limits, and large result sets waste money and degrade reasoning. Add a LIMIT clause automatically or wrap queries in a subquery:
def run_sql(query: str, max_rows: int = 50) -> str:
with Session(engine) as session:
result = session.execute(text(query))
rows = result.fetchmany(max_rows)
truncated = " (truncated)" if len(rows) == max_rows else ""
# ...format and return...
Validate and Sanitize SQL
Consider parsing generated SQL with sqlglot or sqlparse before execution. You can reject queries that reference disallowed tables, contain comments, or use dangerous functions.
import sqlglot
def validate_sql(sql: str, allowed_tables: set[str]) -> bool:
try:
parsed = sqlglot.parse_one(sql)
except Exception:
return False
tables = {t.name for t in parsed.find_all(sqlglot.exp.Table)}
return tables.issubset(allowed_tables)
Log Every Query
Maintain an audit log of every SQL statement the agent executes, along with the originating question and timestamp. This is essential for compliance, debugging, and detecting misuse.
import logging
from datetime import datetime
query_logger = logging.getLogger("agent_queries")
query_logger.setLevel(logging.INFO)
query_logger.addHandler(logging.FileHandler("agent_queries.log"))
@tool
def run_query(sql: str) -> str:
query_logger.info(f"{datetime.utcnow()} | {sql}")
# ...rest of implementation...
Provide Few-Shot Examples
For complex schemas, include example question-SQL pairs in the system prompt. Few-shot examples dramatically improve accuracy on tricky joins and aggregations.
FEW_SHOT = """
Example 1:
Q: How many customers are in Spain?
SQL: SELECT COUNT(*) FROM customers WHERE country = 'Spain';
Example 2:
Q: Total revenue per category?
SQL: SELECT p.category, SUM(p.price * o.quantity) AS revenue
FROM orders o JOIN products p ON o.product_id = p.id
GROUP BY p.category;
"""
Handle Ambiguity Gracefully
Train the agent to ask clarifying questions rather than guessing. Add instructions like: "If a question is ambiguous (e.g., 'top customers' without specifying a metric), ask the user to clarify before querying."
Conclusion
Building an AI agent for database querying is a powerful way to make data accessible across your organization. By combining an LLM with structured tools — schema inspection, query execution, and a reasoning loop — you create a system that handles ambiguous, multi-step questions far better than a one-shot text-to-SQL model. The key to a production-ready agent lies in defense in depth: read-only database roles, SQL validation, result limits, audit logging, and clear prompting. Start with the small SQLite example in this tutorial, then graduate to PostgreSQL or Snowflake with the same architecture. With thoughtful guardrails, your agent can turn natural language into reliable, governed data insights.