Documents → Chunking → Embedding → Vector DB
                                       ↓
User Query → Embed Query → Retrieve Top-K → Rerank → Context Assembly → LLM → Answer

☐ Split documents into chunks (300–600 tokens is typical for dense text)
☐ Preserve metadata per chunk: source URL, page number, section heading, date, author
☐ Handle multiple formats: PDF, HTML, Markdown, DOCX, plain text
☐ Strip boilerplate from web sources (nav, headers, footers, cookie banners)
☐ Deduplicate chunks with a content hash before embedding
☐ Test chunking on your actual data — verify no splits mid-sentence or mid-table
☐ Store chunk text alongside its vector — never rely on ID-only lookups

def fixed_chunk(text: str, size: int = 500, overlap: int = 50) -> list[str]:
    words = text.split()
    chunks = []
    for i in range(0, len(words), size - overlap):
        chunk = " ".join(words[i : i + size])
        if chunk:
            chunks.append(chunk)
    return chunks

from langchain_text_splitters import RecursiveCharacterTextSplitter

splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", ". ", " ", ""],
)
chunks = splitter.split_text(document_text)

from langchain_experimental.text_splitter import SemanticChunker
from langchain_openai import OpenAIEmbeddings  # or any embed model

chunker = SemanticChunker(
    OpenAIEmbeddings(),
    breakpoint_threshold_type="percentile",
    breakpoint_threshold_amount=95,
)
chunks = chunker.split_text(document_text)

☐ Choose model appropriate for your domain (general vs code vs legal vs medical)
☐ Use the exact same model at query time as at ingestion time
☐ Normalize embeddings (most models expect cosine similarity on unit vectors)
☐ Batch embed during ingestion — avoid per-chunk API calls
☐ Store raw text + metadata alongside each vector

from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer("all-MiniLM-L6-v2")   # 384 dims, fast

chunks = ["The capital of France is Paris.", "Python 3.12 released in 2023."]
embeddings = model.encode(chunks, normalize_embeddings=True, batch_size=32)

print(embeddings.shape)   # (2, 384)

(2, 384)

import chromadb
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("all-MiniLM-L6-v2")
client = chromadb.Client()   # in-memory; use PersistentClient for disk
collection = client.create_collection("docs")

# Ingest
texts = ["Paris is the capital of France.", "Berlin is the capital of Germany."]
embeddings = model.encode(texts, normalize_embeddings=True).tolist()
collection.add(
    documents=texts,
    embeddings=embeddings,
    ids=["doc-0", "doc-1"],
    metadatas=[{"source": "geography"}, {"source": "geography"}],
)

# Query
query_vec = model.encode(["What is the capital of France?"],
                          normalize_embeddings=True).tolist()
results = collection.query(query_embeddings=query_vec, n_results=2)
print(results["documents"][0])

['Paris is the capital of France.', 'Berlin is the capital of Germany.']

-- Enable extension
CREATE EXTENSION vector;

-- Table with embedding column
CREATE TABLE doc_chunks (
    id       SERIAL PRIMARY KEY,
    source   TEXT,
    chunk    TEXT,
    embedding VECTOR(384)
);

-- Approximate nearest-neighbor index (HNSW — fast)
CREATE INDEX ON doc_chunks USING hnsw (embedding vector_cosine_ops);

import psycopg2
import numpy as np

conn = psycopg2.connect("postgresql://user:pass@localhost/mydb")
cur = conn.cursor()

query_vec = model.encode(["What is the capital of France?"],
                          normalize_embeddings=True)[0]

cur.execute(
    """
    SELECT source, chunk, 1 - (embedding <=> %s::vector) AS similarity
    FROM doc_chunks
    ORDER BY embedding <=> %s::vector
    LIMIT 5
    """,
    (query_vec.tolist(), query_vec.tolist())
)
rows = cur.fetchall()
for source, chunk, sim in rows:
    print(f"{sim:.3f}  [{source}]  {chunk[:80]}")

