Build a Natural Language SQL Analyst Over Postgres Using Groq and Llama 3

What We're Building

We are building a command-line natural language SQL analyst. You type a question like “Which customers spent over $500 last month?” and the agent inspects your database schema, generates a safe SQL query, executes it on a free-tier Postgres instance, and explains the returned rows in plain English.

Feature list:

• Schema-aware prompting – introspects tables, columns, types, and foreign keys so the LLM writes correct JOINs.
• LLM SQL generation – uses Llama 3 70B on Groq’s free tier (insanely fast inference).
• Safety gate – restricts statements to SELECT only; strips DDL/DML before execution.
• Result narration – second LLM call turns raw tuples into a human-readable summary.
• Zero infrastructure cost – Supabase free tier for Postgres, Groq free credits for the LLM, local Python runtime.

If you enjoy stitching LLMs into real workflows, you might also like the Gmail Triage Agent or the Invoice Extractor guides — same philosophy, different data sources.

Architecture Overview

The system is a linear pipeline with four stages. No vector database, no caching — just deterministic plumbing between a database driver and an LLM API.

1. Schema Introspector queries information_schema and returns a compact DDL summary plus a few sample rows per table.
2. Prompt Assembler stitches the user question, DDL, samples, and strict instructions into a single prompt.
3. Groq returns a SQL string. We run a regex safety check to allow only SELECT statements.
4. Postgres Executor runs the query through SQLAlchemy with a read-only connection.
5. Result Explainer sends the rows back to Groq for a conversational summary.

Prerequisites

All free. No credit card required for Groq or Supabase to start.

Create a Supabase project, grab the connection string (Settings → Database → Connection string). Create a Groq API key (API Keys → Create). Store both as environment variables:

export SUPABASE_DB_URL="postgresql://postgres:[YOUR-PASSWORD]@db.[PROJECT-ID].supabase.co:5432/postgres"
export GROQ_API_KEY="gsk_..."

Step 1: Provision the Database

We need a table to query. Run this in the Supabase SQL Editor (or via psql):

CREATE TABLE customers (
    id SERIAL PRIMARY KEY,
    name TEXT NOT NULL,
    email TEXT UNIQUE,
    joined_at TIMESTAMP DEFAULT now()
);

CREATE TABLE orders (
    id SERIAL PRIMARY KEY,
    customer_id INTEGER REFERENCES customers(id),
    amount NUMERIC(10,2),
    order_date DATE DEFAULT CURRENT_DATE
);

INSERT INTO customers (name, email) VALUES
    ('Alice', 'alice@example.com'),
    ('Bob', 'bob@example.com'),
    ('Charlie', 'charlie@example.com');

INSERT INTO orders (customer_id, amount, order_date) VALUES
    (1, 150.00, '2025-03-01'),
    (1, 200.00, '2025-03-15'),
    (2, 450.00, '2025-03-10'),
    (3, 600.00, '2025-03-05'),
    (3, 50.00, '2025-03-20');

We now have three customers and five orders with a foreign-key relationship — enough to test JOINs and aggregations.

Step 2: Set Up the Python Environment

mkdir sql-analyst && cd sql-analyst
python -m venv .venv && source .venv/bin/activate
pip install sqlalchemy psycopg2-binary groq python-dotenv

Create a .env file with the two variables above, and a config.py:

import os
from dotenv import load_dotenv

load_dotenv()

SUPABASE_DB_URL = os.getenv("SUPABASE_DB_URL")
GROQ_API_KEY = os.getenv("GROQ_API_KEY")

Step 3: Build the Schema Introspector

We query information_schema to build a compact schema representation. This is what the LLM sees before writing SQL.

# introspector.py
from sqlalchemy import create_engine, text
from config import SUPABASE_DB_URL

engine = create_engine(SUPABASE_DB_URL)

