# Build a RAG-Powered Database Assistant with PostgreSQL and pgvector

> Source: <https://dev.to/blockchainlab/build-a-rag-powered-database-assistant-with-postgresql-and-pgvector-328i>
> Published: 2026-07-31 10:01:59+00:00

Build a RAG-Powered Database Assistant with PostgreSQL and pgvector

Tags: ai, database, postgres, tutorial

The Problem

Every company has the same conversation with their database:

"How many orders did we lose in the last 30 days?" "Which customers ordered more than three times but never left a review?" "Show me the revenue trend by region for the past quarter."

These questions are simpleblog_post_en.mdblog_post_en.md for a human analyst, but they never make it into a SQL query fast enough. Business teams wait for engineering. Engineering is busy. The query finally gets written three days later, and the answer is already stale.

What if the database itself could answer natural-language questions in seconds?

That is exactly what this tutorial builds: a RAG-powered database assistant that retrieves relevant schema context, generates SQL with an LLM, and returns the answer directly.

Why RAG instead of "just ask ChatGPT"?

Schema-aware: the LLM sees the actual tables, columns, and relationships in your database.

Up-to-date: metadata changes are reflected immediately without retraining.

Safe: the generated SQL runs on a read-only connection, so it cannot mutate data.

Explainable: users can see which schema context was used to answer the question.

Why PostgreSQL + pgvector

PostgreSQL is the most versatile production database in use today. It already supports JSON, full-text search, window functions, and rich indexes. Since 2023, the pgvector extension gives it production-grade vector search without adding another database to your stack.

Using the same database for both operational data and retrieval means:

One backup strategy, one monitoring stack, one set of credentials.

SQL joins between metadata and your real tables.

No extra network hop to a separate vector store.

Architecture

User question

|

v

[1] Embed question (text-embedding-3-small)

|

v

[2] pgvector similarity search over schema_metadata

|

v

[3] Build prompt: question + relevant tables/columns/examples

|

v

[4] LLM generates SQL (read-only)

|

v

[5] Execute SQL -> return answer + explanation

The key idea is that we are not storing documents; we are storing database schema metadata as the retrieval corpus. That is the difference between a generic chatbot and a database assistant.

Project Setup

Prerequisites

PostgreSQL 15+ with pgvector installed

Python 3.11+

An OpenAI-compatible API key

pip install "psycopg[binary]" openai

import os

import json

import psycopg

from openai import OpenAI

client = OpenAI() # reads OPENAI_API_KEY from environment

DB_URL = os.getenv(

"DATABASE_URL",

"postgresql://postgres:postgres@localhost:5432/rag_db",

)

Step 1: Enable pgvector and Create the Metadata Table

with psycopg.connect(DB_URL, autocommit=True) as conn:

conn.execute("CREATE EXTENSION IF NOT EXISTS vector")

conn.execute("""

CREATE TABLE IF NOT EXISTS schema_metadata (

id BIGSERIAL PRIMARY KEY,

object_name TEXT NOT NULL,

object_type TEXT NOT NULL, -- table | column | example

definition TEXT NOT NULL, -- DDL or column description

description TEXT NOT NULL, -- natural-language semantics

embedding vector(1536) -- text-embedding-3-small

);

""")

conn.execute("""

CREATE INDEX IF NOT EXISTS schema_metadata_embedding_idx

ON schema_metadata

USING hnsw (embedding vector_cosine_ops);

""")

The HNSW index gives approximate nearest-neighbor search with sub-10ms latency on modest datasets. For a few thousand metadata rows, this is effectively instant.

Step 2: Extract Schema Metadata from the Database

Instead of manually writing table descriptions, we extract the real schema from information_schema:

def fetch_schema_metadata(conn) -> list[dict]:

rows = conn.execute("""

SELECT

c.table_name,

c.column_name,

c.data_type,

pg_catalog.col_description(

('"' || c.table_schema || '"."' || c.table_name || '"')::regclass,

c.ordinal_position

) AS column_comment

FROM information_schema.columns c

WHERE c.table_schema = 'public'

ORDER BY c.table_name, c.ordinal_position

""").fetchall()

```
metadata = []
for table_name, column_name, data_type, comment in rows:
    metadata.append({
        "object_name": f"{table_name}.{column_name}",
        "object_type": "column",
        "definition": (
            f"Column: {column_name}\n"
            f"Type: {data_type}\n"
            f"Table: {table_name}"
        ),
        "description": comment or f"{column_name} column in {table_name} table",
    })
return metadata
```

We also add query examples. These teach the LLM the query style used by your team:

QUERY_EXAMPLES = [

{

"object_name": "example:monthly_revenue",

"object_type": "example",

"definition": (

"Question: What is the monthly revenue?\n"

"SQL: SELECT DATE_TRUNC('month', created_at) AS month, "

"SUM(total_amount) FROM orders GROUP BY 1 ORDER BY 1;"

),

"description": "Monthly revenue aggregation on orders",

},

{

"object_name": "example:top_customers",

"object_type": "example",

"definition": (

"Question: Who are the top 10 customers by spend?\n"

"SQL: SELECT c.name, SUM(o.total_amount) AS spend "

"FROM customers c JOIN orders o ON o.customer_id = c.id "

"GROUP BY c.id ORDER BY spend DESC LIMIT 10;"

),

"description": "Top customers by total order amount",

},

]

Step 3: Embed and Store Metadata

def embed_texts(texts: list[str]) -> list[list[float]]:

resp = client.embeddings.create(

model="text-embedding-3-small",

input=texts,

)

return [item.embedding for item in resp.data]

def ingest_metadata():

with psycopg.connect(DB_URL) as conn:

