Building a SQL Query Agent with llama.cpp: Complete Guide
Large language models have transformed how developers interact with structured data. Instead of writing raw SQL queries, users can describe what they want in natural language and let an AI agent translate that intent into executable SQL. In this guide, you'll learn how to build a fully functional SQL Query Agent using llama.cpp, the lightweight C++ inference engine that runs LLMs locally without requiring a GPU server or cloud API keys.
What Is a SQL Query Agent?
A SQL Query Agent is an AI-powered system that bridges the gap between natural language and database queries. It takes a user's plain-English request, understands the database schema, generates a valid SQL query, executes it against the database, and returns the results in a human-readable format. With llama.cpp, you can run this entire pipeline locally, ensuring data privacy and eliminating API costs.
The agent typically performs three core tasks: schema awareness (understanding tables, columns, and relationships), query generation (translating intent to SQL), and result interpretation (explaining the output back to the user). This makes database interaction accessible to non-technical stakeholders while giving developers a powerful tool for rapid data exploration.
Why It Matters
Building a SQL agent with llama.cpp offers several compelling advantages over cloud-based alternatives:
- Data Privacy: All inference happens locally. Sensitive schemas and query results never leave your machine.
- Zero API Costs: No per-token charges. Run unlimited queries without worrying about usage limits.
- Offline Capability: Once the model is downloaded, the agent works without an internet connection.
- Customizable: You control the prompt engineering, schema injection, and validation logic entirely.
- Lightweight:
llama.cppruns efficiently on consumer hardware, including laptops with modest RAM.
Architecture Overview
Before diving into code, let's understand the architecture. The agent follows a multi-step pipeline that loops until it produces a valid, executable query or determines it cannot fulfill the request.
The flow is: user input → schema context injection → LLM generates SQL → validation layer checks syntax → execution against database → result formatting → LLM interprets results → final answer returned to user. If validation or execution fails, the error is fed back to the LLM for correction.
Prerequisites and Setup
You'll need Python 3.9+, a SQLite database for testing, and llama.cpp compiled with Python bindings. You'll also need a quantized model file. For SQL generation, models like CodeLlama-7B-Instruct or Llama-3-8B-Instruct in GGUF format work well.
Installing Dependencies
# Clone and build llama.cpp
git clone https://github.com/ggerganov/llama.cpp
cd llama.cpp
make
# Install Python bindings
pip install llama-cpp-python
# Install supporting libraries
pip install sqlite3 sqlalchemy python-dotenv
# Download a quantized model (example: CodeLlama 7B Q4_K_M)
# Place the .gguf file in your project directory
# Example path: ./models/codellama-7b-instruct.Q4_K_M.gguf
Setting Up a Sample Database
Let's create a sample e-commerce database to test our agent against:
import sqlite3
def create_sample_database(db_path="ecommerce.db"):
conn = sqlite3.connect(db_path)
cursor = conn.cursor()
cursor.execute('''
CREATE TABLE IF NOT EXISTS customers (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
email TEXT UNIQUE,
country TEXT,
signup_date TEXT
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS products (
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
category TEXT,
price REAL,
stock INTEGER
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS orders (
id INTEGER PRIMARY KEY,
customer_id INTEGER,
order_date TEXT,
total_amount REAL,
FOREIGN KEY (customer_id) REFERENCES customers(id)
)
''')
cursor.execute('''
CREATE TABLE IF NOT EXISTS order_items (
id INTEGER PRIMARY KEY,
order_id INTEGER,
product_id INTEGER,
quantity INTEGER,
unit_price REAL,
FOREIGN KEY (order_id) REFERENCES orders(id),
FOREIGN KEY (product_id) REFERENCES products(id)
)
''')
# Insert sample data
customers = [
(1, "Alice Johnson", "alice@email.com", "USA", "2024-01-15"),
(2, "Bob Smith", "bob@email.com", "UK", "2024-02-20"),
(3, "Carol Lee", "carol@email.com", "USA", "2024-03-10"),
]
products = [
(1, "Laptop", "Electronics", 999.99, 50),
(2, "Mouse", "Electronics", 29.99, 200),
(3, "Desk Chair", "Furniture", 199.99, 30),
(4, "Notebook", "Stationery", 4.99, 500),
]
orders = [
(1, 1, "2024-04-01", 1029.98),
(2, 2, "2024-04-05", 229.98),
(3, 1, "2024-04-10", 199.99),
(4, 3, "2024-04-15", 34.98),
]
order_items = [
(1, 1, 1, 1, 999.99),
(2, 1, 2, 1, 29.99),
(3, 2, 2, 1, 29.99),
(4, 2, 3, 1, 199.99),
(5, 3, 3, 1, 199.99),
(6, 4, 2, 5, 29.99),
(7, 4, 4, 5, 4.99),
]
cursor.executemany("INSERT OR REPLACE INTO customers VALUES (?,?,?,?,?)", customers)
cursor.executemany("INSERT OR REPLACE INTO products VALUES (?,?,?,?,?)", products)
cursor.executemany("INSERT OR REPLACE INTO orders VALUES (?,?,?,?)", orders)
cursor.executemany("INSERT OR REPLACE INTO order_items VALUES (?,?,?,?,?)", order_items)
conn.commit()
conn.close()
print("Sample database created successfully.")
if __name__ == "__main__":
create_sample_database()
Building the SQL Query Agent
Now let's build the core agent. We'll structure it as a class that handles schema extraction, prompt construction, SQL generation, validation, execution, and result interpretation.
Step 1: Schema Extraction
The agent needs to understand the database structure. We'll extract table definitions, column types, and foreign key relationships automatically.
import sqlite3
from typing import Dict, List
class SchemaExtractor:
def __init__(self, db_path: str):
self.db_path = db_path
def get_schema(self) -> str:
"""Extract the full database schema as a formatted string."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(
"SELECT name, sql FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'"
)
tables = cursor.fetchall()
schema_parts = []
for table_name, table_sql in tables:
schema_parts.append(f"-- Table: {table_name}")
schema_parts.append(table_sql + ";")
# Get column details
cursor.execute(f"PRAGMA table_info({table_name})")
columns = cursor.fetchall()
col_descriptions = []
for col in columns:
col_id, col_name, col_type, not_null, default_val, pk = col
constraints = []
if pk:
constraints.append("PRIMARY KEY")
if not_null:
constraints.append("NOT NULL")
if default_val is not None:
constraints.append(f"DEFAULT {default_val}")
constraint_str = " ".join(constraints) if constraints else ""
col_descriptions.append(f" - {col_name} ({col_type}) {constraint_str}".strip())
schema_parts.append("\n".join(col_descriptions))
# Get foreign keys
cursor.execute(f"PRAGMA foreign_key_list({table_name})")
fks = cursor.fetchall()
for fk in fks:
_, seq, ref_table, from_col, to_col, _, _ = fk
schema_parts.append(
f" - FK: {from_col} -> {ref_table}.{to_col}"
)
# Get row count for context
cursor.execute(f"SELECT COUNT(*) FROM {table_name}")
count = cursor.fetchone()[0]
schema_parts.append(f" - Row count: {count}")
schema_parts.append("")
conn.close()
return "\n".join(schema_parts)
def get_sample_rows(self, table_name: str, limit: int = 3) -> str:
"""Get sample rows from a table for additional context."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(f"SELECT * FROM {table_name} LIMIT {limit}")
rows = cursor.fetchall()
columns = [desc[0] for desc in cursor.description]
conn.close()
result = f"Sample rows from {table_name}:\n"
result += " | ".join(columns) + "\n"
for row in rows:
result += " | ".join(str(v) for v in row) + "\n"
return result
Step 2: LLM Integration with llama.cpp
Now we'll set up the LLM interface using llama-cpp-python. This handles loading the model and generating responses with proper prompt formatting.
from llama_cpp import Llama
class LLMInterface:
def __init__(self, model_path: str, n_ctx: int = 4096, n_threads: int = 4):
self.llm = Llama(
model_path=model_path,
n_ctx=n_ctx,
n_threads=n_threads,
n_gpu_layers=0, # Set to higher value if you have GPU support
verbose=False,
)
def generate(self, system_prompt: str, user_prompt: str,
max_tokens: int = 512, temperature: float = 0.1,
stop: List[str] = None) -> str:
"""Generate a response using the LLM."""
if stop is None:
stop = ["[/INST]", "[/ANSWER]"]
# Format prompt for Llama-2/CodeLlama chat format
formatted_prompt = f"[INST] <>\n{system_prompt}\n< >\n\n{user_prompt} [/INST]"
response = self.llm(
formatted_prompt,
max_tokens=max_tokens,
temperature=temperature,
top_p=0.95,
stop=stop,
echo=False,
)
return response["choices"][0]["text"].strip()
Step 3: SQL Validation and Execution
Before executing any generated SQL, we need a validation layer to prevent dangerous operations and catch syntax errors. This is critical for safety.
import re
class SQLExecutor:
# SQL keywords that indicate potentially dangerous operations
DANGEROUS_KEYWORDS = [
"DROP", "DELETE", "TRUNCATE", "ALTER", "INSERT",
"UPDATE", "CREATE", "ATTACH", "DETACH", "PRAGMA"
]
def __init__(self, db_path: str):
self.db_path = db_path
def validate_sql(self, sql: str) -> tuple[bool, str]:
"""Validate that SQL is safe and well-formed."""
if not sql or not sql.strip():
return False, "Empty SQL query."
sql_upper = sql.upper().strip()
# Must be a SELECT query or WITH (CTE)
if not (sql_upper.startswith("SELECT") or sql_upper.startswith("WITH")):
return False, "Only SELECT queries are allowed."
# Check for dangerous keywords
for keyword in self.DANGEROUS_KEYWORDS:
# Use word boundary matching to avoid false positives
pattern = r'\b' + keyword + r'\b'
if re.search(pattern, sql_upper):
return False, f"Potentially dangerous keyword detected: {keyword}"
# Check for semicolons that might indicate multiple statements
if ";" in sql.rstrip(";"):
return False, "Multiple SQL statements are not allowed."
return True, "Valid."
def execute(self, sql: str) -> tuple[bool, any, str]:
"""Execute SQL and return results."""
is_valid, message = self.validate_sql(sql)
if not is_valid:
return False, None, message
try:
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute(sql)
results = cursor.fetchall()
columns = [desc[0] for desc in cursor.description] if cursor.description else []
conn.close()
return True, {"columns": columns, "rows": results}, "Success."
except sqlite3.Error as e:
return False, None, f"SQLite error: {str(e)}"
except Exception as e:
return False, None, f"Unexpected error: {str(e)}"
def format_results(self, results: dict, max_rows: int = 20) -> str:
"""Format query results as a readable table."""
if not results["rows"]:
return "Query returned no rows."
columns = results["columns"]
rows = results["rows"][:max_rows]
# Calculate column widths
col_widths = [len(col) for col in columns]
for row in rows:
for i, val in enumerate(row):
col_widths[i] = max(col_widths[i], len(str(val)))
# Build formatted output
header = " | ".join(col.ljust(col_widths[i]) for i, col in enumerate(columns))
separator = "-+-".join("-" * w for w in col_widths)
data_rows = []
for row in rows:
data_rows.append(" | ".join(str(val).ljust(col_widths[i]) for i, val in enumerate(row)))
output = f"{header}\n{separator}\n" + "\n".join(data_rows)
if len(results["rows"]) > max_rows:
output += f"\n... and {len(results['rows']) - max_rows} more rows."
return output
Step 4: Assembling the Agent
Now we combine all components into the main agent class with a retry loop for self-correction.
class SQLQueryAgent:
def __init__(self, model_path: str, db_path: str, max_retries: int = 3):
self.schema_extractor = SchemaExtractor(db_path)
self.llm = LLMInterface(model_path)
self.executor = SQLExecutor(db_path)
self.max_retries = max_retries
self.schema = self.schema_extractor.get_schema()
def _build_sql_generation_prompt(self, question: str) -> str:
"""Build the prompt for SQL generation."""
system_prompt = (
"You are an expert SQL assistant. Your task is to write a SQLite query "
"that answers the user's question based on the provided database schema. "
"Rules:\n"
"1. Only write SELECT queries. Never use INSERT, UPDATE, DELETE, or DROP.\n"
"2. Use proper SQLite syntax.\n"
"3. Return ONLY the SQL query, no explanations.\n"
"4. If the question cannot be answered with the schema, return 'CANNOT_ANSWER'.\n"
"5. Use table aliases for readability.\n"
"6. Limit results to 100 rows unless specifically asked for more."
)
user_prompt = (
f"Database Schema:\n{self.schema}\n\n"
f"Question: {question}\n\n"
f"SQL Query:"
)
return system_prompt, user_prompt
def _build_interpretation_prompt(self, question: str,
sql: str, results: str) -> str:
"""Build the prompt for interpreting results."""
system_prompt = (
"You are a helpful data analyst. Based on the user's question, "
"the SQL query that was executed, and the results, provide a clear "
"and concise natural language answer. Include relevant numbers and "
"insights. Do not mention the SQL query itself."
)
user_prompt = (
f"User's question: {question}\n\n"
f"Query results:\n{results}\n\n"
f"Natural language answer:"
)
return system_prompt, user_prompt
def _extract_sql(self, response: str) -> str:
"""Extract SQL from the LLM response."""
response = response.strip()
# Remove markdown code blocks if present
if "sql" in response:
response = response.split("sql")[1].split("")[0]
elif "" in response:
response = response.split("")[1].split("")[0]
# Remove trailing explanation if present
response = response.strip().rstrip(";") + ";"
return response.strip()
def query(self, question: str) -> dict:
"""Process a natural language question and return the answer."""
print(f"\n{'='*60}")
print(f"Question: {question}")
print(f"{'='*60}")
for attempt in range(self.max_retries):
if attempt == 0:
system_prompt, user_prompt = self._build_sql_generation_prompt(question)
else:
# Include error feedback for self-correction
system_prompt, user_prompt = self._build_sql_generation_prompt(question)
user_prompt += (
f"\n\nPrevious attempt failed with error: {last_error}\n"
f"Please fix the query and try again."
)
print(f"\n[Attempt {attempt + 1}/{self.max_retries}] Generating SQL...")
# Generate SQL
raw_response = self.llm.generate(system_prompt, user_prompt, max_tokens=256)
if "CANNOT_ANSWER" in raw_response:
return {
"success": False,
"answer": "I cannot answer this question with the available database schema.",
"sql": None,
"results": None,
}
sql = self._extract_sql(raw_response)
print(f"Generated SQL: {sql}")
# Validate and execute
success, results, message = self.executor.execute(sql)
if success:
formatted_results = self.executor.format_results(results)
print(f"\nResults:\n{formatted_results}")
# Interpret results
interp_system, interp_user = self._build_interpretation_prompt(
question, sql, formatted_results
)
answer = self.llm.generate(
interp_system, interp_user, max_tokens=256, temperature=0.3
)
return {
"success": True,
"answer": answer,
"sql": sql,
"results": results,
}
else:
last_error = message
print(f"Error: {message}")
return {
"success": False,
"answer": f"Failed to generate a valid query after {self.max_retries} attempts. Last error: {last_error}",
"sql": None,
"results": None,
}
Step 5: Running the Agent
Let's put it all together and test with several questions:
if __name__ == "__main__":
# Initialize
MODEL_PATH = "./models/codellama-7b-instruct.Q4_K_M.gguf"
DB_PATH = "ecommerce.db"
# Create sample database
create_sample_database(DB_PATH)
# Create agent
agent = SQLQueryAgent(
model_path=MODEL_PATH,
db_path=DB_PATH,
max_retries=3
)
# Test questions
questions = [
"What are the top 3 products by total sales quantity?",
"How many orders did each customer place?",
"What is the total revenue from Electronics category?",
"Which customers are from the USA?",
]
for question in questions:
result = agent.query(question)
print(f"\nAnswer: {result['answer']}")
print(f"SQL used: {result['sql']}")
print("\n" + "="*60 + "\n")
Advanced Features
Adding Few-Shot Examples
Including example question-SQL pairs in your prompt dramatically improves accuracy, especially for complex joins and aggregations.
class SQLQueryAgent:
# ... existing code ...
FEW_SHOT_EXAMPLES = """
Example 1:
Question: What is the most expensive product?
SQL: SELECT name, price FROM products ORDER BY price DESC LIMIT 1;
Example 2:
Question: How many products are in each category?
SQL: SELECT category, COUNT(*) as count FROM products GROUP BY category;
Example 3:
Question: What is the total amount spent by each customer?
SQL: SELECT c.name, SUM(o.total_amount) as total_spent
FROM customers c
JOIN orders o ON c.id = o.customer_id
GROUP BY c.id, c.name
ORDER BY total_spent DESC;
"""
def _build_sql_generation_prompt(self, question: str) -> tuple:
system_prompt = (
"You are an expert SQL assistant. Your task is to write a SQLite query "
"that answers the user's question based on the provided database schema.\n"
"Rules:\n"
"1. Only write SELECT queries.\n"
"2. Use proper SQLite syntax.\n"
"3. Return ONLY the SQL query.\n"
"4. If the question cannot be answered, return 'CANNOT_ANSWER'.\n"
)
user_prompt = (
f"Database Schema:\n{self.schema}\n\n"
f"{self.FEW_SHOT_EXAMPLES}\n\n"
f"Question: {question}\n"
f"SQL Query:"
)
return system_prompt, user_prompt
Adding Conversation Memory
For follow-up questions, maintaining conversation context allows the agent to understand references like "those customers" or "that category."
class ConversationalSQLAgent(SQLQueryAgent):
def __init__(self, model_path: str, db_path: str, max_retries: int = 3):
super().__init__(model_path, db_path, max_retries)
self.conversation_history = []
def _build_sql_generation_prompt(self, question: str) -> tuple:
system_prompt = (
"You are an expert SQL assistant. Write a SQLite query that answers "
"the user's question. Consider the conversation history for context. "
"Return ONLY the SQL query."
)
history_text = ""
for entry in self.conversation_history[-3:]: # Keep last 3 exchanges
history_text += f"Previous Q: {entry['question']}\n"
history_text += f"Previous SQL: {entry['sql']}\n\n"
user_prompt = (
f"Database Schema:\n{self.schema}\n\n"
f"Conversation History:\n{history_text}\n"
f"Current Question: {question}\n"
f"SQL Query:"
)
return system_prompt, user_prompt
def query(self, question: str) -> dict:
result = super().query(question)
if result["success"]:
self.conversation_history.append({
"question": question,
"sql": result["sql"],
})
return result
Best Practices
- Always validate SQL before execution: Never trust LLM-generated SQL blindly. Your validation layer should enforce read-only operations and reject any data modification commands.
- Use low temperature for SQL generation: Set temperature between 0.1 and 0.2 for query generation to reduce randomness. Use slightly higher temperature (0.3-0.5) for result interpretation where creativity is acceptable.
- Provide rich schema context: Include not just table definitions but also foreign key relationships, row counts, and sample data. The more context the LLM has, the better its queries will be.
- Implement retry with error feedback: When a query fails, feed the error message back to the LLM. Models can often self-correct when they see the specific error.
- Limit result sets: Add
LIMITclauses or enforce them programmatically to prevent the agent from returning massive result sets that consume context window space. - Log all queries: Maintain an audit log of all generated SQL and execution results for debugging and compliance purposes.
- Choose the right model: Code-focused models like CodeLlama generally produce better SQL than general-purpose models. Larger models (13B+) handle complex joins better but require more RAM.
- Test with edge cases: Validate your agent handles ambiguous questions, requests for non-existent data, and questions requiring multi-table joins correctly.
Performance Optimization Tips
Running LLMs locally requires attention to performance. Here are key optimizations:
# Optimize llama.cpp settings for your hardware
llm = Llama(
model_path=model_path,
n_ctx=4096, # Context window size
n_threads=8, # Match your CPU core count
n_gpu_layers=35, # Offload layers to GPU if available
use_mlock=True, # Lock model in RAM to prevent swapping
use_mmap=True, # Memory-map the model file
n_batch=512, # Batch size for prompt processing
verbose=False,
)
# For repeated queries, cache the schema string
# to avoid re-extracting it every time
# Use quantized models (Q4_K_M is a good balance of speed and quality)
# Q5_K_M offers slightly better quality at a modest speed cost
Conclusion
Building a SQL Query Agent with llama.cpp gives you a powerful, private, and cost-effective way to enable natural language database interaction. By combining schema extraction, careful prompt engineering, robust validation, and a self-correcting retry loop, you can create an agent that handles real-world queries reliably. The architecture presented here — with its clear separation between schema awareness, SQL generation, validation, execution, and interpretation — is extensible. You can add features like multi-database support, query caching, visualization generation, or integration with business intelligence tools. As local LLMs continue to improve in capability and efficiency, self-hosted SQL agents will become an increasingly practical alternative to cloud-based solutions, especially for organizations with strict data privacy requirements or limited budgets. Start with the sample database in this guide, experiment with different models and prompt strategies, and gradually adapt the agent to your specific database schemas and use cases.