def get_schema_summary() -> str:
    """Return a DDL-like summary of all tables, columns, types, and foreign keys."""
    with engine.connect() as conn:
        tables = conn.execute(text("""
            SELECT table_name FROM information_schema.tables
            WHERE table_schema = 'public' AND table_type = 'BASE TABLE'
        """)).fetchall()

        lines = []
        for (table_name,) in tables:
            columns = conn.execute(text("""
                SELECT column_name, data_type, is_nullable
                FROM information_schema.columns
                WHERE table_schema = 'public' AND table_name = :t
            """), {"t": table_name}).fetchall()

            col_strs = []
            for col_name, data_type, nullable in columns:
                null_str = "NULL" if nullable == "YES" else "NOT NULL"
                col_strs.append(f"  {col_name} {data_type} {null_str}")

            fks = conn.execute(text("""
                SELECT kcu.column_name, ccu.table_name, ccu.column_name
                FROM information_schema.table_constraints AS tc
                JOIN information_schema.key_column_usage AS kcu
                    ON tc.constraint_name = kcu.constraint_name
                JOIN information_schema.constraint_column_usage AS ccu
                    ON ccu.constraint_name = tc.constraint_name
                WHERE tc.constraint_type = 'FOREIGN KEY'
                  AND tc.table_schema = 'public'
                  AND tc.table_name = :t
            """), {"t": table_name}).fetchall()

            fk_strs = [f"  FK {col} -> {ref_table}({ref_col})" for col, ref_table, ref_col in fks]

            lines.append(f"TABLE {table_name} (")
            lines.extend(col_strs)
            lines.extend(fk_strs)
            lines.append(")")

        return "\n".join(lines)

def get_sample_rows(table_name: str, limit: int = 3) -> str:
    """Return a few sample rows for a table."""
    with engine.connect() as conn:
        rows = conn.execute(text(f"SELECT * FROM {table_name} LIMIT {limit}")).fetchall()
        if not rows:
            return f"-- {table_name}: no rows"
        cols = conn.execute(text(f"SELECT column_name FROM information_schema.columns WHERE table_name = :t"), {"t": table_name}).fetchall()
        col_names = [c[0] for c in cols]
        return "\n".join([",".join(str(v) for v in row) for row in rows])

Step 4: Build the LLM Query Generator

We use Groq's Python SDK. The prompt is everything — we include the schema, sample rows, and strict formatting instructions.

# llm_client.py
from groq import Groq
from config import GROQ_API_KEY

client = Groq(api_key=GROQ_API_KEY)

def generate_sql(question: str, schema: str, samples: str) -> str:
    system_prompt = """You are a PostgreSQL expert. Given a database schema, sample rows, and a user question, output ONLY a valid SQL SELECT statement.

Rules:
- Output ONLY the SQL, no markdown fences, no explanations.
- Use proper JOINs based on foreign keys.
- Use appropriate aggregations (SUM, COUNT, AVG) when the question implies them.
- Qualify column names with table aliases when ambiguous.
- Always end with a semicolon.
- If the question cannot be answered with the given schema, output: -- CANNOT_ANSWER
"""

    user_prompt = f"""Schema:
{schema}

Sample rows:
{samples}

Question: {question}

SQL:"""

    response = client.chat.completions.create(
        model="llama3-70b-8192",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_prompt}
        ],
        temperature=0.0,
        max_tokens=500
    )

    sql = response.choices[0].message.content.strip()
    # Remove accidental markdown fences
    if sql.startswith("```"):
        sql = sql.split("```")[1]
        if sql.startswith("sql"):
            sql = sql[3:]
    return sql.strip()

Step 5: Build the Safe Query Executor

We never pass raw LLM output to the database. A safety filter blocks anything that isn't a SELECT.

# executor.py
import re
from sqlalchemy import text
from introspector import engine

def is_safe_sql(sql: str) -> bool:
    """Only allow SELECT statements. Block DML, DDL, and multi-statement injections."""
    cleaned = sql.strip().rstrip(";").strip()
    # Must start with SELECT (case-insensitive)
    if not re.match(r"^\s*SELECT\b", cleaned, re.IGNORECASE):
        return False
    # Block multiple statements (semicolons inside the query)
    if ";" in cleaned[:-1]:  # allow trailing semicolon
        return False
    # Block dangerous keywords even inside comments
    dangerous = ["DROP", "DELETE", "INSERT", "UPDATE", "ALTER", "TRUNCATE", "CREATE", "EXEC", "GRANT", "REVOKE"]
    for keyword in dangerous:
        if re.search(rf"\b{keyword}\b", cleaned, re.IGNORECASE):
            return False
    return True

def execute_query(sql: str):
    """Execute a safe SELECT and return (columns, rows)."""
    if not is_safe_sql(sql):
        raise ValueError(f"Unsafe SQL blocked: {sql[:100]}")

    with engine.connect() as conn:
        result = conn.execute(text(sql))
        columns = list(result.keys())
        rows = [tuple(row) for row in result.fetchall()]
        return columns, rows

Step 6: Build the Result Explainer

After we get rows back, we send them to Groq again for a plain-English summary.

# explainer.py
from llm_client import client

