What You Can Build

Why This Matters

Keep Everything Together

Embedding models and retrieval tricks change every few months. Your database shouldn't have to. On Postgres you can swap models, try a new retrieval strategy, or add graph traversal without re-architecting anything.

Storing and Querying Vectors

Enable pgvector

CREATE EXTENSION vector;

Create a Table with Embeddings

CREATE TABLE documents (
  id         bigint PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  title      text NOT NULL,
  content    text NOT NULL,
  metadata   jsonb DEFAULT '{}',
  embedding  vector(1536)
);

Distance Operators

-- Cosine distance (most common for normalized embeddings)
SELECT title, 1 - (embedding <=> $1::vector) AS similarity
FROM documents
ORDER BY embedding <=> $1::vector
LIMIT 5;

-- L2 (Euclidean) distance
SELECT title, embedding <-> $1::vector AS distance
FROM documents
ORDER BY embedding <-> $1::vector
LIMIT 5;

-- Inner product (returns the negative inner product, so multiply by -1)
SELECT title, (embedding <#> $1::vector) * -1 AS score
FROM documents
ORDER BY embedding <#> $1::vector
LIMIT 5;

Indexing for Performance

CREATE INDEX ON documents
  USING hnsw (embedding vector_cosine_ops)
  WITH (m = 16, ef_construction = 128);

-- Tune query-time recall vs. speed (default is 40)
SET hnsw.ef_search = 100;

-- Build after loading data (it needs a representative sample)
CREATE INDEX ON documents
  USING ivfflat (embedding vector_cosine_ops)
  WITH (lists = 100);

-- Tune: higher probes = better recall, more latency
SET ivfflat.probes = 10;

Filtering and Iterative Index Scans

-- Results come back in approximate distance order
SET hnsw.iterative_scan = relaxed_order;

-- Or require exact distance order (slightly more work)
SET hnsw.iterative_scan = strict_order;

-- Cap how far it will go (default 20000 tuples)
SET hnsw.max_scan_tuples = 20000;

Half-Precision, Sparse, and Binary Vectors

-- Half-precision vectors use about half the storage and index memory
CREATE TABLE documents_half (
  id        bigint PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  content   text NOT NULL,
  embedding halfvec(1536)
);

CREATE INDEX ON documents_half
  USING hnsw (embedding halfvec_cosine_ops);

CREATE INDEX ON documents
  USING hnsw ((embedding::halfvec(1536)) halfvec_cosine_ops);

SELECT title
FROM documents
ORDER BY embedding::halfvec(1536) <=> $1::halfvec(1536)
LIMIT 5;

CREATE INDEX ON documents
  USING hnsw ((binary_quantize(embedding)::bit(1536)) bit_hamming_ops);

SELECT id, title FROM (
  SELECT id, title, embedding
  FROM documents
  ORDER BY binary_quantize(embedding)::bit(1536) <~> binary_quantize($1::vector)
  LIMIT 50
) candidates
ORDER BY embedding <=> $1::vector
LIMIT 5;

Reach for halfvec first when memory gets tight. For most applications the recall loss is small and you cut index memory in half. Binary quantization saves more but needs the re-rank step, so test recall on your own data.

Semantic Search

Why Hybrid Search?

Setting Up Hybrid Search

CREATE TABLE articles (
  id         bigint PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  title      text NOT NULL,
  body       text NOT NULL,
  -- Full-text search column (auto-updated)
  tsv        tsvector GENERATED ALWAYS AS (
               to_tsvector('english', title || ' ' || body)
             ) STORED,
  -- Vector embedding for semantic similarity
  embedding  vector(1536)
);

-- Index both search methods
CREATE INDEX ON articles USING gin (tsv);
CREATE INDEX ON articles USING hnsw (embedding vector_cosine_ops);

Hybrid Search Query

WITH keyword_results AS (
  SELECT id,
         ROW_NUMBER() OVER (ORDER BY ts_rank_cd(tsv, query) DESC) AS rank_pos
  FROM articles, plainto_tsquery('english', $1) query
  WHERE tsv @@ query
  ORDER BY rank_pos
  LIMIT 20
),
vector_results AS (
  SELECT id,
         ROW_NUMBER() OVER (ORDER BY embedding <=> $2::vector) AS rank_pos
  FROM articles
  ORDER BY embedding <=> $2::vector
  LIMIT 20
),
combined AS (
  SELECT
    COALESCE(k.id, v.id) AS id,
    -- RRF: sum of 1/(60 + rank position) across both lists.
    -- A row missing from one list contributes 0 from that list.
    COALESCE(1.0 / (60 + k.rank_pos), 0) +
    COALESCE(1.0 / (60 + v.rank_pos), 0) AS rrf_score
  FROM keyword_results k
  FULL OUTER JOIN vector_results v ON k.id = v.id
)
SELECT a.id, a.title, a.body, c.rrf_score
FROM combined c
JOIN articles a ON a.id = c.id
ORDER BY c.rrf_score DESC
LIMIT 10;

aside negative The 60 constant comes from the original RRF paper. It keeps a top result in one list from drowning out everything else. You can tune it, but 60 is a fine default.

Simpler Hybrid Search with Weighted Scoring

SELECT id, title, body,
  (0.7 * (1 - (embedding <=> $2::vector))) +
  (0.3 * ts_rank_cd(tsv, plainto_tsquery('english', $1))) AS score
FROM articles
WHERE tsv @@ plainto_tsquery('english', $1)
   OR embedding <=> $2::vector < 0.5
ORDER BY score DESC
LIMIT 10;

RAG (Retrieval-Augmented Generation)

The RAG Pattern

Schema for RAG

-- Source documents
CREATE TABLE documents (
  id           bigint PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  title        text NOT NULL,
  source_url   text,
  content      text NOT NULL,
  metadata     jsonb DEFAULT '{}',
  created_at   timestamptz DEFAULT now()
);

-- Chunked and embedded content
CREATE TABLE chunks (
  id           bigint PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  document_id  bigint REFERENCES documents(id) ON DELETE CASCADE,
  chunk_index  int NOT NULL,
  content      text NOT NULL,
  token_count  int,
  embedding    vector(1536),
  UNIQUE (document_id, chunk_index)
);

-- Index for fast retrieval
CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);
CREATE INDEX ON chunks (document_id);

