pip install psycopg2-binary pgvector langchain langchain-openai openai sqlalchemy python-dotenv

# Docker: run PostgreSQL with pgvector support
docker run --name pgvector-demo \
  -e POSTGRES_PASSWORD=secret \
  -p 5432:5432 \
  -d pgvector/pgvector:pg16

# .env
OPENAI_API_KEY=sk-...
DATABASE_URL=postgresql://postgres:secret@localhost:5432/aidb

-- Enable the extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Create a products table with an embedding column
CREATE TABLE products (
    id         SERIAL PRIMARY KEY,
    name       TEXT NOT NULL,
    description TEXT,
    category   TEXT,
    price      DECIMAL(10, 2),
    embedding  VECTOR(1536)   -- OpenAI text-embedding-3-small dimensions
);

-- Create an HNSW index for fast approximate nearest-neighbor search
CREATE INDEX ON products
USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

import os
import psycopg2
import numpy as np
from openai import OpenAI
from dotenv import load_dotenv

load_dotenv()

client = OpenAI(api_key=os.environ["OPENAI_API_KEY"])
conn = psycopg2.connect(os.environ["DATABASE_URL"])
cur = conn.cursor()

def get_embedding(text: str) -> list[float]:
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

products = [
    ("Wireless Noise-Cancelling Headphones", "Premium audio with 30hr battery", "electronics", 299.99),
    ("Ergonomic Office Chair",              "Lumbar support, adjustable arms",  "furniture",   449.00),
    ("Python Programming Book",             "Comprehensive guide for beginners", "books",        39.99),
    ("Mechanical Keyboard",                 "Cherry MX switches, RGB backlit",   "electronics", 129.00),
    ("Standing Desk",                       "Electric height-adjustable desk",   "furniture",   599.00),
]

for name, description, category, price in products:
    embedding = get_embedding(f"{name}: {description}")
    cur.execute(
        """
        INSERT INTO products (name, description, category, price, embedding)
        VALUES (%s, %s, %s, %s, %s)
        """,
        (name, description, category, price, embedding)
    )

conn.commit()
print("Products inserted with embeddings.")

def semantic_search(query: str, max_price: float = None, category: str = None, limit: int = 5):
    query_embedding = get_embedding(query)

    sql = """
        SELECT
            name,
            description,
            category,
            price,
            1 - (embedding <=> %s::vector) AS similarity
        FROM products
        WHERE 1=1
    """
    params = [query_embedding]

    if max_price:
        sql += " AND price <= %s"
        params.append(max_price)

    if category:
        sql += " AND category = %s"
        params.append(category)

    sql += " ORDER BY embedding <=> %s::vector LIMIT %s"
    params += [query_embedding, limit]

    cur.execute(sql, params)
    return cur.fetchall()

# "I want something for my home office under $500"
results = semantic_search(
    query="comfortable workspace setup for long work sessions",
    max_price=500.0
)

for name, description, category, price, similarity in results:
    print(f"[{similarity:.3f}] {name} (${price}) — {description}")

[0.891] Ergonomic Office Chair ($449.00) — Lumbar support, adjustable arms
[0.832] Mechanical Keyboard ($129.00) — Cherry MX switches, RGB backlit
[0.801] Standing Desk ($599.00) — Electric height-adjustable desk

User Question
     │
     ▼
Embed the Question
     │
     ▼
Vector Search in PostgreSQL  ──▶  Top-K relevant chunks
     │
     ▼
Build LLM Prompt with context
     │
     ▼
LLM generates answer grounded in your data

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import PGVector
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA
from langchain_community.document_loaders import TextLoader

# 1. Load and chunk your documents
loader = TextLoader("company_docs.txt")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50
)
chunks = splitter.split_documents(documents)

# 2. Store chunks + embeddings in PostgreSQL
CONNECTION_STRING = os.environ["DATABASE_URL"]
COLLECTION_NAME = "company_knowledge_base"

vector_store = PGVector.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(model="text-embedding-3-small"),
    collection_name=COLLECTION_NAME,
    connection_string=CONNECTION_STRING,
)

# 3. Build a retrieval-augmented QA chain
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
retriever = vector_store.as_retriever(
    search_type="similarity",
    search_kwargs={"k": 4}
)

qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",
    retriever=retriever,
    return_source_documents=True
)