0.932  [geography]  Paris is the capital of France.
0.801  [geography]  France is a country in Western Europe.

def retrieve(query: str, k: int = 5) -> list[dict]:
    query_vec = model.encode([query], normalize_embeddings=True)[0]

    # Vector similarity search (top 2k candidates)
    vector_results = collection.query(
        query_embeddings=[query_vec.tolist()],
        n_results=k * 2
    )
    candidates = [
        {"text": doc, "metadata": meta, "score": None}
        for doc, meta in zip(
            vector_results["documents"][0],
            vector_results["metadatas"][0]
        )
    ]

    # Optional: cross-encoder reranking (high-value, ~100ms)
    # from sentence_transformers import CrossEncoder
    # reranker = CrossEncoder("cross-encoder/ms-marco-MiniLM-L-6-v2")
    # pairs = [(query, c["text"]) for c in candidates]
    # scores = reranker.predict(pairs)
    # candidates = sorted(zip(scores, candidates), reverse=True)
    # candidates = [c for _, c in candidates]

    return candidates[:k]

def build_context(chunks: list[dict], max_tokens: int = 6000) -> str:
    context_parts = []
    token_count = 0

    for chunk in chunks:
        # Rough token estimate: 1 token ≈ 4 chars
        chunk_tokens = len(chunk["text"]) // 4
        if token_count + chunk_tokens > max_tokens:
            break
        source = chunk["metadata"].get("source", "unknown")
        context_parts.append(f"[Source: {source}]\n{chunk['text']}")
        token_count += chunk_tokens

    return "\n\n---\n\n".join(context_parts)

Answer the question using ONLY the sources provided below.
If the answer is not in the sources, say "I don't have enough information."
Do not speculate or draw on outside knowledge.
Cite sources by their [Source: ...] label.

Sources:
{context}

Question: {question}

Answer:

import anthropic

anthropic_client = anthropic.Anthropic()

def answer(question: str) -> str:
    chunks = retrieve(question, k=5)
    context = build_context(chunks)

    response = anthropic_client.messages.create(
        model="claude-opus-4-7",
        max_tokens=1024,
        messages=[{
            "role": "user",
            "content": (
                f"Answer using ONLY the sources below. "
                f"Cite sources. If not in sources, say so.\n\n"
                f"Sources:\n{context}\n\n"
                f"Question: {question}"
            )
        }]
    )
    return response.content[0].text

print(answer("What is the capital of France?"))

According to [Source: geography], Paris is the capital of France.

search_tool = {
    "name": "search_docs",
    "description": (
        "Search the documentation for relevant information. "
        "Call this when you need specific facts to answer the question. "
        "You may call it multiple times with different queries."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "query": {"type": "string", "description": "Search query"},
            "max_results": {"type": "integer", "default": 5}
        },
        "required": ["query"]
    }
}

def handle_search(inputs: dict) -> str:
    chunks = retrieve(inputs["query"], k=inputs.get("max_results", 5))
    return build_context(chunks)

# Let Claude drive the retrieval loop
answer = run_agent(
    user_message="What are the differences between Chroma and pgvector?",
    tools=[search_tool],
    max_turns=8,
)

☐ Faithfulness — does the answer use only retrieved chunks? (no hallucination)
☐ Answer relevance — does the answer address the actual question?
☐ Context recall — does the top-k contain the chunk needed to answer?
☐ Context precision — are retrieved chunks on-topic, or noisy?
☐ Latency — p50/p95 retrieval + generation time within SLA
☐ Hallucination rate — spot-check a sample against source documents

from rank_bm25 import BM25Okapi

# Pre-build BM25 index over the same corpus you embedded
tokenized_corpus = [doc.split() for doc in corpus_texts]
bm25 = BM25Okapi(tokenized_corpus)