Retrieval with Metadata Filtering

-- Find relevant chunks, filtered by category and recency
SET hnsw.iterative_scan = relaxed_order;

SELECT c.content, c.chunk_index,
       d.title, d.source_url,
       1 - (c.embedding <=> $1::vector) AS similarity
FROM chunks c
JOIN documents d ON d.id = c.document_id
WHERE d.metadata->>'category' = 'engineering'
  AND d.created_at > now() - interval '90 days'
ORDER BY c.embedding <=> $1::vector
LIMIT 5;

Chunking Strategies

-- Example: inserting chunks with overlap tracking
INSERT INTO chunks (document_id, chunk_index, content, token_count, embedding)
VALUES
  ($1, 0, $2, $3, $4),  -- tokens 0-500
  ($1, 1, $5, $6, $7),  -- tokens 450-950 (50-token overlap)
  ($1, 2, $8, $9, $10); -- tokens 900-1400

aside negative Chunk size matters. Too large and the embedding loses specificity; too small and you lose context. Start with 500 to 1000 tokens and 10 to 20% overlap, then iterate based on retrieval quality.

Context Window Assembly

-- Get retrieved chunks with their previous and next chunks
SELECT content, prev_chunk, next_chunk
FROM (
  SELECT c.id, c.content, c.embedding,
         lag(c.content)  OVER w AS prev_chunk,
         lead(c.content) OVER w AS next_chunk
  FROM chunks c
  WHERE c.document_id IN (
    SELECT document_id FROM chunks WHERE id IN (SELECT id FROM retrieved_chunk_ids)
  )
  WINDOW w AS (PARTITION BY c.document_id ORDER BY c.chunk_index)
) with_context
WHERE id IN (SELECT id FROM retrieved_chunk_ids)
ORDER BY embedding <=> $1::vector;

GraphRAG

When to Use GraphRAG

Setting Up a Knowledge Graph with Apache AGE

-- Enable the AGE extension
CREATE EXTENSION age;
LOAD 'age';
SET search_path = ag_catalog, "$user", public;

-- Create a knowledge graph
SELECT create_graph('knowledge_base');

Building the Graph