# 4. Ask questions
result = qa_chain.invoke({"query": "What is our refund policy?"})
print(result["result"])
print("\nSources:")
for doc in result["source_documents"]:
    print(f"  - {doc.metadata.get('source', 'unknown')}")

# Store documents with tenant metadata
vector_store = PGVector.from_documents(
    documents=chunks,
    embedding=OpenAIEmbeddings(),
    collection_name="docs",
    connection_string=CONNECTION_STRING,
    # Each chunk carries metadata you can filter on
)

# Retrieve only documents for a specific tenant
retriever = vector_store.as_retriever(
    search_kwargs={
        "k": 5,
        "filter": {"tenant_id": "acme-corp"}
    }
)

from openai import OpenAI

client = OpenAI()

SCHEMA = """
Tables:
  orders(id, customer_id, product_id, quantity, total_price, created_at, status)
  customers(id, name, email, country, created_at)
  products(id, name, category, price, stock)
"""

def nl_to_sql(question: str) -> str:
    prompt = f"""You are a SQL expert. Given the following database schema, write a PostgreSQL query to answer the user's question.
Return ONLY the SQL query, no explanation.

Schema:
{SCHEMA}

Question: {question}

SQL:"""

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": prompt}],
        temperature=0
    )
    return response.choices[0].message.content.strip()


# Real-world examples
questions = [
    "What are the top 5 countries by total revenue this year?",
    "Which products have never been ordered?",
    "Show me customers who placed more than 3 orders in the last 30 days",
]

for q in questions:
    print(f"Q: {q}")
    print(f"SQL: {nl_to_sql(q)}")
    print()

-- "What are the top 5 countries by total revenue this year?"
SELECT c.country, SUM(o.total_price) AS total_revenue
FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE EXTRACT(YEAR FROM o.created_at) = EXTRACT(YEAR FROM CURRENT_DATE)
GROUP BY c.country
ORDER BY total_revenue DESC
LIMIT 5;

-- "Which products have never been ordered?"
SELECT p.id, p.name, p.category
FROM products p
LEFT JOIN orders o ON p.id = o.product_id
WHERE o.id IS NULL;

import sqlparse
import re

ALLOWED_STATEMENTS = {"SELECT"}

def is_safe_sql(sql: str) -> bool:
    """Reject anything that isn't a SELECT statement."""
    parsed = sqlparse.parse(sql)
    if not parsed:
        return False
    statement_type = parsed[0].get_type()
    return statement_type in ALLOWED_STATEMENTS