schema_metadata = fetch_schema_metadata(conn)

```
docs = []
for item in schema_metadata + QUERY_EXAMPLES:
    docs.append(f"{item['object_name']}\n{item['definition']}\n{item['description']}")

vectors = embed_texts(docs)

with psycopg.connect(DB_URL) as conn:
    conn.execute("TRUNCATE schema_metadata")
    for item, vector in zip(schema_metadata + QUERY_EXAMPLES, vectors):
        conn.execute("""
            INSERT INTO schema_metadata
                (object_name, object_type, definition, description, embedding)
            VALUES (%s, %s, %s, %s, %s)
        """, (
            item["object_name"],
            item["object_type"],
            item["definition"],
            item["description"],
            vector,
        ))
```

Note: vector accepts a Python list directly through psycopg; the extension serializes it to its native format.

Step 4: Retrieve Relevant Schema Context

def retrieve_context(question: str, top_k: int = 5) -> str:

q_vector = embed_texts([question])[0]

```
with psycopg.connect(DB_URL) as conn:
    rows = conn.execute("""
        SELECT object_name, object_type, definition, description,
               1 - (embedding <=> %s::vector) AS similarity
        FROM schema_metadata
        ORDER BY embedding <=> %s::vector
        LIMIT %s
    """, (q_vector, q_vector, top_k)).fetchall()

context_blocks = []
for name, obj_type, definition, description, sim in rows:
    context_blocks.append(
        f"### {name} ({obj_type}, similarity={sim:.3f})\n"
        f"{definition}\n{description}"
    )
return "\n\n".join(context_blocks)
```

The cosine operator <=> is the key line. It ranks schema fragments by semantic relevance to the user's question.

Step 5: Generate SQL and Return an Answer

SYSTEM_PROMPT = """

You are a senior database analyst. Your job is to convert

natural-language questions into safe, correct PostgreSQL.

Rules:

def generate_sql(question: str, context: str) -> str:

response = client.chat.completions.create(

model="gpt-4o-mini",

messages=[

{"role": "system", "content": SYSTEM_PROMPT},

{"role": "user", "content": (

f"Database schema context:\n{context}\n\n"

f"Question: {question}\n\n"

"Generate PostgreSQL."

)},

],

temperature=0,

)

content = response.choices[0].message.content

return extract_sql(content)

def extract_sql(content: str) -> str:

if "

`sql" in content:`

return content.split("

sql")[1].split("

``` php
")[0].strip()
    return content.strip()

def execute_read_only(sql: str) -> list[tuple]:
    # The application user owns the data; the assistant
    # connects with a read-only role in production.
    with psycopg.connect(DB_URL) as conn:
        cur = conn.execute(sql)
        return cur.fetchall()
Putting It All Together
def ask_database(question: str) -> None:
    print(f"Q: {question}\n")

    context = retrieve_context(question)
    print(f"Retrieved context:\n{context}\n")

    sql = generate_sql(question, context)
    print(f"Generated SQL:\n{sql}\n")

    rows = execute_read_only(sql)
    print(f"Result rows: {len(rows)}")
    for row in rows[:10]:
        print(row)

if __name__ == "__main__":
    ask_database("Show monthly revenue for the last six months.")
Example output:

Q: Show monthly revenue for the last six months.

Generated SQL:
SELECT DATE_TRUNC('month', created_at) AS month,
       SUM(total_amount) AS revenue
FROM orders
WHERE created_at >= CURRENT_DATE - INTERVAL '6 months'
GROUP BY 1
ORDER BY 1;

Result rows:
(datetime.date(2026, 2, 1), Decimal('128450.00'))
(datetime.date(2026, 3, 1), Decimal('153900.25'))
...
Production Optimization Tips
1. Add a schema refresh job
Run ingest_metadata() nightly, or hook it into your CI migration pipeline so new columns are searchable the same day they ship.

2. Use a read-only database role
CREATE ROLE assistant_readonly LOGIN PASSWORD '...';
GRANT CONNECT ON DATABASE rag_db TO assistant_readonly;
GRANT USAGE ON SCHEMA public TO assistant_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO assistant_readonly;
ALTER DEFAULT PRIVILEGES IN SCHEMA public
    GRANT SELECT ON TABLES TO assistant_readonly;
The application should connect with this role, not the superuser.

3. Cache embeddings
Embedding calls cost time and money. Cache by metadata content hash; re-embed only rows that changed.

4. Add guardrails
Validate the generated SQL with EXPLAIN before execution.
Reject queries containing forbidden keywords as a second safety net.
Log every question, SQL, and result count for auditability.
5. Let the LLM explain the result
After execution, send the row summary back to the LLM and ask for a concise, business-friendly interpretation. This is what makes the tool feel useful to non-technical teams.

When Not to Use This Pattern
Ad-hoc data exploration on massive tables: the LLM will write SELECT * or miss an index. Keep a human in the loop.
Financial reporting with strict audit requirements: use a governed BI tool instead.
Low-latency APIs: LLM inference adds seconds. Pre-generate SQL for known questions or cache results.
Complex domain semantics: if your schema names are cryptic, invest in comments first; retrieval is only as good as the metadata.
Conclusion
This pattern is not about replacing SQL. It is about making the database accessible to people who should not need to learn SQL to get an answer.

The stack is deliberately boring:

PostgreSQL for storage and search
pgvector for vector similarity
OpenAI-compatible embeddings and chat completions
A Python script that ties them together
The result is a database assistant that understands your schema, respects your query style, and answers business questions in seconds. Start with one table and one example query. Add more metadata as you trust it, and your team will stop waiting for engineering to write the next report.

Found this useful? Follow me for more practical AI + database engineering posts.
```