-- Add entities as graph nodes
SELECT * FROM cypher('knowledge_base', $$
  CREATE (:Document {
    id: 1,
    title: 'Postgres Performance Guide',
    category: 'engineering'
  })
$$) AS (v agtype);

SELECT * FROM cypher('knowledge_base', $$
  CREATE (:Concept {
    name: 'Index Tuning',
    description: 'Optimizing database indexes for query performance'
  })
$$) AS (v agtype);

-- Create relationships
SELECT * FROM cypher('knowledge_base', $$
  MATCH (d:Document {id: 1}), (c:Concept {name: 'Index Tuning'})
  CREATE (d)-[:COVERS {section: 'chapter_3', depth: 'detailed'}]->(c)
$$) AS (e agtype);

GraphRAG Retrieval Pattern

-- Step 1: Find relevant chunks via vector search
WITH seed_chunks AS (
  SELECT c.id, c.document_id, c.content,
         1 - (c.embedding <=> $1::vector) AS similarity
  FROM chunks c
  ORDER BY c.embedding <=> $1::vector
  LIMIT 5
),
-- Step 2: Documents connected to other documents within 2 hops
doc_links AS (
  SELECT (src::text)::bigint AS src_doc_id,
         (dst::text)::bigint AS related_doc_id
  FROM cypher('knowledge_base', $$
    MATCH (d:Document)-[*1..2]-(related:Document)
    RETURN d.id, related.id
  $$) AS (src agtype, dst agtype)
),
-- Step 3: Chunks from documents linked to the seed documents
graph_chunks AS (
  SELECT c.content, c.embedding, 'graph' AS source
  FROM chunks c
  WHERE c.document_id IN (
    SELECT l.related_doc_id
    FROM doc_links l
    WHERE l.src_doc_id IN (SELECT document_id FROM seed_chunks)
  )
  ORDER BY c.embedding <=> $1::vector
  LIMIT 5
)
-- Combine vector-similar and graph-connected chunks
SELECT content, similarity, source FROM (
  SELECT content, similarity, 'vector' AS source FROM seed_chunks
  UNION ALL
  SELECT content, 1 - (embedding <=> $1::vector), source FROM graph_chunks
) combined
ORDER BY similarity DESC
LIMIT 10;

Entity Extraction and Graph Maintenance

-- Table to track extracted entities
CREATE TABLE entities (
  id            bigint PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  document_id   bigint REFERENCES documents(id) ON DELETE CASCADE,
  entity_name   text NOT NULL,
  entity_type   text NOT NULL,  -- 'person', 'concept', 'tool', etc.
  properties    jsonb DEFAULT '{}'
);

CREATE TABLE entity_relationships (
  id            bigint PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
  source_id     bigint REFERENCES entities(id) ON DELETE CASCADE,
  target_id     bigint REFERENCES entities(id) ON DELETE CASCADE,
  relationship  text NOT NULL,  -- 'mentions', 'depends_on', 'authored_by'
  properties    jsonb DEFAULT '{}'
);

-- Indexes for fast graph traversal
CREATE INDEX ON entity_relationships (source_id);
CREATE INDEX ON entity_relationships (target_id);
CREATE INDEX ON entities (entity_type, entity_name);

GraphRAG with Common Table Expression (CTE)

-- Traverse relationships up to 3 hops from seed entities
WITH RECURSIVE graph_walk AS (
  -- Start from entities mentioned in relevant chunks
  SELECT e.id, e.entity_name, e.entity_type, 0 AS depth
  FROM entities e
  JOIN chunks c ON c.document_id = e.document_id
  WHERE c.id IN (SELECT id FROM seed_chunk_ids)

  UNION ALL

  -- Walk relationships
  SELECT e2.id, e2.entity_name, e2.entity_type, gw.depth + 1
  FROM graph_walk gw
  JOIN entity_relationships er ON er.source_id = gw.id
  JOIN entities e2 ON e2.id = er.target_id
  WHERE gw.depth < 3
)
SELECT entity_name, entity_type, min(depth) AS min_depth
FROM graph_walk
GROUP BY entity_name, entity_type
ORDER BY min_depth, entity_name;

You don't need a dedicated graph database for GraphRAG. Recursive CTEs cover most knowledge graph traversal patterns. AGE gives you openCypher syntax for more complex graph queries, but plain relational tables work well for simpler graphs.

Snowflake Cortex with Postgres from Python