def execute_nl_query(question: str, conn) -> list[dict]:
    sql = nl_to_sql(question)

    # Strip markdown fences if LLM added them
    sql = re.sub(r"```

sql|

```", "", sql).strip()

    if not is_safe_sql(sql):
        raise ValueError(f"Unsafe SQL rejected: {sql}")

    cur = conn.cursor()
    cur.execute(sql)
    columns = [desc[0] for desc in cur.description]
    rows = cur.fetchall()
    return [dict(zip(columns, row)) for row in rows]

FEW_SHOT_EXAMPLES = """
Example 1:
Question: How many orders were placed last month?
SQL: SELECT COUNT(*) FROM orders WHERE created_at >= DATE_TRUNC('month', NOW() - INTERVAL '1 month') AND created_at < DATE_TRUNC('month', NOW());

Example 2:
Question: What is the average order value by product category?
SQL: SELECT p.category, AVG(o.total_price) AS avg_order_value FROM orders o JOIN products p ON o.product_id = p.id GROUP BY p.category ORDER BY avg_order_value DESC;
"""

def nl_to_sql_fewshot(question: str) -> str:
    prompt = f"""You are a SQL expert for a PostgreSQL e-commerce database.

Schema:
{SCHEMA}

{FEW_SHOT_EXAMPLES}

Question: {question}
SQL:"""
    # ... same API call as before

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain.tools import tool
from langchain import hub
import psycopg2

conn = psycopg2.connect(os.environ["DATABASE_URL"])

@tool
def query_database(sql: str) -> str:
    """
    Execute a read-only SQL SELECT query against the e-commerce database.
    Use this to retrieve data needed to answer user questions.
    Never use INSERT, UPDATE, DELETE, or DROP.
    """
    if not is_safe_sql(sql):
        return "Error: Only SELECT queries are allowed."
    try:
        cur = conn.cursor()
        cur.execute(sql)
        columns = [desc[0] for desc in cur.description]
        rows = cur.fetchmany(50)  # cap results
        result = [dict(zip(columns, row)) for row in rows]
        return str(result)
    except Exception as e:
        return f"Query error: {e}"

@tool
def get_schema() -> str:
    """
    Returns the database schema — table names, columns, and types.
    Call this first to understand what data is available before writing queries.
    """
    cur = conn.cursor()
    cur.execute("""
        SELECT table_name, column_name, data_type
        FROM information_schema.columns
        WHERE table_schema = 'public'
        ORDER BY table_name, ordinal_position;
    """)
    rows = cur.fetchall()
    schema = {}
    for table, column, dtype in rows:
        schema.setdefault(table, []).append(f"{column} ({dtype})")
    return "\n".join(
        f"{table}: {', '.join(cols)}"
        for table, cols in schema.items()
    )

@tool
def semantic_product_search(query: str) -> str:
    """
    Search for products using semantic/natural language similarity.
    Use this when the user describes what they want rather than specifying exact product names.
    """
    results = semantic_search(query, limit=5)
    return str([{"name": r[0], "description": r[1], "price": r[3]} for r in results])

# Assemble the agent
llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [get_schema, query_database, semantic_product_search]

prompt = hub.pull("hwchase17/openai-tools-agent")

agent = create_openai_tools_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=10
)

# Complex multi-step question
response = agent_executor.invoke({
    "input": """
    I'm looking for ergonomic office products. 
    Find the top 3 most relevant products, then check if we have customers 
    from Germany who bought office furniture in the past 6 months. 
    Give me a summary of both findings.
    """
})

print(response["output"])

┌─────────────────────────────────────────────────────┐
│                   PostgreSQL Database               │
│                                                     │
│  ┌─────────────┐  ┌──────────────┐  ┌───────────┐  │
│  │  products   │  │  embeddings  │  │  orders   │  │
│  │  customers  │  │  (pgvector)  │  │  ...      │  │
│  └─────────────┘  └──────────────┘  └───────────┘  │
└───────────────────────┬─────────────────────────────┘
                        │
           ┌────────────┼────────────┐
           ▼            ▼            ▼
     ┌──────────┐ ┌──────────┐ ┌──────────┐
     │ Semantic │ │  NL2SQL  │ │  RAG     │
     │  Search  │ │  Agent   │ │ Pipeline │
     └──────────┘ └──────────┘ └──────────┘
           │            │            │
           └────────────┼────────────┘
                        ▼
                  ┌──────────┐
                  │  LLM     │
                  │ (GPT-4o) │
                  └──────────┘
                        │
                        ▼
                  User Response

-- HNSW: faster queries, slower inserts — best for production read-heavy workloads
CREATE INDEX ON products USING hnsw (embedding vector_cosine_ops)
WITH (m = 16, ef_construction = 64);

-- IVFFlat: faster inserts, slightly less accurate — good for bulk loading
CREATE INDEX ON products USING ivfflat (embedding vector_l2_ops)
WITH (lists = 100);
-- Rule of thumb: lists ≈ sqrt(number_of_rows)

# Don't call the embeddings API one row at a time
texts = [f"{p['name']}: {p['description']}" for p in products]

response = client.embeddings.create(
    model="text-embedding-3-small",
    input=texts  # batch up to 2048 inputs at once
)
embeddings = [item.embedding for item in response.data]

import hashlib
import json

query_cache: dict[str, str] = {}

def cached_nl_to_sql(question: str) -> str:
    key = hashlib.md5(question.lower().strip().encode()).hexdigest()
    if key not in query_cache:
        query_cache[key] = nl_to_sql(question)
    return query_cache[key]

-- Create a restricted role for your AI application
CREATE ROLE ai_readonly;
GRANT CONNECT ON DATABASE aidb TO ai_readonly;
GRANT USAGE ON SCHEMA public TO ai_readonly;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ai_readonly;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO ai_readonly;

sql-ai-database-solutions/
├── docker-compose.yml          # PostgreSQL + pgvector, ready to go
├── schema/
│   └── 001_init.sql            # Tables, pgvector extension, indexes
├── examples/
│   ├── 01_pgvector_search.py   # Semantic search
│   ├── 02_rag_pipeline.py      # RAG with LangChain
│   ├── 03_nl2sql.py            # Natural language to SQL
│   └── 04_sql_agent.py         # Autonomous SQL agent
├── .env.example
└── README.md