def explain_results(question: str, sql: str, columns: list[str], rows: list[tuple]) -> str:
    if not rows:
        return "The query returned no results."

    rows_str = "\n".join([",".join(str(v) for v in row) for row in rows])
    prompt = f"""A user asked: "{question}"

The system generated this SQL:
{sql}

And got these results:
Columns: {columns}
Rows:
{rows_str}

Explain the results in 1-3 clear, conversational sentences. Be specific about numbers and entities."""

    response = client.chat.completions.create(
        model="llama3-70b-8192",
        messages=[{"role": "user", "content": prompt}],
        temperature=0.3,
        max_tokens=200
    )
    return response.choices[0].message.content.strip()

Step 7: Assemble the Chat Agent

A single analyst.py script ties everything together with an interactive loop.

# analyst.py
from introspector import get_schema_summary, get_sample_rows
from llm_client import generate_sql
from executor import execute_query
from explainer import explain_results

def main():
    print("Loading schema...")
    schema = get_schema_summary()
    # Grab sample rows from all public tables
    tables = ["customers", "orders"]  # you can auto-discover this in production
    samples = "\n".join([f"--- {t} ---\n{get_sample_rows(t)}" for t in tables])

    print("Schema loaded. Ask questions about your data.\n")

    while True:
        question = input("\n📊 Question: ").strip()
        if question.lower() in ("exit", "quit"):
            break
        if not question:
            continue

        print("  Generating SQL...")
        sql = generate_sql(question, schema, samples)
        if "CANNOT_ANSWER" in sql:
            print(f"  ❌ Cannot answer: {sql}")
            continue

        print(f"  SQL: {sql}")

        try:
            columns, rows = execute_query(sql)
        except ValueError as e:
            print(f"  ❌ Safety blocked: {e}")
            continue
        except Exception as e:
            print(f"  ❌ Execution error: {e}")
            continue

        print(f"  Rows returned: {len(rows)}")

        print("  Explaining...")
        explanation = explain_results(question, sql, columns, rows)
        print(f"\n  📝 {explanation}")

if __name__ == "__main__":
    main()

Running the Analyst

source .venv/bin/activate
python analyst.py

Example session:

📊 Question: who spent the most money and how much?
  Generating SQL...
  SQL: SELECT c.name, SUM(o.amount) as total_spent FROM customers c JOIN orders o ON c.id = o.customer_id GROUP BY c.name ORDER BY total_spent DESC LIMIT 1;
  Rows returned: 1
  Explaining...

  📝 Charlie spent the most money, with a total of $650.00 across two orders.

Extensions and Next Steps

• Streamlit UI — wrap the agent in a Streamlit app for a browser-based chat interface. Free hosting on Streamlit Community Cloud.
• Multi-turn memory — keep conversation context so users can ask follow-ups like “break that down by month.”
• Chart generation — add a Matplotlib step that renders bar/pie charts when the result set is small.
• Row-level security — if you add Supabase Auth, the agent can respect RLS policies by setting the current user ID in the session.
• Slack bot — deploy as a Slack slash command using Bolt for Python (free tier works).

The patterns here (schema → prompt → safety → execute → explain) transfer directly to the PR Review Bot and Gmail Triage Agent — both use the same “structured prompt + safety gate + action” architecture.

Common Pitfalls

FAQ

Q: Can I use this with MySQL or SQLite? Yes. Swap the SQLAlchemy connection URL and adjust information_schema queries for MySQL (INFORMATION_SCHEMA works similarly) or SQLite (PRAGMA table_info).

Q: What if my question requires a window function? Llama 3 70B handles ROW_NUMBER(), RANK(), LAG(), etc. well. Include a note in the system prompt that window functions are allowed.

Q: Is sending my schema to Groq safe? Groq does not train on API inputs. Your schema metadata (table names, column names) is sent — no actual row data goes in the SQL generation prompt, only sample rows you choose to include.

Q: How do I handle time-based questions like “last month”? The LLM will generate CURRENT_DATE - INTERVAL '1 month'. Make sure your order_date column is type DATE or TIMESTAMP so the introspection shows the correct type.

Q: Can I use a local model instead of Groq? Absolutely. Replace the Groq client with Ollama running llama3:70b or codellama. Latency will be higher on consumer hardware, but the architecture stays identical.

If you want to go deeper on prompt engineering for these agent pipelines, Context Engineering for Claude covers the principles behind structuring prompts when the model actually reads what you give it — applicable to Llama 3 as well.