The Definitive Guide to Vector Databases & RAG

# Langchain provides excellent utilities for this
from langchain_text_splitters import RecursiveCharacterTextSplitter

raw_document = """
Neo4j is a graph database management system. 
It was developed by Neo4j, Inc.

... (thousands of words) ...

Pgvector is an open-source extension for PostgreSQL.
It enables similarity search.
"""

# Define the chunking strategy
# chunk_size: Target length of each chunk (in characters or tokens)
# chunk_overlap: How many characters to bleed over into the next chunk to retain context
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    length_function=len,
    is_separator_regex=False,
)

# Execute the split
chunks = text_splitter.create_documents([raw_document])

print(f"Split into {len(chunks)} contextual chunks.")
for i, chunk in enumerate(chunks[:2]):
    print(f"--- Chunk {i} ---")
    print(chunk.page_content)

def contextualize_chunks(document_text, chunks):
    # 1. Ask a fast LLM (like GPT-4o-mini or Claude Haiku) for a summary
    document_summary = llm.invoke(f"Briefly summarize this document to provide context: {document_text}")
    
    contextualized_chunks = []
    for chunk in chunks:
        # 2. Prepend the global summary to the local chunk
        enriched_text = f"DOCUMENT CONTEXT: {document_summary}\n\nCHUNK CONTENT: {chunk}"
        contextualized_chunks.append(enriched_text)
        
    return contextualized_chunks

# Now, when you embed 'contextualized_chunks', the vector math
# will always accurately match searches for "Acme Corp 2023 Revenue"
# because "Acme Corp 2023" is hardcoded into every chunk's vector space.

import psycopg
from pgvector.psycopg import register_vector
from sentence_transformers import SentenceTransformer

# 1. Load the open-source embedding model locally
# This model outputs arrays of 384 floats.
model = SentenceTransformer('all-MiniLM-L6-v2')

# 2. Connect to Postgres and register the custom vector type
# (Assumes you have run `CREATE EXTENSION vector;` in your DB)
conn = psycopg.connect("dbname=postgres user=postgres password=secret")
register_vector(conn)

# 3. Create Table and HNSW Index for fast ANN search
with conn.cursor() as cur:
    cur.execute("""
        CREATE TABLE IF NOT EXISTS chunks (
            id serial PRIMARY KEY,
            content text,
            department varchar,
            embedding vector(384)
        );
    """)
    
    # HNSW is vastly superior to IVFFlat for scaling. 
    # vector_cosine_ops tells the index we will be using Cosine distance.
    cur.execute("""
        CREATE INDEX ON chunks USING hnsw (embedding vector_cosine_ops);
    """)
    
    # 4. Generate Vectors and Insert
    docs = ["Policy 1A: Remote work is allowed on Tuesdays.", "Policy 2B: Expense reports..."]
    depts = ["HR", "Finance"]
    embeddings = model.encode(docs)
    
    for doc, dept, emb in zip(docs, depts, embeddings):
        cur.execute(
            "INSERT INTO chunks (content, department, embedding) VALUES (%s, %s, %s)", 
            (doc, dept, emb)
        )

    # 5. Search with a CONFIDENCE THRESHOLD directly in SQL
    user_query = "When can I work from home?"
    query_vector = model.encode([user_query])[0]
    
    # Operator <=> calculates Cosine Distance. 
    # We invert it (1 - distance) to get a Similarity Score (0.0 to 1.0).
    # We also filter by department using standard SQL!
    cur.execute("""
        SELECT content, department, (1 - (embedding <=> %s::vector)) AS similarity
        FROM chunks
        WHERE (embedding <=> %s::vector) < 0.4  -- Distance < 0.4 means > 60% similarity
          AND department = 'HR'              -- Standard relational filter!
        ORDER BY embedding <=> %s::vector
        LIMIT 5;
    """, (query_vector, query_vector, query_vector))
    
    print("--- pgvector Search Results ---")
    for row in cur.fetchall():
        print(f"[Sim: {row[2]:.2f}] {row[0]} (Dept: {row[1]})")

import requests

# Assume `retrieved_content` is the string we got from pgvector
retrieved_content = "Policy 1A: Remote work is allowed on Tuesdays."

prompt = f"""
You are a helpful HR bot. Answer the user's question using ONLY the provided context.
If the context does not contain the answer, say "I don't know".

Context: {retrieved_content}
Question: {user_query}
"""

response = requests.post('http://localhost:11434/api/generate', json={
    "model": "llama3",
    "prompt": prompt,
    "stream": False
})

print(response.json()['response'])
# Output: "According to Policy 1A, remote work is allowed on Tuesdays."

import chromadb

# 1. Initialize a persistent database on disk (creates a SQLite file locally)
client = chromadb.PersistentClient(path="./my_chroma_db")

# 2. Tell Chroma to use Cosine distance (default is L2 Euclidean)
# Chroma automatically downloads the 'all-MiniLM-L6-v2' model locally.
collection = client.get_or_create_collection(
    name="company_policies",
    metadata={"hnsw:space": "cosine"}
)

# 3. Insert documents (Chroma vectorizes them automatically under the hood)
collection.add(
    documents=[
        "Policy 1A: Remote work is allowed on Tuesdays.",
        "Policy 2B: Expense reports must be filed by Friday.",
        "The cafeteria serves pizza on Wednesdays."
    ],
    metadatas=[{"dept": "HR"}, {"dept": "Finance"}, {"dept": "General"}],
    ids=["doc1", "doc2", "doc3"]
)

# 4. Search with a CONFIDENCE THRESHOLD
user_query = "When can I work from home?"
# Note: Chroma returns 'Distance'. Closer to 0.0 is better. 
# A distance of 0.4 means roughly 60% similarity match.
threshold_distance = 0.4  

# We can pass metadata filters directly into the query method
results = collection.query(
    query_texts=[user_query],
    where={"dept": "HR"}, # Metadata filtering
    n_results=3
)

# 5. Filter results manually 
# (Chroma does not yet natively support distance filtering in the query API)
print("--- Chroma Search Results ---")
for doc, dist, meta in zip(results['documents'][0], results['distances'][0], results['metadatas'][0]):
    if dist < threshold_distance:
        print(f"[PASS - Dist: {dist:.2f}] {doc} (Dept: {meta['dept']})")
    else:
        print(f"[REJECTED - Dist: {dist:.2f}] {doc}")

def agentic_router(user_query: str):
    # We ask the LLM to choose a tool based on the query
    routing_prompt = f"""
    Given the user query: '{user_query}'
    Decide which tool to use. Output ONLY the tool name:
    - 'VECTOR_DB': For questions about internal company policies or past reports.
    - 'SQL_DB': For questions about exact numbers, metrics, or active users.
    - 'WEB_SEARCH': For questions about current events or live weather.
    """
    
    decision = llm.invoke(routing_prompt).strip()
    
    if decision == 'VECTOR_DB':
        return execute_pgvector_search(user_query)
    elif decision == 'SQL_DB':
        # Agent writes SQL and queries the tabular database
        return execute_text_to_sql(user_query)
    else:
        return execute_tavily_web_search(user_query)