Connect to Both Databases

pip install snowflake-connector-python "psycopg[binary]" pandas

import os

import pandas as pd
import psycopg
import snowflake.connector

# Snowflake: reads credentials from ~/.snowflake/connections.toml
sf = snowflake.connector.connect(connection_name="my_connection")

# Postgres: a normal libpq connection string, e.g.
# postgresql://application:password@<your-instance-host>:5432/postgres?sslmode=require
pg = psycopg.connect(os.environ["DATABASE_URL"], autocommit=True)


def pg_df(sql: str, params=None) -> pd.DataFrame:
    """Run a query in Postgres and return a DataFrame."""
    with pg.cursor() as cur:
        cur.execute(sql, params)
        cols = [c.name for c in cur.description]
        return pd.DataFrame(cur.fetchall(), columns=cols)

A Small AI_COMPLETE Helper

MODEL = "llama3.1-70b"  # pick any model available in your region


def complete(prompt: str) -> str:
    """Send a prompt to Cortex AI_COMPLETE and return the response text."""
    with sf.cursor() as cur:
        cur.execute("SELECT AI_COMPLETE(%s, %s)", (MODEL, prompt))
        return cur.fetchone()[0]

Ask Questions About Your Data

def data_context() -> str:
    """Build a text snapshot of the data for the LLM."""
    parts = []

    daily = pg_df("""
        SELECT date_trunc('day', created_at)::date AS day,
               count(*)                            AS orders,
               round(sum(total)::numeric, 2)       AS revenue
        FROM orders
        WHERE created_at >= now() - interval '14 days'
        GROUP BY 1
        ORDER BY 1
    """)
    parts.append("DAILY ORDERS (last 14 days):\n" + daily.to_string(index=False))

    top = pg_df("""
        SELECT p.name, sum(oi.quantity) AS units
        FROM order_items oi
        JOIN products p ON p.id = oi.product_id
        JOIN orders o   ON o.id = oi.order_id
        WHERE o.created_at >= now() - interval '14 days'
        GROUP BY p.name
        ORDER BY units DESC
        LIMIT 10
    """)
    parts.append("TOP PRODUCTS (last 14 days):\n" + top.to_string(index=False))

    return "\n\n".join(parts)


def ask(question: str) -> str:
    prompt = (
        "You are a data analyst. Answer the question using only the data below. "
        "Give specific numbers. If the data doesn't answer it, say so.\n\n"
        f"{data_context()}\n\n"
        f"Question: {question}"
    )
    return complete(prompt)


print(ask("Was last week better than the week before? What drove it?"))

Natural Language to SQL to Chart

SCHEMA = """
orders(id bigint, customer_id bigint, created_at timestamptz, total numeric, region text)
order_items(order_id bigint, product_id bigint, quantity int, price numeric)
products(id bigint, name text, category text)
"""


def nl_to_sql(question: str) -> str:
    prompt = (
        "Write a single Postgres SELECT query that answers the question. "
        "Return only the SQL, no explanation and no markdown.\n\n"
        f"Schema:\n{SCHEMA}\n"
        f"Question: {question}"
    )
    sql = complete(prompt).strip()
    # models sometimes wrap answers in code fences anyway
    return sql.removeprefix("```sql").removeprefix("```").removesuffix("```").strip()


question = "average order value by week for the last 3 months"
sql = nl_to_sql(question)
print(sql)

df = pg_df(sql)
df.plot(x=df.columns[0], y=df.columns[1])  # or st.line_chart(df) in Streamlit

Retrieval Plus AI_COMPLETE (RAG)

def answer(question: str, query_embedding) -> str:
    chunks = pg_df(
        """
        SELECT id, content
        FROM chunks
        ORDER BY embedding <=> %s::vector
        LIMIT 5
        """,
        (str(list(query_embedding)),),
    )
    context = "\n\n".join(f"[{r.id}] {r.content}" for r in chunks.itertuples())

    prompt = (
        "Answer the question using only the context below. "
        "Cite the chunk ids you used in square brackets. "
        "If the context doesn't contain the answer, say so.\n\n"
        f"Context:\n{context}\n\nQuestion: {question}"
    )
    return complete(prompt)

Other Cortex AI Functions

Running It in Streamlit

Production Notes

Conclusion

Related Resources

---

---

---

---

---