def hybrid_retrieve(query: str, k: int = 5, k_fetch: int = 20) -> list[dict]:
    # Dense vector candidates
    query_vec = model.encode([query], normalize_embeddings=True)[0]
    dense = collection.query(query_embeddings=[query_vec.tolist()], n_results=k_fetch)
    dense_ids = dense["ids"][0]

    # Sparse BM25 candidates
    sparse_scores = bm25.get_scores(query.split())
    sparse_ids = [str(i) for i in sparse_scores.argsort()[::-1][:k_fetch]]

    # Reciprocal Rank Fusion (constant k=60 is community standard)
    rrf_scores: dict[str, float] = {}
    for rank, doc_id in enumerate(dense_ids):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + 1.0 / (60 + rank)
    for rank, doc_id in enumerate(sparse_ids):
        rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + 1.0 / (60 + rank)

    top = sorted(rrf_scores.items(), key=lambda x: x[1], reverse=True)[:k]
    return [{"id": did, "score": score} for did, score in top]

def hyde_retrieve(query: str, k: int = 5) -> list[dict]:
    # Generate a hypothetical answer to embed instead of the question
    hyde_prompt = (
        f"Write a short hypothetical answer to the following question. "
        f"Style should match a technical documentation excerpt. "
        f"Do not say 'I don't know' — invent a plausible answer.\n\n"
        f"Question: {query}"
    )
    fake_answer = call_claude(hyde_prompt, max_tokens=300)
    fake_vec = model.encode([fake_answer], normalize_embeddings=True)[0]

    results = collection.query(query_embeddings=[fake_vec.tolist()], n_results=k)
    return [
        {"text": d, "metadata": m}
        for d, m in zip(results["documents"][0], results["metadatas"][0])
    ]

def decompose_query(query: str) -> list[str]:
    """Break a compound question into atomic sub-queries."""
    prompt = (
        "Break the following question into 1-4 atomic sub-questions, each of which "
        "can be answered from a single document. Return ONE per line. No prose.\n\n"
        f"Question: {query}"
    )
    raw = call_claude(prompt, max_tokens=300)
    return [line.strip("-* ").strip() for line in raw.splitlines() if line.strip()]

# Chroma example: filter by metadata before similarity search
results = collection.query(
    query_embeddings=[query_vec.tolist()],
    n_results=5,
    where={
        "$and": [
            {"published_date": {"$gte": "2024-01-01"}},
            {"language": {"$eq": "en"}},
            {"access_tier": {"$in": ["public", "internal"]}},
        ]
    },
)

-- pgvector equivalent: combine vector search with WHERE clause
SELECT source, chunk, 1 - (embedding <=> $1::vector) AS similarity
FROM doc_chunks
WHERE published_date >= '2024-01-01'
  AND language = 'en'
  AND access_tier IN ('public', 'internal')
ORDER BY embedding <=> $1::vector
LIMIT 5;

import time, json, uuid

def traced_answer(question: str) -> dict:
    trace_id = str(uuid.uuid4())
    t0 = time.time()

    # Retrieval
    t1 = time.time()
    chunks = retrieve(question, k=5)
    t_retrieve = time.time() - t1

    # Context assembly
    t1 = time.time()
    context = build_context(chunks)
    t_assemble = time.time() - t1

    # Generation
    t1 = time.time()
    answer = generate(question, context)
    t_generate = time.time() - t1

    trace = {
        "trace_id": trace_id,
        "question": question,
        "retrieved_ids": [c.get("id") for c in chunks],
        "retrieved_scores": [c.get("score") for c in chunks],
        "context_chars": len(context),
        "answer": answer,
        "latency_ms": {
            "retrieve": int(t_retrieve * 1000),
            "assemble": int(t_assemble * 1000),
            "generate": int(t_generate * 1000),
            "total": int((time.time() - t0) * 1000),
        },
    }
    print(json.dumps(trace))    # ship to your logging stack
    return {"answer": answer, "trace_id": trace_id}